/** * 台账存储 adapter —— node-postgres 直写(编码规范 §1 / ADR-002)。 * * 运行时 DDL/DML 由声明生成,ORM 静态 schema 不适用;`ledger` schema 的**唯一写者**就是本文件。 * 标识符全部来自 @contracts 的 identifier 正则(小写字母开头、字母数字下划线), * 落库前再过一道 `quoteIdent` —— 拼接 SQL 的地方只有这里。 */ import type { Pool, PoolClient } from 'pg'; import type { LedgerKey, LedgerQuery, LedgerRow } from '@contracts'; import { LEDGER_SCHEMA } from '@shared/db.ts'; import type { DraftStore, LedgerStore, ObjectRef } from '../../port.ts'; import type { MaterializedObject, MergedField } from '../domain/schema-merge.ts'; import type { DraftEntry } from '../domain/task-view.ts'; import { keyOf } from '../domain/task-view.ts'; const IDENT = /^[a-z][a-z0-9_]*$/; function quoteIdent(name: string): string { if (!IDENT.test(name)) throw new Error(`非法标识符:${name}`); return `"${name}"`; } /** 物理表名:<域>__<对象>;域与对象都过 IDENT,拼接安全 */ export function tableNameOf(ref: ObjectRef): string { if (!IDENT.test(ref.domain) || !IDENT.test(ref.object)) { throw new Error(`非法台账地址:${ref.domain}.${ref.object}`); } return `${ref.domain}__${ref.object}`; } const PG_TYPE: Readonly> = { text: 'text', number: 'double precision', integer: 'bigint', boolean: 'boolean', date: 'date', timestamp: 'timestamptz', json: 'jsonb', }; const FROM_PG: Readonly> = { text: 'text', 'character varying': 'text', 'double precision': 'number', numeric: 'number', bigint: 'integer', integer: 'integer', boolean: 'boolean', date: 'date', 'timestamp with time zone': 'timestamp', 'timestamp without time zone': 'timestamp', jsonb: 'json', json: 'json', }; type Queryable = Pick; export function createLedgerStore(pool: Pool): LedgerStore { return buildStore(pool, pool); } function buildStore(runner: Queryable, pool: Pool): LedgerStore { const store: LedgerStore = { async describe(ref) { const table = tableNameOf(ref); // 列注释即人读名:与列定义同一次查询取回,不多跑一趟 const cols = await runner.query<{ column_name: string; data_type: string; label: string | null }>( `select c.column_name, c.data_type, col_description(($1 || '.' || quote_ident($2))::regclass, c.ordinal_position) as label from information_schema.columns c where c.table_schema = $1 and c.table_name = $2 order by c.ordinal_position`, [LEDGER_SCHEMA, table], ); if (cols.rows.length === 0) return undefined; const tableLabel = await runner.query<{ label: string | null }>( `select obj_description(($1 || '.' || quote_ident($2))::regclass) as label`, [LEDGER_SCHEMA, table], ); const pk = await runner.query<{ attname: string }>( `select a.attname from pg_index i join pg_attribute a on a.attrelid = i.indrelid and a.attnum = any(i.indkey) where i.indrelid = $1::regclass and i.indisprimary`, [`${LEDGER_SCHEMA}.${quoteIdent(table)}`], ); const label = tableLabel.rows[0]?.label ?? undefined; const materialized: MaterializedObject = { domain: ref.domain, object: ref.object, ...(label === null || label === undefined ? {} : { label }), columns: cols.rows.map((c) => ({ name: c.column_name, type: FROM_PG[c.data_type] ?? 'text', ...(c.label === null ? {} : { label: c.label }), })), primaryKey: pk.rows.map((p) => p.attname), }; return materialized; }, /* * 记下对象性质 —— **每次应用声明都写**,不只建表那一次。 * 挂在 create-table 分支上是错的(踩过):表早就建好了,那条分支不走, * meta 永远是空的,治理台上全部表都成了「无人声明」。 */ async setObjectMeta(ref, kind, caseKey) { await runner.query( `insert into ledger.__object_meta (domain, object, kind, case_key) values ($1,$2,$3,$4) on conflict (domain, object) do update set kind = excluded.kind, case_key = excluded.case_key`, [ref.domain, ref.object, kind, caseKey ?? null], ); }, async listObjects() { // 表注释即人读名 —— 与表名同一次查询取回,治理台不必逐个 describe 才知道叫什么 const res = await runner.query<{ table_name: string; label: string | null; kind: string | null; case_key: string | null; }>( /* * 性质与人读名**同一次查询取回**:治理台要按性质分区,而性质写在活动清单里 —— * 若让路由每次去把清单捞回来重建,那是 12 个活动 × 全版本清单 ≈ 1.5MB 的 JSON * 解析(实测热连接 73ms、并发八次 350ms,慢请求还占住浏览器连接连累别的页)。 * 物化的东西记在物化侧:表建出来那一刻性质就定了。 */ `select t.table_name, obj_description(($1 || '.' || quote_ident(t.table_name))::regclass) as label, m.kind, m.case_key from information_schema.tables t left join ledger.__object_meta m on m.domain || '__' || m.object = t.table_name where t.table_schema = $1 and t.table_name not like '\\_\\_%' order by t.table_name`, [LEDGER_SCHEMA], ); return res.rows.flatMap((r) => { const idx = r.table_name.indexOf('__'); if (idx <= 0) return []; return [ { domain: r.table_name.slice(0, idx), object: r.table_name.slice(idx + 2), ...(r.label === null ? {} : { label: r.label }), ...(r.kind === null ? {} : { kind: r.kind }), ...(r.case_key === null ? {} : { caseKey: r.case_key }), }, ]; }); }, /** 主数据清单 —— 放行写拿它挡试跑单;性质在 `__object_meta`,一次 where 取回 */ async masterRefs() { const res = await runner.query<{ domain: string; object: string }>( `select domain, object from ledger.__object_meta where kind = 'master'`, ); return res.rows.map((r) => ({ domain: r.domain, object: r.object })); }, async applyChanges(ref, changes) { const table = `${LEDGER_SCHEMA}.${quoteIdent(tableNameOf(ref))}`; /* * 注释即人读名;没有 label 就不写,读回来是 null,呈现自然退回键名。 * `comment on` 是**工具语句**,不吃 $1 占位符 —— 只能拼字面量,故自己转义单引号。 */ const comment = async (target: string, label: string | undefined) => { if (label === undefined || label.trim() === '') return; await runner.query(`comment on ${target} is '${label.replaceAll("'", "''")}'`); }; for (const change of changes) { switch (change.kind) { case 'create-table': { const cols = change.object.fields.map((f) => `${quoteIdent(f.name)} ${PG_TYPE[f.type]}`); const pk = change.object.primaryKey.length > 0 ? `, primary key (${change.object.primaryKey.map(quoteIdent).join(', ')})` : ''; await runner.query(`create table if not exists ${table} (${cols.join(', ')}${pk})`); /* * 人读名落**表/列注释** —— 声明里带的 label 不能在建表这一步丢掉: * 丢了,治理台就只剩 `project__intake_record` 这种键给人看。 * 用注释而不是另建元数据表:名字与表同生共死,不会对不上。 */ await comment(`table ${table}`, change.object.label); for (const f of change.object.fields) { await comment(`column ${table}.${quoteIdent(f.name)}`, f.label); } break; } case 'add-column': await runner.query( `alter table ${table} add column if not exists ${quoteIdent(change.field.name)} ${PG_TYPE[change.field.type]}`, ); await comment(`column ${table}.${quoteIdent(change.field.name)}`, change.field.label); break; case 'widen-column': await runner.query( `alter table ${table} alter column ${quoteIdent(change.field.name)} type ${PG_TYPE[change.field.type]} ` + `using ${quoteIdent(change.field.name)}::${PG_TYPE[change.field.type]}`, ); break; } } }, async select(query: LedgerQuery) { const table = `${LEDGER_SCHEMA}.${quoteIdent(tableNameOf(query.address))}`; const params: unknown[] = []; const clauses = Object.entries(query.where).map(([field, value]) => { params.push(value); return `${quoteIdent(field)} = $${params.length}`; }); const where = clauses.length > 0 ? ` where ${clauses.join(' and ')}` : ''; const order = query.orderBy.length > 0 ? ` order by ${query.orderBy.map((o) => `${quoteIdent(o.field)} ${o.dir === 'desc' ? 'desc' : 'asc'}`).join(', ')}` : ''; const countRes = await runner.query<{ n: string }>(`select count(*)::text as n from ${table}${where}`, params); params.push(query.limit, query.offset); const rows = await runner.query( `select * from ${table}${where}${order} limit $${params.length - 1} offset $${params.length}`, params, ); return { rows: rows.rows, total: Number(countRes.rows[0]?.n ?? '0') }; }, async selectByKeys(ref, keys) { if (keys.length === 0) return []; const table = `${LEDGER_SCHEMA}.${quoteIdent(tableNameOf(ref))}`; const out: LedgerRow[] = []; for (const key of keys) { const params: unknown[] = []; const clauses = Object.entries(key).map(([field, value]) => { params.push(value); return `${quoteIdent(field)} = $${params.length}`; }); if (clauses.length === 0) continue; const res = await runner.query( `select * from ${table} where ${clauses.join(' and ')} limit 1`, params, ); out.push(...res.rows); } return out; }, async upsert(ref, key, values) { const table = `${LEDGER_SCHEMA}.${quoteIdent(tableNameOf(ref))}`; const merged: LedgerRow = { ...key, ...values }; const fields = Object.keys(merged); if (fields.length === 0) return; const params = fields.map((f) => merged[f]); const placeholders = fields.map((_, i) => `$${i + 1}`); const keyFields = Object.keys(key); const updatable = fields.filter((f) => !keyFields.includes(f)); const onConflict = keyFields.length === 0 ? '' : updatable.length === 0 ? ` on conflict (${keyFields.map(quoteIdent).join(', ')}) do nothing` : ` on conflict (${keyFields.map(quoteIdent).join(', ')}) do update set ` + updatable.map((f) => `${quoteIdent(f)} = excluded.${quoteIdent(f)}`).join(', '); await runner.query( `insert into ${table} (${fields.map(quoteIdent).join(', ')}) values (${placeholders.join(', ')})${onConflict}`, params, ); }, async deleteRow(ref, key) { const table = `${LEDGER_SCHEMA}.${quoteIdent(tableNameOf(ref))}`; const params: unknown[] = []; const clauses = Object.entries(key).map(([field, value]) => { params.push(value); return `${quoteIdent(field)} = $${params.length}`; }); if (clauses.length === 0) return; await runner.query(`delete from ${table} where ${clauses.join(' and ')}`, params); }, async transaction(fn) { const client = await pool.connect(); try { await client.query('begin'); const scoped = buildStore(client, pool); const result = await fn(scoped, createDraftStore(client)); await client.query('commit'); return result; } catch (error) { await client.query('rollback'); throw error; } finally { client.release(); } }, }; return store; } /* ── 任务级草稿层:ledger.__draft ─────────────────────────────────────── */ export const DRAFT_TABLE = `${LEDGER_SCHEMA}.__draft`; export const DRAFT_DDL = ` create table if not exists ${DRAFT_TABLE} ( task_id text not null, domain text not null, object text not null, key_hash text not null, key jsonb not null, patch jsonb not null, baseline jsonb not null, tombstone boolean not null default false, updated_at timestamptz not null default now(), primary key (task_id, domain, object, key_hash) ); -- 已落地记号(2026-08-27,数据工单确认即入库):非空 = 这一行已在一道更早的门上并进正式表, -- 放行时不再写它。加列走 if not exists —— 这张表由启动时的 DDL 自建,不走迁移。 alter table ${DRAFT_TABLE} add column if not exists landed_at timestamptz;`; type DraftRow = { task_id: string; domain: string; object: string; key: Record; patch: LedgerRow; baseline: LedgerRow; tombstone: boolean; landed_at: Date | string | null; }; function toEntry(row: DraftRow): DraftEntry & ObjectRef { return { domain: row.domain, object: row.object, key: row.key, patch: row.patch, baseline: row.baseline, tombstone: row.tombstone, landed: row.landed_at !== null && row.landed_at !== undefined, }; } export function createDraftStore(runner: Queryable): DraftStore { return { async list(taskId) { const res = await runner.query( `select * from ${DRAFT_TABLE} where task_id = $1 order by updated_at`, [taskId], ); return res.rows.map(toEntry); }, async listForObject(taskId, ref) { const res = await runner.query( `select * from ${DRAFT_TABLE} where task_id = $1 and domain = $2 and object = $3 order by updated_at`, [taskId, ref.domain, ref.object], ); return res.rows.map(toEntry); }, async put(taskId, ref, entry) { await runner.query( `insert into ${DRAFT_TABLE} (task_id, domain, object, key_hash, key, patch, baseline, tombstone) values ($1,$2,$3,$4,$5,$6,$7,$8) on conflict (task_id, domain, object, key_hash) do update set patch = ${DRAFT_TABLE}.patch || excluded.patch, -- 基线按**字段**首次写入时捕获:右操作数胜,已有基线保持不变、 -- 本次新写的字段补上它们各自的基线。若只在建行时捕获一次, -- 后续步新写的字段就没有基线,冲突检查会把它们一律当成「他单改过」。 baseline = excluded.baseline || ${DRAFT_TABLE}.baseline, tombstone = excluded.tombstone, -- 再写一笔就不再是「已落地」:新的这一笔要等它自己的门(放行) landed_at = null, updated_at = now()`, [ taskId, ref.domain, ref.object, keyOf(entry.key as LedgerKey), JSON.stringify(entry.key), JSON.stringify(entry.patch), JSON.stringify(entry.baseline), entry.tombstone, ], ); }, /** * 撤掉本单草稿层的**一行** —— 写错了得有退路。 * * 只删 `task_id` 那一份:正式版一个字都动不了。草稿层本来就是这一单的过程事实, * 过程里写错一行是常事;没有退路的后果实测过 —— * 客户说本轮跳过的那颗料被写进了有效清单,条目数报 9 而表里躺着 10, * 判据 `record_rows = line_total` 正确地拦住,而模型**改不动**: * 唯一能改的是那个数(改了 skipped_total 就不对),于是 13 次拒回、23 拍、234k token。 */ async drop(taskId, ref, key) { await runner.query( `delete from ${DRAFT_TABLE} where task_id = $1 and domain = $2 and object = $3 and key_hash = $4`, [taskId, ref.domain, ref.object, keyOf(key)], ); }, async markLanded(taskId, ref, key, baseline) { await runner.query( `update ${DRAFT_TABLE} set landed_at = now(), baseline = $5 where task_id = $1 and domain = $2 and object = $3 and key_hash = $4`, [taskId, ref.domain, ref.object, keyOf(key), JSON.stringify(baseline)], ); }, async clear(taskId) { await runner.query(`delete from ${DRAFT_TABLE} where task_id = $1`, [taskId]); }, }; }