// Plan 3 Task 3 失败测试（先于实现）：内容写入器（稳定身份 + checksum skip）。
// 覆盖契约（对照计划 Task 3 与前置事实 6/8/11/12、ADR-006/010）：
// 1. mini.xml 全链（3 内容行 + 2 attachment）：3 content_entries（typeKey 映射前置事实 11）
//    + 3 entry_locales（slug/locale/current_revision_id 设、**published NULL**）+
//    3 content_revisions（source_kind='import'、checksum 按前置事实 6 公式**独立重算对拍**、
//    data={title?,slug?,status,bodyHtml,seo,knownFields,extras}）+ 5 source_records
//    （attachment 2 行 content_entry_id NULL、parse_status='imported'）；
// 2. 二导同快照（换幂等键重跑）：0 新修订 0 新 entry（checksum skip）、counts.skipped=3；
//    跨快照（变体文件新 snapshot、业务内容不变）：同样 0 新修订 0 新 entry——五列稳定
//    身份复用既有 entry（ADR-006），本快照 source_records 补行（六列身份含 snapshot_id）；
// 3. 内容变化（改 knownFields.title）：新修订 + current 前移 + published 仍 NULL
//    （ADR-010 导入不发布）+ 旧修订原样（不可变触发器——写入路径纯 INSERT 纪律）；
// 4. permalink 变化（改 sourceUrl）：**不建新 entry**（身份键不含 URL）+ 新修订
//    （sourceUrl 在 route_source_values 哈希域内）+ source_records.source_url 更新；
// 5. unsupported/error 也写 source_records（前置事实 12）：parse_status='unsupported'/
//    'failed'、locale=站点默认、content_entry_id NULL；error 无 postId → source_id 以
//    rawLocation 回退（不硬造）；
// 6. locale 校验（前置事实 8）：row.locale ≠ site.default_locale → issue + 跳过（对照行
//    证明拒收不断流）。
// 期望值全部从适配器真实产物行派生（collect 后独立组装 checksum 输入对拍，不与写入器
// 共享组装代码），不猜清洗细节。运行于 globalSetup 临时库（ADR-015）；报告根用
// os.tmpdir 临时目录，finally 清理。
import { createHash } from 'node:crypto'
import { createReadStream } from 'node:fs'
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { describe, expect, it } from 'vitest'
import {
  ADAPTER_VERSION,
  MAPPING_VERSION,
  SANITIZER_VERSION,
  runWxrAdapter,
  type AdapterEvent,
} from '@sf/adapter-wp'
import { UNIVERSAL_ROW_VERSION, type UniversalRow } from '@sf/contracts'
import { canonicalJson } from '../../apps/api/src/common/canonical.js'
import { createImportJob } from '../../apps/api/src/modules/import/import.service.js'
import { createContentWriter } from '../../apps/api/src/modules/import/content-writer.js'
import { registerSource } from '../../apps/api/src/modules/source/register.service.js'
import { rand, withImmutableCleanup, withTestDb, type Db } from './helpers.js'

const FIXTURES_ROOT = fileURLToPath(new URL('../fixtures', import.meta.url))
const WXR_FIXTURE = resolve(FIXTURES_ROOT, 'wxr', 'mini.xml')

/** 前置事实 11 的 typeKey 映射（断言侧独立持有——与实现常量对拍而非引用） */
const EXPECTED_TYPE_KEYS: Readonly<Record<string, string>> = Object.freeze({
  'wp:post': 'news_article',
  'wp:page': 'basic_page',
  'wp:product': 'product',
})

/** 手构事件流的 AsyncIterable 包装（每次调用产新迭代器——重跑各流独立） */
async function* eventsOf(events: AdapterEvent[]): AsyncGenerator<AdapterEvent, void, void> {
  yield* events
}

/** db.execute 行对象规整：postgres.js 行对象的原型差异会让 toStrictEqual 报"无视觉差异"失败——浅拷贝到普通对象 */
function plain(rows: any[]): any[] {
  return rows.map((r) => ({ ...r }))
}

/** 事件流收集（mini 全链一饮而尽；断言期望从产物行派生，不猜清洗细节） */
async function collectMiniEvents(): Promise<AdapterEvent[]> {
  const out: AdapterEvent[] = []
  for await (const ev of runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } })) {
    out.push(ev)
  }
  return out
}

/** 事件流中的 row 事件提取（UniversalRow 列表） */
function rowsOf(events: AdapterEvent[]): UniversalRow[] {
  return events.flatMap((ev) => (ev.kind === 'row' ? [ev.row] : []))
}

/** 深拷贝事件数组并对指定 sourceId 的 row 施加变异（整组克隆——两导的事件互不串扰） */
function withRowMutated(
  events: AdapterEvent[],
  sourceId: string,
  mutate: (row: UniversalRow) => void,
): AdapterEvent[] {
  return events.map((ev) => {
    const copy = structuredClone(ev)
    if (copy.kind === 'row' && copy.row.sourceId === sourceId) mutate(copy.row)
    return copy
  })
}

/**
 * 独立重算 normalized checksum（前置事实 6 公式对拍）：测试自行组装输入对象（mapped_content
 * 结构钉死 {title?,slug?,status,bodyHtml,seo,knownFields,extras}、rawLocation 显式排除、
 * 版本三常量必入），与写入器零共享代码；序列化基形用被 tests/unit/canonical.test.ts 独立
 * 钉死的 canonicalJson，哈希与 register 同法（sha256 hex）。
 */
function expectedChecksum(row: UniversalRow): string {
  const mappedContent: Record<string, unknown> = {
    status: row.status,
    seo: row.seo,
    knownFields: row.knownFields,
    extras: row.extras,
  }
  if (row.title !== undefined) mappedContent.title = row.title
  if (row.slug !== undefined) mappedContent.slug = row.slug
  if (row.bodyHtml !== undefined) mappedContent.bodyHtml = row.bodyHtml
  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: ADAPTER_VERSION,
        mapping_version: MAPPING_VERSION,
        sanitizer_version: SANITIZER_VERSION,
        locale: row.locale,
        mapped_content: mappedContent,
        taxonomy_relations: row.taxonomyRefs,
        media_reference_identities: row.attachmentRefs,
        route_source_values: routeSourceValues,
      }),
    )
    .digest('hex')
}

/** revision.data 期望结构（前置事实 6 钉死的顶层形状——与 checksum 的 mapped_content 同构） */
function expectedRevisionData(row: UniversalRow): Record<string, unknown> {
  const data: Record<string, unknown> = {
    status: row.status,
    seo: row.seo,
    knownFields: row.knownFields,
    extras: row.extras,
  }
  if (row.title !== undefined) data.title = row.title
  if (row.slug !== undefined) data.slug = row.slug
  if (row.bodyHtml !== undefined) data.bodyHtml = row.bodyHtml
  return data
}

/** jsonb 往返规整（期望侧：去 null 原型/undefined 键——与 PG 取回的普通对象同形） */
function jsonbNorm(value: unknown): Record<string, unknown> {
  return JSON.parse(JSON.stringify(value)) as Record<string, unknown>
}

/** 手构 row 事件（全必填字段的最小 UniversalRow；可选字段按 spec 传入） */
function rowEvent(spec: {
  sourceType?: string
  sourceId: string
  locale?: string
  slug?: string
  title?: string
  bodyHtml?: string
  sourceUrl?: string
}): AdapterEvent {
  const row: UniversalRow = {
    sourceType: spec.sourceType ?? 'wp:post',
    sourceId: spec.sourceId,
    status: 'publish',
    locale: spec.locale ?? 'en',
    taxonomyRefs: [],
    attachmentRefs: [],
    seo: {},
    knownFields: {},
    extras: {},
    rawLocation: `wxr://item/${spec.sourceId}`,
  }
  if (spec.slug !== undefined) row.slug = spec.slug
  if (spec.title !== undefined) row.title = spec.title
  if (spec.bodyHtml !== undefined) row.bodyHtml = spec.bodyHtml
  if (spec.sourceUrl !== undefined) row.sourceUrl = spec.sourceUrl
  return { kind: 'row', row, issues: [], sanitizeDiffs: [], changed: false }
}

/** 本文件写入的行清场：FK 反序（两张不可变表走触发器开关封装） */
async function cleanupSiteRows(db: Db, siteId: string): Promise<void> {
  await db.execute(`delete from content_terms where site_id = '${siteId}'`)
  await db.execute(`delete from terms where site_id = '${siteId}'`)
  await db.execute(`delete from taxonomies where site_id = '${siteId}'`)
  await db.execute(`delete from source_records where site_id = '${siteId}'`)
  await db.execute(`delete from entry_locales where site_id = '${siteId}'`)
  await withImmutableCleanup(db, 'content_revisions', 'content_revisions_immutable', `site_id = '${siteId}'`)
  await db.execute(`delete from content_entries where site_id = '${siteId}'`)
  await db.execute(`delete from import_jobs where site_id = '${siteId}'`)
  await withImmutableCleanup(db, 'source_snapshots', 'source_snapshots_immutable', `site_id = '${siteId}'`)
  await db.execute(`delete from source_connections where site_id = '${siteId}'`)
  await db.execute(`delete from sites where id = '${siteId}'`)
}

/** mini 登记（各用例公共前置）：register → {siteId, wxr 快照} */
async function registerMini(db: Db, work: string) {
  const key = `sfcontent-${rand()}`
  const reg = await registerSource(db, {
    siteKey: key,
    wxrPath: WXR_FIXTURE,
    allowedRoots: [FIXTURES_ROOT, work],
    storageRoot: join(work, 'storage'),
    reportRoot: join(work, 'artifacts', 'register'),
  })
  return { key, reg, wxr: reg.snapshots.find((s) => s.kind === 'wxr')! }
}

/** 导入报告读取（import.service 的报告落盘纪律：终态先行落文件） */
async function readReport(work: string, jobId: string): Promise<{
  status: string
  counts: Record<string, number>
  issues: string[]
}> {
  return JSON.parse(await readFile(join(work, 'artifacts', 'import', jobId, 'reports', 'import.json'), 'utf8'))
}

describe('内容写入器（稳定身份 + checksum skip）', () => {
  it('mini 首导全链：typeKey 映射 + entry_locales 双指针（published NULL）+ revision（checksum 独立对拍/data 钉死结构）+ 5 source_records + counts', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-'))
      try {
        const events = await collectMiniEvents()
        // 流形状 sanity：5 row（3 内容 + 2 attachment）+ 2 term + 0 unsupported/error
        expect(rowsOf(events)).toHaveLength(5)
        expect(events.filter((e) => e.kind === 'term')).toHaveLength(2)
        expect(events.filter((e) => e.kind !== 'row' && e.kind !== 'term')).toStrictEqual([])
        const contentRows = rowsOf(events).filter((r) => r.sourceType !== 'wp:attachment')
        const attachmentRows = rowsOf(events).filter((r) => r.sourceType === 'wp:attachment')
        expect(contentRows).toHaveLength(3)
        expect(attachmentRows).toHaveLength(2)

        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: createContentWriter(),
          })
          expect(result.reused).toBe(false)
          expect(result.status).toBe('succeeded')

          // content_entries：typeKey 映射（前置事实 11）+ lifecycle active
          const entries = (await db.execute(`select type_key, lifecycle_status from content_entries
            where site_id = '${reg.siteId}' order by type_key`)) as any
          expect(plain(entries)).toStrictEqual([
            { type_key: 'basic_page', lifecycle_status: 'active' },
            { type_key: 'news_article', lifecycle_status: 'active' },
            { type_key: 'product', lifecycle_status: 'active' },
          ])

          // source_records：5 行；内容行挂 entry、checksum=独立重算；attachment 行
          // content_entry_id NULL、parse_status='imported'（前置事实 11：不建 entry）
          const recs = (await db.execute(`select source_type, source_id, locale, parse_status,
            content_entry_id is null as no_entry, normalized_checksum, source_url, raw_location
            from source_records where site_id = '${reg.siteId}' and source_snapshot_id = '${wxr.id}'
            order by source_id`)) as any
          expect(recs).toHaveLength(5)
          for (const row of contentRows) {
            const rec = recs.find((r: any) => r.source_id === row.sourceId)!
            expect(rec.no_entry).toBe(false)
            expect(rec.parse_status).toBe('imported')
            expect(rec.locale).toBe('en')
            expect(rec.normalized_checksum).toBe(expectedChecksum(row))
            expect(rec.source_url).toBe(row.sourceUrl ?? null)
          }
          for (const row of attachmentRows) {
            const rec = recs.find((r: any) => r.source_id === row.sourceId)!
            expect(rec.no_entry).toBe(true)
            expect(rec.parse_status).toBe('imported')
            expect(rec.locale).toBe('en')
          }

          // revision + 双指针（经 current 指针 join）：source='import'、schema_version=
          // UNIVERSAL_ROW_VERSION、data 钉死结构、published NULL（导入不发布）
          const cur = (await db.execute(`select r.source_id, e.type_key, el.slug, el.locale as locale_of_row,
            el.current_revision_id, el.published_revision_id, cr.checksum, cr.data, cr.source_kind,
            cr.schema_version, cr.locale as locale_of_rev
            from source_records r
            join content_entries e on e.id = r.content_entry_id
            join entry_locales el on el.content_entry_id = e.id and el.locale = r.locale
            join content_revisions cr on cr.id = el.current_revision_id
            where r.site_id = '${reg.siteId}' and r.content_entry_id is not null
            order by r.source_id`)) as any
          expect(cur).toHaveLength(3)
          for (const row of contentRows) {
            const c = cur.find((x: any) => x.source_id === row.sourceId)!
            expect(c.type_key).toBe(EXPECTED_TYPE_KEYS[row.sourceType])
            expect(c.slug).toBe(row.slug)
            expect(c.locale_of_row).toBe('en')
            expect(c.locale_of_rev).toBe('en')
            expect(c.current_revision_id).toBeTruthy()
            expect(c.published_revision_id).toBe(null)
            expect(c.source_kind).toBe('import')
            expect(c.schema_version).toBe(UNIVERSAL_ROW_VERSION)
            expect(c.checksum).toBe(expectedChecksum(row))
            expect(jsonbNorm(c.data)).toStrictEqual(jsonbNorm(expectedRevisionData(row)))
          }

          // 报告：counts 四键 + issues 空（正常路径）
          const report = await readReport(work, result.jobId)
          expect(report.status).toBe('succeeded')
          expect(report.counts).toStrictEqual({
            contentEntries: 3,
            revisionsCreated: 3,
            sourceRecords: 5,
            skipped: 0,
          })
          expect(report.issues).toStrictEqual([])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('二导同快照（换幂等键）：0 新修订 0 新 entry（checksum skip）、counts.skipped=3', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-'))
      try {
        const events = await collectMiniEvents()
        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const base = {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            pipeline: createContentWriter(),
          }
          const first = await createImportJob(db, { ...base, events: eventsOf(events) })
          expect(first.status).toBe('succeeded')

          // 同流重跑：显式幂等键（默认键会复用既有 job 不重跑——Task 1 语义，此处钉写入器幂等）
          const again = await createImportJob(db, {
            ...base,
            idempotencyKey: `sfcontent-rerun-${rand()}`,
            events: eventsOf(events),
          })
          expect(again.reused).toBe(false)
          expect(again.status).toBe('succeeded')

          // 全表不增行
          const after = (await db.execute(`select
            (select count(*)::int from content_entries where site_id = '${reg.siteId}') as entries,
            (select count(*)::int from entry_locales where site_id = '${reg.siteId}') as locales,
            (select count(*)::int from content_revisions where site_id = '${reg.siteId}') as revisions,
            (select count(*)::int from source_records where site_id = '${reg.siteId}') as records`)) as any
          expect(plain(after)[0]).toStrictEqual({ entries: 3, locales: 3, revisions: 3, records: 5 })

          // skipped=3：三内容行 checksum 命中（attachment 命中不计——媒体域是 Task 4 主场）
          const report = await readReport(work, again.jobId)
          expect(report.counts).toStrictEqual({
            contentEntries: 0,
            revisionsCreated: 0,
            sourceRecords: 0,
            skipped: 3,
          })
          expect(report.issues).toStrictEqual([])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('跨快照（变体文件、业务内容不变）：0 新修订 0 新 entry（五列稳定身份）+ 本快照 source_records 补行', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-xsnap-'))
      try {
        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const first = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }),
            pipeline: createContentWriter(),
          })
          expect(first.status).toBe('succeeded')

          // 变体文件：mini 复制 + 通道内注释（字节不同 → 新 snapshot；item 业务内容不变）
          const variantPath = join(work, 'mini-v2.xml')
          const xml = await readFile(WXR_FIXTURE, 'utf8')
          await writeFile(variantPath, xml.replace('<channel>', '<channel>\n  <!-- v2: byte-different, business-identical -->'))
          const reg2 = await registerSource(db, {
            siteKey: key,
            wxrPath: variantPath,
            allowedRoots: [FIXTURES_ROOT, work],
            storageRoot: join(work, 'storage'),
            reportRoot: join(work, 'artifacts', 'register'),
          })
          const wxr2 = reg2.snapshots.find((s) => s.kind === 'wxr')!
          expect(wxr2.id).not.toBe(wxr.id)
          expect(reg2.connectionId).toBe(reg.connectionId) // 同连接——五列稳定身份的前提

          const second = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr2.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: runWxrAdapter(createReadStream(variantPath), { mapConfig: { locale: 'en' } }),
            pipeline: createContentWriter(),
          })
          expect(second.status).toBe('succeeded')

          // 0 新 entry、0 新修订；source_records 每快照各行（六列身份含 snapshot_id）：5+5
          const after = (await db.execute(`select
            (select count(*)::int from content_entries where site_id = '${reg.siteId}') as entries,
            (select count(*)::int from content_revisions where site_id = '${reg.siteId}') as revisions,
            (select count(*)::int from source_records where site_id = '${reg.siteId}') as records`)) as any
          expect(plain(after)[0]).toStrictEqual({ entries: 3, revisions: 3, records: 10 })

          // 新快照的内容行挂**同一批 entry**（稳定身份跨快照复用）
          const linkage = (await db.execute(`select r.source_id, r.source_snapshot_id, r.content_entry_id
            from source_records r
            where r.site_id = '${reg.siteId}' and r.source_snapshot_id in ('${wxr.id}', '${wxr2.id}')
              and r.content_entry_id is not null
            order by r.source_id, r.source_snapshot_id`)) as any
          expect(linkage).toHaveLength(6)
          for (const sourceId of ['101', '111', '121']) {
            const pair = linkage.filter((l: any) => l.source_id === sourceId)
            expect(pair).toHaveLength(2)
            expect(pair[0].content_entry_id).toBe(pair[1].content_entry_id)
          }

          const report = await readReport(work, second.jobId)
          expect(report.counts).toStrictEqual({
            contentEntries: 0,
            revisionsCreated: 0,
            sourceRecords: 5,
            skipped: 3,
          })
          expect(report.issues).toStrictEqual([])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('内容变化（改 knownFields.title）：新修订 + current 前移 + published 仍 NULL（ADR-010）+ 旧修订原样', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-chg-'))
      try {
        const events = await collectMiniEvents()
        const row101 = rowsOf(events).find((r) => r.sourceId === '101')!
        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const base = {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            pipeline: createContentWriter(),
          }
          const first = await createImportJob(db, { ...base, events: eventsOf(events) })
          expect(first.status).toBe('succeeded')
          const before = (await db.execute(`select el.content_entry_id as entry, el.current_revision_id as cur,
            el.published_revision_id as pub
            from entry_locales el join source_records r on r.content_entry_id = el.content_entry_id
            where r.site_id = '${reg.siteId}' and r.source_id = '101'`)) as any
          expect(before).toHaveLength(1)
          expect(before[0].pub).toBe(null)

          // 手构内容变化：WXR 标题变更在适配器产物上是 knownFields.title 变化（mapItem 提升
          // 纪律：title 留在 knownFields）
          const changed = withRowMutated(events, '101', (row) => {
            row.knownFields.title = 'Fixture News A (edited)'
          })
          const changed101 = rowsOf(changed).find((r) => r.sourceId === '101')!
          expect(changed101.knownFields.title).not.toBe(row101.knownFields.title)
          const second = await createImportJob(db, {
            ...base,
            idempotencyKey: `sfcontent-chg-${rand()}`,
            events: eventsOf(changed),
          })
          expect(second.status).toBe('succeeded')

          // current 前移、published 仍 NULL（导入不发布）、同 entry（不建新）
          const after = (await db.execute(`select el.content_entry_id as entry, el.current_revision_id as cur,
            el.published_revision_id as pub
            from entry_locales el join source_records r on r.content_entry_id = el.content_entry_id
            where r.site_id = '${reg.siteId}' and r.source_id = '101'`)) as any
          expect(after[0].entry).toBe(before[0].entry)
          expect(after[0].cur).not.toBe(before[0].cur)
          expect(after[0].pub).toBe(null)

          // 该 entry 两修订（旧原样——不可变触发器下写入路径纯 INSERT；新旧 checksum 各归位）
          const revs = (await db.execute(`select checksum from content_revisions
            where site_id = '${reg.siteId}' and content_entry_id = '${before[0].entry}'
            order by created_at`)) as any
          expect(plain(revs)).toStrictEqual([
            { checksum: expectedChecksum(row101) },
            { checksum: expectedChecksum(changed101) },
          ])
          expect((await db.execute(`select count(*)::int as n from content_entries
            where site_id = '${reg.siteId}'`)) as any).toMatchObject([{ n: 3 }])

          // 其余两行未变 → skipped=2；一行新修订
          const report = await readReport(work, second.jobId)
          expect(report.counts).toStrictEqual({
            contentEntries: 0,
            revisionsCreated: 1,
            sourceRecords: 0,
            skipped: 2,
          })
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('permalink 变化（改 sourceUrl）：不建新 entry（身份键不含 URL）+ 新修订（sourceUrl 在哈希内）+ source_url 更新', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-perma-'))
      try {
        const events = await collectMiniEvents()
        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const base = {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            pipeline: createContentWriter(),
          }
          const first = await createImportJob(db, { ...base, events: eventsOf(events) })
          expect(first.status).toBe('succeeded')
          const before = (await db.execute(`select r.content_entry_id as entry, r.source_url
            from source_records r
            where r.site_id = '${reg.siteId}' and r.source_id = '101'`)) as any

          // 手构 permalink 变化：sourceUrl 进 route_source_values 哈希域（前置事实 6）
          const newUrl = 'https://fixture.example/news/fixture-news-a-renamed.html'
          const changed = withRowMutated(events, '101', (row) => {
            row.sourceUrl = newUrl
          })
          const changed101 = rowsOf(changed).find((r) => r.sourceId === '101')!
          const second = await createImportJob(db, {
            ...base,
            idempotencyKey: `sfcontent-perma-${rand()}`,
            events: eventsOf(changed),
          })
          expect(second.status).toBe('succeeded')

          // 不建新 entry（count 仍 3）；source_records 同行 source_url 前移 + checksum 前移
          const entries = (await db.execute(`select count(*)::int as n from content_entries
            where site_id = '${reg.siteId}'`)) as any
          expect(entries[0].n).toBe(3)
          const rec101 = (await db.execute(`select content_entry_id, source_url, normalized_checksum
            from source_records where site_id = '${reg.siteId}' and source_id = '101'`)) as any
          expect(rec101[0].content_entry_id).toBe(before[0].entry) // 同 entry——稳定身份
          expect(rec101[0].source_url).toBe(newUrl)
          expect(rec101[0].normalized_checksum).toBe(expectedChecksum(changed101))

          // 新修订（permalink 产新修订不建新 entry）；其余两行 skipped
          const revs = (await db.execute(`select count(*)::int as n from content_revisions
            where site_id = '${reg.siteId}' and content_entry_id = '${before[0].entry}'`)) as any
          expect(revs[0].n).toBe(2)
          const report = await readReport(work, second.jobId)
          expect(report.counts).toStrictEqual({
            contentEntries: 0,
            revisionsCreated: 1,
            sourceRecords: 0,
            skipped: 2,
          })
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('unsupported/error 也写 source_records（前置事实 12）：parse_status/locale/无 entry；error 无 postId 以 rawLocation 回退', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-unsup-'))
      try {
        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const events: AdapterEvent[] = [
            { kind: 'unsupported', sourceType: 'wp:nav_menu_item', reason: 'post_type 未配置映射', rawLocation: 'wxr://item/7', postId: '68' },
            { kind: 'error', sourceType: 'wp:post', issue: 'post_id 缺失或非法', rawLocation: 'wxr://item/8' },
          ]
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: createContentWriter(),
          })
          expect(result.status).toBe('succeeded')

          // 两行：locale=站点默认、content_entry_id NULL；error 无 postId → source_id 用
          // rawLocation 回退（不硬造）
          const recs = (await db.execute(`select source_type, source_id, locale, parse_status,
            content_entry_id is null as no_entry
            from source_records where site_id = '${reg.siteId}' order by source_id`)) as any
          expect(plain(recs)).toStrictEqual([
            { source_type: 'wp:nav_menu_item', source_id: '68', locale: 'en', parse_status: 'unsupported', no_entry: true },
            { source_type: 'wp:post', source_id: 'wxr://item/8', locale: 'en', parse_status: 'failed', no_entry: true },
          ])
          // 零 entry 零修订
          expect(((await db.execute(`select count(*)::int as n from content_entries
            where site_id = '${reg.siteId}'`)) as any)[0].n).toBe(0)
          expect(((await db.execute(`select count(*)::int as n from content_revisions
            where site_id = '${reg.siteId}'`)) as any)[0].n).toBe(0)

          const report = await readReport(work, result.jobId)
          expect(report.counts).toStrictEqual({
            contentEntries: 0,
            revisionsCreated: 0,
            sourceRecords: 2,
            skipped: 0,
            unsupportedRecords: 1,
            failedRecords: 1,
          })
          // unsupported/error 是"有明确结果"的预期输入（source_records 行即审计），非 issue
          expect(report.issues).toStrictEqual([])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('locale 校验（前置事实 8）：row.locale ≠ site.default_locale → issue + 跳过（对照行不断流）', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-content-locale-'))
      try {
        const { key, reg, wxr } = await registerMini(db, work)
        try {
          const events: AdapterEvent[] = [
            rowEvent({ sourceId: '301', locale: 'fr', slug: 'post-301', bodyHtml: '<p>bonjour</p>' }),
            rowEvent({ sourceId: '302', slug: 'post-302', bodyHtml: '<p>ok</p>' }),
          ]
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: createContentWriter(),
          })
          expect(result.status).toBe('succeeded_with_issues')

          // 只落对照行 302（en）；301 无任何行
          const recs = (await db.execute(`select source_id from source_records
            where site_id = '${reg.siteId}' order by source_id`)) as any
          expect(plain(recs)).toStrictEqual([{ source_id: '302' }])
          expect(((await db.execute(`select count(*)::int as n from content_entries
            where site_id = '${reg.siteId}'`)) as any)[0].n).toBe(1)

          const report = await readReport(work, result.jobId)
          expect(report.counts).toStrictEqual({
            contentEntries: 1,
            revisionsCreated: 1,
            sourceRecords: 1,
            skipped: 1,
          })
          expect(report.issues).toHaveLength(1)
          expect(report.issues[0]).toContain('301')
          expect(report.issues[0]).toContain('fr')
          expect(report.issues[0]).toContain('en')
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })
})