cms-website/apps/api/src/modules/import/import.service.ts

// import_jobs 生命周期服务（Plan 3 Task 1 + Task 6 编排，spec §7.2/§7.5、前置事实 6/8/13）：
// 解析站点与快照 → 幂等建 job → 状态推进（queued → running → succeeded /
// succeeded_with_issues / failed）→ 报告落 <reportRoot>/<jobId>/reports/import.json
// （temp+rename 原子写、**先于终态落行**——行内 report_path 出现即文件存在）。
// 管道产出 Outcome（counts/issues）汇入报告；issues 非空 → succeeded_with_issues。
// 复用（幂等键命中）只返回 job 行锚定字段——快照/版本解析不回读（权威在既有报告）。
//
// 本文件与 register.service.ts 同纪律保持 **零装饰器**：CLI（scripts/sf.ts 经 tsx）与
// 集成测试直接 import 本文件，不拉起 Nest 装饰器执行；Nest 侧由 import.module.ts 薄包装。
//
// 三快照绑定（前置事实 13）：job.source_snapshot_id 指 **wxr 快照**（主输入）；
// media_bundle / manual_url_list 不入 job 行（不扩列），按 (wxr 快照的 connection, kind)
// 解析最新记入 job 报告 snapshots 段——导入管道（Task 4/6）从报告/上下文取用。
//
// 幂等：import_jobs.idempotency_key UNIQUE——重复提交同 key **复用既有 job 不重跑**
// （含失败 job：重试须换 key 或等 Task 6 的重试语义，本任务钉死“不静默重跑”）；并发
// 同 key 由唯一约束收敛（插入撞 23505 → 重查复用）。
//
// 版本块：四常量单源来自 @sf/adapter-wp 的 versions（ADAPTER/MAPPING/SANITIZER/PARSER）；
// job 行只落 mapping_version（MAPPING_VERSION），全四项进报告。默认幂等键 =
// site + wxr 快照 + 三语义版本（adapter/mapping/sanitizer——任一 bump 即新 job，对齐
// spec §7.5 “版本变化可显式触发重新标准化”；parser 是诊断信息不进键）。
//
// 管道注入：Task 6 起默认 = **编排管道**（四家写入器汇流，orchestratedPipeline）——真
// 事件流（runWxrAdapter 读 wxr 快照副本，mapConfig.locale 接 site.default_locale——前置
// 事实 8）→ content/term/media/route 严格串行分派 → finish 硬序（media 先于 route）→
// counts 按键**求和**聚合（Task 6 钉死，见 ImportPipelineOutcome.counts 契约）。注入
// pipeline 钩子供集成测试观察中间态/失败路径（注入时缺省事件流为空——Task 1 语义保持）。
//
// stale 回收（Task 6 义务④）：每次建 job 前把同 site 超过 24h 仍 queued/running 的 job
// 置 stale（崩溃/中断残留终态化）。心跳列决策留 ADR-017——现阶段无心跳列，以
// created_at 近似（job 为单进程就地推进，24h 远超任何真实执行时长）。
import { createReadStream } from 'node:fs'
import { mkdir, rename, rm, writeFile } from 'node:fs/promises'
import { dirname, join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { and, desc, eq, inArray, lt } from 'drizzle-orm'
import type { NodePgDatabase } from 'drizzle-orm/node-postgres'
import {
  ADAPTER_VERSION,
  MAPPING_VERSION,
  PARSER_VERSION,
  SANITIZER_VERSION,
  runWxrAdapter,
  type AdapterEvent,
  type AdapterTimings,
} from '@sf/adapter-wp'
import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js'
import type * as schema from '../../db/schema.js'
import { isUniqueViolation } from '../source/register.service.js'
import { createContentEventHandler } from './content-writer.js'
import { createTermEventHandler } from './term-writer.js'
import { createMediaEventHandlerFromContext } from './media-writer.js'
import { DEFAULT_NEWS_PATH, DEFAULT_PRODUCTS_PATH, createRouteEventHandler, normalizeListPath } from './route-writer.js'

/** 本文件位于 <repo>/apps/api/src/modules/import/ → 上五级是仓库根 */
const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..')

export type ImportDb = NodePgDatabase<typeof schema>

export type ImportJobStatus = (typeof schema.importJobStatus.enumValues)[number]

export type ImportErrorCode =
  | 'SITE_NOT_FOUND'
  | 'SNAPSHOT_NOT_FOUND'
  | 'NO_WXR_SNAPSHOT'
  | 'PIPELINE_FAILED'
  | 'REPORT_WRITE_FAILED'

/** 导入失败（可预期错误）：CLI 以 code 区分呈现；不裹程序 bug（那些原样上抛） */
export class ImportError extends Error {
  readonly code: ImportErrorCode
  constructor(code: ImportErrorCode, message: string) {
    super(message)
    this.name = 'ImportError'
    this.code = code
  }
}

/** 版本块（报告/上下文携带；job 行只落 mapping_version） */
export interface ImportVersionBlock {
  adapter: string
  mapping: string
  sanitizer: string
  parser: string
}

/** 已解析快照引用（报告与管道上下文的载荷；metadata 含 storageRoot 等登记期事实） */
export interface SnapshotRef {
  id: string
  kind: 'wxr' | 'media_bundle' | 'manual_url_list'
  sha256: string
  byteSize: number
  storagePath: string
  capturedAt: Date
  metadata: Record<string, unknown>
}

/** 管道执行上下文：Task 6 真实管道的全部输入（站点 locale 接缝见前置事实 8） */
export interface ImportJobContext {
  /**
   * 写入器自行管理事务：ctx.db 为直连（非事务代理），需要原子性的多语句写入用
   * db.transaction 自行包裹；单语句 upsert（on conflict do nothing/update）本身原子，
   * 无需显式事务。
   */
  db: ImportDb
  jobId: string
  siteId: string
  siteKey: string
  defaultLocale: string
  connectionId: string
  /** 主输入：wxr 快照（job.source_snapshot_id 指向它） */
  wxrSnapshot: SnapshotRef
  /** 侧输入（前置事实 13）：按 (connection, kind) 解析的最新快照，可能缺席（null） */
  mediaBundleSnapshot: SnapshotRef | null
  manualUrlListSnapshot: SnapshotRef | null
  versions: ImportVersionBlock
  /**
   * 报告累加器（与 Outcome 同形）：写入器可就地累加（ctx.report.counts.k++ /
   * ctx.report.issues.push(...)）。合并规则见 {@link ImportPipeline}——返回值优先。
   */
  report: ImportPipelineOutcome
  /**
   * 适配器事件流（runWxrAdapter 产物：row/unsupported/error/term 按文档序交错）。
   * Task 6 兑现：缺省 = 从 wxr 快照副本构造的真流（mapConfig.locale=site.default_locale
   * ——前置事实 8）；测试经 {@link CreateImportJobInput.events} 手构注入。单消费方
   * （编排器）一遍 for-await 分派给四家写入器——只承诺"可迭代一次"，无 tee 无重迭代。
   */
  events: AsyncIterable<AdapterEvent>
}

/**
 * 管道产出契约（Task 1 钉死、Task 6 修订——counts 聚合改**求和**、加法扩 issueGroups/
 * sanitize）：
 * - counts：计数块（键名由写入器约定，如 contentEntries/terms/routes）——原样进报告；
 * - issues：问题清单（简单字符串——机器可读错误码按 Plan 2 台账推迟）；**非空 → job
 *   终态 succeeded_with_issues**（问题全部入报告，不静默）；
 * - issueGroups / sanitize：可选分组视图与清洗统计（编排管道产出；issues 仍是全量扁平清单）。
 *
 * 汇合规则（Task 6 钉死）：最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键
 * **求和**、issues 串联（累加器在前、返回值在后）、issueGroups 逐键串联、sanitize
 * 数值相加。管道抛错时无返回值，累加器现值原样进报告（部分产出可审计）。
 */
export interface ImportPipelineOutcome {
  /**
   * 计数块（原样进报告）。**键契约：counts 键是跨写入器的扁平全局命名空间**——
   * 汇合按"同名键**求和**"聚合（ctx.report.counts ⊕ outcome.counts，Task 6 钉死：
   * 四家写入器键前缀互斥，求和是防覆盖的安全网——Task 1 的 CLOBBER 语义废弃）。
   * 各写入器仍须各持独占键名并在自身 JSDoc 声明归属（term：termsProcessed/
   * autoCreatedTerms/contentTermsCreated；content：contentEntries/revisionsCreated/
   * sourceRecords/skipped/unsupportedRecords/failedRecords；media 六键 media 前缀；
   * route 五键 routes 前缀——新写入器勿复用他家键名）。
   */
  counts: Record<string, number>
  issues: string[]
  /**
   * 可选：按来源分组的 issues 视图（报告分组呈现）。**issues 字段仍是无分组全量
   * 清单**（既有消费方契约不变）；分组键 = 四写入器前缀 + adapter（row 事件自带行级
   * 问题）+ orchestrator（编排器自身）。
   */
  issueGroups?: Partial<Record<ImportIssueGroup, string[]>>
  /** 可选：清洗统计（row 事件 sanitizeDiffs/changed 的聚合——编排层在分派循环收集） */
  sanitize?: ImportSanitizeStats
}

/** issues 分组键：四写入器前缀 + adapter（row 事件 issues 面）+ orchestrator（编排器） */
export type ImportIssueGroup = 'content' | 'term' | 'media' | 'route' | 'adapter' | 'orchestrator'

/** 清洗统计（Plan 2 报告纪律的导入侧镜像：changed 语义 = html !== 输入原文） */
export interface ImportSanitizeStats {
  /** changed=true 的 row 数（零 diff 等价变形也计——如 img 自闭合规范化） */
  changedRows: number
  /** 按 diff 类型计数（'removed-tag'/'removed-attr'——sparse：零计类型不占键） */
  byDiffType: Record<string, number>
}

/**
 * 写入器共享身份基类型（Task 6 义务③：编译期防漂移）——四家写入器的 identity 参数均以
 * 此为基础（`WriterContext & { 各家专属列 }`），db/siteId/connectionId 三列的形状与命名
 * 单源约束；ImportJobContext 结构性满足（Pick 自它），编排注入即传 ctx 的这三列。
 */
export type WriterContext = Pick<ImportJobContext, 'db' | 'siteId' | 'connectionId'>

/** 管道（缺省 = Task 6 编排管道 orchestratedPipeline——四家写入器汇流）。抛错 → job failed */
export type ImportPipeline = (ctx: ImportJobContext) => Promise<ImportPipelineOutcome>

export interface CreateImportJobInput {
  siteKey: string
  /** 显式 wxr 快照 id；缺省按 (connection, kind='wxr') 选最新 */
  snapshotId?: string
  /** 幂等键；缺省 = site + wxr 快照 + 三语义版本 自动派生 */
  idempotencyKey?: string
  /** 连接名（最新快照解析的 connection 侧）；缺省 'default'（与 register 缺省一致） */
  connectionName?: string
  /** 报告根（落 <reportRoot>/<jobId>/reports/import.json）；缺省 repoRoot/artifacts/import */
  reportRoot?: string
  /**
   * 适配器事件流（ctx.events 的注入位）：测试手构注入。注入而缺省 pipeline 时仍走编排
   * 管道（手构流 + 真编排）；注入 pipeline 而无 events 时为空流（Task 1 语义保持）；
   * 两者皆缺席 = 真流（wxr 快照副本，前置事实 8 的 locale 接线在真流构造处）。
   */
  events?: AsyncIterable<AdapterEvent>
  /** 管道钩子（测试注入）；缺省 = 编排管道 */
  pipeline?: ImportPipeline
  /** 主列表页路径（前置事实 9：CLI 参数来源，规范化值记入 job 报告）；缺省 /products/ */
  productsPath?: string
  /** 新闻列表页路径（同上）；缺省 /news/ */
  newsPath?: string
}

export interface CreateImportJobResult {
  jobId: string
  /** job 终态（或复用时既有状态——并发窗口下可能读到中间态） */
  status: ImportJobStatus
  /** true = 命中既有 job（幂等键复用，未重跑管道、未改行） */
  reused: boolean
  siteId: string
  /** job 行的 mapping_version（新建=当前 MAPPING_VERSION；复用=行值——权威记录在行） */
  mappingVersion: string
  reportPath: string | null
  durationMs: number
  /**
   * 以下字段**仅新建路径填充**（复用口径忠实化：reuse 不回读快照/版本解析，快照与
   * 版本块的权威记录是既有 job 报告——reportPath 指向它；复用返回只认 job 行锚定字段）
   */
  siteKey?: string
  connectionId?: string
  wxrSnapshot?: SnapshotRef
  mediaBundleSnapshot?: SnapshotRef | null
  manualUrlListSnapshot?: SnapshotRef | null
  versions?: ImportVersionBlock
}

const VERSIONS: ImportVersionBlock = {
  adapter: ADAPTER_VERSION,
  mapping: MAPPING_VERSION,
  sanitizer: SANITIZER_VERSION,
  parser: PARSER_VERSION,
}

// ---- 快照解析 ----

const SNAPSHOT_COLUMNS = {
  id: sourceSnapshots.id,
  siteId: sourceSnapshots.siteId,
  kind: sourceSnapshots.kind,
  sha256: sourceSnapshots.sha256,
  byteSize: sourceSnapshots.byteSize,
  storagePath: sourceSnapshots.storagePath,
  sourceConnectionId: sourceSnapshots.sourceConnectionId,
  capturedAt: sourceSnapshots.capturedAt,
  metadata: sourceSnapshots.metadata,
}

type SnapshotRow = {
  id: string
  siteId: string
  kind: 'wxr' | 'sitemap' | 'media_bundle' | 'manual_url_list'
  sha256: string
  byteSize: number
  storagePath: string
  sourceConnectionId: string
  capturedAt: Date
  metadata: Record<string, unknown>
}

function toRef(row: SnapshotRow): SnapshotRef {
  return {
    id: row.id,
    // 声音窄化：调用方按 kind 过滤查询（wxr/media_bundle/manual_url_list 三值）或已过
    // kind==='wxr' 校验——drizzle 列类型不携带过滤谓词，四值联合在此收窄是安全的
    kind: row.kind as SnapshotRef['kind'],
    sha256: row.sha256,
    byteSize: row.byteSize,
    storagePath: row.storagePath,
    capturedAt: row.capturedAt,
    metadata: row.metadata,
  }
}

/** 按 (connection, kind) 解析最新快照（capturedAt 降序、createdAt 降序兜底平局）；无则 null */
async function latestSnapshot(
  db: ImportDb,
  connectionId: string,
  kind: 'wxr' | 'media_bundle' | 'manual_url_list',
): Promise<SnapshotRow | null> {
  const rows = await db
    .select(SNAPSHOT_COLUMNS)
    .from(sourceSnapshots)
    .where(and(eq(sourceSnapshots.sourceConnectionId, connectionId), eq(sourceSnapshots.kind, kind)))
    .orderBy(desc(sourceSnapshots.capturedAt), desc(sourceSnapshots.createdAt))
    .limit(1)
  return rows.length > 0 ? (rows[0] as SnapshotRow) : null
}

// ---- 幂等键 ----

/** 默认幂等键：site + wxr 快照 + 三语义版本（定义钉死——报告可复算、CLI 可显式覆盖） */
export function defaultIdempotencyKey(siteKey: string, wxrSnapshotId: string): string {
  return `import:${siteKey}:${wxrSnapshotId}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}`
}

// ---- 主入口 ----

/**
 * 创建（并就地推进到底）一个导入 job：
 * 解析（site → wxr 快照 → 侧快照）→ 幂等建行（queued）→ running（attempt=1、started_at）
 * → 空管道/注入管道 → succeeded｜failed（error_code/error_summary/finished_at）→ 报告落盘。
 * 失败口径：解析阶段失败零 job 行；管道阶段失败 job 行保留（failed 终态 + error 列）。
 */
export async function createImportJob(
  db: ImportDb,
  input: CreateImportJobInput,
): Promise<CreateImportJobResult> {
  const startedAt = performance.now()

  // 阶段 0：站点解析（不自动建站——建站是 register 的职责）
  const siteRows = await db
    .select({ id: sites.id, key: sites.key, defaultLocale: sites.defaultLocale })
    .from(sites)
    .where(eq(sites.key, input.siteKey))
    .limit(1)
  if (siteRows.length === 0) {
    throw new ImportError('SITE_NOT_FOUND', `站点不存在：${input.siteKey}（先运行 sf source:register）`)
  }
  const site = siteRows[0]!

  // 阶段 0.5：stale 回收（Task 6 义务④）——同 site 超过 24h 仍 queued/running 的 job 置
  // stale（崩溃/中断残留的终态化，finished_at 记宣判时刻）。心跳列决策留 ADR-017：现阶
  // 段无心跳列，以 created_at 近似（job 单进程就地推进，24h 远超任何真实执行时长）。
  await db
    .update(importJobs)
    .set({ status: 'stale', finishedAt: new Date() })
    .where(
      and(
        eq(importJobs.siteId, site.id),
        inArray(importJobs.status, ['queued', 'running']),
        lt(importJobs.createdAt, new Date(Date.now() - STALE_AFTER_MS)),
      ),
    )

  // 阶段 1：wxr 主输入快照解析（显式 id 优先；缺省按 (connection, kind='wxr') 最新）
  let wxrRow: SnapshotRow | null = null
  if (input.snapshotId !== undefined) {
    const rows = await db
      .select(SNAPSHOT_COLUMNS)
      .from(sourceSnapshots)
      .where(eq(sourceSnapshots.id, input.snapshotId))
      .limit(1)
    if (rows.length === 0) {
      throw new ImportError('SNAPSHOT_NOT_FOUND', `快照不存在：${input.snapshotId}`)
    }
    const row = rows[0] as SnapshotRow
    if (row.siteId !== site.id) {
      throw new ImportError(
        'SNAPSHOT_NOT_FOUND',
        `快照 ${input.snapshotId} 不属于站点 ${input.siteKey}`,
      )
    }
    if (row.kind !== 'wxr') {
      throw new ImportError(
        'NO_WXR_SNAPSHOT',
        `快照 ${input.snapshotId} 的 kind 是 ${row.kind}，不是 wxr——主输入必须是 wxr 快照`,
      )
    }
    wxrRow = row
  } else {
    const connectionName = input.connectionName ?? 'default'
    // connection 判重键 (site_id, name) 无库级唯一约束（Plan 1 决定，register 是唯一写入方）
    const connRows = await db
      .select({ id: sourceConnections.id })
      .from(sourceConnections)
      .where(and(eq(sourceConnections.siteId, site.id), eq(sourceConnections.name, connectionName)))
      .limit(1)
    if (connRows.length > 0) {
      wxrRow = await latestSnapshot(db, connRows[0]!.id, 'wxr')
    }
    if (wxrRow === null) {
      throw new ImportError(
        'NO_WXR_SNAPSHOT',
        `站点 ${input.siteKey}（connection=${connectionName}）没有任何 wxr 快照——先运行 sf source:register --wxr`,
      )
    }
  }
  const connectionId = wxrRow.sourceConnectionId

  // 阶段 2：侧快照（前置事实 13）——挂在 wxr 快照的 connection 下按 kind 解析最新
  const mediaBundleRow = await latestSnapshot(db, connectionId, 'media_bundle')
  const manualUrlListRow = await latestSnapshot(db, connectionId, 'manual_url_list')
  const wxrSnapshot = toRef(wxrRow)
  const mediaBundleSnapshot = mediaBundleRow === null ? null : toRef(mediaBundleRow)
  const manualUrlListSnapshot = manualUrlListRow === null ? null : toRef(manualUrlListRow)

  // 阶段 3：幂等建 job（queued）——命中既有 key 直接复用返回（不重跑）。
  // 复用口径忠实化：只返回 job 行锚定字段（含行内 mapping_version/report_path），
  // 快照与版本块不回读（权威记录在既有 job 报告——reportPath 指向它）。
  const idempotencyKey = input.idempotencyKey ?? defaultIdempotencyKey(input.siteKey, wxrSnapshot.id)
  const existing = await reuseExisting(db, idempotencyKey)
  if (existing !== null) {
    return { ...existing, durationMs: performance.now() - startedAt }
  }
  let jobId: string
  try {
    const inserted = await db
      .insert(importJobs)
      .values({
        siteId: site.id,
        sourceSnapshotId: wxrSnapshot.id,
        status: 'queued',
        mappingVersion: MAPPING_VERSION,
        idempotencyKey,
      })
      .returning({ id: importJobs.id })
    jobId = inserted[0]!.id
  } catch (err) {
    // 并发同 key：唯一约束收敛——重查复用（插入撞 23505 说明行已存在，重查必命中）
    if (isUniqueViolation(err)) {
      const raced = await reuseExisting(db, idempotencyKey)
      if (raced !== null) {
        return { ...raced, durationMs: performance.now() - startedAt }
      }
    }
    throw err
  }

  // 阶段 4：状态推进 queued → running（attempt 记首次执行）
  const attempt = 1
  await db
    .update(importJobs)
    .set({ status: 'running', startedAt: new Date(), attempt })
    .where(eq(importJobs.id, jobId))

  // 阶段 5：管道——缺省 = 编排管道（四家写入器汇流）；注入钩子优先。抛错 → failed 终态。
  // 事件流三分：注入 events 优先（手构流 + 缺省编排管道 = 测试形态）→ 注入 pipeline 而
  // 无 events = 空流（Task 1 语义保持）→ 双缺席 = 真流（wxr 快照副本 + locale 接线，
  // 前置事实 8——缺省空串会让全部行被 content-writer 以 locale 不匹配拒收）。
  const productsPath = normalizeListPath(input.productsPath ?? DEFAULT_PRODUCTS_PATH)
  const newsPath = normalizeListPath(input.newsPath ?? DEFAULT_NEWS_PATH)
  const useRealStream = input.events === undefined && input.pipeline === undefined
  const adapterTimings: AdapterTimings | null = useRealStream ? { parseMs: 0, sanitizeMs: 0 } : null
  const events: AsyncIterable<AdapterEvent> =
    input.events !== undefined
      ? input.events
      : input.pipeline !== undefined
        ? emptyEvents()
        : realEventStream(wxrSnapshot, site.defaultLocale, adapterTimings!)
  const pipeline: ImportPipeline =
    input.pipeline !== undefined ? input.pipeline : orchestratedPipeline({ productsPath, newsPath })

  let failed: { code: ImportErrorCode; summary: string } | null = null
  const ctx: ImportJobContext = {
    db,
    jobId,
    siteId: site.id,
    siteKey: site.key,
    defaultLocale: site.defaultLocale,
    connectionId,
    wxrSnapshot,
    mediaBundleSnapshot,
    manualUrlListSnapshot,
    versions: VERSIONS,
    report: { counts: {}, issues: [] },
    events,
  }
  let outcome: ImportPipelineOutcome | null = null
  try {
    outcome = await pipeline(ctx)
  } catch (err) {
    failed = {
      code: err instanceof ImportError ? err.code : 'PIPELINE_FAILED',
      // '[pipeline]' 前缀：管道阶段错误与解析阶段错误在 error_summary 上可区分
      //（ImportError 保留原 code——管道内可预期失败复用同一套错误码面）
      summary: `[pipeline] ${err instanceof Error ? `${err.name}: ${err.message}` : String(err)}`,
    }
  }
  // 抛错时无返回值：累加器现值进报告（部分产出可审计）
  const merged: ImportPipelineOutcome = outcome === null ? ctx.report : mergeOutcomes(ctx.report, outcome)
  let finalStatus: ImportJobStatus
  if (failed !== null) {
    finalStatus = 'failed'
  } else {
    finalStatus = merged.issues.length > 0 ? 'succeeded_with_issues' : 'succeeded'
  }

  // 阶段 6：报告先行落盘（temp+rename 原子替换），终态落行在后——行内 report_path
  // 出现即文件存在；写盘失败 → job failed（REPORT_WRITE_FAILED、report_path 不落，
  // 不留“succeeded 但报告失踪”或悬挂路径）
  const reportRoot = resolve(input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'import'))
  const reportDir = join(reportRoot, jobId, 'reports')
  const reportPath = join(reportDir, 'import.json')
  const report = {
    jobId,
    generatedAt: new Date().toISOString(),
    status: finalStatus,
    siteKey: site.key,
    siteId: site.id,
    connectionId,
    idempotencyKey,
    attempt,
    snapshots: {
      wxr: slimRef(wxrSnapshot),
      media_bundle: mediaBundleSnapshot === null ? null : slimRef(mediaBundleSnapshot),
      manual_url_list: manualUrlListSnapshot === null ? null : slimRef(manualUrlListSnapshot),
    },
    versions: VERSIONS,
    // 主列表路径（前置事实 9：CLI 参数来源记入报告——route-writer 实际消费的规范化值）
    paths: { productsPath, newsPath },
    counts: merged.counts,
    issues: merged.issues,
    ...(merged.issueGroups === undefined ? {} : { issueGroups: merged.issueGroups }),
    ...(merged.sanitize === undefined ? {} : { sanitize: merged.sanitize }),
    // 适配器分段计时（仅真流编排——Plan 2 报告纪律的 parse/sanitize 分段在导入侧续录）
    ...(adapterTimings === null ? {} : { adapterTiming: { parseMs: adapterTimings.parseMs, sanitizeMs: adapterTimings.sanitizeMs } }),
    error: failed,
    durationMs: performance.now() - startedAt,
  }
  let reportWritten = true
  try {
    await mkdir(reportDir, { recursive: true })
    const tmpPath = `${reportPath}.tmp`
    try {
      await writeFile(tmpPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8')
      await rename(tmpPath, reportPath)
    } catch (err) {
      await rm(tmpPath, { force: true })
      throw err
    }
  } catch (err) {
    reportWritten = false
    failed = {
      code: 'REPORT_WRITE_FAILED',
      summary: `[report] ${err instanceof Error ? `${err.name}: ${err.message}` : String(err)}`,
    }
    finalStatus = 'failed'
  }
  const finishedAt = new Date()
  // 终态落行：failed===null ⟹ 写盘成功（report_path 同步入行）；failed≠null 时按
  // 写盘成败决定 report_path 去留（写失败不落——文件不存在不悬挂路径）
  const finalUpdate =
    failed === null
      ? { status: finalStatus, finishedAt, reportPath }
      : reportWritten
        ? { status: finalStatus, finishedAt, reportPath, errorCode: failed.code, errorSummary: failed.summary }
        : { status: finalStatus, finishedAt, errorCode: failed.code, errorSummary: failed.summary }
  await db.update(importJobs).set(finalUpdate).where(eq(importJobs.id, jobId))

  return {
    jobId,
    status: finalStatus,
    reused: false,
    siteId: site.id,
    mappingVersion: MAPPING_VERSION,
    siteKey: site.key,
    connectionId,
    wxrSnapshot,
    mediaBundleSnapshot,
    manualUrlListSnapshot,
    versions: VERSIONS,
    reportPath: reportWritten ? reportPath : null,
    durationMs: performance.now() - startedAt,
  }
}

/** 既有 job 的复用返回（不动行、不跑管道；仅 job 行锚定字段）；无则 null */
async function reuseExisting(
  db: ImportDb,
  idempotencyKey: string,
): Promise<Pick<CreateImportJobResult, 'jobId' | 'status' | 'reused' | 'siteId' | 'mappingVersion' | 'reportPath'> | null> {
  const found = await db
    .select({
      id: importJobs.id,
      status: importJobs.status,
      siteId: importJobs.siteId,
      mappingVersion: importJobs.mappingVersion,
      reportPath: importJobs.reportPath,
    })
    .from(importJobs)
    .where(eq(importJobs.idempotencyKey, idempotencyKey))
    .limit(1)
  if (found.length === 0) return null
  const row = found[0]!
  return {
    jobId: row.id,
    status: row.status,
    reused: true,
    siteId: row.siteId,
    mappingVersion: row.mappingVersion,
    reportPath: row.reportPath,
  }
}

/** 报告里的快照引用瘦身版：不含 metadata（media_bundle 清单可达数千条，报告只留身份锚点） */
function slimRef(ref: SnapshotRef): {
  id: string
  kind: string
  sha256: string
  byteSize: number
  storagePath: string
} {
  return { id: ref.id, kind: ref.kind, sha256: ref.sha256, byteSize: ref.byteSize, storagePath: ref.storagePath }
}

/** 缺省空事件流（input.events 缺席时的 ctx.events 占位；每次调用产新迭代器） */
async function* emptyEvents(): AsyncGenerator<AdapterEvent, void, void> {}

// ---- 编排管道（Task 6 汇流——Task 5 审查移交的契约底本在此兑现） ----

/** stale 回收阈值：queued/running 滞留超过此时长即宣判（毫秒） */
const STALE_AFTER_MS = 24 * 60 * 60 * 1000

/** wxr 快照副本绝对路径（metadata.storageRoot + storagePath——登记期固化的存储根事实） */
function wxrSnapshotCopyPath(snapshot: SnapshotRef): string {
  const root = snapshot.metadata.storageRoot
  if (typeof root !== 'string' || root === '') {
    throw new Error(
      `wxr 快照 ${snapshot.id} metadata 缺 storageRoot（登记期不变量破裂——无法读取副本构造事件流）`,
    )
  }
  return join(root, snapshot.storagePath)
}

/**
 * 真事件流（惰性生成器：路径解析/读盘错误推迟到首拍 next()——落进管道执行期，由
 * createImportJob 的 PIPELINE_FAILED 收口，不留"job 已 running 而流构造抛错"的悬挂行）。
 * 前置事实 8：mapConfig.locale 必须接 site.default_locale（缺省空串 = 全行被拒收）。
 */
async function* realEventStream(
  snapshot: SnapshotRef,
  locale: string,
  timings: AdapterTimings,
): AsyncGenerator<AdapterEvent, void, void> {
  yield* runWxrAdapter(createReadStream(wxrSnapshotCopyPath(snapshot)), {
    mapConfig: { locale },
    timings,
  })
}

/** counts 按键求和（Task 6 钉死的聚合策略——就地累加 target） */
function sumCounts(target: Record<string, number>, source: Record<string, number>): void {
  for (const [key, value] of Object.entries(source)) {
    target[key] = (target[key] ?? 0) + value
  }
}

/** Outcome 汇合（ImportPipelineOutcome JSDoc 契约）：counts 求和、issues/issueGroups 串联、sanitize 相加 */
function mergeOutcomes(acc: ImportPipelineOutcome, ret: ImportPipelineOutcome): ImportPipelineOutcome {
  const counts: Record<string, number> = { ...acc.counts }
  sumCounts(counts, ret.counts)
  const issueGroups: Partial<Record<ImportIssueGroup, string[]>> = {}
  let hasGroups = false
  for (const key of Object.keys(acc.issueGroups ?? {}) as ImportIssueGroup[]) {
    issueGroups[key] = [...(acc.issueGroups![key] ?? [])]
    hasGroups = true
  }
  for (const key of Object.keys(ret.issueGroups ?? {}) as ImportIssueGroup[]) {
    issueGroups[key] = [...(issueGroups[key] ?? []), ...(ret.issueGroups![key] ?? [])]
    hasGroups = true
  }
  const sanitize: ImportSanitizeStats | undefined =
    acc.sanitize === undefined && ret.sanitize === undefined
      ? undefined
      : {
          changedRows: (acc.sanitize?.changedRows ?? 0) + (ret.sanitize?.changedRows ?? 0),
          byDiffType: {},
        }
  if (sanitize !== undefined) {
    for (const src of [acc.sanitize?.byDiffType, ret.sanitize?.byDiffType]) {
      if (src !== undefined) sumCounts(sanitize.byDiffType, src)
    }
  }
  return {
    counts,
    issues: [...acc.issues, ...ret.issues],
    ...(hasGroups ? { issueGroups } : {}),
    ...(sanitize === undefined ? {} : { sanitize }),
  }
}

/**
 * 编排管道（Task 6 汇流契约底本——Task 5 审查移交钉死）：
 * - 事件期**严格串行 await** 分派 content → term → media → route：三家（term/media/
 *   route）的 entry 锚点解析依赖 content 对**同事件**的落库产物（source_records.
 *   content_entry_id / entry_locales），禁 Promise.all——并行下锚点查询可能抢在 content
 *   写入前发生；
 * - finish 唯一硬序：term/content（同步、无库依赖）任意 → `await media.finish()` 先于
 *   `await route.finish()`（attachment 301 反查与 asset 全表都读 media_assets——资产须已齐）；
 * - 分派规则（Task 3 审查移交⑥）：wp:attachment 行的媒体域归 media-writer
 *   （media_assets 唯一写入方），content-writer 对其只留 source_records record-only 足迹
 *   （前置事实 11）——两家各写各域，单一 owner；
 * - counts 按键求和聚合（安全网——四家键前缀互斥）；清洗统计与 row 事件行级问题在
 *   分派循环收集（RowResult 契约：issues 逐条收录为行级错误）。
 */
function orchestratedPipeline(paths: { productsPath: string; newsPath: string }): ImportPipeline {
  return async (ctx) => {
    const content = createContentEventHandler({
      db: ctx.db,
      siteId: ctx.siteId,
      connectionId: ctx.connectionId,
      snapshotId: ctx.wxrSnapshot.id,
      jobId: ctx.jobId,
      defaultLocale: ctx.defaultLocale,
      versions: ctx.versions,
    })
    const term = createTermEventHandler({
      db: ctx.db,
      siteId: ctx.siteId,
      connectionId: ctx.connectionId,
    })
    const media = await createMediaEventHandlerFromContext(ctx)
    const route = createRouteEventHandler({
      db: ctx.db,
      siteId: ctx.siteId,
      connectionId: ctx.connectionId,
      snapshotId: ctx.wxrSnapshot.id,
      defaultLocale: ctx.defaultLocale,
      productsPath: paths.productsPath,
      newsPath: paths.newsPath,
    })

    const sanitize: ImportSanitizeStats = { changedRows: 0, byDiffType: {} }
    const adapterIssues: string[] = []
    for await (const ev of ctx.events) {
      if (ev.kind === 'row') {
        if (ev.changed) sanitize.changedRows++
        for (const diff of ev.sanitizeDiffs) {
          sanitize.byDiffType[diff.type] = (sanitize.byDiffType[diff.type] ?? 0) + 1
        }
        for (const issue of ev.issues) {
          adapterIssues.push(`行级问题（${ev.row.sourceType}/${ev.row.sourceId}）：${issue}`)
        }
      }
      await content.handleEvent(ev)
      await term.handleEvent(ev)
      await media.handleEvent(ev)
      await route.handleEvent(ev)
    }
    const t = term.finish()
    const c = content.finish()
    const m = await media.finish()
    const r = await route.finish()

    const counts: Record<string, number> = {}
    sumCounts(counts, c.counts)
    sumCounts(counts, t.counts)
    sumCounts(counts, m.counts)
    sumCounts(counts, r.counts)
    return {
      counts,
      issues: [...c.issues, ...t.issues, ...m.issues, ...r.issues, ...adapterIssues],
      issueGroups: {
        content: c.issues,
        term: t.issues,
        media: m.issues,
        route: r.issues,
        adapter: adapterIssues,
        orchestrator: [],
      },
      sanitize,
    }
  }
}