// 内容稳定身份写入器（Plan 3 Task 3，spec §5.3/§7.5、前置事实 6/8/11/12、ADR-006/010）：
// 消费适配器事件流把内容行落库——幂等三态（miss / checksum 同 skip / checksum 异新修订）。
//
// 稳定身份两层（ADR-006）：
// - 快照行（六列）：site + connection + source_type + source_id + locale + **snapshot_id**
//   （source_records_identity_uq）——每快照各一行，checksum 对比的对象；
// - 内容身份（五列）：上面去掉 snapshot_id——跨快照复用同一 content_entry（permalink
//   变化、快照重导都不建新 entry）。五列解析与 term-writer.resolveContentEntryId 同口径
//   （latest first），本写入器是 source_records.content_entry_id 的**唯一写入方**（Task 2
//   测试以 seed 模拟的接缝在此成为真实落库）。
//
// normalized checksum（前置事实 6 公式钉死，P0 评审 Issue 2）：
// canonicalJson({ universal_row_version, adapter_version, mapping_version, sanitizer_version,
// locale, mapped_content, taxonomy_relations, media_reference_identities, route_source_values })
// 再 sha256 hex（与 register 快照哈希同法）。要点：
// - **rawLocation 显式排除**（位置性字段，进哈希会破坏跨快照 skip）；
// - 版本三常量必入（bump 语义——任一 bump 即 checksum 变化 → 新修订，spec §7.5
//   "版本变化可显式触发重新标准化"）；parser 是诊断信息不入；
// - status 必入（draft↔publish 翻转必须可见）；sourceUrl 经 route_source_values 入
//   （permalink 变化产新修订、不建新 entry——spec §7.5 一致）；
// - mapped_content 与 revision.data 同构：{ title?, slug?, status, bodyHtml, seo,
//   knownFields, extras }——可选字段**缺席即无值**（契约纪律），不以 undefined/null 占位。
//
// title 提升纪律（有意不做）：mapItem 把 title 留在 knownFields（契约冻结，升版须评审），
// data.title 只镜像 row.title（适配器行该键缺席——title 在 knownFields 可取）。
//
// attachment 不建 content_entries（前置事实 11）：wp:attachment 行只记 source_records
// （content_entry_id NULL、parse_status='imported'）——媒体管道（Task 4）与 301 路由
// （Task 5）的消费锚点。typeKey 映射钉死：wp:post→news_article、wp:page→basic_page、
// wp:product→product。
//
// unsupported/error 也写 source_records（前置事实 12）：parse_status='unsupported'/'failed'、
// locale=站点默认、不关联 entry——"不支持对象有明确结果"的可审计性；postId 缺失时
// source_id 以 rawLocation 回退（**不硬造**身份）。这两类是预期输入——写入器侧不推
// issue；Task 6 起编排层将 reason/issue 归档 adapter 组（进 report.issues，终态随之
// succeeded_with_issues——两层纪律有意分层）。
//
// 事务边界（ctx.db 直连约定下的自行包裹）：跨语句不变量的路径（全新建四行 / 变化三步）
// 用 db.transaction；单语句路径（跨快照 skip 补行）单 INSERT 自原子。写入路径对
// content_revisions **纯 INSERT**（不可变触发器兼容，ADR-010）；UPDATE 只发生在可变表
// entry_locales（current 指针/slug 前移，published 永不碰）与 source_records（checksum/
// source_url 前移）。
//
// locale 校验（前置事实 8）：row.locale 必须等于站点 default_locale（显式支持列表本切片
// 无输入源——site 表只有 default_locale）；不匹配 → issue + 跳过（不落任何行）。
// Task 6 编排负责把 site.default_locale 接进 mapConfig（否则适配器产 locale 空串行，
// 1047 行全被此处拒收——接缝两侧双保险）。
import { createHash } from 'node:crypto'
import { and, eq } from 'drizzle-orm'
import { UNIVERSAL_ROW_VERSION, type UniversalRow } from '@sf/contracts'
import type { AdapterEvent } from '@sf/adapter-wp'
import { canonicalJson } from '../../common/canonical.js'
import { contentEntries, contentRevisions, entryLocales, sourceRecords } from '../../db/schema.js'
import { findLatestLinkedSourceRecord } from './anchors.js'
import type { ImportDb, ImportPipeline, ImportPipelineOutcome, ImportVersionBlock, WriterContext } from './import.service.js'

/** 事务句柄类型（db.transaction 回调参数——从 ImportDb 的事务签名提取，避免重复泛型展开） */
type Tx = Parameters<Parameters<ImportDb['transaction']>[0]>[0]

/** sourceType → content_entries.type_key（前置事实 11 钉死；wp:attachment 不在此表——不建 entry） */
export const CONTENT_TYPE_KEY_MAP: Readonly<Record<string, string>> = Object.freeze({
  'wp:post': 'news_article',
  'wp:page': 'basic_page',
  'wp:product': 'product',
})

/** row 事件产出的 source_records.parse_status（attachment/内容行同值；unsupported/error 另有两值） */
const PARSE_STATUS_IMPORTED = 'imported'
const PARSE_STATUS_UNSUPPORTED = 'unsupported'
const PARSE_STATUS_FAILED = 'failed'

/** content_revisions 固定落位：导入修订的 source_kind / created_by / schema_version */
const REVISION_SOURCE_KIND = 'import' as const
const REVISION_CREATED_BY = 'import'

/**
 * mapped_content / revision.data 的共用结构（前置事实 6 钉死）：可选字段缺席即无值。
 * extras 可能是 null 原型对象（mapItem 的原型污染防御）——canonicalJson 与 jsonb 序列化
 * 均兼容，此处不复制。
 */
function mappedContentOf(row: UniversalRow): Record<string, unknown> {
  const mapped: Record<string, unknown> = {
    status: row.status,
    seo: row.seo,
    knownFields: row.knownFields,
    extras: row.extras,
  }
  if (row.title !== undefined) mapped.title = row.title
  if (row.slug !== undefined) mapped.slug = row.slug
  if (row.bodyHtml !== undefined) mapped.bodyHtml = row.bodyHtml
  return mapped
}

/**
 * normalized checksum（前置事实 6 公式精确实现）：canonicalJson 输入域后 sha256 hex。
 * 导出供 Task 4 媒体写入器复用（attachment 行的 source_records.normalized_checksum 同式）。
 */
export function normalizedContentChecksum(
  row: UniversalRow,
  versions: Pick<ImportVersionBlock, 'adapter' | 'mapping' | 'sanitizer'>,
): string {
  const routeSourceValues: Record<string, unknown> = {}
  if (row.sourceUrl !== undefined) routeSourceValues.sourceUrl = row.sourceUrl
  return createHash('sha256')
    .update(
      canonicalJson({
        universal_row_version: UNIVERSAL_ROW_VERSION,
        adapter_version: versions.adapter,
        mapping_version: versions.mapping,
        sanitizer_version: versions.sanitizer,
        locale: row.locale,
        mapped_content: mappedContentOf(row),
        taxonomy_relations: row.taxonomyRefs,
        media_reference_identities: row.attachmentRefs,
        route_source_values: routeSourceValues,
      }),
    )
    .digest('hex')
}

/**
 * 可分派的内容写入器句柄（Task 6 编排形状，与 term-writer 同构）：编排器单消费方
 * for-await 拉一遍事件流、逐事件分派；两态（counts/issues）经 finish 取回。句柄非可重入。
 */
export interface ContentEventHandler {
  /** 单事件分派（term 事件 no-op——term-writer 主场） */
  handleEvent(ev: AdapterEvent): Promise<void>
  /** 流终取回累计产出（counts/issues；调用后句柄弃用） */
  finish(): ImportPipelineOutcome
}

/**
 * 本写入器的 counts 键与归属（**counts 键是跨写入器的扁平全局命名空间**——import.service
 * 按"同名键**求和**"合并（Task 6 钉死——防覆盖安全网），各写入器必须各持独占键名，见
 * ImportPipelineOutcome.counts 的键契约注记）。本写入器独占：contentEntries /
 * revisionsCreated / sourceRecords / skipped / unsupportedRecords / failedRecords——
 * **Task 4 media-writer 起新键名，勿复用 'skipped'**（语义不同会互相污染）。
 * 前四键恒出现在报告；unsupportedRecords/failedRecords 只在 >0 时出现（分项按事件计）。
 */
interface ContentWriterCounts {
  contentEntries: number
  revisionsCreated: number
  sourceRecords: number
  skipped: number
  unsupportedRecords?: number
  failedRecords?: number
}

/**
 * 内容写入器的事件分派句柄。产出 counts：
 * - contentEntries：新建 content_entries 数（幂等重跑/变化路径为 0）；
 * - revisionsCreated：新建 content_revisions 数（checksum 命中为 0——G2 验收锚点）；
 * - sourceRecords：新建 source_records 数（内容/attachment/unsupported/error 全计；
 *   跨快照重导按快照补行故可为正；NULL-entry 占位行收编**不计**——行已在，UPDATE
 *   挂接非新建，见 handleContentRow 态 1b）；
 * - skipped：内容行未产修订的跳过数（同快照 checksum 命中、跨快照无业务变化、locale
 *   拒收三类；attachment 命中不计——媒体域是 Task 4 主场）；
 * - unsupportedRecords/failedRecords：unsupported/error 事件分项数（**按事件计**非按
 *   插入行——重跑行幂等而计数如实；拆分只在分派点可知，Task 6 报告层无从重derive）。
 * issues：locale 不匹配、未知 sourceType 兜底——显式报告（§7.6 禁静默）。
 */
export function createContentEventHandler(identity: WriterContext & {
  /** 本 job 的 wxr 快照 id（六列身份的 snapshot 侧 + 新行 source_snapshot_id） */
  snapshotId: string
  /** 本 job id（source_records.import_job_id 与 content_revisions.import_job_id 落库锚点） */
  jobId: string
  /** 站点默认 locale（前置事实 8：row.locale 校验基准） */
  defaultLocale: string
  /** 版本块（checksum 的三常量输入；parser 不入哈希） */
  versions: ImportVersionBlock
}): ContentEventHandler {
  const { db, siteId } = identity
  // 交叉 Record：finish() 要把 counts 交还给 ImportPipelineOutcome（Record<string, number>）
  // ——接口无索引签名不可赋值；交叉后已知键类型不变、动态赋值兼容
  const counts: ContentWriterCounts & Record<string, number> = {
    contentEntries: 0,
    revisionsCreated: 0,
    sourceRecords: 0,
    skipped: 0,
  }
  const issues: string[] = []

  /** 修订插入（纯 INSERT——不可变表；data 与 checksum 同构于 mapped_content） */
  async function insertRevision(
    tx: Tx,
    args: { entryId: string; row: UniversalRow; data: Record<string, unknown>; checksum: string; sourceRecordId: string },
  ): Promise<string> {
    const inserted = await tx
      .insert(contentRevisions)
      .values({
        siteId,
        contentEntryId: args.entryId,
        locale: args.row.locale,
        schemaVersion: UNIVERSAL_ROW_VERSION,
        data: args.data,
        checksum: args.checksum,
        sourceKind: REVISION_SOURCE_KIND,
        sourceRecordId: args.sourceRecordId,
        importJobId: identity.jobId,
        createdBy: REVISION_CREATED_BY,
      })
      .returning({ id: contentRevisions.id })
    return inserted[0]!.id
  }

  /**
   * current 指针前移 + slug 跟随最新行（可变表 UPDATE——published 永不碰，ADR-010）。
   * 行缺席时补建（跨写入器接缝兜底：本写入器正常路径首导必先建 entry_locales）。
   */
  async function pointCurrent(tx: Tx, entryId: string, row: UniversalRow, revisionId: string): Promise<void> {
    const updated = await tx
      .update(entryLocales)
      .set({ currentRevisionId: revisionId, slug: row.slug ?? row.sourceId, updatedAt: new Date() })
      .where(
        and(
          eq(entryLocales.siteId, siteId),
          eq(entryLocales.contentEntryId, entryId),
          eq(entryLocales.locale, row.locale),
        ),
      )
      .returning({ id: entryLocales.id })
    if (updated.length === 0) {
      await tx.insert(entryLocales).values({
        siteId,
        contentEntryId: entryId,
        locale: row.locale,
        slug: row.slug ?? row.sourceId,
        currentRevisionId: revisionId,
      })
    }
  }

  /** 快照行（六列身份）查找：本快照内该源对象的 source_records 行（唯一索引保证至多一行） */
  async function findSnapshotRecord(row: UniversalRow) {
    const rows = await db
      .select({
        id: sourceRecords.id,
        contentEntryId: sourceRecords.contentEntryId,
        normalizedChecksum: sourceRecords.normalizedChecksum,
      })
      .from(sourceRecords)
      .where(
        and(
          eq(sourceRecords.siteId, siteId),
          eq(sourceRecords.sourceConnectionId, identity.connectionId),
          eq(sourceRecords.sourceType, row.sourceType),
          eq(sourceRecords.sourceId, row.sourceId),
          eq(sourceRecords.locale, row.locale),
          eq(sourceRecords.sourceSnapshotId, identity.snapshotId),
        ),
      )
      .limit(1)
    return rows[0] ?? null
  }

  /** 五列内容身份 → 既有 entry 的最新快照行（anchors.ts 共享 helper——latest first，跨快照） */
  async function findPriorLinkedRecord(row: UniversalRow) {
    return findLatestLinkedSourceRecord(db, { siteId, connectionId: identity.connectionId }, row)
  }

  /** 内容行主路径：幂等三态（miss / checksum 同 skip / checksum 异新修订） */
  async function handleContentRow(row: UniversalRow, typeKey: string): Promise<void> {
    const checksum = normalizedContentChecksum(row, identity.versions)
    const data = mappedContentOf(row)

    // 态 1：本快照行已存在（挂 entry——本写入器是 content 行 source_records 的唯一写入方，
    // 挂接不变量由写入路径自身保证）。
    const snapRow = await findSnapshotRecord(row)
    if (snapRow !== null && snapRow.contentEntryId !== null) {
      if (snapRow.normalizedChecksum === checksum) {
        counts.skipped++
        return
      }
      // checksum 异：快照行前移（checksum/source_url）+ 新修订 + current 前移（事务）。
      // lineage 语义（有意）：source_records.import_job_id 留在建行 job（行的溯源=创建它的
      // 导入），本次 job 记在 content_revisions.import_job_id（谁产了新修订）
      const entryId = snapRow.contentEntryId // 事务回调外的 const 收窄（属性收窄不进闭包）
      await db.transaction(async (tx) => {
        await tx
          .update(sourceRecords)
          .set({ normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null })
          .where(eq(sourceRecords.id, snapRow.id))
        const revisionId = await insertRevision(tx, {
          entryId,
          row,
          data,
          checksum,
          sourceRecordId: snapRow.id,
        })
        await pointCurrent(tx, entryId, row, revisionId)
      })
      counts.revisionsCreated++
      return
    }
    if (snapRow !== null) {
      // 态 1b：NULL-entry 占位行收编（Task 6 义务①——Task 3 审查移交的已知死角在此闭合）：
      // 六列命中而 content_entry_id NULL 的行（外源 seed / attachment record-only 足迹 +
      // 映射扩充后该 sourceType 升格为内容类型重导）→ 走全新建分支会撞六列唯一约束
      // 23505，改为就地 UPDATE 挂 entry。语义与态 2/3 对齐：
      // - 无五列先行者 → 全新建变体（entry+revision+locale），占位行 UPDATE 挂接；
      // - 先行者 checksum 同 → 只挂接（skipped——0 新修订）；
      // - 先行者 checksum 异 → 挂接 + 新修订 + current 前移。
      // counts：sourceRecords 不计（行已在，收编非新建）；lineage 留在占位行建行 job。
      const prior = await findPriorLinkedRecord(row)
      const adoptValues = { contentEntryId: null as string | null, normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null }
      if (prior === null) {
        await db.transaction(async (tx) => {
          const entry = await tx
            .insert(contentEntries)
            .values({ siteId, typeKey, lifecycleStatus: 'active' })
            .returning({ id: contentEntries.id })
          const entryId = entry[0]!.id
          await tx
            .update(sourceRecords)
            .set({ ...adoptValues, contentEntryId: entryId })
            .where(eq(sourceRecords.id, snapRow.id))
          const revisionId = await insertRevision(tx, {
            entryId,
            row,
            data,
            checksum,
            sourceRecordId: snapRow.id,
          })
          await pointCurrent(tx, entryId, row, revisionId)
        })
        counts.contentEntries++
        counts.revisionsCreated++
        return
      }
      const entryId = prior.contentEntryId // 事务回调外的 const 收窄（属性收窄不进闭包）
      if (prior.normalizedChecksum === checksum) {
        await db
          .update(sourceRecords)
          .set({ ...adoptValues, contentEntryId: entryId })
          .where(eq(sourceRecords.id, snapRow.id))
        counts.skipped++
        return
      }
      await db.transaction(async (tx) => {
        await tx
          .update(sourceRecords)
          .set({ ...adoptValues, contentEntryId: entryId })
          .where(eq(sourceRecords.id, snapRow.id))
        const revisionId = await insertRevision(tx, {
          entryId,
          row,
          data,
          checksum,
          sourceRecordId: snapRow.id,
        })
        await pointCurrent(tx, entryId, row, revisionId)
      })
      counts.revisionsCreated++
      return
    }

    // 态 2/3 前置：内容身份（五列）解析——跨快照复用既有 entry（permalink/快照变化不建新）
    const prior = await findPriorLinkedRecord(row)
    if (prior === null) {
      // 态 3 全新建：entry + revision + locale + 快照行（事务——四行原子）
      await db.transaction(async (tx) => {
        const entry = await tx
          .insert(contentEntries)
          .values({ siteId, typeKey, lifecycleStatus: 'active' })
          .returning({ id: contentEntries.id })
        const entryId = entry[0]!.id
        const record = await tx
          .insert(sourceRecords)
          .values(recordValues(row, checksum, entryId))
          .returning({ id: sourceRecords.id })
        const revisionId = await insertRevision(tx, {
          entryId,
          row,
          data,
          checksum,
          sourceRecordId: record[0]!.id,
        })
        await tx.insert(entryLocales).values({
          siteId,
          contentEntryId: entryId,
          locale: row.locale,
          slug: row.slug ?? row.sourceId,
          currentRevisionId: revisionId,
          // published_revisionId 缺省 NULL——导入不发布（ADR-010）
        })
      })
      counts.contentEntries++
      counts.revisionsCreated++
      counts.sourceRecords++
      return
    }
    if (prior.normalizedChecksum === checksum) {
      // 态 2 跨快照无业务变化：只补本快照行（挂同一 entry），0 新修订（spec §7.5 Skip）
      await db.insert(sourceRecords).values(recordValues(row, checksum, prior.contentEntryId))
      counts.sourceRecords++
      counts.skipped++
      return
    }
    // 跨快照有变化：补本快照行 + 新修订 + current 前移（事务）
    await db.transaction(async (tx) => {
      const record = await tx
        .insert(sourceRecords)
        .values(recordValues(row, checksum, prior.contentEntryId))
        .returning({ id: sourceRecords.id })
      const revisionId = await insertRevision(tx, {
        entryId: prior.contentEntryId,
        row,
        data,
        checksum,
        sourceRecordId: record[0]!.id,
      })
      await pointCurrent(tx, prior.contentEntryId, row, revisionId)
    })
    counts.sourceRecords++
    counts.revisionsCreated++
  }

  /** source_records 行值组装（source_checksum NULL——行级原始哈希无输入源，快照级在快照行） */
  function recordValues(row: UniversalRow, checksum: string, contentEntryId: string | null) {
    return {
      siteId,
      sourceConnectionId: identity.connectionId,
      sourceSnapshotId: identity.snapshotId,
      importJobId: identity.jobId,
      sourceType: row.sourceType,
      sourceId: row.sourceId,
      locale: row.locale,
      sourceUrl: row.sourceUrl ?? null,
      rawLocation: row.rawLocation,
      normalizedChecksum: checksum,
      parseStatus: PARSE_STATUS_IMPORTED,
      contentEntryId,
    }
  }

  /**
   * record-only 行（attachment / 未知 sourceType）：只记 source_records（不建 entry）。
   * checksum 命中静默跳过（不计 skipped——媒体域计数是 Task 4 主场）；变化则快照行前移。
   */
  async function handleRecordOnlyRow(row: UniversalRow, knownType: boolean): Promise<void> {
    if (!knownType) {
      issues.push(
        `未知 sourceType：${JSON.stringify(row.sourceType)}（${row.sourceId}）不在 typeKey 映射表——只记 source_records，不建 content entry`,
      )
    }
    const checksum = normalizedContentChecksum(row, identity.versions)
    const snapRow = await findSnapshotRecord(row)
    if (snapRow !== null) {
      if (snapRow.normalizedChecksum === checksum) return
      await db
        .update(sourceRecords)
        .set({ normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null })
        .where(eq(sourceRecords.id, snapRow.id))
      return
    }
    await db.insert(sourceRecords).values(recordValues(row, checksum, null))
    counts.sourceRecords++
  }

  /** unsupported/error 事件（前置事实 12）：也写 source_records，可审计且幂等 */
  async function handleUnmappable(
    ev: Extract<AdapterEvent, { kind: 'unsupported' }> | Extract<AdapterEvent, { kind: 'error' }>,
    parseStatus: 'unsupported' | 'failed',
  ): Promise<void> {
    // postId 缺失不硬造身份——source_id 以 rawLocation 回退（行可审计、身份不虚构）
    const sourceId = ev.postId ?? ev.rawLocation
    // 分项计数按事件计（非按插入行）：unsupported/failed 的拆分只在分派点可知，
    // Task 6 报告层无法从 sourceRecords 总数重derive；重跑时行幂等但计数如实。
    if (parseStatus === 'unsupported') counts.unsupportedRecords = (counts.unsupportedRecords ?? 0) + 1
    else counts.failedRecords = (counts.failedRecords ?? 0) + 1
    const inserted = await db
      .insert(sourceRecords)
      .values({
        siteId,
        sourceConnectionId: identity.connectionId,
        sourceSnapshotId: identity.snapshotId,
        importJobId: identity.jobId,
        sourceType: ev.sourceType,
        sourceId,
        locale: identity.defaultLocale,
        rawLocation: ev.rawLocation,
        parseStatus: parseStatus === 'unsupported' ? PARSE_STATUS_UNSUPPORTED : PARSE_STATUS_FAILED,
      })
      .onConflictDoNothing()
      .returning({ id: sourceRecords.id })
    if (inserted.length > 0) counts.sourceRecords++
  }

  /** row 事件：locale 校验 → 类型分派（typeKey 映射 / attachment / 未知兜底） */
  async function handleRow(row: UniversalRow): Promise<void> {
    if (row.locale !== identity.defaultLocale) {
      issues.push(
        `row locale 不匹配站点默认（${row.sourceType}/${row.sourceId}，row.locale=${JSON.stringify(row.locale)} ≠ site.default_locale=${JSON.stringify(identity.defaultLocale)}）——跳过（前置事实 8）`,
      )
      counts.skipped++
      return
    }
    const typeKey = Object.hasOwn(CONTENT_TYPE_KEY_MAP, row.sourceType)
      ? CONTENT_TYPE_KEY_MAP[row.sourceType]
      : undefined
    if (typeKey === undefined) {
      await handleRecordOnlyRow(row, row.sourceType === 'wp:attachment')
      return
    }
    await handleContentRow(row, typeKey)
  }

  return {
    async handleEvent(ev) {
      if (ev.kind === 'row') {
        await handleRow(ev.row)
      } else if (ev.kind === 'unsupported') {
        await handleUnmappable(ev, 'unsupported')
      } else if (ev.kind === 'error') {
        await handleUnmappable(ev, 'failed')
      }
      // term 事件：内容写入器不消费（term-writer 主场；Task 6 编排各写入器各取所需）
    },
    finish() {
      return { counts, issues }
    },
  }
}

/**
 * 内容写入管道（{@link createContentEventHandler} 的自消费整流便捷壳）：单遍拉完
 * ctx.events、逐事件分派、流终返回 outcome。Task 6 编排时可改用句柄形状直接分派
 * （单消费方拉流分派给各写入器，无 tee）。
 */
export function createContentWriter(): ImportPipeline {
  return async (ctx) => {
    const handler = createContentEventHandler({
      db: ctx.db,
      siteId: ctx.siteId,
      connectionId: ctx.connectionId,
      snapshotId: ctx.wxrSnapshot.id,
      jobId: ctx.jobId,
      defaultLocale: ctx.defaultLocale,
      versions: ctx.versions,
    })
    for await (const ev of ctx.events) {
      await handler.handleEvent(ev)
    }
    return handler.finish()
  }
}