diff --git a/apps/api/src/modules/import/content-writer.ts b/apps/api/src/modules/import/content-writer.ts new file mode 100644 index 0000000..68e7038 --- /dev/null +++ b/apps/api/src/modules/import/content-writer.ts @@ -0,0 +1,474 @@ +// 内容稳定身份写入器（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（job 不因此 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, desc, eq, isNotNull } 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 type { ImportDb, ImportPipeline, ImportPipelineOutcome, ImportVersionBlock } from './import.service.js' + +/** 事务句柄类型（db.transaction 回调参数——从 ImportDb 的事务签名提取，避免重复泛型展开） */ +type Tx = Parameters[0]>[0] + +/** sourceType → content_entries.type_key（前置事实 11 钉死；wp:attachment 不在此表——不建 entry） */ +export const CONTENT_TYPE_KEY_MAP: Readonly> = 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 { + const mapped: Record = { + 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, +): string { + const routeSourceValues: Record = {} + 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 + /** 流终取回累计产出（counts/issues；调用后句柄弃用） */ + finish(): ImportPipelineOutcome +} + +/** + * 内容写入器的事件分派句柄。产出 counts： + * - contentEntries：新建 content_entries 数（幂等重跑/变化路径为 0）； + * - revisionsCreated：新建 content_revisions 数（checksum 命中为 0——G2 验收锚点）； + * - sourceRecords：新建 source_records 数（内容/attachment/unsupported/error 全计； + * 跨快照重导按快照补行故可为正）； + * - skipped：内容行未产修订的跳过数（同快照 checksum 命中、跨快照无业务变化、locale + * 拒收三类；attachment 命中不计——媒体域是 Task 4 主场）。 + * issues：locale 不匹配、未知 sourceType 兜底——显式报告（§7.6 禁静默）。 + */ +export function createContentEventHandler(identity: { + db: ImportDb + siteId: string + /** 快照行六列身份的 connection 侧 + 五列内容身份的解析锚点 */ + connectionId: string + /** 本 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 + const counts = { 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; checksum: string; sourceRecordId: string }, + ): Promise { + 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 { + 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 的最新快照行（跨快照；latest first——与 term-writer 同口径） */ + async function findPriorLinkedRecord(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), + isNotNull(sourceRecords.contentEntryId), + ), + ) + .orderBy(desc(sourceRecords.createdAt)) + .limit(1) + // contentEntryId 非 null：查询谓词 isNotNull 保证（drizzle 谓词不细化 select 类型） + return rows[0] === undefined ? null : { ...rows[0], contentEntryId: rows[0].contentEntryId! } + } + + /** 内容行主路径：幂等三态（miss / checksum 同 skip / checksum 异新修订） */ + async function handleContentRow(row: UniversalRow, typeKey: string): Promise { + 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 前移（事务） + 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 + } + + // 态 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 { + 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 | Extract, + parseStatus: 'unsupported' | 'failed', + ): Promise { + // postId 缺失不硬造身份——source_id 以 rawLocation 回退（行可审计、身份不虚构） + const sourceId = ev.postId ?? ev.rawLocation + 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 { + 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() + } +} diff --git a/apps/api/src/modules/import/term-writer.ts b/apps/api/src/modules/import/term-writer.ts index 39e8a61..8e9745d 100644 --- a/apps/api/src/modules/import/term-writer.ts +++ b/apps/api/src/modules/import/term-writer.ts @@ -88,7 +88,8 @@ export interface TermEventHandler { * term 物化的事件分派句柄。产出 counts： * - termsProcessed：处理的 wp:term 事件数； * - autoCreatedTerms：**终态**自建数（引用侧自建且未被 wp:term 收编的行数——瞬态自建 - * 被收编即回退，不虚报"wp:term 表缺失"）； + * 被收编即回退，不虚报"wp:term 表缺失"；null-id wp:term 事件照建行（source_id= + * auto:）但**不计入**——建行源自 wp:term 表侧在场，非引用侧缺表自建）； * - contentTermsCreated：新建的 (entry, term) 关联数（幂等重跑为 0）。 * issues：无 entry 可关联的 row 分类引用、表外 taxonomy 兜底、缺 term_id 的 wp:term * 事件、同 slug 双 term_id、收编撞唯一约束——全部显式报告（§7.6 禁静默）。 diff --git a/tests/integration/import-content.test.ts b/tests/integration/import-content.test.ts new file mode 100644 index 0000000..88b6175 --- /dev/null +++ b/tests/integration/import-content.test.ts @@ -0,0 +1,674 @@ +// 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> = Object.freeze({ + 'wp:post': 'news_article', + 'wp:page': 'basic_page', + 'wp:product': 'product', +}) + +/** 手构事件流的 AsyncIterable 包装（每次调用产新迭代器——重跑各流独立） */ +async function* eventsOf(events: AdapterEvent[]): AsyncGenerator { + yield* events +} + +/** db.execute 行对象规整：postgres.js 行对象的原型差异会让 toStrictEqual 报"无视觉差异"失败——浅拷贝到普通对象 */ +function plain(rows: any[]): any[] { + return rows.map((r) => ({ ...r })) +} + +/** 事件流收集（mini 全链一饮而尽；断言期望从产物行派生，不猜清洗细节） */ +async function collectMiniEvents(): Promise { + 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 = { + 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 = {} + 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 { + const data: Record = { + 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 { + return JSON.parse(JSON.stringify(value)) as Record +} + +/** 手构 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 { + 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 + 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('', '\n ')) + 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, + }) + // 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: '

bonjour

' }), + rowEvent({ sourceId: '302', slug: 'post-302', bodyHtml: '

ok

' }), + ] + 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 }) + } + }) + }) +})