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
|
// SQLite-backed state for the mailroom.
//
// The service keeps one in-memory `state` object and calls persist() after
// mutating it. This module keeps that model but makes each persist a single
// transaction that writes only what changed: every plain-object collection
// (names, offers, invoices, ...) is one row per entry, and everything else
// (lastPayIndex, the pending queue, the forwarder wallet) is one row per
// top-level key. WAL + synchronous=FULL, so a crash leaves either the old
// rows or the new ones — never a truncated file.
//
// When the database has no rows yet, a legacy JSON state file is imported
// (in one transaction) and left in place, so a rollback to the JSON-file
// build finds its data instead of starting empty.
import { Database } from 'bun:sqlite';
import { existsSync, mkdirSync, readFileSync } from 'node:fs';
import { dirname } from 'node:path';
// Top-level keys stored whole even though they are objects: the forwarder
// wallet's state is one unit and must never be written half-updated.
const WHOLE = new Set(['fwdArk']);
const SCALAR = ''; // collection name used for whole-value keys
const isMap = (k, v) => !WHOLE.has(k) && v !== null && typeof v === 'object' && !Array.isArray(v);
export function openStore(path, { legacyJson } = {}) {
mkdirSync(dirname(path), { recursive: true });
const db = new Database(path, { create: true, strict: true });
db.exec('PRAGMA journal_mode = WAL');
db.exec('PRAGMA synchronous = FULL');
db.exec('PRAGMA busy_timeout = 5000');
db.exec(`CREATE TABLE IF NOT EXISTS kv (
coll TEXT NOT NULL, key TEXT NOT NULL, val TEXT NOT NULL, updated INTEGER NOT NULL,
PRIMARY KEY (coll, key)) WITHOUT ROWID`);
// which top-level keys are maps, so an empty map survives a reload
db.exec('CREATE TABLE IF NOT EXISTS colls (coll TEXT PRIMARY KEY)');
const upsert = db.prepare(`INSERT INTO kv (coll, key, val, updated) VALUES (?, ?, ?, ?)
ON CONFLICT (coll, key) DO UPDATE SET val = excluded.val, updated = excluded.updated`);
const del = db.prepare('DELETE FROM kv WHERE coll = ? AND key = ?');
const addColl = db.prepare('INSERT OR IGNORE INTO colls (coll) VALUES (?)');
const dropColl = db.prepare('DELETE FROM colls WHERE coll = ?');
const dropRows = db.prepare('DELETE FROM kv WHERE coll = ?');
// last-written serialization of every row, to diff against
const written = new Map(); // `${coll}\0${key}` -> json
const maps = new Set();
function load() {
const state = {};
for (const { coll } of db.query('SELECT coll FROM colls').all()) { state[coll] = {}; maps.add(coll); }
for (const { coll, key, val } of db.query('SELECT coll, key, val FROM kv').all()) {
written.set(`${coll}\0${key}`, val);
const v = JSON.parse(val);
if (coll === SCALAR) state[key] = v;
else (state[coll] ||= {})[key] = v;
}
return state;
}
const persistTx = db.transaction((state) => {
const now = Date.now();
const seen = new Set();
for (const [k, v] of Object.entries(state)) {
if (v === undefined) continue;
if (isMap(k, v)) {
if (!maps.has(k)) {
// a key that used to be stored whole is now a map: clear the old row
if (written.has(`${SCALAR}\0${k}`)) { del.run(SCALAR, k); written.delete(`${SCALAR}\0${k}`); }
addColl.run(k); maps.add(k);
}
for (const [rk, rv] of Object.entries(v)) {
if (rv === undefined) continue;
const id = `${k}\0${rk}`, json = JSON.stringify(rv);
seen.add(id);
if (written.get(id) !== json) { upsert.run(k, rk, json, now); written.set(id, json); }
}
} else {
if (maps.has(k)) { dropRows.run(k); dropColl.run(k); maps.delete(k); }
const id = `${SCALAR}\0${k}`, json = JSON.stringify(v);
seen.add(id);
if (written.get(id) !== json) { upsert.run(SCALAR, k, json, now); written.set(id, json); }
}
}
for (const k of [...maps]) if (!(k in state)) { dropColl.run(k); maps.delete(k); }
for (const id of [...written.keys()]) {
if (seen.has(id)) continue;
const [coll, key] = id.split('\0');
del.run(coll, key); written.delete(id);
}
});
let state;
const empty = !db.query('SELECT 1 FROM kv LIMIT 1').get();
if (empty && legacyJson && existsSync(legacyJson)) {
// JSON.parse throws on a corrupt file: refuse to start rather than import nothing
state = JSON.parse(readFileSync(legacyJson, 'utf8'));
persistTx(state);
} else {
state = load();
}
state.names ||= {};
return {
state,
persist: () => persistTx(state),
// a consistent copy for backups, safe while the service runs
snapshot: (to) => db.exec(`VACUUM INTO '${to.replace(/'/g, "''")}'`),
close: () => db.close(),
};
}
|
