fix(source): DB 事务化；storageRoot 入 metadata；siteCreated 与并发测试；CLI 体验与校验小修

评审修复（4 项 Important + 8 项 Minor）：
- DB 事务化：site→connection→snapshots 全部落库包进 db.transaction 单事务
  （"任何失败零落库"字面成立）；助手改收 Pick<select|insert|execute> 结构类型
  （tx 与直连 db 通用）。并发撞唯一约束不在事务内补查（语句失败后续查 25P02），
  改为 registerSource 层事务级重试（≤3 次，第二次 select 命中复用分支）
- connection 并发：select-then-insert 无约束可兜底，事务内 pg_advisory_xact_lock
  (siteId:name) 串行化同键登记（kind 枚举扩展推迟 REST 通道，注释注明）
- storageRoot 绝对路径固化进每条 snapshot metadata 与报告 inputs（快照不可变，
  事后无法补写——永久缺口封堵）
- ensureSite 返回 created；RegisterResult.siteCreated；CLI 站点已存在时提示
  --name/--locale/--origin 未生效
- 媒体复制期逐文件 sha 复核（copyAndHash 对拍清单，源中途变更 → COPY_FAILED）
- 空媒体目录拒绝（EMPTY_MEDIA_DIR）
- CLI：PATH_NOT_ALLOWED 追加 --allow 提示；--origin 经 new URL 校验（http/s）、
  --locale 语言标签形状校验，fail fast；开跑打印生效白名单（审计）
- makeRunId 服务导出、CLI 复用；.tmp 残留手工清理策略写入文件头注释
- 测试 +2：并发同输入恰好一组行/双方同 id/新建标志恰一（3 次重跑稳定）；
  isUniqueViolation 形状单测（drizzle cause.code / 裸 code / 其他）

全套 263 绿；typecheck / lint 绿；CLI 冒烟对临时库验证事务路径、siteCreated
输出、defaults 落库（en-US / https://qa.example）、白名单打印与校验报错

Co-Authored-By: Claude Code <noreply@anthropic.com>

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/<kind>/.tmp/<uuid> 暂存、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<typeof schema>
 
+/**
+ * 事务/直连两用的最小写入面：db.transaction 回调的 tx（PgTransaction）与
+ * NodePgDatabase 都结构满足此形状——三个 ensure* 助手借此在两种上下文通用。
+ */
+type SourceDbWriter = Pick<SourceDb, 'select' | 'insert' | 'execute'>
+
 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<Inge
     )
   }
   // 清单哈希输入只含 path+sha256（bytes 只进 metadata，不进哈希——定义钉死）
+  if (scan.files.length === 0) {
+    throw new SourceRegisterError(
+      'EMPTY_MEDIA_DIR',
+      `media 目录为空（未发现任何常规文件）：${srcDir}——请检查路径是否正确`,
+    )
+  }
   const manifestHashInput = scan.files.map((f) => ({ 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<Inge
     await mkdir(tmpRoot, { recursive: true })
     const tmp = join(tmpRoot, randomUUID())
     try {
+      // 复制期逐文件复核 sha：登记窗口内源目录被改写会使副本与清单哈希脱钩——发现即失败
       for (const f of scan.files) {
         const destFile = join(tmp, f.path)
         await mkdir(dirname(destFile), { recursive: true })
-        await pipeline(
-          createReadStream(join(srcDir, f.path), { highWaterMark: CHUNK_SIZE }),
-          createWriteStream(destFile, { highWaterMark: CHUNK_SIZE }),
-        )
+        const { sha256: copiedSha } = await copyAndHash(join(srcDir, f.path), destFile)
+        if (copiedSha !== f.sha256) {
+          throw new Error(`源文件在登记期间变更：${join(srcDir, f.path)}（清单 ${f.sha256.slice(0, 12)}… ≠ 复制 ${copiedSha.slice(0, 12)}…）`)
+        }
       }
       copied = await renameIntoPlace(tmp, dest)
     } catch (err) {
@@ -419,80 +439,81 @@ async function validateInputPath(
 
 // ---- DB 写入（drizzle） ----
 
-/** postgres unique violation（23505）——drizzle 会包成 DrizzleQueryError，真 code 在 cause */
-function isUniqueViolation(err: unknown): boolean {
+/** postgres unique violation（23505）——drizzle 会包成 DrizzleQueryError，真 code 在 cause（导出供单测钉形状） */
+export function isUniqueViolation(err: unknown): boolean {
   const e = err as { code?: unknown; cause?: { code?: unknown } }
   const code = e?.cause?.code ?? e?.code
   return code === '23505'
 }
 
-/** site upsert：select-then-insert；插入撞 key 唯一约束（并发竞态）则重查一次 */
-async function ensureSite(db: SourceDb, siteKey: string, defaults: RegisterSiteDefaults): Promise<string> {
+/**
+ * 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<string, unknown>
+    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<RegisterResult> {
   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<ReturnType<typeof writeAll>>
+  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 归档——<reportRoot>/<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,