feat(import): 内容稳定身份写入
Plan 3 Task 3：content-writer（row→entry/locale/revision/source_record， checksum skip 幂等三态）。checksum 按前置事实 6 公式精确实现（canonicalJson 六输入域 + sha256；rawLocation 排除、版本三常量入、status 入、sourceUrl 经 route_source_values 入）；六列快照行身份 + 五列内容身份（跨快照复用 entry， permalink 变化产新修订不建新 entry）；attachment/unsupported/error 只记 source_records（前置事实 11/12，error 无 postId 以 rawLocation 回退不硬造）； 导入全程不触 published 指针（ADR-010）。term-writer JSDoc 补 null-id 建行 不计入 autoCreatedTerms 注脚（Task 2 遗留）。 测试：tests/integration/import-content.test.ts 7 用例（mini 全链对拍/同快照 skip/跨快照稳定身份/内容变化/ permalink 变化/unsupported+error/locale 校验）， 296 全绿；typecheck 四段链/lint 通过。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -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<Parameters<ImportDb['transaction']>[0]>[0]
|
||||
|
||||
/** sourceType → content_entries.type_key（前置事实 11 钉死；wp:attachment 不在此表——不建 entry） */
|
||||
export const CONTENT_TYPE_KEY_MAP: Readonly<Record<string, string>> = Object.freeze({
|
||||
'wp:post': 'news_article',
|
||||
'wp:page': 'basic_page',
|
||||
'wp:product': 'product',
|
||||
})
|
||||
|
||||
/** row 事件产出的 source_records.parse_status（attachment/内容行同值；unsupported/error 另有两值） */
|
||||
const PARSE_STATUS_IMPORTED = 'imported'
|
||||
const PARSE_STATUS_UNSUPPORTED = 'unsupported'
|
||||
const PARSE_STATUS_FAILED = 'failed'
|
||||
|
||||
/** content_revisions 固定落位：导入修订的 source_kind / created_by / schema_version */
|
||||
const REVISION_SOURCE_KIND = 'import' as const
|
||||
const REVISION_CREATED_BY = 'import'
|
||||
|
||||
/**
|
||||
* mapped_content / revision.data 的共用结构（前置事实 6 钉死）：可选字段缺席即无值。
|
||||
* extras 可能是 null 原型对象（mapItem 的原型污染防御）——canonicalJson 与 jsonb 序列化
|
||||
* 均兼容，此处不复制。
|
||||
*/
|
||||
function mappedContentOf(row: UniversalRow): Record<string, unknown> {
|
||||
const mapped: Record<string, unknown> = {
|
||||
status: row.status,
|
||||
seo: row.seo,
|
||||
knownFields: row.knownFields,
|
||||
extras: row.extras,
|
||||
}
|
||||
if (row.title !== undefined) mapped.title = row.title
|
||||
if (row.slug !== undefined) mapped.slug = row.slug
|
||||
if (row.bodyHtml !== undefined) mapped.bodyHtml = row.bodyHtml
|
||||
return mapped
|
||||
}
|
||||
|
||||
/**
|
||||
* normalized checksum（前置事实 6 公式精确实现）：canonicalJson 输入域后 sha256 hex。
|
||||
* 导出供 Task 4 媒体写入器复用（attachment 行的 source_records.normalized_checksum 同式）。
|
||||
*/
|
||||
export function normalizedContentChecksum(
|
||||
row: UniversalRow,
|
||||
versions: Pick<ImportVersionBlock, 'adapter' | 'mapping' | 'sanitizer'>,
|
||||
): string {
|
||||
const routeSourceValues: Record<string, unknown> = {}
|
||||
if (row.sourceUrl !== undefined) routeSourceValues.sourceUrl = row.sourceUrl
|
||||
return createHash('sha256')
|
||||
.update(
|
||||
canonicalJson({
|
||||
universal_row_version: UNIVERSAL_ROW_VERSION,
|
||||
adapter_version: versions.adapter,
|
||||
mapping_version: versions.mapping,
|
||||
sanitizer_version: versions.sanitizer,
|
||||
locale: row.locale,
|
||||
mapped_content: mappedContentOf(row),
|
||||
taxonomy_relations: row.taxonomyRefs,
|
||||
media_reference_identities: row.attachmentRefs,
|
||||
route_source_values: routeSourceValues,
|
||||
}),
|
||||
)
|
||||
.digest('hex')
|
||||
}
|
||||
|
||||
/**
|
||||
* 可分派的内容写入器句柄（Task 6 编排形状，与 term-writer 同构）：编排器单消费方
|
||||
* for-await 拉一遍事件流、逐事件分派；两态（counts/issues）经 finish 取回。句柄非可重入。
|
||||
*/
|
||||
export interface ContentEventHandler {
|
||||
/** 单事件分派（term 事件 no-op——term-writer 主场） */
|
||||
handleEvent(ev: AdapterEvent): Promise<void>
|
||||
/** 流终取回累计产出（counts/issues；调用后句柄弃用） */
|
||||
finish(): ImportPipelineOutcome
|
||||
}
|
||||
|
||||
/**
|
||||
* 内容写入器的事件分派句柄。产出 counts：
|
||||
* - 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<string, unknown>; checksum: string; sourceRecordId: string },
|
||||
): Promise<string> {
|
||||
const inserted = await tx
|
||||
.insert(contentRevisions)
|
||||
.values({
|
||||
siteId,
|
||||
contentEntryId: args.entryId,
|
||||
locale: args.row.locale,
|
||||
schemaVersion: UNIVERSAL_ROW_VERSION,
|
||||
data: args.data,
|
||||
checksum: args.checksum,
|
||||
sourceKind: REVISION_SOURCE_KIND,
|
||||
sourceRecordId: args.sourceRecordId,
|
||||
importJobId: identity.jobId,
|
||||
createdBy: REVISION_CREATED_BY,
|
||||
})
|
||||
.returning({ id: contentRevisions.id })
|
||||
return inserted[0]!.id
|
||||
}
|
||||
|
||||
/**
|
||||
* current 指针前移 + slug 跟随最新行（可变表 UPDATE——published 永不碰，ADR-010）。
|
||||
* 行缺席时补建（跨写入器接缝兜底：本写入器正常路径首导必先建 entry_locales）。
|
||||
*/
|
||||
async function pointCurrent(tx: Tx, entryId: string, row: UniversalRow, revisionId: string): Promise<void> {
|
||||
const updated = await tx
|
||||
.update(entryLocales)
|
||||
.set({ currentRevisionId: revisionId, slug: row.slug ?? row.sourceId, updatedAt: new Date() })
|
||||
.where(
|
||||
and(
|
||||
eq(entryLocales.siteId, siteId),
|
||||
eq(entryLocales.contentEntryId, entryId),
|
||||
eq(entryLocales.locale, row.locale),
|
||||
),
|
||||
)
|
||||
.returning({ id: entryLocales.id })
|
||||
if (updated.length === 0) {
|
||||
await tx.insert(entryLocales).values({
|
||||
siteId,
|
||||
contentEntryId: entryId,
|
||||
locale: row.locale,
|
||||
slug: row.slug ?? row.sourceId,
|
||||
currentRevisionId: revisionId,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/** 快照行（六列身份）查找：本快照内该源对象的 source_records 行（唯一索引保证至多一行） */
|
||||
async function findSnapshotRecord(row: UniversalRow) {
|
||||
const rows = await db
|
||||
.select({
|
||||
id: sourceRecords.id,
|
||||
contentEntryId: sourceRecords.contentEntryId,
|
||||
normalizedChecksum: sourceRecords.normalizedChecksum,
|
||||
})
|
||||
.from(sourceRecords)
|
||||
.where(
|
||||
and(
|
||||
eq(sourceRecords.siteId, siteId),
|
||||
eq(sourceRecords.sourceConnectionId, identity.connectionId),
|
||||
eq(sourceRecords.sourceType, row.sourceType),
|
||||
eq(sourceRecords.sourceId, row.sourceId),
|
||||
eq(sourceRecords.locale, row.locale),
|
||||
eq(sourceRecords.sourceSnapshotId, identity.snapshotId),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
return rows[0] ?? null
|
||||
}
|
||||
|
||||
/** 五列内容身份 → 既有 entry 的最新快照行（跨快照；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<void> {
|
||||
const checksum = normalizedContentChecksum(row, identity.versions)
|
||||
const data = mappedContentOf(row)
|
||||
|
||||
// 态 1：本快照行已存在（挂 entry——本写入器是 content 行 source_records 的唯一写入方，
|
||||
// 挂接不变量由写入路径自身保证）
|
||||
const snapRow = await findSnapshotRecord(row)
|
||||
if (snapRow !== null && snapRow.contentEntryId !== null) {
|
||||
if (snapRow.normalizedChecksum === checksum) {
|
||||
counts.skipped++
|
||||
return
|
||||
}
|
||||
// checksum 异：快照行前移（checksum/source_url）+ 新修订 + current 前移（事务）
|
||||
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<void> {
|
||||
if (!knownType) {
|
||||
issues.push(
|
||||
`未知 sourceType：${JSON.stringify(row.sourceType)}（${row.sourceId}）不在 typeKey 映射表——只记 source_records，不建 content entry`,
|
||||
)
|
||||
}
|
||||
const checksum = normalizedContentChecksum(row, identity.versions)
|
||||
const snapRow = await findSnapshotRecord(row)
|
||||
if (snapRow !== null) {
|
||||
if (snapRow.normalizedChecksum === checksum) return
|
||||
await db
|
||||
.update(sourceRecords)
|
||||
.set({ normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null })
|
||||
.where(eq(sourceRecords.id, snapRow.id))
|
||||
return
|
||||
}
|
||||
await db.insert(sourceRecords).values(recordValues(row, checksum, null))
|
||||
counts.sourceRecords++
|
||||
}
|
||||
|
||||
/** unsupported/error 事件（前置事实 12）：也写 source_records，可审计且幂等 */
|
||||
async function handleUnmappable(
|
||||
ev: Extract<AdapterEvent, { kind: 'unsupported' }> | Extract<AdapterEvent, { kind: 'error' }>,
|
||||
parseStatus: 'unsupported' | 'failed',
|
||||
): Promise<void> {
|
||||
// postId 缺失不硬造身份——source_id 以 rawLocation 回退（行可审计、身份不虚构）
|
||||
const sourceId = ev.postId ?? ev.rawLocation
|
||||
const inserted = await db
|
||||
.insert(sourceRecords)
|
||||
.values({
|
||||
siteId,
|
||||
sourceConnectionId: identity.connectionId,
|
||||
sourceSnapshotId: identity.snapshotId,
|
||||
importJobId: identity.jobId,
|
||||
sourceType: ev.sourceType,
|
||||
sourceId,
|
||||
locale: identity.defaultLocale,
|
||||
rawLocation: ev.rawLocation,
|
||||
parseStatus: parseStatus === 'unsupported' ? PARSE_STATUS_UNSUPPORTED : PARSE_STATUS_FAILED,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: sourceRecords.id })
|
||||
if (inserted.length > 0) counts.sourceRecords++
|
||||
}
|
||||
|
||||
/** row 事件：locale 校验 → 类型分派（typeKey 映射 / attachment / 未知兜底） */
|
||||
async function handleRow(row: UniversalRow): Promise<void> {
|
||||
if (row.locale !== identity.defaultLocale) {
|
||||
issues.push(
|
||||
`row locale 不匹配站点默认（${row.sourceType}/${row.sourceId}，row.locale=${JSON.stringify(row.locale)} ≠ site.default_locale=${JSON.stringify(identity.defaultLocale)}）——跳过（前置事实 8）`,
|
||||
)
|
||||
counts.skipped++
|
||||
return
|
||||
}
|
||||
const typeKey = Object.hasOwn(CONTENT_TYPE_KEY_MAP, row.sourceType)
|
||||
? CONTENT_TYPE_KEY_MAP[row.sourceType]
|
||||
: undefined
|
||||
if (typeKey === undefined) {
|
||||
await handleRecordOnlyRow(row, row.sourceType === 'wp:attachment')
|
||||
return
|
||||
}
|
||||
await handleContentRow(row, typeKey)
|
||||
}
|
||||
|
||||
return {
|
||||
async handleEvent(ev) {
|
||||
if (ev.kind === 'row') {
|
||||
await handleRow(ev.row)
|
||||
} else if (ev.kind === 'unsupported') {
|
||||
await handleUnmappable(ev, 'unsupported')
|
||||
} else if (ev.kind === 'error') {
|
||||
await handleUnmappable(ev, 'failed')
|
||||
}
|
||||
// term 事件：内容写入器不消费（term-writer 主场；Task 6 编排各写入器各取所需）
|
||||
},
|
||||
finish() {
|
||||
return { counts, issues }
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 内容写入管道（{@link createContentEventHandler} 的自消费整流便捷壳）：单遍拉完
|
||||
* ctx.events、逐事件分派、流终返回 outcome。Task 6 编排时可改用句柄形状直接分派
|
||||
* （单消费方拉流分派给各写入器，无 tee）。
|
||||
*/
|
||||
export function createContentWriter(): ImportPipeline {
|
||||
return async (ctx) => {
|
||||
const handler = createContentEventHandler({
|
||||
db: ctx.db,
|
||||
siteId: ctx.siteId,
|
||||
connectionId: ctx.connectionId,
|
||||
snapshotId: ctx.wxrSnapshot.id,
|
||||
jobId: ctx.jobId,
|
||||
defaultLocale: ctx.defaultLocale,
|
||||
versions: ctx.versions,
|
||||
})
|
||||
for await (const ev of ctx.events) {
|
||||
await handler.handleEvent(ev)
|
||||
}
|
||||
return handler.finish()
|
||||
}
|
||||
}
|
||||
@@ -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:<slug>）但**不计入**——建行源自 wp:term 表侧在场，非引用侧缺表自建）；
|
||||
* - contentTermsCreated：新建的 (entry, term) 关联数（幂等重跑为 0）。
|
||||
* issues：无 entry 可关联的 row 分类引用、表外 taxonomy 兜底、缺 term_id 的 wp:term
|
||||
* 事件、同 slug 双 term_id、收编撞唯一约束——全部显式报告（§7.6 禁静默）。
|
||||
|
||||
@@ -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<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,
|
||||
})
|
||||
// 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 })
|
||||
}
|
||||
})
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user