feat(source): 输入登记服务与 sf CLI（content-addressed 快照）

- registerSource(db, input)：wxr/sitemap/媒体目录三类输入登记为 content-addressed 快照——流式 copy+sha256 单遍双算（64KB 块），副本落 storage/snapshots/<kind>/<sha256前2>/<sha256>（文件）或 <sha256>/（媒体整树），后续任务只读副本；UNIQUE(site,kind,sha256) 幂等（重复登记复用既有行） - media_bundle 哈希钉死：排序相对路径（'/' 归一）+ 每文件 sha256 → canonical JSON（键排序无空白）→ 再 sha256；不含 mtime、不依赖遍历顺序；完整清单（path/bytes/sha256）入 metadata jsonb；符号链接（含 junction）跳过并记录 - site 按 key select-then-insert upsert（23505 竞态重查）；connection 判重键 (site_id,name) 为应用层 select-then-insert（表无库级唯一，注释注明） - 路径安全：输入 realpath 后须在 allowedRoots 白名单内（缺省 storage/incoming）， ENOENT/越界/类型不符抛带 code 的 SourceRegisterError；全部输入先校验后摄取，任何失败零落库零复制 - 报告落 artifacts/register/<runId>/reports/register.json（runId=时间戳+随机；输入清单/哈希/大小/耗时） - scripts/sf.ts（commander）：wxr:probe 只读诊断（任意可读路径，IO/损坏文件出报告不崩——Task 6 fatalError 契约）出 JSON/HTML 报告；source:register 调服务出摘要。CLI 机制钉死：根 devDep tsx + 根 script sf + commander 根依赖，TS 源码直连（@sf/* 经 tsconfig paths，不经 dist） - source.module.ts：Nest 注册 SourceRegisterService（standalone context 用）； register.service.ts 刻意零装饰器（CLI/测试直 import 不拉 reflect-metadata） - launder 类型垫片改经 triple-slash 引用随 html.ts 进任意编译程序（apps/api 与根程序直连适配器源码的新场景）；apps/api 增 @sf/adapter-wp workspace 依赖与 paths 映射（register.service 消费 ADAPTER_VERSION 单源） - 测试 +7（tests/integration/source-register.test.ts，临时库 + tmpdir storage/ 报告根 finally 清理）：三输入登记/清单哈希定义对拍/重复登记幂等/mtime 排除/ 副本布局与内容一致/路径安全六分支/junction 跳过/报告契约；全套 260 绿 Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-21 12:31:15 +08:00
parent a535cde270
commit 586d715455
10 changed files with 1246 additions and 3 deletions
@@ -0,0 +1,654 @@
+// source:register 输入登记服务（Plan 2 Task 7，spec §7.1 / ADR-004/008）：
+// 把 WXR / sitemap / 媒体目录三类输入登记为 content-addressed 快照——
+// 副本落 storage/snapshots/<kind>/<sha256前2>/<sha256>（文件）或 <sha256>/（媒体整树），
+// 源输入文件从此不再被后续任务读取（导入只寻址副本）。
+//
+// 本文件刻意保持 **零装饰器**：CLI（scripts/sf.ts 经 tsx）与集成测试直接 import 本文件，
+// 不应因此拉起 Nest 装饰器执行（@Module/@Injectable 需 reflect-metadata 先行，见
+// source.module.ts 的 standalone 说明）。Nest 侧由 SourceRegisterService 薄包装注入。
+//
+// 哈希定义（钉死，测试独立重算对拍）：
+// - 文件输入：内容 sha256（流式，64KB 块，copy 与 hash 单遍双算）；
+// - media_bundle：排序相对路径（分隔符统一 '/'）+ 每文件 sha256 → canonical JSON
+//   （对象键排序、无空白）→ 对该 JSON 串再 sha256。不含 mtime、不依赖遍历顺序。
+//
+// 幂等：source_snapshots UNIQUE(site_id, kind, sha256)——同输入重复 register 复用既有行；
+// site 按 key select-then-insert upsert；connection 判重键 (site_id, name) **无库级唯一
+// 约束**，靠应用层 select-then-insert 判重（本服务是唯一写入方；引入并发写入方时须补库级约束）。
+//
+// 路径安全：输入路径（realpath 后）必须位于 allowedRoots 白名单内——storage/incoming 是
+// 常规通道，显式白名单供测试/工具放行夹具目录；越界与 ENOENT 均抛带 reason code 的
+// SourceRegisterError。wxr:probe（CLI 只读诊断）不走本服务，无白名单。
+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 type { NodePgDatabase } from 'drizzle-orm/node-postgres'
+import { ADAPTER_VERSION } from '@sf/adapter-wp'
+import { sites, sourceConnections, sourceSnapshots } from '../../db/schema.js'
+import type * as schema from '../../db/schema.js'
+
+/** 本文件位于 <repo>/apps/api/src/modules/source/ → 上五级是仓库根 */
+const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../../../..')
+
+/** 流式读写块大小（copy 与 hash 同块；64KB 与 maxWxrBytes 限额家族同一量级约定） */
+const CHUNK_SIZE = 64 * 1024
+
+export type SourceDb = NodePgDatabase<typeof schema>
+
+export type SourceSnapshotKind = 'wxr' | 'sitemap' | 'media_bundle'
+
+/** 各 kind 快照的 adapter_version：wxr 用适配器包单源版本；sitemap/media 尚无解析器，占位钉死 */
+const SNAPSHOT_ADAPTER_VERSION: Record<SourceSnapshotKind, string> = {
+  wxr: `adapter-wp@${ADAPTER_VERSION}`,
+  sitemap: 'sitemap@0',
+  media_bundle: 'media-bundle@1',
+}
+
+export type SourceRegisterErrorCode =
+  | 'NO_INPUTS'
+  | 'PATH_NOT_ALLOWED'
+  | 'INPUT_NOT_FOUND'
+  | 'INPUT_NOT_FILE'
+  | 'INPUT_NOT_DIR'
+  | 'INPUT_READ_FAILED'
+  | 'COPY_FAILED'
+
+/** 登记失败（可预期错误）：CLI 以 code 区分呈现；不裹程序 bug（那些原样上抛） */
+export class SourceRegisterError extends Error {
+  readonly code: SourceRegisterErrorCode
+  constructor(code: SourceRegisterErrorCode, message: string) {
+    super(message)
+    this.name = 'SourceRegisterError'
+    this.code = code
+  }
+}
+
+export interface RegisterSiteDefaults {
+  name?: string
+  defaultLocale?: string
+  productionOrigin?: string
+}
+
+export interface RegisterInput {
+  siteKey: string
+  wxrPath?: string
+  sitemapPath?: string
+  mediaDir?: string
+  /** 输入路径白名单（realpath 后必须位于其中之一）；缺省 [repoRoot/storage/incoming] */
+  allowedRoots?: string[]
+  /** 快照存储根（副本落 <storageRoot>/snapshots/...）；缺省 repoRoot/storage */
+  storageRoot?: string
+  /** 报告根（落 <reportRoot>/<runId>/reports/register.json）；缺省 repoRoot/artifacts/register */
+  reportRoot?: string
+  /** 连接名——(site_id, name) 应用层判重键的 name 侧；缺省 'default' */
+  connectionName?: string
+  /** 站点不存在时的建站默认列 */
+  siteDefaults?: RegisterSiteDefaults
+}
+
+export interface RegisteredSnapshot {
+  kind: SourceSnapshotKind
+  /** snapshot 行 id */
+  id: string
+  sha256: string
+  byteSize: number
+  /** 相对 storageRoot 的副本路径（posix 分隔符，跨平台稳定） */
+  storagePath: string
+  /** 原输入绝对路径（审计用；副本才是后续任务的读取对象） */
+  sourcePath: string
+  /** 本次是否新插入行（false = UNIQUE(site,kind,sha256) 复用既有快照） */
+  reused: boolean
+  /** 本次是否写入副本文件（false = content-addressed 副本已存在） */
+  copied: boolean
+  durationMs: number
+  metadata: Record<string, unknown>
+}
+
+export interface RegisterResult {
+  runId: string
+  siteKey: string
+  siteId: string
+  connectionId: string
+  connectionCreated: boolean
+  snapshots: RegisteredSnapshot[]
+  reportPath: string
+  durationMs: number
+}
+
+export interface MediaManifestEntry {
+  /** 相对媒体目录根的路径，分隔符统一 '/' */
+  path: string
+  bytes: number
+  sha256: string
+}
+
+// ---- 哈希与副本 ----
+
+/** 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<string, unknown>
+    const body = Object.keys(record)
+      .sort()
+      .map((k) => `${JSON.stringify(k)}:${canonicalJson(record[k])}`)
+      .join(',')
+    return `{${body}}`
+  }
+  return JSON.stringify(value)
+}
+
+/** 流式单遍 copy+sha256 双算：读 64KB 块 → 哈希更新 → 写目标（pipeline 管背压与错误传播） */
+async function copyAndHash(src: string, dest: string): Promise<{ sha256: string; bytes: number }> {
+  const hash = createHash('sha256')
+  let bytes = 0
+  const through = async function* (chunks: AsyncIterable<Buffer>) {
+    for await (const chunk of chunks) {
+      hash.update(chunk)
+      bytes += chunk.length
+      yield chunk
+    }
+  }
+  try {
+    await pipeline(
+      createReadStream(src, { highWaterMark: CHUNK_SIZE }),
+      through,
+      createWriteStream(dest, { highWaterMark: CHUNK_SIZE }),
+    )
+  } catch (err) {
+    await rm(dest, { force: true })
+    throw err
+  }
+  return { sha256: hash.digest('hex'), bytes }
+}
+
+/** 流式哈希单文件（媒体扫描阶段用——只读不复制） */
+async function hashFile(src: string): Promise<{ sha256: string; bytes: number }> {
+  const hash = createHash('sha256')
+  let bytes = 0
+  for await (const chunk of createReadStream(src, { highWaterMark: CHUNK_SIZE })) {
+    hash.update(chunk)
+    bytes += chunk.length
+  }
+  return { sha256: hash.digest('hex'), bytes }
+}
+
+interface MediaScan {
+  /** 按 path 字典序排序（哈希定义要求与遍历顺序无关） */
+  files: MediaManifestEntry[]
+  /** 被跳过的符号链接（含 junction）与其他非常规文件的相对路径 */
+  skipped: string[]
+}
+
+/** 递归枚举媒体目录：跳过符号链接（记录之），文件逐个流式哈希 */
+async function scanMediaDir(root: string): Promise<MediaScan> {
+  const files: MediaManifestEntry[] = []
+  const skipped: string[] = []
+  const walk = async (abs: string, rel: string): Promise<void> => {
+    for (const entry of await readdir(abs, { withFileTypes: true })) {
+      const childRel = rel === '' ? entry.name : `${rel}/${entry.name}`
+      const childAbs = join(abs, entry.name)
+      if (entry.isSymbolicLink()) {
+        skipped.push(childRel)
+        continue
+      }
+      if (entry.isDirectory()) {
+        await walk(childAbs, childRel)
+        continue
+      }
+      if (!entry.isFile()) {
+        skipped.push(childRel)
+        continue
+      }
+      const { sha256, bytes } = await hashFile(childAbs)
+      files.push({ path: childRel, bytes, sha256 })
+    }
+  }
+  await walk(root, '')
+  files.sort((a, b) => (a.path < b.path ? -1 : a.path > b.path ? 1 : 0))
+  return { files, skipped }
+}
+
+/** 并发写同一 content-addressed 目标的兜底：rename 撞已有副本视为命中（幂等），删临时件 */
+async function renameIntoPlace(tmp: string, dest: string): Promise<boolean> {
+  if (existsSync(dest)) {
+    await rm(tmp, { recursive: true, force: true })
+    return false
+  }
+  await mkdir(dirname(dest), { recursive: true })
+  try {
+    await rename(tmp, dest)
+    return true
+  } catch (err) {
+    if (existsSync(dest)) {
+      await rm(tmp, { recursive: true, force: true })
+      return false
+    }
+    throw err
+  }
+}
+
+interface Ingested {
+  kind: SourceSnapshotKind
+  sourcePath: string
+  sha256: string
+  byteSize: number
+  storagePath: string
+  copied: boolean
+  metadata: Record<string, unknown>
+  durationMs: number
+}
+
+/** 文件输入（wxr/sitemap）：temp 副本上单遍 copy+hash → 按 sha 定名落位（已存在则命中复用） */
+async function ingestFile(
+  kind: 'wxr' | 'sitemap',
+  src: string,
+  storageRoot: string,
+): Promise<Ingested> {
+  const t0 = performance.now()
+  const tmpDir = join(storageRoot, 'snapshots', kind, '.tmp')
+  await mkdir(tmpDir, { recursive: true })
+  const tmp = join(tmpDir, `${randomUUID()}.part`)
+  let sha256: string
+  let bytes: number
+  try {
+    ;({ sha256, bytes } = await copyAndHash(src, tmp))
+  } catch (err) {
+    throw new SourceRegisterError(
+      'INPUT_READ_FAILED',
+      `${kind} 输入读取失败：${src}（${err instanceof Error ? err.message : String(err)}）`,
+    )
+  }
+  const rel = `snapshots/${kind}/${sha256.slice(0, 2)}/${sha256}`
+  let copied: boolean
+  try {
+    copied = await renameIntoPlace(tmp, join(storageRoot, rel))
+  } catch (err) {
+    throw new SourceRegisterError(
+      'COPY_FAILED',
+      `${kind} 快照副本落位失败：${join(storageRoot, rel)}（${err instanceof Error ? err.message : String(err)}）`,
+    )
+  }
+  return {
+    kind,
+    sourcePath: src,
+    sha256,
+    byteSize: bytes,
+    storagePath: rel,
+    copied,
+    metadata: { sourceName: basename(src) },
+    durationMs: performance.now() - t0,
+  }
+}
+
+/** 媒体目录输入：扫描（每文件哈希）→ 清单哈希 → 缺副本时整树复制到 <sha>/ 目录 */
+async function ingestMediaDir(srcDir: string, storageRoot: string): Promise<Ingested> {
+  const t0 = performance.now()
+  let scan: MediaScan
+  try {
+    scan = await scanMediaDir(srcDir)
+  } catch (err) {
+    throw new SourceRegisterError(
+      'INPUT_READ_FAILED',
+      `media 目录扫描失败：${srcDir}（${err instanceof Error ? err.message : String(err)}）`,
+    )
+  }
+  // 清单哈希输入只含 path+sha256（bytes 只进 metadata，不进哈希——定义钉死）
+  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)
+  const rel = `snapshots/media_bundle/${sha256.slice(0, 2)}/${sha256}`
+  const dest = join(storageRoot, rel)
+  let copied = false
+  if (!existsSync(dest)) {
+    const tmpRoot = join(storageRoot, 'snapshots', 'media_bundle', '.tmp')
+    await mkdir(tmpRoot, { recursive: true })
+    const tmp = join(tmpRoot, randomUUID())
+    try {
+      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 }),
+        )
+      }
+      copied = await renameIntoPlace(tmp, dest)
+    } catch (err) {
+      await rm(tmp, { recursive: true, force: true })
+      throw new SourceRegisterError(
+        'COPY_FAILED',
+        `media_bundle 快照副本落位失败：${dest}（${err instanceof Error ? err.message : String(err)}）`,
+      )
+    }
+  }
+  return {
+    kind: 'media_bundle',
+    sourcePath: srcDir,
+    sha256,
+    byteSize: totalBytes,
+    storagePath: rel,
+    copied,
+    metadata: {
+      fileCount: scan.files.length,
+      totalBytes,
+      manifest: scan.files,
+      skippedSymlinks: scan.skipped,
+    },
+    durationMs: performance.now() - t0,
+  }
+}
+
+// ---- 路径安全 ----
+
+/** realpath（不存在则回退词法 resolve）——白名单根可能尚未创建（如首次运行无 storage/incoming） */
+async function realpathOrResolve(p: string): Promise<string> {
+  try {
+    return await realpath(p)
+  } catch {
+    return resolve(p)
+  }
+}
+
+/** 路径是否位于某根目录内（双方均 realpath 后比较；首段 .. 或绝对化即越界） */
+function isUnderRoot(target: string, root: string): boolean {
+  const rel = relative(root, target)
+  if (rel === '') return true
+  if (isAbsolute(rel)) return false
+  return rel.split(/[\\/]/)[0] !== '..'
+}
+
+async function validateInputPath(
+  inputPath: string,
+  expected: 'file' | 'directory',
+  allowedRoots: string[],
+  label: string,
+): Promise<string> {
+  const abs = resolve(inputPath)
+  let stat
+  try {
+    stat = await lstat(abs)
+  } catch (err) {
+    const code = (err as NodeJS.ErrnoException).code
+    if (code === 'ENOENT' || code === 'ENOTDIR') {
+      throw new SourceRegisterError('INPUT_NOT_FOUND', `${label} 输入不存在：${abs}`)
+    }
+    throw new SourceRegisterError(
+      'INPUT_READ_FAILED',
+      `${label} 输入无法访问（${code ?? 'unknown'}）：${abs}`,
+    )
+  }
+  // 输入本身是符号链接：以 realpath 后的目标做白名单与类型判定（防白名单内链接指向外界）
+  let real: string
+  if (stat.isSymbolicLink()) {
+    try {
+      real = await realpath(abs)
+    } catch {
+      // 断链（目标不存在）：realpath 抛 ENOENT——归入清晰的"输入不存在"，而非裸系统错误
+      throw new SourceRegisterError('INPUT_NOT_FOUND', `${label} 输入不存在（符号链接目标缺失）：${abs}`)
+    }
+  } else {
+    real = abs
+  }
+  let allowed = false
+  for (const root of allowedRoots) {
+    const realRoot = await realpathOrResolve(resolve(root))
+    if (isUnderRoot(real, realRoot)) {
+      allowed = true
+      break
+    }
+  }
+  if (!allowed) {
+    throw new SourceRegisterError(
+      'PATH_NOT_ALLOWED',
+      `${label} 输入路径越界：${abs}（不在允许目录：${allowedRoots.join(' ; ')}）`,
+    )
+  }
+  const realStat = await lstat(real)
+  if (expected === 'file' && !realStat.isFile()) {
+    throw new SourceRegisterError('INPUT_NOT_FILE', `${label} 输入不是文件：${abs}`)
+  }
+  if (expected === 'directory' && !realStat.isDirectory()) {
+    throw new SourceRegisterError('INPUT_NOT_DIR', `${label} 输入不是目录：${abs}`)
+  }
+  return real
+}
+
+// ---- DB 写入（drizzle） ----
+
+/** postgres unique violation（23505）——drizzle 会包成 DrizzleQueryError，真 code 在 cause */
+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> {
+  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
+  }
+}
+
+/**
+ * connection 判重：键 (site_id, name)。**表无此库级唯一约束**（Plan 1 schema 决定），
+ * 判重在应用层 select-then-insert——本服务是 source_connections 唯一写入方；出现第二个
+ * 写入方（REST 通道等）时必须先补库级唯一索引再谈并发。
+ * configuration 保持空对象：Connection 只描述来源类别（ADR-004），文件事实属于 snapshots。
+ */
+async function ensureConnection(
+  db: SourceDb,
+  siteId: string,
+  name: string,
+): Promise<{ id: string; created: boolean }> {
+  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
+  }
+}
+
+/** snapshot 幂等落库：UNIQUE(site_id, kind, sha256) 命中复用；不可变表只 INSERT */
+async function ensureSnapshot(
+  db: SourceDb,
+  args: { siteId: string; connectionId: string; ingested: Ingested; capturedAt: Date },
+): Promise<{ id: string; reused: boolean }> {
+  const { siteId, connectionId, ingested, capturedAt } = args
+  const found = 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 (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
+  }
+}
+
+// ---- runId 与报告 ----
+
+/** runId = 紧凑时间戳 + 6 位随机（同秒多次运行可区分） */
+function makeRunId(): string {
+  const ts = new Date()
+  const compact = `${ts.getFullYear()}${String(ts.getMonth() + 1).padStart(2, '0')}${String(
+    ts.getDate(),
+  ).padStart(2, '0')}T${String(ts.getHours()).padStart(2, '0')}${String(ts.getMinutes()).padStart(
+    2,
+    '0',
+  )}${String(ts.getSeconds()).padStart(2, '0')}`
+  return `${compact}-${Math.random().toString(36).slice(2, 8)}`
+}
+
+// ---- 主入口 ----
+
+/**
+ * 登记来源输入：校验（全部输入先过白名单/存在性，任何失败零落库零复制）→
+ * 逐输入摄取（copy+hash）→ site/connection/snapshot 落库 → 报告写盘。
+ * 返回 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 reportRoot = input.reportRoot ?? join(REPO_ROOT, 'artifacts', 'register')
+
+  // 阶段 0：全部输入先校验（fail fast，不留下半程副本）
+  let wxrReal: string | undefined
+  let sitemapReal: string | undefined
+  let mediaReal: string | undefined
+  if (input.wxrPath !== undefined) wxrReal = await validateInputPath(input.wxrPath, 'file', allowedRoots, 'wxr')
+  if (input.sitemapPath !== undefined)
+    sitemapReal = await validateInputPath(input.sitemapPath, 'file', allowedRoots, 'sitemap')
+  if (input.mediaDir !== undefined)
+    mediaReal = await validateInputPath(input.mediaDir, 'directory', allowedRoots, 'media')
+  if (wxrReal === undefined && sitemapReal === undefined && mediaReal === undefined) {
+    throw new SourceRegisterError('NO_INPUTS', '未提供任何输入（wxrPath / sitemapPath / mediaDir 至少其一）')
+  }
+
+  // 阶段 1：摄取（hash + 副本落位）——顺序固定 wxr → sitemap → media（报告输入清单稳定序）
+  const ingested: Ingested[] = []
+  if (wxrReal !== undefined) ingested.push(await ingestFile('wxr', wxrReal, storageRoot))
+  if (sitemapReal !== undefined) ingested.push(await ingestFile('sitemap', sitemapReal, storageRoot))
+  if (mediaReal !== undefined) ingested.push(await ingestMediaDir(mediaReal, storageRoot))
+
+  // 阶段 2：DB 写入
+  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,
+    })
+  }
+
+  // 阶段 3：报告（阶段 2 无 import_jobs，报告按 runId 归档——<reportRoot>/<runId>/reports/）
+  const runId = makeRunId()
+  const durationMs = performance.now() - startedAt
+  const report = {
+    runId,
+    generatedAt: new Date().toISOString(),
+    siteKey: input.siteKey,
+    siteId,
+    connectionId: connection.id,
+    connectionCreated: connection.created,
+    inputs: snapshots.map((s) => ({
+      kind: s.kind,
+      sourcePath: s.sourcePath,
+      sha256: s.sha256,
+      byteSize: s.byteSize,
+      storagePath: s.storagePath,
+      reused: s.reused,
+      copied: s.copied,
+      durationMs: s.durationMs,
+      metadata: s.metadata,
+    })),
+    totals: {
+      snapshots: snapshots.length,
+      reused: snapshots.filter((s) => s.reused).length,
+      byteSize: snapshots.reduce((sum, s) => sum + s.byteSize, 0),
+    },
+    durationMs,
+  }
+  const reportDir = join(reportRoot, runId, 'reports')
+  await mkdir(reportDir, { recursive: true })
+  const reportPath = join(reportDir, 'register.json')
+  await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8')
+
+  return {
+    runId,
+    siteKey: input.siteKey,
+    siteId,
+    connectionId: connection.id,
+    connectionCreated: connection.created,
+    snapshots,
+    reportPath,
+    durationMs,
+  }
+}
@@ -0,0 +1,25 @@
+// source 域 Nest 模块（Plan 2 Task 7）：注册 SourceRegisterService 供 standalone context
+// 使用（NestFactory.createApplicationContext(SourceModule)——本切片 API 不监听端口，ADR-002）。
+// 注意：本文件是仓库中第一个会执行 Nest 装饰器的模块文件——import 'reflect-metadata' 置于
+// 顶部（装饰器依赖 Reflect.defineMetadata；CLI/测试直接 import register.service.ts 纯函数
+// 文件，不经本模块，故无需提前加载 reflect-metadata）。
+import 'reflect-metadata'
+import { Injectable, Module } from '@nestjs/common'
+import { registerSource, type RegisterInput, type RegisterResult, type SourceDb } from './register.service.js'
+
+/**
+ * registerSource 的 Nest 注入壳：逻辑全在 register.service.ts 的独立函数里，
+ * 本类只提供 DI 语境下的调用面（standalone context / 后续 REST 通道复用同一实现）。
+ */
+@Injectable()
+export class SourceRegisterService {
+  register(db: SourceDb, input: RegisterInput): Promise<RegisterResult> {
+    return registerSource(db, input)
+  }
+}
+
+@Module({
+  providers: [SourceRegisterService],
+  exports: [SourceRegisterService],
+})
+export class SourceModule {}