PostgreSQL store

A PostgreSQL Store for inkan: background jobs, sessions, logins, idempotency keys and rate limits kept in one table, so every process and every server of an app sees the same state. It runs on the database client you already have, pg (a Pool or a Client) or postgres (postgres.js), and has no dependencies of its own.

npm install @vxnsin/inkan @inkanjs/postgres pg   # or postgres

import { inkan, rateLimit } from "@vxnsin/inkan";
import pg from "pg";
import { postgresStore } from "@inkanjs/postgres";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const store = postgresStore(pool); // or postgresStore(postgres(url)) with postgres.js

const app = inkan({ workers: "auto" })
  .register(rateLimit({ max: 100, window: 60_000, store }))
  .job("/exports", { store }, async (job) => buildExport(job));

One store serves everything that keeps state; each part keeps its keys apart by a prefix of its own:

import { session } from "@inkanjs/session";
import { auth } from "@inkanjs/auth";
import { idempotency } from "@inkanjs/idempotency";

app
  .register(session({ secret, store }))
  .register(auth({ /* … */ store }))
  .register(idempotency({ store }));

With workers, inkan refuses jobs kept in memory; this store is shared: true, so jobs run in any worker, and progress and cancels reach the others through LISTEN/NOTIFY.

OptionsCopy Anchor Link

postgresStore(db, options?)

A sweep deletes 10000 rows a statement (for update skip locked, so two processes do not wait on each other) until a statement comes back short, at most a million rows a sweep; the next one goes on. store.sweep() runs one now. store.close() stops the sweep and the listening. The pool (or client, or sql) you passed in stays open: end it yourself when the app stops.

The tableCopy Anchor Link

create table if not exists "inkan_store" (key text primary key, value text not null, expires_at timestamptz);
create index if not exists "inkan_store_expires_at" on "inkan_store" (expires_at) where expires_at is not null;

With create: false, create it yourself (in a migration) with exactly these columns. value is the JSON text, text and not jsonb: jsonb cannot hold \u0000, reorders the keys of objects and drops duplicate ones, and a store has to give back what JSON.stringify made. value::jsonb still works in a query of your own.

What it runsCopy Anchor Link

Every value is a parameter. table and schema are checked against /^[A-Za-z_][A-Za-z0-9_]*$/ before any SQL and then quoted; anything else throws. Channel names never reach SQL as given: each channel is a database channel named by a hash of the table, prefix and channel (inkan_<40 hex>), so any name fits the 63 bytes of an identifier.

The driver is told apart by its methods: postgres.js has unsafe and listen, pg has query (a Pool also totalCount). A single pg Client gets one query at a time.

GuaranteesCopy Anchor Link

• add, take and increment are each one statement on a row lock, so they hold across every process and server on the same database, table and prefix.
• Expiry is by the database's clock, now(), never the app's, so servers whose clocks disagree still agree on what is over. An expired row counts as gone for every method (add takes its key, increment starts over) before the sweep deletes it.
• A key is never gone before its TTL (microsecond timestamps) and never returned after it.
• increment counts in numeric: 0.1 + 0.2 is 0.3. On a live key whose value is not a JSON number (a string, even "5", a boolean, null, an object) it rejects with a TypeError and leaves the value as it was, as the other stores do.
• The suite from @vxnsin/inkan/testing (storeConformance, see writing an adapter) passes on a pg Pool, a pg Client and postgres.js.

SecurityCopy Anchor Link

• Publish/subscribe has no authentication. A message is whatever arrives on the channel: any role that can connect to the database can NOTIFY (or select pg_notify(…)) any channel, without a grant, so anyone with a login to that database can send one, and inkan's jobs act on them (a forged message can cancel a running job or report false progress). The channel names are hashes of table, prefix and channel, which hides nothing from someone who knows them. Give the app a database of its own, and CONNECT on it only to the roles that should be able to talk to the app. Do not share the database with tenants you do not trust.
• Writing the store is trusting it. Whoever can write the keys can forge sessions, idempotent replies, job records and rate-limit counts, and whoever can read them reads every session. Give the app's database user access to the store's table (select, insert, update, delete on it) and nothing else, and nobody else write access to it.
• Values are stored as JSON text, as they are given: not encrypted, not signed. Keep secrets out of them or encrypt them yourself; keys are stored as given too (under the prefix), so do not put secrets in keys.
• Errors carry no keys or values. A failing statement rejects with a new Error that keeps code, severity, where and the schema, table, column and constraint names, its message with every quoted part replaced by "…"; pg's detail ("Key (value)=(…)") and postgres.js's query and parameters are not passed on. Log them freely.

LimitsCopy Anchor Link

• NOTIFY payloads are under 8000 bytes. publish rejects a larger message (as JSON) with a RangeError. inkan's jobs send small messages; keep your own small too and put large data in the store.
• Publish/subscribe is at most once. A message published while nobody listens, or while the listening connection reconnects, is lost. A dropped pg listening connection is replaced (tried every second); postgres.js reconnects on its own.
• With a pg Pool, the first subscribe takes one connection out of the pool for good (until close()): size the pool for it, or pass a connected pg.Client as listener. Behind PgBouncer in transaction mode LISTEN does not work; give it a direct connection.
• now() is the start of the transaction, so the TTL counts from when the statement began.
• The sweep runs in every process that has the store; the deletes overlap harmlessly. Rows without a TTL stay until deleted.
• The database should be UTF-8, or values with characters its encoding lacks fail.
• Keys and the prefix. The prefix (default "inkan:") ends in :; one without gets it ("app" is "app:"), so prefix "app" with key "1x" and prefix "app1" with key "x" stay apart. "" puts nothing in front. Do not give one use a prefix that starts another's ("app:" and "app:sessions:" can meet). A key or channel holding NUL (\0, which PostgreSQL text cannot hold) or a lone UTF-16 surrogate is refused with a TypeError, in all four inkan stores alike; in a value either is escaped by JSON and fine.
• Long keys. A key (with its prefix) longer than 2048 bytes of UTF-8 is kept under \x01sha256:<hex> of it (storedKey), as in @inkanjs/mysql, so it fits the primary key's btree. Two different keys share a row only if SHA-256 collides; keys starting with \x01 are hashed too, so no key can pose as a hash.
• Every call is a round trip. For thousands of increments a second, Redis is the better store.