浅念
fa27a3a308
- validation_issues 落库改分批（VALIDATION_ISSUES_CHUNK=1000 行/批循环）： 单 INSERT 占位符上限（pg ~65535 ÷ 6 列）会被超大 issue 集撞穿； run 行先落、批次间无序依赖。新增测试：1101 坏链注入 → 全量 1101 行 落库、收口 VALIDATION_FAILED（非 GATE_INTERNAL） - import 域报告写序换 HTML 先（与 build 域对齐——json 在则 html 必在） - gates canonical 指错分支 evidence 改 kind:'canonical-mismatch' （mismatch 语义不再误用 missing 键；缺席分支保持 missing:'canonical'） - build.html 尾换行对齐 import 域 - 计划文档 Task 9 补审查遗留清单（helper 迁移/Plan 5 前置门禁面） Co-Authored-By: Claude Code <noreply@anthropic.com>
756 lines
34 KiB
TypeScript
756 lines
34 KiB
TypeScript
// import_jobs 生命周期服务（Plan 3 Task 1 + Task 6 编排，spec §7.2/§7.5、前置事实 6/8/13）：
|
||
// 解析站点与快照 → 幂等建 job → 状态推进（queued → running → succeeded /
|
||
// succeeded_with_issues / failed）→ 报告落 <reportRoot>/<jobId>/reports/import.json
|
||
// （temp+rename 原子写、**先于终态落行**——行内 report_path 出现即文件存在）。
|
||
// 管道产出 Outcome（counts/issues）汇入报告；issues 非空 → succeeded_with_issues。
|
||
// 复用（幂等键命中）只返回 job 行锚定字段——快照/版本解析不回读（权威在既有报告）。
|
||
//
|
||
// 本文件与 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 起默认 = **编排管道**（四家写入器汇流）——orchestratedPipeline 自
|
||
// Task 8 起抽取于 orchestrator.ts（纯函数 over ctx，零行为变更）。本文件负责：真事件流
|
||
// 构造（runWxrAdapter 读 wxr 快照副本，mapConfig.locale 接 site.default_locale——前置
|
||
// 事实 8）→ Outcome 汇合（mergeOutcomes）→ 报告固化。注入 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, 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 { mergeOutcomes, orchestratedPipeline } from './orchestrator.js'
|
||
import {
|
||
DEFAULT_NEWS_PATH,
|
||
DEFAULT_PRODUCTS_PATH,
|
||
isSafePath,
|
||
normalizeListPath,
|
||
} from './route-writer.js'
|
||
import { renderImportReportHtml } from './report-html.js'
|
||
|
||
/** 本文件位于 <repo>/apps/api/src/modules/import/ → 上五级是仓库根 */
|
||
const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..')
|
||
|
||
export type ImportDb = NodePgDatabase<typeof schema>
|
||
|
||
export type ImportJobStatus = (typeof schema.importJobStatus.enumValues)[number]
|
||
|
||
export type ImportErrorCode =
|
||
| 'SITE_NOT_FOUND'
|
||
| 'SNAPSHOT_NOT_FOUND'
|
||
| 'NO_WXR_SNAPSHOT'
|
||
| 'PIPELINE_FAILED'
|
||
| 'REPORT_WRITE_FAILED'
|
||
| 'INVALID_LIST_PATH'
|
||
|
||
/** 导入失败（可预期错误）：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<string, unknown>
|
||
}
|
||
|
||
/** 管道执行上下文：Task 6 真实管道的全部输入（站点 locale 接缝见前置事实 8） */
|
||
export interface ImportJobContext {
|
||
/**
|
||
* 写入器自行管理事务：ctx.db 为直连（非事务代理），需要原子性的多语句写入用
|
||
* db.transaction 自行包裹；单语句 upsert（on conflict do nothing/update）本身原子，
|
||
* 无需显式事务。
|
||
*/
|
||
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
|
||
/**
|
||
* 报告累加器（与 Outcome 同形）：写入器可就地累加（ctx.report.counts.k++ /
|
||
* ctx.report.issues.push(...)）。合并规则见 {@link ImportPipeline}——返回值优先。
|
||
*/
|
||
report: ImportPipelineOutcome
|
||
/**
|
||
* 适配器事件流（runWxrAdapter 产物：row/unsupported/error/term 按文档序交错）。
|
||
* Task 6 兑现：缺省 = 从 wxr 快照副本构造的真流（mapConfig.locale=site.default_locale
|
||
* ——前置事实 8）；测试经 {@link CreateImportJobInput.events} 手构注入。单消费方
|
||
* （编排器）一遍 for-await 分派给四家写入器——只承诺"可迭代一次"，无 tee 无重迭代。
|
||
*/
|
||
events: AsyncIterable<AdapterEvent>
|
||
}
|
||
|
||
/**
|
||
* 管道产出契约（Task 1 钉死、Task 6 修订——counts 聚合改**求和**、加法扩 issueGroups/
|
||
* sanitize）：
|
||
* - counts：计数块（键名由写入器约定，如 contentEntries/terms/routes）——原样进报告；
|
||
* - issues：问题清单（简单字符串——机器可读错误码按 Plan 2 台账推迟）；**非空 → job
|
||
* 终态 succeeded_with_issues**（问题全部入报告，不静默）；
|
||
* - issueGroups / sanitize：可选分组视图与清洗统计（编排管道产出；issues 仍是全量扁平清单）。
|
||
*
|
||
* 汇合规则（Task 6 钉死）：最终产出 = ctx.report 累加器 ⊕ 管道返回值——counts 按键
|
||
* **求和**、issues 串联（累加器在前、返回值在后）、issueGroups 逐键串联、sanitize
|
||
* 数值相加。管道抛错时无返回值，累加器现值进报告（部分产出可审计）——缺省编排管道
|
||
* 在重抛前自行冲刷部分产出（term/content 纯返回 finish + adapter 问题/清洗统计 +
|
||
* eventsDispatched 进度计数，见 orchestratedPipeline），注入管道的冲刷行为由注入方自理。
|
||
*/
|
||
export interface ImportPipelineOutcome {
|
||
/**
|
||
* 计数块（原样进报告）。**键契约：counts 键是跨写入器的扁平全局命名空间**——
|
||
* 汇合按"同名键**求和**"聚合（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 前缀；orchestrator：eventsDispatched——仅失败路径部分冲刷时
|
||
* 出现，成功路径不占键——新写入器勿复用他家键名）。
|
||
*/
|
||
counts: Record<string, number>
|
||
issues: string[]
|
||
/**
|
||
* 可选：按来源分组的 issues 视图（报告分组呈现）。**issues 字段仍是无分组全量
|
||
* 清单**（既有消费方契约不变）；分组键 = 四写入器前缀 + adapter（row 事件自带行级
|
||
* 问题）+ orchestrator（编排器自身）。
|
||
*/
|
||
issueGroups?: Partial<Record<ImportIssueGroup, string[]>>
|
||
/** 可选：清洗统计（row 事件 sanitizeDiffs/changed 的聚合——编排层在分派循环收集） */
|
||
sanitize?: ImportSanitizeStats
|
||
}
|
||
|
||
/** 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<string, number>
|
||
}
|
||
|
||
/**
|
||
* 写入器共享身份基类型（Task 6 义务③：编译期防漂移）——四家写入器的 identity 参数均以
|
||
* 此为基础（`WriterContext & { 各家专属列 }`），db/siteId/connectionId 三列的形状与命名
|
||
* 单源约束；ImportJobContext 结构性满足（Pick 自它），编排注入即传 ctx 的这三列。
|
||
*/
|
||
export type WriterContext = Pick<ImportJobContext, 'db' | 'siteId' | 'connectionId'>
|
||
|
||
/** 管道（缺省 = Task 6 编排管道 orchestratedPipeline——四家写入器汇流）。抛错 → job failed */
|
||
export type ImportPipeline = (ctx: ImportJobContext) => Promise<ImportPipelineOutcome>
|
||
|
||
export interface CreateImportJobInput {
|
||
siteKey: string
|
||
/** 显式 wxr 快照 id；缺省按 (connection, kind='wxr') 选最新 */
|
||
snapshotId?: string
|
||
/** 幂等键；缺省 = site + wxr 快照 + 三语义版本 自动派生 */
|
||
idempotencyKey?: string
|
||
/** 连接名（最新快照解析的 connection 侧）；缺省 'default'（与 register 缺省一致） */
|
||
connectionName?: string
|
||
/** 报告根（落 <reportRoot>/<jobId>/reports/import.json）；缺省 repoRoot/artifacts/import */
|
||
reportRoot?: string
|
||
/**
|
||
* 适配器事件流（ctx.events 的注入位）：测试手构注入。注入而缺省 pipeline 时仍走编排
|
||
* 管道（手构流 + 真编排）；注入 pipeline 而无 events 时为空流（Task 1 语义保持）；
|
||
* 两者皆缺席 = 真流（wxr 快照副本，前置事实 8 的 locale 接线在真流构造处）。
|
||
*/
|
||
events?: AsyncIterable<AdapterEvent>
|
||
/** 管道钩子（测试注入）；缺省 = 编排管道 */
|
||
pipeline?: ImportPipeline
|
||
/**
|
||
* 主列表页路径（前置事实 9：CLI 参数来源，规范化值记入 job 报告）；缺省 /products/。
|
||
* 解析期防线（阶段 2.5，建 job 行前）：不安全段（穿越/控制字符等）→ INVALID_LIST_PATH
|
||
*/
|
||
productsPath?: string
|
||
/** 新闻列表页路径（同上，缺省 /news/；防线同 productsPath） */
|
||
newsPath?: string
|
||
}
|
||
|
||
export interface CreateImportJobResult {
|
||
jobId: string
|
||
/** job 终态（或复用时既有状态——并发窗口下可能读到中间态） */
|
||
status: ImportJobStatus
|
||
/** true = 命中既有 job（幂等键复用，未重跑管道、未改行） */
|
||
reused: boolean
|
||
siteId: string
|
||
/** job 行的 mapping_version（新建=当前 MAPPING_VERSION；复用=行值——权威记录在行） */
|
||
mappingVersion: string
|
||
reportPath: string | null
|
||
durationMs: number
|
||
/**
|
||
* 以下字段**仅新建路径填充**（复用口径忠实化：reuse 不回读快照/版本解析，快照与
|
||
* 版本块的权威记录是既有 job 报告——reportPath 指向它；复用返回只认 job 行锚定字段）
|
||
*/
|
||
siteKey?: string
|
||
connectionId?: string
|
||
wxrSnapshot?: SnapshotRef
|
||
mediaBundleSnapshot?: SnapshotRef | null
|
||
manualUrlListSnapshot?: SnapshotRef | null
|
||
versions?: ImportVersionBlock
|
||
}
|
||
|
||
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<string, unknown>
|
||
}
|
||
|
||
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<SnapshotRow | null> {
|
||
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}`
|
||
}
|
||
|
||
/**
|
||
* 主列表路径解析（阶段 2.5 输入防线，Task 6 审查 Fix 3）：安全检查（route-writer 的
|
||
* isSafePath 同判据）→ normalizeListPath 规范化（补首尾斜杠）。不安全段 →
|
||
* INVALID_LIST_PATH（CLI 无需新码——既有 ImportError 通道按 code 呈现）。
|
||
*/
|
||
function resolveListPath(raw: string, cliFlag: string): string {
|
||
if (!isSafePath(raw)) {
|
||
throw new ImportError(
|
||
'INVALID_LIST_PATH',
|
||
`${cliFlag}=${raw} 含不安全段（NUL/反斜杠/控制字符/穿越段 '..'）——拒绝（完整校验规则 SP3 验证器）`,
|
||
)
|
||
}
|
||
try {
|
||
return normalizeListPath(raw)
|
||
} catch (err) {
|
||
// normalizeListPath 自带同判据拒绝（双覆盖的理论重叠分支）——归码为可预期错误
|
||
throw new ImportError('INVALID_LIST_PATH', `${cliFlag}=${raw}：${err instanceof Error ? err.message : String(err)}`)
|
||
}
|
||
}
|
||
|
||
// ---- 主入口 ----
|
||
|
||
/**
|
||
* 创建（并就地推进到底）一个导入 job：
|
||
* 解析（site → wxr 快照 → 侧快照）→ 幂等建行（queued）→ running（attempt=1、started_at）
|
||
* → 空管道/注入管道 → succeeded｜failed（error_code/error_summary/finished_at）→ 报告落盘。
|
||
* 失败口径：解析阶段失败零 job 行（含主列表路径防线 INVALID_LIST_PATH——建行前抛出，
|
||
* 不留悬挂 running 行）；管道阶段失败 job 行保留（failed 终态 + error 列）。
|
||
*/
|
||
export async function createImportJob(
|
||
db: ImportDb,
|
||
input: CreateImportJobInput,
|
||
): Promise<CreateImportJobResult> {
|
||
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]!
|
||
|
||
// 阶段 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) {
|
||
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)
|
||
|
||
// 阶段 2.5：主列表路径解析（Task 6 审查 Fix 3——**建 job 行前**的输入防线：穿越段等
|
||
// 不安全输入 → INVALID_LIST_PATH 零 job 行残留；建行/置 running 后再抛会留悬挂行）。
|
||
// normalizeListPath 自带同判据安全拒绝（route-writer 侧双覆盖），此处前置检查给出
|
||
// CLI 参数归属的错误信息。复用命中同样过检：非法输入恒拒（与 SITE_NOT_FOUND 同纪律）。
|
||
const productsPath = resolveListPath(input.productsPath ?? DEFAULT_PRODUCTS_PATH, '--products-path')
|
||
const newsPath = resolveListPath(input.newsPath ?? DEFAULT_NEWS_PATH, '--news-path')
|
||
|
||
// 阶段 3：幂等建 job（queued）——命中既有 key 直接复用返回（不重跑）。
|
||
// 复用口径忠实化：只返回 job 行锚定字段（含行内 mapping_version/report_path），
|
||
// 快照与版本块不回读（权威记录在既有 job 报告——reportPath 指向它）。
|
||
const idempotencyKey = input.idempotencyKey ?? defaultIdempotencyKey(input.siteKey, wxrSnapshot.id)
|
||
const existing = await reuseExisting(db, idempotencyKey)
|
||
if (existing !== null) {
|
||
return { ...existing, 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, durationMs: performance.now() - startedAt }
|
||
}
|
||
}
|
||
throw err
|
||
}
|
||
|
||
// 阶段 4：状态推进 queued → running（attempt 记首次执行）
|
||
const attempt = 1
|
||
await db
|
||
.update(importJobs)
|
||
.set({ status: 'running', startedAt: new Date(), attempt })
|
||
.where(eq(importJobs.id, jobId))
|
||
|
||
// 阶段 5：管道——缺省 = 编排管道（四家写入器汇流）；注入钩子优先。抛错 → failed 终态。
|
||
// 事件流三分：注入 events 优先（手构流 + 缺省编排管道 = 测试形态）→ 注入 pipeline 而
|
||
// 无 events = 空流（Task 1 语义保持）→ 双缺席 = 真流（wxr 快照副本 + locale 接线，
|
||
// 前置事实 8——缺省空串会让全部行被 content-writer 以 locale 不匹配拒收）。
|
||
// productsPath/newsPath 已在阶段 2.5 解析（建行前防线 + 报告/管道共用规范化值）。
|
||
const useRealStream = input.events === undefined && input.pipeline === undefined
|
||
const adapterTimings: AdapterTimings | null = useRealStream ? { parseMs: 0, sanitizeMs: 0 } : null
|
||
const events: AsyncIterable<AdapterEvent> =
|
||
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,
|
||
jobId,
|
||
siteId: site.id,
|
||
siteKey: site.key,
|
||
defaultLocale: site.defaultLocale,
|
||
connectionId,
|
||
wxrSnapshot,
|
||
mediaBundleSnapshot,
|
||
manualUrlListSnapshot,
|
||
versions: VERSIONS,
|
||
report: { counts: {}, issues: [] },
|
||
events,
|
||
}
|
||
let outcome: ImportPipelineOutcome | null = null
|
||
try {
|
||
outcome = await pipeline(ctx)
|
||
} catch (err) {
|
||
failed = {
|
||
code: err instanceof ImportError ? err.code : 'PIPELINE_FAILED',
|
||
// '[pipeline]' 前缀：管道阶段错误与解析阶段错误在 error_summary 上可区分
|
||
//（ImportError 保留原 code——管道内可预期失败复用同一套错误码面）
|
||
summary: `[pipeline] ${err instanceof Error ? `${err.name}: ${err.message}` : String(err)}`,
|
||
}
|
||
}
|
||
// 抛错时无返回值：累加器现值进报告（部分产出可审计）
|
||
const merged: ImportPipelineOutcome = outcome === null ? ctx.report : mergeOutcomes(ctx.report, outcome)
|
||
let finalStatus: ImportJobStatus
|
||
if (failed !== null) {
|
||
finalStatus = 'failed'
|
||
} else {
|
||
// 终态判定在**全量** issues 长度上做（截断是报告固化层语义，不吞终态事实）
|
||
finalStatus = merged.issues.length > 0 ? 'succeeded_with_issues' : 'succeeded'
|
||
}
|
||
|
||
// 报告 issues 封顶（Task 8 兑现 Task 6 推迟项，对齐 Plan 2 ROW_ERRORS_CAP 纪律）：
|
||
// flat 清单与 issueGroups 各组按同一常量截断——对抗性输入（单条 item 上万行级问题）
|
||
// 不使报告体积失去上界；counts/sanitize 统计不受影响始终完整。分组是同一集合的
|
||
// 视图，各组溢出 ≤ flat 溢出（分划性质），不单独入账。
|
||
const cappedIssues = merged.issues.slice(0, REPORT_ISSUES_CAP)
|
||
const issuesTruncated = merged.issues.length - cappedIssues.length
|
||
const cappedGroups = capIssueGroups(merged.issueGroups)
|
||
|
||
// 阶段 6：报告先行落盘（temp+rename 原子替换），终态落行在后——行内 report_path
|
||
// 出现即文件存在；写盘失败 → job failed（REPORT_WRITE_FAILED、report_path 不落，
|
||
// 不留“succeeded 但报告失踪”或悬挂路径）。import.json 为权威产物（机器消费），
|
||
// import.html 为同数据的渲染视图（Task 8 兑现 Task 6 推迟项）——同 try 同失败口径。
|
||
const reportRoot = resolve(input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'import'))
|
||
const reportDir = join(reportRoot, jobId, 'reports')
|
||
const reportPath = join(reportDir, 'import.json')
|
||
const htmlPath = join(reportDir, 'import.html')
|
||
const report: ImportReport = {
|
||
jobId,
|
||
generatedAt: new Date().toISOString(),
|
||
status: finalStatus,
|
||
siteKey: site.key,
|
||
siteId: site.id,
|
||
connectionId,
|
||
idempotencyKey,
|
||
attempt,
|
||
snapshots: {
|
||
wxr: slimRef(wxrSnapshot),
|
||
media_bundle: mediaBundleSnapshot === null ? null : slimRef(mediaBundleSnapshot),
|
||
manual_url_list: manualUrlListSnapshot === null ? null : slimRef(manualUrlListSnapshot),
|
||
},
|
||
versions: VERSIONS,
|
||
// 主列表路径（前置事实 9：CLI 参数来源记入报告——route-writer 实际消费的规范化值）
|
||
paths: { productsPath, newsPath },
|
||
counts: merged.counts,
|
||
issues: cappedIssues,
|
||
issuesTruncated,
|
||
...(cappedGroups === undefined ? {} : { issueGroups: cappedGroups }),
|
||
...(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,
|
||
}
|
||
let reportWritten = true
|
||
try {
|
||
await mkdir(reportDir, { recursive: true })
|
||
// HTML 先写（与 build 域对齐——json 在则 html 必在：JSON 落盘即"双件齐备"信号）
|
||
await writeAtomically(htmlPath, `${renderImportReportHtml(report)}\n`)
|
||
await writeAtomically(reportPath, `${JSON.stringify(report, null, 2)}\n`)
|
||
} catch (err) {
|
||
reportWritten = false
|
||
failed = {
|
||
code: 'REPORT_WRITE_FAILED',
|
||
summary: `[report] ${err instanceof Error ? `${err.name}: ${err.message}` : String(err)}`,
|
||
}
|
||
finalStatus = 'failed'
|
||
}
|
||
const finishedAt = new Date()
|
||
// 终态落行：failed===null ⟹ 写盘成功（report_path 同步入行）；failed≠null 时按
|
||
// 写盘成败决定 report_path 去留（写失败不落——文件不存在不悬挂路径）
|
||
const finalUpdate =
|
||
failed === null
|
||
? { status: finalStatus, finishedAt, reportPath }
|
||
: reportWritten
|
||
? { status: finalStatus, finishedAt, reportPath, errorCode: failed.code, errorSummary: failed.summary }
|
||
: { status: finalStatus, finishedAt, errorCode: failed.code, errorSummary: failed.summary }
|
||
await db.update(importJobs).set(finalUpdate).where(eq(importJobs.id, jobId))
|
||
|
||
return {
|
||
jobId,
|
||
status: finalStatus,
|
||
reused: false,
|
||
siteId: site.id,
|
||
mappingVersion: MAPPING_VERSION,
|
||
siteKey: site.key,
|
||
connectionId,
|
||
wxrSnapshot,
|
||
mediaBundleSnapshot,
|
||
manualUrlListSnapshot,
|
||
versions: VERSIONS,
|
||
reportPath: reportWritten ? reportPath : null,
|
||
durationMs: performance.now() - startedAt,
|
||
}
|
||
}
|
||
|
||
/** 既有 job 的复用返回（不动行、不跑管道；仅 job 行锚定字段）；无则 null */
|
||
async function reuseExisting(
|
||
db: ImportDb,
|
||
idempotencyKey: string,
|
||
): Promise<Pick<CreateImportJobResult, 'jobId' | 'status' | 'reused' | 'siteId' | 'mappingVersion' | 'reportPath'> | null> {
|
||
const found = await db
|
||
.select({
|
||
id: importJobs.id,
|
||
status: importJobs.status,
|
||
siteId: importJobs.siteId,
|
||
mappingVersion: importJobs.mappingVersion,
|
||
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,
|
||
mappingVersion: row.mappingVersion,
|
||
reportPath: row.reportPath,
|
||
}
|
||
}
|
||
|
||
/** 报告里的快照引用瘦身版：不含 metadata（media_bundle 清单可达数千条，报告只留身份锚点） */
|
||
export interface SnapshotRefSlim {
|
||
id: string
|
||
kind: string
|
||
sha256: string
|
||
byteSize: number
|
||
storagePath: string
|
||
}
|
||
|
||
function slimRef(ref: SnapshotRef): SnapshotRefSlim {
|
||
return { id: ref.id, kind: ref.kind, sha256: ref.sha256, byteSize: ref.byteSize, storagePath: ref.storagePath }
|
||
}
|
||
|
||
/**
|
||
* 报告 issues 固化上限（条）：flat 清单与 issueGroups 各组按此截断，超出只计数
|
||
* （issuesTruncated——见 {@link ImportReport.issuesTruncated}）。Plan 2 Task 6
|
||
* ROW_ERRORS_CAP 纪律的导入侧镜像：对抗性输入下单条 item 可产出上万行级问题，
|
||
* 固化全部会使报告体积失去上界；统计维度（counts/sanitize）不受影响始终完整。
|
||
*/
|
||
const REPORT_ISSUES_CAP = 10_000
|
||
|
||
/** issueGroups 各组按 {@link REPORT_ISSUES_CAP} 截断（组溢出不单独入账——分划性质下 ≤ flat 溢出） */
|
||
function capIssueGroups(
|
||
groups: Partial<Record<ImportIssueGroup, string[]>> | undefined,
|
||
): Partial<Record<ImportIssueGroup, string[]>> | undefined {
|
||
if (groups === undefined) return undefined
|
||
const capped: Partial<Record<ImportIssueGroup, string[]>> = {}
|
||
for (const [key, value] of Object.entries(groups) as Array<[ImportIssueGroup, string[] | undefined]>) {
|
||
if (value === undefined) continue
|
||
capped[key] = value.slice(0, REPORT_ISSUES_CAP)
|
||
}
|
||
return capped
|
||
}
|
||
|
||
/**
|
||
* 导入报告对象（import.json 的固化形状——renderImportReportHtml 的输入契约）。
|
||
* issues/issueGroups 已按 {@link REPORT_ISSUES_CAP} 截断（issuesTruncated 记 flat
|
||
* 溢出条数）；对象层保留**原文**（未转义——JSON 归档/机器消费的语义层），HTML
|
||
* 渲染层（report-html.ts）负责转义。
|
||
*/
|
||
export interface ImportReport {
|
||
jobId: string
|
||
generatedAt: string
|
||
status: ImportJobStatus
|
||
siteKey: string
|
||
siteId: string
|
||
connectionId: string
|
||
idempotencyKey: string
|
||
attempt: number
|
||
snapshots: {
|
||
wxr: SnapshotRefSlim
|
||
media_bundle: SnapshotRefSlim | null
|
||
manual_url_list: SnapshotRefSlim | null
|
||
}
|
||
versions: ImportVersionBlock
|
||
/** 主列表路径（前置事实 9：CLI 参数来源——route-writer 实际消费的规范化值） */
|
||
paths: { productsPath: string; newsPath: string }
|
||
counts: Record<string, number>
|
||
/** 行级问题清单（封顶 {@link REPORT_ISSUES_CAP} 条——超出只计数） */
|
||
issues: string[]
|
||
/** 超出 {@link REPORT_ISSUES_CAP} 被丢弃固化的条数（只计数；issueGroups 各组同限截断不另计） */
|
||
issuesTruncated: number
|
||
/** 可选：按来源分组的 issues 视图（各组同限截断） */
|
||
issueGroups?: Partial<Record<ImportIssueGroup, string[]>>
|
||
/** 可选：清洗统计（row 事件 sanitizeDiffs/changed 的聚合——编排管道产出） */
|
||
sanitize?: ImportSanitizeStats
|
||
/** 可选：适配器分段计时（仅真流编排） */
|
||
adapterTiming?: { parseMs: number; sanitizeMs: number }
|
||
error: { code: ImportErrorCode; summary: string } | null
|
||
durationMs: number
|
||
}
|
||
|
||
/** temp+rename 原子写（失败清走 tmp 不留半成品——报告落盘纪律的共用形状） */
|
||
async function writeAtomically(path: string, content: string): Promise<void> {
|
||
const tmpPath = `${path}.tmp`
|
||
try {
|
||
await writeFile(tmpPath, content, 'utf8')
|
||
await rename(tmpPath, path)
|
||
} catch (err) {
|
||
await rm(tmpPath, { force: true })
|
||
throw err
|
||
}
|
||
}
|
||
|
||
/** 缺省空事件流（input.events 缺席时的 ctx.events 占位；每次调用产新迭代器） */
|
||
async function* emptyEvents(): AsyncGenerator<AdapterEvent, void, void> {}
|
||
|
||
// ---- 事件流构造与 stale 回收（编排本体见 orchestrator.ts） ----
|
||
|
||
/** stale 回收阈值：queued/running 滞留超过此时长即宣判（毫秒）——决策与理由见 ADR-017 */
|
||
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<AdapterEvent, void, void> {
|
||
yield* runWxrAdapter(createReadStream(wxrSnapshotCopyPath(snapshot)), {
|
||
mapConfig: { locale },
|
||
timings,
|
||
})
|
||
}
|
||
|