feat(db): 0001 来源域——connection/snapshot/job/record + 快照不可变触发器

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-20 19:27:31 +08:00
parent 5da883cd40
commit 32ceba58f0
5 changed files with 1002 additions and 2 deletions
@@ -0,0 +1,106 @@
+// 来源域集成测试：快照不可变触发器（UPDATE/DELETE 拒绝）+ 三处唯一约束在真 PG 上生效。
+// 生命周期纪律：随机后缀插入 + finally 反序清理，可重复运行不残留脏数据。
+// 注意：source_snapshots 不可变触发器连 DELETE 也拒——清理时先 DISABLE 该触发器再删行后 ENABLE
+// （不可变表的生产语义是永不删；测试清场是唯一例外，走显式 DDL 开关且删除限定本站点行）。
+import { describe, it, expect } from 'vitest'
+import { getDb } from '../../apps/api/src/db/client.js'
+
+type Db = ReturnType<typeof getDb>
+type Fixture = { siteId: string; connectionId: string; snapshotId: string; sha256: string; idempotencyKey: string }
+
+// 建链（FK 依赖顺序）：site → source_connection → source_snapshot → import_job
+async function setupSourceFixture(db: Db): Promise<Fixture> {
+  const rand = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`
+  const site = (await db.execute(`insert into sites (id, key, name, default_locale, production_origin)
+    values (gen_random_uuid(), 'src-${rand}', 'SrcFixture', 'en', 'https://src.example') returning id`)) as any
+  const siteId = site[0].id
+  const conn = (await db.execute(`insert into source_connections (id, site_id, kind, name, configuration)
+    values (gen_random_uuid(), '${siteId}', 'wxr', 'conn-${rand}', '{}') returning id`)) as any
+  const connectionId = conn[0].id
+  const snap = (await db.execute(`insert into source_snapshots (id, site_id, source_connection_id, kind, storage_path, sha256, byte_size, adapter_version, captured_at, metadata)
+    values (gen_random_uuid(), '${siteId}', '${connectionId}', 'wxr', 'wxr://${rand}.xml', 'sha256-${rand}', 123, 'wxr@1', now(), '{}') returning id, sha256`)) as any
+  const job = (await db.execute(`insert into import_jobs (id, site_id, source_snapshot_id, status, mapping_version, idempotency_key, attempt)
+    values (gen_random_uuid(), '${siteId}', '${snap[0].id}', 'queued', 'mapping@1', 'idem-${rand}', 0) returning idempotency_key`)) as any
+  return { siteId, connectionId, snapshotId: snap[0].id, sha256: snap[0].sha256, idempotencyKey: job[0].idempotency_key }
+}
+
+// 反序清理：子表常规删除；快照表临时 DISABLE 不可变触发器；最后删 site。
+async function cleanupSourceFixture(db: Db, f: Fixture) {
+  await db.execute(`delete from source_records where site_id = '${f.siteId}'`)
+  await db.execute(`delete from import_jobs where site_id = '${f.siteId}'`)
+  await db.execute(`alter table source_snapshots disable trigger source_snapshots_immutable`)
+  try {
+    await db.execute(`delete from source_snapshots where site_id = '${f.siteId}'`)
+  } finally {
+    await db.execute(`alter table source_snapshots enable trigger source_snapshots_immutable`)
+  }
+  await db.execute(`delete from source_connections where site_id = '${f.siteId}'`)
+  await db.execute(`delete from sites where id = '${f.siteId}'`)
+}
+
+async function withSourceFixture(fn: (db: Db, f: Fixture) => Promise<void>) {
+  const db = getDb()
+  const f = await setupSourceFixture(db)
+  try {
+    await fn(db, f)
+  } finally {
+    await cleanupSourceFixture(db, f)
+    await (db as any).$client.end()
+  }
+}
+
+// drizzle 会把 PostgresError 包成 DrizzleQueryError（message 只有 "Failed query: ..."），
+// 服务端原始错误文本在 cause 里——断言触发器/约束身份时必须看 cause。
+async function rejectionMessage(db: Db, query: string): Promise<string> {
+  try {
+    await db.execute(query)
+  } catch (e: any) {
+    return String(e?.cause?.message ?? e?.message ?? e)
+  }
+  throw new Error(`expected query to be rejected: ${query}`)
+}
+
+describe('来源域约束', () => {
+  it('source_snapshots 不可变：UPDATE 与 DELETE 均被触发器拒绝', async () => {
+    await withSourceFixture(async (db, f) => {
+      expect(await rejectionMessage(db, `update source_snapshots set sha256 = 'tampered' where id = '${f.snapshotId}'`))
+        .toMatch(/immutable table source_snapshots/)
+      expect(await rejectionMessage(db, `delete from source_snapshots where id = '${f.snapshotId}'`))
+        .toMatch(/immutable table source_snapshots/)
+    })
+  })
+
+  it('source_snapshots UNIQUE(site_id, kind, sha256)：同三元组重复插入拒绝，不同 sha256 放行', async () => {
+    await withSourceFixture(async (db, f) => {
+      await expect(db.execute(`insert into source_snapshots (id, site_id, source_connection_id, kind, storage_path, sha256, byte_size, adapter_version, captured_at, metadata)
+        values (gen_random_uuid(), '${f.siteId}', '${f.connectionId}', 'wxr', 'wxr://dup.xml', '${f.sha256}', 456, 'wxr@1', now(), '{}')`))
+        .rejects.toThrow()
+      // 对照组：同 site+kind 但 sha256 不同 → 允许（证明约束打在三元组上，由 finally 按 site 清理）
+      await db.execute(`insert into source_snapshots (id, site_id, source_connection_id, kind, storage_path, sha256, byte_size, adapter_version, captured_at, metadata)
+        values (gen_random_uuid(), '${f.siteId}', '${f.connectionId}', 'wxr', 'wxr://other.xml', 'sha256-other-${f.sha256}', 789, 'wxr@1', now(), '{}')`)
+    })
+  })
+
+  it('source_records 稳定身份键 UNIQUE(site_id, source_connection_id, source_type, source_id, locale, source_snapshot_id)：同键重复拒绝，换 snapshot 放行', async () => {
+    await withSourceFixture(async (db, f) => {
+      const insertRecord = (snapshotId: string, rawLocation: string) =>
+        db.execute(`insert into source_records (id, site_id, source_connection_id, source_snapshot_id, import_job_id, source_type, source_id, locale, raw_location, parse_status)
+          values (gen_random_uuid(), '${f.siteId}', '${f.connectionId}', '${snapshotId}', (select id from import_jobs where site_id = '${f.siteId}' limit 1), 'post', 'source-1', 'en', '${rawLocation}', 'parsed')`)
+      await insertRecord(f.snapshotId, 'raw://1')
+      // 同稳定身份键 + 同 snapshot：第二次插入拒绝
+      await expect(insertRecord(f.snapshotId, 'raw://1-dup')).rejects.toThrow()
+      // 对照组：同稳定身份键、不同 snapshot（跨快照再导入同一来源项）→ 允许，由 finally 按 site 清理
+      const snap2 = (await db.execute(`insert into source_snapshots (id, site_id, source_connection_id, kind, storage_path, sha256, byte_size, adapter_version, captured_at, metadata)
+        values (gen_random_uuid(), '${f.siteId}', '${f.connectionId}', 'wxr', 'wxr://v2.xml', 'sha256-v2-${f.sha256}', 999, 'wxr@1', now(), '{}') returning id`)) as any
+      await insertRecord(snap2[0].id, 'raw://1-v2')
+    })
+  })
+
+  it('import_jobs.idempotency_key 唯一：重复拒绝', async () => {
+    await withSourceFixture(async (db, f) => {
+      await expect(db.execute(`insert into import_jobs (id, site_id, source_snapshot_id, status, mapping_version, idempotency_key, attempt)
+        values (gen_random_uuid(), '${f.siteId}', '${f.snapshotId}', 'queued', 'mapping@1', '${f.idempotencyKey}', 0)`))
+        .rejects.toThrow()
+    })
+  })
+})