Create one pool per database, with timeouts, and close it at shutdown:
ts
import { createPostgres } from 'jg-postgres';const postgres = createPostgres({ host: 'db.internal', database: 'app', user: 'app', password: 'secret', max: 10, connectionTimeoutMillis: 30_000, // waiting for a connection statement_timeout: 60_000, // one statement, on the server lock_timeout: 30_000, // waiting for a row or table lock idle_in_transaction_session_timeout: 300_000, // a transaction left open onError: (error) => console.error('PostgreSQL', error),});const users = await postgres.query<{ id: string }>('SELECT id FROM user_account LIMIT $1;', [10]);await postgres.close();
The pool reads NUMERIC as a number and TIMESTAMPTZ as an ISO 8601 UTC string, such as '2026-10-17T08:30:00.000Z', and keeps BIGINT as a string, so no digit of a large amount is lost. The parsers are set on the pool, not node-postgres’s global registry, so other pools in the program keep theirs. Errors no caller awaits, such as an idle connection the server closed, go to onError, console.error by default, instead of crashing the process.
transaction commits when the operation resolves, rolls back when it throws, and rejects with the operation’s own error. It always returns its client to the pool, or destroys it when its state is unknown. When the connection breaks during COMMIT, nobody can tell whether the transaction committed, so it rejects with TRANSACTION_OUTCOME_UNKNOWN: check the data before retrying anything that isn’t idempotent. Run work inside a transaction on its client, and never start a transaction inside another.
Timeouts
A profile says how long a kind of work may run: a statement on the server, a wait for a lock, and the client’s wait for the answer, which adds a grace, 15 seconds by default, so the server’s own error arrives first. Give the pool the profile of everyday work, and a slow transaction one of its own:
timeouts sets the statement and lock timeouts for that transaction only, so the client goes back to the pool with the pool’s. The pool’s query_timeout still applies to each query that doesn’t set its own, so pass the transaction’s queries through withQueryTimeout with the same profile. A query the client stops waiting for is rolled back: the client goes back to the pool once ROLLBACK, which has the pool’s query_timeout too, confirms the connection, and is destroyed otherwise.
Optional reads
createOptionalReader runs reads that are nice to have, such as a counter in a page’s header, without ever delaying or crowding out essential work. A read is skipped, giving null at once, when the reader already runs maxActiveReads reads, a request waits for a client, or the pool can’t give more clients than reservedIdleClients, counting idle ones and the room left to open new ones. It runs in a read-only transaction with short timeouts, gives null after admissionSeconds, and never throws:
ts
import { createOptionalReader } from 'jg-postgres';const counters = createOptionalReader({ postgres }); // 2 reads at most, 2 clients left free, 1 secondconst unread = counters.read( `unread:${accountId}`, // callers of the same key share one read { text: 'SELECT count(*)::int AS count FROM notification WHERE account_id = $1;', values: [accountId], }, (row) => (row as { count: number }).count,);const count = await unread.completion; // a number, or null
Usage limits
admitUsage decides, atomically, whether an actor may do a rate-limited action now, such as sending an invitation, and records it when they may. It refuses with cooldown while the same action on the same scope was accepted less than cooldownSeconds ago, by any actor, and with limit once the actor did the action limit times within windowSeconds:
ts
import { admitUsage } from 'jg-postgres';const admission = await admitUsage(postgres, { actorId: userId, actionId: 'send-invitation', scopeKey: inviteeEmailHash, // the cooldown applies per invitee limit: 20, windowSeconds: 86_400, cooldownSeconds: 60,});// { outcome: 'accepted' | 'cooldown' | 'limit', retryAt: an ISO timestamp, or null }
Concurrent calls queue on transaction advisory locks, the scope’s then the actor’s, so two can never both pass, and every comparison uses the database’s clock. The accepted actions go in a usage_event table:
sql
CREATE TABLE usage_event ( id bigint GENERATED ALWAYS AS IDENTITY, actor_id varchar(256) NOT NULL, action_id varchar(128) NOT NULL, scope_key varchar(256) NULL, created_at timestamptz NOT NULL DEFAULT now(), CONSTRAINT pk_ue__id PRIMARY KEY (id), CONSTRAINT chk_ue__actor_id_non_empty CHECK (actor_id <> ''), CONSTRAINT chk_ue__action_id_non_empty CHECK (action_id <> ''), CONSTRAINT chk_ue__scope_key_non_empty CHECK (scope_key <> ''));CREATE INDEX idx_ue__actor_id_action_id_created_at ON usage_event (actor_id, action_id, created_at);CREATE INDEX idx_ue__action_id_scope_key_created_at_scoped ON usage_event (action_id, scope_key, created_at) WHERE scope_key IS NOT NULL;CREATE INDEX idx_ue__created_at ON usage_event (created_at);
The table grows with every accepted action. Prune it on a schedule, deleting in batches the rows older than the longest window or cooldown, $1 seconds, $2 rows at a time, such as 1024, and running it again while it deletes a whole batch. It compares with now(), which in a statement of its own is the statement’s start, because the index can bound a scan by now() but not by clock_timestamp():
sql
DELETE FROM usage_eventWHERE id IN ( SELECT id FROM usage_event WHERE created_at < now() - make_interval(secs => $1) ORDER BY created_at LIMIT $2);
Verification challenges
createVerificationChallenges issues and checks short codes bound to an actor, an action, and a target, such as the code that confirms a new email address. Each action has a policy: how long a code is valid, how many wrong codes it takes, and how many codes an actor may be issued within a window, over every target:
ts
import { createVerificationChallenges } from 'jg-postgres';const challenges = createVerificationChallenges(postgres, { policies: { 'verify-email': { validitySeconds: 600, maxFailedAttempts: 5, issueLimit: 5, issueWindowSeconds: 3_600, }, }, codeLength: 6, // the default});const binding = { actorId: userId, actionId: 'verify-email', targetId: newEmail };const issued = await challenges.issue(binding, ({ code, challengeId }) => mailer.send(newEmail, code, { idempotencyKey: challengeId }),);// { outcome: 'issued', challengeToken, expiresAt } or { outcome: 'limit', retryAt }// later, with the token the client kept and the code the user typedconst isVerified = await challenges.consume({ ...binding, challengeToken: savedToken, code: typedCode,});
issue refuses with limit once the actor was issued issueLimit codes within the window, with retryAt, when the oldest of them leaves it. Otherwise it supersedes the binding’s active challenge and commits the new one, then calls deliver with the code, the only time it’s available. A delivery that throws makes issue throw VERIFICATION_DELIVERY_FAILED, with the code and the token redacted from it and its cause; the challenge still counts toward the limit. Concurrent issues for an actor and an action queue on a transaction advisory lock, and every comparison uses the database’s clock.
consume resolves to true once, for the right code. A wrong code counts a failed attempt, and a challenge stops accepting codes after maxFailedAttempts of them, so the policy’s current value applies. A used, superseded, or expired challenge, another binding’s token, or a token or code of the wrong shape resolves to false without counting, and every false looks the same to the caller.
Only the SHA-256 hashes of the token and of the code, combined with its token, are stored, in a verification_challenge table:
sql
CREATE TABLE verification_challenge ( id bigint GENERATED ALWAYS AS IDENTITY, actor_id varchar(256) NOT NULL, action_id varchar(128) NOT NULL, target_id varchar(256) NOT NULL, token_hash text NOT NULL, code_hash text NOT NULL, failed_attempt_count smallint NOT NULL DEFAULT 0, expires_at timestamptz NOT NULL, used_at timestamptz NULL, superseded_at timestamptz NULL, created_at timestamptz NOT NULL DEFAULT now(), CONSTRAINT pk_vc__id PRIMARY KEY (id), CONSTRAINT uq_vc__token_hash UNIQUE (token_hash), CONSTRAINT chk_vc__actor_id_non_empty CHECK (actor_id <> ''), CONSTRAINT chk_vc__action_id_non_empty CHECK (action_id <> ''), CONSTRAINT chk_vc__target_id_non_empty CHECK (target_id <> ''), CONSTRAINT chk_vc__token_hash_sha256 CHECK (token_hash ~ '^[0-9a-f]{64}$'), CONSTRAINT chk_vc__code_hash_sha256 CHECK (code_hash ~ '^[0-9a-f]{64}$'), CONSTRAINT chk_vc__failed_attempt_count_non_negative CHECK (failed_attempt_count >= 0), CONSTRAINT chk_vc__expires_at_chronological CHECK (expires_at > created_at), CONSTRAINT chk_vc__used_at_superseded_at_mutually_exclusive CHECK (used_at IS NULL OR superseded_at IS NULL));CREATE UNIQUE INDEX uidx_vc__actor_id_action_id_target_id_active ON verification_challenge (actor_id, action_id, target_id) WHERE used_at IS NULL AND superseded_at IS NULL;CREATE INDEX idx_vc__actor_id_action_id_created_at ON verification_challenge (actor_id, action_id, created_at);CREATE INDEX idx_vc__created_at ON verification_challenge (created_at);
Prune it on a schedule as usage_event is pruned, deleting in batches the rows older than the longest validitySeconds or issueWindowSeconds of any policy, $1 seconds, $2 rows at a time. A younger row may still be consumed, or count toward a limit:
sql
DELETE FROM verification_challengeWHERE id IN ( SELECT id FROM verification_challenge WHERE created_at < now() - make_interval(secs => $1) ORDER BY created_at LIMIT $2);
Database summaries
readDatabaseSummary and listTableSizes read what an admin dashboard or a health check shows about the database, and diffTables compares the tables with the ones a migration should have created:
ts
import { diffTables, listTableSizes, readDatabaseSummary } from 'jg-postgres';const summary = await readDatabaseSummary(postgres.pool); // the public schema, or { schema }// { name, serverVersion, serverVersionNumber, sizeBytes, activeConnections, maxConnections, tableCount }const tables = await listTableSizes(postgres.pool); // [{ name, totalBytes }], sorted by nameconst { missing, unexpected } = diffTables( tables.map(({ name }) => name), ['app_user', 'session'],);
Sizes and counts are numbers, and sizes include indexes and TOAST data. activeConnections counts the connections to this database running a query. Without the pg_read_all_stats role, PostgreSQL shows a role only its own connections, so the count is lower than the server’s.
Counter tables
defineCounterTable builds the queries that change counter rows, such as an account’s daily usage or a balance, quoting every name once and sending every value as a parameter. applyCounterQueries runs them in the caller’s transaction, sorted by row, so two transactions that change the same rows lock them in the same order and never deadlock:
ts
import { applyCounterQueries, defineCounterTable } from 'jg-postgres';const dailyUsage = defineCounterTable({ table: 'daily_usage', keyColumns: ['account_id', 'day'], counters: ['request_count', 'token_count'],});await postgres.transaction((client) => applyCounterQueries(client, [ // inserts the row, or adds to it dailyUsage.buildIncrement( { account_id: accountId, day }, { request_count: 1, token_count: tokens }, ), // adds to a row that must exist, or throws COUNTER_ROW_NOT_FOUND dailyUsage.buildAdjustment({ account_id: accountId, day }, { token_count: -refunded }), ]),);
An increment is an INSERT … ON CONFLICT (key columns) DO UPDATE, so the table needs a unique constraint on exactly the key columns, and a default, such as 0, for every counter. Amounts are whole numbers or bigints within BIGINT: positive to increment, and non-zero to adjust. A counter that would leave BIGINT’s range fails with SQLSTATE 22003, which isNumericValueOutOfRangeError checks, and the transaction rolls back.
Constraint errors
ts
import { isUniqueViolationError } from 'jg-postgres';declare const error: unknown;if (isUniqueViolationError(error, 'uq_user_account__email')) { // the email is already registered}
isDatabaseError and the checks for unique, foreign key, not-null, check, and restrict violations, string truncation, numeric overflow, invalid text, serialization failures, deadlocks, and canceled queries read the SQLSTATE code. The constraint checks take an optional constraint name.
Advisory locks
createAdvisoryLockKey turns a name into the 64-bit key PostgreSQL’s advisory locks take. lockTransaction holds one until the transaction ends, waiting while another holds it. withSessionLock holds one on a connection across several transactions, or skips the work when another session holds it:
ts
import { createAdvisoryLockKey, createPostgres, withSessionLock } from 'jg-postgres';declare const postgres: ReturnType<typeof createPostgres>;const outcome = await withSessionLock( postgres, createAdvisoryLockKey('user-account-deletion:42'), async (lock) => { lock.assertActive(); // throws SESSION_LOCK_LOST once the lock's connection is gone return 'deleted'; },);// { acquired: false } when another request holds the lock
Cursor pages and ordered lists
listByCursor lists a page of rows after a cursor, and buildCursorPage turns a page listed with one extra row into { items, nextCursor }. moveUp and moveDown swap a row’s sort order with its neighbor’s, in their own transaction under the list’s lock, and lockOrderedList takes the same lock for an append. Every table and column name is quoted.
A record store
jg-postgres/record-store keeps small JSON records, such as counters or a feature’s settings, one row each in a record_store table, which the app creates:
sql
CREATE TABLE IF NOT EXISTS record_store ( id varchar(512), value jsonb NOT NULL, CONSTRAINT pk_rs__id PRIMARY KEY (id), CONSTRAINT chk_rs__id_non_empty CHECK (id <> ''), CONSTRAINT chk_rs__value_object CHECK (jsonb_typeof(value) = 'object'));
ts
import { createPostgres } from 'jg-postgres';import { createRecordStore } from 'jg-postgres/record-store';import { z } from 'zod';declare const postgres: ReturnType<typeof createPostgres>;const counters = createRecordStore({ postgres, id: 'admin:unread-counts', defaults: { notifications: 0, reports: 0 }, schema: z.object({ notifications: z.number(), reports: z.number() }),});await counters.incrementNumber('notifications', 1); // atomic, so concurrent increments all countawait counters.setNumber('reports', 0); // leaves the other fields as they areawait counters.get(); // { notifications: 1, reports: 0 }, checked with the schema
The defaults fill every missing or null field when the record is read, and are the only keys a write keeps. Every method takes an optional transaction’s client, to run in that transaction without committing it.
Errors
Code
When
TRANSACTION_OUTCOME_UNKNOWN
The connection was lost during COMMIT
SESSION_LOCK_LOST
The connection that held a session lock was lost
INVALID_IDENTIFIER
A name is empty, holds a null character, or is longer than 63 bytes
ROW_NOT_FOUND
The row to move doesn’t exist in the ordered list
ROW_ALREADY_FIRST
The row to move up is already first
ROW_ALREADY_LAST
The row to move down is already last
MOVE_FAILED
The move didn’t update exactly two rows
INVALID_TIMEOUT_PROFILE
A timeout profile’s seconds aren’t positive and finite, or its lock timeout is longer than its statement timeout
INVALID_OPTIONAL_READ_OPTIONS
An optional reader’s database, limits, or timeouts aren’t valid
INVALID_USAGE_ADMISSION
A usage admission’s ids, limit, or seconds aren’t valid, or it has a cooldown without a scope
INVALID_VERIFICATION_POLICY
The verification policies or the code length aren’t valid
INVALID_VERIFICATION_INPUT
A verification binding’s ids aren’t valid, its action has no policy, or issue has no delivery function
VERIFICATION_DELIVERY_FAILED
The delivery function threw, after the challenge was committed
INVALID_COUNTER_TABLE
A counter table has no key column or no counter, or repeats a name
INVALID_COUNTER_CHANGE
A counter row’s key doesn’t have exactly the key columns, or an amount isn’t a whole, non-zero amount within BIGINT
COUNTER_ROW_NOT_FOUND
A counter query changed no row, because the row to adjust doesn’t exist
A query’s own error, such as a constraint violation, is node-postgres’s DatabaseError, unchanged.
Generated from TSDoc
API reference
Every export of [email protected], 76 in all, with its signature, its errors, and an example.
Decides, atomically, whether an actor may do a rate-limited action now, such as sending an invitation, and records it when they may. It refuses with cooldown while the same action on the same scope was accepted less than cooldownSeconds ago, by any actor, and with limit once the actor did the action limit times within windowSeconds. Concurrent calls queue on transaction advisory locks, the scope's then the actor's, so they never both pass, and every comparison uses the database's clock. It needs the usage_event table the README defines.
The actor, the action, the optional scope, the limit and its window, and the optional cooldown.
Returns
Promise<IUsageAdmission>
A promise that resolves to the outcome, with retryAt, an ISO 8601 UTC timestamp: when the scope's cooldown ends, or null without one.
Throws
INVALID_USAGE_ADMISSIONThe usage admission needs an actor id of 1 to 256 characters, an action id of 1 to 128, a scope key of 1 to 256 when given, a whole positive limit, and positive seconds, with a cooldown only for a scope.
Example
import { admitUsage, createPostgres } from 'jg-postgres';declare const postgres: ReturnType<typeof createPostgres>;const admission = await admitUsage(postgres, { actorId: '42', actionId: 'send-invitation', scopeKey: '[email protected]', limit: 20, windowSeconds: 86_400, cooldownSeconds: 60,});if (admission.outcome !== 'accepted') { // tell the user when to try again: admission.retryAt, or tomorrow for 'limit'}
Runs counter queries one after another on a transaction's client, sorted by their lock keys, so two transactions changing the same rows lock them in the same order and never deadlock each other. Queries for the same row keep their order. Each must change exactly one row, so an adjustment of a row that doesn't exist throws, and the caller's transaction rolls back. A counter that would leave BIGINT's range fails with SQLSTATE 22003, as isNumericValueOutOfRangeError checks.
Turns a lock's name into the signed 64-bit key PostgreSQL's advisory locks take: the first 8 bytes of the name's SHA-256 hash. The same name always gives the same key, in every process.
Signature
createAdvisoryLockKey(name: string): string
Parameters
namestring
The lock's name, such as user-account-deletion:42.
Returns
string
The key, as a string, such as '7480330268761432310'.
Example
import { createAdvisoryLockKey } from 'jg-postgres';createAdvisoryLockKey('moldea'); // => '7480330268761432310'
Creates a reader for optional data, such as counters in a page's header, that must never delay or crowd out essential work. A read is skipped, giving null at once, when the reader already runs maxActiveReads reads, a request waits for a client, or the pool can't give more clients than reservedIdleClients, counting idle ones and the room left to open new ones. A running read gives null once admissionSeconds pass, and is shared by every caller of the same key until it ends. Each read runs in a read-only transaction with short timeouts of its own, and never throws.
The database, and optionally the limits and the reads' timeouts.
Returns
IOptionalReader
The reader.
Throws
INVALID_OPTIONAL_READ_OPTIONSThe optional reader needs a database, whole positive maxActiveReads, whole reservedIdleClients from 0, positive admissionSeconds, and a valid timeout profile.
Example
import { createOptionalReader, createPostgres } from 'jg-postgres';declare const postgres: ReturnType<typeof createPostgres>;declare const accountId: string;const counters = createOptionalReader({ postgres });const unread = counters.read( `unread:${accountId}`, { text: 'SELECT count(*)::int AS count FROM notification WHERE account_id = $1 AND read_at IS NULL;', values: [accountId] }, (row) => (row as { count: number }).count,);const count = await unread.completion; // a number, or null when the database is busy
Creates a PostgreSQL pool with the helpers every query, transaction, and lock uses. The pool parses NUMERIC as a number and TIMESTAMPTZ as an ISO 8601 UTC string, keeps BIGINT as a string, and sends errors no caller awaits, such as an idle connection the server closed, to onError instead of crashing the process.
Creates the functions that issue and consume verification challenges, such as the code that confirms a new email address. A challenge is bound to an actor, an action, and a target; only the SHA-256 hashes of its token and code are stored, in a verification_challenge table the README defines, and every comparison uses the database's clock.
issue refuses with limit once the actor was issued the action's issueLimit challenges within issueWindowSeconds, over every target, so concurrent issues queue on a transaction advisory lock. Otherwise it supersedes the binding's active challenge, commits the new one, and calls deliver with the code. consume uses the challenge when the code matches, or counts a failed attempt, and every failure resolves to false alike.
Each action's policy, keyed by its action id, and the code's digits.
Returns
IVerificationChallenges
The functions.
Throws
INVALID_VERIFICATION_POLICYThe verification challenges need at least one policy, each for an action id of 1 to 128 characters, with whole validity and window seconds of at least 1, a whole issue limit of at least 1, and a whole failed-attempt limit from 1 to 32767, and a whole code length from 4 to 10.
Defines a table of counters, such as an account's daily usage, and gives the functions that build the queries changing its rows. Every name is quoted once, and every value is a parameter. The table needs a unique constraint on exactly the key columns, for the increments' ON CONFLICT, and a default for every counter. Run the queries with applyCounterQueries, which orders them so concurrent transactions lock rows in the same order.
Defines how long a kind of work may run: a statement on the server, a wait for a lock, and the client's wait for the server's answer, which is the statement timeout plus a grace, so the server's own timeout error arrives first. A slow report and an interactive request each get one.
Compares the tables a schema has with the ones an app expects, such as in a health check after a migration. Names compare exactly, as PostgreSQL stores them.
Lists a page of a table's rows after a cursor, ordered by the cursor column. Every name is quoted, and the order and the cursor comparison use the table's own column, so a selected column can't change them.
Locks an ordered list for the rest of a transaction, so no move or append changes it meanwhile, such as before appending a row after the list's last sort order. moveUp and moveDown take the same lock.
Takes an advisory lock for the rest of a transaction, waiting while another transaction holds it. PostgreSQL releases it when the transaction commits or rolls back.
import { createAdvisoryLockKey, createPostgres, lockTransaction } from 'jg-postgres';declare const postgres: ReturnType<typeof createPostgres>;await postgres.transaction(async (client) => { await lockTransaction(client, createAdvisoryLockKey('invoice-numbers')); // only one transaction at a time gets here});
Quotes a table, column, or constraint name for SQL that can't take it as a parameter. A name PostgreSQL would cut short, an empty one, or one with a null character throws instead.
Signature
quoteIdentifier(name: string): string
Parameters
namestring
The name, such as order or sort_order.
Returns
string
The quoted name, such as "order".
Throws
INVALID_IDENTIFIERThe identifier is empty, holds a null character, or is longer than 63 bytes.
Reads a summary of the database the executor is connected to, such as for an admin dashboard: its name, the server's version, its size, its active connections out of the server's maximum, and the ordinary tables in a schema. Without the pg_read_all_stats role, PostgreSQL shows only the role's own connections, so activeConnections undercounts.
Gives a query the client timeout of a profile, which node-postgres's pool otherwise sets for every query, so a query in a transaction with longer timeouts isn't cut off by the pool's.
Runs an operation while holding an advisory lock on one connection, across as many transactions as it needs, or skips it when another session holds the lock. The lock lasts as long as its connection, so the operation can check it's still held. Afterwards the lock is released, and the client is destroyed instead of reused when the release fails or the connection was lost, which releases the lock too.
runs an operation in a transaction on one client, committing when it resolves and rolling back when it throws, optionally with timeouts of its own. Never start a transaction inside another: run the inner work on the client instead
closeproperty
() => Promise<void>
ends every connection, once queries in flight finish
what createPostgres takes: node-postgres's pool configuration, such as the host, the pool size, and connectionTimeoutMillis, statement_timeout, lock_timeout, and idle_in_transaction_session_timeout, plus where errors outside any query go
how long a query may run, in seconds: on the server, waiting for a lock, and on the client, which stops waiting for the server's answer. defineTimeoutProfile builds it
Declaration
type ITimeoutProfile = Readonly<{ statementTimeoutSeconds: number; lockTimeoutSeconds: number; queryTimeoutSeconds: number;}>;
type ITransactionOptions = { timeouts?: ITimeoutProfile | undefined;};
Members
timeoutsproperty
ITimeoutProfile
the statement and lock timeouts for this transaction only, from defineTimeoutProfile. The pool's query_timeout still applies to each query that doesn't set its own, so pass the operation's queries through withQueryTimeout with the same profile. A query the client stops waiting for is rolled back: the client goes back to the pool once ROLLBACK, which waits for the server and has the pool's query_timeout too, confirms the connection, and is destroyed otherwise
what admitUsage decides, with when to try again: accepted, refused by the scope's cooldown, or refused because the actor reached the limit within the window
What issue resolves to: the challenge's token, which the client keeps and sends back with the code, and when it expires; or, when the actor reached the policy's issue limit, when they may be issued one again.
Creates a store for one JSON record, such as a set of counters or a feature's settings, kept in the record_store table the package's README describes, which the app creates. The defaults fill every missing field when it's read, and are the only keys a write keeps. Every read is checked with the schema. A method given a transaction's client runs in it, and never commits it.