/*
 * 업무 KPI 조회 — **창 해소 + 저널 조회 + 폴드**를 한 곳에.
 *
 * 왜 한 곳인가: 처음에 GraphQL 질의와 AI 도구가 각각 같은 조회를 적었다. 그러면 창 규칙이 두 곳에서
 * 갈라지고(한쪽만 고치면 화면과 AI 가 다른 숫자를 말한다) 조회 상한·테넌트 조건도 복제된다.
 * 계산은 `kpi-fold`(순수)가 소유하고, 이 파일은 **무엇을 읽을지**만 정한다.
 *
 * ── 창의 기준 시각은 "트윈의 시계" 다 ──────────────────────────────────────
 * 시뮬 트윈은 자기 클록으로 산다(예: 저널의 eventTime 이 2026-01-01T00:25Z 인데 실제 지금은 7월).
 * 그런 트윈에 "최근 1시간" 을 **실제 시각**으로 물으면 그 구간에는 아무 기록이 없어 항상
 * "측정된 기록 없음" 이 된다 — 폴드는 정상인데 답이 쓸모없다.
 *
 * 그래서 호출부가 끝 시각을 주지 않으면 **그 트윈의 마지막 기록 시각**을 끝으로 잡는다. 커널의 시각
 * 상수를 앱에서 재현하지 않고(무방언), 라이브 트윈에서도 자연스럽다(마지막 기록 ≈ 지금).
 * 어떤 기준을 썼는지는 결과의 `window.basis` 로 밝힌다 — 숨기면 사용자가 숫자를 오해한다.
 */
import { twinWarn } from './log.js'
import { mergeStreams } from '../service/twin-journal/event-type-split-shape.js'
import { hydratedDate } from './read-time.js'
/* 시간대 이름을 순간마다 푸는 함수 — engine 이 service 를 들이는 것은 이 파일들의 관행이다(§`twin-engine.ts`). */
import { utcOffsetOf } from '../service/reference/reference-master.js'
import { Between, In } from 'typeorm'

import { type TariffDeclaration } from '@operato/twin-kernel'
import { hierarchyOf, levelOfLocationType, electricalUpstreamOf, EMS_LOCATION_TYPES, EMS_PROPERTY } from '@operato/ops-contract'

import { getReadRepository, getRepository } from '@things-factory/shell'

import { TwinEvent } from '../service/twin-event/twin-event.js'
import { TwinInstance } from '../service/twin-instance/twin-instance.js'
import { TwinArea } from '../service/twin-space/twin-area.js'
import { TwinSpace } from '../service/twin-space/twin-space.js'
import { TwinTarget } from '../service/twin-target/twin-target.js'
import { TWIN_METRIC, judgeTargets, judgeGroupTargets, scopedTargets, type ScopedTargets, type TargetVerdict, type TwinMetricId } from './kpi-target.js'
import { TwinEngine } from './twin-engine.js'
import { buildEnergyTopology, type EnergyTopology } from './energy-topology.js'
import {
  crossGroups,
  deltaP50,
  foldKpi,
  foldKpiBuckets,
  shiftWindow,
  type CrossedGroup,
  type KpiEvent,
  type KpiGroupBy,
  type KpiResult
} from './kpi-fold.js'
import { MAX_BASELINE_PERIODS, baselineWindows, summarizeBaseline, type BaselineStride, type MetricBaseline } from './kpi-baseline.js'

/**
 * 업무 전이만 읽는다 — 설비 상태(equipment.status)는 OEE 누적기의 몫이고 여기선 쓰지 않는다.
 *
 * **에너지 구간 마감**은 함께 읽는다(2026-08-14): 원단위는 분자(전력량)와 분모(산출)를 **같은 창**에서
 * 나눈 값이라, 두 사실이 한 폴드에 들어와야 창이 어긋날 수 없다. 표본(`energy.measured`)은 읽지 않는다 —
 * 저널에 없고(§4.1 인제스트 계약) 여기서 필요한 것은 마감된 구간뿐이다.
 */
const BUSINESS_EVENTS = ['task.status', 'order.status', 'energy.demand.window', 'energy.generation.period', 'energy.usage.period', 'energy.bill', 'energy.generation.price']

/** 한 번에 계산할 이벤트 상한 — 큰 구간이 서버를 붙잡지 않게. */
const MAX_EVENTS = 20000

/**
 * 종류를 나눠 물을 때 만들 갈래 수의 상한 — 대상 × 종류.
 *
 * 갈래마다 상한(`MAX_EVENTS`)까지 받아야 합친 결과가 정확하다(§`mergeStreams`). 그래서 갈래가 늘면
 * 메모리가 그만큼 든다. 셋(종류) × 둘(트윈)까지는 최악에 12만 행이고, 그 이상은 예전 한 질의로 간다 —
 * 느리지만 답은 같다.
 */
/*
 * ── 나눠 물을 수 있는 갈래의 최대 (2026-08-30 실측) ──────────────────────────
 *
 * 6 이었다. 그런데 **한 현장에 트윈이 여럿인 것이 이 제품의 정상**이다(창고·공정·야드가 한 현실을
 * 각각 본다). 트윈 3개 × 종류 6가지 = 18 이라 조건에 안 걸렸고, 한 질의로 떨어졌다.
 *
 * 같은 창·같은 행으로 두 모양을 쟀다(hatio-us, order.status 345만 건이 있는 하루):
 *
 *   한 질의(종류 IN + 시각순)   ix_twin_event_1 로 그 날 전체를 걷는다   25.987초 · 20,000건(잘림)
 *   갈래 하나(종류 정확일치)     ix_twin_event_7 이 순서까지 준다          0.286초
 *
 * 91배다. 그리고 25초를 기다린 끝에 나오는 값이 **잘린 값**이다 — 사용자는 그 사실을 25초 뒤에 본다.
 *
 * ── 늘리는 대가 ────────────────────────────────────────────────────────────
 * 갈래마다 상한까지 받아야 합칠 때 정확하므로(§`mergeStreams`), 갈래가 많으면 그만큼 든다.
 * 최악은 `갈래 수 × MAX_EVENTS` 행이다. 다만 그 최악은 **모든 갈래가 창을 가득 채울 때**이고,
 * 그때는 어차피 결과가 잘린다(그 사실은 `capped` 로 나간다).
 *
 * 32 는 트윈 4개 × 종류 8가지를 덮는다. 어휘가 더 늘면 이 수를 다시 재고 정한다.
 */
const KPI_MAX_SPLIT = 32

export interface TwinKpiInput {
  domainId: string
  /** 한 트윈. `spaceId` 와 둘 중 하나는 있어야 한다. */
  instanceId?: string
  /**
   * 한 **공간** — 그 공간에서 가동 중인 트윈 전부를 **한 번에 계산한다**.
   *
   * 트윈별로 계산해 평균을 다시 평균하면 분위수가 거짓이 된다(p90 의 평균은 p90 이 아니다).
   * 그래서 이벤트를 모아 **한 번** 계산한다. 사용자는 "부산 창고" 를 보고, 그 안에 트윈이 몇 개든
   * 합쳐서 본다(헤더 트윈 셀렉터를 없앤 결정과 같은 규율).
   */
  spaceId?: string
  /** 창의 끝(ISO). 미지정이면 **그 트윈의 마지막 기록 시각**. */
  toTime?: string
  /** 창의 시작(ISO). 미지정이면 끝에서 windowMinutes 만큼 뒤로. */
  fromTime?: string
  /** 창 길이(분). fromTime 이 있으면 무시. 기본 60, 1~1440. */
  windowMinutes?: number
  /** 시작 이벤트를 찾기 위해 창보다 앞으로 더 읽는 시간(분). 기본 60, 0~1440. */
  lookbackMinutes?: number
  /**
   * 창을 N 등분해 **구간별 값도** 함께 낸다(추세용). 1~48, 기본 0(쪼개지 않음).
   * 조회는 한 번이고 계산만 N 번 반복한다 — 폴드가 순수하므로 값싸다.
   */
  buckets?: number
  /**
   * 같은 사실을 **다른 관점으로** 쪼갠다: 자원별·작업종류별·도착지점별·오더별·구역별.
   *
   * 고정된 한 렌즈만 낼 수 있으면 "구역별로는 어떤가" 라는 질문에 화면도 AI 도 답할 수 없다.
   * 축은 조회 인자 하나이고, 계산는 순수 함수라 축을 하나 더 보는 값이 거의 없다.
   */
  groupBy?: KpiGroupBy
  /** 축의 상한(기본 20, 최대 100). 넘치면 잘라내고 잘린 수를 함께 알린다. */
  groupLimit?: number
  /**
   * **무엇과 비교할 것인가** — 같은 길이의 앞 구간(previous)이나 하루 전 같은 시간(yesterday).
   *
   * 숫자 하나로는 좋아졌는지 나빠졌는지 알 수 없다("29초" 가 개선인지 악화인지). 창만 옮겨 한 번 더
   * 계산하면 되므로(폴드는 순수) 값싸다. 비교 구간에 기록이 없으면 **delta 를 만들지 않는다** —
   * 없는 것을 0% 로 그리면 "변화 없음" 이라는 거짓이 된다.
   */
  compareTo?: 'previous' | 'yesterday'
  /**
   * **여러 구간 동시 비교** — 최근 N 구간의 같은 시간대를 함께 계산해 **분포**와 그 안의 자리를 낸다.
   *
   * `compareTo` 는 한 구간만 봐서 판단이 "어제보다 나아졌다" 에서 멈춘다 — 어제도 나빴으면 쓸모가 없다.
   * 최근 며칠의 같은 시간대를 보면 목표 없이도 "지난 7일 중 가장 낮다" 를 말할 수 있다.
   *
   * `stride: 'week'` 는 같은 요일을 본다(주말·평일이 섞이지 않게). `periods` 는 1~14
   * (창마다 조회가 한 번 더 붙는다 — 값을 공짜로 말하지 않는다).
   *
   * `compareTo` 와 함께 쓸 수 있다: 다른 질문이기 때문이다(직전 한 구간 vs 최근 분포).
   */
  baseline?: { periods: number; stride?: BaselineStride }
}

export interface TwinKpiOutput extends Omit<Partial<KpiResult>, 'window'> {
  /** 사람이 읽는 창(ISO) + 기준. 폴드의 raw 창(ms)은 여기에 흡수한다. */
  window: {
    fromTime: string
    toTime: string
    minutes: number
    /** 끝 시각을 어디서 얻었나 — 'given'(호출부 지정) | 'latest-event'(트윈의 마지막 기록). */
    basis: 'given' | 'latest-event'
  }
  /**
   * 조회가 **상한에 걸렸는가**(구간이 너무 커서 일부만 읽었을 수 있다).
   *
   * true 면 이 구간의 숫자는 **하한**이다 — 실제로는 더 있었을 수 있다. 알리지 않고 잘라내면 숫자가 알리지 않고
   * 작아지고 사용자는 그것을 성과 하락으로 읽는다. 구간을 좁혀 다시 물어야 한다.
   */
  eventsCapped?: boolean
  /**
   * 이 구간에 **측정할 기록이 있었는가**. false 는 "처리량 0" 이 아니라 **"기록이 없다"** 는 뜻이다 —
   * 두 상황은 완전히 다르고, 섞으면 사용자가 잘못된 결론을 낸다.
   */
  measured: boolean
  note?: string
  /** 무엇을 집계했나 — 공간이면 합쳐 계산한 트윈 목록을 밝힌다(숫자의 출처를 숨기지 않는다). */
  /**
   * 무엇을 집계했나 — 공간이면 합쳐 계산한 트윈 목록.
   *
   * `restartPolicies` 는 **이 숫자가 무엇의 성과인지**를 말한다: mirror(실제 시스템을 반영) ·
   * `resume`(이어지는 가상 세계) · `reset`(재현 실험 사본). 같은 "345건" 이 현실의 실적일 수도, 시뮬레이션의
   * 산출일 수도 있다 — 그 구별 없이 숫자만 보여주면 사용자가 시뮬 결과를 실적으로 착각한다.
   */
  scope: { spaceId?: string; instanceIds: string[]; restartPolicies: string[] }
  /** 구간별 값(요청했을 때만) — 추세를 그리기 위한 최소 정보. */
  buckets?: { fromTime: string; toTime: string; tasks: number; orders: number; workP50Ms: number }[]
  /**
   * **목표 대비 판정** — 숫자를 판단으로 바꾸는 값(§kpi-target).
   *
   * 목표가 걸린 지표만 들어온다. 목표가 없거나 그 구간에 값이 없는 지표는 **여기 나타나지 않는다** —
   * 전부 "목표 없음" 으로 채우면 화면이 회색 노이즈가 되고, 없는 값을 0 으로 놓으면 "목표 크게 미달"
   * 이라는 없는 사실이 만들어진다.
   */
  targets?: TargetVerdict[]
  /**
   * 비교 구간(요청했을 때만) — 같은 규칙·같은 길이로 계산한 앞 구간과의 차이.
   *
   * `measured:false` 면 그 구간에 기록이 없다는 뜻이고 `delta` 는 없다(0 이 아니다).
   * delta 는 **현재 − 비교구간** 이다. 좋고 나쁨은 판단하지 않는다 — 처리량은 클수록 좋고 소요는
   * 작을수록 좋다는 해석은 소비처(화면·AI)의 몫이고, 계산 층이 판단을 섞으면 목표 없이 판정하게 된다.
   */
  comparison?: {
    basis: 'previous' | 'yesterday'
    window: { fromTime: string; toTime: string }
    measured: boolean
    throughput?: { tasks: number; orders: number }
    workTime?: KpiResult['workTime']
    leadTime?: KpiResult['leadTime']
    waitTime?: KpiResult['waitTime']
    delta?: { tasks: number; orders: number; workP50Ms?: number; leadP50Ms?: number; waitP50Ms?: number }
  }
  /**
   * **여러 구간 분포**(요청했을 때만) — 지표별로 최근 N 구간의 분포와 지금의 자리.
   *
   * `samples` 는 **요청한 창 전부**다(기록 없는 창도 `measured:false` 로 남는다) — 빼고 말하면
   * "지난 7일" 이 실제로는 3일이었다는 것을 아무도 모른다. 자리를 말할 수 없으면 `noPosition` 이
   * 이유를 밝힌다(표본 부족·현재 값 없음). 방향은 지표의 성질이라 `standing` 이 그것을 적용한 읽기다.
   */
  baseline?: {
    stride: BaselineStride
    /** 요청한 과거 구간 수(상한 적용 후). */
    periods: number
    metrics: MetricBaseline[]
    /** 과거 창 조회 중 하나라도 상한에 걸렸나 — 그 창의 값은 하한이라 분포가 낮게 보일 수 있다. */
    capped?: boolean
  }
  /**
   * 관점별 값(요청했을 때만). `key` 는 식별자, `label` 은 사람이 읽는 이름(있을 때만 — 구역은 이름이
   * 따로 저장되어 있다). `truncated` 는 상한 때문에 빠진 축의 수다.
   */
  groups?: {
    by: KpiGroupBy
    items: (KpiGroupOut & { label?: string })[]
    /**
     * 비교 구간에는 일이 있었는데 **지금은 없는** 축. 표에서 그냥 빠지면 멈춘 구역·대기 중인 자원이
     * 눈에 보이지 않는다 — 운영에서는 이것이 가장 중요한 신호일 수 있다.
     */
    disappeared?: { key: string; label?: string; tasks: number }[]
    /** 상한 때문에 빠진 축의 수와, 그 축들이 안고 있던 건수(둘 다 알려야 꼬리를 놓치지 않는다). */
    truncated?: number
    truncatedTasks?: number
    note?: string
  }
}

/** 축 하나의 값 — 폴드의 KpiGroup 과 같은 모양(시각 변환이 없어 그대로 통과한다). */
type KpiGroupOut = NonNullable<KpiResult['groups']>['items'][number] &
  Partial<Pick<CrossedGroup, 'prevTasks' | 'deltaTasks' | 'deltaWorkP50Ms' | 'isNew'>>

/**
 * 노드 → 구역 지도. **저널에는 구역이 없다**(노드까지만 있다) — 모델 정의의 `locations[].parentId` 가
 * 마스터 계층에서 온 소속이다. 기동하지 않은 트윈도 모델은 남아 있어 이 경로가 가장 튼튼하다.
 *
 * **한 홉만 보면 안 된다.** `parentId` 는 구역일 수도 있고 중간 단(라인·셀)일 수도 있다. 라인이 끼면
 * 한 홉짜리 코드는 "구역이 LINE-1" 이라고 답하고, 그 id 는 구역 목록에 없으므로 그 아래 스테이션이
 * **집계에서 오류 없이 빠진다**(에러 없이 틀린 숫자). 그래서 커널 `hierarchyOf` 로 사슬을 끝까지 걷는다 —
 * 걷는 규칙은 커널에 한 벌뿐이고 소비처는 그것만 부른다.
 */
/**
 * 교대 축의 재료 — **교대 선언과 현장의 시각 기준.**
 *
 * 저널에는 시각만 있고 교대가 없다(일부러 그렇게 뒀다 — 교대는 파생이라 찍어 두면 나중에 교대 시간을
 * 고쳤을 때 과거가 옛 구분으로 굳는다). 그래서 폴드가 계산할 재료를 여기서 모은다.
 *
 * 선언의 출처는 **모델의 설비**다. 여러 설비가 서로 다른 교대를 가질 수 있지만, 축은 하나여야 하므로
 * **가장 많이 쓰이는 선언 한 벌**을 쓴다 — 3교대 공장은 전 설비가 같은 교대이므로 이것이 곧 정답이고,
 * 자원마다 교대가 다른 현장에서는 이 축이 거칠다는 뜻이다(그 사실을 여기 적어 둔다).
 *
 * 시각 기준은 모델에 실려 있다(공간이 갖는 값을 기동 시 호스트가 풀어 넣는다). 없으면 UTC 로 읽고,
 * 그 기본값은 커널 계약에 밝혀져 있다.
 *
 * **이름이 있으면 이름을 쓴다**(`timezone`). 접기는 지난 구간을 훑으므로 고정 오프셋으로는 일광절약시간
 * 전환을 넘을 수 없다 — 1년치를 한 오프셋으로 읽으면 그 안의 전환 두 번이 사실을 다른 교대로 세게 하고,
 * 그 어긋남은 오류를 내지 않는다(2026-09-07).
 */
/**
 * 이 범위에 걸린 운영 목표 — **좁은 선언이 넓은 선언을 덮는다.**
 *
 * 공간 전체 목표 위에 인스턴스 목표를 얹을 수 있어야 예외를 표현할 수 있다("이 라인만 다르다").
 * 같은 지표에 둘 다 있으면 인스턴스가 이긴다.
 */
async function loadTargets(
  domainId: string,
  spaceId: string | undefined,
  instanceIds: string[],
  axis?: string
): Promise<ScopedTargets> {
  /* 조회는 **가져오기만** 한다 — 어느 목표가 이기는지는 순수 함수(`scopedTargets`)가 정한다.
     그 규칙이 DB 접근 뒤에 있으면 화면을 눌러 봐야만 확인된다(그리고 실제로 확인되지 않았다). */
  const rows = await getRepository(TwinTarget).find({ where: { domain: { id: domainId } } }).catch(() => [])
  return scopedTargets(rows, { spaceId, instanceIds, axis })
}

async function shiftContext(
  domainId: string,
  instanceIds: string[]
): Promise<{ shifts?: any[]; utcOffsetMinutes?: number; offsetAt?: (atMs: number) => number | undefined }> {
  if (!instanceIds.length) return {}
  const rows = await getRepository(TwinInstance).find({ where: { domain: { id: domainId }, instanceId: In(instanceIds) } })
  const tally = new Map<string, { decl: any[]; n: number }>()
  let utcOffsetMinutes: number | undefined
  let timezone: string | undefined
  for (const row of rows) {
    const model = row.model as any
    if (utcOffsetMinutes === undefined && typeof model?.utcOffsetMinutes === 'number') utcOffsetMinutes = model.utcOffsetMinutes
    if (timezone === undefined && typeof model?.timezone === 'string' && model.timezone) timezone = model.timezone
    for (const e of (model?.equipment ?? []) as any[]) {
      const decl = e?.workCalendar
      if (!Array.isArray(decl) || !decl.length) continue
      const key = JSON.stringify(decl)
      const hit = tally.get(key)
      if (hit) hit.n++
      else tally.set(key, { decl, n: 1 })
    }
  }
  let best: { decl: any[]; n: number } | undefined
  for (const v of tally.values()) if (!best || v.n > best.n) best = v
  /*
   * 이름이 있으면 **푸는 함수**로 바꿔 넘긴다. 접기가 레코드마다 부르므로 그 순간의 오프셋이 나오고,
   * 일광절약시간 전환이 든 구간도 맞게 세어진다(§`FoldOptions.offsetAt`).
   *
   * 수(`utcOffsetMinutes`)도 함께 남긴다 — 이름이 없는 공간에서는 그것이 유일한 답이고, 이름이 있어도
   * 그 순간을 못 푸는 경우의 마지막 기댈 자리다.
   */
  return {
    ...(best ? { shifts: best.decl } : {}),
    ...(utcOffsetMinutes !== undefined ? { utcOffsetMinutes } : {}),
    ...(timezone ? { offsetAt: (atMs: number) => utcOffsetOf(timezone, new Date(atMs)) } : {})
  }
}

async function locationAreaMap(
  domainId: string,
  instanceIds: string[],
  spaceId?: string
): Promise<{ locationArea: Record<string, string>; labels: Record<string, string> }> {
  const locationArea: Record<string, string> = {}
  if (instanceIds.length) {
    const rows = await getRepository(TwinInstance).find({
      where: { domain: { id: domainId }, instanceId: In(instanceIds) }
    })
    for (const row of rows) {
      const locations = (((row.model as any)?.locations ?? []) as any[]).filter(n => n?.id)
      if (!locations.length) continue
      /* 순환은 hierarchyOf 가 오류를 낸다 — 한 트윈의 잘못된 계층이 다른 트윈의 집계까지 죽이지 않게 여기서 가둔다. */
      let h
      try {
        /*
         * 단 해석기를 함께 넘긴다 — 자리가 `level` 을 적지 않으면 **타입이 답한다**(카탈로그 선언).
         * 이 집계는 지금 구역 롤업(`rollupOf`)만 쓰므로 결과가 달라지지 않는다. 그래도 여기서
         * 잇는다: 단으로 묻는 질의(`ancestorOfLevel`)가 생길 때 배선을 다시 찾지 않게 한다.
         */
        h = hierarchyOf({ locations }, t => levelOfLocationType(t, row.kind))
      } catch (e: any) {
        twinWarn(`[twin-kpi] "${row.instanceId}": location hierarchy is broken — area rollup skipped for this twin: ${e?.message}`)
        continue
      }
      for (const n of locations) {
        const area = h.rollupOf(String(n.id))
        if (area) locationArea[String(n.id)] = area // 구역 미상은 넣지 않는다 — 없는 구역을 만들지 않는다
      }
    }
  }

  const labels: Record<string, string> = {}
  const sid = spaceId
  if (sid) {
    const space = await getRepository(TwinSpace).findOne({ where: { domain: { id: domainId }, spaceId: sid } })
    if (space) {
      const areas = await getRepository(TwinArea).find({
        where: { domain: { id: domainId } as any, space: { id: space.id } as any }
      })
      for (const a of areas) if (a.areaId && a.name) labels[a.areaId] = a.name
    }
  }
  return { locationArea, labels }
}

/**
 * 자원 → **배전 계통의 자리**(분기·수전) 지도와 그 자리의 이름.
 *
 * ── 무엇을 읽나 ─────────────────────────────────────────────────────────────
 * 에너지 트윈의 모델이 「이 설비가 어느 자리에 매달렸나」를 선언한다(`homeLocation`). 그 자리가 분기가
 * 아니라 하위 자리라면 부모를 타고 올라가 **전기 계통의 자리**를 찾는다 — 전기의 계통은 물리 구역과
 * 다른 나무이므로 구역 롤업(`locationAreaMap`)을 쓸 수 없다.
 *
 * ── 왜 에너지 트윈만 읽나 ───────────────────────────────────────────────────
 * 같은 설비가 공장 트윈에서는 자원(작업을 한다)이고 에너지 트윈에서는 부하(전기를 먹는다)다 — 사실 둘,
 * id 하나. 가동시간은 공장 트윈의 저널에서 오고, 전기 계통의 소속은 이 모델에서 온다. 그 둘을 id 로
 * 잇는 것이 이 지도의 일이다.
 *
 * 자리를 말하지 않는 설비는 **넣지 않는다** — 없는 소속을 만들면 그 몫이 엉뚱한 분기의 것이 된다.
 */
/**
 * 선언된 **전기요금 단가** — 에너지 트윈의 자리 속성에서 읽는다(계약전력과 같은 자리).
 *
 * 여러 에너지 트윈이 한 현장에 있으면 **처음 선언된 것 하나**를 쓴다. 평균을 내거나 합치지 않는다:
 * 요금은 계약이고, 두 계약을 섞은 단가는 어느 계약도 아니다. 서로 다른 단가가 선언돼 있으면 그것을
 * 밝힌다(알리지 않고 하나를 고르면 왜 그 금액인지 아무도 설명할 수 없다).
 */
async function tariffOf(
  domainId: string,
  instanceIds: string[]
): Promise<{ tariff?: TariffDeclaration; conflict?: boolean }> {
  if (!instanceIds.length) return {}
  const rows = await getRepository(TwinInstance).find({ where: { domain: { id: domainId }, instanceId: In(instanceIds) } })
  const found: TariffDeclaration[] = []
  for (const row of rows) {
    if (row.kind !== 'ems') continue
    for (const loc of ((row.model as any)?.locations ?? []) as any[]) {
      const props = new Map<string, any>(((loc?.properties ?? []) as any[]).map(p => [String(p?.id), p?.value]))
      const num = (id: string): number | undefined => {
        const v = Number(props.get(id))
        return Number.isFinite(v) && v > 0 ? v : undefined
      }
      const d = num(EMS_PROPERTY.demandChargePerKW)
      const e = num(EMS_PROPERTY.energyChargePerKWh)
      /* 요금적용전력 — 있으면 기본요금이 조건부 파생이 아니라 사실이 된다(§`electricityCost`). */
      const b = num(EMS_PROPERTY.billingDemandKW)
      const cur = props.get(EMS_PROPERTY.currency)
      if (d === undefined && e === undefined) continue
      found.push({
        ...(d !== undefined ? { demandChargePerKW: d } : {}),
        ...(e !== undefined ? { energyChargePerKWh: e } : {}),
        ...(b !== undefined ? { billingDemandKW: b } : {}),
        ...(cur ? { currency: String(cur) } : {})
      })
    }
  }
  if (!found.length) return {}
  const first = found[0]
  const same = found.every(
    t =>
      t.demandChargePerKW === first.demandChargePerKW &&
      t.energyChargePerKWh === first.energyChargePerKWh &&
      t.billingDemandKW === first.billingDemandKW &&
      t.currency === first.currency
  )
  return { tariff: first, ...(same ? {} : { conflict: true }) }
}

async function resourceFeederMap(
  domainId: string,
  instanceIds: string[]
): Promise<{ resourceFeeder: Record<string, string>; meterFeeder: Record<string, string>; labels: Record<string, string> }> {
  const resourceFeeder: Record<string, string> = {}
  /** 계량 지점 → 그것이 재는 분기. 있으면 그 분기 값은 **측정**이다(배분보다 강한 근거). */
  const meterFeeder: Record<string, string> = {}
  const labels: Record<string, string> = {}
  if (!instanceIds.length) return { resourceFeeder, meterFeeder, labels }

  const rows = await getRepository(TwinInstance).find({ where: { domain: { id: domainId }, instanceId: In(instanceIds) } })
  for (const row of rows) {
    if (row.kind !== 'ems') continue // 전기 계통을 선언하는 것은 에너지 트윈이다
    const model: any = row.model ?? {}
    const locations: any[] = (model.locations ?? []).filter((n: any) => n?.id)
    const byId = new Map(locations.map((n: any) => [String(n.id), n]))
    const electrical = new Set(EMS_LOCATION_TYPES as readonly string[])

    /**
     * 이 자리에서 **전기 상류**로 올라가 처음 만나는 계통의 자리. 없으면 undefined.
     *
     * 상류의 뜻은 커널이 읽는다(`electricalUpstreamOf`) — 공간 부모(공장·구역)를 따라 올라가면 전기적으로
     * 뜻이 없는 길을 걷게 되고, 옛 세대(분기의 `parentId` 에 수전)와 새 세대(`upstreamId`)가 섞여 있으므로
     * 그 판정을 여기서 다시 쓰면 두 벌이 된다.
     */
    const feederOf = (startId: string): string | undefined => {
      const seen = new Set<string>()
      let cur = byId.get(startId)
      while (cur && !seen.has(String(cur.id))) {
        seen.add(String(cur.id)) // 잘못된 모델의 순환이 여기서 멈춘다
        if (electrical.has(String(cur.type))) return String(cur.id)
        const up = electricalUpstreamOf(locations as any, String(cur.id))
        cur = up ? byId.get(up) : cur.parentId ? byId.get(String(cur.parentId)) : undefined
      }
      return undefined
    }

    for (const e of (model.equipment ?? []) as any[]) {
      const home = e?.homeLocation ?? e?.homeLocationId
      if (!e?.id || !home) continue
      /*
       * 계량기는 부하가 아니다 — 그것의 「몫」을 세면 총량이 이중으로 나뉜다. 대신 **어디를 재는지**를
       * 따로 모은다: 계량기가 분기에 있으면 그 분기 값은 배분이 아니라 **측정**이다(가장 강한 근거).
       */
      if (e.kind === 'meter') {
        const at = feederOf(String(home))
        /* 뿌리(수전)를 재는 계량기는 분기의 것이 아니다 — 그 값은 총량이고, 나머지는 뺄셈으로 나온다. */
        if (at && String(byId.get(at)?.type) !== 'incoming') meterFeeder[String(e.id)] = at
        continue
      }
      /* 차단기·태양광·축전지도 부하가 아니다. */
      if (e.kind === 'breaker' || e.kind === 'pv-array' || e.kind === 'battery') continue
      const feeder = feederOf(String(home))
      if (feeder) resourceFeeder[String(e.id)] = feeder
    }
    for (const n of locations) if (electrical.has(String(n.type)) && n.name) labels[String(n.id)] = String(n.name)
  }
  return { resourceFeeder, meterFeeder, labels }
}

/**
 * 이 공간의 에너지 트윈들의 계통도 — 모델을 읽어 트리로 처리한다.
 *
 * 여러 트윈이 한 공간에 있으면 **각 트윈의 뿌리를 나란히** 둔다(하나로 합치면 없는 결선을 그리게 된다).
 * 에너지 트윈이 없으면 만들지 않는다 — 빈 그림은 고장으로 읽힌다.
 */
/**
 * 이 트윈들이 이어받고 있는 **마지막 청구서** — 요금적용전력의 우선 근거.
 *
 * 저널이 아니라 상태에서 읽는 이유: 청구는 한 달에 한 번이라 하루 창의 조회에는 걸리지 않는다.
 * 그런데 그 값은 **그 창에도 적용된다**(요금 제도가 그렇게 군다). 커널이 재기동에서도 이어받으므로
 * 상태가 그 자리다.
 *
 * 트윈이 여럿이면 가장 늦게 끝난 것을 쓴다 — 현장 하나에 청구가 여럿이면 그것은 다른 문제이고,
 * 여기서 골라 감추지 않는다.
 */
function latestBillOf(domainId: string, instanceIds: string[]): { to: string; billingDemandKW?: number } | undefined {
  let best: { to: string; billingDemandKW?: number } | undefined
  for (const instanceId of instanceIds) {
    const bill = (TwinEngine.snapshot(domainId, instanceId) as any)?.energy?.lastBill
    if (!bill?.to) continue
    if (!best || Date.parse(bill.to) > Date.parse(best.to)) {
      best = { to: String(bill.to), ...(Number.isFinite(Number(bill.billingDemandKW)) ? { billingDemandKW: Number(bill.billingDemandKW) } : {}) }
    }
  }
  return best
}

/** 이 트윈들이 이어받고 있는 **발전 단가** — 창 안에 단가 사실이 없을 때 쓴다. */
function latestGenerationPriceOf(
  domainId: string,
  instanceIds: string[]
): { from: string; to: string; unitPrice: number; currency: string } | undefined {
  let best: { from: string; to: string; unitPrice: number; currency: string } | undefined
  for (const instanceId of instanceIds) {
    const p = (TwinEngine.snapshot(domainId, instanceId) as any)?.energy?.generationPrice
    if (!p?.from || !Number.isFinite(Number(p.unitPrice)) || !p.currency) continue
    if (!best || Date.parse(p.from) > Date.parse(best.from)) {
      best = { from: String(p.from), to: String(p.to), unitPrice: Number(p.unitPrice), currency: String(p.currency) }
    }
  }
  return best
}

/**
 * 이 트윈들의 **발전 정격** — 이용률의 분모.
 *
 * 커널이 고른다(설비 합 → 자리 합계, 교류 → 직류). 여러 트윈이면 합한다 — 한 현장의 발전은
 * 트윈마다 나뉘어 있어도 같은 현장의 발전이다. 근거가 갈리면 약한 쪽(직류·자리)으로 적는다.
 */
function generationRatedOf(
  domainId: string,
  instanceIds: string[]
): { kW: number; basis: 'ac' | 'dc'; from: 'equipment' | 'location' } | undefined {
  let kW = 0
  let basis: 'ac' | 'dc' = 'ac'
  let from: 'equipment' | 'location' = 'equipment'
  for (const instanceId of instanceIds) {
    /* 스냅샷에서 읽는다 — 커널이 고른 값이다. 여기서 모델을 다시 뒤지면 고르는 규칙이 두 벌이 된다. */
    const r = (TwinEngine.snapshot(domainId, instanceId) as any)?.energy?.generationRated
    if (!r || !(Number(r.kW) > 0)) continue
    kW += Number(r.kW)
    if (r.basis === 'dc') basis = 'dc'
    if (r.from === 'location') from = 'location'
  }
  return kW > 0 ? { kW, basis, from } : undefined
}

/** 이 트윈들이 이어받고 있는 **요금 기준** — 이 주기에 적용되는 값(청구서보다 먼저다). */
function latestTariffBasisOf(domainId: string, instanceIds: string[]): { from: string; billingDemandKW?: number } | undefined {
  let best: { from: string; billingDemandKW?: number } | undefined
  for (const instanceId of instanceIds) {
    const b = (TwinEngine.snapshot(domainId, instanceId) as any)?.energy?.tariffBasis
    if (!b?.from) continue
    if (!best || Date.parse(b.from) > Date.parse(best.from)) {
      best = { from: String(b.from), ...(Number.isFinite(Number(b.billingDemandKW)) ? { billingDemandKW: Number(b.billingDemandKW) } : {}) }
    }
  }
  return best
}

async function energyTopologyOf(
  domainId: string,
  instanceIds: string[],
  energy: { byFeeder?: { key: string; kWh: number; basis?: 'measured' | 'apportioned'; peakKW?: number }[] } | undefined
): Promise<EnergyTopology | undefined> {
  if (!instanceIds.length) return undefined
  const rows = await getRepository(TwinInstance).find({ where: { domain: { id: domainId }, instanceId: In(instanceIds) } })
  const ems = rows.filter(r => r.kind === 'ems' && r.model)
  if (!ems.length) return undefined
  const merged: EnergyTopology = { roots: [], unplaced: [], unmetered: [], loadsOnIncoming: 0 }
  for (const row of ems) {
    /*
     * 지금의 설비 상태를 함께 넘긴다 — 발전·저장의 **지금 값**은 구간 폴드가 아니라 스냅샷이 안다.
     * 없으면(정지된 트윈) 구조만 그린다.
     */
    const snap = TwinEngine.snapshot(domainId, row.instanceId)
    const t = buildEnergyTopology(row.model, energy, snap?.equipment)
    merged.roots.push(...t.roots)
    merged.unplaced.push(...t.unplaced)
    merged.unmetered.push(...t.unmetered)
    merged.loadsOnIncoming += t.loadsOnIncoming
  }
  return merged.roots.length ? merged : undefined
}

/**
 * 이 대상들의 **마지막 업무 기록 시각**(ISO). 기록이 없으면 null.
 *
 * ── 두 가지를 함께 고쳤다 (2026-08-28 실측) ─────────────────────────────────
 * 예전 버전은 이랬다.
 *
 *   findOne({ where: { domain: { id }, instanceId: In(…), eventType: In(셋) }, order: { revision: 'DESC' } })
 *
 * **① 관계로 물어 두 단계 질의가 된다.** 안쪽에 상한이 없어 조건에 맞는 행을 **전부 만든 뒤** 하나를
 * 고른다 — 그 트윈의 업무 사실이 97만 행이고 `payload` 까지 실려 온다. 이 파일의 다른 자리가 같은 것을
 * 실측해 적어 두었다(§`twinEvents`: 0.024초 대 12초). 서버 로그에 「느린 질의」로 찍혀 있던 것이
 * 이것이고, 현장 성과 화면의 7.2초가 여기서 나왔다.
 *
 * **② 「마지막 시각」을 원하면서 `revision` 으로 정렬했다.** 그 둘은 어긋난다 — 연결된 시스템을 되읽는
 * 트윈은 원본의 시각을 싣고 도착 순서로 리비전을 받으므로, 리비전이 가장 큰 행이 4월 사실일 수 있다
 * (같은 어긋남을 오늘 저널 목록에서도 고쳤다). 창의 끝을 그 값으로 잡으면 **창 자체가 틀린다.**
 *
 * 지금은 종류마다 `MAX(eventTime)` 을 묻는다. `(domain, instance, event_type, event_time)` 색인의
 * 끝을 집으므로 종류당 한 번이고, 셋 중 가장 늦은 것이 답이다.
 */
async function latestEventTime(domainId: string, instanceIds: string[]): Promise<string | null> {
  if (instanceIds.length === 0) return null
  const rows = await Promise.all(
    instanceIds.flatMap(instanceId =>
      BUSINESS_EVENTS.map(eventType =>
        getReadRepository(TwinEvent)
          .createQueryBuilder('e')
          .select('MAX(e.eventTime)', 'at')
          .where('e.domain = :domainId', { domainId })
          .andWhere('e.instanceId = :instanceId', { instanceId })
          .andWhere('e.eventType = :eventType', { eventType })
          .getRawOne<{ at: unknown }>()
          .catch(() => null)
      )
    )
  )
  /* 드라이버가 준 값은 시간대가 없을 수 있다 — 컬럼 변환을 지나게 한다(§`hydratedDate`). */
  const repo = getReadRepository(TwinEvent)
  let latest: Date | null = null
  for (const r of rows) {
    const at = hydratedDate(repo, 'eventTime', r?.at)
    if (at && (!latest || at.getTime() > latest.getTime())) latest = at
  }
  return latest ? latest.toISOString() : null
}

/**
 * 창 구간의 업무 이벤트 — 창보다 `lookbackMs` 만큼 앞까지 읽는다(완료된 작업의 시작이 창 밖일 수 있다).
 *
 * eventTime 은 날짜 컬럼이라 DB 가 시각으로 비교한다 — 예전 문자열 컬럼 시절의 "UTC ISO 만 들어오면
 * 사전식 비교가 맞는다" 는 전제가 필요 없다.
 */
async function fetchEvents(domainId: string, instanceIds: string[], fromMs: number, toMs: number, lookbackMs: number) {
  if (instanceIds.length === 0) return { events: [] as KpiEvent[], capped: false }
  /*
   * ── 필요한 세 컬럼만, 그리고 **읽기 연결로** (2026-08-22 실측) ─────────────
   * 이 읽기가 서버를 붙잡고 있었다. 실측(개발 DB 16.3GB · 저널 1,134만 행): 2만 행을 실어 오는 동안
   * 다른 가벼운 읽기가 **47.6초** 기다렸다. 로그에서 이 자리 뒤로 사용자·도메인 조회까지 30.9초로
   * 찍혔다 — TypeORM sqlite 드라이버는 연결을 하나만 들기 때문이다.
   *
   * **행 단위로 읽는 것 자체는 정당하다**: 이 폴드는 `p10·p50·p90` 을 낸다(§`kpi-fold`). 분위수는
   * 카운터로 만들 수 없고 트윈별 사전집계로도 만들 수 없다(p90 의 평균은 p90 이 아니다). 개별 값이
   * 있어야 한다.
   *
   * 정당하지 않은 것은 **읽는 방식**이었다. 세 가지를 고친다.
   *
   *  ① **엔티티를 만들지 않는다.** 폴드가 쓰는 것은 `eventType`·`eventTime`·`payload` 뿐인데
   *     `find()` 는 열여섯 컬럼을 실어 엔티티로 만들고 관계 로더를 지난다. CPU 프로파일에서
   *     `RawSqlResultsToEntityTransformer` 와 `stringToSimpleJson` 이 상위였다.
   *  ② **읽기 연결로 보낸다**(`getReadRepository`). 순수 읽기이므로 쓰기와 순서를 다툴 것이 없다.
   *     실측: 같은 2만 행 읽기에서 다른 읽기의 대기가 47.6초 → 2.8초.
   *  ③ **정렬 축을 거르는 축과 맞춘다.** 여기가 이 자리의 진짜 비용이었다.
   *
   * ── `ORDER BY revision` 이 `LIMIT` 을 무력화했다 (실측) ────────────────────
   * 거르는 축은 `event_time`(창)이고 정렬 축은 `revision` 이었다. 그러면 sqlite 는 창에 맞는 행을
   * **전부** 찾아 `revision` 으로 **전부 정렬한 뒤** 상한을 적용한다 — 상한이 정렬 뒤에 걸리므로
   * 아무것도 줄여 주지 않는다. 실측이 그것을 그대로 보였다(인스턴스 하나가 375만 행):
   *
   *   ORDER BY revision            LIMIT 20000 → 32.2초 · LIMIT 5000 → 28.3초 · LIMIT 2000 → 28.2초
   *   ORDER BY eventTime, revision LIMIT 20000 →  0.41초                                     (78배)
   *
   * 계획도 그렇게 말한다: 앞은 `USE TEMP B-TREE FOR ORDER BY`(전체 정렬), 뒤는
   * `USE TEMP B-TREE FOR LAST TERM OF ORDER BY` — 시각은 인덱스(`ix_twin_event_1`)에서 순서대로
   * 나오고 **동시각만** 임시 정렬한다. 즉 범위를 훑다가 상한에서 멈춘다.
   *
   * ── 순서의 뜻은 **완전히 같지는 않다** (측정해서 고친 문장) ────────────────
   * 처음에 「순서의 뜻은 같다」고 적었다. 재 보니 아니었다 — `revision` 순서와 `eventTime` 순서가
   * 어긋나는 자리가 실재한다(hatio-mx2: 426만 행 중 **7건** · rosarito-mes: 11만 중 8건). 미러는
   * 원본의 시각을 싣고 도착 순서로 리비전을 받으므로, 늦게 도착한 사건이 그 어긋남을 만든다.
   *
   * **그러나 폴드의 수치는 같다.** `foldTaskRecords`(커널 `task-fold.ts`)가 스트림 위치가 아니라
   * **시각으로** 판정한다:
   *
   *   startedMs   = Math.min(startedMs, at)
   *   completedMs = Math.max(completedMs, at)
   *
   * `min`·`max` 이므로 행이 어떤 순서로 와도 답이 같다. 소요시간·분위수·처리량은 순서와 무관하다.
   *
   * 순서에 민감한 자리는 **하나**다: 그 폴드의 `location` 이 「마지막에 본 것」이다. 같은 밀리초에 같은
   * 작업의 자리가 두 번 바뀌면 어느 것이 남는지가 순서로 갈린다(라이브 미러가 같은 밀리초에 여러 행을
   * 보내는 경우다). 수치가 아니라 **표시되는 자리 하나**이고, 그래서 이 바꿈의 값이 그 위험보다 크다.
   *
   * `revision` 을 둘째 축으로 남기는 이유도 그것이다 — 동시각의 순서가 실행마다 달라지지 않게 한다.
   *
   * **상한을 낮추는 것으로는 고쳐지지 않는다**(위 숫자가 그것을 말한다). 정렬 축이 원인이다.
   *
   * 남는 비용을 감추지 않는다: `payload` 파싱은 여전히 메인 스레드 일이고 연결을 나눠도 나뉘지 않는다
   * (실측 평균 1.5초). 그것을 더 줄이려면 폴드가 payload 에서 무엇을 쓰는지 좁혀야 한다 — 별 건이다.
   */
  const from = new Date(fromMs - lookbackMs)
  const to = new Date(toMs)

  /**
   * 한 갈래를 묻는다 — **종류를 정확일치로** 준다. 그것이 이 고침의 요점이다.
   */
  const readOne = (instanceId: string, eventType: string) =>
    getReadRepository(TwinEvent)
      .createQueryBuilder('e')
      .select(['e.eventType AS "eventType"', 'e.eventTime AS "eventTime"', 'e.revision AS "revision"', 'e.payload AS "payload"'])
      .where('e.domain = :domainId', { domainId })
      .andWhere('e.instanceId = :instanceId', { instanceId })
      .andWhere('e.eventType = :eventType', { eventType })
      .andWhere('e.eventTime BETWEEN :from AND :to', { from, to })
      .orderBy('e.eventTime', 'ASC')
      .addOrderBy('e.revision', 'ASC')
      .limit(MAX_EVENTS)
      .getRawMany<{ eventType: string; eventTime: unknown; revision: unknown; payload: unknown }>()

  /*
   * ── **종류를 나눠 묻는다** (2026-08-28 실측) ────────────────────────────────
   *
   * 예전에는 `eventType IN (셋)` + 시각 범위 + 시각순이었다. 그 모양에서 sqlite 는 종류 색인을 버리고
   * 시각 색인으로 **구간 전체를 걸으며 행마다 종류를 확인한다.** 승화푸드(저널 3,140만행) 실측:
   *
   *   창 1시간    구간 안 0행 · 쓰는 행 0          40밀리초
   *   창 24시간   구간 안 254,185행 · 쓰는 행 0   3,776밀리초   ← 25만 행을 걸어 0건을 쓴다
   *   창 7일      구간 안 2,070,625행 · 34,143    173밀리초    ← 상한에 일찍 닿아 오히려 빠르다
   *
   * 창을 넓히면 빨라지는 것 자체가 신호였다. 종류를 정확일치로 물으면
   * `ix_twin_event_7 (domain, instance, event_type, event_time)` 이 범위까지 색인에서 준다.
   *
   * 나눌 수를 제한한다 — 갈래마다 상한까지 받아야 정확하므로(§`mergeStreams`) 갈래가 많으면 메모리가
   * 그만큼 든다. 넘으면 예전 한 질의로 간다(느리지만 답은 같다).
   */
  const pairs = instanceIds.length * BUSINESS_EVENTS.length
  let raw: { eventType: string; eventTime: unknown; revision: unknown; payload: unknown }[]
  let mergedCapped = false
  if (pairs && pairs <= KPI_MAX_SPLIT) {
    const streams = await Promise.all(
      instanceIds.flatMap(instanceId => BUSINESS_EVENTS.map(eventType => readOne(instanceId, eventType)))
    )
    const merged = mergeStreams(streams, MAX_EVENTS, (a, b) => {
      const at = String(a.eventTime ?? '').localeCompare(String(b.eventTime ?? ''))
      if (at !== 0) return at
      return Number(a.revision ?? 0) - Number(b.revision ?? 0)
    })
    if (merged.underfilled) {
      /* 있을 수 없는 일이다 — 갈래가 상한보다 많이 받아 왔다. 조용히 쓰지 않고 말한다. */
      twinWarn('[twin-kpi] a per-type stream returned more than the cap — the merge cannot be trusted; falling back')
      raw = []
    } else {
      raw = merged.items
      mergedCapped = merged.capped
    }
  } else {
    raw = await getReadRepository(TwinEvent)
      .createQueryBuilder('e')
      .select(['e.eventType AS "eventType"', 'e.eventTime AS "eventTime"', 'e.revision AS "revision"', 'e.payload AS "payload"'])
      .where('e.domain = :domainId', { domainId })
      .andWhere('e.instanceId IN (:...instanceIds)', { instanceIds })
      .andWhere('e.eventType IN (:...types)', { types: BUSINESS_EVENTS })
      .andWhere('e.eventTime BETWEEN :from AND :to', { from, to })
      .orderBy('e.eventTime', 'ASC')
      .addOrderBy('e.revision', 'ASC')
      .limit(MAX_EVENTS)
      .getRawMany<{ eventType: string; eventTime: unknown; revision: unknown; payload: unknown }>()
    mergedCapped = raw.length >= MAX_EVENTS
  }

  /* 상한에 정확히 닿으면 **더 있었을 수 있다** — 그 사실을 돌려준다. 알리지 않고 잘라내면 숫자가 알리지 않고
   * 작아지고, 사용자는 그것을 성과 하락으로 읽는다(가장 나쁜 종류의 거짓이다). */
  return {
    /* 폴드는 ISO 문자열 축으로 수행된다(kpi-fold) — 경계에서 한 번만 옮긴다. */
    events: raw.map(r => ({ eventType: r.eventType, eventTime: isoOf(r.eventTime), payload: jsonOf(r.payload) })),
    capped: mergedCapped
  }
}

/**
 * 원시 결과의 시각을 ISO 로 — **드라이버마다 모양이 다르다.**
 *
 * 엔티티 경로에서는 TypeORM 이 `Date` 로 바꿔 주지만, 원시 결과는 드라이버가 준 그대로다(sqlite 는
 * `'2026-08-07 11:00:00.000'`, pg 는 `Date`). 경계에서 한 번 정규화한다 — 폴드는 ISO 문자열만 안다.
 */
function isoOf(v: unknown): string | undefined {
  if (v == null) return undefined
  if (v instanceof Date) return v.toISOString()
  const s = String(v)
  /* sqlite 의 공백 구분 형식을 ISO 로 — 그대로 `Date` 에 주면 시간대 해석이 갈린다. */
  const d = new Date(s.includes('T') ? s : s.replace(' ', 'T') + (s.endsWith('Z') ? '' : 'Z'))
  return Number.isNaN(d.getTime()) ? undefined : d.toISOString()
}

/**
 * 원시 결과의 `payload` 를 객체로 — **엔티티 경로의 `simple-json` 변환을 여기서 한다.**
 *
 * 깨진 JSON 은 **버리지 않고 비운다**: 그 사건이 있었다는 사실은 남고, 안을 읽을 수 없다는 것만
 * 폴드에 전해진다(폴드는 없는 필드를 건너뛴다). 던지면 창 하나가 통째로 사라진다.
 */
function jsonOf(v: unknown): any {
  if (v == null) return undefined
  if (typeof v !== 'string') return v
  try {
    return JSON.parse(v)
  } catch {
    return undefined
  }
}

/**
 * 집계 대상 트윈 목록.
 *
 * 공간이 주어지면 그 공간의 **운영 목적** 트윈을 모은다 — 벤치 사본은 사용자가 보는 현장이 아니다.
 * 트윈이 지정되면 그것 하나. 둘 다 없으면 무엇을 재야 할지 알 수 없으므로 명시 실패한다.
 *
 * ── **「지금 도는가」로 걸러내지 않는다** (2026-08-28 실측) ──────────────────
 * 여기에 `running` 조건이 있었다. 그래서 멈춘 트윈은 현장으로 물었을 때 대상에서 빠졌고, 화면은
 * 「기록이 없습니다」라고 말했다 — **저널에는 있는데.**
 *
 *   twinEventList(instanceId: "ems-plant-1")      1,792건
 *   twinEventList(spaceId:    "ems-plant-1-site")     0건   ← 같은 트윈, 같은 도메인
 *
 * 규모로 보면 이것은 예외가 아니라 기본값이었다: 등록된 트윈 30개 중 **도는 것이 2개, 멈춘 것이 28개**다.
 * 즉 현장으로 묻는 저널 조회가 대부분의 현장에서 거짓 「없다」를 냈다.
 *
 * 지운 이유는 두 가지다.
 *
 *  ① **저널은 기록이고, 프로세스가 지금 도는지는 다른 축이다.** 창 안에 있는 사실은 사실이다.
 *     아래 줄이 트윈을 지정한 경우에 대해 이미 그 원칙을 적어 두었다(「정지·비등록이어도 저널은 남아
 *     있을 수 있다」). 같은 원칙이 현장 경로에만 적용되지 않고 있었다.
 *  ② 옛 주석이 대는 이유(「벤치 사본은 사용자가 보는 현장이 아니다」)는 **벤치 제외만** 설명한다.
 *     `running` 에는 근거가 적혀 있지 않았다.
 *
 * **바뀌는 것**: 현장 KPI 가 이제 멈춘 트윈의 지난 기록도 센다. 창 안의 사실을 세는 것이 그 지표의
 * 정의이므로 그것이 맞다. 「지금 도는 것만」이 필요한 화면이 생기면 그때 **이름 있는 축**으로 받는다 —
 * 여기에 숨겨 두면 두 물음이 한 칸을 쓴다.
 */
async function resolveTargets(input: TwinKpiInput): Promise<{ instanceIds: string[]; restartPolicies: string[] }> {
  if (!input.instanceId && !input.spaceId) throw new Error('either instanceId or spaceId is required')
  const list = await TwinEngine.list(input.domainId)
  const rows = input.instanceId
    ? list.filter((x: any) => x?.instanceId === input.instanceId)
    : list.filter((x: any) => x?.spaceId === input.spaceId && x.purpose !== 'bench' && x.instanceId)
  /* 목록에서 못 찾아도(정지·비등록) 지정된 트윈은 그대로 조회한다 — 저널은 남아 있을 수 있다. */
  const instanceIds = rows.length ? rows.map((x: any) => x.instanceId as string) : input.instanceId ? [input.instanceId] : []
  /* 컬럼이 NOT NULL 이라 'unknown' 으로 메울 일이 없다 — 비면 계약이 깨진 것이지 모르는 상태가 아니다. */
  const restartPolicies = [...new Set(rows.map((x: any) => String(x.restartPolicy)))]
  return { instanceIds, restartPolicies }
}

/**
 * 대상 해소 공개 창구 — **저널 목록 질의도 KPI 와 같은 규칙을 써야 한다.**
 *
 * 화면에서 "이 공간" 은 하나의 뜻이어야 한다. 성과 화면과 원장 화면이 각자 co-located 트윈을
 * 추리면 같은 공간을 보면서 다른 대상 집합을 세게 되고, 두 숫자가 어긋나는 이유를 아무도 설명하지 못한다.
 */
export async function resolveTwinTargets(domainId: string, instanceId?: string, spaceId?: string): Promise<string[]> {
  const { instanceIds } = await resolveTargets({ domainId, instanceId, spaceId } as TwinKpiInput)
  return instanceIds
}

/**
 * 시간창 업무 KPI. 창 해소 → 저널 조회 → 순수 폴드.
 *
 * 테넌트 조건은 호출부가 준 `domainId` 로만 좁힌다(클라이언트 값을 그대로 쓰지 않는 것은 호출부 책임).
 */
export async function computeTwinKpi(input: TwinKpiInput): Promise<TwinKpiOutput> {
  const minutes = Math.max(1, Math.min(24 * 60, Number(input.windowMinutes) || 60))
  const lookbackMs = Math.max(0, Math.min(24 * 60, input.lookbackMinutes ?? 60)) * 60_000

  const { instanceIds, restartPolicies } = await resolveTargets(input)
  const scope = { spaceId: input.spaceId, instanceIds, restartPolicies }

  /* 끝 시각 — 지정이 없으면 대상들의 마지막 기록. 그것도 없으면 측정할 것이 없다. */
  let basis: 'given' | 'latest-event' = 'given'
  let toMs = input.toTime ? Date.parse(input.toTime) : NaN
  if (!Number.isFinite(toMs)) {
    const latest = await latestEventTime(input.domainId, instanceIds)
    if (!latest) {
      const now = Date.now()
      return {
        window: {
          fromTime: new Date(now - minutes * 60_000).toISOString(),
          toTime: new Date(now).toISOString(),
          minutes,
          basis: 'latest-event'
        },
        measured: false,
        scope,
        note:
          instanceIds.length === 0
            ? 'no running twin in this space — nothing to measure.'
            : 'no operational events in the journal yet — there is nothing to measure (this is NOT "zero throughput").'
      }
    }
    /*
     * ── **미래를 성과 구간으로 삼지 않는다** (2026-08-30) ──────────────────────
     *
     * 마지막 기록이 미래일 수 있다. 한 원본이 **끝나지 않은 이번 달**의 요금 정보를 청구서로 보냈고,
     * 청구서의 사건 시각은 기간의 끝이라 그 사실이 **내일** 일어난 것으로 적혔다. 그러자 구간이
     * `내일 00:00 → 모레 00:00` 이 되어 그 현장의 화면에서 **어제 발전량이 사라졌다.** 오류는 나지
     * 않았고 화면만 비어 있었다.
     *
     * 커널이 이제 그런 사실을 받지 않는다. 그래도 여기서 막는다 — **이미 적힌 것은 사라지지 않고**,
     * 저널은 지우지 않기 때문이다. 그리고 성과는 지나간 일이다. 아직 오지 않은 구간의 성과라는 것은
     * 없다.
     */
    toMs = Math.min(Date.parse(latest), Date.now())
    basis = 'latest-event'
  }

  const fromMs = input.fromTime ? Date.parse(input.fromTime) : toMs - minutes * 60_000
  if (!Number.isFinite(fromMs) || !Number.isFinite(toMs) || fromMs >= toMs) {
    throw new Error('invalid window: fromTime must be a valid time before toTime')
  }

  const { events, capped } = await fetchEvents(input.domainId, instanceIds, fromMs, toMs, lookbackMs)

  /* 구역·교대 축은 **이벤트 밖의 지식**을 필요로 한다 — 다른 축은 payload 에서 바로 나온다. */
  const areaMap =
    input.groupBy === 'area' ? await locationAreaMap(input.domainId, instanceIds, input.spaceId) : { locationArea: {}, labels: {} }
  const shiftCtx = input.groupBy === 'shift' ? await shiftContext(input.domainId, instanceIds) : {}
  /*
   * 전기 계통 소속은 **축과 무관하게** 읽는다 — 분기별 몫은 관점 축이 아니라 에너지 사실의 일부라
   * (같은 창의 총량을 나눈 것) 어느 축을 보고 있든 함께 나가야 한다.
   */
  const feederMap = await resourceFeederMap(input.domainId, instanceIds)
  /* 요금 단가도 축과 무관하게 읽는다 — 금액은 같은 창의 전력량·최대수요에서 나오는 사실이다. */
  const tariffFound = await tariffOf(input.domainId, instanceIds)
  /*
   * ── 요금적용전력은 **청구서가 먼저다** (2026-08-29) ─────────────────────────
   *
   * 그 값은 최근 몇 달의 최고 수요로 **매달 다시 정해진다.** 선언에 박아 두면 매달 사람이 갱신해야
   * 하고, 안 하면 낡은 값으로 기본요금이 조용히 틀린다.
   *
   * 원본이 청구서에 그 값을 실어 매일 보낸다. 창 밖의 청구서는 저널 조회에 걸리지 않으므로
   * **커널 상태에 이어져 있는 마지막 청구서**를 함께 넘긴다(§`EnergyState.lastBill`).
   */
  const lastBill = latestBillOf(input.domainId, instanceIds)
  const tariffBasis = latestTariffBasisOf(input.domainId, instanceIds)
  /* 이용률의 분모 — 커널이 모델의 선언에서 고른다(설비 합 → 자리 합계, 교류 → 직류). */
  const generationRated = generationRatedOf(input.domainId, instanceIds)
  const generationPrice = latestGenerationPriceOf(input.domainId, instanceIds)
  const kpi = foldKpi(events, { fromMs, toMs }, {
    groupBy: input.groupBy,
    locationArea: areaMap.locationArea,
    resourceFeeder: feederMap.resourceFeeder,
    meterFeeder: feederMap.meterFeeder,
    ...(tariffFound.tariff ? { tariff: tariffFound.tariff } : {}),
    ...(lastBill ? { lastBill } : {}),
    ...(tariffBasis ? { tariffBasis } : {}),
    ...(generationRated ? { generationRated } : {}),
    ...(generationPrice ? { generationPrice } : {}),
    ...shiftCtx,
    groupLimit: input.groupLimit
  })
  /*
   * **완료로 세지 않은 오더 상태 낱말을 말한다** (2026-08-23).
   *
   * 폴드는 순수하다(로그를 모른다) — 그래서 부르는 쪽이 말한다. 왜 말해야 하나: 이 폴드는 오더의
   * 완료를 **동의어 목록**으로 판정하고, 그것은 원본의 낱말에서 뜻을 추론하는 것이다. 커넥터가
   * 자기 표로 옮기지 않으면 그 낱말이 목록에 없어 **완료가 알리지 않고 0 으로 세어진다.**
   *
   * 실제로 그랬다: 첫 실 원본의 `FINISHED` 가 목록에 없어 완료 오더가 전부 0 이었고, 사람이 화면을
   * 보고 의심해서야 드러났다(그 커넥터에는 상태 표가 정의돼 **있으면서** 오더 경로에서 쓰이지 않았다).
   * 그때 목록에 낱말을 더한 것은 그 사실을 가린 땜빵이었다.
   *
   * 그러니 이 로그는 「목록을 늘려라」가 아니라 **「그 커넥터가 옮기지 않았다」**를 뜻한다. 화면을 보지
   * 않는 사람도 알아야 하므로 로그에 남긴다 — 들어온 것이 없는 0 이 이 시스템에서 가장 비싼 침묵이다.
   *
   * 완료가 하나도 없을 때만 말한다: `picking` 처럼 완료가 아닌 낱말이 섞여 있는 것은 정상이고,
   * 그것까지 매번 경고하면 진짜 신호가 묻힌다.
   */
  const notCounted = Object.entries(kpi.orderStatusNotCounted ?? {})
  if (notCounted.length && kpi.throughput.orders === 0) {
    twinWarn(
      `[twin-kpi] no order counted as completed in this window, but these order status words were seen: ` +
        `${notCounted.map(([w, n]) => `${w}×${n}`).join(' · ')} — the connector may not be translating them ` +
        `to the kernel's vocabulary (see its own status table).`
    )
  }
  /*
   * 목표를 **폴드 뒤, 그룹 조립 앞**에 읽는다 — 전체 판정과 축별 판정이 같은 목록에서 나와야
   * 두 곳의 임계가 갈라지지 않는다. 폴드 자체는 목표를 모른다(사실 층과 판정 층의 분리).
   */
  const scoped = await loadTargets(input.domainId, input.spaceId, instanceIds, input.groupBy)
  const verdicts = judgeTargets(kpi, scoped.overall)
  const groupTargets = scoped.byAxisKey

  const window = {
    fromTime: new Date(fromMs).toISOString(),
    toTime: new Date(toMs).toISOString(),
    minutes: Math.round((toMs - fromMs) / 60_000),
    basis
  }
  /**
   * 여러 구간 분포를 만든다 — **현재 창에 기록이 없어도 낸다.**
   *
   * 그때가 오히려 가장 알고 싶은 순간이다("지금 아무 기록이 없는데 평소엔 60건이었다" = 멈춘 신호).
   * 그래서 조기 반환보다 앞에 두고 두 경로가 함께 쓴다.
   */
  const computeBaseline = async (): Promise<TwinKpiOutput['baseline']> => {
    if (!input.baseline || !(input.baseline.periods > 0)) return undefined
    const stride = input.baseline.stride ?? 'day'
    const windows = baselineWindows(fromMs, toMs, input.baseline.periods, stride)
    let baseCapped = false
    const samples: { fromMs: number; toMs: number; values: Partial<Record<TwinMetricId, number | undefined>> }[] = []

    for (const w of windows) {
      const { events: pastEvents, capped: pastCapped } = await fetchEvents(input.domainId, instanceIds, w.fromMs, w.toMs, lookbackMs)
      baseCapped = baseCapped || pastCapped
      /* 축 없이, **같은 교대 정의**로 처리한다(정의가 다르면 같은 시간대가 아니게 된다). */
      const past = foldKpi(pastEvents, w, { locationArea: areaMap.locationArea, ...shiftCtx })
      const values: Partial<Record<TwinMetricId, number | undefined>> = {}
      for (const metric of Object.keys(TWIN_METRIC) as TwinMetricId[]) values[metric] = TWIN_METRIC[metric].of(past)
      samples.push({ fromMs: w.fromMs, toMs: w.toMs, values })
    }

    const current: Partial<Record<TwinMetricId, number | undefined>> = {}
    for (const metric of Object.keys(TWIN_METRIC) as TwinMetricId[]) current[metric] = TWIN_METRIC[metric].of(kpi)

    return {
      stride,
      periods: windows.length,
      metrics: summarizeBaseline(TWIN_METRIC, current, samples),
      ...(baseCapped ? { capped: true } : {})
    }
  }


  if (kpi.eventsInWindow === 0) {
    const baseline = await computeBaseline()
    return {
      window,
      measured: false,
      scope,
      note: 'no operational events in this window — nothing to measure (this is NOT "zero throughput").',
      ...(baseline ? { baseline } : {})
    }
  }

  /* 구간별 값 — 쪼개기 규칙은 순수 모듈이 소유한다(테스트 가능). 여기서는 시각을 ISO 로 옮기기만. */
  const rawBuckets = foldKpiBuckets(events, { fromMs, toMs }, Number(input.buckets) || 0)
  const buckets = rawBuckets.length
    ? rawBuckets.map(b => ({
        fromTime: new Date(b.fromMs).toISOString(),
        toTime: new Date(b.toMs).toISOString(),
        tasks: b.tasks,
        orders: b.orders,
        workP50Ms: b.workP50Ms
      }))
    : undefined

  /* 비교 구간 — **같은 길이·같은 규칙**으로 한 번 더 계산한다. 기록이 없으면 delta 를 만들지 않는다
   * (없는 것을 0 으로 그리면 "변화 없음" 이라는 거짓이 된다). */
  let comparison: TwinKpiOutput['comparison']
  /* 관점 × 비교 교차 — 축별 delta 와 **사라진 축**(전에 있었고 지금 없는 것). */
  let crossed: ReturnType<typeof crossGroups> | undefined
  /* 비교 구간 조회도 상한에 걸릴 수 있다 — 그 경우 delta 가 실제보다 작아 보인다. */
  let prevCappedFlag = false
  if (input.compareTo === 'previous' || input.compareTo === 'yesterday') {
    const w = shiftWindow(fromMs, toMs, input.compareTo)
    const { events: prevEvents, capped: prevCapped } = await fetchEvents(
      input.domainId,
      instanceIds,
      w.fromMs,
      w.toMs,
      lookbackMs
    )
    /* 축이 요청됐으면 비교 구간도 **같은 축**으로 처리한다 — 교차("구역별로 어제 대비")를 내기 위해.
     * 상한은 축이 서로 어긋나지 않게 넉넉히 둔다(현재 축이 상한 안에 있는데 비교 축에서 밀려나면
     * 있는데 없다고 말하게 된다). */
    const prev = foldKpi(prevEvents, w, {
      groupBy: input.groupBy,
      locationArea: areaMap.locationArea,
      ...shiftCtx, // 비교 구간도 **같은 교대 정의**로 처리한다(정의가 다르면 축이 어긋난다)
      groupLimit: input.groupBy ? 100 : undefined
    })
    prevCappedFlag = prevCapped
    if (input.groupBy && kpi.groups && prev.groups && prev.eventsInWindow > 0) {
      crossed = crossGroups(kpi.groups.items, prev.groups.items)
    }
    const base = {
      basis: input.compareTo,
      window: { fromTime: new Date(w.fromMs).toISOString(), toTime: new Date(w.toMs).toISOString() }
    }
    comparison =
      prev.eventsInWindow === 0
        ? { ...base, measured: false }
        : {
            ...base,
            measured: true,
            throughput: prev.throughput,
            workTime: prev.workTime,
            leadTime: prev.leadTime,
            waitTime: prev.waitTime,
            delta: {
              tasks: kpi.throughput.tasks - prev.throughput.tasks,
              orders: kpi.throughput.orders - prev.throughput.orders,
              workP50Ms: deltaP50(kpi.workTime, prev.workTime),
              leadP50Ms: deltaP50(kpi.leadTime, prev.leadTime),
              waitP50Ms: deltaP50(kpi.waitTime, prev.waitTime)
            }
          }
  }

  /*
   * 여러 구간 동시 비교 — **같은 시간대를 며칠 뒤로** 옮겨 분포를 만든다(성과 맵 ③).
   *
   * 창마다 조회가 한 번 더 붙지만 각 창은 현재 창과 같은 길이라(보통 1~24h) 값싸다. 축(groupBy)은
   * 넘기지 않는다 — 분포는 **전체 지표**의 질문이고, 축별 분포까지 내면 응답이 축 수만큼 부푼다
   * (필요해지면 그때 축을 지정해 다시 묻는 것이 정직하다).
   *
   * 지표 어휘는 `TWIN_METRIC` 이 계속 소유하고, 여기서는 **값만 꺼내** 분포 모듈에 넘긴다.
   */
  /* 관점 값은 시각 변환이 없어 그대로 통과한다. 붙이는 것은 ① 사람이 읽는 이름과 ② 축이 통째로
   * 비었을 때의 이유다 — "구역별 0건" 과 "노드에 구역이 안 붙어 있다" 는 완전히 다른 상황이다. */
  const groups = kpi.groups
    ? {
        by: kpi.groups.by,
        items: (crossed?.items ?? kpi.groups.items).map(g => {
          /* 축의 이 값에 목표가 걸려 있으면 판정을 함께 낸다 — "이 구역만 목표가 다르다" 를
             표에서 바로 볼 수 있게. 목표가 없는 축값에는 필드를 만들지 않는다. */
          const gt = judgeGroupTargets(g as any, groupTargets[g.key])
          return {
            ...g,
            ...(areaMap.labels[g.key] ? { label: areaMap.labels[g.key] } : {}),
            ...(gt.length ? { targets: gt } : {})
          }
        }),
        ...(crossed?.disappeared.length
          ? {
              disappeared: crossed.disappeared.map(d => ({
                ...d,
                ...(areaMap.labels[d.key] ? { label: areaMap.labels[d.key] } : {})
              }))
            }
          : {}),
        ...(kpi.groups.truncated
          ? { truncated: kpi.groups.truncated, truncatedTasks: kpi.groups.truncatedTasks ?? 0 }
          : {}),
        ...(input.groupBy === 'area' &&
        Object.keys(areaMap.locationArea).length === 0 &&
        kpi.groups.items.some(g => g.key === 'unknown')
          ? { note: 'locations carry no area (parentId) in this twin — cannot break down by area. this is NOT "no work in areas".' }
          : {})
      }
    : undefined

  /*
   * 목표 대비 판정 — **사실 계산 뒤에 얹는다.**
   *
   * 폴드는 목표를 모른다(사실만 낸다). 목표를 아는 순간 "세는 규칙" 과 "잘했나 판정하는 규칙" 이
   * 한 함수에 섞이고, 목표가 바뀔 때마다 사실 계산을 건드리게 된다. 그래서 조회 층에서 목표를
   * 실어 오고, 판정은 순수 함수(`judgeTargets`)가 한다.
   *
   * **목표가 없으면 판정하지 않는다** — 필드 자체를 만들지 않는다. 전부 "목표 없음" 으로 채우면
   * 화면이 회색 노이즈가 되고, 목표가 있는 지표가 묻힌다.
   */
  /* 분포는 **한 번만** 계산한다 — 창마다 조회가 붙으므로 두 번 부르면 조회가 두 배가 된다. */
  const baseline = await computeBaseline()

  /* 폴드가 돌려준 raw 창(ms)은 버린다 — 위의 ISO 창이 사람이 읽을 정본이다. */
  const { window: _rawWindow, groups: _rawGroups, ...rest } = kpi
  /*
   * 자리에 **부르는 이름**을 붙인다 — `feeder-2` 는 id 이고 사람은 「가공 라인」으로 부른다.
   * 이름을 모르는 자리는 id 그대로 둔다(지어내지 않는다).
   */
  /* 단가가 갈려 있으면 밝힌다 — 우리가 고른 하나로 낸 금액이라는 사실을 화면이 말할 수 있게. */
  if (rest.energy?.cost && tariffFound.conflict) rest.energy = { ...rest.energy, tariffConflict: true } as any
  if (rest.energy?.byFeeder?.length) {
    rest.energy = {
      ...rest.energy,
      byFeeder: rest.energy.byFeeder.map(row => ({ ...row, ...(feederMap.labels[row.key] ? { label: feederMap.labels[row.key] } : {}) }))
    }
  }
  /*
   * 계통도 — **선언된 관계와 계량이 덮는 범위**를 함께 낸다.
   *
   * 값은 여기서 다시 계산하지 않는다(폴드가 낸 분기별 몫을 그대로 노드에 얹는다). 계통은 모델이 아는
   * 사실이므로 저널만 보는 폴드가 아니라 여기서 조립한다 — 화면이 모델을 따로 또 조회하지 않아도 되고,
   * AI 도 같은 값을 읽는다(규칙 한 벌).
   */
  if (rest.energy) {
    const topo = await energyTopologyOf(input.domainId, instanceIds, rest.energy)
    if (topo) rest.energy = { ...rest.energy, topology: topo } as any
  }
  return {
    window,
    measured: true,
    scope,
    ...(capped || prevCappedFlag ? { eventsCapped: true } : {}),
    ...rest,
    ...(buckets ? { buckets } : {}),
    ...(groups ? { groups } : {}),
    ...(verdicts.length ? { targets: verdicts } : {}),
    ...(comparison ? { comparison } : {}),
    ...(baseline ? { baseline } : {})
  }
}