openvibers/openvibe.sdk Commits

tools/asyncify after Deals, Coupons, Trade and Reviews: gen-migration.js (the first migration from server/db.js at HEAD, evaluated when the SCHEMA has interpolations, plus the Publishing and SDK tables under any names), g1db.py (server/db.js to openDb/createStore/openStore), g1docs.py (README and STATUS.json), sqlfix.py (the SQL rewrites that are safe by text: INSERT OR IGNORE, NOCASE, ILIKE, json_each, IFNULL, sqlite_master, qualified upsert targets, sequencer.stamp's handle; and CHECK lines for the rest). asyncify.js wraps .map() callbacks that become async in Promise.all, recursively outwards, and reports unawaited .map(async …). sqlite-schema-to-pg converts immutable-row triggers (UPDATE OF too) to PL/pgSQL and NOCASE indexes to lower(). docs/migrating-to-postgresql.md §7 gains the four services' lessons. Not part of the published package.

OP OpenVibers committed 28 Sept 2026 · 580f50fcaa · 9 files, +377 −20


Changes

ddiff --git a/docs/migrating-to-postgresql.md b/docs/migrating-to-postgresql.md
iindex 925e84b..79641c6 100644
---- a/docs/migrating-to-postgresql.md
++++ b/docs/migrating-to-postgresql.md
@@@ -148,3 +148,11 @@ It changes nothing in the SQLite file.
- **Work after the commit.** A `setImmediate` after a synchronous better-sqlite3 transaction ran after the commit. Around an async transaction it does not. Use `db.afterCommit(fn)` (SDK ≥ 0.20).
- **The inbox.** `createInbox` (better-sqlite3) becomes `createPgInbox`, awaited. Its `idempotency_receipts` table goes in `migrations/0001_initial.sql` (`inboxSchema()`), not an unawaited `ensureSchema()`.
- **A script run as a child process in tests** cannot reach the test's PGlite. Call the function it wraps in-process, and `node --check` the script.
++- **Deals, Coupons, Trade and Reviews (2026-09-28)** added these:
++ - **Higher-order wrappers.** A helper that takes a function (`conflictGuard(fn)`, `unique(fn)`, `staffAction(fn)`, a batch) must `await fn()`. Otherwise its `try` catches nothing and the caller races the write. Match PostgreSQL's error codes (`23505` unique, `23503` foreign key) with `err.table` and `err.detail`, not SQLite's message.
++ - **Per-process counters around a transaction** (`batchDepth++ … finally batchDepth--`) break once requests interleave. Keep per-call state in `AsyncLocalStorage`.
++ - **Factories stay synchronous.** A factory that seeded a row at construction became `async`, and every caller got a promise. Write such rows lazily on first use.
++ - **SQL.** Name aliases are not allowed in `HAVING`: repeat the aggregate. A bare column next to `MAX()` in a `GROUP BY` takes its value from the max row in SQLite; use `DISTINCT ON … ORDER BY` instead. `SELECT DISTINCT` over a `LEFT JOIN` with `ORDER BY lower(x)` fails; use `WHERE … OR EXISTS (…)`. Cast a parameter tested only with `IS NULL` (`@p::bigint IS NULL`). Give a recursive CTE's seed the column's type and collation (`?::text COLLATE "C"`). `json_each(?)` becomes `jsonb_array_elements_text(?::jsonb)`.
++ - **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.
ddiff --git a/tools/asyncify/README.md b/tools/asyncify/README.md
iindex 0690286..caa96e6 100644
---- a/tools/asyncify/README.md
++++ b/tools/asyncify/README.md
@@@ -37,9 +37,25 @@ What stays for a person:
Also by hand: `db.transaction(fn)()` → `db.tx(fn)` (the codemod leaves it), scalar `MAX(a, b)` → `GREATEST`, `rowid` tiebreaks → an identity `seq` column, and a `setImmediate` meant to run after a commit → `db.afterCommit`. The lessons list in `docs/migrating-to-postgresql.md` §7 has the rest.
++## The order for a Blog-shaped service
++
++```
++git checkout -b postgres
++python3 g1pass.py <svc> <Name> <ENV> # package pins, config, outbox, limits, readiness, boot, CI, unit, env, import script
++npm install
++node gen-migration.js <Name> --publishing … # migrations/0001_initial.sql from the SQLite SCHEMA
++python3 g1db.py <svc> <Name> # server/db.js: openDb / createStore / openStore
++node asyncify.js --config <svc>.json … # await everything; wraps .map(async …) in Promise.all
++python3 sqlfix.py # the SQL rewrites, and CHECK lines for a person
++npm test && npm run test:pg # then rehearse: scripts/migrate-to-postgres.js --sqlite <copy> --pglite
++python3 g1docs.py <svc> <Name> <ENV> # README and STATUS.json
++```
++
++`asyncify.js` now turns `xs.map(cb)` with a callback that becomes async (or an async function passed by name) into `(await Promise.all(xs.map(…)))`, recursively outwards. It reports `.map(async …)` that is not awaited as a whole, and callbacks of other array methods.
++
## sqlite-schema-to-pg.js
--`toPg(sqliteDdl)` turns the SCHEMA string a service's `db.js` ran into the first migration: `INTEGER PRIMARY KEY AUTOINCREMENT` → identity, `TEXT` → `text COLLATE "C"`, `INTEGER` → `bigint`, `REAL` → `double precision`, `BLOB` → `bytea`, and no `IF NOT EXISTS`. It refuses triggers.
++`toPg(sqliteDdl)` turns the SCHEMA string a service's `db.js` ran into the first migration: `INTEGER PRIMARY KEY AUTOINCREMENT` → identity, `TEXT` → `text COLLATE "C"`, `INTEGER` → `bigint`, `REAL` → `double precision`, `BLOB` → `bytea`, and no `IF NOT EXISTS`. An index on `x COLLATE NOCASE` is on `lower(x)`. An immutable-row trigger (`BEFORE UPDATE [OF …]/DELETE … RAISE(ABORT, msg)`) becomes PL/pgSQL. Any other trigger is refused. `gen-migration.js` runs it on `server/db.js` at HEAD (it evaluates a SCHEMA with `${…}`) and adds the Publishing and SDK tables.
## g1pass.py
ddiff --git a/tools/asyncify/asyncify.js b/tools/asyncify/asyncify.js
iindex 97e6b1f..fed858b 100644
---- a/tools/asyncify/asyncify.js
++++ b/tools/asyncify/asyncify.js
@@@ -107,14 +107,57 @@ function analyse(file, src, cfg, asyncNames) {
const becameAsync = new Set();
const newlyAsyncNames = new Set();
const fnsToAsync = new Set();
++ const wrappedMaps = new Set();
++ const isFnN = (x) => x && (x.type === 'FunctionDeclaration' || x.type === 'FunctionExpression' || x.type === 'ArrowFunctionExpression');
++ const inPromiseArg = (call) => { const p = ancestors.get(call) || []; const up = p[p.length - 1]; return up && up.type === 'CallExpression' && up.arguments[0] === call && up.callee.type === 'MemberExpression' && up.callee.object.name === 'Promise'; };
++ /** fn will contain an await: make it async; when it is a .map() callback, wrap that map in (await Promise.all(…)) and
++ * make its own enclosing function async too (recursively); another array method's callback needs a person. */
++ function makeAsync(fn) {
++ if (!fn || fnsToAsync.has(fn)) return;
++ fnsToAsync.add(fn);
++ const anc = ancestors.get(fn) || [];
++ const call = anc[anc.length - 1];
++ if (!(call && call.type === 'CallExpression' && call.arguments.includes(fn) && call.callee.type === 'MemberExpression' && ARRAY_CB.has(call.callee.property.name))) return;
++ if (call.callee.property.name !== 'map') { manual.push(`${file}:${call.loc.start.line}: a .${call.callee.property.name}() callback becomes async`); return; }
++ if (inPromiseArg(call) || wrappedMaps.has(call)) return;
++ wrappedMaps.add(call);
++ edits.push({ at: call.start, text: '(await Promise.all(' });
++ edits.push({ at: call.end, text: '))' });
++ makeAsync([...anc].reverse().find((x) => x !== fn && isFnN(x)));
++ }
++ // An async callback already there whose map result is not awaited as a whole is almost always a bug.
++ walk.full(ast, (n) => {
++ if (n.type === 'CallExpression' && n.callee.type === 'MemberExpression' && n.callee.property.name === 'map' && isFnN(n.arguments[0]) && n.arguments[0].async && !inPromiseArg(n)) {
++ const p = ancestors.get(n) || []; const up = p[p.length - 1];
++ if (!(up && up.type === 'VariableDeclarator')) manual.push(`${file}:${n.loc.start.line}: .map(async …) whose promises are not awaited together (Promise.all)`);
++ }
++ });
-- // An async function passed by name to an array method (xs.map(asyncFn)) needs a hand fix too.
++ // An async function passed by name to an array method: xs.map(asyncFn) / xs.map(api.asyncMethod) is wrapped in
++ // (await Promise.all(…)) like an async callback; any other array method needs a person.
++ const isFnNode = (x) => x.type === 'FunctionDeclaration' || x.type === 'FunctionExpression' || x.type === 'ArrowFunctionExpression';
walk.full(ast, (n) => {
-- if (n.type === 'CallExpression' && n.callee.type === 'MemberExpression' && ARRAY_CB.has(n.callee.property.name)
-- && n.arguments[0] && n.arguments[0].type === 'Identifier' && localAsync.has(n.arguments[0].name)) {
-- const gp = null;
-- manual.push(`${file}:${n.loc.start.line}: async function ${n.arguments[0].name} passed to .${n.callee.property.name}()`);
++ if (n.type !== 'CallExpression' || n.callee.type !== 'MemberExpression' || !ARRAY_CB.has(n.callee.property.name) || !n.arguments[0]) return;
++ const arg = n.arguments[0];
++ let what = null;
++ if (arg.type === 'Identifier' && localAsync.has(arg.name) && !syncNames.has(arg.name)) what = `async function ${arg.name}`;
++ if (arg.type === 'MemberExpression' && !arg.computed && arg.object.type === 'Identifier' && apis[arg.object.name]) {
++ const set = asyncNames.api.get(apis[arg.object.name]);
++ if (set && set.has(arg.property.name) && !syncNames.has(arg.property.name)) what = `async method ${arg.object.name}.${arg.property.name}`;
++ }
++ if (!what) return;
++ const p = ancestors.get(n) || [];
++ const up = p[p.length - 1];
++ if (n.callee.property.name === 'map' && up && up.type === 'CallExpression' && up.callee.type === 'MemberExpression' && up.callee.object.name === 'Promise') return;
++ const outer = [...p].reverse().find(isFnNode);
++ if (n.callee.property.name === 'map' && outer && !wrappedMaps.has(n)) {
++ wrappedMaps.add(n);
++ edits.push({ at: n.start, text: '(await Promise.all(' });
++ edits.push({ at: n.end, text: '))' });
++ makeAsync(outer);
++ return;
}
++ manual.push(`${file}:${n.loc.start.line}: ${what} passed to .${n.callee.property.name}()`);
});
walk.fullAncestor(ast, (node, _s, anc) => {
if (node.type !== 'CallExpression' || !asyncCall(node)) return;
@@@ -135,11 +178,25 @@ function analyse(file, src, cfg, asyncNames) {
const fn = fnIdx >= 0 ? parents[fnIdx] : null;
if (fn) {
const fnParent = parents[fnIdx - 1];
-- if (fnParent && fnParent.type === 'CallExpression' && fnParent.arguments.includes(fn) && fnParent.callee.type === 'MemberExpression' && ARRAY_CB.has(fnParent.callee.property.name)) {
-- manual.push(`${file}:${node.loc.start.line}: async call inside a .${fnParent.callee.property.name}() callback`);
-- return;
++ const inPromiseAll = (call) => { const p = ancestors.get(call) || []; const up = p[p.length - 1]; return up && up.type === 'CallExpression' && up.arguments[0] === call && up.callee.type === 'MemberExpression' && up.callee.object.name === 'Promise'; };
++ if (fnParent && fnParent.type === 'CallExpression' && fnParent.arguments.includes(fn) && fnParent.callee.type === 'MemberExpression' && ARRAY_CB.has(fnParent.callee.property.name)
++ && !(fnParent.callee.property.name === 'map' && inPromiseAll(fnParent))) {
++ // xs.map(cb) → (await Promise.all(xs.map(async cb))): same order, same length. Other array methods
++ // change meaning with an async callback (filter keeps every promise), so they stay for a person.
++ const outer = (() => { for (let i = fnIdx - 2; i >= 0; i--) if (isFn(parents[i])) return parents[i]; return null; })();
++ if (fnParent.callee.property.name === 'map' && outer) {
++ if (!wrappedMaps.has(fnParent)) {
++ wrappedMaps.add(fnParent);
++ edits.push({ at: fnParent.start, text: '(await Promise.all(' });
++ edits.push({ at: fnParent.end, text: '))' });
++ makeAsync(outer);
++ }
++ } else {
++ manual.push(`${file}:${node.loc.start.line}: async call inside a .${fnParent.callee.property.name}() callback`);
++ return;
++ }
}
-- fnsToAsync.add(fn);
++ makeAsync(fn);
}
const wrap = parent && parent.type === 'MemberExpression' && parent.object === node;
edits.push({ at: node.start, text: wrap ? '(await ' : 'await ' });
@@@ -161,7 +218,8 @@ function analyse(file, src, cfg, asyncNames) {
}
function apply(src, edits) {
-- const sorted = edits.slice().sort((a, b) => b.at - a.at || (a.text === ')' ? -1 : 1));
++ // At one position, closing text goes in first so it ends up after what opens there.
++ const sorted = edits.slice().sort((a, b) => b.at - a.at || ((a.text[0] === ')' ? 0 : 1) - (b.text[0] === ')' ? 0 : 1)));
let out = src;
for (const e of sorted) out = out.slice(0, e.at) + e.text + out.slice(e.at);
return out;
ddiff --git a/tools/asyncify/g1db.py b/tools/asyncify/g1db.py
nnew file mode 100644
iindex 0000000..c4a11b0
---- /dev/null
++++ b/tools/asyncify/g1db.py
@@@ -0,0 +1,62 @@
++#!/usr/bin/env python3
++"""server/db.js of a Blog-shaped service, from better-sqlite3 to openvibe-sdk/db (ADR-035): openDb (DATABASE_URL through
++PgBouncer, migrations as the owner, PGlite in development), createStore (the stores openStore returned, on the async
++handle) and openStore(config). The SCHEMA literal goes (it is migrations/0001_initial.sql now: gen-migration.js).
++usage: g1db.py <svc> <Name> run in the service's checkout after gen-migration.js."""
++import re, sys
++svc, Name = sys.argv[1], sys.argv[2]
++p = 'server/db.js'
++s = open(p).read()
++if 'openvibe-sdk/db' in s: sys.exit('server/db.js is converted already')
++m = re.search(r"\nfunction openStore\(dbPath, \{ now = \(\) => Date\.now\(\) \} = \{\}\) \{\n[\s\S]*?\n return \{\n([\s\S]*?)\n \};\n\}\n", s)
++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]
++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)
++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.
++ */
++async function openDb(config, {{ log = console, registry }} = {{}}) {{
++ if (!config.db.url) {{
++ if (config.isProduction) throw new Error('DATABASE_URL is not set: production serves from PostgreSQL (OpenVibe.Host roles/data add-service.sh {svc})');
++ log.warn(`[{Name}] DATABASE_URL unset: embedded PGlite database in ${{DEV_PGLITE}} (development only, one process)`);
++ fs.mkdirSync(DEV_PGLITE, {{ recursive: true }});
++ const db = createDb({{ pglite: DEV_PGLITE, service: '{svc}', registry, log }});
++ await db.migrate({{ dir: MIGRATIONS, log }});
++ return db;
++ }}
++ if (!config.db.directUrl) throw new Error('DATABASE_DIRECT_URL is not set: migrations run with the owner role on a direct connection');
++ const owner = createDb({{ url: config.db.directUrl, service: '{svc}-migrate', max: 1, log }});
++ try {{ await owner.migrate({{ dir: MIGRATIONS, log }}); }} finally {{ await owner.close(); }}
++ return createDb({{ url: config.db.url, service: '{svc}', registry, log }});
++}}
++
++/**
++ * Every store on a migrated database handle. opts.now — injectable clock (epoch ms), so tests and replays are
++ * deterministic. store.tx(fn) is a transaction; inside it, plain db calls join it (ambient).
++ */
++function createStore(db, {{ now = () => Date.now() }} = {{}}) {{
++ return {{
++ db,
++ now,
++{chr(10).join(props)}
++ tx: async (fn) => await db.tx(() => fn()),
++ close: () => db.close(),
++ }};
++}}
++
++/** openDb + createStore. */
++async function openStore(config, {{ now, log }} = {{}}) {{
++ return createStore(await openDb(config, {{ log }}), {{ now }});
++}}
++"""
++tail = re.sub(r"module\.exports = \{ openStore(, )?", r"module.exports = { openDb, openStore, createStore, MIGRATIONS\1", tail)
++s = head + body + tail
++open(p, 'w').write(s)
++print('server/db.js converted; stores kept:', ', '.join(re.findall(r"^\s+(\w+):", '\n'.join(props), re.M)) or 'none')
ddiff --git a/tools/asyncify/g1docs.py b/tools/asyncify/g1docs.py
nnew file mode 100644
iindex 0000000..d4bc2dd
---- /dev/null
++++ b/tools/asyncify/g1docs.py
@@@ -0,0 +1,42 @@
++#!/usr/bin/env python3
++"""README.md and STATUS.json edits after a service's PostgreSQL pass (the wording Blog, News and Deals use).
++usage: g1docs.py <svc> <Name> <ENVPREFIX> run in the service's checkout, after package.json has the new pins.
++Each edit is skipped when its anchor is missing (it prints what it could not place, for a hand edit)."""
++import json, re, sys
++svc, Name, ENV = sys.argv[1], sys.argv[2], sys.argv[3]
++pkg = json.load(open('package.json'))
++ver = lambda n: (re.search(r'/v([0-9.]+)$', pkg['dependencies'].get(n, '')) or [None, None])[1]
++s = open('README.md').read(); o = s
++miss = []
++def sub(pattern, repl, flags=0, what=''):
++ global s
++ n = re.subn(pattern, repl, s, count=1, flags=flags)
++ if n[1]: s = n[0]
++ else: miss.append(what or pattern[:60])
++sub(r"live in %s'?s? own SQLite \(`%s_DB_PATH`\)" % (re.escape(Name), ENV),
++ f"live in {Name}' own PostgreSQL database (`ov_{svc}` on the host's data role, ADR-035; schema in [migrations/](migrations/))", what='tables line')
++if 'PostgreSQL 18 and Valkey 9' not in s:
++ sub(r"\n- \*\*Packages\*\*", "\n- **PostgreSQL 18 and Valkey 9** (OpenVibe.Host `roles/data/`, ADR-035): every read and write is async through\n `openvibe-sdk/db`; Valkey holds the per-actor limit counters (optional: without `VALKEY_URL` they count per process).\n- **Packages**", what='PG bullet')
++for name in ['openvibe-publishing', 'openvibe-contracts', 'openvibe-sdk', 'openvibe-shared']:
++ v = ver(name)
++ if v: s = re.sub(r"(`%s`\s+)v[0-9.]+" % re.escape(name), r"\g<1>v" + v, s)
++s = s.replace('(chrome, app icon', '(Frame, app icon')
++unit = re.search(r"The unit is `openvibe-%s\.service` on `[^`]+`, the env file `/etc/openvibe/%s\.env`\." % (svc, svc), s)
++if unit and 'add-service.sh' not in s:
++ s = s[:unit.end()] + f""" The database is
++`ov_{svc}` on the host's data role (`sudo /opt/openvibe.host/roles/data/add-service.sh {svc}` writes its settings); the
++release migrates it at boot. The one-time move from SQLite is `scripts/migrate-to-postgres.js` (openvibe-sdk
++`runSqliteMigration`, with a `--pglite` rehearsal mode), run while the service is stopped; the old
++`/var/lib/openvibe-{svc}/{svc}.db` stays read-only for 7 days as the rollback.""" + s[unit.end():]
++elif not unit: miss.append('unit paragraph')
++s = re.sub(r"Nothing blocks a rollback: the schema\s+code only adds tables and columns\.", "Migrations only add tables and columns.", s)
++if s != o: open('README.md', 'w').write(s)
++st = json.load(open('STATUS.json'))
++st['runtime'] = re.sub(r'better-sqlite3 \d+', 'PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9', st.get('runtime', ''))
++if ver('openvibe-contracts'): st['contracts'] = f"openvibe-contracts v{ver('openvibe-contracts')}"
++if 'packages' in st:
++ st['packages'] = [f"{n} v{ver(n)}" if ver(n) else p for p in st['packages'] for n in [p.split(' ')[0]]]
++open('STATUS.json', 'w').write(json.dumps(st, indent=2, ensure_ascii=False) + '\n')
++left = [l for l in s.splitlines() if re.search(r'SQLite|better-sqlite3', l)]
++print('README/STATUS updated' + (f"; not placed: {', '.join(miss)}" if miss else ''))
++for l in left: print(' still says SQLite:', l[:140])
ddiff --git a/tools/asyncify/g1pass.py b/tools/asyncify/g1pass.py
iindex 0d9a6ec..74bc05a 100644
---- a/tools/asyncify/g1pass.py
++++ b/tools/asyncify/g1pass.py
@@@ -9,12 +9,24 @@ def edit(p, pairs, required=True):
if not os.path.exists(p): print('skip (missing)', p); return
s = open(p).read(); o = s
for a, b in pairs:
-- if b in s and a not in s: continue
++ if b in s: continue # applied already (b may contain a)
if a not in s:
if required: raise SystemExit(f'{p}: anchor not found: {a[:90]}')
continue
s = s.replace(a, b, 1)
if s != o: open(p, 'w').write(s); print('edited', p)
++# package.json: the PostgreSQL stack and the pins it needs
++pkg = json.load(open('package.json'))
++tag = lambda repo, v: f'https://codeload.github.com/OpenVibers/{repo}/tar.gz/refs/tags/{v}'
++d = pkg['dependencies']
++d['openvibe-sdk'] = tag('OpenVibe.SDK', os.environ.get('G1_SDK', 'v0.20.0'))
++if 'openvibe-publishing' in d: d['openvibe-publishing'] = tag('OpenVibe.Publishing', os.environ.get('G1_PUBLISHING', 'v1.0.0'))
++if 'openvibe-contracts' in d: d['openvibe-contracts'] = tag('OpenVibe.Contracts', os.environ.get('G1_CONTRACTS', 'v0.76.0'))
++d.setdefault('pg', '^8.23.0'); d.setdefault('iovalkey', '^0.4.0')
++pkg['dependencies'] = dict(sorted(d.items()))
++pkg.setdefault('devDependencies', {})['@electric-sql/pglite'] = '^0.5.8'
++pkg['scripts']['test:pg'] = f'{ENV}_TEST_STORE=pg node test/run.js'
++open('package.json', 'w').write(json.dumps(pkg, indent=2) + '\n'); print('package.json')
# config
edit('server/config.js', [(f" dbPath: env.{ENV}_DB_PATH || './data/{svc}.db',", f""" // PostgreSQL (ADR-035): DATABASE_URL serves (PgBouncer), DATABASE_DIRECT_URL migrates (owner role).
db: {{ url: env.DATABASE_URL || '', directUrl: env.DATABASE_DIRECT_URL || '' }},
@@@ -22,13 +34,15 @@ edit('server/config.js', [(f" dbPath: env.{ENV}_DB_PATH || './data/{svc}.
// The SQLite file of releases before the switch: read once by scripts/migrate-to-postgres.js.
dbPath: env.{ENV}_DB_PATH || './data/{svc}.db',""")])
# outbox
--s = open('server/events/outbox.js').read()
--edit('server/events/outbox.js', [
++if os.path.exists('server/events/outbox.js'):
++ edit('server/events/outbox.js', [
("const { createEventsClient, createOutbox } = require('openvibe-sdk/events');", "const { createEventsClient, createPgOutbox } = require('openvibe-sdk/events');"),
(" const outbox = createOutbox(db, {", " // The PostgreSQL outbox: rows are written in the change's own transaction (enqueue(db, …) joins the ambient\n // transaction); several processes relay one table safely (leases).\n const outbox = createPgOutbox(db, {"),
-- (" outbox.ensureSchema();\n", ""),
])
s = open('server/events/outbox.js').read()
++s2 = re.sub(r"\n[ \t]*(await )?outbox\.ensureSchema\(\);[^\n]*", '', s)
++if s2 != s: open('server/events/outbox.js', 'w').write(s2); print('outbox: schema is the migration now')
++s = open('server/events/outbox.js').read()
s2 = re.sub(r'outbox\.enqueue\((?!db, )', 'outbox.enqueue(db, ', s)
if s2 != s: open('server/events/outbox.js', 'w').write(s2); print('enqueue(db, …)')
# app.js store
@@@ -45,9 +59,9 @@ edit('server/http/actor-limits.js', [
(" now,\n", " now,\n // Shared across processes on Valkey (ADR-035) when VALKEY_URL is set; in-process otherwise.\n ...(valkey ? { store: createValkeyLimitStore(valkey) } : {}),\n"),
])
s = open('server/app.js').read()
--m = re.search(r" ctx\.limits = createActorLimits\(\{ config, now: opts\.limitsNow \|\| \(\(\) => Date\.now\(\)\), registry: metrics\.registry, log \}\);", s)
++m = re.search(r" ctx\.(limits|actorLimits) = createActorLimits\(\{ config, now: opts\.limitsNow \|\| \(\(\) => Date\.now\(\)\), registry: metrics\.registry, log \}\);", s)
if m:
-- s = s.replace(m.group(0), " // Valkey (ADR-035): shared, never-authoritative state (per-actor limit counters). Optional.\n const valkey = opts.valkey !== undefined ? opts.valkey : (config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null);\n ctx.valkey = valkey;\n ctx.limits = createActorLimits({ config, now: opts.limitsNow || (() => Date.now()), registry: metrics.registry, log, valkey });")
++ s = s.replace(m.group(0), " // Valkey (ADR-035): shared, never-authoritative state (per-actor limit counters). Optional.\n const valkey = opts.valkey !== undefined ? opts.valkey : (config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null);\n ctx.valkey = valkey;\n ctx." + m.group(1) + " = createActorLimits({ config, now: opts.limitsNow || (() => Date.now()), registry: metrics.registry, log, valkey });")
open('server/app.js', 'w').write(s); print('app valkey')
# readiness
s = open('server/observability.js').read(); o = s
@@@ -118,6 +132,7 @@ VALKEY_URL=
VALKEY_PREFIX=
# The SQLite file of releases before PostgreSQL: read once by scripts/migrate-to-postgres.js.
{ENV}_DB_PATH=./data/{svc}.db""")], required=False)
++os.makedirs('scripts', exist_ok=True)
cfgmod = "require('../server/config')"
open('scripts/migrate-to-postgres.js', 'w').write(f"""#!/usr/bin/env node
'use strict';
ddiff --git a/tools/asyncify/gen-migration.js b/tools/asyncify/gen-migration.js
nnew file mode 100644
iindex 0000000..3450be3
---- /dev/null
++++ b/tools/asyncify/gen-migration.js
@@@ -0,0 +1,68 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * The first migration of a better-sqlite3 service (ADR-035), written once: the SCHEMA template literal of its
++ * server/db.js through sqlite-schema-to-pg, then the openvibe-publishing stores it opens and the openvibe-sdk inbox
++ * and outbox. Run in the service's checkout, after `npm install` of openvibe-publishing ≥ 1.0 and openvibe-sdk ≥ 0.18:
++ *
++ * 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]
++ * → migrations/0001_initial.sql
++ *
++ * It lists every `rowid` the server code still orders by: those tables need a `seq bigint GENERATED ALWAYS AS
++ * IDENTITY UNIQUE` column (a person adds it; see docs/migrating-to-postgresql.md §7).
++ */
++const fs = require('fs');
++const path = require('path');
++const { execFileSync } = require('child_process');
++const { toPg } = require('./sqlite-schema-to-pg');
++
++const args = process.argv.slice(2);
++const Name = args[0];
++if (!Name) { console.error('usage: gen-migration.js <Name> [--publishing store:prefix,…] [--no-inbox]'); process.exit(2); }
++const pi = args.indexOf('--publishing');
++const pubs = pi >= 0 ? args[pi + 1].split(',').map((x) => x.split(':')) : [];
++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 m = /const SCHEMA = `([\s\S]*?)`;/.exec(src);
++if (!m) { console.error('server/db.js (HEAD) has no const SCHEMA = `…`;'); process.exit(1); }
++if (m[1].includes('${')) {
++ const tmp = path.resolve('server', '.db-head-schema.js');
++ fs.writeFileSync(tmp, src);
++ try {
++ const schema = require(tmp).SCHEMA;
++ if (typeof schema !== 'string') throw new Error('it does not export SCHEMA');
++ m[1] = schema;
++ } catch (err) { console.error(`the SCHEMA literal has interpolations and could not be evaluated: ${err.message}`); process.exit(1); } finally { fs.unlinkSync(tmp); }
++}
++let out = `-- phase: expand
++-- OpenVibe.${Name} on PostgreSQL (ADR-035, roadmap WS-X2): the tables as they were on SQLite (converted by openvibe-sdk
++-- tools/asyncify/sqlite-schema-to-pg: text COLLATE "C" compares like SQLite, integers are bigint, identities keep their ids),
++-- then the openvibe-publishing stores and the openvibe-sdk inbox and outbox. Generated once on ${new Date().toISOString().slice(0, 10)}; never edited after it runs.
++${toPg(m[1]).replace(/\n{3,}/g, '\n\n')}`;
++const SCHEMA_OF = { 'index-hooks': (x, p) => x.sequencerSchema(p), seo: (x, p) => x.redirectsSchema(p) };
++const opt = (name, def) => { const i = args.indexOf(name); return i >= 0 ? args[i + 1] : def; };
++for (const [store, prefix] of pubs) {
++ const mod = req(`openvibe-publishing/${store}`);
++ out += `\n-- openvibe-publishing/${store} (prefix ${prefix})\n${(SCHEMA_OF[store] ? SCHEMA_OF[store](mod, prefix) : mod.schema(prefix)).trim()}\n`;
++}
++const ev = req('openvibe-sdk/events');
++const inboxTable = opt('--inbox-table', 'idempotency_receipts');
++const outboxTable = opt('--outbox-table', 'event_outbox');
++if (!args.includes('--no-inbox')) out += `\n-- openvibe-sdk/events inbox: one receipt per (consumer, event) handled\n${ev.inboxSchema(inboxTable).trim()}\n`;
++out += `\n-- openvibe-sdk/events PostgreSQL outbox\n${ev.outboxSchema(outboxTable).trim()}\n`;
++// SQLite's own copies of the tables above (created by the packages at boot) would be defined twice.
++for (const t of [outboxTable, inboxTable]) {
++ const re = new RegExp(`\\nCREATE TABLE ${t} \\([\\s\\S]*?\\n\\);\\n`, 'g');
++ if ((out.match(re) || []).length && out.includes(`CREATE TABLE IF NOT EXISTS ${t}`)) out = out.replace(re, '\n');
++}
++fs.mkdirSync('migrations', { recursive: true });
++fs.writeFileSync(path.join('migrations', '0001_initial.sql'), out);
++console.log(`migrations/0001_initial.sql: ${(out.match(/CREATE TABLE/g) || []).length} tables`);
++try {
++ const hits = execFileSync('git', ['grep', '-n', 'rowid', '--', 'server', 'scripts'], { encoding: 'utf8' }).trim();
++ if (hits) console.log(`rowid still used (add a seq identity column to these tables and order by it):\n${hits}`);
++} catch { /* no rowid */ }
ddiff --git a/tools/asyncify/sqlfix.py b/tools/asyncify/sqlfix.py
nnew file mode 100644
iindex 0000000..34aaa35
---- /dev/null
++++ b/tools/asyncify/sqlfix.py
@@@ -0,0 +1,76 @@
++#!/usr/bin/env python3
++"""The SQLite → PostgreSQL SQL rewrites that are safe to do by text (ADR-035), over a service's server/, scripts/ and
++test/ files; and a report of what needs a person.
++usage: sqlfix.py [files…] (default: git ls-files server scripts test, *.js)
++
++Rewrites:
++ INSERT OR IGNORE INTO t … → INSERT INTO t … ON CONFLICT DO NOTHING (at the end of that string literal)
++ x COLLATE NOCASE → lower(x)
++ LIKE / NOT LIKE → ILIKE / NOT ILIKE (SQLite's LIKE ignores ASCII case)
++ json_each(?) → jsonb_array_elements_text(?::jsonb) (a JSON text parameter; column "value")
++ IFNULL( → COALESCE(
++ sqlite_master table lists → information_schema.tables of the current schema
++ pragma_table_info('t') → information_schema.columns of t
++ sequencer.stamp(doc) → sequencer.stamp(store.db, doc) (openvibe-publishing 1.0)
++Reported (fix by hand): INSERT OR REPLACE, rowid, scalar MAX(a, b)/MIN(a, b), HAVING on an alias, WITH RECURSIVE (the
++seed needs the column's collation), datetime/strftime/julianday, IFNULL, GROUP_CONCAT, json_extract, lastInsertRowid,
++.transaction(, setImmediate/nextTick near a transaction, createInbox, PRAGMA."""
++import re, subprocess, sys
++files = sys.argv[1:] or [f for f in subprocess.run(['git', 'ls-files', 'server', 'scripts', 'test'], capture_output=True, text=True).stdout.split() if f.endswith('.js')]
++REPORT = [
++ (r'INSERT OR REPLACE', 'INSERT OR REPLACE → ON CONFLICT (key) DO UPDATE SET … (qualify the target columns)'),
++ (r'\browid\b', 'rowid → a seq identity column'),
++ (r'\b(MAX|MIN)\(\s*[\w.]+\s*,', 'scalar MAX/MIN → GREATEST/LEAST'),
++ (r'HAVING\s+[a-z_]+\s*[<>=!]', 'HAVING on an alias → repeat the aggregate'),
++ (r'SELECT DISTINCT', 'SELECT DISTINCT with ORDER BY on an expression not selected fails; a DISTINCT over a LEFT JOIN is usually WHERE … OR EXISTS (…)'),
++ (r'GROUP BY', 'GROUP BY: every selected bare column must be grouped (SQLite took a bare column from the MAX/MIN row: DISTINCT ON)'),
++ (r'WITH RECURSIVE', 'recursive seed: cast ? to the column type and collation (?::text COLLATE "C")'),
++ (r"datetime\(|strftime\(|julianday\(", 'SQLite date functions'),
++ (r'GROUP_CONCAT|json_extract|json_group_array|json_object\(', 'SQLite functions'),
++ (r'lastInsertRowid', 'lastInsertRowid → RETURNING'),
++ (r'\.transaction\(', 'db.transaction(fn)() → db.tx(fn)'),
++ (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'(@\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 = []
++for p in files:
++ s = open(p).read(); o = s
++ while 'INSERT OR IGNORE INTO' in s:
++ i = s.index('INSERT OR IGNORE INTO'); q = s[i - 1] if s[i - 1] in '\'"`' else '`'
++ j = s.index(q, i)
++ s = s[:i] + 'INSERT INTO' + s[i + len('INSERT OR IGNORE INTO'):j] + ' ON CONFLICT DO NOTHING' + s[j:]
++ s = re.sub(r'([\w.]+) COLLATE NOCASE', r'lower(\1)', s)
++ 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)
++ # ON CONFLICT … DO UPDATE SET x = x + 1 / COALESCE(excluded.x, x): a bare column on the right is ambiguous in
++ # 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)
++ k = rest.index('DO UPDATE SET') + len('DO UPDATE SET')
++ head, sets = rest[:k], rest[k:]
++ def fix_rhs(a):
++ lhs, eq, rhs = a.partition('=')
++ if not eq: return a
++ for c in cols:
++ rhs = re.sub(r'(?<![.\w@:$])%s\b(?!\s*\()' % re.escape(c), f'{table}.{c}', rhs)
++ return lhs + eq + rhs
++ parts = re.split(r'(,\s*(?=\w+\s*=))', sets)
++ return f'INSERT INTO {table} ({m.group(2)}){head}' + ''.join(fix_rhs(x) if not re.match(r',\s*$', x) else x for x in parts)
++ s = re.sub(r'INSERT INTO (\w+) \(([^)]*)\)((?:(?!INSERT INTO)[^`\'"])*?DO UPDATE SET[^`\'"]*)', qualify, s)
++ # 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'\))",
++ "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+)'\)",
++ r"SELECT column_name AS name FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = '\1' ORDER BY ordinal_position", s)
++ if s != o: open(p, 'w').write(s); changed.append(p)
++ for n, line in enumerate(s.split('\n'), 1):
++ for pat, why in REPORT:
++ if re.search(pat, line) and not line.lstrip().startswith(('//', '*')):
++ print(f'CHECK {p}:{n}: {why}: {line.strip()[:120]}')
++print('rewrote: ' + (', '.join(changed) or 'nothing'))
ddiff --git a/tools/asyncify/sqlite-schema-to-pg.js b/tools/asyncify/sqlite-schema-to-pg.js
iindex eb7b8f4..c6ea611 100644
---- a/tools/asyncify/sqlite-schema-to-pg.js
++++ b/tools/asyncify/sqlite-schema-to-pg.js
@@@ -7,13 +7,25 @@
* INTEGER → bigint (epoch milliseconds and counters; openvibe-sdk/db returns int8 as Number)
* REAL → double precision, BLOB → bytea
* CREATE … IF NOT EXISTS → CREATE … (a migration runs once)
-- * Types only in column position (upper case, as SQLite schemas are written); comments are kept. Triggers
-- * (RAISE(ABORT, …)) are refused: write them as PL/pgSQL by hand.
++ * Types only in column position (upper case, as SQLite schemas are written); comments are kept. An immutable-row
++ * trigger (BEFORE UPDATE/DELETE … RAISE(ABORT, msg)) becomes a PL/pgSQL function and trigger; any other trigger is
++ * refused (write it by hand).
*
* const { toPg } = require('openvibe-sdk/tools/asyncify/sqlite-schema-to-pg'); (from a checkout)
*/
function toPg(sql) {
-- if (/CREATE\s+TRIGGER/i.test(sql)) throw new Error('sqlite-schema-to-pg: triggers need writing as PL/pgSQL by hand');
++ // The one trigger shape services use: a row that must never change (BEFORE UPDATE/DELETE … RAISE(ABORT, msg)).
++ sql = sql.replace(/CREATE TRIGGER(?: IF NOT EXISTS)? (\w+) BEFORE (UPDATE(?: OF [\w\s,]+?)?|DELETE) ON (\w+)\s+BEGIN SELECT RAISE\(ABORT, '([^']*)'\); END;/g,
++ (m, name, op, table, 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 EXECUTE FUNCTION ${name}();`);
++ if (sql.split(';').some((st) => /CREATE\s+TRIGGER/i.test(st) && !/FOR EACH ROW 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');
++ }
++ // COLLATE NOCASE: in an index, the index is on lower(col) (queries compare lower(col) too: sqlfix.py); on a column
++ // definition it goes (a case-insensitive UNIQUE there needs a lower() unique index, written by hand).
++ sql = sql.replace(/(CREATE (?:UNIQUE )?INDEX[^;]*?\()([^;]*?)\)/g, (m, head, cols) => head + cols.replace(/(\w+) COLLATE NOCASE/g, 'lower($1)') + ')');
++ if (/^\s*\w+\s+TEXT\b[^\n]*COLLATE NOCASE[^\n]*UNIQUE|UNIQUE[^\n]*COLLATE NOCASE/m.test(sql)) throw new Error('sqlite-schema-to-pg: a case-insensitive UNIQUE column needs a unique index on lower(col), by hand');
++ sql = sql.replace(/^(\s*\w+\s+TEXT\b[^\n]*?) COLLATE NOCASE/gm, '$1');
return sql
.replace(/\bINTEGER PRIMARY KEY AUTOINCREMENT\b/g, 'bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY')
.replace(/CREATE (UNIQUE )?INDEX IF NOT EXISTS/g, (m, u) => `CREATE ${u || ''}INDEX`)