openvibers/openvibe.sdk Commits

Migration tools and lessons from the Events pass (G3).

OP OpenVibers committed 28 Sept 2026 · 999e889b7f · 5 files, +46 −12

gen-migration.js takes --from <module> (a SCHEMA outside server/db.js); the schema converter moves a trigger's WHEN
into the function body (PostgreSQL forbids subqueries there); g1db.py keeps a file header when openStore has no doc
comment; sqlfix.py qualifies upsert right-hand sides with the table's columns from migrations/, drops SQLite's
sqlite_% filter when it rewrites table lists, and reports DDL at boot (db.exec). docs/migrating-to-postgresql.md §7
gains the Events lessons: serialized worker passes, registering before the first await, functions passed as
arguments, no DDL at boot, test boots with their own database, redaction and vacuum.

Changes

ddiff --git a/docs/migrating-to-postgresql.md b/docs/migrating-to-postgresql.md
iindex 5f84444..1189ec2 100644
---- a/docs/migrating-to-postgresql.md
++++ b/docs/migrating-to-postgresql.md
@@@ -158,3 +158,10 @@ It changes nothing in the SQLite file.
- **Values.** `jsonb` columns come back as objects, so do not `JSON.parse` them.
- **Gauges.** A metrics `collect()` is synchronous. Read the database one scrape behind.
- **`npm run test:pg` finds races PGlite hides.** PGlite has one connection, so an unawaited write finished before the next read. On a pool it does not.
++- **Events (2026-09-28)** added these:
++ - **A loop that read and then claimed in one synchronous step** (a delivery worker's "select due rows, mark them busy") can now run twice at once, from a timer and a kick, and send the same row twice. Serialize the passes: a call during a pass waits for one more pass after it.
++ - **Register before the first `await`.** A connection counted against a cap, or joined to an in-process fan-out, after its catch-up queries lets the cap overrun and loses what was published meanwhile. Take the slot and join first; buffer live rows until the catch-up ends, then send those it did not cover.
++ - **A function passed as an argument** (`flush(insertBatch)`) is async now; the codemod does not see through parameters. Await it by hand.
++ - **No DDL at boot.** A module that ran `CREATE TABLE IF NOT EXISTS` on the serving handle fails on production roles (`permission denied for schema`) while PGlite allows it. The table goes in the migration; `sqlfix.py` reports `db.exec(`.
++ - **Test boots that call `start()` directly** fall back to the development PGlite directory and share rows across runs. Pass every boot a `createTestDb` handle.
++ - **Redaction.** PostgreSQL has no `secure_delete`: an overwritten row version stays until vacuum. Lower the table's `autovacuum_vacuum_scale_factor` and say so in the README.
ddiff --git a/tools/asyncify/g1db.py b/tools/asyncify/g1db.py
iindex c4a11b0..e03e9e6 100644
---- a/tools/asyncify/g1db.py
++++ b/tools/asyncify/g1db.py
@@@ -12,12 +12,14 @@ m = re.search(r"\nfunction openStore\(dbPath, \{ now = \(\) => Date\.now\(\) \}
if not m: sys.exit('openStore(dbPath, { now }) with a returned object literal not found: convert by hand')
props = [l for l in m.group(1).split('\n') if not re.match(r"\s+(db|now|tx: .*|close: .*),?$", l)]
doc = s.rfind('/**', 0, m.start() + 1)
--head = s[:doc] if doc != -1 and s[doc:m.start()].count('*/') == 1 else s[:m.start() + 1]
++end = s.find('*/', doc) if doc != -1 else -1
++# openStore's own doc comment only: the comment must end right before the function (whitespace between).
++head = s[:doc] if doc != -1 and end != -1 and not s[end + 2:m.start() + 1].strip() else s[:m.start() + 1]
tail = s[m.end():]
head = re.sub(r"const Database = require\('better-sqlite3'\);\n", "const { createDb } = require('openvibe-sdk/db');\n", head)
head = re.sub(r"\nconst SCHEMA = `[\s\S]*?`;\n", "\nconst MIGRATIONS = path.join(__dirname, '..', 'migrations');\nconst DEV_PGLITE = path.join(__dirname, '..', 'data', 'pglite');\n", head)
--head = re.sub(r"%s'?s? own SQLite database[:,] created on boot, idempotently\." % re.escape(Name),
-- f"{Name}' own PostgreSQL database (ADR-035, roadmap WS-X2): the schema is migrations/NNNN_*.sql, applied at boot.", head)
++head = re.sub(r"(%s'?s?) own SQLite database(?: \(WAL\))?[:,] created on boot, idempotently\." % re.escape(Name),
++ lambda x: x.group(1) + " own PostgreSQL database (ADR-035, roadmap WS-X2): the schema is migrations/NNNN_*.sql, applied at boot.", head)
body = f"""/**
* The serving handle (ADR-035): DATABASE_URL through PgBouncer; in development without it, an embedded PGlite database
* in data/pglite. Migrations run first, as the owner (DATABASE_DIRECT_URL), or on the embedded handle.
ddiff --git a/tools/asyncify/gen-migration.js b/tools/asyncify/gen-migration.js
iindex 25b7670..908beba 100644
---- a/tools/asyncify/gen-migration.js
++++ b/tools/asyncify/gen-migration.js
@@@ -7,6 +7,7 @@
*
* node gen-migration.js <Name> [--publishing revisions:blog_post,citations:blog_post,index-hooks:blog,seo:blog_entity,…]
* [--no-inbox] [--outbox-table t] [--inbox-table t] [--no-outbox] [--ensure a.js,b.js]
++ * [--from server/store.js] (the module holding SCHEMA; default server/db.js)
* → migrations/0001_initial.sql
*
* It lists every `rowid` the server code still orders by: those tables need a `seq bigint GENERATED ALWAYS AS
@@@ -26,11 +27,12 @@ const req = (m) => require(require.resolve(m, { paths: [process.cwd()] }));
// The schema as the SQLite release ran it: server/db.js at HEAD (the conversion edits the working copy). A literal with
// ${…} interpolations is evaluated by loading that file and reading its SCHEMA export.
--const src = execFileSync('git', ['show', 'HEAD:server/db.js'], { encoding: 'utf8' });
++const dbFile = (() => { const i = args.indexOf('--from'); return i >= 0 ? args[i + 1] : 'server/db.js'; })();
++const src = execFileSync('git', ['show', `HEAD:${dbFile}`], { encoding: 'utf8' });
const m = /const SCHEMA = `([\s\S]*?)`;/.exec(src);
--if (!m) { console.error('server/db.js (HEAD) has no const SCHEMA = `…`;'); process.exit(1); }
++if (!m) { console.error(`${dbFile} (HEAD) has no const SCHEMA = \`…\`;`); process.exit(1); }
if (m[1].includes('${')) {
-- const tmp = path.resolve('server', '.db-head-schema.js');
++ const tmp = path.resolve(path.dirname(dbFile), '.db-head-schema.js');
fs.writeFileSync(tmp, src);
try {
const schema = require(tmp).SCHEMA;
ddiff --git a/tools/asyncify/sqlfix.py b/tools/asyncify/sqlfix.py
iindex 4082738..4cbf509 100644
---- a/tools/asyncify/sqlfix.py
++++ b/tools/asyncify/sqlfix.py
@@@ -33,10 +33,23 @@ REPORT = [
(r'setImmediate|process\.nextTick', 'deferred work: after a commit → db.afterCommit'),
(r'createInbox\b', 'createInbox → createPgInbox (awaited; receipts table in the migration)'),
(r'\bPRAGMA\b|\.pragma\(', 'PRAGMA'),
++ (r'\b(db|this\.db)\.exec\(', 'DDL at boot on the serving handle: the serving role cannot create tables; the table belongs in a migration'),
++ (r'\b(SUM|AVG)\((?![^)]*\bCASE\b)[^)]*\)(?!\s*::)(?!, *0\)::)', 'SUM/AVG of a bigint column is numeric, returned as text: cast it (COALESCE(SUM(x), 0)::bigint) unless the column is double precision'),
(r'(@\w+|[^\w]\?) IS (NOT )?NULL(?!::)', 'a parameter tested with IS NULL may need a cast (@p::bigint IS NULL) when PostgreSQL cannot infer its type'),
(r'UNIQUE constraint|constraint failed|SQLITE_[A-Z]', "SQLite error text → err.code ('23505' unique, '23503' foreign key, '23514' check) with err.table/err.constraint/err.detail"),
]
changed = []
++# Table columns from the service's migrations, so upsert right-hand sides can be qualified even for columns the INSERT
++# does not list (a counter bumped in place, a timestamp cleared).
++TABLE_COLS = {}
++import glob as _glob
++for mf in sorted(_glob.glob('migrations/*.sql')):
++ ddl = open(mf).read()
++ for tm in re.finditer(r'CREATE TABLE (?:IF NOT EXISTS )?(\w+) \(([\s\S]*?)\n\);', ddl):
++ cols = [c.group(1).strip('"') for c in re.finditer(r'^\s*("?\w+"?)\s+(?:text|bigint|integer|double|boolean|bytea|jsonb|timestamptz|numeric|real)', tm.group(2), re.M)]
++ TABLE_COLS.setdefault(tm.group(1), set()).update(cols)
++ for am in re.finditer(r'ALTER TABLE (\w+) ADD COLUMN (\w+)', ddl):
++ TABLE_COLS.setdefault(am.group(1), set()).add(am.group(2))
for p in files:
s = open(p).read(); o = s
while 'INSERT OR IGNORE INTO' in s:
@@@ -48,10 +61,15 @@ for p in files:
s = re.sub(r'(?<![IN\w])LIKE (?=[@?:$\'(]|[a-z])', 'ILIKE ', s)
s = s.replace('json_each(?)', 'jsonb_array_elements_text(?::jsonb)')
s = re.sub(r'\bIFNULL\(', 'COALESCE(', s)
++ # SQLite's null-safe x IS ? / x IS NOT ? (a parameter) are IS [NOT] DISTINCT FROM in PostgreSQL.
++ s = re.sub(r'([\w.]+) IS NOT (\?|@\w+)', r'\1 IS DISTINCT FROM \2', s)
++ s = re.sub(r'([\w.]+) IS (\?|@\w+)', r'\1 IS NOT DISTINCT FROM \2', s)
# assert.throws(() => db.prepare(…).run(…), re): the statement rejects now.
s = re.sub(r"assert\.throws\(\(\) => ((?:t\.)?[\w.]*db\.prepare\((?:[^()]|\([^()]*\))*\)\.(?:run|get|all)\((?:[^()]|\([^()]*\))*\)), ", r"await assert.rejects(\1, ", s)
# A jsonb column (the SDK outbox's envelope) comes back as an object.
s = re.sub(r"JSON\.parse\((\w+)\.envelope\)", r"(typeof \1.envelope === 'string' ? JSON.parse(\1.envelope) : \1.envelope)", s)
++ # const f = db.transaction((a, b) => { … }) (called later as f(a, b)) → const f = (a, b) => db.tx(async () => { … })
++ s = re.sub(r'const (\w+) = (\w+(?:\.\w+)?)\.transaction\(\(([^()]*)\) => \{', r'const \1 = (\3) => \2.tx(async () => {', s)
# db.transaction(fn)() → db.tx(fn) (balanced parentheses; the codemod then awaits it and makes fn async).
out, i = [], 0
while True:
@@@ -78,6 +96,7 @@ for p in files:
# PostgreSQL; qualify it with the table (columns of the INSERT list only; excluded.x and the targets stay).
def qualify(m):
table, cols, rest = m.group(1), [c.strip() for c in m.group(2).split(',')], m.group(3)
++ cols = sorted(set(cols) | TABLE_COLS.get(table, set()), key=len, reverse=True)
k = rest.index('DO UPDATE SET') + len('DO UPDATE SET')
head, sets = rest[:k], rest[k:]
def fix_rhs(a):
@@@ -92,6 +111,7 @@ for p in files:
# openvibe-publishing 1.0: the index sequencer stamps through a handle (the ambient transaction joins through store.db).
handle = 'store.db' if 'store.db' in s else 'db'
s = re.sub(r'((?:store\.)?sequencer\.stamp\()(?!store\.db|db,|t,)', lambda m: m.group(1) + handle + ', ', s)
++ s = re.sub(r"(SELECT name FROM sqlite_master WHERE type (?:= 'table'|IN \('table', ?'view'\))) AND name NOT (?:I)?LIKE 'sqlite_%'", r"\1", s)
s = re.sub(r"SELECT name FROM sqlite_master WHERE type (?:= 'table'|IN \('table', ?'view'\))",
"SELECT table_name AS name FROM information_schema.tables WHERE table_schema = current_schema()", s)
s = re.sub(r"SELECT name FROM pragma_table_info\('(\w+)'\)",
ddiff --git a/tools/asyncify/sqlite-schema-to-pg.js b/tools/asyncify/sqlite-schema-to-pg.js
iindex d89b27e..8444520 100644
---- a/tools/asyncify/sqlite-schema-to-pg.js
++++ b/tools/asyncify/sqlite-schema-to-pg.js
@@@ -15,12 +15,15 @@
*/
function toPg(sql) {
// The one trigger shape services use: a row that must never change (BEFORE UPDATE/DELETE … RAISE(ABORT, msg)).
-- // With an optional WHEN condition (x IS NOT y between OLD and NEW becomes IS DISTINCT FROM).
-- sql = sql.replace(/CREATE TRIGGER(?: IF NOT EXISTS)? (\w+) BEFORE (UPDATE(?: OF [\w\s,]+?)?|DELETE) ON (\w+)(?:\s+WHEN ([\s\S]+?))?\s+BEGIN SELECT RAISE\(ABORT, '([^']*)'\); END;/g,
-- (m, name, op, table, when, msg) => `CREATE FUNCTION ${name}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION '${msg}'; END $$;\n`
-- + `CREATE TRIGGER ${name} BEFORE ${op.replace(/\s+/g, ' ')} ON ${table} FOR EACH ROW`
-- + (when ? ` WHEN (${when.replace(/\s+/g, ' ').replace(/ IS NOT (OLD|NEW)\./g, ' IS DISTINCT FROM $1.')})` : '')
-- + ` EXECUTE FUNCTION ${name}();`);
++ // The condition (SQLite's WHEN, which may hold a subquery PostgreSQL does not allow in a trigger's WHEN) goes inside
++ // the function: IF cond THEN RAISE; the row passes otherwise (RETURN NEW for an update, OLD for a delete).
++ sql = sql.replace(/CREATE TRIGGER(?: IF NOT EXISTS)?\s+(\w+)\s+BEFORE\s+(UPDATE(?:\s+OF\s+[\w\s,]+?)?|DELETE)\s+ON\s+(\w+)(?:\s+WHEN\s+([\s\S]+?))?\s+BEGIN SELECT RAISE\(ABORT, '([^']*)'\); END;/g,
++ (m, name, op, table, when, msg) => {
++ const cond = when ? when.replace(/\s+/g, ' ').replace(/ IS NOT (OLD|NEW)\./g, ' IS DISTINCT FROM $1.') : 'TRUE';
++ const row = /^DELETE/.test(op) ? 'OLD' : 'NEW';
++ return `CREATE FUNCTION ${name}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF ${cond} THEN RAISE EXCEPTION '${msg}'; END IF; RETURN ${row}; END $$;\n`
++ + `CREATE TRIGGER ${name} BEFORE ${op.replace(/\s+/g, ' ')} ON ${table} FOR EACH ROW EXECUTE FUNCTION ${name}();`;
++ });
if (sql.split(';').some((st) => /CREATE\s+TRIGGER/i.test(st) && !/FOR EACH ROW[\s\S]*EXECUTE FUNCTION/.test(st))) {
throw new Error('sqlite-schema-to-pg: a trigger other than the immutable-row shape needs writing as PL/pgSQL by hand');
}