add a postgres transport to bus-postgres

[adenhertog]

Closes #266

Summary

Adds PostgresTransport to @node-ts/bus-postgres: the bus' queues kept in Postgres tables, so a service needs no message broker.

Background

#266 asks for a database-as-queue transport: SKIP LOCKED queue tables, LISTEN/NOTIFY, pub/sub through a subscription table, and passing the bus-test conformance suite. The design was agreed with the maintainer first. The accepted decisions are in this comment.

Problem

Every transport today needs a broker (RabbitMQ, or SQS and SNS). A single service, or services that share one database, have to run a broker just to send messages to themselves.

Approach

• Package: PostgresTransport lives in bus-postgres. It shares the persistence's SQL helpers, which are moved to an internal src/postgres-sql.ts; the persistence's behaviour is unchanged.

• Tables: shared by every queue in schemaName: transport_messages (index on (queue, visible_at)), transport_queues, transport_subscriptions and transport_dead_letters.

• Provisioning:

  • provision() creates the tables and, unless the bus is send-only, inserts the queue row and one subscription row per handled message and custom-handler topic. It never deletes anything.
  • initialize() only checks that everything exists, and throws ResourcesNotProvisioned.
  • The plan includes SQL grants for <runtime_role>.

• Sending: send and publish fan out to each subscribed queue in one statement, which also sends pg_notify for each queue reached. sendToAddress inserts straight into a provisioned queue, and throws EndpointNotFound otherwise.

• Receiving:

  • Uses a lease: one for update skip locked claim sets visible_at = now + visibilityTimeoutMs (default 30 s, no heartbeat) and a new lease token.
  • Delete, return and dead-letter only act while the lease token still matches. A receipt whose lease another receiver has taken settles nothing, and logs a warning.
  • failedAttempts is the number of earlier deliveries, as on SQS.

• Waiting for messages:

  • LISTEN runs on a connection of its own. Polling every pollIntervalMs (default 1 s) is always on as a fallback, and listen: false turns listening off for PgBouncer in transaction mode.
  • Each poll, each notification and each received message wakes one waiting read, so an idle process makes one query per poll whatever its concurrency.
  • A wake that comes while no read is waiting is kept, so the next read that finds the queue empty checks again rather than waiting for the next poll.
  • The listening connection uses the settings of a pool that was passed in, and has TCP keepalive. If it's lost, it reconnects with backoff, which only resets once the connection has stayed up for 30 s.
  • stop() waits for a listen that's still in flight, and a listen that ends up stale closes its own connection.
  • pollIntervalMs and visibilityTimeoutMs must be above 0, or the constructor throws InvalidTransportDuration.

• Retries and dead letters:

  • returnMessage sets visible_at.
  • fail moves the row to transport_dead_letters with the bus-failure header, in one statement.
  • Unparseable messages are dead-lettered when read.
  • The docs give SQL for redriving and purging dead letters, and the integration test runs the redrive SQL.

• Delayed delivery uses the existing outgoing-messages dispatcher, with no bus-core change. Joining the transport to the withOutbox() transaction is follow-up postgres transport: exactly-once by enlisting the transport in the outbox transaction #343, and the docs say the transport isn't part of that transaction.

• Tests:

  • transportTests passes.
  • Integration tests also cover:
    • provisioning: missing resources, and a dry run;
    • receiving through LISTEN and through polling;
    • an expired lease;
    • stop() while a claim that finds nothing is in flight (a regression test for a hang found in review);
    • a notification that arrives while no read is waiting;
    • an unparseable message;
    • withOutbox() on the same database: commit, rollback, inbox dedupe, a delayed send, and dead-letter redrive;
    • a second copy of a message that arrives while the first still holds its inbox row, after its lease expired, which runs the handler only once.
  • There are also unit tests.
  • outboxTests and inboxTests were not changed to take a transport. They run on in-memory queues they control, such as the failing queue in outboxTests, so the outbox and inbox are covered with this transport in its own integration test instead.

• Docs: a new docs/transports/postgres.md, which also shows how an external system publishes with pg_notify, with a sidebar entry, a card on the transports page, a row in the provisioning guide, the bus-postgres README, and the CLAUDE.md files. Minor changeset.

• This is original work under the clean-room policy, not ported, translated or copied from another messaging framework

• Added a changeset (pnpm changeset) for user-facing changes to published packages, or none is needed

• Docs: updated docs/ for user-facing changes, or none needed

🤖 Generated with Claude Code