Postgres

@postel/pg is a standalone adapter for Postgres: hand it a pg.Pool you already manage, or a connection string and Postel owns the pool. It's the adapter for multi-node, high-throughput delivery.

pnpm add @postel/pg pg

npm install @postel/pg pg

yarn add @postel/pg pg

bun add @postel/pg pg

Hand PgStorage a pg.Pool your app already manages — Postel reuses it, so the outbox insert can share your transactions:

import { Pool } from "pg";
import { Postel } from "@postel/core";
import { PgStorage } from "@postel/pg";
import { config } from "./config.js";

const pool = new Pool({ connectionString: config.databaseUrl });

export const postel = Postel({
  outbound: {
    storage: PgStorage({ pool }),
  },
});

Or pass a connection string and let Postel open and own the pool:

import { Postel } from "@postel/core";
import { PgStorage } from "@postel/pg";
import { config } from "./config.js";

export const postel = Postel({
  outbound: {
    storage: PgStorage({ connectionString: config.databaseUrl }),
  },
});

Options

Pass exactly one of connectionString or pool. pg (node-postgres) is a peer dependency you install alongside.

SKIP LOCKED reservation

Workers reserve outbox rows with UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED …). Concurrent workers each grab a disjoint batch without blocking on one another — this is what lets you scale delivery horizontally across many worker processes against one database.

LISTEN/NOTIFY

The adapter declares capabilities.notify = true. Every send() fires NOTIFY postel_messages_new, and idle workers LISTEN for it — so a freshly enqueued message is picked up with near-zero latency instead of waiting for the next poll tick.

Transactions compose with yours

Because Postel runs through node-postgres, you can share a transaction: open one on a client and pass it to send() so the outbox insert commits atomically with your business writes. See the Storage overview.

Receiver-side dedup

@postel/pg also exports PgDedup (and ensurePgDedupTable) — the inbound idempotency-dedup helper, independent of the outbound storage. Wire it onto an inbound source, reusing the same pool:

import { PgDedup } from "@postel/pg";

const postel = Postel({
  inbound: {
    vendor: {
      verify: Secret(config.webhookSecret),
      dedup: PgDedup({
        client: pool,
        // tableName: "postel_received_messages",  (default)
        // schema: undefined,                       (search_path)
        // autoMigrate: true,                       (creates the table on first use)
      }),
      dedupTtl: "24h",
    },
  },
});

The adapter uses INSERT … ON CONFLICT … DO UPDATE WHERE expires_at <= now() — atomic at the DB level, no application-level locks. The table is two columns: message_id text PRIMARY KEY, expires_at timestamptz, with an index on expires_at for cheap cleanup. See Deduplication for the dedup contract and where to place the call.