# Transactional recipient selection

> Choose recipients from your own routing table inside the same PostgreSQL transaction that commits the business write.

Source: https://good-webhooks.dev/docs/guides/recipient-selection



Normally, an event goes to every endpoint subscribed to its type. Use this recipe when **who receives it depends on the payload** or on rows you haven't committed yet. For example: send `invoice.paid` only to endpoints that handle the invoice's currency.

1. **BEGIN** (your app): Open a transaction on one client.
2. **Write and route**: Update the invoice, then query your routing table on the same client.
3. **publish()**: Pass the frozen recipient list through a one-off `EndpointSource`.
4. **COMMIT**: The invoice, the event, and its deliveries commit together.

Your tables and the delivery tables must share one PostgreSQL database. The recipe uses only the public API and starts no worker.

## Routing rules in your schema [#routing-rules-in-your-schema]

The example routes `invoice.paid` to endpoints whose currency and minimum amount match the invoice. An `invoice_routes` table owned by your application decides recipients for this event.

Include every eligible endpoint, paused ones too. Check endpoint ownership when you create routes. If you also require managed subscriptions, add that to the routing query.

```sql title="examples/transactions/schema.sql"
-- Application-owned example tables. Apply separately from the Good Webhooks migration.
CREATE SCHEMA transaction_example;

CREATE TABLE transaction_example.invoices (
  organization_id text NOT NULL,
  invoice_id text NOT NULL,
  amount integer NOT NULL CHECK (amount >= 0),
  currency text NOT NULL,
  paid boolean NOT NULL DEFAULT false,
  PRIMARY KEY (organization_id, invoice_id)
);

-- The host checks endpoint ownership and maintains these invoice.paid routing rules.
-- Paused endpoints retain their routes. The worker resolves current endpoint lifecycle.
CREATE TABLE transaction_example.invoice_routes (
  organization_id text NOT NULL,
  endpoint_id text NOT NULL,
  currency text NOT NULL,
  minimum_amount integer NOT NULL DEFAULT 0 CHECK (minimum_amount >= 0),
  PRIMARY KEY (organization_id, endpoint_id, currency)
);

```

## Capture recipients, then publish [#capture-recipients-then-publish]

The function updates the invoice and queries routes on the transaction client. It then builds an inert delivery instance whose `matchRecipients` returns the frozen list. Each call gets its own source, so concurrent requests cannot overwrite each other.

```ts title="examples/transactions/invoices.ts"
import type { Database, SqlClient } from 'good-webhooks'
import { createDelivery, type EndpointSource } from 'good-webhooks/delivery'
import { z } from 'zod'

export const events = {
  'invoice.paid': z.object({
    invoiceId: z.string(),
    amount: z.number().int().nonnegative(),
    currency: z.string(),
  }),
}
export const deliverySchema = 'public'

/** The host authorizes the organization and supplies a client inside BEGIN. */
export async function recordInvoicePayment(options: {
  database: Database
  pgClient: SqlClient
  source: EndpointSource
  organizationId: string
  invoiceId: string
}) {
  const { database, pgClient, source, organizationId, invoiceId } = options
  const scope = Object.freeze({ type: 'organization', id: organizationId })
  const invoice = await pgClient.query<{
    invoice_id: string
    amount: number
    currency: string
  }>(
    `UPDATE transaction_example.invoices SET paid = true
     WHERE organization_id = $1 AND invoice_id = $2
     RETURNING invoice_id, amount, currency`,
    [organizationId, invoiceId],
  )
  const row = invoice.rows[0]
  if (!row) throw new Error('Invoice not found in the authorized organization.')
  const data = { invoiceId: row.invoice_id, amount: row.amount, currency: row.currency }

  // Host-owned routes define eligible endpoints for this payload, including paused endpoints.
  // The host must check endpoint ownership when creating routes. Lookup errors must reject.
  const selected = await pgClient.query<{ endpoint_id: string }>(
    `SELECT endpoint_id FROM transaction_example.invoice_routes
     WHERE organization_id = $1 AND currency = $2 AND minimum_amount <= $3
     ORDER BY endpoint_id`,
    [organizationId, data.currency, data.amount],
  )
  const recipientIds = Object.freeze(selected.rows.map(({ endpoint_id }) => endpoint_id))
  const publicationSource: EndpointSource = {
    async matchRecipients(requestedScope, eventType) {
      if (
        requestedScope?.type !== scope.type ||
        requestedScope.id !== scope.id ||
        eventType !== 'invoice.paid'
      )
        throw new Error('This recipient selection belongs to one scope and event type.')
      return recipientIds
    },
    // Resolution still uses current management state, never this transaction's client.
    resolveEndpoint: (requestedScope, id) => source.resolveEndpoint(requestedScope, id),
  }

  // Construction opens no connections and starts no worker. Never mutate a shared source.
  const publication = createDelivery({
    database,
    schema: deliverySchema,
    events,
    source: publicationSource,
  })
  // The result is provisional until the caller commits. On retry, call this function again
  // inside a fresh transaction so it repeats the business reads and recipient selection.
  return publication
    .forScope(scope)
    .publish(
      { type: 'invoice.paid', data, idempotencyKey: `invoice.paid:${invoiceId}` },
      { transaction: pgClient },
    )
}

```

Two details are easy to miss:

* `matchRecipients()` receives only the scope and event type. The closure supplies the selection and rejects any other scope or type.
* `resolveEndpoint()` delegates to the normal source. It must keep returning current URLs, secrets, and status. Never capture the transaction client in it.

## Commit from the caller [#commit-from-the-caller]

```ts title="pay-invoice.ts"
import { recordInvoicePayment } from './invoices.js'

const pgClient = await pool.connect()
try {
  await pgClient.query('BEGIN')
  const publication = await recordInvoicePayment({
    database: pool,
    pgClient,
    source,
    organizationId: 'org_123',
    invoiceId: 'inv_123',
  })
  await pgClient.query('COMMIT')
  // Only now is the publication durable.
} catch (error) {
  await pgClient.query('ROLLBACK').catch(() => {})
  throw error
} finally {
  pgClient.release()
}
```

A later rollback removes the invoice update, the event, and its pending deliveries together. The routing query runs at your transaction's isolation level. Default `READ COMMITTED` sees your earlier writes and a per-statement snapshot. Choose stronger isolation only if the business rule needs it.

## Retrying and running the worker [#retrying-and-running-the-worker]

If the transaction aborts, retry the whole operation with a fresh transaction and a fresh call. Do not reuse a captured selection.

After a commit, repeating the same key, type, and payload returns the original event with its original recipients. The wrapper still queries routes first, so it can fail on a routing-storage outage even though the publication is already accepted. Different content with the same key throws `IDEMPOTENCY_CONFLICT`.

The worker needs only the normal source and the same events, database, and schema:

```ts title="worker.ts"
import { createDelivery } from 'good-webhooks/delivery'
import { deliverySchema, events } from './invoices.js'

const delivery = createDelivery({ database: pool, schema: deliverySchema, events, source })
await delivery.worker.run({ signal: shutdown.signal })
```

It uses the committed recipient IDs and resolves current endpoint state before each attempt. It never reruns your routing query.

> **Warning.** This recipe cannot pull a separate management database into the PostgreSQL transaction. If
> recipient authorization lives elsewhere, define how that state is synchronized or use a durable
> recovery workflow.