add a postgres transport to bus-postgres

.changeset/postgres-transport.md

@@ -0,0 +1,5 @@
+---
+'@node-ts/bus-postgres': minor
+---
+
+Add `PostgresTransport`, a transport that keeps the bus' queues in Postgres tables, so a service needs no message broker (#266). It receives with `for update skip locked` and a visibility timeout (`visibilityTimeoutMs`, default 30 s), fans published events and sent commands out to the queues subscribed to them, receives new messages straight away with `LISTEN/NOTIFY` and polls every second as a fallback (`listen: false` for PgBouncer in transaction mode), retries after the recoverability policy's delay, and moves dead letters to a `transport_dead_letters` table. `bus provision` creates its tables, queue and subscriptions, and prints the grants it needs at runtime. It needs Postgres 13 or later, and throws `InvalidTransportDuration` for a `visibilityTimeoutMs` or `pollIntervalMs` that isn't above 0. See https://node-ts.github.io/bus/transports/postgres.

CLAUDE.md

@@ -46,7 +46,7 @@ pnpm exec dotenv -e test.env -- jest packages/bus-sqs/src/sqs-transport.spec.ts
 
 - `bus-messages` — `Message`/`Command`/`Event` base types and `MessageAttributes`. Messages identify themselves by `$name` (usually a static `NAME`) and `$version`.
 - `bus-core` — the bus itself, the `Transport`/`Persistence`/`Serializer`/`Receiver` interfaces (`Transport.endpointName` is the service queue's name), and in-memory defaults (`InMemoryQueue`, `InMemoryPersistence`, `JsonSerializer`).
-- Transports: `bus-sqs` (SNS topics fanned out to SQS queues, with policy generation), `bus-rabbitmq`.
+- Transports: `bus-sqs` (SNS topics fanned out to SQS queues, with policy generation), `bus-rabbitmq`, and `PostgresTransport` in `bus-postgres` (queues in shared Postgres tables, no broker; see `packages/bus-postgres/CLAUDE.md`).
 - Persistence (workflow state and delayed messages): `bus-postgres`, `bus-mongodb`.
 - `bus-sqs-lambda` — a `Receiver` that feeds Lambda SQS events into the bus instead of the bus polling the transport.
 - `bus-opentelemetry` — `openTelemetry()` middleware for spans, W3C trace context (carried in message attributes) and messaging metrics, with only `@opentelemetry/api` as a peer. See `packages/bus-opentelemetry/CLAUDE.md`.

README.md

@@ -20,7 +20,7 @@ Requires Node.js 24 or later.
 - [@node-ts/bus-rabbitmq](https://github.com/node-ts/bus/tree/master/packages/bus-rabbitmq) - A RabbitMQ transport
 - [@node-ts/bus-sqs](https://github.com/node-ts/bus/tree/master/packages/bus-sqs) - An Amazon SQS transport
 - [@node-ts/bus-sqs-lambda](https://github.com/node-ts/bus/tree/master/packages/bus-sqs-lambda) - A receiver that handles SQS messages in AWS Lambda
-- [@node-ts/bus-postgres](https://github.com/node-ts/bus/tree/master/packages/bus-postgres) - A Postgres persistence for workflow state
+- [@node-ts/bus-postgres](https://github.com/node-ts/bus/tree/master/packages/bus-postgres) - A Postgres persistence for workflow state, and a Postgres transport that needs no broker
 - [@node-ts/bus-mongodb](https://github.com/node-ts/bus/tree/master/packages/bus-mongodb) - A MongoDB persistence for workflow state
 - [@node-ts/bus-opentelemetry](https://github.com/node-ts/bus/tree/master/packages/bus-opentelemetry) - OpenTelemetry tracing and metrics, as bus middleware
 - [@node-ts/bus-nestjs](https://github.com/node-ts/bus/tree/master/packages/bus-nestjs) - A NestJS module: handlers and workflows as providers, and the bus run by Nest's lifecycle

docs/.vitepress/config.mts

@@ -126,6 +126,7 @@ const infrastructure: DefaultTheme.SidebarItem[] = [
       { text: 'RabbitMQ', link: '/transports/rabbitmq' },
       { text: 'Amazon SQS', link: '/transports/amazon-sqs' },
       { text: 'SQS and Lambda', link: '/transports/sqs-lambda' },
+      { text: 'Postgres', link: '/transports/postgres' },
       { text: 'Custom transports', link: '/transports/custom' }
     ]
   },

docs/guide/provisioning.md

@@ -11,12 +11,13 @@ A bus needs infrastructure before it can run: a queue, a topic or exchange for e
 
 Everything is worked out from the bus' configuration: its transport, persistence, handlers, workflows, custom handlers and message types.
 
-| Adapter                                           | Provisions                                                                                                                           |
-| ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
-| [Amazon SQS](/transports/amazon-sqs#provisioning) | An SNS topic per message, the service queue and dead letter queue, a subscription per handled message, and the queue's access policy |
-| [RabbitMQ](/transports/rabbitmq#provisioning)     | A fanout exchange per message, the service queue with its retry and dead letter queues, and the bindings                             |
-| [Postgres](/persistence/postgres#provisioning)    | The schema, a table per workflow state with indexes on its lookups, and the outgoing messages table                                  |
-| [MongoDB](/persistence/mongodb#provisioning)      | A collection per workflow state with indexes on its lookups, and the outgoing messages collection                                    |
+| Adapter                                                 | Provisions                                                                                                                           |
+| ------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
+| [Amazon SQS](/transports/amazon-sqs#provisioning)       | An SNS topic per message, the service queue and dead letter queue, a subscription per handled message, and the queue's access policy |
+| [RabbitMQ](/transports/rabbitmq#provisioning)           | A fanout exchange per message, the service queue with its retry and dead letter queues, and the bindings                             |
+| [Postgres transport](/transports/postgres#provisioning) | The schema, the messages, queues, subscriptions and dead letters tables, the service's queue, and a subscription per handled message |
+| [Postgres](/persistence/postgres#provisioning)          | The schema, a table per workflow state with indexes on its lookups, and the outgoing messages table                                  |
+| [MongoDB](/persistence/mongodb#provisioning)            | A collection per workflow state with indexes on its lookups, and the outgoing messages collection                                    |
 
 A topic or exchange is provisioned for every message the bus handles, and every message in the [message types](/guide/serializers/message-types) passed to `withMessageTypes()`, other than workflow state. That way a service that sends a message doesn't depend on the service that handles it being deployed first. A send-only bus provisions only those topics or exchanges, so **pass a send-only bus the message types of everything it sends**: without them it provisions nothing, its runtime permissions allow sending nothing, and `bus provision` logs a warning. The in-memory queue and persistence have nothing to provision.
 
@@ -121,7 +122,7 @@ SqsTransport (7 resources)
 
 ## Checking resources at startup
 
-`initialize()` creates nothing. It checks that what the bus receives through exists, such as its queues and subscriptions, and its tables and indexes, with read-only calls such as SQS's `GetQueueUrl` or Postgres' `to_regclass`, and throws `ResourcesNotProvisioned` naming each missing resource, so a service that was deployed before its infrastructure fails straight away rather than when the first message arrives. A send-only bus has nothing to check on its transport. The topics or exchanges of messages a bus only sends aren't checked at startup: a send to one that doesn't exist fails when it's sent. Each adapter's page lists the permissions the check needs. They're included in the permissions `--permissions` prints.
+`initialize()` creates nothing. It checks that what the bus receives through exists, such as its queues and subscriptions, and its tables and indexes, with read-only calls such as SQS's `GetQueueUrl` or Postgres' `to_regclass`, and throws `ResourcesNotProvisioned` naming each missing resource, so a service that was deployed before its infrastructure fails straight away rather than when the first message arrives. A send-only bus checks little or nothing on its transport, such as only the Postgres transport's tables. The topics or exchanges of messages a bus only sends aren't checked at startup: a send to one that doesn't exist fails when it's sent. Each adapter's page lists the permissions the check needs. They're included in the permissions `--permissions` prints.
 
 If the service's credentials can't be given those permissions, turn the check off:
 
@@ -157,5 +158,5 @@ A [custom transport](/transports/custom) or [persistence](/persistence/custom) i
 
 ## See also
 
-- [Amazon SQS](/transports/amazon-sqs#provisioning), [RabbitMQ](/transports/rabbitmq#provisioning), [Postgres](/persistence/postgres#provisioning) and [MongoDB](/persistence/mongodb#provisioning), for what each provisions and the permissions it needs
+- [Amazon SQS](/transports/amazon-sqs#provisioning), [RabbitMQ](/transports/rabbitmq#provisioning), the [Postgres transport](/transports/postgres#provisioning), [Postgres](/persistence/postgres#provisioning) and [MongoDB](/persistence/mongodb#provisioning), for what each provisions and the permissions it needs
 - [`BusInstance.provision`](/api/bus-core/classes/BusInstance#provision), [`ProvisioningPlan`](/api/bus-core/interfaces/ProvisioningPlan) and [`ResourcesNotProvisioned`](/api/bus-core/classes/ResourcesNotProvisioned) in the API reference

docs/persistence/postgres.md

@@ -5,7 +5,7 @@ description: Store workflow state in Postgres with @node-ts/bus-postgres.
 
 # Postgres
 
-`@node-ts/bus-postgres` stores workflow state, and messages sent with [delayed delivery](/guide/delayed-delivery), in [PostgreSQL](https://www.postgresql.org/). This page covers installing, configuring and provisioning it.
+`@node-ts/bus-postgres` stores workflow state, and messages sent with [delayed delivery](/guide/delayed-delivery), in [PostgreSQL](https://www.postgresql.org/). This page covers installing, configuring and provisioning it. The package also has a transport that keeps the bus' queues in Postgres: see [Postgres transport](/transports/postgres).
 
 <PackageBadge pkg="bus-postgres" />
 

docs/snippets/postgres-transport.ts

@@ -0,0 +1,37 @@
+import { Bus } from '@node-ts/bus-core'
+import {
+  PostgresPersistence,
+  PostgresTransport,
+  PostgresTransportConfiguration
+} from '@node-ts/bus-postgres'
+import { reserveRoomHandler } from './handlers/reserve-room-handler'
+import { messageTypes } from './message-types.generated'
+
+const connection = {
+  connectionString: 'postgres://postgres:password@localhost:5432/postgres'
+}
+
+const transportConfiguration: PostgresTransportConfiguration = {
+  queueName: 'reservations-service',
+  schemaName: 'bus',
+  connection,
+  // Longer than the slowest handler takes, or a message still being handled is received again
+  visibilityTimeoutMs: 30_000
+}
+const postgresTransport = new PostgresTransport(transportConfiguration)
+
+const bus = Bus.configure()
+  .withMessageTypes(messageTypes)
+  .withTransport(postgresTransport)
+  // Optional: workflow state and delayed messages in the same database
+  .withPersistence(
+    new PostgresPersistence({ connection, schemaName: 'workflows' })
+  )
+  .withHandler(reserveRoomHandler)
+  // For local development: creates the tables, the queue and its subscriptions when the bus initializes. In
+  // production, create them at deploy time with `bus provision` instead.
+  .withAutoProvision()
+  .build()
+
+await bus.initialize()
+await bus.start()

docs/transports.md

@@ -13,6 +13,7 @@ By default the bus uses an in-memory queue, `InMemoryQueue`, which needs nothing
   <Card title="RabbitMQ" tag="@node-ts/bus-rabbitmq" link="/transports/rabbitmq">An AMQP broker. The transport declares the exchanges and queues, and retries messages after the delay the recoverability policy chooses.</Card>
   <Card title="Amazon SQS" tag="@node-ts/bus-sqs" link="/transports/amazon-sqs">AWS' managed queues, with SNS topics for events. `bus provision` creates the topics, queues and subscriptions, or you manage them.</Card>
   <Card title="SQS and Lambda" tag="@node-ts/bus-sqs-lambda" link="/transports/sqs-lambda">Handle SQS messages in AWS Lambda, which reads the queue and passes batches to the bus.</Card>
+  <Card title="Postgres" tag="@node-ts/bus-postgres" link="/transports/postgres">Queues in the Postgres database you already have, with no broker to run. Suits one service, or services that share a database.</Card>
   <Card title="Custom transports" link="/transports/custom">Adapt another broker by implementing the Transport interface, and check it with the conformance suite.</Card>
 </FeatureGrid>