sync-company-to-graph-db.js
6.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
#!/usr/bin/env node
/**
* sync-company-to-graph-db.js — 把一批公司在 graph.sqlite 里的全部行，从一个库合并进另一个库
*
* 背景: 2026-08-28 把本地 mycompany / innovator-fde-trial 同步到生产时手搓过一次这个逻辑，
* 踩了两个坑，见 docs/done/production-deploy.md「跨环境同步公司数据」一节。这个脚本是那次
* 手搓版本的正式化，以后再有「把某几家公司搬到另一个 graph.sqlite」的需求，直接用它，
* 不要重新手写 SELECT * 的合并逻辑。
*
* 两个坑，这个脚本已经绕开：
* 1. 列序不能信。本地库和生产库的表结构字段名一样，但物理列序可能不同——同一张表，
* 有的库是老版本 CREATE TABLE 之后靠 ALTER TABLE 一路加列，有的库是某次重建/清空后
* 吃到了当时最新的 CREATE TABLE 字面量，两边列的物理顺序会对不上。SELECT * 位置对齐
* 的 INSERT 会静默把值写进错误的列（NOT NULL 约束能接住一部分，其余的没有任何报错）。
* 这里一律按列名显式对齐（用 PRAGMA table_info 在运行时取目标库的列名）。
* 2. graph_runs.parent_node_run_id ⇄ graph_run_nodes.run_id 是循环外键。先插 graph_runs
* （parent_node_run_id 强制置 NULL），再插 graph_run_nodes，最后回填 parent_node_run_id。
* 回填那条 UPDATE 的关联子查询**必须**给内外表都加别名——用裸表名的话 SQLite 会把子查询
* 里的列解析错关联对象，静默回填成 NULL，不报错、不报变更数异常（changes 照样是「成功」）。
*
* 用法:
* node scripts/sync-company-to-graph-db.js \
* --source data/graph.sqlite \
* --target /path/to/other/graph.sqlite \
* --companies mycompany,innovator-fde-trial \
* [--dry-run]
*
* --dry-run 整个合并跑在一个事务里，最后主动抛出回滚，不留任何行；控制台照常打印各表行数。
*
* 表顺序即依赖顺序（父表先于子表插入）：
* system_agents(公司私有的) → company_agents → org_graph_nodes → org_graph_edges
* → work_graphs → work_graph_nodes → work_graph_edges → graph_runs → graph_run_nodes
* → graph_run_loop_records → graph_files → graph_edge_contracts → goals → tasks
*
* 不做的事: 不搬 data/companies/{cid}/ 下的工作文件（memory / conversations / workspace）——
* 那部分是普通文件，直接 rsync 就行，不需要脚本。这个脚本只管 graph.sqlite 里的行。
*/
const Database = require('better-sqlite3');
function parseArgs(argv) {
const out = { dryRun: argv.includes('--dry-run') };
for (const flag of ['--source', '--target', '--companies']) {
const i = argv.indexOf(flag);
if (i !== -1) out[flag.slice(2)] = argv[i + 1];
}
return out;
}
const { source, target, companies, dryRun } = parseArgs(process.argv.slice(2));
if (!source || !target || !companies) {
console.error(
'usage: node scripts/sync-company-to-graph-db.js --source <db> --target <db> --companies id1,id2 [--dry-run]'
);
process.exit(1);
}
const COMPANIES = companies.split(',').map((s) => s.trim()).filter(Boolean);
const ph = COMPANIES.map(() => '?').join(',');
const db = new Database(target);
db.pragma('foreign_keys = ON');
db.exec(`ATTACH DATABASE '${source}' AS src`);
// 列名一律从目标库运行时读取（PRAGMA table_info），不写死顺序——见文件头坑 1。
const targetCols = (table) => db.prepare(`PRAGMA table_info(${table})`).all().map((c) => c.name);
const log = (...a) => console.log(dryRun ? '[dry-run]' : '[sync]', ...a);
function insert(table, mode, whereSql, colOverrides = {}) {
const cols = targetCols(table);
const selectExprs = cols.map((c) => (c in colOverrides ? `${colOverrides[c]} AS ${c}` : c));
const sql = `INSERT ${mode} INTO ${table} (${cols.join(', ')}) SELECT ${selectExprs.join(', ')} FROM src.${table} ${whereSql}`;
const info = db.prepare(sql).run(...COMPANIES);
log(`${table}: ${info.changes} rows`);
}
const tx = db.transaction(() => {
insert('system_agents', 'OR REPLACE', `WHERE owner_company_id IN (${ph})`);
insert('company_agents', 'OR IGNORE', `WHERE company_id IN (${ph})`);
insert('org_graph_nodes', 'OR IGNORE', `WHERE company_id IN (${ph})`);
insert('org_graph_edges', 'OR IGNORE', `WHERE company_id IN (${ph})`);
insert('work_graphs', 'OR IGNORE', `WHERE company_id IN (${ph})`);
insert('work_graph_nodes', 'OR IGNORE',
`WHERE work_graph_id IN (SELECT id FROM src.work_graphs WHERE company_id IN (${ph}))`);
insert('work_graph_edges', 'OR IGNORE',
`WHERE work_graph_id IN (SELECT id FROM src.work_graphs WHERE company_id IN (${ph}))`);
// 坑 2：先插 graph_runs，parent_node_run_id 强制 NULL，等 graph_run_nodes 落地后再回填。
insert('graph_runs', 'OR IGNORE', `WHERE company_id IN (${ph})`, { parent_node_run_id: 'NULL' });
insert('graph_run_nodes', 'OR IGNORE',
`WHERE run_id IN (SELECT id FROM src.graph_runs WHERE company_id IN (${ph}))`);
// 回填必须给内外表都加别名，否则子查询里的列会解析错关联对象，静默写成 NULL（坑 2 的后半）。
const backfill = db.prepare(`
UPDATE graph_runs AS tgt SET parent_node_run_id = (
SELECT s.parent_node_run_id FROM src.graph_runs AS s WHERE s.id = tgt.id
)
WHERE tgt.company_id IN (${ph})
`).run(...COMPANIES);
log(`graph_runs.parent_node_run_id backfilled: ${backfill.changes} rows touched`);
insert('graph_run_loop_records', 'OR IGNORE',
`WHERE run_id IN (SELECT id FROM src.graph_runs WHERE company_id IN (${ph}))`);
insert('graph_files', 'OR IGNORE', `WHERE company_id IN (${ph})`);
insert('graph_edge_contracts', 'OR IGNORE',
`WHERE edge_layer = 'org' AND edge_id IN (SELECT CAST(id AS TEXT) FROM src.org_graph_edges WHERE company_id IN (${ph}))`);
insert('graph_edge_contracts', 'OR IGNORE',
`WHERE edge_layer = 'work' AND edge_id IN (
SELECT id FROM src.work_graph_edges
WHERE work_graph_id IN (SELECT id FROM src.work_graphs WHERE company_id IN (${ph}))
)`);
insert('goals', 'OR IGNORE', `WHERE company_id IN (${ph})`);
insert('tasks', 'OR IGNORE', `WHERE company_id IN (${ph})`);
// FK 校验放在事务内部：同一连接能看见本事务未提交的写入，dry-run 也能校验到真实结果。
const fkErrors = db.pragma('foreign_key_check');
if (fkErrors.length) {
throw new Error('foreign_key_check failed: ' + JSON.stringify(fkErrors));
}
if (dryRun) throw new Error('__DRY_RUN_ROLLBACK__');
});
let failed = false;
try {
tx();
console.log('sync complete');
} catch (e) {
if (e.message === '__DRY_RUN_ROLLBACK__') {
log('rolled back (dry run) — no rows persisted, foreign_key_check passed');
console.log('dry run complete, no changes persisted');
} else {
console.error(e.message);
failed = true;
}
} finally {
db.exec('DETACH DATABASE src');
db.close();
}
if (failed) process.exit(1);