diff --git a/apps/api/src/modules/source/register.service.ts b/apps/api/src/modules/source/register.service.ts index 473a75b..4a462a8 100644 --- a/apps/api/src/modules/source/register.service.ts +++ b/apps/api/src/modules/source/register.service.ts @@ -19,13 +19,17 @@ // 路径安全：输入路径（realpath 后）必须位于 allowedRoots 白名单内——storage/incoming 是 // 常规通道，显式白名单供测试/工具放行夹具目录；越界与 ENOENT 均抛带 reason code 的 // SourceRegisterError。wxr:probe（CLI 只读诊断）不走本服务，无白名单。 +// +// .tmp 残留策略：摄取用 storage/snapshots//.tmp/ 暂存、rename 落位；中途 +// 崩溃/失败可能遗留 .tmp 残段。它们无害（content-addressed 的部分副本，重跑会另起 +// 新 uuid 重写）但不再被本服务清理——手工清理策略：CLI 未运行时整目录删除 .tmp/ 即可。 import { createHash, randomUUID } from 'node:crypto' import { createReadStream, createWriteStream, existsSync } from 'node:fs' import { lstat, mkdir, readdir, realpath, rename, rm, writeFile } from 'node:fs/promises' import { basename, dirname, isAbsolute, join, relative, resolve } from 'node:path' import { fileURLToPath } from 'node:url' import { pipeline } from 'node:stream/promises' -import { and, eq } from 'drizzle-orm' +import { and, eq, sql } from 'drizzle-orm' import type { NodePgDatabase } from 'drizzle-orm/node-postgres' import { ADAPTER_VERSION } from '@sf/adapter-wp' import { sites, sourceConnections, sourceSnapshots } from '../../db/schema.js' @@ -39,6 +43,12 @@ const CHUNK_SIZE = 64 * 1024 export type SourceDb = NodePgDatabase +/** + * 事务/直连两用的最小写入面：db.transaction 回调的 tx（PgTransaction）与 + * NodePgDatabase 都结构满足此形状——三个 ensure* 助手借此在两种上下文通用。 + */ +type SourceDbWriter = Pick + export type SourceSnapshotKind = 'wxr' | 'sitemap' | 'media_bundle' /** 各 kind 快照的 adapter_version：wxr 用适配器包单源版本；sitemap/media 尚无解析器，占位钉死 */ @@ -54,6 +64,7 @@ export type SourceRegisterErrorCode = | 'INPUT_NOT_FOUND' | 'INPUT_NOT_FILE' | 'INPUT_NOT_DIR' + | 'EMPTY_MEDIA_DIR' | 'INPUT_READ_FAILED' | 'COPY_FAILED' @@ -112,6 +123,8 @@ export interface RegisterResult { runId: string siteKey: string siteId: string + /** 本次是否新建站点（false = 按 key 复用既有行，siteDefaults 未生效） */ + siteCreated: boolean connectionId: string connectionCreated: boolean snapshots: RegisteredSnapshot[] @@ -298,6 +311,12 @@ async function ingestMediaDir(srcDir: string, storageRoot: string): Promise ({ path: f.path, sha256: f.sha256 })) const sha256 = createHash('sha256').update(canonicalJson(manifestHashInput)).digest('hex') const totalBytes = scan.files.reduce((sum, f) => sum + f.bytes, 0) @@ -309,13 +328,14 @@ async function ingestMediaDir(srcDir: string, storageRoot: string): Promise { +/** + * site upsert：select-then-insert。插入撞 key 唯一约束（并发同键）时**不在此处重查**—— + * 事务内语句失败即整体中止（25P02），续查必须等回滚后；冲突经 isUniqueViolation 上抛， + * 由 registerSource 的事务级重试整体重开（第二次 select 自然命中复用分支）。 + */ +async function ensureSite( + db: SourceDbWriter, + siteKey: string, + defaults: RegisterSiteDefaults, +): Promise<{ id: string; created: boolean }> { const found = await db.select({ id: sites.id }).from(sites).where(eq(sites.key, siteKey)).limit(1) - if (found.length > 0) return found[0]!.id - try { - const inserted = await db - .insert(sites) - .values({ - key: siteKey, - name: defaults.name ?? siteKey, - defaultLocale: defaults.defaultLocale ?? 'en', - productionOrigin: defaults.productionOrigin ?? `https://${siteKey}.example`, - }) - .returning({ id: sites.id }) - return inserted[0]!.id - } catch (err) { - if (isUniqueViolation(err)) { - const again = await db.select({ id: sites.id }).from(sites).where(eq(sites.key, siteKey)).limit(1) - if (again.length > 0) return again[0]!.id - } - throw err - } + if (found.length > 0) return { id: found[0]!.id, created: false } + const inserted = await db + .insert(sites) + .values({ + key: siteKey, + name: defaults.name ?? siteKey, + defaultLocale: defaults.defaultLocale ?? 'en', + productionOrigin: defaults.productionOrigin ?? `https://${siteKey}.example`, + }) + .returning({ id: sites.id }) + return { id: inserted[0]!.id, created: true } } /** * connection 判重：键 (site_id, name)。**表无此库级唯一约束**（Plan 1 schema 决定）， * 判重在应用层 select-then-insert——本服务是 source_connections 唯一写入方；出现第二个 - * 写入方（REST 通道等）时必须先补库级唯一索引再谈并发。 + * 写入方（REST 通道等）时必须先补库级唯一索引再谈并发（kind 枚举的扩展同样推迟到 + * 那时一并决策，故 kind 暂以 'wxr' 字面量写入）。 + * 并发安全：select-then-insert 无法靠约束兜底，取 pg_advisory_xact_lock(siteId+name) + * 串行化同键并发（事务级锁，COMMIT/ROLLBACK 自动释放——**必须**在事务内调用）。 * configuration 保持空对象：Connection 只描述来源类别（ADR-004），文件事实属于 snapshots。 */ async function ensureConnection( - db: SourceDb, + db: SourceDbWriter, siteId: string, name: string, ): Promise<{ id: string; created: boolean }> { + await db.execute(sql`select pg_advisory_xact_lock(hashtext(${`${siteId}:${name}`}))`) const found = await db .select({ id: sourceConnections.id }) .from(sourceConnections) .where(and(eq(sourceConnections.siteId, siteId), eq(sourceConnections.name, name))) .limit(1) if (found.length > 0) return { id: found[0]!.id, created: false } - try { - const inserted = await db - .insert(sourceConnections) - .values({ siteId, kind: 'wxr', name, configuration: {} }) - .returning({ id: sourceConnections.id }) - return { id: inserted[0]!.id, created: true } - } catch (err) { - if (isUniqueViolation(err)) { - // (site_id,name) 无库级唯一——此分支只可能是站点级联竞态；重查兜底 - const again = await db - .select({ id: sourceConnections.id }) - .from(sourceConnections) - .where(and(eq(sourceConnections.siteId, siteId), eq(sourceConnections.name, name))) - .limit(1) - if (again.length > 0) return { id: again[0]!.id, created: false } - } - throw err - } + const inserted = await db + .insert(sourceConnections) + .values({ siteId, kind: 'wxr', name, configuration: {} }) + .returning({ id: sourceConnections.id }) + return { id: inserted[0]!.id, created: true } } -/** snapshot 幂等落库：UNIQUE(site_id, kind, sha256) 命中复用；不可变表只 INSERT */ +/** + * snapshot 幂等落库：UNIQUE(site_id, kind, sha256) 命中复用；不可变表只 INSERT。 + * 撞唯一约束（并发同输入）同 ensureSite：上抛由事务级重试收敛，不在此处重查（25P02）。 + */ async function ensureSnapshot( - db: SourceDb, - args: { siteId: string; connectionId: string; ingested: Ingested; capturedAt: Date }, + db: SourceDbWriter, + args: { + siteId: string + connectionId: string + ingested: Ingested + /** 落库 metadata（调用方已并入 storageRoot 等登记期事实） */ + metadata: Record + capturedAt: Date + }, ): Promise<{ id: string; reused: boolean }> { - const { siteId, connectionId, ingested, capturedAt } = args + const { siteId, connectionId, ingested, metadata, capturedAt } = args const found = await db .select({ id: sourceSnapshots.id }) .from(sourceSnapshots) @@ -505,45 +526,27 @@ async function ensureSnapshot( ) .limit(1) if (found.length > 0) return { id: found[0]!.id, reused: true } - try { - const inserted = await db - .insert(sourceSnapshots) - .values({ - siteId, - sourceConnectionId: connectionId, - kind: ingested.kind, - storagePath: ingested.storagePath, - sha256: ingested.sha256, - byteSize: ingested.byteSize, - adapterVersion: SNAPSHOT_ADAPTER_VERSION[ingested.kind], - capturedAt, - metadata: ingested.metadata, - }) - .returning({ id: sourceSnapshots.id }) - return { id: inserted[0]!.id, reused: false } - } catch (err) { - if (isUniqueViolation(err)) { - const again = await db - .select({ id: sourceSnapshots.id }) - .from(sourceSnapshots) - .where( - and( - eq(sourceSnapshots.siteId, siteId), - eq(sourceSnapshots.kind, ingested.kind), - eq(sourceSnapshots.sha256, ingested.sha256), - ), - ) - .limit(1) - if (again.length > 0) return { id: again[0]!.id, reused: true } - } - throw err - } + const inserted = await db + .insert(sourceSnapshots) + .values({ + siteId, + sourceConnectionId: connectionId, + kind: ingested.kind, + storagePath: ingested.storagePath, + sha256: ingested.sha256, + byteSize: ingested.byteSize, + adapterVersion: SNAPSHOT_ADAPTER_VERSION[ingested.kind], + capturedAt, + metadata, + }) + .returning({ id: sourceSnapshots.id }) + return { id: inserted[0]!.id, reused: false } } // ---- runId 与报告 ---- -/** runId = 紧凑时间戳 + 6 位随机（同秒多次运行可区分） */ -function makeRunId(): string { +/** runId = 紧凑时间戳 + 6 位随机（同秒多次运行可区分）；CLI 复用同一形态（导出去重） */ +export function makeRunId(): string { const ts = new Date() const compact = `${ts.getFullYear()}${String(ts.getMonth() + 1).padStart(2, '0')}${String( ts.getDate(), @@ -559,14 +562,16 @@ function makeRunId(): string { /** * 登记来源输入：校验（全部输入先过白名单/存在性/类型）→ 逐输入摄取（copy+hash）→ * site/connection/snapshot 落库 → 报告写盘。 - * 失败口径：校验阶段失败零落库零复制；摄取阶段中途失败**不回滚已落位的副本** - * （content-addressed、重跑幂等无害），但 DB 写入在全部摄取完成后才开始—— - * 故任何失败都零落库。返回 RegisteredSnapshot[] 供 CLI 摘要与调用方断言。 + * 失败口径：校验阶段失败零落库零复制；摄取阶段中途失败不回滚已落位的副本 + * （content-addressed、重跑幂等无害）——DB 写入在全部摄取完成后以**单事务**执行 + * （site→connection→snapshots 全成或全回滚），故任何失败都零落库。 + * 每条 snapshot 的 metadata 固化登记期事实（storageRoot 绝对路径等——快照不可变， + * 事后无法补写）。返回 RegisteredSnapshot[] 供 CLI 摘要与调用方断言。 */ export async function registerSource(db: SourceDb, input: RegisterInput): Promise { const startedAt = performance.now() const allowedRoots = input.allowedRoots ?? [join(REPO_ROOT, 'storage', 'incoming')] - const storageRoot = input.storageRoot ?? join(REPO_ROOT, 'storage') + const storageRoot = resolve(input.storageRoot ?? join(REPO_ROOT, 'storage')) const reportRoot = input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'register') // 阶段 0：全部输入先校验（fail fast——校验失败零落库零复制；摄取中途失败口径见函数 JSDoc） @@ -588,26 +593,53 @@ export async function registerSource(db: SourceDb, input: RegisterInput): Promis if (sitemapReal !== undefined) ingested.push(await ingestFile('sitemap', sitemapReal, storageRoot)) if (mediaReal !== undefined) ingested.push(await ingestMediaDir(mediaReal, storageRoot)) - // 阶段 2：DB 写入 + // 阶段 2：DB 写入——单事务（site→connection→snapshots 全成或全回滚；"任何失败零落库" + // 由此成立）。connection 的 select-then-insert 在事务内靠 advisory xact lock 串行化。 + // 并发同键撞唯一约束时事务整体中止（语句失败后续查会 25P02）——不做事务内补救， + // 由这层事务级重试整体重开：第二次 select-then-insert 自然命中复用分支（上限 3 次）。 const capturedAt = new Date() - const siteId = await ensureSite(db, input.siteKey, input.siteDefaults ?? {}) - const connection = await ensureConnection(db, siteId, input.connectionName ?? 'default') - const snapshots: RegisteredSnapshot[] = [] - for (const item of ingested) { - const { id, reused } = await ensureSnapshot(db, { siteId, connectionId: connection.id, ingested: item, capturedAt }) - snapshots.push({ - kind: item.kind, - id, - sha256: item.sha256, - byteSize: item.byteSize, - storagePath: item.storagePath, - sourcePath: item.sourcePath, - reused, - copied: item.copied, - durationMs: item.durationMs, - metadata: item.metadata, + const writeAll = async () => + db.transaction(async (tx) => { + const site = await ensureSite(tx, input.siteKey, input.siteDefaults ?? {}) + const connection = await ensureConnection(tx, site.id, input.connectionName ?? 'default') + const snapshots: RegisteredSnapshot[] = [] + for (const item of ingested) { + // storageRoot 固化进每条 snapshot 的 metadata：storagePath 只是相对锚，登记后 + // 无法从快照行反推它当初落在哪个存储根（快照不可变，不可事后补写） + const metadata = { ...item.metadata, storageRoot } + const { id, reused } = await ensureSnapshot(tx, { + siteId: site.id, + connectionId: connection.id, + ingested: item, + metadata, + capturedAt, + }) + snapshots.push({ + kind: item.kind, + id, + sha256: item.sha256, + byteSize: item.byteSize, + storagePath: item.storagePath, + sourcePath: item.sourcePath, + reused, + copied: item.copied, + durationMs: item.durationMs, + metadata, + }) + } + return { site, connection, snapshots } }) + let written: Awaited> + for (let attempt = 0; ; attempt++) { + try { + written = await writeAll() + break + } catch (err) { + if (isUniqueViolation(err) && attempt < 3) continue + throw err + } } + const { site, connection, snapshots } = written // 阶段 3：报告（阶段 2 无 import_jobs，报告按 runId 归档——//reports/） const runId = makeRunId() @@ -616,15 +648,18 @@ export async function registerSource(db: SourceDb, input: RegisterInput): Promis runId, generatedAt: new Date().toISOString(), siteKey: input.siteKey, - siteId, + siteId: site.id, + siteCreated: site.created, connectionId: connection.id, connectionCreated: connection.created, + storageRoot, inputs: snapshots.map((s) => ({ kind: s.kind, sourcePath: s.sourcePath, sha256: s.sha256, byteSize: s.byteSize, storagePath: s.storagePath, + storageRoot, reused: s.reused, copied: s.copied, durationMs: s.durationMs, @@ -645,7 +680,8 @@ export async function registerSource(db: SourceDb, input: RegisterInput): Promis return { runId, siteKey: input.siteKey, - siteId, + siteId: site.id, + siteCreated: site.created, connectionId: connection.id, connectionCreated: connection.created, snapshots, diff --git a/docs/superpowers/plans/2026-10-20-slice-wxr-adapter-phase2.md b/docs/superpowers/plans/2026-10-20-slice-wxr-adapter-phase2.md index 4a27f92..7a49960 100644 --- a/docs/superpowers/plans/2026-10-20-slice-wxr-adapter-phase2.md +++ b/docs/superpowers/plans/2026-10-20-slice-wxr-adapter-phase2.md @@ -141,6 +141,7 @@ cms-one/ ### Task 9: 阶段 2 收尾 - [ ] ADR-016（saxes/commander/sanitize-html 选型与许可证）；根 typecheck 链加 @sf/adapter-wp；lint 全绿；contracts 首个消费者接线验证（adapter-wp workspace 依赖 contracts——**Plan 1 终审前瞻项落实**：import 路径用 dist 还是 src 的决策一并入 ADR-016） +- [ ] **Task 7 审查移交**：① connection (site_id,name) 库级唯一索引须在第二个写入方（REST 通道）出现前完成 ② media_bundle→WXR attachment URL 的映射策略（uploads 路径约定）记入阶段 3 计划输入 ③ REST 化时重审：白名单 TOCTOU、cwd 自动放行、connection.kind 枚举扩 ④ sanitize-html engines>=22.12 与根 engines>=22 的口径（升根或记 ADR） - [ ] 全量三绿 + push；plan 勾选；**移交阶段 3 清单**（内容/分类/媒体入库：media 魔数校验、source_records 落库、稳定身份 upsert、checksum skip、route_registry 生成；**续传携带：Task 0.2 cms 数据备份（用户暂缓，阶段 6 L3 验收前必须补做**——责任链不断） ## 验收（对照 spec §15 阶段 2） diff --git a/scripts/sf.ts b/scripts/sf.ts index 4d57a90..016ab69 100644 --- a/scripts/sf.ts +++ b/scripts/sf.ts @@ -13,18 +13,15 @@ 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 { registerSource, SourceRegisterError, type RegisterInput } from '../apps/api/src/modules/source/register.service.js' +import { + makeRunId, + registerSource, + SourceRegisterError, + type RegisterInput, +} from '../apps/api/src/modules/source/register.service.js' const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..') -/** runId = 紧凑时间戳 + 6 位随机（与 register.service 同形；报告目录按 runId 归档） */ -function makeRunId(): string { - const ts = new Date() - const p = (n: number) => String(n).padStart(2, '0') - const compact = `${ts.getFullYear()}${p(ts.getMonth() + 1)}${p(ts.getDate())}T${p(ts.getHours())}${p(ts.getMinutes())}${p(ts.getSeconds())}` - return `${compact}-${Math.random().toString(36).slice(2, 8)}` -} - /** 探测报告摘要（stdout）：版本块之外的计数/清洗/耗时 + 致命错误 */ function printProbeSummary(report: WxrAdapterReport): void { console.log(` rows=${report.totals.rows} terms=${report.totals.terms} unsupported=${report.totals.unsupported} failed=${report.totals.failed}`) @@ -91,30 +88,48 @@ program if (opts.wxr === undefined && opts.sitemap === undefined && opts.media === undefined) { throw new SourceRegisterError('NO_INPUTS', '至少提供 --wxr / --sitemap / --media 之一') } + // fail fast：建站默认列在服务端只在"站点新建"时生效——先做形状校验，避免跑完摄取才报错 + if (opts.origin !== undefined) { + let parsed: URL + try { + parsed = new URL(opts.origin) + } catch { + throw new Error(`--origin 不是合法 URL：${opts.origin}`) + } + if (parsed.protocol !== 'https:' && parsed.protocol !== 'http:') { + throw new Error(`--origin 必须是 http(s) URL：${opts.origin}`) + } + } + if (opts.locale !== undefined && !/^[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*$/.test(opts.locale)) { + throw new Error(`--locale 不符合语言标签形状（如 en、zh-CN）：${opts.locale}`) + } // CLI 是可信操作者通道：白名单 = storage/incoming（常规通道）+ 操作者所在 cwd（显式 // 于此运行的动作本身即放行依据）+ --allow 追加根。服务内的白名单校验（PATH_NOT_ALLOWED） - // 防的是未来程序化/REST 通道的越界读取。 + // 防的是未来程序化/REST 通道的越界读取。生效白名单先行打印（审计：路径放行决策可回溯）。 + const allowedRoots = [join(REPO_ROOT, 'storage', 'incoming'), process.cwd(), ...opts.allow.map((d) => resolve(d))] const input: RegisterInput = { siteKey: opts.site, - allowedRoots: [join(REPO_ROOT, 'storage', 'incoming'), process.cwd(), ...opts.allow.map((d) => resolve(d))], + allowedRoots, storageRoot: join(REPO_ROOT, 'storage'), reportRoot: join(REPO_ROOT, 'artifacts', 'register'), } if (opts.wxr !== undefined) input.wxrPath = opts.wxr if (opts.sitemap !== undefined) input.sitemapPath = opts.sitemap if (opts.media !== undefined) input.mediaDir = opts.media - if (opts.name !== undefined || opts.locale !== undefined || opts.origin !== undefined) { + const defaultsPassed = opts.name !== undefined || opts.locale !== undefined || opts.origin !== undefined + if (defaultsPassed) { const defaults: NonNullable = {} if (opts.name !== undefined) defaults.name = opts.name if (opts.locale !== undefined) defaults.defaultLocale = opts.locale if (opts.origin !== undefined) defaults.productionOrigin = opts.origin input.siteDefaults = defaults } + console.log(`[source:register] 允许目录：${allowedRoots.join(' ; ')}`) // getDb 走 DATABASE_URL_TEST ?? DATABASE_URL 回退链——对临时库冒烟只需 env 覆盖，无需改代码 const db = getDb() try { const result = await registerSource(db, input) - console.log(`[source:register] runId=${result.runId} site=${result.siteKey} (${result.siteId})`) + console.log(`[source:register] runId=${result.runId} site=${result.siteKey} (${result.siteId}${result.siteCreated ? ' 新建' : ' 已存在'})`) console.log( ` connection=${result.connectionId}${result.connectionCreated ? ' (created)' : ' (existing)'}`, ) @@ -123,6 +138,9 @@ program ` ${s.kind}: sha256=${s.sha256.slice(0, 12)}… bytes=${s.byteSize} ${s.reused ? 'snapshot复用' : 'snapshot新建'}${s.copied ? '' : ' 副本已存在'} (${s.durationMs.toFixed(1)}ms)`, ) } + if (!result.siteCreated && defaultsPassed) { + console.error(' 注意：站点已存在，--name/--locale/--origin 未生效') + } console.log(` report: ${result.reportPath} (total ${result.durationMs.toFixed(1)}ms)`) } finally { await (db as { $client: { end: () => Promise } }).$client.end() @@ -134,6 +152,9 @@ program 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 { console.error(`[sf] ${err instanceof Error ? err.message : String(err)}`) } diff --git a/tests/integration/source-register.test.ts b/tests/integration/source-register.test.ts index 73eb815..097bb51 100644 --- a/tests/integration/source-register.test.ts +++ b/tests/integration/source-register.test.ts @@ -10,7 +10,9 @@ // 副本内容与原输入一致（后续任务只读副本）； // 5. 路径安全：白名单外 / 越界遍历 / **中间段链接逃逸（整路径 realpath 后判定）** / // ENOENT / 类型不符均清晰报错且零落库； -// 6. 报告落 artifacts/register//reports/register.json（runId=时间戳+随机）。 +// 6. 报告落 artifacts/register//reports/register.json（runId=时间戳+随机）； +// 7. 并发登记同输入恰好一组行（DB 单事务 + 唯一约束/advisory lock 收敛）、双方同 id； +// 8. 空媒体目录拒绝（EMPTY_MEDIA_DIR）、metadata 固化 storageRoot、siteCreated 语义。 // 运行于 globalSetup 临时库（ADR-015）；storage/报告根用 os.tmpdir 临时目录，finally 清理。 // 注：符号链接跳过用目录 junction 验证（Windows 免特权；非 Windows 退化为普通符号链接）； // 创建失败（极端沙箱）时该断言静默放行——junction 创建不是本测试的被测契约。 @@ -22,6 +24,7 @@ import { join, resolve } from 'node:path' import { fileURLToPath } from 'node:url' import { describe, expect, it } from 'vitest' import { + isUniqueViolation, registerSource, SourceRegisterError, type RegisterInput, @@ -204,6 +207,8 @@ describe('source:register 服务', () => { const r2 = await registerSource(db, makeInput(ctx, key)) expect(r2.siteId).toBe(r1.siteId) + expect(r2.siteCreated).toBe(false) + expect(r1.siteCreated).toBe(true) expect(r2.connectionId).toBe(r1.connectionId) expect(r2.connectionCreated).toBe(false) expect(r2.snapshots).toHaveLength(3) @@ -248,6 +253,8 @@ describe('source:register 服务', () => { } // 副本路径 ≠ 源路径（后续任务只读副本） expect(dest).not.toBe(s.sourcePath) + // storageRoot 固化进 metadata（快照不可变，事后无法补写） + expect((s.metadata as { storageRoot: string }).storageRoot).toBe(resolve(ctx.storageRoot)) } } finally { await cleanupSiteRows(db, r.siteId) @@ -302,10 +309,16 @@ describe('source:register 服务', () => { input: { ...base }, code: 'NO_INPUTS', }, + { + label: '空媒体目录', + input: { ...base, mediaDir: join(ctx.work, 'empty-media') }, + code: 'EMPTY_MEDIA_DIR', + }, ] - // 越界遍历用例的配套文件：work 根下放 escape.txt，incoming/ 只是白名单根 + // 用例配套布局：work 根放 escape.txt（越界目标）、incoming/（收窄白名单根）、empty-media/（空目录） await writeFile(join(ctx.work, 'escape.txt'), 'escape') await mkdir(join(ctx.work, 'incoming'), { recursive: true }) + await mkdir(join(ctx.work, 'empty-media'), { recursive: true }) for (const c of cases) { const err: SourceRegisterError = await registerSource(db, c.input).then( @@ -403,17 +416,20 @@ describe('source:register 服务', () => { runId: string siteKey: string siteId: string - inputs: { kind: string; sourcePath: string; sha256: string; byteSize: number; durationMs: number }[] + storageRoot: string + inputs: { kind: string; sourcePath: string; sha256: string; byteSize: number; storageRoot: string; durationMs: number }[] totals: { snapshots: number; reused: number; byteSize: number } durationMs: number } expect(report.runId).toBe(r.runId) expect(report.siteKey).toBe(key) expect(report.siteId).toBe(r.siteId) + expect(report.storageRoot).toBe(resolve(ctx.storageRoot)) expect(report.inputs).toHaveLength(3) const wxrInput = report.inputs.find((i) => i.kind === 'wxr')! expect(wxrInput.sha256).toBe(sha256(await readFile(WXR_FIXTURE))) expect(wxrInput.byteSize).toBe((await readFile(WXR_FIXTURE)).length) + expect(wxrInput.storageRoot).toBe(resolve(ctx.storageRoot)) expect(wxrInput.durationMs).toBeGreaterThanOrEqual(0) expect(report.totals.snapshots).toBe(3) expect(report.totals.reused).toBe(0) @@ -426,4 +442,59 @@ describe('source:register 服务', () => { }) }) }) + + it('并发 register 同输入：恰好一组行，双方返回同一组 id（事务 + 唯一约束/advisory lock 收敛）', async () => { + await withTestDb(async (db) => { + await withCtx(async (ctx) => { + const key = `sfreg-${rand()}` + const input: RegisterInput = { + siteKey: key, + wxrPath: WXR_FIXTURE, + allowedRoots: ctx.allowedRoots, + storageRoot: ctx.storageRoot, + reportRoot: ctx.reportRoot, + } + const [r1, r2] = await Promise.all([registerSource(db, input), registerSource(db, input)]) + try { + // 并发终态契约：同 id、同快照；"谁新建"分支按竞态分配，但恰好各一方 + expect(r1.siteId).toBe(r2.siteId) + expect(r1.connectionId).toBe(r2.connectionId) + expect(r1.snapshots.map((s: RegisteredSnapshot) => s.id)).toEqual( + r2.snapshots.map((s: RegisteredSnapshot) => s.id), + ) + expect(Number(r1.siteCreated) + Number(r2.siteCreated)).toBe(1) + expect(Number(r1.connectionCreated) + Number(r2.connectionCreated)).toBe(1) + expect( + r1.snapshots.map((s: RegisteredSnapshot) => !s.reused).filter(Boolean).length + + r2.snapshots.map((s: RegisteredSnapshot) => !s.reused).filter(Boolean).length, + ).toBe(1) + const counts = (await db.execute( + `select (select count(*) from source_snapshots where site_id = '${r1.siteId}') as snaps, + (select count(*) from source_connections where site_id = '${r1.siteId}') as conns, + (select count(*) from sites where key = '${key}') as sites`, + )) as any + expect(counts[0]).toEqual({ snaps: '1', conns: '1', sites: '1' }) + } finally { + await cleanupSiteRows(db, r1.siteId) + } + }) + }) + }) + + it('isUniqueViolation 形状识别：drizzle 包裹（cause.code）/裸 postgres code/其他均正确', () => { + // drizzle 会把 postgres 错误包成 DrizzleQueryError：message 只剩 "Failed query"，真 code 在 cause + const drizzleWrapped = Object.assign(new Error('Failed query: insert into sites ...'), { + cause: Object.assign(new Error('duplicate key value violates unique constraint "sites_key_unique"'), { + code: '23505', + }), + }) + expect(isUniqueViolation(drizzleWrapped)).toBe(true) + // 裸 postgres 错误（code 直挂） + expect(isUniqueViolation(Object.assign(new Error('dup'), { code: '23505' }))).toBe(true) + // 其他错误码 / 无 code → false + expect(isUniqueViolation(Object.assign(new Error('x'), { cause: { code: '23508' } }))).toBe(false) + expect(isUniqueViolation(Object.assign(new Error('x'), { code: '42P01' }))).toBe(false) + expect(isUniqueViolation(new Error('plain'))).toBe(false) + expect(isUniqueViolation(null)).toBe(false) + }) })