import { Arg, Ctx, Int, Query, Resolver } from 'type-graphql'
import { Between } from 'typeorm'

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

import { TwinEngine } from '../../engine/index.js'
import { TwinIngestWindow } from './twin-ingest-window.js'

/*
 * 인제스트 창 **이력** — 재기동 뒤에도 추이를 이어 본다.
 *
 * ── 왜 따로 있나 (2026-08-20) ───────────────────────────────────────────────
 * 창을 표로 남기기 시작했는데 **읽는 코드가 없었다.** 21행이 쌓여 있고 아무도 보지 않았다 — 이
 * 프로젝트가 결함으로 보는 상태다(「필드만 만들고 아무도 읽지 않는다」). 표를 만든 이유가 「메모리
 * 장부는 재기동에 사라지고 6시간까지인데 추이는 그보다 오래 봐야 한다」였으니, 목적이 미완이었다.
 *
 * ── 왜 `twinIngestHealth` 에 합치지 않나 ────────────────────────────────────
 * 그 조회는 **화면이 자주 묻는다**(상세를 열 때마다, 목록은 응답에 요약이 함께 온다). 거기에 DB 주사를
 * 얹으면 이 세션에서 없앤 것과 같은 종류의 비용이 돌아온다 — 자주 묻는 조회가 자라는 표를 읽는 것.
 * 이력은 **다른 질문**이고 덜 자주 묻는다: 「더 뒤까지 보여 달라」.
 *
 * ── 이어 붙이기는 서버가 한다 ───────────────────────────────────────────────
 * 표와 메모리는 겹친다(프로세스가 살아 있는 동안 닫힌 창은 둘 다에 있다). 겹침을 화면이 정리하게 두면
 * 화면마다 규칙이 갈리고, 같은 구간이 두 번 그려지거나 사라진다. 그래서 여기서 한 벌로 이어 붙이고
 * **어디서부터 메모리인지**(`liveFrom`)를 함께 낸다 — 화면이 근거를 말할 수 있게.
 */

/**
 * 한 번에 돌려주는 창 수 상한 — 144창 × 10분 = 24시간.
 *
 * 상한이 필요한 이유는 이 표가 **자란다**는 것이다(트윈 1,000개에 하루 14.4만 행). 상한에서 밀려난
 * 수는 **세어서 낸다** — 알리지 않고 자르면 화면이 「그 앞에는 아무 일도 없었다」로 읽는다.
 */
export const HISTORY_WINDOW_LIMIT = 144

/** 물을 수 있는 기간의 상한(시간). 넘겨도 여기서 자르고, 자른 사실을 낸다. */
export const HISTORY_HOURS_MAX = 168

@Resolver()
export class TwinIngestWindowQuery {
  @Query(returns => ScalarObject, {
    nullable: true,
    description:
      'Ingest windows for one twin over a past span, stitched from two sources so the trend survives a restart: closed windows persisted in the table (older) and the in-memory ledger (recent, roughly the last 6 hours). Overlap is resolved here rather than on screen — a window present in both is reported once, from memory, and `liveFrom` says where memory takes over so a screen can state its basis. Each window carries offered / accepted / rejected / undelivered plus the journal rows written and the write round-trips behind them; `acceptedRatio` is null when nothing was offered, because an empty window is not a perfect one. Deliberately separate from twinIngestHealth: that query is polled whenever a screen opens, and making it scan a growing table would reintroduce the cost this codebase just removed elsewhere. Counts that could not be returned are reported rather than dropped — `storedDropped` when the row cap was hit and `hoursClamped` when the requested span was longer than allowed. Null if the instance is not in the current domain.'
  })
  async twinIngestHistory(
    @Arg('instanceId') instanceId: string,
    @Arg('hours', type => Int, { nullable: true, description: 'How far back to read, in hours. Defaults to 24 and is clamped to 168.' })
    hours: number | undefined,
    @Ctx() context: ResolverContext
  ): Promise<any> {
    const domainId = context.state.domain.id
    /* 테넌트 격리 — 멈춘 트윈의 이력도 볼 수 있어야 한다(그때가 가장 보고 싶은 때다). */
    if (!TwinEngine.owns(domainId, instanceId) && !(await TwinEngine.ownsRegistered(domainId, instanceId))) return null

    const asked = Number.isFinite(Number(hours)) && Number(hours) > 0 ? Math.floor(Number(hours)) : 24
    const span = Math.min(asked, HISTORY_HOURS_MAX)
    const now = Date.now()
    const fromMs = now - span * 3_600_000

    /*
     * 표에서 그 구간의 닫힌 창을 읽는다. **상한을 둔다** — 자라는 표이고, 이 조회가 화면에서 열린다.
     * 최신 쪽이 중요하므로 내려서 정렬해 가져오고, 낼 때 다시 올린다.
     */
    const repo = getRepository(TwinIngestWindow)
    const where = {
      domain: { id: domainId } as any,
      instanceId,
      closedAt: Between(new Date(fromMs), new Date(now))
    }
    const rows = await repo.find({ where, order: { closedAt: 'DESC' }, take: HISTORY_WINDOW_LIMIT })

    /*
     * 상한에 닿았으면 **몇 개가 밀렸는지 센다.** 「더 있음」만 말하면 화면이 그 앞을 짐작하게 된다.
     * 상한에 닿지 않았으면 셀 필요가 없다(왕복을 아낀다).
     */
    const storedDropped = rows.length === HISTORY_WINDOW_LIMIT ? Math.max(0, (await repo.count({ where })) - rows.length) : 0

    const stored = rows.reverse().map(r => ({
      from: r.openedAt?.toISOString?.() ?? new Date(r.openedAt as any).toISOString(),
      to: r.closedAt?.toISOString?.() ?? new Date(r.closedAt as any).toISOString(),
      offered: r.offered,
      /* 통과는 **트윈에 닿은** 것만. 표의 값을 그대로 쓴다 — 읽는 쪽에서 다시 나누면 반올림이 갈린다. */
      accepted: r.offered - r.rejected - (r.undelivered ?? 0),
      rejected: r.rejected,
      undelivered: r.undelivered ?? 0,
      acceptedRatio: typeof r.acceptedRatio === 'number' ? r.acceptedRatio : null,
      rows: r.journaledRows ?? 0,
      flushes: r.journalFlushes ?? 0,
      source: 'stored' as const
    }))

    /* 메모리 장부의 닫힌 창 — 이쪽이 최신이고, 겹치는 구간은 이쪽을 쓴다(같은 사실의 원본이다). */
    const live = (TwinEngine.ingestHealthOf(domainId, instanceId).trend ?? []).map((w: any) => ({
      ...w,
      source: 'live' as const
    }))

    /*
     * 겹침 정리 — 창의 **여는 시각**이 키다. 같은 창이 표와 메모리에 둘 다 있으면 하나만 낸다.
     * 시각 문자열을 그대로 키로 쓰지 않는다: 드라이버가 밀리초를 자르거나 형식을 바꿀 수 있어
     * 같은 창이 다른 키가 된다. epoch 로 견준다.
     */
    const liveStarts = new Set(live.map((w: any) => Date.parse(w.from)))
    const windows = [...stored.filter(w => !liveStarts.has(Date.parse(w.from))), ...live]

    return {
      windows,
      hours: span,
      /* 물은 것보다 짧게 잘렸으면 밝힌다. */
      ...(span < asked ? { hoursClamped: asked } : {}),
      from: new Date(fromMs).toISOString(),
      to: new Date(now).toISOString(),
      /** 표에서 읽은 창 수 · 상한에서 밀려난 수. */
      storedCount: stored.length,
      storedDropped,
      /**
       * 어디서부터 메모리인가 — 화면이 근거를 말할 수 있게. 메모리에 닫힌 창이 없으면 `null` 이다
       * (재기동 직후가 그렇다: 그때 보이는 것은 전부 표에서 온 것이다).
       */
      liveFrom: live.length ? live[0].from : null,
      liveCount: live.length
    }
  }
}