Postgres · any driver

Postgres without an ORM. One function — run a parameterised statement, return rows — is the entire contract.

Why this exists

The Drizzle adapter came first, but it only ever used Drizzle as a SQL builder — it imports exactly two things, sql and its type, and writes raw SQL with them. Nothing about this library needs an ORM.

So this adapter takes the query function directly. The whole contract is one signature:

type SqlQuery = (
  text: string,
  params: readonly unknown[],
) => Promise<readonly Record<string, unknown>[]>;

Run a parameterised statement, give back rows. Every Postgres driver already does this, which is why the same adapter covers all of them.

Setup

import { postgresAdapter } from "easy-ping/adapters/postgres";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

export const notify = easyPing({
  database: postgresAdapter(
    async (text, params) => (await pool.query(text, params as unknown[])).rows,
  ),
  // ...
});

Create the tables with renderPostgresDdl(coreSchema) from easy-ping/schema, the same as any other non-Drizzle setup.

Other drivers

import postgres from "postgres";

const sql = postgres(process.env.DATABASE_URL);

postgresAdapter(
  async (text, params) => await sql.unsafe(text, params),
  { transaction: (fn) => sql.begin((tx) => fn((t, p) => tx.unsafe(t, p))) },
);

import { Kysely, PostgresDialect } from "kysely";

const db = new Kysely({ dialect: new PostgresDialect({ pool }) });

postgresAdapter(
  async (text, params) => (await db.executeQuery({ sql: text, parameters: [...params] })).rows,
);

import { neon } from "@neondatabase/serverless";

const sql = neon(process.env.DATABASE_URL);

// No transaction option: HTTP-mode serverless drivers have no session to hold
// one open. createNotifications still works, it just is not atomic.
postgresAdapter(async (text, params) => await sql.query(text, [...params]));

Prisma works too, via $queryRawUnsafe — though if you're already on Prisma you're paying for an ORM this adapter exists to avoid.

Transactions

The optional transaction option is what makes createNotifications atomic: a notification and its deliveries land together, or neither does.

import { pgTransaction, postgresAdapter } from "easy-ping/adapters/postgres";

postgresAdapter(query, { transaction: pgTransaction(pool) });

pgTransaction is the same sixteen lines everyone wrote by hand, in one place. The part worth centralising is the finally: drop the client.release() and every send leaks a connection until the pool is exhausted, which only shows up under load.

On another driver, write it yourself. The shape is the same:

transaction: async (fn) => {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await fn(async (t, p) => (await client.query(t, p as unknown[])).rows);
    await client.query("COMMIT");
    return result;
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

Creating the tables

createPostgresTables renders the DDL from the installed version's schema and runs it, so the tables cannot drift from the package you have. Every statement is IF NOT EXISTS, so it is safe on every boot.

import { createPostgresTables } from "easy-ping/adapters/postgres";
import { pushSchema } from "easy-ping/plugins/push";

// Plugins own their own tables and export the declaration, so this needs no
// plugin instance — and therefore no provider invented just to read a schema.
await createPostgresTables(query, { plugins: [pushSchema] });

Postgres-specific, on purpose

This is not a generic SQL adapter, and calling it one would be a lie. The statements use ON CONFLICT, IS DISTINCT FROM, RETURNING and — the one that matters — FOR UPDATE SKIP LOCKED, which is what lets two concurrent sweeps step around each other instead of blocking.

The other SQL engines are their own adapters, sharing one dialect-parameterised core rather than copies of this file: MySQL spells the claim two ways depending on whether you pass a transaction, and SQLite leans on its single writer. All six backends in the repo run the same conformance suite; this one was verified against it before it shipped, including the row-lock case.

LISTEN/NOTIFY wake-ups

With more than one replica, a send() on one process should wake the worker and the bell's event stream on the others. postgresSignals does that over LISTEN/NOTIFY, with no extra table:

import postgres from "postgres";
import { postgresAdapter, postgresSignals } from "easy-ping/adapters/postgres";

const sql = postgres(process.env.DATABASE_URL!);

export const notify = easyPing({
  database: postgresAdapter((text, params) => sql.unsafe(text, params as never[])),
  signals: postgresSignals(sql),   // { channel?: "easy_ping", onError? }
  // ...
});

Every logical channel (inbox:<userId>, deliveries) travels as the payload on one Postgres channel, because Postgres channel names are identifiers and user ids are not. Each process tags its own publishes so it does not react to its own echo.

For node-postgres, LISTEN is per connection, so hand it a dedicated connected Client, not the pool:

import { Client } from "pg";
import { pgListenNotify, postgresSignals } from "easy-ping/adapters/postgres";

const listener = new Client({ connectionString: process.env.DATABASE_URL });
await listener.connect();
signals: postgresSignals(pgListenNotify(listener)),