/*
 * **빠진 구간을 다시 받는 일을 실제로 부른다** — `fillSince` 에 연결과 어댑터를 물려 주는 자리.
 *
 * ── 왜 이 파일이 없었나 ─────────────────────────────────────────────────────
 * `fillSince` 는 2026-08-31 에 만들어졌고 **부르는 곳이 0곳이었습니다.** 2026-09-05 에 저는 그것을
 * 못 보고 같은 일을 하는 것을 하나 더 만들었습니다(`backfillScope`). 두 벌이 되었고 둘 다 아무도
 * 부르지 않았습니다.
 *
 * 그래서 이런 상태였습니다.
 *
 * ```
 * 번호가 비면 어디서부터 다시 받을지 안다   ✔  takeInSequence 가 판정한다
 * 그 빈 것을 메운다                        ✖  부를 곳이 없다
 * ```
 *
 * 실제로 plant 아웃박스의 1~20 을 잃고 되찾을 방법이 없는 상태를 만났습니다.
 *
 * ── 같은 검사를 지난다 ──────────────────────────────────────────────────────
 * `fillSince` 가 `takeInSequence` 를 지나고, 유입은 `liveIngest` 를 지납니다 — 밀어 받는 길과 같은
 * 함수입니다. 여기서 따로 만들면 두 길이 다른 규칙으로 받고, 한쪽만 수정되는 날이 옵니다.
 *
 * ── 커서 열쇠는 프레임워크가 만든다 ─────────────────────────────────────────
 * 열쇠는 `instanceId::scope` 이고 그것을 아는 것은 프레임워크입니다. 어댑터가 만들면 두 곳이 같은
 * 규칙을 갖게 되고, 어긋나는 날 **다시 받기가 자기 커서를 0부터 새로 셉니다.**
 */
import { getRepository } from '@things-factory/shell'

import { TwinInstance } from '../twin-instance/twin-instance.js'
import { TwinReference } from './twin-reference.js'
import { getAdapter, type SeqReport, type SiteDescriptor } from './reference-adapter.js'
import { firstSeqOf, pushCursorKey, seqOf, withSeq } from './hook-contract.js'
import { fillSince, type FillOutcome } from './reference-fill.js'
import { liveIngest } from './live-ingest.js'
import { recordReferenceReconciliation } from './reference-reconciliation-writer.js'
import { journalIdentities, spanOf } from './fill-dedupe.js'
import { TwinEngine } from '../../engine/index.js'
import { twinLog, twinWarn } from '../../engine/log.js'

/** 한 연결의 한 단위를 다시 받은 결과. */
export interface FillRunResult extends Partial<FillOutcome> {
  instanceId: string
  scope: string
  /**
   * 못 한 이유 — 성공이면 없습니다.
   *
   * `not-supported` 는 결함이 아닙니다. 주기로 물어보는 원본(chef · ppms)은 다음 주기가 곧 다시
   * 받는 것입니다.
   */
  skipped?: 'unknown-twin' | 'no-source' | 'no-adapter' | 'not-supported'
}

/**
 * 그 연결의 한 단위를 다시 받는다.
 *
 * **커서를 되돌리지 않습니다.** 받은 것이 번호 검사를 지나야 커서가 나아가고, 그 판정은 밀어 받는
 * 길과 같은 함수가 합니다.
 *
 * @param from 여기서부터 다시 받는다 — 주면 커서를 무시합니다. 잃은 구간을 되찾을 때만 씁니다.
 */
export async function runFill(args: {
  domainId: string
  instanceId: string
  scope: string
  from?: number
  maxPages?: number
}): Promise<FillRunResult> {
  const { domainId, instanceId, scope, from, maxPages } = args
  const base: FillRunResult = { instanceId, scope }

  const inst = await getRepository(TwinInstance).findOne({ where: { domain: { id: domainId }, instanceId } })
  if (!inst) return { ...base, skipped: 'unknown-twin' }

  const repo = getRepository(TwinReference)
  const ref = await repo.findOne({ where: { domain: { id: domainId }, source: instanceId } })
  if (!ref) return { ...base, skipped: 'no-source' }

  const adapter = getAdapter(ref.adapterType)
  if (!adapter) return { ...base, skipped: 'no-adapter' }
  /* 다시 받기를 안 내는 원본이 있습니다 — 주기로 물어보는 쪽은 다음 주기가 곧 그것입니다. */
  if (typeof adapter.fetchSince !== 'function') return { ...base, skipped: 'not-supported' }

  const siteId = (inst.origin as any)?.siteId
  const site: SiteDescriptor = siteId ? { siteId } : { siteId: instanceId }
  const cursorKey = pushCursorKey(instanceId, scope)

  /*
   * `from` 을 주면 커서를 그 값으로 갈아 끼웁니다 — **저장된 커서를 건드리지 않습니다.** `fillSince`
   * 가 나아간 만큼만 적으므로, 잃은 구간을 되찾은 뒤 커서는 원래 자리보다 앞서지 않습니다.
   */
  const cursor = from === undefined ? ((ref as any).liveCursor ?? {}) : withSeq((ref as any).liveCursor, cursorKey, from)
  /* fillSince가 커서를 저장할 때 이 지역 값도 같이 바꾼다. 저장본을 다시 읽어 대사를 늦추지 않는다. */
  let persistedCursor: unknown = cursor

  const outcome = await fillSince({
    cfg: ref.connectionConfig ?? {},
    site,
    scope,
    cursorKey,
    cursor,
    fetch: (cfg, s, sc, since) => adapter.fetchSince!(cfg, s as SiteDescriptor, sc, since),
    ingest: records => liveIngest(domainId, instanceId, records, 'fill'),
    /*
     * ── 넣기 전에 기억을 채운다 (2026-09-06) ──────────────────────────────────
     * 중복 판정 기억은 프로세스 안에만 있다. 재기동하면 비워지고, 그러면 backfill 이 **이미 저널에
     * 있는 사실도 다시 앉힌다.** 실측으로 39건이 통째로 다시 앉았다(저널 44 → 83).
     *
     * push 경로에서는 안 한다 — 거기서는 재전송이 창 안에 들어와 이미 걸리고, 봉투마다 저널을
     * 조회하면 뜨거운 길이 느려진다.
     */
    beforeIngest: async records => {
      const span = spanOf(records)
      if (!span) return
      const { identities, scanned, truncated } = await journalIdentities(domainId, instanceId, span)
      const added = TwinEngine.warmDeduper(domainId, instanceId, identities)
      if (truncated) {
        twinWarn(
          `[twin-fill] "${instanceId}" 의 저널을 ${scanned}줄까지만 읽었습니다 — 그보다 앞의 사실은 ` +
            '다시 앉을 수 있습니다. 구간을 좁혀 부르거나 정체 색인이 필요합니다'
        )
      }
      if (added) twinLog(`[twin-fill] "${instanceId}" — 저널에서 정체 ${added}개를 기억에 채웠습니다`)
    },
    saveCursor: async next => {
      await repo.update({ id: (ref as any).id }, { liveCursor: next } as any)
      persistedCursor = next
    },
    ...(maxPages ? { maxPages } : {})
  })

  /* 원본이 실제로 답한 범위가 있을 때만 기록한다. 범위를 모르는 것은 "정상"으로 덮지 않는다. */
  await recordReferenceReconciliation({
    domainId,
    instanceId,
    scope,
    sourceRange: outcome.sourceRange,
    cursorSeq: seqOf(persistedCursor, cursorKey),
    firstSeqSeen: firstSeqOf(persistedCursor, cursorKey)
  }).catch(err => twinWarn(`[twin-reconcile] "${instanceId}" · ${scope} 상태를 적지 못했습니다: ${err?.message ?? err}`))

  const say = `[twin-fill] "${instanceId}" · ${scope} — ${outcome.applied}건 넣음 · ${outcome.pages}쪽 · ${outcome.stopped}${outcome.detail ? ` (${outcome.detail})` : ''}`
  /* 끝난 것과 막힌 것을 다른 소리로 남깁니다 — 구멍은 사람이 원본을 봐야 합니다. */
  if (outcome.stopped === 'done' || outcome.stopped === 'no-progress') twinLog(say)
  else twinWarn(say)

  return { ...base, ...outcome }
}

/**
 * **그 번호들이 무엇이었나** — 구멍을 만났을 때 사람이 묻는다.
 *
 * 커넥터는 「418 다음이 420」까지만 말한다. 그 419 가 무엇이었는지 알려면 원본에 물어야 하고,
 * 그 길이 없어서 사람이 아웃박스를 손으로 뒤졌다(2026-09-06).
 *
 * **판정하지 않는다.** 원본이 아는 것을 그대로 낸다 — 「그러니 포기해도 된다」는 되돌릴 수 없는
 * 판단이라 사람의 몫이다.
 */
export async function seqReportOf(args: {
  domainId: string
  instanceId: string
  scope: string
  from: number
  to: number
}): Promise<{ instanceId: string; scope: string; report?: SeqReport; skipped?: 'unknown-twin' | 'no-source' | 'no-adapter' | 'not-supported' }> {
  const { domainId, instanceId, scope, from, to } = args
  const base = { instanceId, scope }

  const inst = await getRepository(TwinInstance).findOne({ where: { domain: { id: domainId }, instanceId } })
  if (!inst) return { ...base, skipped: 'unknown-twin' }
  const ref = await getRepository(TwinReference).findOne({ where: { domain: { id: domainId }, source: instanceId } })
  if (!ref) return { ...base, skipped: 'no-source' }
  const adapter = getAdapter(ref.adapterType)
  if (!adapter) return { ...base, skipped: 'no-adapter' }
  /* 번호를 매기지 않는 원본은 물을 것이 없다 — 결함이 아니다. */
  if (typeof adapter.seqReport !== 'function') return { ...base, skipped: 'not-supported' }

  const siteId = (inst.origin as any)?.siteId
  const site: SiteDescriptor | undefined = siteId ? { siteId } : undefined
  return { ...base, report: await adapter.seqReport(ref.connectionConfig ?? {}, site, scope, from, to) }
}