浅念
77687e2330
Task 6 质量评审四项修复： - 失败报告部分冲刷（Fix 1）：orchestratedPipeline 分派/finish 包 try/catch—— 中途抛错时把可安全取得的部分产出冲入 ctx.report 后重抛：term/content 的 finish 是纯返回（已算成复用、未算成就地补算 trySyncFinish，自身抛错放弃 该部分不掩盖原错）、已收集的 adapter 问题/清洗统计、进度计数 eventsDispatched（orchestrator 键，仅失败路径出现——成功路径不占 18 键面）。 media/route finish 依赖库与先序产物，失败路径不调用（其键缺席如实反映）。 - 列表路径校验（Fix 3）：ImportErrorCode 增 INVALID_LIST_PATH；解析期防线 移至阶段 2.5（建 job 行前——建行/置 running 后再抛会留悬挂行）；复用 route-writer 导出的 isSafePath；normalizeListPath 自带安全拒绝（写入器 自调用位双覆盖）。CLI 无需新码（既有 ImportError 通道按 code 呈现）。 - JSDoc 求和措辞（Fix 2）：content-writer/media-writer 的 counts 键契约 注记由"返回值覆盖同名键 CLOBBER"改为"同名键求和（Task 6 钉死——防覆盖 安全网）"，与 ImportPipelineOutcome.counts 契约对齐。 - reason 归档（Fix 4）：分派循环对 unsupported/error 事件把 reason/issue 推入 adapter 问题组（此前被静默丢弃）。 测试：import-full 增 2 例（手构流 3 事件后爆炸 → failed + eventsDispatched=3 + 部分 content/term 计数 + unsupported/error reason 双归档；--products-path /../x/ 等 → INVALID_LIST_PATH 零 job 行）。全量 325 通过；typecheck/lint 干净。 Co-Authored-By: Claude Code <noreply@anthropic.com>
565 lines
25 KiB
TypeScript
565 lines
25 KiB
TypeScript
// 内容稳定身份写入器（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, WriterContext } from './import.service.js'
|
||
|
||
/** 事务句柄类型（db.transaction 回调参数——从 ImportDb 的事务签名提取，避免重复泛型展开） */
|
||
type Tx = Parameters<Parameters<ImportDb['transaction']>[0]>[0]
|
||
|
||
/** sourceType → content_entries.type_key（前置事实 11 钉死；wp:attachment 不在此表——不建 entry） */
|
||
export const CONTENT_TYPE_KEY_MAP: Readonly<Record<string, string>> = Object.freeze({
|
||
'wp:post': 'news_article',
|
||
'wp:page': 'basic_page',
|
||
'wp:product': 'product',
|
||
})
|
||
|
||
/** row 事件产出的 source_records.parse_status（attachment/内容行同值；unsupported/error 另有两值） */
|
||
const PARSE_STATUS_IMPORTED = 'imported'
|
||
const PARSE_STATUS_UNSUPPORTED = 'unsupported'
|
||
const PARSE_STATUS_FAILED = 'failed'
|
||
|
||
/** content_revisions 固定落位：导入修订的 source_kind / created_by / schema_version */
|
||
const REVISION_SOURCE_KIND = 'import' as const
|
||
const REVISION_CREATED_BY = 'import'
|
||
|
||
/**
|
||
* mapped_content / revision.data 的共用结构（前置事实 6 钉死）：可选字段缺席即无值。
|
||
* extras 可能是 null 原型对象（mapItem 的原型污染防御）——canonicalJson 与 jsonb 序列化
|
||
* 均兼容，此处不复制。
|
||
*/
|
||
function mappedContentOf(row: UniversalRow): Record<string, unknown> {
|
||
const mapped: Record<string, unknown> = {
|
||
status: row.status,
|
||
seo: row.seo,
|
||
knownFields: row.knownFields,
|
||
extras: row.extras,
|
||
}
|
||
if (row.title !== undefined) mapped.title = row.title
|
||
if (row.slug !== undefined) mapped.slug = row.slug
|
||
if (row.bodyHtml !== undefined) mapped.bodyHtml = row.bodyHtml
|
||
return mapped
|
||
}
|
||
|
||
/**
|
||
* normalized checksum（前置事实 6 公式精确实现）：canonicalJson 输入域后 sha256 hex。
|
||
* 导出供 Task 4 媒体写入器复用（attachment 行的 source_records.normalized_checksum 同式）。
|
||
*/
|
||
export function normalizedContentChecksum(
|
||
row: UniversalRow,
|
||
versions: Pick<ImportVersionBlock, 'adapter' | 'mapping' | 'sanitizer'>,
|
||
): string {
|
||
const routeSourceValues: Record<string, unknown> = {}
|
||
if (row.sourceUrl !== undefined) routeSourceValues.sourceUrl = row.sourceUrl
|
||
return createHash('sha256')
|
||
.update(
|
||
canonicalJson({
|
||
universal_row_version: UNIVERSAL_ROW_VERSION,
|
||
adapter_version: versions.adapter,
|
||
mapping_version: versions.mapping,
|
||
sanitizer_version: versions.sanitizer,
|
||
locale: row.locale,
|
||
mapped_content: mappedContentOf(row),
|
||
taxonomy_relations: row.taxonomyRefs,
|
||
media_reference_identities: row.attachmentRefs,
|
||
route_source_values: routeSourceValues,
|
||
}),
|
||
)
|
||
.digest('hex')
|
||
}
|
||
|
||
/**
|
||
* 可分派的内容写入器句柄（Task 6 编排形状，与 term-writer 同构）：编排器单消费方
|
||
* for-await 拉一遍事件流、逐事件分派；两态（counts/issues）经 finish 取回。句柄非可重入。
|
||
*/
|
||
export interface ContentEventHandler {
|
||
/** 单事件分派（term 事件 no-op——term-writer 主场） */
|
||
handleEvent(ev: AdapterEvent): Promise<void>
|
||
/** 流终取回累计产出（counts/issues；调用后句柄弃用） */
|
||
finish(): ImportPipelineOutcome
|
||
}
|
||
|
||
/**
|
||
* 本写入器的 counts 键与归属（**counts 键是跨写入器的扁平全局命名空间**——import.service
|
||
* 按"同名键**求和**"合并（Task 6 钉死——防覆盖安全网），各写入器必须各持独占键名，见
|
||
* ImportPipelineOutcome.counts 的键契约注记）。本写入器独占：contentEntries /
|
||
* revisionsCreated / sourceRecords / skipped / unsupportedRecords / failedRecords——
|
||
* **Task 4 media-writer 起新键名，勿复用 'skipped'**（语义不同会互相污染）。
|
||
* 前四键恒出现在报告；unsupportedRecords/failedRecords 只在 >0 时出现（分项按事件计）。
|
||
*/
|
||
interface ContentWriterCounts {
|
||
contentEntries: number
|
||
revisionsCreated: number
|
||
sourceRecords: number
|
||
skipped: number
|
||
unsupportedRecords?: number
|
||
failedRecords?: number
|
||
}
|
||
|
||
/**
|
||
* 内容写入器的事件分派句柄。产出 counts：
|
||
* - contentEntries：新建 content_entries 数（幂等重跑/变化路径为 0）；
|
||
* - revisionsCreated：新建 content_revisions 数（checksum 命中为 0——G2 验收锚点）；
|
||
* - sourceRecords：新建 source_records 数（内容/attachment/unsupported/error 全计；
|
||
* 跨快照重导按快照补行故可为正；NULL-entry 占位行收编**不计**——行已在，UPDATE
|
||
* 挂接非新建，见 handleContentRow 态 1b）；
|
||
* - skipped：内容行未产修订的跳过数（同快照 checksum 命中、跨快照无业务变化、locale
|
||
* 拒收三类；attachment 命中不计——媒体域是 Task 4 主场）；
|
||
* - unsupportedRecords/failedRecords：unsupported/error 事件分项数（**按事件计**非按
|
||
* 插入行——重跑行幂等而计数如实；拆分只在分派点可知，Task 6 报告层无从重derive）。
|
||
* issues：locale 不匹配、未知 sourceType 兜底——显式报告（§7.6 禁静默）。
|
||
*/
|
||
export function createContentEventHandler(identity: WriterContext & {
|
||
/** 本 job 的 wxr 快照 id（六列身份的 snapshot 侧 + 新行 source_snapshot_id） */
|
||
snapshotId: string
|
||
/** 本 job id（source_records.import_job_id 与 content_revisions.import_job_id 落库锚点） */
|
||
jobId: string
|
||
/** 站点默认 locale（前置事实 8：row.locale 校验基准） */
|
||
defaultLocale: string
|
||
/** 版本块（checksum 的三常量输入；parser 不入哈希） */
|
||
versions: ImportVersionBlock
|
||
}): ContentEventHandler {
|
||
const { db, siteId } = identity
|
||
// 交叉 Record：finish() 要把 counts 交还给 ImportPipelineOutcome（Record<string, number>）
|
||
// ——接口无索引签名不可赋值；交叉后已知键类型不变、动态赋值兼容
|
||
const counts: ContentWriterCounts & Record<string, number> = {
|
||
contentEntries: 0,
|
||
revisionsCreated: 0,
|
||
sourceRecords: 0,
|
||
skipped: 0,
|
||
}
|
||
const issues: string[] = []
|
||
|
||
/** 修订插入（纯 INSERT——不可变表；data 与 checksum 同构于 mapped_content） */
|
||
async function insertRevision(
|
||
tx: Tx,
|
||
args: { entryId: string; row: UniversalRow; data: Record<string, unknown>; checksum: string; sourceRecordId: string },
|
||
): Promise<string> {
|
||
const inserted = await tx
|
||
.insert(contentRevisions)
|
||
.values({
|
||
siteId,
|
||
contentEntryId: args.entryId,
|
||
locale: args.row.locale,
|
||
schemaVersion: UNIVERSAL_ROW_VERSION,
|
||
data: args.data,
|
||
checksum: args.checksum,
|
||
sourceKind: REVISION_SOURCE_KIND,
|
||
sourceRecordId: args.sourceRecordId,
|
||
importJobId: identity.jobId,
|
||
createdBy: REVISION_CREATED_BY,
|
||
})
|
||
.returning({ id: contentRevisions.id })
|
||
return inserted[0]!.id
|
||
}
|
||
|
||
/**
|
||
* current 指针前移 + slug 跟随最新行（可变表 UPDATE——published 永不碰，ADR-010）。
|
||
* 行缺席时补建（跨写入器接缝兜底：本写入器正常路径首导必先建 entry_locales）。
|
||
*/
|
||
async function pointCurrent(tx: Tx, entryId: string, row: UniversalRow, revisionId: string): Promise<void> {
|
||
const updated = await tx
|
||
.update(entryLocales)
|
||
.set({ currentRevisionId: revisionId, slug: row.slug ?? row.sourceId, updatedAt: new Date() })
|
||
.where(
|
||
and(
|
||
eq(entryLocales.siteId, siteId),
|
||
eq(entryLocales.contentEntryId, entryId),
|
||
eq(entryLocales.locale, row.locale),
|
||
),
|
||
)
|
||
.returning({ id: entryLocales.id })
|
||
if (updated.length === 0) {
|
||
await tx.insert(entryLocales).values({
|
||
siteId,
|
||
contentEntryId: entryId,
|
||
locale: row.locale,
|
||
slug: row.slug ?? row.sourceId,
|
||
currentRevisionId: revisionId,
|
||
})
|
||
}
|
||
}
|
||
|
||
/** 快照行（六列身份）查找：本快照内该源对象的 source_records 行（唯一索引保证至多一行） */
|
||
async function findSnapshotRecord(row: UniversalRow) {
|
||
const rows = await db
|
||
.select({
|
||
id: sourceRecords.id,
|
||
contentEntryId: sourceRecords.contentEntryId,
|
||
normalizedChecksum: sourceRecords.normalizedChecksum,
|
||
})
|
||
.from(sourceRecords)
|
||
.where(
|
||
and(
|
||
eq(sourceRecords.siteId, siteId),
|
||
eq(sourceRecords.sourceConnectionId, identity.connectionId),
|
||
eq(sourceRecords.sourceType, row.sourceType),
|
||
eq(sourceRecords.sourceId, row.sourceId),
|
||
eq(sourceRecords.locale, row.locale),
|
||
eq(sourceRecords.sourceSnapshotId, identity.snapshotId),
|
||
),
|
||
)
|
||
.limit(1)
|
||
return rows[0] ?? null
|
||
}
|
||
|
||
/** 五列内容身份 → 既有 entry 的最新快照行（跨快照；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 前移（事务）。
|
||
// lineage 语义（有意）：source_records.import_job_id 留在建行 job（行的溯源=创建它的
|
||
// 导入），本次 job 记在 content_revisions.import_job_id（谁产了新修订）
|
||
const entryId = snapRow.contentEntryId // 事务回调外的 const 收窄（属性收窄不进闭包）
|
||
await db.transaction(async (tx) => {
|
||
await tx
|
||
.update(sourceRecords)
|
||
.set({ normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null })
|
||
.where(eq(sourceRecords.id, snapRow.id))
|
||
const revisionId = await insertRevision(tx, {
|
||
entryId,
|
||
row,
|
||
data,
|
||
checksum,
|
||
sourceRecordId: snapRow.id,
|
||
})
|
||
await pointCurrent(tx, entryId, row, revisionId)
|
||
})
|
||
counts.revisionsCreated++
|
||
return
|
||
}
|
||
if (snapRow !== null) {
|
||
// 态 1b：NULL-entry 占位行收编（Task 6 义务①——Task 3 审查移交的已知死角在此闭合）：
|
||
// 六列命中而 content_entry_id NULL 的行（外源 seed / attachment record-only 足迹 +
|
||
// 映射扩充后该 sourceType 升格为内容类型重导）→ 走全新建分支会撞六列唯一约束
|
||
// 23505，改为就地 UPDATE 挂 entry。语义与态 2/3 对齐：
|
||
// - 无五列先行者 → 全新建变体（entry+revision+locale），占位行 UPDATE 挂接；
|
||
// - 先行者 checksum 同 → 只挂接（skipped——0 新修订）；
|
||
// - 先行者 checksum 异 → 挂接 + 新修订 + current 前移。
|
||
// counts：sourceRecords 不计（行已在，收编非新建）；lineage 留在占位行建行 job。
|
||
const prior = await findPriorLinkedRecord(row)
|
||
const adoptValues = { contentEntryId: null as string | null, normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null }
|
||
if (prior === null) {
|
||
await db.transaction(async (tx) => {
|
||
const entry = await tx
|
||
.insert(contentEntries)
|
||
.values({ siteId, typeKey, lifecycleStatus: 'active' })
|
||
.returning({ id: contentEntries.id })
|
||
const entryId = entry[0]!.id
|
||
await tx
|
||
.update(sourceRecords)
|
||
.set({ ...adoptValues, contentEntryId: entryId })
|
||
.where(eq(sourceRecords.id, snapRow.id))
|
||
const revisionId = await insertRevision(tx, {
|
||
entryId,
|
||
row,
|
||
data,
|
||
checksum,
|
||
sourceRecordId: snapRow.id,
|
||
})
|
||
await pointCurrent(tx, entryId, row, revisionId)
|
||
})
|
||
counts.contentEntries++
|
||
counts.revisionsCreated++
|
||
return
|
||
}
|
||
const entryId = prior.contentEntryId // 事务回调外的 const 收窄（属性收窄不进闭包）
|
||
if (prior.normalizedChecksum === checksum) {
|
||
await db
|
||
.update(sourceRecords)
|
||
.set({ ...adoptValues, contentEntryId: entryId })
|
||
.where(eq(sourceRecords.id, snapRow.id))
|
||
counts.skipped++
|
||
return
|
||
}
|
||
await db.transaction(async (tx) => {
|
||
await tx
|
||
.update(sourceRecords)
|
||
.set({ ...adoptValues, contentEntryId: entryId })
|
||
.where(eq(sourceRecords.id, snapRow.id))
|
||
const revisionId = await insertRevision(tx, {
|
||
entryId,
|
||
row,
|
||
data,
|
||
checksum,
|
||
sourceRecordId: snapRow.id,
|
||
})
|
||
await pointCurrent(tx, entryId, row, revisionId)
|
||
})
|
||
counts.revisionsCreated++
|
||
return
|
||
}
|
||
|
||
// 态 2/3 前置：内容身份（五列）解析——跨快照复用既有 entry（permalink/快照变化不建新）
|
||
const prior = await findPriorLinkedRecord(row)
|
||
if (prior === null) {
|
||
// 态 3 全新建：entry + revision + locale + 快照行（事务——四行原子）
|
||
await db.transaction(async (tx) => {
|
||
const entry = await tx
|
||
.insert(contentEntries)
|
||
.values({ siteId, typeKey, lifecycleStatus: 'active' })
|
||
.returning({ id: contentEntries.id })
|
||
const entryId = entry[0]!.id
|
||
const record = await tx
|
||
.insert(sourceRecords)
|
||
.values(recordValues(row, checksum, entryId))
|
||
.returning({ id: sourceRecords.id })
|
||
const revisionId = await insertRevision(tx, {
|
||
entryId,
|
||
row,
|
||
data,
|
||
checksum,
|
||
sourceRecordId: record[0]!.id,
|
||
})
|
||
await tx.insert(entryLocales).values({
|
||
siteId,
|
||
contentEntryId: entryId,
|
||
locale: row.locale,
|
||
slug: row.slug ?? row.sourceId,
|
||
currentRevisionId: revisionId,
|
||
// published_revisionId 缺省 NULL——导入不发布（ADR-010）
|
||
})
|
||
})
|
||
counts.contentEntries++
|
||
counts.revisionsCreated++
|
||
counts.sourceRecords++
|
||
return
|
||
}
|
||
if (prior.normalizedChecksum === checksum) {
|
||
// 态 2 跨快照无业务变化：只补本快照行（挂同一 entry），0 新修订（spec §7.5 Skip）
|
||
await db.insert(sourceRecords).values(recordValues(row, checksum, prior.contentEntryId))
|
||
counts.sourceRecords++
|
||
counts.skipped++
|
||
return
|
||
}
|
||
// 跨快照有变化：补本快照行 + 新修订 + current 前移（事务）
|
||
await db.transaction(async (tx) => {
|
||
const record = await tx
|
||
.insert(sourceRecords)
|
||
.values(recordValues(row, checksum, prior.contentEntryId))
|
||
.returning({ id: sourceRecords.id })
|
||
const revisionId = await insertRevision(tx, {
|
||
entryId: prior.contentEntryId,
|
||
row,
|
||
data,
|
||
checksum,
|
||
sourceRecordId: record[0]!.id,
|
||
})
|
||
await pointCurrent(tx, prior.contentEntryId, row, revisionId)
|
||
})
|
||
counts.sourceRecords++
|
||
counts.revisionsCreated++
|
||
}
|
||
|
||
/** source_records 行值组装（source_checksum NULL——行级原始哈希无输入源，快照级在快照行） */
|
||
function recordValues(row: UniversalRow, checksum: string, contentEntryId: string | null) {
|
||
return {
|
||
siteId,
|
||
sourceConnectionId: identity.connectionId,
|
||
sourceSnapshotId: identity.snapshotId,
|
||
importJobId: identity.jobId,
|
||
sourceType: row.sourceType,
|
||
sourceId: row.sourceId,
|
||
locale: row.locale,
|
||
sourceUrl: row.sourceUrl ?? null,
|
||
rawLocation: row.rawLocation,
|
||
normalizedChecksum: checksum,
|
||
parseStatus: PARSE_STATUS_IMPORTED,
|
||
contentEntryId,
|
||
}
|
||
}
|
||
|
||
/**
|
||
* record-only 行（attachment / 未知 sourceType）：只记 source_records（不建 entry）。
|
||
* checksum 命中静默跳过（不计 skipped——媒体域计数是 Task 4 主场）；变化则快照行前移。
|
||
*/
|
||
async function handleRecordOnlyRow(row: UniversalRow, knownType: boolean): Promise<void> {
|
||
if (!knownType) {
|
||
issues.push(
|
||
`未知 sourceType：${JSON.stringify(row.sourceType)}（${row.sourceId}）不在 typeKey 映射表——只记 source_records，不建 content entry`,
|
||
)
|
||
}
|
||
const checksum = normalizedContentChecksum(row, identity.versions)
|
||
const snapRow = await findSnapshotRecord(row)
|
||
if (snapRow !== null) {
|
||
if (snapRow.normalizedChecksum === checksum) return
|
||
await db
|
||
.update(sourceRecords)
|
||
.set({ normalizedChecksum: checksum, sourceUrl: row.sourceUrl ?? null })
|
||
.where(eq(sourceRecords.id, snapRow.id))
|
||
return
|
||
}
|
||
await db.insert(sourceRecords).values(recordValues(row, checksum, null))
|
||
counts.sourceRecords++
|
||
}
|
||
|
||
/** unsupported/error 事件（前置事实 12）：也写 source_records，可审计且幂等 */
|
||
async function handleUnmappable(
|
||
ev: Extract<AdapterEvent, { kind: 'unsupported' }> | Extract<AdapterEvent, { kind: 'error' }>,
|
||
parseStatus: 'unsupported' | 'failed',
|
||
): Promise<void> {
|
||
// postId 缺失不硬造身份——source_id 以 rawLocation 回退（行可审计、身份不虚构）
|
||
const sourceId = ev.postId ?? ev.rawLocation
|
||
// 分项计数按事件计（非按插入行）：unsupported/failed 的拆分只在分派点可知，
|
||
// Task 6 报告层无法从 sourceRecords 总数重derive；重跑时行幂等但计数如实。
|
||
if (parseStatus === 'unsupported') counts.unsupportedRecords = (counts.unsupportedRecords ?? 0) + 1
|
||
else counts.failedRecords = (counts.failedRecords ?? 0) + 1
|
||
const inserted = await db
|
||
.insert(sourceRecords)
|
||
.values({
|
||
siteId,
|
||
sourceConnectionId: identity.connectionId,
|
||
sourceSnapshotId: identity.snapshotId,
|
||
importJobId: identity.jobId,
|
||
sourceType: ev.sourceType,
|
||
sourceId,
|
||
locale: identity.defaultLocale,
|
||
rawLocation: ev.rawLocation,
|
||
parseStatus: parseStatus === 'unsupported' ? PARSE_STATUS_UNSUPPORTED : PARSE_STATUS_FAILED,
|
||
})
|
||
.onConflictDoNothing()
|
||
.returning({ id: sourceRecords.id })
|
||
if (inserted.length > 0) counts.sourceRecords++
|
||
}
|
||
|
||
/** row 事件：locale 校验 → 类型分派（typeKey 映射 / attachment / 未知兜底） */
|
||
async function handleRow(row: UniversalRow): Promise<void> {
|
||
if (row.locale !== identity.defaultLocale) {
|
||
issues.push(
|
||
`row locale 不匹配站点默认（${row.sourceType}/${row.sourceId}，row.locale=${JSON.stringify(row.locale)} ≠ site.default_locale=${JSON.stringify(identity.defaultLocale)}）——跳过（前置事实 8）`,
|
||
)
|
||
counts.skipped++
|
||
return
|
||
}
|
||
const typeKey = Object.hasOwn(CONTENT_TYPE_KEY_MAP, row.sourceType)
|
||
? CONTENT_TYPE_KEY_MAP[row.sourceType]
|
||
: undefined
|
||
if (typeKey === undefined) {
|
||
await handleRecordOnlyRow(row, row.sourceType === 'wp:attachment')
|
||
return
|
||
}
|
||
await handleContentRow(row, typeKey)
|
||
}
|
||
|
||
return {
|
||
async handleEvent(ev) {
|
||
if (ev.kind === 'row') {
|
||
await handleRow(ev.row)
|
||
} else if (ev.kind === 'unsupported') {
|
||
await handleUnmappable(ev, 'unsupported')
|
||
} else if (ev.kind === 'error') {
|
||
await handleUnmappable(ev, 'failed')
|
||
}
|
||
// term 事件：内容写入器不消费（term-writer 主场；Task 6 编排各写入器各取所需）
|
||
},
|
||
finish() {
|
||
return { counts, issues }
|
||
},
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 内容写入管道（{@link createContentEventHandler} 的自消费整流便捷壳）：单遍拉完
|
||
* ctx.events、逐事件分派、流终返回 outcome。Task 6 编排时可改用句柄形状直接分派
|
||
* （单消费方拉流分派给各写入器，无 tee）。
|
||
*/
|
||
export function createContentWriter(): ImportPipeline {
|
||
return async (ctx) => {
|
||
const handler = createContentEventHandler({
|
||
db: ctx.db,
|
||
siteId: ctx.siteId,
|
||
connectionId: ctx.connectionId,
|
||
snapshotId: ctx.wxrSnapshot.id,
|
||
jobId: ctx.jobId,
|
||
defaultLocale: ctx.defaultLocale,
|
||
versions: ctx.versions,
|
||
})
|
||
for await (const ev of ctx.events) {
|
||
await handler.handleEvent(ev)
|
||
}
|
||
return handler.finish()
|
||
}
|
||
}
|