diff --git a/README.md b/README.md index 074f2db..eedfb0d 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,9 @@ version, licence and Node floor for every package: - **Row archiving** — moves aged rows out of hot tables into a mirror `archive` schema on a pg_cron schedule (idempotent DB setup). - **Change-notify triggers** — installs Postgres `NOTIFY` triggers for row - changes. + changes, reporting the schema alongside the table so tables of the same name + in different schemas are told apart. `withoutChangeNotify()` lets a bulk + write commit without a notification per row. - **Down-migrations** — `migrateDown()` undoes applied Prisma migrations (Prisma has no native "down"). - **SQL log helpers** — `prettifySql()` and cooperative log suppression. diff --git a/src/change-notify.ts b/src/change-notify.ts index d37e006..803136e 100644 --- a/src/change-notify.ts +++ b/src/change-notify.ts @@ -30,6 +30,19 @@ export const CHANGE_NOTIFY_CHANNEL = 'record_change_notify'; export const CHANGE_NOTIFY_TRIGGER_NAME = 'record_change_notify'; /** Default name of the trigger's plpgsql notify function. */ export const CHANGE_NOTIFY_FUNCTION_NAME = 'record_change_notify_fn'; +/** + * Setting the trigger reads to decide whether to stay quiet. + * + * @remarks + * Named after the trigger it silences, like the channel, trigger and function + * above it — a reader meeting it in a `SET LOCAL` can tell what it belongs to + * without knowing this package. + * + * A setting of our own rather than `session_replication_role`, which would also + * switch off foreign-key enforcement — a bulk load run that way can leave orphan + * rows behind. This suppresses nothing but these notifications. + */ +export const CHANGE_NOTIFY_SUPPRESS_SETTING = 'record_change_notify.suppressed'; /** Which tables notify, and under which Postgres object names. */ export interface ChangeTriggerConfig { @@ -39,14 +52,32 @@ export interface ChangeTriggerConfig { triggerName?: string; /** Name of the shared plpgsql function (default {@link CHANGE_NOTIFY_FUNCTION_NAME}). */ functionName?: string; + /** + * Setting that silences the trigger (default + * {@link CHANGE_NOTIFY_SUPPRESS_SETTING}). + */ + suppressSetting?: string; + /** + * Default schema of the listed tables (default `public`). + * + * @remarks + * A model may name its own schema as `schema.Table`, which wins over this. + * That is what lets a service reconcile tables it creates at run time — + * a tenant schema, say — without naming the default one. + */ + schema?: string; /** * Tables that should notify — the complete desired set, not an addition. * * @remarks * Reconciliation is two-way: a table listed here without the trigger gets it, - * and a table that HAS the trigger but is not listed here has it dropped. So - * omitting `models` entirely (the default empty array) removes the trigger from - * every table it is currently on. + * and a table that HAS the trigger but is not listed here has it dropped. + * + * It is confined to the schemas these names mention, so a call listing only + * public tables leaves triggers in other schemas alone — otherwise a service + * that installs per-schema would tear down its own work on the next start. + * Within those schemas it is absolute: omitting `models` entirely removes the + * trigger from every table in the default schema. */ models?: readonly string[]; /** Suppress SQL logging for the install DDL (default `true`). */ @@ -73,15 +104,31 @@ export interface RawClient extends RawExecutor { $transaction(fn: (tx: RawExecutor) => Promise): Promise; } +/** A model name as `schema` and `table`, taking `fallback` when unqualified. */ +function split( + model: string, + fallback: string, +): { schema: string; table: string } { + const dot = model.indexOf('.'); + + return dot === -1 + ? { schema: fallback, table: model } + : { schema: model.slice(0, dot), table: model.slice(dot + 1) }; +} + +const qualify = (schema: string, table: string): string => `${schema}.${table}`; + /** * Install Postgres triggers that `NOTIFY` on every row change in the given tables. * * @remarks * Creates one shared plpgsql function and attaches a row-level trigger to each * table in `models`, firing after every insert, update and delete. Each - * notification is a JSON payload with three keys — `table`, `op` (`INSERT`, - * `UPDATE` or `DELETE`) and `row` — where `row` is the new row, or the OLD row for - * a delete. Listen for them with `@imqueue/pg-pubsub`, or any `LISTEN` client. + * notification is a JSON payload with four keys — `schema`, `table`, `op` + * (`INSERT`, `UPDATE` or `DELETE`) and `row` — where `row` is the new row, or the + * OLD row for a delete. `schema` is what makes a listener able to tell two tables + * of the same name apart, which is the ordinary case once a service creates + * schemas of its own. Listen for them with `@imqueue/pg-pubsub`, or any `LISTEN` client. * * The whole thing runs in one transaction and is idempotent, so it is safe on every * start. It also reconciles in both directions: `models` is the complete desired @@ -89,6 +136,10 @@ export interface RawClient extends RawExecutor { * trigger but is not listed has it dropped. Calling this with an empty or omitted * `models` therefore REMOVES every trigger of that name — it is not a no-op. * + * Reconciliation is confined to the schemas `models` mentions, plus `schema` + * itself. A service that installs triggers per tenant schema can therefore + * reconcile one of them without tearing down the others. + * * Two limits worth knowing. Postgres caps a notification payload at 8000 bytes and * raises an error beyond it, so a table with large rows can make its own writes * fail — this is unsuitable for wide or blob-bearing tables. And the trigger fires @@ -110,6 +161,8 @@ export async function installChangeTriggers( channel = CHANGE_NOTIFY_CHANNEL, triggerName = CHANGE_NOTIFY_TRIGGER_NAME, functionName = CHANGE_NOTIFY_FUNCTION_NAME, + schema = 'public', + suppressSetting = CHANGE_NOTIFY_SUPPRESS_SETTING, models = [], silent = true, }: ChangeTriggerConfig, @@ -121,10 +174,16 @@ export async function installChangeTriggers( DECLARE rec record; BEGIN + IF coalesce( + current_setting('${suppressSetting}', true), '' + ) = 'on' THEN + RETURN NULL; + END IF; IF TG_OP = 'DELETE' THEN rec := OLD; ELSE rec := NEW; END IF; PERFORM pg_notify( TG_ARGV[0], json_build_object( + 'schema', TG_TABLE_SCHEMA, 'table', TG_TABLE_NAME, 'op', TG_OP, 'row', row_to_json(rec) @@ -135,31 +194,50 @@ export async function installChangeTriggers( $fn$ LANGUAGE plpgsql; `); - const rows = await tx.$queryRawUnsafe<{ table: string }[]>( - `SELECT event_object_table AS "table" - FROM information_schema.triggers - WHERE trigger_name = $1 - GROUP BY event_object_table`, + const wanted = models.map(model => split(model, schema)); + const schemas = [...new Set(wanted.map(one => one.schema))]; + + if (!schemas.includes(schema)) { + schemas.push(schema); + } + + const rows = await tx.$queryRawUnsafe< + { schema: string; table: string }[] + >( + `SELECT event_object_schema AS "schema", + event_object_table AS "table" + FROM information_schema.triggers + WHERE trigger_name = $1 + AND event_object_schema = ANY ($2) + GROUP BY event_object_schema, event_object_table`, triggerName, + schemas, ); - const installed = new Set(rows.map(row => row.table)); - const required = new Set(models); - for (const model of models) { - if (!installed.has(model)) { + const installed = new Set( + rows.map(row => qualify(row.schema, row.table)), + ); + const required = new Set( + wanted.map(one => qualify(one.schema, one.table)), + ); + + for (const one of wanted) { + if (!installed.has(qualify(one.schema, one.table))) { await tx.$executeRawUnsafe( `CREATE TRIGGER "${triggerName}" - AFTER INSERT OR UPDATE OR DELETE ON "${model}" + AFTER INSERT OR UPDATE OR DELETE + ON "${one.schema}"."${one.table}" FOR EACH ROW EXECUTE PROCEDURE ${functionName}('${channel}')`, ); } } - for (const model of installed) { - if (!required.has(model)) { + for (const row of rows) { + if (!required.has(qualify(row.schema, row.table))) { await tx.$executeRawUnsafe( - `DROP TRIGGER IF EXISTS "${triggerName}" ON "${model}"`, + `DROP TRIGGER IF EXISTS "${triggerName}" + ON "${row.schema}"."${row.table}"`, ); } } @@ -167,3 +245,50 @@ export async function installChangeTriggers( await (silent ? silently(install) : install()); } + +/** + * Run `fn` in a transaction whose row changes notify nobody. + * + * @remarks + * For a bulk write — an import, a backfill, a reconciliation — where the + * trigger would otherwise emit one notification per row. Forty thousand rows + * is forty thousand payloads through a single Postgres notification queue, to + * tell listeners something they would rather hear once. + * + * The suppression is `SET LOCAL`, so it belongs to this transaction alone: it + * reverts on commit or rollback, and no concurrent session is affected. It is + * a setting the trigger itself reads, **not** `session_replication_role` — that + * would silence foreign-key enforcement too, and a bulk load run under it can + * commit orphan rows. + * + * Do the writes on the `tx` handed to `fn`. Writes issued on the outer client + * go out on a different connection, where the setting was never applied, and + * will notify as usual. + * + * Nothing is emitted afterwards to say what changed — a caller that suppresses + * is telling listeners it will account for the change itself, by bumping a + * revision, invalidating a tag, or announcing the import once when it is done. + * + * @param client - A client that can open a transaction. + * @param fn - Work to run with notifications suppressed. + * @param setting - Setting the trigger reads (default + * {@link CHANGE_NOTIFY_SUPPRESS_SETTING}). + * @returns Whatever `fn` resolves to. + * @example + * ```typescript + * await withoutChangeNotify(prisma, async tx => { + * await tx.$executeRawUnsafe(bulkUpsert); + * }); + * ``` + */ +export async function withoutChangeNotify( + client: RawClient, + fn: (tx: RawExecutor) => Promise, + setting: string = CHANGE_NOTIFY_SUPPRESS_SETTING, +): Promise { + return client.$transaction(async tx => { + await tx.$executeRawUnsafe(`SET LOCAL "${setting}" = 'on'`); + + return fn(tx); + }); +} diff --git a/test/unit/barrel.spec.ts b/test/unit/barrel.spec.ts index 411b37f..aedb00e 100644 --- a/test/unit/barrel.spec.ts +++ b/test/unit/barrel.spec.ts @@ -36,6 +36,7 @@ const EXPORTS = [ 'AuditAction', 'CHANGE_NOTIFY_CHANNEL', 'CHANGE_NOTIFY_FUNCTION_NAME', + 'CHANGE_NOTIFY_SUPPRESS_SETTING', 'CHANGE_NOTIFY_TRIGGER_NAME', 'accessScope', 'accessWhere', @@ -50,6 +51,7 @@ const EXPORTS = [ 'silently', 'softDelete', 'toIsoDates', + 'withoutChangeNotify', ]; // The package is ESM, but Node >= 22 lets CommonJS `require()` an ESM module — diff --git a/test/unit/change-notify.spec.ts b/test/unit/change-notify.spec.ts new file mode 100644 index 0000000..ce64a33 --- /dev/null +++ b/test/unit/change-notify.spec.ts @@ -0,0 +1,218 @@ +/*! + * Postgres row-change NOTIFY trigger installer — tests + * + * I'm Queue Software Project + * Copyright (C) 2025 imqueue.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + * If you want to use this code in a closed source (commercial) project, you can + * purchase a proprietary commercial license. Please contact us at + * to get commercial licensing options. + */ + +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { + CHANGE_NOTIFY_SUPPRESS_SETTING, + installChangeTriggers, + withoutChangeNotify, +} from '../../src/change-notify.js'; + +interface Installed { + schema: string; + table: string; +} + +function client(installed: Installed[] = []) { + const statements: string[] = []; + const queries: { sql: string; values: unknown[] }[] = []; + + const executor = { + $executeRawUnsafe: async (sql: string) => { + statements.push(sql.replace(/\s+/g, ' ').trim()); + + return 0; + }, + $queryRawUnsafe: async (sql: string, ...values: unknown[]) => { + queries.push({ sql, values }); + + const schemas = (values[1] ?? []) as string[]; + + return installed.filter(one => + schemas.includes(one.schema), + ) as unknown as T; + }, + }; + + return { + statements, + queries, + ...executor, + $transaction: async (fn: (tx: typeof executor) => Promise) => + fn(executor), + }; +} + +describe('installChangeTriggers()', () => { + it('notifies the schema alongside the table', async () => { + const db = client(); + + await installChangeTriggers(db, { models: ['User'] }); + + const fn = db.statements.find(one => one.includes('pg_notify')); + + assert.ok(fn, 'the notify function is created'); + assert.match(fn, /'schema', TG_TABLE_SCHEMA/); + assert.match(fn, /'table', TG_TABLE_NAME/); + }); + + it('creates the trigger in the default schema', async () => { + const db = client(); + + await installChangeTriggers(db, { models: ['User'] }); + + assert.ok( + db.statements.some(one => one.includes('ON "public"."User"')), + 'the table is schema-qualified', + ); + }); + + it('takes a schema named on the model itself', async () => { + const db = client(); + + await installChangeTriggers(db, { models: ['tenant_a.Loan'] }); + + assert.ok( + db.statements.some(one => one.includes('ON "tenant_a"."Loan"')), + ); + }); + + it('drops a trigger the desired set no longer names', async () => { + const db = client([ + { schema: 'public', table: 'User' }, + { schema: 'public', table: 'Gone' }, + ]); + + await installChangeTriggers(db, { models: ['User'] }); + + assert.ok( + db.statements.some(one => + one.includes( + 'DROP TRIGGER IF EXISTS "record_change_notify" ON "public"."Gone"', + ), + ), + 'the unlisted table is reconciled away', + ); + assert.ok( + !db.statements.some( + one => + one.includes('"public"."User"') && one.startsWith('DROP'), + ), + 'the listed one is left alone', + ); + }); + + /** + * The whole point of confining it: a service that installs per tenant + * schema reconciles one without tearing down the rest. + */ + it('leaves schemas the desired set never mentions alone', async () => { + const db = client([ + { schema: 'public', table: 'User' }, + { schema: 'tenant_b', table: 'Loan' }, + ]); + + await installChangeTriggers(db, { models: ['User'] }); + + assert.deepEqual(db.queries[0]?.values[1], ['public']); + assert.ok( + !db.statements.some(one => one.includes('tenant_b')), + 'another schema is not touched', + ); + }); + + it('reads back only the schemas it is about to reconcile', async () => { + const db = client(); + + await installChangeTriggers(db, { + models: ['User', 'tenant_a.Loan'], + }); + + assert.deepEqual(db.queries[0]?.values[1], ['public', 'tenant_a']); + }); +}); + +describe('withoutChangeNotify()', () => { + it('sets the suppression only for its own transaction', async () => { + const db = client(); + + await withoutChangeNotify(db, async tx => { + await tx.$executeRawUnsafe('INSERT INTO "Term" VALUES (1)'); + }); + + assert.equal( + db.statements[0], + `SET LOCAL "${CHANGE_NOTIFY_SUPPRESS_SETTING}" = 'on'`, + 'the setting is LOCAL, so it reverts with the transaction', + ); + assert.match(db.statements[1] ?? '', /INSERT INTO "Term"/); + }); + + it('hands the transaction to the caller, not the outer client', async () => { + const db = client(); + let handed: unknown; + + await withoutChangeNotify(db, async tx => { + handed = tx; + }); + + assert.ok(handed, 'writes issued elsewhere would notify as usual'); + }); + + it('returns what the work returns', async () => { + const db = client(); + + assert.equal(await withoutChangeNotify(db, async () => 42), 42); + }); + + it('takes a different setting when the trigger was given one', async () => { + const db = client(); + + await withoutChangeNotify(db, async () => undefined, 'app.quiet'); + + assert.equal(db.statements[0], `SET LOCAL "app.quiet" = 'on'`); + }); +}); + +describe('the trigger body', () => { + it('returns early while the suppression is set', async () => { + const db = client(); + + await installChangeTriggers(db, { models: ['User'] }); + + const fn = db.statements.find(one => one.includes('pg_notify')) ?? ''; + + assert.match( + fn, + new RegExp( + `current_setting\\('${CHANGE_NOTIFY_SUPPRESS_SETTING}', true\\)`, + ), + ); + assert.ok( + fn.indexOf('RETURN NULL') < fn.indexOf('pg_notify'), + 'it gives up before building a payload, not after', + ); + }); +});