From 0e8b2b53563f632488a6398c4fda19f6fa1caf1c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 21 Oct 2026 15:30:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(import):=20job=20=E9=AA=A8=E6=9E=B6?= =?UTF-8?q?=E4=B8=8E=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Plan 3 Task 1：sf import 命令骨架 + import_jobs 生命周期（空管道，Task 6 接真实管道）。 - import.service.ts（零装饰器）：createImportJob——站点/快照解析（缺省按 (connection, kind='wxr') 选最新；三快照绑定按前置事实 13 记入报告，job 行不扩列）、幂等建行 （默认 key=site+wxr快照+三语义版本；重复提交复用既有 job 不重跑，并发撞唯一约束 重查收敛）、状态推进 queued→running→succeeded/failed（attempt/started_at/finished_at/ error_code/error_summary）、报告落 artifacts/import//reports/import.json （全四版本块：adapter/mapping/sanitizer/parser；job 行落 mapping_version） - 可预期错误带 code：SITE_NOT_FOUND / SNAPSHOT_NOT_FOUND / NO_WXR_SNAPSHOT / PIPELINE_FAILED——解析期失败零 job 行 - adapter 事件加法式补 postId（前置事实 12）：unsupported/error 事件可选携带 String(postId)，缺失时键缺席不硬造 source_id；row 事件与报告聚合零影响、不触发版本 bump - canonicalJson 提取至 apps/api/src/common/canonical.ts 共享导出（前置事实 6）， register 侧改 import 零行为变更（既有清单哈希测试守护） - sf CLI + import 命令（--site/--snapshot/--key/--connection），输出 job 摘要与版本块 - 测试：tests/integration/import-job.test.ts（5 例：生命周期观察/最新快照解析/幂等复用 与并发收敛/可预期失败零落行/failed 路径）+ adapter postId 3 例 + canonical 单测 6 例 （265→279 全绿）；typecheck 四段链/lint 绿；临时库 CLI 冒烟（register→import→复用）通过 Co-Authored-By: Claude Code --- apps/api/src/common/canonical.ts | 22 + apps/api/src/modules/import/import.module.ts | 27 ++ apps/api/src/modules/import/import.service.ts | 436 ++++++++++++++++++ .../src/modules/source/register.service.ts | 18 +- ...2026-10-21-slice-import-pipeline-phase3.md | 8 +- packages/adapter-wordpress/src/adapter.ts | 19 +- .../adapter-wordpress/test/adapter.test.ts | 36 ++ scripts/sf.ts | 57 ++- tests/integration/import-job.test.ts | 411 +++++++++++++++++ tests/unit/canonical.test.ts | 46 ++ 10 files changed, 1052 insertions(+), 28 deletions(-) create mode 100644 apps/api/src/common/canonical.ts create mode 100644 apps/api/src/modules/import/import.module.ts create mode 100644 apps/api/src/modules/import/import.service.ts create mode 100644 tests/integration/import-job.test.ts create mode 100644 tests/unit/canonical.test.ts diff --git a/apps/api/src/common/canonical.ts b/apps/api/src/common/canonical.ts new file mode 100644 index 0000000..fcaf3bb --- /dev/null +++ b/apps/api/src/common/canonical.ts @@ -0,0 +1,22 @@ +// canonical JSON 序列化（Plan 3 Task 1 从 register.service.ts 提取，前置事实 6）： +// 对象键递归字典序排序、无空白——media_bundle 清单哈希（register）与 normalized +// checksum（阶段 3 导入，spec §7.5）共用的序列化基形，单一实现避免两处定义漂移。 +// 定义钉死（测试独立重算对拍，见 tests/unit/canonical.test.ts）： +// - 数组：元素保序（顺序是语义的一部分），逐元素递归； +// - 对象：键按 Array.prototype.sort 默认序（UTF-16 码元）排序后 `"key":value` 逗号连接； +// - 原始值：JSON.stringify 语义（string/number/boolean/null）。 +// register 侧改 import 本函数——零行为变更（既有 media_bundle 清单哈希测试守护）。 + +/** canonical JSON（对象键递归排序、无空白）——哈希输入的序列化基形 */ +export function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]` + if (value !== null && typeof value === 'object') { + const record = value as Record + const body = Object.keys(record) + .sort() + .map((k) => `${JSON.stringify(k)}:${canonicalJson(record[k])}`) + .join(',') + return `{${body}}` + } + return JSON.stringify(value) +} diff --git a/apps/api/src/modules/import/import.module.ts b/apps/api/src/modules/import/import.module.ts new file mode 100644 index 0000000..d54b5eb --- /dev/null +++ b/apps/api/src/modules/import/import.module.ts @@ -0,0 +1,27 @@ +// import 域 Nest 模块（Plan 3 Task 1）：注册 ImportJobService 供 standalone context +// 使用（NestFactory.createApplicationContext(ImportModule)——本切片 API 不监听端口，ADR-002）。 +// 与 source.module.ts 同构：逻辑全在 import.service.ts 的零装饰器独立函数里，本类只提供 +// DI 语境下的调用面；'reflect-metadata' 置顶（CLI/测试直接 import import.service.ts 纯函数 +// 文件，不经本模块，故无需提前加载 reflect-metadata）。 +import 'reflect-metadata' +import { Injectable, Module } from '@nestjs/common' +import { + createImportJob, + type CreateImportJobInput, + type CreateImportJobResult, + type ImportDb, +} from './import.service.js' + +/** createImportJob 的 Nest 注入壳（standalone context / 后续 REST 通道复用同一实现） */ +@Injectable() +export class ImportJobService { + create(db: ImportDb, input: CreateImportJobInput): Promise { + return createImportJob(db, input) + } +} + +@Module({ + providers: [ImportJobService], + exports: [ImportJobService], +}) +export class ImportModule {} diff --git a/apps/api/src/modules/import/import.service.ts b/apps/api/src/modules/import/import.service.ts new file mode 100644 index 0000000..19b43a0 --- /dev/null +++ b/apps/api/src/modules/import/import.service.ts @@ -0,0 +1,436 @@ +// import_jobs 生命周期服务（Plan 3 Task 1，spec §7.2/§7.5、前置事实 6/13）： +// 解析站点与快照 → 幂等建 job → 状态推进（queued → running → succeeded/failed）→ +// 报告落 //reports/import.json。 +// +// 本文件与 register.service.ts 同纪律保持 **零装饰器**：CLI（scripts/sf.ts 经 tsx）与 +// 集成测试直接 import 本文件，不拉起 Nest 装饰器执行；Nest 侧由 import.module.ts 薄包装。 +// +// 三快照绑定（前置事实 13）：job.source_snapshot_id 指 **wxr 快照**（主输入）； +// media_bundle / manual_url_list 不入 job 行（不扩列），按 (wxr 快照的 connection, kind) +// 解析最新记入 job 报告 snapshots 段——导入管道（Task 4/6）从报告/上下文取用。 +// +// 幂等：import_jobs.idempotency_key UNIQUE——重复提交同 key **复用既有 job 不重跑** +// （含失败 job：重试须换 key 或等 Task 6 的重试语义，本任务钉死“不静默重跑”）；并发 +// 同 key 由唯一约束收敛（插入撞 23505 → 重查复用）。 +// +// 版本块：四常量单源来自 @sf/adapter-wp 的 versions（ADAPTER/MAPPING/SANITIZER/PARSER）； +// job 行只落 mapping_version（MAPPING_VERSION），全四项进报告。默认幂等键 = +// site + wxr 快照 + 三语义版本（adapter/mapping/sanitizer——任一 bump 即新 job，对齐 +// spec §7.5 “版本变化可显式触发重新标准化”；parser 是诊断信息不进键）。 +// +// 管道注入：Task 6 才接真实管道（terms/content/media/routes 写入器）；本任务默认**空 +// 管道**（直接成功），pipeline 钩子供集成测试观察 running 中间态与 failed 路径。 +import { mkdir, 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 type { NodePgDatabase } from 'drizzle-orm/node-postgres' +import { + ADAPTER_VERSION, + MAPPING_VERSION, + PARSER_VERSION, + SANITIZER_VERSION, +} 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' + +/** 本文件位于 /apps/api/src/modules/import/ → 上五级是仓库根 */ +const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..') + +export type ImportDb = NodePgDatabase + +export type ImportJobStatus = (typeof schema.importJobStatus.enumValues)[number] + +export type ImportErrorCode = + | 'SITE_NOT_FOUND' + | 'SNAPSHOT_NOT_FOUND' + | 'NO_WXR_SNAPSHOT' + | 'PIPELINE_FAILED' + +/** 导入失败（可预期错误）：CLI 以 code 区分呈现；不裹程序 bug（那些原样上抛） */ +export class ImportError extends Error { + readonly code: ImportErrorCode + constructor(code: ImportErrorCode, message: string) { + super(message) + this.name = 'ImportError' + this.code = code + } +} + +/** 版本块（报告/上下文携带；job 行只落 mapping_version） */ +export interface ImportVersionBlock { + adapter: string + mapping: string + sanitizer: string + parser: string +} + +/** 已解析快照引用（报告与管道上下文的载荷；metadata 含 storageRoot 等登记期事实） */ +export interface SnapshotRef { + id: string + kind: 'wxr' | 'media_bundle' | 'manual_url_list' + sha256: string + byteSize: number + storagePath: string + capturedAt: Date + metadata: Record +} + +/** 管道执行上下文：Task 6 真实管道的全部输入（站点 locale 接缝见前置事实 8） */ +export interface ImportJobContext { + db: ImportDb + jobId: string + siteId: string + siteKey: string + defaultLocale: string + connectionId: string + /** 主输入：wxr 快照（job.source_snapshot_id 指向它） */ + wxrSnapshot: SnapshotRef + /** 侧输入（前置事实 13）：按 (connection, kind) 解析的最新快照，可能缺席（null） */ + mediaBundleSnapshot: SnapshotRef | null + manualUrlListSnapshot: SnapshotRef | null + versions: ImportVersionBlock +} + +/** 管道钩子（Task 6 接真实写入器；缺省空管道=直接成功）。抛错 → job failed */ +export type ImportPipeline = (ctx: ImportJobContext) => Promise + +export interface CreateImportJobInput { + siteKey: string + /** 显式 wxr 快照 id；缺省按 (connection, kind='wxr') 选最新 */ + snapshotId?: string + /** 幂等键；缺省 = site + wxr 快照 + 三语义版本 自动派生 */ + idempotencyKey?: string + /** 连接名（最新快照解析的 connection 侧）；缺省 'default'（与 register 缺省一致） */ + connectionName?: string + /** 报告根（落 //reports/import.json）；缺省 repoRoot/artifacts/import */ + reportRoot?: string + /** 管道钩子（测试/Task 6 注入）；缺省空管道 */ + pipeline?: ImportPipeline +} + +export interface CreateImportJobResult { + jobId: string + /** job 终态（或复用时既有状态——并发窗口下可能读到中间态） */ + status: ImportJobStatus + /** true = 命中既有 job（幂等键复用，未重跑管道、未改行） */ + reused: boolean + siteId: string + siteKey: string + connectionId: string + wxrSnapshot: SnapshotRef + mediaBundleSnapshot: SnapshotRef | null + manualUrlListSnapshot: SnapshotRef | null + versions: ImportVersionBlock + reportPath: string | null + durationMs: number +} + +const VERSIONS: ImportVersionBlock = { + adapter: ADAPTER_VERSION, + mapping: MAPPING_VERSION, + sanitizer: SANITIZER_VERSION, + parser: PARSER_VERSION, +} + +// ---- 快照解析 ---- + +const SNAPSHOT_COLUMNS = { + id: sourceSnapshots.id, + siteId: sourceSnapshots.siteId, + kind: sourceSnapshots.kind, + sha256: sourceSnapshots.sha256, + byteSize: sourceSnapshots.byteSize, + storagePath: sourceSnapshots.storagePath, + sourceConnectionId: sourceSnapshots.sourceConnectionId, + capturedAt: sourceSnapshots.capturedAt, + metadata: sourceSnapshots.metadata, +} + +type SnapshotRow = { + id: string + siteId: string + kind: 'wxr' | 'sitemap' | 'media_bundle' | 'manual_url_list' + sha256: string + byteSize: number + storagePath: string + sourceConnectionId: string + capturedAt: Date + metadata: Record +} + +function toRef(row: SnapshotRow): SnapshotRef { + return { + id: row.id, + // 声音窄化：调用方按 kind 过滤查询（wxr/media_bundle/manual_url_list 三值）或已过 + // kind==='wxr' 校验——drizzle 列类型不携带过滤谓词，四值联合在此收窄是安全的 + kind: row.kind as SnapshotRef['kind'], + sha256: row.sha256, + byteSize: row.byteSize, + storagePath: row.storagePath, + capturedAt: row.capturedAt, + metadata: row.metadata, + } +} + +/** 按 (connection, kind) 解析最新快照（capturedAt 降序、createdAt 降序兜底平局）；无则 null */ +async function latestSnapshot( + db: ImportDb, + connectionId: string, + kind: 'wxr' | 'media_bundle' | 'manual_url_list', +): Promise { + const rows = await db + .select(SNAPSHOT_COLUMNS) + .from(sourceSnapshots) + .where(and(eq(sourceSnapshots.sourceConnectionId, connectionId), eq(sourceSnapshots.kind, kind))) + .orderBy(desc(sourceSnapshots.capturedAt), desc(sourceSnapshots.createdAt)) + .limit(1) + return rows.length > 0 ? (rows[0] as SnapshotRow) : null +} + +// ---- 幂等键 ---- + +/** 默认幂等键：site + wxr 快照 + 三语义版本（定义钉死——报告可复算、CLI 可显式覆盖） */ +export function defaultIdempotencyKey(siteKey: string, wxrSnapshotId: string): string { + return `import:${siteKey}:${wxrSnapshotId}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}` +} + +// ---- 主入口 ---- + +/** + * 创建（并就地推进到底）一个导入 job： + * 解析（site → wxr 快照 → 侧快照）→ 幂等建行（queued）→ running（attempt=1、started_at） + * → 空管道/注入管道 → succeeded｜failed（error_code/error_summary/finished_at）→ 报告落盘。 + * 失败口径：解析阶段失败零 job 行；管道阶段失败 job 行保留（failed 终态 + error 列）。 + */ +export async function createImportJob( + db: ImportDb, + input: CreateImportJobInput, +): Promise { + const startedAt = performance.now() + + // 阶段 0：站点解析（不自动建站——建站是 register 的职责） + const siteRows = await db + .select({ id: sites.id, key: sites.key, defaultLocale: sites.defaultLocale }) + .from(sites) + .where(eq(sites.key, input.siteKey)) + .limit(1) + if (siteRows.length === 0) { + throw new ImportError('SITE_NOT_FOUND', `站点不存在：${input.siteKey}（先运行 sf source:register）`) + } + const site = siteRows[0]! + + // 阶段 1：wxr 主输入快照解析（显式 id 优先；缺省按 (connection, kind='wxr') 最新） + let wxrRow: SnapshotRow | null = null + if (input.snapshotId !== undefined) { + const rows = await db + .select(SNAPSHOT_COLUMNS) + .from(sourceSnapshots) + .where(eq(sourceSnapshots.id, input.snapshotId)) + .limit(1) + if (rows.length === 0) { + throw new ImportError('SNAPSHOT_NOT_FOUND', `快照不存在：${input.snapshotId}`) + } + const row = rows[0] as SnapshotRow + if (row.siteId !== site.id) { + throw new ImportError( + 'SNAPSHOT_NOT_FOUND', + `快照 ${input.snapshotId} 不属于站点 ${input.siteKey}`, + ) + } + if (row.kind !== 'wxr') { + throw new ImportError( + 'NO_WXR_SNAPSHOT', + `快照 ${input.snapshotId} 的 kind 是 ${row.kind}，不是 wxr——主输入必须是 wxr 快照`, + ) + } + wxrRow = row + } else { + const connectionName = input.connectionName ?? 'default' + // connection 判重键 (site_id, name) 无库级唯一约束（Plan 1 决定，register 是唯一写入方） + const connRows = await db + .select({ id: sourceConnections.id }) + .from(sourceConnections) + .where(and(eq(sourceConnections.siteId, site.id), eq(sourceConnections.name, connectionName))) + .limit(1) + if (connRows.length > 0) { + wxrRow = await latestSnapshot(db, connRows[0]!.id, 'wxr') + } + if (wxrRow === null) { + throw new ImportError( + 'NO_WXR_SNAPSHOT', + `站点 ${input.siteKey}（connection=${connectionName}）没有任何 wxr 快照——先运行 sf source:register --wxr`, + ) + } + } + const connectionId = wxrRow.sourceConnectionId + + // 阶段 2：侧快照（前置事实 13）——挂在 wxr 快照的 connection 下按 kind 解析最新 + const mediaBundleRow = await latestSnapshot(db, connectionId, 'media_bundle') + const manualUrlListRow = await latestSnapshot(db, connectionId, 'manual_url_list') + const wxrSnapshot = toRef(wxrRow) + const mediaBundleSnapshot = mediaBundleRow === null ? null : toRef(mediaBundleRow) + const manualUrlListSnapshot = manualUrlListRow === null ? null : toRef(manualUrlListRow) + + // 阶段 3：幂等建 job（queued）——命中既有 key 直接复用返回（不重跑） + const idempotencyKey = input.idempotencyKey ?? defaultIdempotencyKey(input.siteKey, wxrSnapshot.id) + const resolved = { + siteKey: input.siteKey, + connectionId, + wxrSnapshot, + mediaBundleSnapshot, + manualUrlListSnapshot, + } + const existing = await reuseExisting(db, idempotencyKey) + if (existing !== null) { + return { ...existing, ...resolved, durationMs: performance.now() - startedAt } + } + let jobId: string + try { + const inserted = await db + .insert(importJobs) + .values({ + siteId: site.id, + sourceSnapshotId: wxrSnapshot.id, + status: 'queued', + mappingVersion: MAPPING_VERSION, + idempotencyKey, + }) + .returning({ id: importJobs.id }) + jobId = inserted[0]!.id + } catch (err) { + // 并发同 key：唯一约束收敛——重查复用（插入撞 23505 说明行已存在，重查必命中） + if (isUniqueViolation(err)) { + const raced = await reuseExisting(db, idempotencyKey) + if (raced !== null) { + return { ...raced, ...resolved, durationMs: performance.now() - startedAt } + } + } + throw err + } + + // 阶段 4：状态推进 queued → running（attempt 记首次执行） + await db + .update(importJobs) + .set({ status: 'running', startedAt: new Date(), attempt: 1 }) + .where(eq(importJobs.id, jobId)) + + // 阶段 5：管道（本任务缺省空管道；Task 6 接真实写入器）——抛错 → failed 终态 + let failed: { code: string; summary: string } | null = null + const ctx: ImportJobContext = { + db, + jobId, + siteId: site.id, + siteKey: site.key, + defaultLocale: site.defaultLocale, + connectionId, + wxrSnapshot, + mediaBundleSnapshot, + manualUrlListSnapshot, + versions: VERSIONS, + } + try { + await (input.pipeline ?? emptyPipeline)(ctx) + } catch (err) { + failed = { + code: err instanceof ImportError ? err.code : 'PIPELINE_FAILED', + summary: err instanceof Error ? `${err.name}: ${err.message}` : String(err), + } + } + + // 阶段 6：终态落行（report_path 同步入行）→ 报告写盘 + const reportRoot = resolve(input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'import')) + const reportDir = join(reportRoot, jobId, 'reports') + const reportPath = join(reportDir, 'import.json') + const finishedAt = new Date() + const finalStatus: ImportJobStatus = failed === null ? 'succeeded' : 'failed' + await db + .update(importJobs) + .set( + failed === null + ? { status: finalStatus, finishedAt, reportPath } + : { + status: finalStatus, + finishedAt, + reportPath, + errorCode: failed.code, + errorSummary: failed.summary, + }, + ) + .where(eq(importJobs.id, jobId)) + const report = { + jobId, + generatedAt: finishedAt.toISOString(), + status: finalStatus, + siteKey: site.key, + siteId: site.id, + connectionId, + snapshots: { + wxr: slimRef(wxrSnapshot), + media_bundle: mediaBundleSnapshot === null ? null : slimRef(mediaBundleSnapshot), + manual_url_list: manualUrlListSnapshot === null ? null : slimRef(manualUrlListSnapshot), + }, + versions: VERSIONS, + error: failed, + durationMs: performance.now() - startedAt, + } + await mkdir(reportDir, { recursive: true }) + await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8') + + return { + jobId, + status: finalStatus, + reused: false, + siteId: site.id, + siteKey: site.key, + connectionId, + wxrSnapshot, + mediaBundleSnapshot, + manualUrlListSnapshot, + versions: VERSIONS, + reportPath, + durationMs: performance.now() - startedAt, + } +} + +/** 既有 job 的复用返回（不动行、不跑管道）；无则 null */ +async function reuseExisting( + db: ImportDb, + idempotencyKey: string, +): Promise | null> { + const found = await db + .select({ + id: importJobs.id, + status: importJobs.status, + siteId: importJobs.siteId, + reportPath: importJobs.reportPath, + }) + .from(importJobs) + .where(eq(importJobs.idempotencyKey, idempotencyKey)) + .limit(1) + if (found.length === 0) return null + const row = found[0]! + return { + jobId: row.id, + status: row.status, + reused: true, + siteId: row.siteId, + reportPath: row.reportPath, + versions: VERSIONS, + } +} + +/** 报告里的快照引用瘦身版：不含 metadata（media_bundle 清单可达数千条，报告只留身份锚点） */ +function slimRef(ref: SnapshotRef): { + id: string + kind: string + sha256: string + byteSize: number + storagePath: string +} { + 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 {} diff --git a/apps/api/src/modules/source/register.service.ts b/apps/api/src/modules/source/register.service.ts index cf02d93..244f7bd 100644 --- a/apps/api/src/modules/source/register.service.ts +++ b/apps/api/src/modules/source/register.service.ts @@ -33,6 +33,7 @@ import { pipeline } from 'node:stream/promises' import { and, eq, sql } from 'drizzle-orm' import type { NodePgDatabase } from 'drizzle-orm/node-postgres' import { ADAPTER_VERSION } from '@sf/adapter-wp' +import { canonicalJson } from '../../common/canonical.js' import { sites, sourceConnections, sourceSnapshots } from '../../db/schema.js' import type * as schema from '../../db/schema.js' @@ -149,20 +150,9 @@ export interface MediaManifestEntry { } // ---- 哈希与副本 ---- - -/** canonical JSON（对象键递归排序、无空白）——media_bundle 清单哈希的序列化基形 */ -function canonicalJson(value: unknown): string { - if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]` - if (value !== null && typeof value === 'object') { - const record = value as Record - const body = Object.keys(record) - .sort() - .map((k) => `${JSON.stringify(k)}:${canonicalJson(record[k])}`) - .join(',') - return `{${body}}` - } - return JSON.stringify(value) -} +// canonicalJson 自 Plan 3 Task 1 起提取至 apps/api/src/common/canonical.ts 共享导出 +// （前置事实 6：normalized checksum 与清单哈希共用同一序列化基形）——本文件改 import， +// 定义零变更（tests/unit/canonical.test.ts + 既有清单哈希测试双守护）。 /** 流式单遍 copy+sha256 双算：读 64KB 块 → 哈希更新 → 写目标（pipeline 管背压与错误传播） */ async function copyAndHash(src: string, dest: string): Promise<{ sha256: string; bytes: number }> { diff --git a/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md b/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md index 7d50a46..bbce9f5 100644 --- a/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md +++ b/docs/superpowers/plans/2026-10-21-slice-import-pipeline-phase3.md @@ -53,15 +53,15 @@ tests/integration/import-*.test.ts - [ ] 证据文档：行数、与冒烟 855 attachment 的覆盖率预估（URL 前缀归一后 join 对照） - [ ] commit `feat(source): 媒体映射侧车导出与登记` -### Task 1: import 命令骨架 + import_jobs 生命周期 +### Task 1: import 命令骨架 + import_jobs 生命周期 ✅ 完成 **Files:** `apps/api/src/modules/import/{import.service.ts,import.module.ts}`；sf.ts + `import`；Test `tests/integration/import-job.test.ts` （本任务含前置事实 12 的 adapter 事件加法扩展：unsupported/error 事件补 `postId?: string`） -- [ ] 失败测试：`sf import --site --snapshot `（或按 connection+kind 选最新快照）→ import_jobs 行 created→running→succeeded（或 succeeded_with_issues/failed）；idempotency_key 唯一（重复提交同 key 复用既有 job 不重跑）；job 记录 mapping/adapter/sanitizer 版本块（来自 versions.ts） -- [ ] 实现：job 状态机 + 报告落 artifacts/import//；空流（无 snapshot）→ 明确报错 -- [ ] 绿 → commit `feat(import): job 骨架与生命周期` +- [x] 失败测试：`sf import --site --snapshot `（或按 connection+kind 选最新快照）→ import_jobs 行 created→running→succeeded（或 succeeded_with_issues/failed）；idempotency_key 唯一（重复提交同 key 复用既有 job 不重跑）；job 记录 mapping/adapter/sanitizer 版本块（来自 versions.ts） +- [x] 实现：job 状态机 + 报告落 artifacts/import//；空流（无 snapshot）→ 明确报错 +- [x] 绿 → commit `feat(import): job 骨架与生命周期` ### Task 2: term 物化（含自建缺失） diff --git a/packages/adapter-wordpress/src/adapter.ts b/packages/adapter-wordpress/src/adapter.ts index f2b7ba9..fe674ae 100644 --- a/packages/adapter-wordpress/src/adapter.ts +++ b/packages/adapter-wordpress/src/adapter.ts @@ -37,11 +37,16 @@ export interface RowResult { * 全链事件判别联合。unsupported/error 分支与 mapItem 的 MappedResult 同形，仅 error * 分支补 `sourceType`（`wp:${postType}` 报告锚点：post_id 缺失时 mapItem 拒猜身份， * 但 post_type 本身仍可作计数归桶依据——命名约定同 unsupported 分支，仅为报告可读性）。 + * + * Plan 3 Task 1 加法扩展（前置事实 12）：unsupported/error 事件可选携带 `postId` + * （WxrItem.postId 的 String 化；缺失时**键缺席**——不硬造 source_id）。用途：阶段 3 + * 导入把 unsupported/error 行也写进 source_records（source_id=post_id）的可审计性。 + * 纯加法可选字段：row 事件（UniversalRow 契约）与报告聚合均不受影响，不触发版本 bump。 */ export type AdapterEvent = | RowResult - | { kind: 'unsupported'; sourceType: string; reason: string; rawLocation: string } - | { kind: 'error'; sourceType: string; issue: string; rawLocation: string } + | { kind: 'unsupported'; sourceType: string; reason: string; rawLocation: string; postId?: string } + | { kind: 'error'; sourceType: string; issue: string; rawLocation: string; postId?: string } | { kind: 'term'; term: WxrTerm } /** 分段计时累计（可注入对象，运行中被本模块累加；消费完成后读取终值） */ @@ -103,9 +108,15 @@ function assembleItem( timings: AdapterTimings, ): AdapterEvent { const mapped = mapItem(item, mapConfig) - if (mapped.kind === 'unsupported') return mapped + // postId 附加规则（前置事实 12）：post_id 存在 → String 化挂上（与 row.sourceId 同口径， + // 负值原样）；缺失 → 键缺席。error 分支今天只在 post_id 缺失时触发，条件式保持 + // “存在即挂”的统一不变量，未来 error 触发缘由扩展时无需回改此处。 + if (mapped.kind === 'unsupported') { + return item.postId === null ? mapped : { ...mapped, postId: String(item.postId) } + } if (mapped.kind === 'error') { - return { ...mapped, sourceType: `wp:${item.postType}` } + const base = { ...mapped, sourceType: `wp:${item.postType}` } + return item.postId === null ? base : { ...base, postId: String(item.postId) } } const { row } = mapped diff --git a/packages/adapter-wordpress/test/adapter.test.ts b/packages/adapter-wordpress/test/adapter.test.ts index 7c97a0b..58dfb3d 100644 --- a/packages/adapter-wordpress/test/adapter.test.ts +++ b/packages/adapter-wordpress/test/adapter.test.ts @@ -385,6 +385,42 @@ describe('unsupported / error 汇聚', () => { }) }) +describe('unsupported / error 事件加法式补 postId（Plan 3 Task 1 前置事实 12）', () => { + it('unsupported（nav_menu_item）：post_id 存在 → 事件携带 postId（String 化，含负值）', async () => { + const xml = wxrDoc( + itemXml({ postId: 401, postType: 'nav_menu_item' }) + itemXml({ postId: -7, postType: 'custom_block' }), + ) + const events = await collect(runWxrAdapter(fromXml(xml), { mapConfig: { locale: 'zh-CN' } })) + expect(events[0]).toMatchObject({ + kind: 'unsupported', + sourceType: 'wp:nav_menu_item', + postId: '401', + }) + // 负值 postId 与 row.sourceId 同口径（String 原样，'-7' 不归一为 '7'） + expect(events[1]).toMatchObject({ kind: 'unsupported', sourceType: 'wp:custom_block', postId: '-7' }) + }) + + it('error（post_id 缺失）：postId 键缺席——不硬造 source_id（报告照常按计数归桶）', async () => { + const xml = wxrDoc(itemXml({ postId: null, postType: 'post' })) + const events = await collect(runWxrAdapter(fromXml(xml), { mapConfig: { locale: 'zh-CN' } })) + const ev = events[0] + if (ev === undefined || ev.kind !== 'error') throw new Error('期望 error 事件') + expect('postId' in ev).toBe(false) + expect((ev as { postId?: string }).postId).toBeUndefined() + // 报告聚合不受加法字段影响：failed 计数照常 + const report = await runWxrAdapterReport(fromXml(xml), { mapConfig: { locale: 'zh-CN' } }) + expect(report.totals).toStrictEqual({ rows: 0, terms: 0, unsupported: 0, failed: 1 }) + }) + + it('row 事件不受影响：RowResult 五键结构不变（postId 只加在 unsupported/error 变体上）', async () => { + const events = await collect( + runWxrAdapter(fromXml(wxrDoc(itemXml({ postId: 402 }))), { mapConfig: { locale: 'zh-CN' } }), + ) + const ev = soleRow(events) + expect(Object.keys(ev).sort()).toStrictEqual(['changed', 'issues', 'kind', 'row', 'sanitizeDiffs']) + }) +}) + describe('wppop 解析失败路径全链贯通（Task 4 已实现，Task 6 验证贯通）', () => { const RAW = 'not-php-at-all' diff --git a/scripts/sf.ts b/scripts/sf.ts index 1e6946f..77df3ae 100644 --- a/scripts/sf.ts +++ b/scripts/sf.ts @@ -1,11 +1,12 @@ #!/usr/bin/env node -// sf CLI（Plan 2 Task 7）：commander 两命令—— +// sf CLI（Plan 2 Task 7 + Plan 3 Task 1）：commander 三命令—— // - wxr:probe：只读诊断（任意可读路径；Task 8 直读 cms 用此通道），出 JSON/HTML 报告不入库； -// - source:register：调 registerSource 登记 content-addressed 快照（路径白名单校验在服务内）。 +// - source:register：调 registerSource 登记 content-addressed 快照（路径白名单校验在服务内）； +// - import：调 createImportJob 建 import_jobs 并推进生命周期（Task 1 空管道骨架；Task 6 接真实管道）。 // 执行机制（Plan 2 评审钉死）：根 devDep tsx 直吃 TS 源码（含 apps/api 的 Nest 源码与 // packages 的装饰器/TS 源文件），@sf/* 经根/包 tsconfig paths 直连 src——无需预构建 dist。 -// 注意：CLI 只 import register.service.ts（零装饰器纯函数文件），不拉起 Nest 模块—— -// 反射元数据依赖由此避免；source.module.ts 留给 standalone context。 +// 注意：CLI 只 import register.service.ts / import.service.ts（零装饰器纯函数文件）， +// 不拉起 Nest 模块——反射元数据依赖由此避免；module 文件留给 standalone context。 import { createReadStream } from 'node:fs' import { mkdir, writeFile } from 'node:fs/promises' import { dirname, join, resolve } from 'node:path' @@ -13,6 +14,11 @@ import { fileURLToPath } from 'node:url' import { Command } from 'commander' import { renderReportHtml, runWxrAdapterReport, type WxrAdapterReport } from '@sf/adapter-wp' import { getDb } from '../apps/api/src/db/client.js' +import { + ImportError, + createImportJob, + type CreateImportJobInput, +} from '../apps/api/src/modules/import/import.service.js' import { makeRunId, registerSource, @@ -39,7 +45,7 @@ function printProbeSummary(report: WxrAdapterReport): void { const program = new Command() program .name('sf') - .description('SiteForge CLI（阶段 2：来源输入登记与 WXR 探测）') + .description('SiteForge CLI（阶段 3：来源输入登记、WXR 探测与导入 job 生命周期）') .version('0.1.0') program @@ -155,14 +161,53 @@ program } }) +program + .command('import') + .description('导入已登记的 wxr 快照（Task 1 空管道骨架：job 生命周期 + 报告；真实管道 Task 6 接入）') + .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 }) => { + // getDb 走 DATABASE_URL_TEST ?? DATABASE_URL 回退链——临时库冒烟只需 env 覆盖 + const db = getDb() + // exactOptionalPropertyTypes：缺席键不传 undefined 值（与 source:register 同款） + const input: CreateImportJobInput = { siteKey: opts.site, connectionName: opts.connection } + if (opts.snapshot !== undefined) input.snapshotId = opts.snapshot + if (opts.key !== undefined) input.idempotencyKey = opts.key + try { + const result = await createImportJob(db, input) + console.log( + `[import] job=${result.jobId} status=${result.status}${result.reused ? '（复用既有 job，未重跑）' : ''}`, + ) + console.log(` site=${result.siteKey} connection=${result.connectionId}`) + console.log( + ` wxr 快照: ${result.wxrSnapshot.sha256.slice(0, 12)}… (${result.wxrSnapshot.id})`, + ) + console.log( + ` media_bundle: ${result.mediaBundleSnapshot === null ? '（无）' : result.mediaBundleSnapshot.id} / manual_url_list: ${result.manualUrlListSnapshot === null ? '（无）' : result.manualUrlListSnapshot.id}`, + ) + console.log( + ` 版本: adapter=${result.versions.adapter} mapping=${result.versions.mapping} sanitizer=${result.versions.sanitizer} parser=${result.versions.parser}`, + ) + if (result.reportPath !== null) console.log(` report: ${result.reportPath}`) + console.log(` total ${result.durationMs.toFixed(1)}ms`) + if (result.status === 'failed') process.exitCode = 1 + } finally { + await (db as { $client: { end: () => Promise } }).$client.end() + } + }) + // parseAsync：action 内抛错走此 catch（commander 默认会 process.exit(1) 并打印堆栈， -// 这里收敛为单行错误信息 + exitCode——SourceRegisterError 打 code，其余原样呈现） +// 这里收敛为单行错误信息 + exitCode——SourceRegisterError/ImportError 打 code，其余原样呈现） program.parseAsync(process.argv).catch((err: unknown) => { if (err instanceof SourceRegisterError) { console.error(`[sf] ${err.code}: ${err.message}`) if (err.code === 'PATH_NOT_ALLOWED') { console.error(' 提示：可用 --allow 放行额外目录，或将文件移入上述目录') } + } else if (err instanceof ImportError) { + console.error(`[sf] ${err.code}: ${err.message}`) } else { console.error(`[sf] ${err instanceof Error ? err.message : String(err)}`) } diff --git a/tests/integration/import-job.test.ts b/tests/integration/import-job.test.ts new file mode 100644 index 0000000..0643898 --- /dev/null +++ b/tests/integration/import-job.test.ts @@ -0,0 +1,411 @@ +// Plan 3 Task 1 失败测试（先于实现）：import 命令骨架 + import_jobs 生命周期。 +// 覆盖契约（对照计划 Task 1 与前置事实 6/12/13）： +// 1. register（wxr + media + media-map 三快照）→ createImportJob：job 行 created(queued)→ +// running→succeeded 全程可观察（管道钩子内读 running 中间态）、mapping_version 落 +// MAPPING_VERSION、报告记录全四版本块 + 三快照绑定（wxr 主输入 / media_bundle / +// manual_url_list 按 (connection,kind) 解析最新记入报告，job 行不扩列——前置事实 13）； +// 2. 无 --snapshot：按 (connection, kind='wxr') 选最新快照； +// 3. idempotency_key 唯一：重复提交同默认 key（site+snapshot+版本）→ 复用既有 job 不重跑 +// （reused 标记 + 管道零调用 + 恰一行）；显式 key 覆盖默认；并发同 key 收敛同 job； +// 4. 无 wxr 快照（站点仅有非 wxr 输入 / 显式 id 指向非 wxr 快照）→ NO_WXR_SNAPSHOT； +// 显式 id 不存在或属他站 → SNAPSHOT_NOT_FOUND；站点不存在 → SITE_NOT_FOUND；均零 job 行； +// 5. failed 路径：管道抛错 → running→failed（error_code/error_summary/finished_at）， +// 重提交同 key 复用失败 job 不重跑。 +// 运行于 globalSetup 临时库（ADR-015）；storage/报告根用 os.tmpdir 临时目录，finally 清理。 +import { existsSync } from 'node:fs' +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 { ADAPTER_VERSION, MAPPING_VERSION, PARSER_VERSION, SANITIZER_VERSION } from '@sf/adapter-wp' +import { + ImportError, + createImportJob, + type CreateImportJobInput, +} 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') + +interface Ctx { + work: string + storageRoot: string + registerReportRoot: string + importReportRoot: string + mediaDir: string + mediaMapFile: string + allowedRoots: string[] +} + +/** 独立媒体树 + 侧车 JSON（形态同 source-register.test.ts） */ +async function makeCtx(): Promise { + const work = await mkdtemp(join(tmpdir(), 'sf-import-')) + const mediaDir = join(work, 'media') + await mkdir(join(mediaDir, 'sub'), { recursive: true }) + await writeFile(join(mediaDir, 'a.png'), 'sf-import-media-a') + await writeFile(join(mediaDir, 'sub', 'b.png'), 'sf-import-media-b') + const mediaMapFile = join(work, 'media-map.json') + await writeFile( + mediaMapFile, + `${JSON.stringify([ + { sourceUrl: 'https://fixture.example/wp-content/uploads/2026/10/fixture-a.png', filePath: 'a.png' }, + ])}\n`, + 'utf8', + ) + return { + work, + storageRoot: join(work, 'storage'), + registerReportRoot: join(work, 'artifacts', 'register'), + importReportRoot: join(work, 'artifacts', 'import'), + mediaDir, + mediaMapFile, + allowedRoots: [FIXTURES_ROOT, work], + } +} + +/** 服务写入的 job 行清场：import_jobs 可变直删；快照不可变走触发器开关；FK 反序 */ +async function cleanupSiteRows(db: Db, siteId: string): Promise { + 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}'`) +} + +/** 登记三快照（wxr + media_bundle + manual_url_list）的便捷封装 */ +async function registerAll( + db: Db, + ctx: Ctx, + siteKey: string, +): Promise>> { + return registerSource(db, { + siteKey, + wxrPath: WXR_FIXTURE, + mediaDir: ctx.mediaDir, + mediaMapPath: ctx.mediaMapFile, + allowedRoots: ctx.allowedRoots, + storageRoot: ctx.storageRoot, + reportRoot: ctx.registerReportRoot, + }) +} + +describe('import_jobs 生命周期（Task 1 骨架）', () => { + it('三快照 register → createImportJob：queued→running→succeeded、mapping_version、报告含版本块与三快照绑定', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const reg = await registerAll(db, ctx, key) + try { + const wxr = reg.snapshots.find((s) => s.kind === 'wxr')! + const media = reg.snapshots.find((s) => s.kind === 'media_bundle')! + const map = reg.snapshots.find((s) => s.kind === 'manual_url_list')! + + // 管道钩子内观察中间态：此刻 job 已 created(queued)→running（attempt 记首次执行） + let midStatus: string | undefined + let midAttempt: string | number | undefined + let midStartedAt: Date | null | undefined + let midFinishedAt: Date | null | undefined + const result = await createImportJob(db, { + siteKey: key, + snapshotId: wxr.id, + reportRoot: ctx.importReportRoot, + pipeline: async (jobCtx) => { + expect(jobCtx.jobId).toBeTruthy() + const rows = (await db.execute( + `select status, attempt, started_at, finished_at from import_jobs where id = '${jobCtx.jobId}'`, + )) as any + midStatus = rows[0].status + midAttempt = rows[0].attempt + midStartedAt = rows[0].started_at + midFinishedAt = rows[0].finished_at + }, + }) + + expect(result.reused).toBe(false) + expect(result.status).toBe('succeeded') + expect(midStatus).toBe('running') + expect(Number(midAttempt)).toBe(1) + expect(midStartedAt).not.toBeNull() + expect(midFinishedAt).toBeNull() + + // 终态行：身份/版本/时间戳/report_path + const job = (await db.execute( + `select status, mapping_version, source_snapshot_id, started_at, finished_at, report_path, attempt, error_code + from import_jobs where id = '${result.jobId}'`, + )) as any + expect(job[0].status).toBe('succeeded') + expect(job[0].mapping_version).toBe(MAPPING_VERSION) + expect(job[0].source_snapshot_id).toBe(wxr.id) + expect(job[0].started_at).not.toBeNull() + expect(job[0].finished_at).not.toBeNull() + expect(job[0].error_code).toBeNull() + expect(Number(job[0].attempt)).toBe(1) + + // 报告：全四版本块 + 三快照绑定（前置事实 13：侧快照只入报告，job 行不扩列） + const reportPath = join(ctx.importReportRoot, result.jobId, 'reports', 'import.json') + expect(job[0].report_path).toBe(reportPath) + expect(existsSync(reportPath)).toBe(true) + const report = JSON.parse(await readFile(reportPath, 'utf8')) as { + jobId: string + status: string + versions: Record + snapshots: { + wxr: { id: string } + media_bundle: { id: string } | null + manual_url_list: { id: string } | null + } + } + expect(report.jobId).toBe(result.jobId) + expect(report.status).toBe('succeeded') + expect(report.versions).toStrictEqual({ + adapter: ADAPTER_VERSION, + mapping: MAPPING_VERSION, + sanitizer: SANITIZER_VERSION, + parser: PARSER_VERSION, + }) + expect(report.snapshots.wxr.id).toBe(wxr.id) + expect(report.snapshots.media_bundle?.id).toBe(media.id) + expect(report.snapshots.manual_url_list?.id).toBe(map.id) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) + + it('无 snapshotId：按 (connection, kind=wxr) 选最新；同默认 key 重提交复用不重跑', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const r1 = await registerSource(db, { + siteKey: key, + wxrPath: WXR_FIXTURE, + allowedRoots: ctx.allowedRoots, + storageRoot: ctx.storageRoot, + reportRoot: ctx.registerReportRoot, + }) + try { + // 第二个 wxr 快照：内容加尾随换行（sha 不同、仍合法 XML）；稍歇保证 capturedAt 严格更晚 + const v2 = join(ctx.work, 'mini-v2.xml') + await writeFile(v2, `${await readFile(WXR_FIXTURE, 'utf8')}\n`, 'utf8') + await new Promise((r) => setTimeout(r, 25)) + const r2 = await registerSource(db, { + siteKey: key, + wxrPath: v2, + allowedRoots: ctx.allowedRoots, + storageRoot: ctx.storageRoot, + reportRoot: ctx.registerReportRoot, + }) + const wxr1 = r1.snapshots.find((s) => s.kind === 'wxr')! + const wxr2 = r2.snapshots.find((s) => s.kind === 'wxr')! + expect(wxr2.id).not.toBe(wxr1.id) + + let pipelineCalls = 0 + const input: CreateImportJobInput = { + siteKey: key, + reportRoot: ctx.importReportRoot, + pipeline: async () => { + pipelineCalls++ + }, + } + const a = await createImportJob(db, input) + expect(a.reused).toBe(false) + expect(a.status).toBe('succeeded') + // 最新快照 = 第二个（v2） + const job = (await db.execute( + `select source_snapshot_id from import_jobs where id = '${a.jobId}'`, + )) as any + expect(job[0].source_snapshot_id).toBe(wxr2.id) + + // 同默认 key 重提交：复用既有 job、管道零追加调用 + const b = await createImportJob(db, input) + expect(b.reused).toBe(true) + expect(b.jobId).toBe(a.jobId) + expect(b.status).toBe('succeeded') + expect(pipelineCalls).toBe(1) + const count = (await db.execute( + `select count(*)::int as n from import_jobs where site_id = '${r1.siteId}'`, + )) as any + expect(count[0].n).toBe(1) + + // 显式 idempotencyKey 覆盖默认 key：同 key 复用、异 key（回默认）不新建重复 + const c = await createImportJob(db, { ...input, idempotencyKey: 'manual-key-1' }) + expect(c.reused).toBe(false) + const d = await createImportJob(db, { ...input, idempotencyKey: 'manual-key-1' }) + expect(d.reused).toBe(true) + expect(d.jobId).toBe(c.jobId) + expect(pipelineCalls).toBe(2) + } finally { + await cleanupSiteRows(db, r1.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) + + it('并发同默认 key：恰一 job 行、双方同 jobId、管道恰执行一次（唯一约束收敛）', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const reg = await registerAll(db, ctx, key) + try { + const wxr = reg.snapshots.find((s) => s.kind === 'wxr')! + let pipelineCalls = 0 + const input: CreateImportJobInput = { + siteKey: key, + snapshotId: wxr.id, + reportRoot: ctx.importReportRoot, + pipeline: async () => { + pipelineCalls++ + }, + } + const [a, b] = await Promise.all([createImportJob(db, input), createImportJob(db, input)]) + expect(a.jobId).toBe(b.jobId) + expect(Number(a.reused) + Number(b.reused)).toBe(1) + expect(pipelineCalls).toBe(1) + const count = (await db.execute( + `select count(*)::int as n from import_jobs where site_id = '${reg.siteId}'`, + )) as any + expect(count[0].n).toBe(1) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) + + it('可预期失败均零 job 行：SITE_NOT_FOUND / SNAPSHOT_NOT_FOUND（不存在或属他站）/ NO_WXR_SNAPSHOT（无非 wxr 主输入或显式指向非 wxr）', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + // 站点 A：仅 media-map 输入（有 connection、无非 wxr 主输入） + const keyA = `sfimp-${rand()}` + const regA = await registerSource(db, { + siteKey: keyA, + mediaMapPath: ctx.mediaMapFile, + allowedRoots: ctx.allowedRoots, + storageRoot: ctx.storageRoot, + reportRoot: ctx.registerReportRoot, + }) + // 站点 B：三快照齐全 + const keyB = `sfimp-${rand()}` + const regB = await registerAll(db, ctx, keyB) + try { + const mapSnapA = regA.snapshots[0]! + const mediaSnapB = regB.snapshots.find((s) => s.kind === 'media_bundle')! + const cases: { label: string; input: CreateImportJobInput; code: string }[] = [ + { + label: '站点不存在', + input: { siteKey: `sfimp-no-such-${rand()}` }, + code: 'SITE_NOT_FOUND', + }, + { + label: '仅有 manual_url_list 输入的站点（无任何 wxr 快照）', + input: { siteKey: keyA, reportRoot: ctx.importReportRoot }, + code: 'NO_WXR_SNAPSHOT', + }, + { + label: '显式 snapshotId 指向非 wxr 快照（media_bundle）', + input: { siteKey: keyB, snapshotId: mediaSnapB.id, reportRoot: ctx.importReportRoot }, + code: 'NO_WXR_SNAPSHOT', + }, + { + label: '显式 snapshotId 不存在', + input: { + siteKey: keyB, + snapshotId: '00000000-0000-4000-8000-000000000000', + reportRoot: ctx.importReportRoot, + }, + code: 'SNAPSHOT_NOT_FOUND', + }, + { + label: '显式 snapshotId 属于他站', + input: { siteKey: keyB, snapshotId: mapSnapA.id, reportRoot: ctx.importReportRoot }, + code: 'SNAPSHOT_NOT_FOUND', + }, + ] + for (const c of cases) { + const err: ImportError = await createImportJob(db, c.input).then( + () => { + throw new Error(`用例「${c.label}」不应成功`) + }, + (e: ImportError) => e, + ) + expect(err, c.label).toBeInstanceOf(ImportError) + expect(err.code, c.label).toBe(c.code) + expect(err.message, c.label).not.toBe('') + } + // 零 job 行（两站点均无） + const count = (await db.execute( + `select count(*)::int as n from import_jobs where site_id in ('${regA.siteId}', '${regB.siteId}')`, + )) as any + expect(count[0].n).toBe(0) + } finally { + await cleanupSiteRows(db, regB.siteId) + await cleanupSiteRows(db, regA.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) + + it('failed 路径：管道抛错 → running→failed（error_code/error_summary/finished_at）；重提交同 key 复用失败 job 不重跑', async () => { + await withTestDb(async (db) => { + const ctx = await makeCtx() + try { + const key = `sfimp-${rand()}` + const reg = await registerAll(db, ctx, key) + try { + const wxr = reg.snapshots.find((s) => s.kind === 'wxr')! + let pipelineCalls = 0 + const input: CreateImportJobInput = { + siteKey: key, + snapshotId: wxr.id, + reportRoot: ctx.importReportRoot, + pipeline: async () => { + pipelineCalls++ + throw new Error('模拟管道失败：断链 attachment') + }, + } + const a = await createImportJob(db, input) + expect(a.reused).toBe(false) + expect(a.status).toBe('failed') + const job = (await db.execute( + `select status, error_code, error_summary, finished_at, started_at from import_jobs where id = '${a.jobId}'`, + )) as any + expect(job[0].status).toBe('failed') + expect(job[0].error_code).toBe('PIPELINE_FAILED') + expect(String(job[0].error_summary)).toContain('模拟管道失败') + expect(job[0].started_at).not.toBeNull() + expect(job[0].finished_at).not.toBeNull() + + // 同 key 重提交：复用失败 job（不重跑、不改行） + const b = await createImportJob(db, input) + expect(b.reused).toBe(true) + expect(b.jobId).toBe(a.jobId) + expect(b.status).toBe('failed') + expect(pipelineCalls).toBe(1) + const count = (await db.execute( + `select count(*)::int as n from import_jobs where site_id = '${reg.siteId}'`, + )) as any + expect(count[0].n).toBe(1) + } finally { + await cleanupSiteRows(db, reg.siteId) + } + } finally { + await rm(ctx.work, { recursive: true, force: true }) + } + }) + }) +}) diff --git a/tests/unit/canonical.test.ts b/tests/unit/canonical.test.ts new file mode 100644 index 0000000..63e2250 --- /dev/null +++ b/tests/unit/canonical.test.ts @@ -0,0 +1,46 @@ +// canonicalJson 共享导出测试（Plan 3 Task 1，前置事实 6）：从 register.service.ts 提取到 +// apps/api/src/common/canonical.ts——register 侧改 import 零行为变更（既有 media_bundle 清单 +// 哈希测试继续绿的回归责任在 tests/integration/source-register.test.ts，此处钉导出本身的 +// 序列化定义：对象键递归排序、无空白、数组保序）。 +// 独立重算对照：期望串手写钉死，不 import 实现——定义漂移即爆红。 +import { describe, expect, it } from 'vitest' +import { canonicalJson } from '../../apps/api/src/common/canonical.js' + +describe('canonicalJson（共享导出，前置事实 6）', () => { + it('对象键递归排序：任意嵌套层级都按键名字典序', () => { + expect(canonicalJson({ b: 1, a: 2 })).toBe('{"a":2,"b":1}') + expect(canonicalJson({ b: 1, a: { d: 2, c: 3 } })).toBe('{"a":{"c":3,"d":2},"b":1}') + expect(canonicalJson({ z: { y: { x: { w: 1, v: 2 } } } })).toBe('{"z":{"y":{"x":{"v":2,"w":1}}}}') + }) + + it('无空白：键与值、数组元素间均无空格/换行', () => { + expect(canonicalJson({ a: [1, 2, 3], b: { c: 'x' } })).toBe('{"a":[1,2,3],"b":{"c":"x"}}') + expect(canonicalJson({ a: 1, b: 2 })).not.toContain(' ') + expect(canonicalJson({ a: { b: { c: [1, { d: 2 }] } } })).not.toMatch(/\s/) + }) + + it('数组保序：元素顺序是语义的一部分，不排序', () => { + expect(canonicalJson([3, 1, 2])).toBe('[3,1,2]') + expect(canonicalJson([{ z: 1, a: 2 }, { y: 3, b: 4 }])).toBe('[{"a":2,"z":1},{"b":4,"y":3}]') + }) + + it('原始值走 JSON.stringify 语义：字符串/数字/布尔/null', () => { + expect(canonicalJson('x')).toBe('"x"') + expect(canonicalJson('a"b')).toBe('"a\\"b"') + expect(canonicalJson(3)).toBe('3') + expect(canonicalJson(true)).toBe('true') + expect(canonicalJson(null)).toBe('null') + expect(canonicalJson({ a: null, b: true, c: 0, d: '' })).toBe('{"a":null,"b":true,"c":0,"d":""}') + }) + + it('键排序按 UTF-16 码元（与 Array.prototype.sort 默认一致）——大写在前、非 ASCII 原样', () => { + expect(canonicalJson({ b: 1, A: 2, a: 3 })).toBe('{"A":2,"a":3,"b":1}') + expect(canonicalJson({ 图: 1, 中: 2 })).toBe(JSON.stringify({ 中: 2, 图: 1 })) + }) + + it('空对象/空数组序列化', () => { + expect(canonicalJson({})).toBe('{}') + expect(canonicalJson([])).toBe('[]') + expect(canonicalJson({ a: {}, b: [] })).toBe('{"a":{},"b":[]}') + }) +})