cms-website/tests/integration/import-terms.test.ts

// Plan 3 Task 2 失败测试（先于实现）：term 物化（含自建缺失）。
// 覆盖契约（对照计划 Task 2 与前置事实 1/9、phase2-smoke 移交项①）：
// 1. mini.xml 全链：wp:term 事件 → taxonomies（key→object_type 映射 + public_base_path
//    前置事实 9 常量）+ terms（source_id=term_id 字符串）+ content_terms（entry↔term，
//    is_primary 近似）。mini 文档序 items 先于 wp:term——row 先到时按 nicename 自建、
//    wp:term 后到**收编**（source_id/name 就地扶正，无 auto: 残留），终态与 term 先行序
//    一致（顺序无关性）；
// 2. 自建缺失（phase2-smoke：真实 WXR 的 wp:term 表缺 category/post_tag 整域）：row 的
//    taxonomyRefs (domain, nicename) 无对应 term → 自动建（source_id=auto:<nicename>、
//    slug=nicename、name=nicename——UniversalRow 不携带 <category> 文本，实现层 JSDoc 注明）；
//    同 nicename 多次引用只建一条（应用层索引 + UNIQUE(site,taxonomy,source_id) 兜底）；
// 3. content_terms：is_primary 每 (entry, taxonomy) 首个 true 余 false（WP 单主分类近似）；
//    row 无对应 content entry → term 仍物化、关联跳过 + issue（不静默丢弃）；
// 4. counts：termsProcessed / autoCreatedTerms / contentTermsCreated；正常路径 issues 空；
// 5. 幂等：同流重跑（换幂等键新 job）taxonomies/terms/content_terms 均不增行。
// 运行于 globalSetup 临时库（ADR-015）；报告根用 os.tmpdir 临时目录，finally 清理。
// content entry 由测试内 seed 模拟 Task 3 内容写入器的最小落库足迹——source_records.
// content_entry_id 是 term 关联的消费接缝（Task 6 编排时由内容写入器保证存在）。
import { createReadStream } from 'node:fs'
import { mkdtemp, readFile, rm } 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 { runWxrAdapter, type AdapterEvent } from '@sf/adapter-wp'
import type { UniversalRow } from '@sf/contracts'
import type { ImportJobContext, ImportPipeline } from '../../apps/api/src/modules/import/import.service.js'
import { createImportJob } from '../../apps/api/src/modules/import/import.service.js'
import { createTermWriter } from '../../apps/api/src/modules/import/term-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')

/** 手构事件流的 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 }))
}

/** 手构 row 事件（全必填字段的最小 UniversalRow） */
function rowEvent(spec: {
  sourceType?: string
  sourceId: string
  taxonomyRefs: Array<{ taxonomy: string; sourceTermId: string }>
}): AdapterEvent {
  const row: UniversalRow = {
    sourceType: spec.sourceType ?? 'wp:post',
    sourceId: spec.sourceId,
    status: 'publish',
    locale: 'en',
    taxonomyRefs: spec.taxonomyRefs,
    attachmentRefs: [],
    seo: {},
    knownFields: {},
    extras: {},
    rawLocation: `wxr://item/${spec.sourceId}`,
  }
  return { kind: 'row', row, issues: [], sanitizeDiffs: [], changed: false }
}

/** 手构 wp:term 事件 */
function termEvent(spec: { termId: number | null; taxonomy: string; slug: string; name: string }): AdapterEvent {
  return {
    kind: 'term',
    term: {
      rawLocation: `wxr://term/${spec.slug}`,
      termId: spec.termId,
      taxonomy: spec.taxonomy,
      slug: spec.slug,
      name: spec.name,
    },
  }
}

/**
 * 模拟 Task 3 内容写入器的最小落库足迹：content_entries + source_records（含
 * content_entry_id 接缝列）。幂等（identity 命中即复用）——重跑场景同流再 seed 不增行。
 */
async function seedContentEntry(
  ctx: ImportJobContext,
  spec: { sourceType: string; sourceId: string; typeKey: string; locale?: string },
): Promise<void> {
  const locale = spec.locale ?? 'en'
  const found = (await ctx.db.execute(`select content_entry_id from source_records
    where site_id = '${ctx.siteId}' and source_connection_id = '${ctx.connectionId}'
      and source_type = '${spec.sourceType}' and source_id = '${spec.sourceId}' and locale = '${locale}'
      and content_entry_id is not null
    order by created_at desc limit 1`)) as any
  if (found.length > 0) return
  const entry = (await ctx.db.execute(`insert into content_entries (id, site_id, type_key, lifecycle_status)
    values (gen_random_uuid(), '${ctx.siteId}', '${spec.typeKey}', 'active') returning id`)) as any
  await ctx.db.execute(`insert into source_records (id, site_id, source_connection_id, source_snapshot_id, import_job_id,
    source_type, source_id, locale, raw_location, parse_status, content_entry_id)
    values (gen_random_uuid(), '${ctx.siteId}', '${ctx.connectionId}', '${ctx.wxrSnapshot.id}', '${ctx.jobId}',
      '${spec.sourceType}', '${spec.sourceId}', '${locale}', 'wxr://item/${spec.sourceId}', 'mapped', '${entry[0].id}')`)
}

/** 测试管道：先 seed（模拟内容写入器足迹）再跑 term 写入器——Task 6 编排的雏形 */
function seededTermWriter(seed: (ctx: ImportJobContext) => Promise<void>): ImportPipeline {
  return async (ctx) => {
    await seed(ctx)
    return createTermWriter()(ctx)
  }
}

/** 本文件写入的行清场：FK 反序（terms 自引用单语句全删安全；快照不可变走触发器开关） */
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 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 三内容行的 seed（前置事实 11 的 typeKey 映射：wp:post→news_article、wp:page→basic_page、wp:product→product） */
async function seedMiniEntries(ctx: ImportJobContext): Promise<void> {
  await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '101', typeKey: 'news_article' })
  await seedContentEntry(ctx, { sourceType: 'wp:product', sourceId: '111', typeKey: 'product' })
  await seedContentEntry(ctx, { sourceType: 'wp:page', sourceId: '121', typeKey: 'basic_page' })
}

describe('term 物化（含自建缺失）', () => {
  it('mini.xml 全链：taxonomies 常量映射 + terms(source_id=term_id) + content_terms(is_primary) + 收编（row 先到 wp:term 后到）+ counts/issues', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-'))
      try {
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(work, 'storage'),
          reportRoot: join(work, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }),
            pipeline: seededTermWriter(seedMiniEntries),
          })
          expect(result.reused).toBe(false)
          expect(result.status).toBe('succeeded')

          // taxonomies：key→object_type 映射 + 前置事实 9 常量（public_base_path 仅记录）
          const tax = (await db.execute(`select key, object_type, public_base_path from taxonomies
            where site_id = '${reg.siteId}' order by key`)) as any
          expect(plain(tax)).toStrictEqual([
            { key: 'category', object_type: 'post', public_base_path: '/category/' },
            { key: 'product_cat', object_type: 'product', public_base_path: '/product-category/' },
          ])

          // terms：source_id=term_id 字符串；mini 文档序 items 先于 wp:term——post 101 的
          // category/company 引用先到（自建 auto:company），wp:term 31 后到收编扶正
          const termsRows = (await db.execute(`select t.source_id, t.slug, t.name, x.key as taxonomy
            from terms t join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}' order by x.key`)) as any
          expect(plain(termsRows)).toStrictEqual([
            { source_id: '31', slug: 'company', name: 'Company', taxonomy: 'category' },
            { source_id: '32', slug: 'pipe-fittings', name: 'Pipe Fittings', taxonomy: 'product_cat' },
          ])
          const autoLeft = (await db.execute(`select count(*)::int as n from terms
            where site_id = '${reg.siteId}' and source_id like 'auto:%'`)) as any
          expect(autoLeft[0].n).toBe(0)

          // content_terms：post 101 ↔ category/company（单引用即首引用，is_primary=true）
          const ct = (await db.execute(`select content_entry_id, term_id, is_primary from content_terms
            where site_id = '${reg.siteId}'`)) as any
          expect(ct).toHaveLength(1)
          const entry101 = (await db.execute(`select content_entry_id from source_records
            where site_id = '${reg.siteId}' and source_type = 'wp:post' and source_id = '101'`)) as any
          const term31 = (await db.execute(`select id from terms
            where site_id = '${reg.siteId}' and source_id = '31'`)) as any
          expect(ct[0].content_entry_id).toBe(entry101[0].content_entry_id)
          expect(ct[0].term_id).toBe(term31[0].id)
          expect(ct[0].is_primary).toBe(true)

          // 报告：counts 三键（自建终态 0——瞬态自建已被收编）+ issues 空（正常路径）
          const report = JSON.parse(
            await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { status: string; counts: Record<string, number>; issues: string[] }
          expect(report.status).toBe('succeeded')
          expect(report.counts).toStrictEqual({
            termsProcessed: 2,
            autoCreatedTerms: 0,
            contentTermsCreated: 1,
          })
          expect(report.issues).toStrictEqual([])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('自建缺失 term（手构缺表场景）：auto: 前缀 + 同 nicename 单条 + is_primary 规则 + 无 entry 跳过关联 + 未知 taxonomy 兜底', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-'))
      try {
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(work, 'storage'),
          reportRoot: join(work, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          // 缺表场景：category/post_tag 无任何 wp:term 事件（照真实 hesupipe WXR）；
          // nav_menu 是表外 taxonomy（wp:term 有、常量表无——兜底映射钉行为）
          const events: AdapterEvent[] = [
            rowEvent({ sourceId: '501', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company-news' }] }),
            rowEvent({ sourceId: '502', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company-news' }] }),
            rowEvent({
              sourceId: '503',
              taxonomyRefs: [
                { taxonomy: 'category', sourceTermId: 'news-a' },
                { taxonomy: 'category', sourceTermId: 'news-b' },
              ],
            }),
            rowEvent({
              sourceType: 'wp:product',
              sourceId: '504',
              taxonomyRefs: [
                { taxonomy: 'product_cat', sourceTermId: 'pc-1' },
                { taxonomy: 'product_tag', sourceTermId: 'pt-1' },
              ],
            }),
            // 505 故意不 seed entry：term 仍物化、关联跳过 + issue
            rowEvent({ sourceId: '505', taxonomyRefs: [{ taxonomy: 'post_tag', sourceTermId: 'orphan-tag' }] }),
            termEvent({ termId: 77, taxonomy: 'nav_menu', slug: 'mainmenu', name: 'Main Menu' }),
          ]
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: seededTermWriter(async (ctx) => {
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '502', typeKey: 'news_article' })
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '503', typeKey: 'news_article' })
              await seedContentEntry(ctx, { sourceType: 'wp:product', sourceId: '504', typeKey: 'product' })
            }),
          })
          expect(result.status).toBe('succeeded_with_issues')

          // 自建 terms：source_id=auto:<nicename>、slug=nicename、name=nicename；
          // company-news 两处引用只建一条（应用层索引；UNIQUE(site,taxonomy,source_id) 兜底）
          const autoTerms = (await db.execute(`select t.source_id, t.slug, t.name, x.key as taxonomy
            from terms t join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}' and t.source_id like 'auto:%' order by t.source_id`)) as any
          expect(plain(autoTerms)).toStrictEqual([
            { source_id: 'auto:company-news', slug: 'company-news', name: 'company-news', taxonomy: 'category' },
            { source_id: 'auto:news-a', slug: 'news-a', name: 'news-a', taxonomy: 'category' },
            { source_id: 'auto:news-b', slug: 'news-b', name: 'news-b', taxonomy: 'category' },
            { source_id: 'auto:orphan-tag', slug: 'orphan-tag', name: 'orphan-tag', taxonomy: 'post_tag' },
            { source_id: 'auto:pc-1', slug: 'pc-1', name: 'pc-1', taxonomy: 'product_cat' },
            { source_id: 'auto:pt-1', slug: 'pt-1', name: 'pt-1', taxonomy: 'product_tag' },
          ])

          // 未知 taxonomy 兜底（非静默）：nav_menu → object_type 'post'、/nav_menu/ + issue
          const navTax = (await db.execute(`select object_type, public_base_path from taxonomies
            where site_id = '${reg.siteId}' and key = 'nav_menu'`)) as any
          expect(plain(navTax)).toStrictEqual([{ object_type: 'post', public_base_path: '/nav_menu/' }])
          const navTerm = (await db.execute(`select t.source_id, t.name from terms t
            join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}' and x.key = 'nav_menu'`)) as any
          expect(plain(navTerm)).toStrictEqual([{ source_id: '77', name: 'Main Menu' }])

          // content_terms：is_primary 每 (entry, taxonomy) 首个 true 余 false；
          // 505 无 entry → 不关联（term 已物化）
          const ct = (await db.execute(`select r.source_id as row_id, x.key as taxonomy, t.slug, c.is_primary
            from content_terms c
            join source_records r on r.content_entry_id = c.content_entry_id
            join terms t on t.id = c.term_id
            join taxonomies x on x.id = t.taxonomy_id
            where c.site_id = '${reg.siteId}' order by r.source_id, t.slug`)) as any
          expect(plain(ct)).toStrictEqual([
            { row_id: '501', taxonomy: 'category', slug: 'company-news', is_primary: true },
            { row_id: '502', taxonomy: 'category', slug: 'company-news', is_primary: true },
            { row_id: '503', taxonomy: 'category', slug: 'news-a', is_primary: true },
            { row_id: '503', taxonomy: 'category', slug: 'news-b', is_primary: false },
            { row_id: '504', taxonomy: 'product_cat', slug: 'pc-1', is_primary: true },
            { row_id: '504', taxonomy: 'product_tag', slug: 'pt-1', is_primary: true },
          ])

          // 报告：自建 6（orphan-tag 计入——物化不依赖 entry）；两条 issue（505 无 entry + nav_menu 兜底）
          const report = JSON.parse(
            await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { status: string; counts: Record<string, number>; issues: string[] }
          expect(report.counts).toStrictEqual({
            termsProcessed: 1,
            autoCreatedTerms: 6,
            contentTermsCreated: 6,
          })
          expect(report.issues).toHaveLength(2)
          expect(report.issues.some((i) => i.includes('505') && i.includes('content entry'))).toBe(true)
          expect(report.issues.some((i) => i.includes('nav_menu'))).toBe(true)
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('顺序无关性（term 先行序）：wp:term 先到 → row 后到直连既有 term，零自建、终态与 row 先行序一致', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-'))
      try {
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(work, 'storage'),
          reportRoot: join(work, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          const events: AdapterEvent[] = [
            termEvent({ termId: 31, taxonomy: 'category', slug: 'company', name: 'Company' }),
            termEvent({ termId: 88, taxonomy: 'product_tag', slug: 'special', name: 'Special' }),
            rowEvent({ sourceId: '101', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }] }),
            rowEvent({
              sourceType: 'wp:product',
              sourceId: '111',
              taxonomyRefs: [{ taxonomy: 'product_tag', sourceTermId: 'special' }],
            }),
          ]
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: seededTermWriter(async (ctx) => {
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '101', typeKey: 'news_article' })
              await seedContentEntry(ctx, { sourceType: 'wp:product', sourceId: '111', typeKey: 'product' })
            }),
          })
          expect(result.status).toBe('succeeded')

          // 与 mini 序（row 先到 + 收编）同一终态：source_id=term_id、name=wp:term 权威名
          const termsRows = (await db.execute(`select t.source_id, t.slug, t.name, x.key as taxonomy
            from terms t join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}' order by x.key`)) as any
          expect(plain(termsRows)).toStrictEqual([
            { source_id: '31', slug: 'company', name: 'Company', taxonomy: 'category' },
            { source_id: '88', slug: 'special', name: 'Special', taxonomy: 'product_tag' },
          ])
          const taxRows = (await db.execute(`select key, object_type, public_base_path from taxonomies
            where site_id = '${reg.siteId}' order by key`)) as any
          expect(plain(taxRows)).toStrictEqual([
            { key: 'category', object_type: 'post', public_base_path: '/category/' },
            { key: 'product_tag', object_type: 'product', public_base_path: '/product-tag/' },
          ])

          // term 先行序全程零自建；两条关联均挂真实 term
          const report = JSON.parse(
            await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { status: string; counts: Record<string, number>; issues: string[] }
          expect(report.counts).toStrictEqual({
            termsProcessed: 2,
            autoCreatedTerms: 0,
            contentTermsCreated: 2,
          })
          expect(report.issues).toStrictEqual([])
          const ct = (await db.execute(`select r.source_id as row_id, t.source_id as term_source_id, c.is_primary
            from content_terms c
            join source_records r on r.content_entry_id = c.content_entry_id
            join terms t on t.id = c.term_id
            where c.site_id = '${reg.siteId}' order by r.source_id`)) as any
          expect(plain(ct)).toStrictEqual([
            { row_id: '101', term_source_id: '31', is_primary: true },
            { row_id: '111', term_source_id: '88', is_primary: true },
          ])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('幂等重跑（换幂等键新 job 同流）：taxonomies/terms/content_terms 均不增行、关联零新增', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-'))
      try {
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(work, 'storage'),
          reportRoot: join(work, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          const base = {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            pipeline: seededTermWriter(seedMiniEntries),
          }
          const first = await createImportJob(db, {
            ...base,
            events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }),
          })
          expect(first.status).toBe('succeeded')
          const before = (await db.execute(`select
            (select count(*)::int from taxonomies where site_id = '${reg.siteId}') as taxonomies,
            (select count(*)::int from terms where site_id = '${reg.siteId}') as terms,
            (select count(*)::int from content_terms where site_id = '${reg.siteId}') as content_terms`)) as any
          expect(plain(before)[0]).toStrictEqual({ taxonomies: 2, terms: 2, content_terms: 1 })

          // 同流重跑：显式幂等键（默认键会复用既有 job 不重跑——那是 Task 1 语义，此处钉写入器幂等）
          const again = await createImportJob(db, {
            ...base,
            idempotencyKey: `sfterm-rerun-${rand()}`,
            events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }),
          })
          expect(again.reused).toBe(false)
          expect(again.status).toBe('succeeded')
          const after = (await db.execute(`select
            (select count(*)::int from taxonomies where site_id = '${reg.siteId}') as taxonomies,
            (select count(*)::int from terms where site_id = '${reg.siteId}') as terms,
            (select count(*)::int from content_terms where site_id = '${reg.siteId}') as content_terms`)) as any
          expect(plain(after)[0]).toStrictEqual({ taxonomies: 2, terms: 2, content_terms: 1 })

          // 重跑计数：事件仍处理（termsProcessed=2），但物化/关联零新增
          const report = JSON.parse(
            await readFile(join(work, 'artifacts', 'import', again.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { counts: Record<string, number>; issues: string[] }
          expect(report.counts).toStrictEqual({
            termsProcessed: 2,
            autoCreatedTerms: 0,
            contentTermsCreated: 0,
          })
          expect(report.issues).toStrictEqual([])
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('null-id wp:term 双序等价：name-only 收编——term 先行/row 先行终态 name 与 issue 双序一致', async () => {
    await withTestDb(async (db) => {
      // 两序各起独立站点（终态对比不受他序残留影响）；work 根共用、子目录隔离
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-nullid-'))
      const runScenario = async (order: 'term-first' | 'row-first') => {
        const dir = join(work, order)
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(dir, 'storage'),
          reportRoot: join(dir, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          const nullIdTerm = termEvent({
            termId: null,
            taxonomy: 'category',
            slug: 'company-news',
            name: 'Company News',
          })
          const row = rowEvent({
            sourceId: '501',
            taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company-news' }],
          })
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(dir, 'artifacts', 'import'),
            events: eventsOf(order === 'term-first' ? [nullIdTerm, row] : [row, nullIdTerm]),
            pipeline: seededTermWriter(async (ctx) => {
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
            }),
          })
          expect(result.status).toBe('succeeded_with_issues')
          const termsRows = (await db.execute(`select t.source_id, t.slug, t.name
            from terms t join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}'`)) as any
          const ct = (await db.execute(`select r.source_id as row_id, t.slug, c.is_primary
            from content_terms c
            join source_records r on r.content_entry_id = c.content_entry_id
            join terms t on t.id = c.term_id
            where c.site_id = '${reg.siteId}'`)) as any
          const report = JSON.parse(
            await readFile(join(dir, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { issues: string[] }
          return { terms: plain(termsRows), ct: plain(ct), issues: report.issues }
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      }
      try {
        const termFirst = await runScenario('term-first')
        const rowFirst = await runScenario('row-first')
        // 双序终态一致：单行 auto:company-news、name 为 wp:term 权威名（name-only 收编）
        const expectedTerms = [
          { source_id: 'auto:company-news', slug: 'company-news', name: 'Company News' },
        ]
        expect(termFirst.terms).toStrictEqual(expectedTerms)
        expect(rowFirst.terms).toStrictEqual(expectedTerms)
        expect(termFirst.ct).toStrictEqual(rowFirst.ct)
        expect(termFirst.ct).toStrictEqual([
          { row_id: '501', slug: 'company-news', is_primary: true },
        ])
        // issue 双序均在（缺 term_id 的 wp:term 事件无论先后到都报告）
        expect(termFirst.issues).toHaveLength(1)
        expect(rowFirst.issues).toHaveLength(1)
        expect(termFirst.issues[0]).toMatch(/缺 term_id/)
        expect(rowFirst.issues[0]).toMatch(/缺 term_id/)
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('同 slug 双 term_id：文档序 first-wins——保留先到 id、后到跳过并报告，关联绑保留行、name 不被覆盖', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-dualid-'))
      try {
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(work, 'storage'),
          reportRoot: join(work, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          const events: AdapterEvent[] = [
            termEvent({ termId: 31, taxonomy: 'category', slug: 'company', name: 'Company' }),
            termEvent({ termId: 99, taxonomy: 'category', slug: 'company', name: 'Company Dup' }),
            rowEvent({ sourceId: '501', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }] }),
          ]
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: seededTermWriter(async (ctx) => {
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
            }),
          })
          expect(result.status).toBe('succeeded_with_issues')

          // 单行：先到的 31 保留（name 不被后到 99 的事件覆盖——first-wins）
          const termsRows = (await db.execute(`select t.source_id, t.slug, t.name
            from terms t join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}'`)) as any
          expect(plain(termsRows)).toStrictEqual([
            { source_id: '31', slug: 'company', name: 'Company' },
          ])

          // issue：保留/跳过双 id + rawLocation 均可审计
          const report = JSON.parse(
            await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { counts: Record<string, number>; issues: string[] }
          expect(report.issues).toHaveLength(1)
          expect(report.issues[0]).toMatch(/同 slug 双 term_id：保留 31、跳过 99/)

          // 关联绑保留行（31）；计数零自建
          const ct = (await db.execute(`select r.source_id as row_id, t.source_id as term_source_id, c.is_primary
            from content_terms c
            join source_records r on r.content_entry_id = c.content_entry_id
            join terms t on t.id = c.term_id
            where c.site_id = '${reg.siteId}'`)) as any
          expect(plain(ct)).toStrictEqual([
            { row_id: '501', term_source_id: '31', is_primary: true },
          ])
          expect(report.counts).toStrictEqual({
            termsProcessed: 2,
            autoCreatedTerms: 0,
            contentTermsCreated: 1,
          })
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })

  it('收编撞唯一约束：source_id 已被他行占用 → 自建行原样保留 + issue（写入器唯一 try/catch 路径）', async () => {
    await withTestDb(async (db) => {
      const work = await mkdtemp(join(tmpdir(), 'sf-terms-adopt-'))
      try {
        const key = `sfterm-${rand()}`
        const reg = await registerSource(db, {
          siteKey: key,
          wxrPath: WXR_FIXTURE,
          allowedRoots: [FIXTURES_ROOT],
          storageRoot: join(work, 'storage'),
          reportRoot: join(work, 'artifacts', 'register'),
        })
        try {
          const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
          // 事件 1 占位 source_id=31（slug 不同→不自建命中）；事件 2 按 slug 自建
          // auto:company；事件 3 收编 UPDATE auto:company→31 撞 terms_identity_uq
          const events: AdapterEvent[] = [
            termEvent({ termId: 31, taxonomy: 'category', slug: 'company-alt', name: 'Alt' }),
            rowEvent({ sourceId: '501', taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }] }),
            termEvent({ termId: 31, taxonomy: 'category', slug: 'company', name: 'Company' }),
          ]
          const result = await createImportJob(db, {
            siteKey: key,
            snapshotId: wxr.id,
            reportRoot: join(work, 'artifacts', 'import'),
            events: eventsOf(events),
            pipeline: seededTermWriter(async (ctx) => {
              await seedContentEntry(ctx, { sourceType: 'wp:post', sourceId: '501', typeKey: 'news_article' })
            }),
          })
          expect(result.status).toBe('succeeded_with_issues')

          // 两行并存：占位 31 原样；自建行整语句回退（source_id 与 name 都未被收编改动）
          const termsRows = (await db.execute(`select t.source_id, t.slug, t.name
            from terms t join taxonomies x on x.id = t.taxonomy_id
            where t.site_id = '${reg.siteId}' order by t.source_id`)) as any
          expect(plain(termsRows)).toStrictEqual([
            { source_id: '31', slug: 'company-alt', name: 'Alt' },
            { source_id: 'auto:company', slug: 'company', name: 'company' },
          ])

          const report = JSON.parse(
            await readFile(join(work, 'artifacts', 'import', result.jobId, 'reports', 'import.json'), 'utf8'),
          ) as { counts: Record<string, number>; issues: string[] }
          expect(report.issues).toHaveLength(1)
          expect(report.issues[0]).toMatch(/收编撞唯一约束.*31/)

          // 关联仍挂保留的自建行；自建计数不回退（收编失败，行仍是自建终态）
          const ct = (await db.execute(`select r.source_id as row_id, t.source_id as term_source_id, c.is_primary
            from content_terms c
            join source_records r on r.content_entry_id = c.content_entry_id
            join terms t on t.id = c.term_id
            where c.site_id = '${reg.siteId}'`)) as any
          expect(plain(ct)).toStrictEqual([
            { row_id: '501', term_source_id: 'auto:company', is_primary: true },
          ])
          expect(report.counts).toStrictEqual({
            termsProcessed: 2,
            autoCreatedTerms: 1,
            contentTermsCreated: 1,
          })
        } finally {
          await cleanupSiteRows(db, reg.siteId)
        }
      } finally {
        await rm(work, { recursive: true, force: true })
      }
    })
  })
})