diff --git a/apps/api/src/modules/import/content-writer.ts b/apps/api/src/modules/import/content-writer.ts index 13283aa..da66d6f 100644 --- a/apps/api/src/modules/import/content-writer.ts +++ b/apps/api/src/modules/import/content-writer.ts @@ -50,7 +50,7 @@ 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' +import type { ImportDb, ImportPipeline, ImportPipelineOutcome, ImportVersionBlock, WriterContext } from './import.service.js' /** 事务句柄类型(db.transaction 回调参数——从 ImportDb 的事务签名提取,避免重复泛型展开) */ type Tx = Parameters[0]>[0] @@ -149,18 +149,15 @@ interface ContentWriterCounts { * - 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: { - db: ImportDb - siteId: string - /** 快照行六列身份的 connection 侧 + 五列内容身份的解析锚点 */ - connectionId: string +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 落库锚点) */ @@ -286,8 +283,6 @@ export function createContentEventHandler(identity: { // 态 1:本快照行已存在(挂 entry——本写入器是 content 行 source_records 的唯一写入方, // 挂接不变量由写入路径自身保证)。 - // 已知死角:NULL-entry 占位行(外源 seed/未来写入方)+ 后续映射扩充重导 → 全新建分支 - // 插入撞六列唯一约束 23505;Task 6 收编(UPDATE 该行挂 entry)——见审查记录 const snapRow = await findSnapshotRecord(row) if (snapRow !== null && snapRow.contentEntryId !== null) { if (snapRow.normalizedChecksum === checksum) { @@ -315,6 +310,67 @@ export function createContentEventHandler(identity: { 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) diff --git a/apps/api/src/modules/import/import.service.ts b/apps/api/src/modules/import/import.service.ts index d82ba45..9fa08de 100644 --- a/apps/api/src/modules/import/import.service.ts +++ b/apps/api/src/modules/import/import.service.ts @@ -1,4 +1,4 @@ -// import_jobs 生命周期服务(Plan 3 Task 1,spec §7.2/§7.5、前置事实 6/13): +// import_jobs 生命周期服务(Plan 3 Task 1 + Task 6 编排,spec §7.2/§7.5、前置事实 6/8/13): // 解析站点与快照 → 幂等建 job → 状态推进(queued → running → succeeded / // succeeded_with_issues / failed)→ 报告落 //reports/import.json // (temp+rename 原子写、**先于终态落行**——行内 report_path 出现即文件存在)。 @@ -21,24 +21,37 @@ // site + wxr 快照 + 三语义版本(adapter/mapping/sanitizer——任一 bump 即新 job,对齐 // spec §7.5 “版本变化可显式触发重新标准化”;parser 是诊断信息不进键)。 // -// 管道注入:Task 6 才接真实管道(terms/content/media/routes 写入器);本任务默认**空 -// 管道**(零计数零问题直接成功),pipeline 钩子供集成测试观察 running 中间态、 -// succeeded_with_issues 与 failed 路径。 +// 管道注入:Task 6 起默认 = **编排管道**(四家写入器汇流,orchestratedPipeline)——真 +// 事件流(runWxrAdapter 读 wxr 快照副本,mapConfig.locale 接 site.default_locale——前置 +// 事实 8)→ content/term/media/route 严格串行分派 → finish 硬序(media 先于 route)→ +// counts 按键**求和**聚合(Task 6 钉死,见 ImportPipelineOutcome.counts 契约)。注入 +// pipeline 钩子供集成测试观察中间态/失败路径(注入时缺省事件流为空——Task 1 语义保持)。 +// +// stale 回收(Task 6 义务④):每次建 job 前把同 site 超过 24h 仍 queued/running 的 job +// 置 stale(崩溃/中断残留终态化)。心跳列决策留 ADR-017——现阶段无心跳列,以 +// created_at 近似(job 为单进程就地推进,24h 远超任何真实执行时长)。 +import { createReadStream } from 'node:fs' import { mkdir, rename, rm, writeFile } from 'node:fs/promises' import { dirname, join, resolve } from 'node:path' import { fileURLToPath } from 'node:url' -import { and, desc, eq } from 'drizzle-orm' +import { and, desc, eq, inArray, lt } from 'drizzle-orm' import type { NodePgDatabase } from 'drizzle-orm/node-postgres' import { ADAPTER_VERSION, MAPPING_VERSION, PARSER_VERSION, SANITIZER_VERSION, + runWxrAdapter, type AdapterEvent, + type AdapterTimings, } from '@sf/adapter-wp' import { importJobs, sites, sourceConnections, sourceSnapshots } from '../../db/schema.js' import type * as schema from '../../db/schema.js' import { isUniqueViolation } from '../source/register.service.js' +import { createContentEventHandler } from './content-writer.js' +import { createTermEventHandler } from './term-writer.js' +import { createMediaEventHandlerFromContext } from './media-writer.js' +import { DEFAULT_NEWS_PATH, DEFAULT_PRODUCTS_PATH, createRouteEventHandler, normalizeListPath } from './route-writer.js' /** 本文件位于 /apps/api/src/modules/import/ → 上五级是仓库根 */ const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..') @@ -109,37 +122,66 @@ export interface ImportJobContext { report: ImportPipelineOutcome /** * 适配器事件流(runWxrAdapter 产物:row/unsupported/error/term 按文档序交错)。 - * Task 2 加法扩展:测试经 {@link CreateImportJobInput.events} 手构注入;Task 6 编排 - * 时改为从 wxr 快照构造真流(runWxrAdapter(storagePath 流))——多个写入器共享同流 - * 时的重迭代策略(每写入器一新生成器等)由 Task 6 决定,本字段只承诺"可迭代一次"。 + * Task 6 兑现:缺省 = 从 wxr 快照副本构造的真流(mapConfig.locale=site.default_locale + * ——前置事实 8);测试经 {@link CreateImportJobInput.events} 手构注入。单消费方 + * (编排器)一遍 for-await 分派给四家写入器——只承诺"可迭代一次",无 tee 无重迭代。 */ events: AsyncIterable } /** - * 管道产出契约(Task 1 钉死,Task 6 写入器遵从): + * 管道产出契约(Task 1 钉死、Task 6 修订——counts 聚合改**求和**、加法扩 issueGroups/ + * sanitize): * - counts:计数块(键名由写入器约定,如 contentEntries/terms/routes)——原样进报告; * - issues:问题清单(简单字符串——机器可读错误码按 Plan 2 台账推迟);**非空 → job - * 终态 succeeded_with_issues**(问题全部入报告,不静默)。 + * 终态 succeeded_with_issues**(问题全部入报告,不静默); + * - issueGroups / sanitize:可选分组视图与清洗统计(编排管道产出;issues 仍是全量扁平清单)。 * - * 汇合规则(钉死):最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键合并 - * (**返回值覆盖累加器同名键**)、issues 串联(累加器在前、返回值在后)。管道抛错时 - * 无返回值,累加器现值原样进报告(部分产出可审计)。 + * 汇合规则(Task 6 钉死):最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键 + * **求和**、issues 串联(累加器在前、返回值在后)、issueGroups 逐键串联、sanitize + * 数值相加。管道抛错时无返回值,累加器现值原样进报告(部分产出可审计)。 */ export interface ImportPipelineOutcome { /** * 计数块(原样进报告)。**键契约:counts 键是跨写入器的扁平全局命名空间**—— - * 汇合规则按"返回值覆盖累加器同名键"合并(ctx.report.counts ⊕ outcome.counts), - * 同名键 **CLOBBER 不求和**:各写入器必须各持独占键名并在自身 JSDoc 声明归属 - * (Task 2 term-writer:termsProcessed/autoCreatedTerms/contentTermsCreated; - * Task 3 content-writer:contentEntries/revisionsCreated/sourceRecords/skipped/ - * unsupportedRecords/failedRecords——Task 4 media-writer 起新键名,勿复用 'skipped')。 + * 汇合按"同名键**求和**"聚合(ctx.report.counts ⊕ outcome.counts,Task 6 钉死: + * 四家写入器键前缀互斥,求和是防覆盖的安全网——Task 1 的 CLOBBER 语义废弃)。 + * 各写入器仍须各持独占键名并在自身 JSDoc 声明归属(term:termsProcessed/ + * autoCreatedTerms/contentTermsCreated;content:contentEntries/revisionsCreated/ + * sourceRecords/skipped/unsupportedRecords/failedRecords;media 六键 media 前缀; + * route 五键 routes 前缀——新写入器勿复用他家键名)。 */ counts: Record issues: string[] + /** + * 可选:按来源分组的 issues 视图(报告分组呈现)。**issues 字段仍是无分组全量 + * 清单**(既有消费方契约不变);分组键 = 四写入器前缀 + adapter(row 事件自带行级 + * 问题)+ orchestrator(编排器自身)。 + */ + issueGroups?: Partial> + /** 可选:清洗统计(row 事件 sanitizeDiffs/changed 的聚合——编排层在分派循环收集) */ + sanitize?: ImportSanitizeStats } -/** 管道(Task 6 接真实写入器;缺省空管道=零计数零问题直接成功)。抛错 → job failed */ +/** issues 分组键:四写入器前缀 + adapter(row 事件 issues 面)+ orchestrator(编排器) */ +export type ImportIssueGroup = 'content' | 'term' | 'media' | 'route' | 'adapter' | 'orchestrator' + +/** 清洗统计(Plan 2 报告纪律的导入侧镜像:changed 语义 = html !== 输入原文) */ +export interface ImportSanitizeStats { + /** changed=true 的 row 数(零 diff 等价变形也计——如 img 自闭合规范化) */ + changedRows: number + /** 按 diff 类型计数('removed-tag'/'removed-attr'——sparse:零计类型不占键) */ + byDiffType: Record +} + +/** + * 写入器共享身份基类型(Task 6 义务③:编译期防漂移)——四家写入器的 identity 参数均以 + * 此为基础(`WriterContext & { 各家专属列 }`),db/siteId/connectionId 三列的形状与命名 + * 单源约束;ImportJobContext 结构性满足(Pick 自它),编排注入即传 ctx 的这三列。 + */ +export type WriterContext = Pick + +/** 管道(缺省 = Task 6 编排管道 orchestratedPipeline——四家写入器汇流)。抛错 → job failed */ export type ImportPipeline = (ctx: ImportJobContext) => Promise export interface CreateImportJobInput { @@ -153,13 +195,17 @@ export interface CreateImportJobInput { /** 报告根(落 //reports/import.json);缺省 repoRoot/artifacts/import */ reportRoot?: string /** - * 适配器事件流(ctx.events 的注入位):Task 2 测试手构注入;缺省空流(空管道下无消费 - * 方)。Task 6 编排时由此改为从 wxr 快照构造 runWxrAdapter 真流(前置事实 8:mapConfig - * .locale 必须传 site.default_locale)。 + * 适配器事件流(ctx.events 的注入位):测试手构注入。注入而缺省 pipeline 时仍走编排 + * 管道(手构流 + 真编排);注入 pipeline 而无 events 时为空流(Task 1 语义保持); + * 两者皆缺席 = 真流(wxr 快照副本,前置事实 8 的 locale 接线在真流构造处)。 */ events?: AsyncIterable - /** 管道钩子(测试/Task 6 注入);缺省空管道 */ + /** 管道钩子(测试注入);缺省 = 编排管道 */ pipeline?: ImportPipeline + /** 主列表页路径(前置事实 9:CLI 参数来源,规范化值记入 job 报告);缺省 /products/ */ + productsPath?: string + /** 新闻列表页路径(同上);缺省 /news/ */ + newsPath?: string } export interface CreateImportJobResult { @@ -279,6 +325,20 @@ export async function createImportJob( } const site = siteRows[0]! + // 阶段 0.5:stale 回收(Task 6 义务④)——同 site 超过 24h 仍 queued/running 的 job 置 + // stale(崩溃/中断残留的终态化,finished_at 记宣判时刻)。心跳列决策留 ADR-017:现阶 + // 段无心跳列,以 created_at 近似(job 单进程就地推进,24h 远超任何真实执行时长)。 + await db + .update(importJobs) + .set({ status: 'stale', finishedAt: new Date() }) + .where( + and( + eq(importJobs.siteId, site.id), + inArray(importJobs.status, ['queued', 'running']), + lt(importJobs.createdAt, new Date(Date.now() - STALE_AFTER_MS)), + ), + ) + // 阶段 1:wxr 主输入快照解析(显式 id 优先;缺省按 (connection, kind='wxr') 最新) let wxrRow: SnapshotRow | null = null if (input.snapshotId !== undefined) { @@ -370,8 +430,23 @@ export async function createImportJob( .set({ status: 'running', startedAt: new Date(), attempt }) .where(eq(importJobs.id, jobId)) - // 阶段 5:管道(本任务缺省空管道;Task 6 接真实写入器)——抛错 → failed 终态。 - // 产出汇合规则(ImportPipeline JSDoc 钉死):返回值 counts 覆盖累加器同名键、issues 追加在后。 + // 阶段 5:管道——缺省 = 编排管道(四家写入器汇流);注入钩子优先。抛错 → failed 终态。 + // 事件流三分:注入 events 优先(手构流 + 缺省编排管道 = 测试形态)→ 注入 pipeline 而 + // 无 events = 空流(Task 1 语义保持)→ 双缺席 = 真流(wxr 快照副本 + locale 接线, + // 前置事实 8——缺省空串会让全部行被 content-writer 以 locale 不匹配拒收)。 + const productsPath = normalizeListPath(input.productsPath ?? DEFAULT_PRODUCTS_PATH) + const newsPath = normalizeListPath(input.newsPath ?? DEFAULT_NEWS_PATH) + const useRealStream = input.events === undefined && input.pipeline === undefined + const adapterTimings: AdapterTimings | null = useRealStream ? { parseMs: 0, sanitizeMs: 0 } : null + const events: AsyncIterable = + input.events !== undefined + ? input.events + : input.pipeline !== undefined + ? emptyEvents() + : realEventStream(wxrSnapshot, site.defaultLocale, adapterTimings!) + const pipeline: ImportPipeline = + input.pipeline !== undefined ? input.pipeline : orchestratedPipeline({ productsPath, newsPath }) + let failed: { code: ImportErrorCode; summary: string } | null = null const ctx: ImportJobContext = { db, @@ -385,11 +460,11 @@ export async function createImportJob( manualUrlListSnapshot, versions: VERSIONS, report: { counts: {}, issues: [] }, - events: input.events ?? emptyEvents(), + events, } let outcome: ImportPipelineOutcome | null = null try { - outcome = await (input.pipeline ?? emptyPipeline)(ctx) + outcome = await pipeline(ctx) } catch (err) { failed = { code: err instanceof ImportError ? err.code : 'PIPELINE_FAILED', @@ -399,10 +474,7 @@ export async function createImportJob( } } // 抛错时无返回值:累加器现值进报告(部分产出可审计) - const merged: ImportPipelineOutcome = - outcome === null - ? ctx.report - : { counts: { ...ctx.report.counts, ...outcome.counts }, issues: [...ctx.report.issues, ...outcome.issues] } + const merged: ImportPipelineOutcome = outcome === null ? ctx.report : mergeOutcomes(ctx.report, outcome) let finalStatus: ImportJobStatus if (failed !== null) { finalStatus = 'failed' @@ -431,8 +503,14 @@ export async function createImportJob( manual_url_list: manualUrlListSnapshot === null ? null : slimRef(manualUrlListSnapshot), }, versions: VERSIONS, + // 主列表路径(前置事实 9:CLI 参数来源记入报告——route-writer 实际消费的规范化值) + paths: { productsPath, newsPath }, counts: merged.counts, issues: merged.issues, + ...(merged.issueGroups === undefined ? {} : { issueGroups: merged.issueGroups }), + ...(merged.sanitize === undefined ? {} : { sanitize: merged.sanitize }), + // 适配器分段计时(仅真流编排——Plan 2 报告纪律的 parse/sanitize 分段在导入侧续录) + ...(adapterTimings === null ? {} : { adapterTiming: { parseMs: adapterTimings.parseMs, sanitizeMs: adapterTimings.sanitizeMs } }), error: failed, durationMs: performance.now() - startedAt, } @@ -522,11 +600,163 @@ function slimRef(ref: SnapshotRef): { return { id: ref.id, kind: ref.kind, sha256: ref.sha256, byteSize: ref.byteSize, storagePath: ref.storagePath } } -/** Task 1 空管道:零计数零问题直接成功(Task 6 以真实写入器替换缺省值) */ -async function emptyPipeline(_ctx: ImportJobContext): Promise { - return { counts: {}, issues: [] } -} - /** 缺省空事件流(input.events 缺席时的 ctx.events 占位;每次调用产新迭代器) */ async function* emptyEvents(): AsyncGenerator {} +// ---- 编排管道(Task 6 汇流——Task 5 审查移交的契约底本在此兑现) ---- + +/** stale 回收阈值:queued/running 滞留超过此时长即宣判(毫秒) */ +const STALE_AFTER_MS = 24 * 60 * 60 * 1000 + +/** wxr 快照副本绝对路径(metadata.storageRoot + storagePath——登记期固化的存储根事实) */ +function wxrSnapshotCopyPath(snapshot: SnapshotRef): string { + const root = snapshot.metadata.storageRoot + if (typeof root !== 'string' || root === '') { + throw new Error( + `wxr 快照 ${snapshot.id} metadata 缺 storageRoot(登记期不变量破裂——无法读取副本构造事件流)`, + ) + } + return join(root, snapshot.storagePath) +} + +/** + * 真事件流(惰性生成器:路径解析/读盘错误推迟到首拍 next()——落进管道执行期,由 + * createImportJob 的 PIPELINE_FAILED 收口,不留"job 已 running 而流构造抛错"的悬挂行)。 + * 前置事实 8:mapConfig.locale 必须接 site.default_locale(缺省空串 = 全行被拒收)。 + */ +async function* realEventStream( + snapshot: SnapshotRef, + locale: string, + timings: AdapterTimings, +): AsyncGenerator { + yield* runWxrAdapter(createReadStream(wxrSnapshotCopyPath(snapshot)), { + mapConfig: { locale }, + timings, + }) +} + +/** counts 按键求和(Task 6 钉死的聚合策略——就地累加 target) */ +function sumCounts(target: Record, source: Record): void { + for (const [key, value] of Object.entries(source)) { + target[key] = (target[key] ?? 0) + value + } +} + +/** Outcome 汇合(ImportPipelineOutcome JSDoc 契约):counts 求和、issues/issueGroups 串联、sanitize 相加 */ +function mergeOutcomes(acc: ImportPipelineOutcome, ret: ImportPipelineOutcome): ImportPipelineOutcome { + const counts: Record = { ...acc.counts } + sumCounts(counts, ret.counts) + const issueGroups: Partial> = {} + let hasGroups = false + for (const key of Object.keys(acc.issueGroups ?? {}) as ImportIssueGroup[]) { + issueGroups[key] = [...(acc.issueGroups![key] ?? [])] + hasGroups = true + } + for (const key of Object.keys(ret.issueGroups ?? {}) as ImportIssueGroup[]) { + issueGroups[key] = [...(issueGroups[key] ?? []), ...(ret.issueGroups![key] ?? [])] + hasGroups = true + } + const sanitize: ImportSanitizeStats | undefined = + acc.sanitize === undefined && ret.sanitize === undefined + ? undefined + : { + changedRows: (acc.sanitize?.changedRows ?? 0) + (ret.sanitize?.changedRows ?? 0), + byDiffType: {}, + } + if (sanitize !== undefined) { + for (const src of [acc.sanitize?.byDiffType, ret.sanitize?.byDiffType]) { + if (src !== undefined) sumCounts(sanitize.byDiffType, src) + } + } + return { + counts, + issues: [...acc.issues, ...ret.issues], + ...(hasGroups ? { issueGroups } : {}), + ...(sanitize === undefined ? {} : { sanitize }), + } +} + +/** + * 编排管道(Task 6 汇流契约底本——Task 5 审查移交钉死): + * - 事件期**严格串行 await** 分派 content → term → media → route:三家(term/media/ + * route)的 entry 锚点解析依赖 content 对**同事件**的落库产物(source_records. + * content_entry_id / entry_locales),禁 Promise.all——并行下锚点查询可能抢在 content + * 写入前发生; + * - finish 唯一硬序:term/content(同步、无库依赖)任意 → `await media.finish()` 先于 + * `await route.finish()`(attachment 301 反查与 asset 全表都读 media_assets——资产须已齐); + * - 分派规则(Task 3 审查移交⑥):wp:attachment 行的媒体域归 media-writer + * (media_assets 唯一写入方),content-writer 对其只留 source_records record-only 足迹 + * (前置事实 11)——两家各写各域,单一 owner; + * - counts 按键求和聚合(安全网——四家键前缀互斥);清洗统计与 row 事件行级问题在 + * 分派循环收集(RowResult 契约:issues 逐条收录为行级错误)。 + */ +function orchestratedPipeline(paths: { productsPath: string; newsPath: string }): ImportPipeline { + return async (ctx) => { + const content = createContentEventHandler({ + db: ctx.db, + siteId: ctx.siteId, + connectionId: ctx.connectionId, + snapshotId: ctx.wxrSnapshot.id, + jobId: ctx.jobId, + defaultLocale: ctx.defaultLocale, + versions: ctx.versions, + }) + const term = createTermEventHandler({ + db: ctx.db, + siteId: ctx.siteId, + connectionId: ctx.connectionId, + }) + const media = await createMediaEventHandlerFromContext(ctx) + const route = createRouteEventHandler({ + db: ctx.db, + siteId: ctx.siteId, + connectionId: ctx.connectionId, + snapshotId: ctx.wxrSnapshot.id, + defaultLocale: ctx.defaultLocale, + productsPath: paths.productsPath, + newsPath: paths.newsPath, + }) + + const sanitize: ImportSanitizeStats = { changedRows: 0, byDiffType: {} } + const adapterIssues: string[] = [] + for await (const ev of ctx.events) { + if (ev.kind === 'row') { + if (ev.changed) sanitize.changedRows++ + for (const diff of ev.sanitizeDiffs) { + sanitize.byDiffType[diff.type] = (sanitize.byDiffType[diff.type] ?? 0) + 1 + } + for (const issue of ev.issues) { + adapterIssues.push(`行级问题(${ev.row.sourceType}/${ev.row.sourceId}):${issue}`) + } + } + await content.handleEvent(ev) + await term.handleEvent(ev) + await media.handleEvent(ev) + await route.handleEvent(ev) + } + const t = term.finish() + const c = content.finish() + const m = await media.finish() + const r = await route.finish() + + const counts: Record = {} + sumCounts(counts, c.counts) + sumCounts(counts, t.counts) + sumCounts(counts, m.counts) + sumCounts(counts, r.counts) + return { + counts, + issues: [...c.issues, ...t.issues, ...m.issues, ...r.issues, ...adapterIssues], + issueGroups: { + content: c.issues, + term: t.issues, + media: m.issues, + route: r.issues, + adapter: adapterIssues, + orchestrator: [], + }, + sanitize, + } + } +} + diff --git a/apps/api/src/modules/import/media-writer.ts b/apps/api/src/modules/import/media-writer.ts index c7c6c81..b5af482 100644 --- a/apps/api/src/modules/import/media-writer.ts +++ b/apps/api/src/modules/import/media-writer.ts @@ -43,7 +43,7 @@ import type { UniversalRow } from '@sf/contracts' import { entryLocales, mediaAssets, mediaUsages, sourceRecords } from '../../db/schema.js' import { isUniqueViolation, type MediaManifestEntry } from '../source/register.service.js' import { detectMediaType } from './magic-bytes.js' -import type { ImportDb, ImportJobContext, ImportPipeline, ImportPipelineOutcome } from './import.service.js' +import type { ImportJobContext, ImportPipeline, ImportPipelineOutcome, WriterContext } from './import.service.js' const UPLOADS_MARKER = '/wp-content/uploads/' @@ -119,10 +119,7 @@ interface SidecarEntry { filePath: string } -export function createMediaEventHandler(identity: { - db: ImportDb - siteId: string - connectionId: string +export function createMediaEventHandler(identity: WriterContext & { snapshotId: string /** 引用行 locale 校验基准(与 content-writer 同拒收口径——不双报) */ defaultLocale: string diff --git a/apps/api/src/modules/import/route-writer.ts b/apps/api/src/modules/import/route-writer.ts index cbfdf2c..6693dc1 100644 --- a/apps/api/src/modules/import/route-writer.ts +++ b/apps/api/src/modules/import/route-writer.ts @@ -72,7 +72,7 @@ import type { AdapterEvent } from '@sf/adapter-wp' import type { UniversalRow } from '@sf/contracts' import { mediaAssets, routeRegistry, sites, sourceRecords, taxonomies, terms } from '../../db/schema.js' import { CONTENT_TYPE_KEY_MAP } from './content-writer.js' -import type { ImportDb, ImportPipeline, ImportPipelineOutcome } from './import.service.js' +import type { ImportPipeline, ImportPipelineOutcome, WriterContext } from './import.service.js' /** CLI --products-path 默认值(前置事实 9:记入 job metadata,本写入器收规范化值) */ export const DEFAULT_PRODUCTS_PATH = '/products/' @@ -88,8 +88,8 @@ export function foldCollisionKey(path: string): string { return path.normalize('NFC').toLowerCase().replace(/\/+$/, '') } -/** 主列表路径规范化(CLI 参数防线:补首尾斜杠) */ -function normalizeListPath(path: string): string { +/** 主列表路径规范化(CLI 参数防线:补首尾斜杠);Task 6 起编排层导入用于报告记录实际生效值 */ +export function normalizeListPath(path: string): string { const withLeading = path.startsWith('/') ? path : `/${path}` return withLeading.endsWith('/') ? withLeading : `${withLeading}/` } @@ -176,11 +176,7 @@ interface RoutePayload { * identity 注记:connectionId/snapshotId 是 entry 锚点解析与 source_snapshot_id 溯源的 * 必需输入(与 content/media 同源——Task 6 从 ImportJobContext 共享注入)。 */ -export function createRouteEventHandler(identity: { - db: ImportDb - siteId: string - /** source_records 五列身份解析的连接锚点(content-writer 同口径) */ - connectionId: string +export function createRouteEventHandler(identity: WriterContext & { /** 行溯源(route_registry.source_snapshot_id) */ snapshotId: string /** locale 校验基准(与 content/media 同拒收口径——不双报) */ @@ -443,20 +439,27 @@ export function createRouteEventHandler(identity: { /** * finish 期单条 attachment 处理:URL 层防线(link 在场 → 解析 → ?p=/&page_id= 排除 → 路径 - * 安全)先行,资产反查在后 → 301 载荷(查询/permalink 双形态)。返回 true 仅当实际写入 - * (created/updated——重复事件 no-op 不重复计入降级汇总)。 + * 安全)先行,资产反查在后 → 301 载荷(查询/permalink 双形态)。返回三态: + * - 'written':实际写入(created/updated); + * - 'resolved':目标解析成功且行全等 no-op(幂等重跑——降级处置照常成立,只是零写入); + * - 'blocked':目标缺席/歧义/防线拒绝(issue 已记,路由未登记)。 + * 降级汇总按 **去重 source_id 的 resolved∪written** 计(Task 6 口径:降级是处置事实, + * 重跑如实再现;同 attachment 重复事件不虚报——Plan 3 钉死语义)。 */ - async function writeAttachmentRoute(att: PendingAttachment, host: string): Promise { + async function writeAttachmentRoute( + att: PendingAttachment, + host: string, + ): Promise<'written' | 'resolved' | 'blocked'> { if (att.sourceUrl === '') { issues.push(`attachment 缺 link(wp:attachment/${att.sourceId},${att.rawLocation}):无页面 URL 证据——301 路由不登记`) - return false + return 'blocked' } let parsed: URL try { parsed = new URL(att.sourceUrl) } catch (err) { issues.push(`attachment link 解析失败(wp:attachment/${att.sourceId},${att.sourceUrl}):${err instanceof Error ? err.message : String(err)}——301 路由不登记`) - return false + return 'blocked' } // ?p=/&page_id= 排除口径与内容行一致(Fix 对齐:attachment link 同形态同排除—— // excluded_with_reason 语义见模块头;?attachment_id= 不在排除表,照常 query_key 建模) @@ -464,12 +467,12 @@ export function createRouteEventHandler(identity: { if (parsed.searchParams.has(param)) { counts.routesExcludedQueries++ issues.push(`查询 URL 排除(wp:attachment/${att.sourceId},${att.sourceUrl}):?${param}= 形态不登记 route_registry(route_kind 无 excluded 值——查询路由 SP3;spec §6.1/§6.2 excluded_with_reason 以计数+本 issue 落报告)`) - return false + return 'blocked' } } if (!isSafePath(parsed.pathname)) { issues.push(`attachment 路径安全校验拒绝(wp:attachment/${att.sourceId},${parsed.pathname}):含 NUL/反斜杠/控制字符/穿越段——301 路由不登记(完整校验规则 SP3 验证器)`) - return false + return 'blocked' } const assetRows = await db .select({ publicPath: mediaAssets.publicPath }) @@ -485,12 +488,12 @@ export function createRouteEventHandler(identity: { .limit(2) if (assetRows.length === 0) { issues.push(`attachment 301 目标缺席(wp:attachment/${att.sourceId},${att.sourceUrl}):media_assets 无对应资产(媒体三级对拍 miss?)——attachment 页路由未登记`) - return false + return 'blocked' } if (assetRows.length > 1) { // LIMIT 2 探针只知下界——歧义消息按 ≥2 表述(不自动选,spec §5.5) issues.push(`attachment 多资产歧义(wp:attachment/${att.sourceId}):同 source_id 命中 ≥2 行——不自动选(spec §5.5),attachment 页路由未登记`) - return false + return 'blocked' } const publicPath = assetRows[0]!.publicPath const queryKey = canonicalQueryKey(parsed) @@ -513,7 +516,7 @@ export function createRouteEventHandler(identity: { priority: 0, verificationStatus: 'verified_redirect', }) - return outcome === 'created' || outcome === 'updated' + return outcome === 'created' || outcome === 'updated' ? 'written' : 'resolved' } return { @@ -637,13 +640,15 @@ export function createRouteEventHandler(identity: { } } - // ③ 暂存 attachment 301——成功计数汇总为一条降级 issue(前置事实 10) - let downgraded = 0 + // ③ 暂存 attachment 301——降级汇总为一条 issue(前置事实 10);计数按去重 + // source_id 的已解析处置数(written∪resolved):幂等重跑 no-op 零写入但降级处置 + // 照常成立、如实再现(Plan 3 钉死);同 attachment 重复事件不虚报(去重) + const downgradedSources = new Set() for (const att of pendingAttachments) { - if (await writeAttachmentRoute(att, host)) downgraded++ + if ((await writeAttachmentRoute(att, host)) !== 'blocked') downgradedSources.add(att.sourceId) } - if (downgraded > 0) { - issues.push(`attachment 页降级(前置事实 10):${downgraded} 条 attachment 页一律 301 → 媒体文件——仅 WXR 证据、无 sitemap 解析(sitemap→html 分支留待 SP3)`) + if (downgradedSources.size > 0) { + issues.push(`attachment 页降级(前置事实 10):${downgradedSources.size} 条 attachment 页一律 301 → 媒体文件——仅 WXR 证据、无 sitemap 解析(sitemap→html 分支留待 SP3)`) } // ④ asset 行(media_assets 全表——site 的旧 URL 空间台账;source_url 是来源原文) diff --git a/apps/api/src/modules/import/term-writer.ts b/apps/api/src/modules/import/term-writer.ts index 8e9745d..cbb1303 100644 --- a/apps/api/src/modules/import/term-writer.ts +++ b/apps/api/src/modules/import/term-writer.ts @@ -36,7 +36,7 @@ import type { AdapterEvent, WxrTerm } from '@sf/adapter-wp' import type { UniversalRow } from '@sf/contracts' import { contentTerms, sourceRecords, taxonomies, terms } from '../../db/schema.js' import { isUniqueViolation } from '../source/register.service.js' -import type { ImportDb, ImportPipeline, ImportPipelineOutcome } from './import.service.js' +import type { ImportPipeline, ImportPipelineOutcome, WriterContext } from './import.service.js' /** 自建 term 的 source_id 前缀(与 wp:term 的 term_id 字符串源身份空间隔离) */ const AUTO_SOURCE_ID_PREFIX = 'auto:' @@ -94,12 +94,7 @@ export interface TermEventHandler { * issues:无 entry 可关联的 row 分类引用、表外 taxonomy 兜底、缺 term_id 的 wp:term * 事件、同 slug 双 term_id、收编撞唯一约束——全部显式报告(§7.6 禁静默)。 */ -export function createTermEventHandler(identity: { - db: ImportDb - siteId: string - /** entry 解析(source_records 稳定身份查询)用的连接锚点 */ - connectionId: string -}): TermEventHandler { +export function createTermEventHandler(identity: WriterContext): TermEventHandler { const db = identity.db const siteId = identity.siteId const counts = { termsProcessed: 0, autoCreatedTerms: 0, contentTermsCreated: 0 } diff --git a/scripts/sf.ts b/scripts/sf.ts index 7cbb904..d8e288b 100644 --- a/scripts/sf.ts +++ b/scripts/sf.ts @@ -2,7 +2,7 @@ // sf CLI(Plan 2 Task 7 + Plan 3 Task 1):commander 三命令—— // - wxr:probe:只读诊断(任意可读路径;Task 8 直读 cms 用此通道),出 JSON/HTML 报告不入库; // - source:register:调 registerSource 登记 content-addressed 快照(路径白名单校验在服务内); -// - import:调 createImportJob 建 import_jobs 并推进生命周期(Task 1 空管道骨架;Task 6 接真实管道)。 +// - import:调 createImportJob 建 import_jobs 并推进生命周期(Task 6:默认编排管道——四家写入器汇流)。 // 执行机制(Plan 2 评审钉死):根 devDep tsx 直吃 TS 源码(含 apps/api 的 Nest 源码与 // packages 的装饰器/TS 源文件),@sf/* 经根/包 tsconfig paths 直连 src——无需预构建 dist。 // 注意:CLI 只 import register.service.ts / import.service.ts(零装饰器纯函数文件), @@ -163,16 +163,30 @@ program program .command('import') - .description('导入已登记的 wxr 快照(Task 1 空管道骨架:job 生命周期 + 报告;真实管道 Task 6 接入)') + .description('导入已登记的 wxr 快照(Task 6 编排管道:term/content/media/route 四家写入器汇流 + 导入报告)') .requiredOption('--site ', '站点 key') .option('--snapshot ', 'wxr 快照 id(缺省按 connection+kind=wxr 选最新)') .option('--key ', '幂等键(缺省 = site+snapshot+版本块 自动派生;重复提交同 key 复用既有 job 不重跑)') .option('--connection ', '连接名(快照解析的 connection 侧,缺省 default)', 'default') - .action(async (opts: { site: string; snapshot?: string; key?: string; connection: string }) => { + .option('--products-path ', '产品主列表页路径(前置事实 9,规范化值记入 job 报告)', '/products/') + .option('--news-path ', '新闻主列表页路径(同上)', '/news/') + .action(async (opts: { + site: string + snapshot?: string + key?: string + connection: string + productsPath: string + newsPath: string + }) => { // getDb 走 DATABASE_URL_TEST ?? DATABASE_URL 回退链——临时库冒烟只需 env 覆盖 const db = getDb() // exactOptionalPropertyTypes:缺席键不传 undefined 值(与 source:register 同款) - const input: CreateImportJobInput = { siteKey: opts.site, connectionName: opts.connection } + const input: CreateImportJobInput = { + siteKey: opts.site, + connectionName: opts.connection, + productsPath: opts.productsPath, + newsPath: opts.newsPath, + } if (opts.snapshot !== undefined) input.snapshotId = opts.snapshot if (opts.key !== undefined) input.idempotencyKey = opts.key try { diff --git a/tests/integration/import-full.test.ts b/tests/integration/import-full.test.ts new file mode 100644 index 0000000..5341bcf --- /dev/null +++ b/tests/integration/import-full.test.ts @@ -0,0 +1,470 @@ +// Plan 3 Task 6 失败测试(先于实现):编排集成 + 导入报告——四家写入器汇流。 +// 覆盖契约(对照计划 Task 6 与前置事实 8/11/13、Task 3/5 审查移交清单): +// 1. 全链(register wxr+media+media-map → 默认编排管道·真事件流):content_entries=3 / +// source_records=5 / terms=2+自建 0 / content_terms=1 / media_assets=2 / routes=12; +// published_revision_id 全库 NULL(导入不发布——ADR-010);mapConfig.locale 接线 +// (site.default_locale='zh-CN' → 全部 source_records.locale='zh-CN'——locale 只可能 +// 来自编排接线,mini 无自带 locale);报告 = 四家 counts 18 键全量 + 清洗统计 + +// 版本块 + issues 分组 + 路径块 + 适配器分段计时; +// 2. 二导幂等(换 key 重跑):全部 counts 归零语义(termsProcessed/mediaDanglingRefs 等 +// 按事件计键如实再现)、数据库行数不变、job 仍 succeeded_with_issues +// (attachment 降级汇总 issue 必在——Plan 3 钉死语义:降级是处置事实非写入计数); +// 3. 无 media 快照失败路径:media 域全 missing 报告不失败(空 manifest 语义已备); +// 4. NULL-entry 占位行收编(Task 3 审查移交①):六列命中 content_entry_id NULL 行 → +// UPDATE 挂 entry(不撞 23505、sourceRecords 不虚增); +// 5. stale 回收(义务④):同 site 超 24h 的 queued/running 置 stale,新近 job 不受影响; +// 6. REPORT_WRITE_FAILED(Task 1 遗留):报告写盘失败 → failed 终态、report_path 不落。 +// 运行于 globalSetup 临时库(ADR-015);storage/报告根 os.tmpdir,finally 清理。 +import { mkdir, 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 { createImportJob } from '../../apps/api/src/modules/import/import.service.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') + +const PNG_SIG = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]) + +/** 伪 PNG 体(魔数校验只看签名——载荷唯一避免跨测试撞哈希) */ +function pngBytes(seed: string): Buffer { + return Buffer.concat([PNG_SIG, Buffer.from(`png-body:${seed}`, 'latin1')]) +} + +/** 本文件写入的行清场:FK 反序(与 import-routes 同清单;两张不可变表走触发器开关) */ +async function cleanupSiteRows(db: Db, siteId: string): Promise { + await db.execute(`delete from route_registry where site_id = '${siteId}'`) + await db.execute(`delete from media_usages where site_id = '${siteId}'`) + await db.execute(`delete from media_assets where site_id = '${siteId}'`) + 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}'`) +} + +interface Work { + work: string + storageRoot: string + reportRoot: string + allowedRoots: string[] +} + +async function makeWork(prefix: string): Promise { + const work = await mkdtemp(join(tmpdir(), `${prefix}-`)) + return { + work, + storageRoot: join(work, 'storage'), + reportRoot: join(work, 'artifacts', 'import'), + allowedRoots: [FIXTURES_ROOT, work], + } +} + +/** 全量登记(wxr + 媒体目录 + 侧车)——形态同 import-routes 全链用例;locale 可选注入 */ +async function registerFull( + db: Db, + w: Work, + spec: { locale?: string } = {}, +): Promise<{ key: string; siteId: string; wxrId: string }> { + const mediaDir = join(w.work, 'media') + await mkdir(join(mediaDir, '2026/10'), { recursive: true }) + await writeFile(join(mediaDir, '2026/10/fixture-a.png'), pngBytes(`full-a-${rand()}`)) + const guidB = `b-${rand()}.png` + await writeFile(join(mediaDir, guidB), pngBytes(`full-b-${rand()}`)) + const sidecarPath = join(w.work, 'media-map.json') + await writeFile( + sidecarPath, + JSON.stringify([ + { sourceUrl: 'https://fixture.example/wp-content/uploads/2026/10/fixture-b.png', filePath: guidB }, + ]), + ) + const key = `sffull-${rand()}` + const reg = await registerSource(db, { + siteKey: key, + wxrPath: WXR_FIXTURE, + mediaDir, + mediaMapPath: sidecarPath, + allowedRoots: w.allowedRoots, + storageRoot: w.storageRoot, + reportRoot: join(w.work, 'artifacts', 'register'), + ...(spec.locale === undefined ? {} : { siteDefaults: { defaultLocale: spec.locale } }), + }) + return { key, siteId: reg.siteId, wxrId: reg.snapshots.find((s) => s.kind === 'wxr')!.id } +} + +/** 仅 wxr 登记(失败路径/收编场景——media_bundle/manual_url_list 缺席) */ +async function registerWxrOnly(db: Db, w: Work): Promise<{ key: string; siteId: string; wxrId: string }> { + const key = `sffull-${rand()}` + const reg = await registerSource(db, { + siteKey: key, + wxrPath: WXR_FIXTURE, + allowedRoots: w.allowedRoots, + storageRoot: w.storageRoot, + reportRoot: join(w.work, 'artifacts', 'register'), + }) + return { key, siteId: reg.siteId, wxrId: reg.snapshots.find((s) => s.kind === 'wxr')!.id } +} + +/** 导入报告读取(默认编排管道的产物断言面) */ +async function readReport(w: Work, jobId: string): Promise { + return JSON.parse(await readFile(join(w.reportRoot, jobId, 'reports', 'import.json'), 'utf8')) +} + +/** 单值计数便捷 */ +async function countOf(db: Db, sql: string): Promise { + const rows = (await db.execute(sql)) as any + return Number(rows[0].n) +} + +describe('编排集成 + 导入报告(四家写入器汇流)', () => { + it('全链(真事件流 + 默认编排管道):18 键 counts 全量、published 全 NULL、locale 接线、清洗统计、issues 分组、报告路径/版本/计时块', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-chain') + try { + // locale 接线断言的实证设计(前置事实 8):site 默认 locale 取 'zh-CN'——mini 无 + // 自带 locale,行 locale 只可能来自编排接线 mapConfig.locale=site.default_locale + const { key, siteId, wxrId } = await registerFull(db, w, { locale: 'zh-CN' }) + try { + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxrId, + reportRoot: w.reportRoot, + }) + expect(result.reused).toBe(false) + // 悬空引用(media 203/204)×2 + attachment 降级汇总 ×1 → issues 非空 + expect(result.status).toBe('succeeded_with_issues') + + // 数据库行数对拍(计划 Task 6 钉死清单) + expect(await countOf(db, `select count(*)::int as n from content_entries where site_id = '${siteId}'`)).toBe(3) + expect(await countOf(db, `select count(*)::int as n from source_records where site_id = '${siteId}'`)).toBe(5) + expect(await countOf(db, `select count(*)::int as n from terms where site_id = '${siteId}'`)).toBe(2) + expect(await countOf(db, `select count(*)::int as n from content_terms where site_id = '${siteId}'`)).toBe(1) + expect(await countOf(db, `select count(*)::int as n from media_assets where site_id = '${siteId}'`)).toBe(2) + expect(await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}'`)).toBe(12) + + // published 全库 NULL(导入不发布——ADR-010)+ current 全设(导入产出工作修订) + expect( + await countOf(db, `select count(*)::int as n from entry_locales where site_id = '${siteId}' and published_revision_id is not null`), + ).toBe(0) + expect( + await countOf(db, `select count(*)::int as n from entry_locales where site_id = '${siteId}' and current_revision_id is not null`), + ).toBe(3) + + // locale 接线(前置事实 8):全部 source_records 行 locale = site 默认(非空串) + const locales = (await db.execute(`select distinct locale from source_records where site_id = '${siteId}'`)) as any + expect(locales.map((r: any) => r.locale).sort()).toStrictEqual(['zh-CN']) + + // attachment 分派规则(前置事实 11):2 条 attachment 行是 record-only 足迹(无 entry) + expect( + await countOf(db, `select count(*)::int as n from source_records where site_id = '${siteId}' and source_type = 'wp:attachment' and content_entry_id is null`), + ).toBe(2) + + // content_terms:post 101 的 category 引用(is_primary=true——WP 单主分类近似) + // (投影转 Boolean——postgres.js 原型差异会让 toStrictEqual 报"无视觉差异"失败) + const ct = (await db.execute(`select ct.is_primary, t.slug from content_terms ct + join terms t on t.id = ct.term_id where ct.site_id = '${siteId}'`)) as any + expect(ct.map((r: any) => ({ isPrimary: Boolean(r.is_primary), slug: r.slug }))).toStrictEqual([ + { isPrimary: true, slug: 'company' }, + ]) + + // 报告:四家 counts 18 键全量(全局命名空间求和聚合——键前缀互斥) + const report = await readReport(w, result.jobId) + expect(report.counts).toStrictEqual({ + contentEntries: 3, + revisionsCreated: 3, + sourceRecords: 5, + skipped: 0, + termsProcessed: 2, + autoCreatedTerms: 0, + contentTermsCreated: 1, + mediaRegistered: 2, + mediaFromSidecar: 1, + mediaMissing: 0, + mediaQuarantined: 0, + mediaDanglingRefs: 2, + mediaAmbiguous: 0, + routesCreated: 12, + routesUpdated: 0, + routesRedirects: 2, + routesCollisions: 0, + routesExcludedQueries: 0, + }) + + // 清洗统计(分派循环收集 row 事件 sanitizeDiffs/changed):mini 仅 post 101 正文 + // img 自闭合规范化零 diff 变形(changedRows=1、无 diff 条目——sparse 形态) + expect(report.sanitize).toStrictEqual({ changedRows: 1, byDiffType: {} }) + + // issues 分组(按写入器前缀):flat issues 保持全量(3 条),分组归位 + expect(report.issues).toHaveLength(3) + expect(report.issueGroups).toStrictEqual({ + content: [], + term: [], + media: report.issues.filter((i: string) => i.includes('悬空')), + route: report.issues.filter((i: string) => i.includes('attachment 页降级')), + adapter: [], + orchestrator: [], + }) + expect(report.issueGroups.media).toHaveLength(2) + expect(report.issueGroups.route).toHaveLength(1) + + // 路径块(前置事实 9:CLI 参数来源记入报告——缺省值)+ 版本块 + 分段计时(真流) + expect(report.paths).toStrictEqual({ productsPath: '/products/', newsPath: '/news/' }) + expect(report.versions).toStrictEqual({ + adapter: report.versions.adapter, + mapping: report.versions.mapping, + sanitizer: report.versions.sanitizer, + parser: report.versions.parser, + }) + expect(Object.keys(report.versions).sort()).toStrictEqual(['adapter', 'mapping', 'parser', 'sanitizer']) + expect(typeof report.adapterTiming.parseMs).toBe('number') + expect(typeof report.adapterTiming.sanitizeMs).toBe('number') + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) + + it('二导幂等(换 key):全部写入计数归零、数据库行数不变、job 仍 succeeded_with_issues(降级汇总 issue 必在)', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-idem') + try { + const { key, siteId, wxrId } = await registerFull(db, w) + try { + const base = { siteKey: key, snapshotId: wxrId, reportRoot: w.reportRoot } as const + const first = await createImportJob(db, { ...base }) + expect(first.status).toBe('succeeded_with_issues') + + const again = await createImportJob(db, { ...base, idempotencyKey: `sffull-rerun-${rand()}` }) + expect(again.reused).toBe(false) + expect(again.status).toBe('succeeded_with_issues') + + // 写入计数全零语义:按事件计的键(termsProcessed/mediaDanglingRefs)如实再现 + const report = await readReport(w, again.jobId) + expect(report.counts).toStrictEqual({ + contentEntries: 0, + revisionsCreated: 0, + sourceRecords: 0, + skipped: 3, + termsProcessed: 2, + autoCreatedTerms: 0, + contentTermsCreated: 0, + mediaRegistered: 0, + mediaFromSidecar: 0, + mediaMissing: 0, + mediaQuarantined: 0, + mediaDanglingRefs: 2, + mediaAmbiguous: 0, + routesCreated: 0, + routesUpdated: 0, + routesRedirects: 0, + routesCollisions: 0, + routesExcludedQueries: 0, + }) + + // 数据库行数不变(幂等:零新行零修订零更新) + expect(await countOf(db, `select count(*)::int as n from content_entries where site_id = '${siteId}'`)).toBe(3) + expect(await countOf(db, `select count(*)::int as n from content_revisions where site_id = '${siteId}'`)).toBe(3) + expect(await countOf(db, `select count(*)::int as n from source_records where site_id = '${siteId}'`)).toBe(5) + expect(await countOf(db, `select count(*)::int as n from terms where site_id = '${siteId}'`)).toBe(2) + expect(await countOf(db, `select count(*)::int as n from content_terms where site_id = '${siteId}'`)).toBe(1) + expect(await countOf(db, `select count(*)::int as n from media_assets where site_id = '${siteId}'`)).toBe(2) + expect(await countOf(db, `select count(*)::int as n from media_usages where site_id = '${siteId}'`)).toBe(3) + expect(await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}'`)).toBe(12) + + // 降级汇总 issue 必在(Plan 3 钉死:降级是处置事实,重跑如实再现)+ 悬空 2 条 + const downgrade = report.issues.filter((i: string) => i.includes('attachment 页降级')) + expect(downgrade).toHaveLength(1) + expect(downgrade[0]).toContain('2 条') + expect(report.issues.filter((i: string) => i.includes('悬空'))).toHaveLength(2) + expect(report.issues).toHaveLength(3) + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) + + it('无 media 快照(只 wxr):media 域全 missing 报告不失败(空 manifest 语义)+ 主列表路径参数入报告', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-nomedia') + try { + const { key, siteId, wxrId } = await registerWxrOnly(db, w) + try { + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxrId, + reportRoot: w.reportRoot, + productsPath: '/shop/', + }) + expect(result.status).toBe('succeeded_with_issues') + + // 空清单语义:2 attachment 全 missing(不插行)、5 引用全悬空、路由 8 行 + // (3 站点页 + 3 内容 + 2 分类——301/asset 无资产可指) + expect(await countOf(db, `select count(*)::int as n from media_assets where site_id = '${siteId}'`)).toBe(0) + expect(await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}'`)).toBe(8) + expect( + await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}' and route_path = '/shop/'`), + ).toBe(1) + expect( + await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}' and route_path = '/products/'`), + ).toBe(0) + + const report = await readReport(w, result.jobId) + expect(report.counts.mediaMissing).toBe(2) + expect(report.counts.mediaRegistered).toBe(0) + expect(report.counts.mediaDanglingRefs).toBe(5) + expect(report.counts.routesCreated).toBe(8) + expect(report.paths).toStrictEqual({ productsPath: '/shop/', newsPath: '/news/' }) + // attachment 301 目标缺席逐条报告(§7.6)+ missing/悬空——全部在 issues + expect(report.issues.filter((i: string) => i.includes('301 目标缺席'))).toHaveLength(2) + expect(report.issues.filter((i: string) => i.includes('missing'))).toHaveLength(2) + expect(report.issues.filter((i: string) => i.includes('悬空'))).toHaveLength(5) + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) + + it('NULL-entry 占位行收编(Task 3 审查移交①):六列命中 content_entry_id NULL 行 → UPDATE 挂 entry、不撞 23505、sourceRecords 不虚增', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-adopt') + try { + const { key, siteId, wxrId } = await registerWxrOnly(db, w) + try { + // 外源占位行(seed 模拟:六列身份 = wp:post/101 本快照、entry NULL) + const seedJob = (await db.execute(`insert into import_jobs (id, site_id, source_snapshot_id, status, + mapping_version, idempotency_key, attempt) values (gen_random_uuid(), '${siteId}', '${wxrId}', + 'succeeded', 'seed', 'seed-${rand()}', 1) returning id`)) as any + const connRow = (await db.execute(`select source_connection_id from source_snapshots where id = '${wxrId}'`)) as any + const seeded = (await db.execute(`insert into source_records (id, site_id, source_connection_id, + source_snapshot_id, import_job_id, source_type, source_id, locale, raw_location, parse_status, + content_entry_id) values (gen_random_uuid(), '${siteId}', '${connRow[0].source_connection_id}', + '${wxrId}', '${seedJob[0].id}', 'wp:post', '101', 'en', 'wxr://item/101', 'imported', NULL) + returning id`)) as any + + const result = await createImportJob(db, { siteKey: key, snapshotId: wxrId, reportRoot: w.reportRoot }) + expect(result.status).toBe('succeeded_with_issues') + + // 占位行被收编:挂 entry + checksum 前移(非新建行——sourceRecords=4 不是 5) + const adopted = (await db.execute(`select content_entry_id, normalized_checksum from source_records + where id = '${seeded[0].id}'`)) as any + expect(adopted[0].content_entry_id).not.toBeNull() + expect(adopted[0].normalized_checksum).not.toBeNull() + const el = (await db.execute(`select current_revision_id, published_revision_id from entry_locales + where site_id = '${siteId}' and content_entry_id = '${adopted[0].content_entry_id}'`)) as any + expect(el[0].current_revision_id).not.toBeNull() + expect(el[0].published_revision_id).toBeNull() + + const report = await readReport(w, result.jobId) + expect(report.counts.contentEntries).toBe(3) + expect(report.counts.revisionsCreated).toBe(3) + expect(report.counts.sourceRecords).toBe(4) // 101 收编非新建(2 attachment + 111/121) + expect(report.counts.skipped).toBe(0) + // 收编行照常参与 term 关联与路由(101 的 entry 锚点就位) + expect(report.counts.contentTermsCreated).toBe(1) + expect(report.counts.routesCreated).toBe(8) + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) + + it('stale 回收(义务④):同 site 超 24h 的 queued/running 置 stale;新近 job 不受影响;新 job 正常推进', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-stale') + try { + const { key, siteId, wxrId } = await registerWxrOnly(db, w) + try { + const oldH = 25 + const freshH = 1 + const seed = async (status: string, hoursAgo: number, idem: string): Promise => { + const rows = (await db.execute(`insert into import_jobs (id, site_id, source_snapshot_id, status, + mapping_version, idempotency_key, attempt, started_at, created_at) values (gen_random_uuid(), + '${siteId}', '${wxrId}', '${status}', 'seed', '${idem}', 1, + now() - interval '${hoursAgo} hours', now() - interval '${hoursAgo} hours') returning id`)) as any + return rows[0].id + } + const oldRunning = await seed('running', oldH, `stale-run-${rand()}`) + const oldQueued = await seed('queued', oldH + 2, `stale-q-${rand()}`) + const freshRunning = await seed('running', freshH, `stale-fresh-${rand()}`) + + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxrId, + reportRoot: w.reportRoot, + pipeline: async () => ({ counts: {}, issues: [] }), + }) + expect(result.status).toBe('succeeded') + + const st = (rows: string[]) => + (db.execute(`select status, finished_at from import_jobs where id in (${rows.map((r) => `'${r}'`).join(',')})`) as any) as Promise + const oldRows = await st([oldRunning, oldQueued]) + for (const r of oldRows) { + expect(r.status).toBe('stale') + expect(r.finished_at).not.toBeNull() + } + const freshRows = await st([freshRunning]) + expect(freshRows[0].status).toBe('running') // 未超时不动(心跳列决策留 ADR-017) + expect(freshRows[0].finished_at).toBeNull() + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) + + it('REPORT_WRITE_FAILED(Task 1 遗留):报告目录不可建 → failed 终态、error_code 落列、report_path 不落', async () => { + await withTestDb(async (db) => { + const w = await makeWork('sf-full-rwfail') + try { + const { key, siteId, wxrId } = await registerWxrOnly(db, w) + try { + // 用普通文件占住报告根:mkdir //reports 必败(ENOTDIR/EEXIST) + const blocker = join(w.work, 'blocker') + await writeFile(blocker, 'not a directory') + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxrId, + reportRoot: blocker, + pipeline: async () => ({ counts: { ok: 1 }, issues: [] }), + }) + expect(result.status).toBe('failed') + expect(result.reportPath).toBeNull() + const job = (await db.execute(`select status, error_code, error_summary, report_path, finished_at + from import_jobs where id = '${result.jobId}'`)) as any + expect(job[0].status).toBe('failed') + expect(job[0].error_code).toBe('REPORT_WRITE_FAILED') + expect(String(job[0].error_summary)).toMatch(/^\[report\] /) + expect(job[0].report_path).toBeNull() // 文件不存在不悬挂路径 + expect(job[0].finished_at).not.toBeNull() + } finally { + await cleanupSiteRows(db, siteId) + } + } finally { + await rm(w.work, { recursive: true, force: true }) + } + }) + }) +}) diff --git a/tests/integration/import-job.test.ts b/tests/integration/import-job.test.ts index b11fd7b..fe9eb13 100644 --- a/tests/integration/import-job.test.ts +++ b/tests/integration/import-job.test.ts @@ -11,7 +11,7 @@ // 显式 id 不存在或属他站 → SNAPSHOT_NOT_FOUND;站点不存在 → SITE_NOT_FOUND;均零 job 行; // 5. failed 路径:管道抛错 → running→failed(error_code/error_summary/finished_at, // '[pipeline]' 前缀),重提交同 key 复用失败 job 不重跑;管道抛 ImportError 保留原 code; -// 6. Outcome 契约:ctx.report 累加 ⊕ 返回值(counts 同名键返回值覆盖)→ succeeded_with_issues; +// 6. Outcome 契约:ctx.report 累加 ⊕ 返回值(counts 同名键**求和**——Task 6 编排聚合)→ succeeded_with_issues; // 7. 复用口径忠实化:reuse 返回只含 job 行锚定字段(快照/版本字段缺席); // 8. 23505 收敛确定性探针:未提交孪生 job 持有幂等键(真实 PG 锁语义,零代理)。 // 运行于 globalSetup 临时库(ADR-015);storage/报告根用 os.tmpdir 临时目录,finally 清理。 @@ -434,7 +434,7 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { }) }) - it('Outcome 契约:ctx.report 累加 ⊕ 返回值(返回值 counts 覆盖同名键),issues 非空 → succeeded_with_issues,计数与问题入报告', async () => { + it('Outcome 契约:ctx.report 累加 ⊕ 返回值按**键求和**(Task 6 编排聚合),issues 非空 → succeeded_with_issues,计数与问题入报告', async () => { await withTestDb(async (db) => { const ctx = await makeCtx() try { @@ -449,7 +449,7 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { pipeline: async (jobCtx) => { // 写入器式累加(Task 6 写入器的使用形态) jobCtx.report.counts['terms'] = 2 - jobCtx.report.counts['contentEntries'] = 999 // 被返回值同名键覆盖 + jobCtx.report.counts['contentEntries'] = 100 // 与返回值同名键**求和**(Task 6 钉死:防覆盖安全网) jobCtx.report.issues.push('自建缺失 term:1(category/uncategorized)') return { counts: { contentEntries: 3, routes: 8 }, @@ -468,8 +468,8 @@ describe('import_jobs 生命周期(Task 1 骨架)', () => { await readFile(join(ctx.importReportRoot, result.jobId, 'reports', 'import.json'), 'utf8'), ) as { status: string; counts: Record; issues: string[] } expect(report.status).toBe('succeeded_with_issues') - // 汇合规则:counts 按键合并(返回值覆盖累加器同名键)、issues 累加器在前返回值在后 - expect(report.counts).toStrictEqual({ terms: 2, contentEntries: 3, routes: 8 }) + // 汇合规则:counts 按键**求和**(Task 6 钉死——CLOBBER 废弃)、issues 累加器在前返回值在后 + expect(report.counts).toStrictEqual({ terms: 2, contentEntries: 103, routes: 8 }) expect(report.issues).toStrictEqual([ '自建缺失 term:1(category/uncategorized)', '悬空 attachment 引用:2 条(mini 203/204)',