Webhooks

@voltro/plugin-webhooks adds first-class outgoing and incoming webhooks. Both directions are declared in *.webhook.tsx files: outgoing events the app emits, and incoming HTTP endpoints the app exposes to third parties. Delivery is durable — each outgoing delivery runs as an @effect/workflow, so signing, retries, rate limits, and Retry-After honouring all happen inside a crash-safe workflow body.

The framework discovers *.webhook.tsx files and wires the queries + codegen. The managed tables (_voltro_webhook_targets, _voltro_webhook_deliveries, and the rate-limit counter table _voltro_webhook_rate_windows) are created by the bootstrap migration; you can register them explicitly in your schema via the webhookTables() mixin from @voltro/plugin-webhooks/mixin.

The targets + deliveries tables carry tenant() (the rate-window table holds only operational counters — no tenant data), so webhook subscriptions are per-tenant: a target subscribed under tenant A is invisible to tenant B, and emit() from a tenant fans out only to that tenant's targets. The runtime stamps tenantId from the caller's subject on subscribe and carries it through to delivery — you don't pass it by hand.

#Outgoing — defineOutgoingEvent(...)

Declare an event the app emits. External systems subscribe to it at runtime via ctx.webhooks.subscribe(...). Each subscribed target is a row in _voltro_webhook_targets; each delivery attempt is a row in _voltro_webhook_deliveries.

// events/order.completed.webhook.tsx
import { defineOutgoingEvent } from '@voltro/plugin-webhooks'
import { Schema } from 'effect'

export default defineOutgoingEvent({
  id: 'order.completed',
  description: 'Order moved into "fulfilled" — payment captured, items shipped.',
  payload: Schema.Struct({
    orderId:  Schema.String,
    tenantId: Schema.String,
    total:    Schema.Number,
    items:    Schema.Array(Schema.Struct({ sku: Schema.String, qty: Schema.Number })),
  }),
  version: 1,
})

Recipients receive { event, eventId, occurredAt, payload } with payload matching the schema — emit() decodes every payload against it and rejects a mismatch with the typed WebhookPayloadInvalid before any delivery is created. version defaults to 1; increment it whenever the payload shape changes in a way subscribers must adapt to — new subscriptions pin to the event's current version. Optional defaultRetry and defaultSigning apply to new subscriptions when the subscriber doesn't pass retry / signing explicitly (explicit input wins, then the event default, then the package default). Optional globalRateLimit: { perMinute } caps ALL deliveries of this event combined: the delivery workflow claims a slot in a fixed one-minute window counter at the shared store (so the cap holds across replicas) before every POST; over-limit deliveries are deferred — parked as status='pending' rows and durable-slept until the next window opens — never dropped.

#Subscribing targets

Subscribers register at runtime — from a settings page, an admin mutation, or a seed. Targets persist; the dashboard surfaces them under a "Webhooks" tab.

// mutations/webhooks.subscribe.mutation.server.ts
const execute = async (input: { url: string; secret?: string }, ctx: AppContext) => {
  const target = await ctx.webhooks.subscribe({
    event: 'order.completed',
    url:   input.url,
    // Optional — the server generates a 32-byte hex secret if absent.
    // The caller receives it ONCE in the response; subsequent reads
    // never expose it again.
    secret: input.secret,
    retry: {                                  // overrides the event's defaultRetry
      strategy:     'exponential',
      maxAttempts:  8,
      initialDelay: '5s',
      maxDelay:     '1h',
      retryOn:      [408, 429, 500, 502, 503, 504],
      jitter:       'full',
    },
    filter:  { 'payload.total': { gt: 1000 } }, // optional predicate filter
    headers: { 'X-Tenant': 'acme' },            // optional per-target headers
    rateLimitPerMinute: 60,                     // optional — excess deliveries defer to the next minute window
    format:  'json',                            // optional wire format: 'json' (default) | 'form' | 'xml'
    autoDisableAfter: 20,                       // optional — auto-pause after N consecutive terminal failures (off by default)
  })
  return { targetId: target.id, secret: target.secret }
}

rateLimitPerMinute caps wire POSTs to this target (retries count too): the delivery workflow claims a fixed-window slot at the shared store before each POST and defers over-limit deliveries to the next window — queued as status='pending' rows, never dropped.

Delivery format — json (default) · form · xml

format decides the bytes the delivery workflow signs and POSTs. The producer side always stringifies the payload to JSON; the workflow re-encodes it into the target's format at delivery time and signs the re-encoded bytes (never the source JSON). If a format can't represent the payload shape the delivery is marked failed with a typed WebhookPayloadUnrepresentable reason before any wire POST — a mis-formatted target never silently ships JSON under a wrong content-type.

• json (default) — content-type: application/json, the JSON string verbatim. Represents any JSON value.
• form — content-type: application/x-www-form-urlencoded. The payload MUST be a JSON object (a top-level array/scalar is unrepresentable). Nested structure flattens to bracketed keys (the PHP/Rails convention): { a: 1, b: { c: 2 }, items: [{ sku: 'x' }, 'y'] } → a=1&b[c]=2&items[0][sku]=x&items[1]=y. Leaf primitives become their string form; null becomes an empty value (key=); keys/values are percent-encoded.
• xml — content-type: application/xml. The payload is wrapped in a single <webhook> root: object keys → child elements (<key>…</key>), arrays → repeated <item>…</item> elements under the array's key, primitives → escaped text, null → an empty element. Prefixed with the XML declaration. An object key that isn't a valid XML element name is unrepresentable.

Auto-disable — dead-letter guard for failing endpoints

autoDisableAfter (off by default; a positive integer when set) auto-pauses a target after that many consecutive terminal delivery failures. The delivery workflow tracks a per-target streak at the shared store (multi-replica-correct): each terminal failed delivery increments it, any succeeded resets it to zero. When the streak reaches the threshold the target flips to active:false (stamped with autoDisabledAt + the failure reason, surfaced on the inspect panel) and a warn log fires.

Auto-disable reuses the same pause machinery as a manual pauseTarget: while disabled, subsequent emits queue as status='pending' rows (nothing is dropped — see the pause/flush section below). A manual resumeTarget re-activates the target, flushes the accumulated queue, and clears the streak so the next failure starts a fresh count.

#Emitting

Emit from any handler that imports the plugin:

import { useWebhooks } from '@voltro/plugin-webhooks'

const execute = async (input: { orderId: string }, ctx: AppContext) => {
  // ... do the mutation ...
  const webhooks = useWebhooks(ctx)
  await webhooks.emit('order.completed', {
    orderId:  newOrder.id,
    tenantId: newOrder.tenantId,
    total:    newOrder.total,
    items:    newOrder.items,
  })
}

useWebhooks(ctx) returns the typed WebhooksServiceShape the framework attached to AppContext; it throws a clear message if the plugin isn't active. emit() accepts the event id or the defineOutgoingEvent descriptor itself — emit(orderCompleted, payload) types payload against the event schema at the call site. Either way the payload is decoded against the schema and a mismatch throws WebhookPayloadInvalid. It returns { eventId, deliveries } — one entry per matched target, each 'dispatched' (workflow kicked off) or 'queued' (the target is paused — see below), for log correlation. The actual HTTP POSTs happen in the background delivery workflow; the call returns as soon as the workflows are kicked off, not when they complete.

The service also exposes replay(deliveryId), listTargets(event?), pauseTarget / resumeTarget, deleteTarget, rotateSecret, and updateTargetPayloadVersion — consumed by the dashboard's Webhooks panel.

While a target is paused — whether by a manual pauseTarget or by the autoDisableAfter dead-letter guard above — emits against it queue as status='pending' rows in _voltro_webhook_deliveries instead of POSTing — nothing is dropped. resumeTarget re-activates the target and flushes its queue through the normal delivery workflow in emit order (createdAt ascending, millisecond granularity) per target, and clears any auto-disable streak. deleteTarget removes the target's queued pending rows along with it. An auto-disabled target is distinguishable from a manually-paused one by its non-null autoDisabledAt / autoDisableReason.

#Typed errors — @voltro/plugin-webhooks/errors

The outgoing WebhooksService ops throw Schema.TaggedErrors, not bare Errors — catch them with Effect.catchTag server-side, or declare them on a mutation/action descriptor's error: union so the client decodes them typed:

• WebhookSubscribeInvalid ({ field, reason }) — subscribe input failed a policy check (a non-http(s) URL, a secret below the minimum length, a degenerate retry policy, or an out-of-range rate limit).
• WebhookDeliveryNotFound ({ deliveryId }) — replay found no delivery row for that id.
• WebhookPayloadVersionInvalid ({ version }) — updateTargetPayloadVersion got a non-positive-integer version.
• WebhookPayloadInvalid ({ event, issues }) — emit got a payload that doesn't decode against the event's payload schema; rejected before any delivery row or workflow run exists.
• WebhookPayloadUnrepresentable ({ format, reason }) — the target's form/xml encoder can't represent the payload shape (e.g. a form target whose payload is a top-level array/scalar, or an xml object key that isn't a valid element name). The delivery is marked failed with this reason before any wire POST.

Import these from the browser-safe @voltro/plugin-webhooks/errors subpath when a descriptor (*.mutation.ts / *.action.ts) names one in its error: schema. The package root re-exports them too but is server-only (it also re-exports node:crypto / @voltro/database modules), so a descriptor importing from the root would pull server code into the browser bundle — the same rule as @voltro/plugin-multitenancy/guard.

// mutations/webhooks.subscribe.mutation.ts — DESCRIPTOR (browser-safe)
import { defineMutation } from '@voltro/protocol'
import { WebhookSubscribeInvalid } from '@voltro/plugin-webhooks/errors'
import { Schema } from 'effect'

export const subscribeWebhook = defineMutation({
  name:   'webhooks.subscribe',
  input:  Schema.Struct({ url: Schema.String }),
  output: Schema.Struct({ targetId: Schema.String }),
  error:  WebhookSubscribeInvalid,
})

#Incoming — defineIncomingWebhook(...)

Declare a public HTTP endpoint the app exposes to receive webhooks from a third party (Stripe, GitHub, Slack, a custom partner). The framework's incoming middleware verifies the signature, decodes the body, claims an idempotency key, and only then calls your handler.

// webhooks/stripe.webhook.tsx
import { defineIncomingWebhook } from '@voltro/plugin-webhooks'
import { stripeProvider } from '@voltro/plugin-webhooks/providers'
import { Schema } from 'effect'

export default defineIncomingWebhook({
  id:       'stripe',
  provider: stripeProvider(),
  payload:  Schema.Struct({
    id:   Schema.String,
    type: Schema.String,
    // Stripe payloads vary by event type — narrow per handler via ctx.body.type.
    data: Schema.Any,
  }),
  handler: async (ctx) => {
    // ctx.body is the validated payload.
    // ctx.idempotencyKey is the Stripe event id (from the provider preset).
    // ctx.headers carries the original request headers (lowercased keys).
    // ctx.rawBody is the raw bytes, if you need to recompute a MAC.
    if (ctx.body.type === 'invoice.payment_succeeded') {
      await store.update('invoices', ctx.body.data.object.metadata.invoiceId, { paidAt: new Date() })
    }
  },
})

The endpoint mounts at /webhooks/<id> by default; override with path: '/integrations/stripe/v1'. The signing secret comes from the env var VOLTRO_WEBHOOK_SECRET_<UPPER_ID> — for id: 'stripe', that's VOLTRO_WEBHOOK_SECRET_STRIPE (non-alphanumeric characters in the id become _).

#Provider presets

Built-in presets configure the signature scheme, idempotency extraction, and body type for the common integrations. Import them from @voltro/plugin-webhooks/providers:

Preset	Signature header	Idempotency key
stripeProvider()	Stripe-Signature	body.id
githubProvider()	X-Hub-Signature-256	X-GitHub-Delivery header
slackProvider()	X-Slack-Signature	X-Slack-Request-Timestamp
genericProvider()	X-Webhook-Signature	Idempotency-Key header

Explicit fields on defineIncomingWebhook always override the preset.

#Custom providers — defineWebhookProvider(...)

For partners not covered by the built-ins, define your own preset in a *.webhook.tsx file:

import { defineWebhookProvider } from '@voltro/plugin-webhooks'

export default defineWebhookProvider({
  id:   'acme.custom',
  name: 'Acme Custom',
  signature: {
    _tag: 'hmac', algorithm: 'hmacSha256', header: 'X-Acme-Sig',
    includeTimestamp: true, replayWindowSeconds: 600,
    encoding: 'hex', versionPrefix: 'v1=',
  },
  idempotency: { from: 'x-acme-delivery-id', ttl: '7d' },
  bodyType: 'json',
})

Then reference it from another *.webhook.tsx:

import { defineIncomingWebhook } from '@voltro/plugin-webhooks'
import acmeProvider from '../providers/acme.webhook'
import { Schema } from 'effect'

export default defineIncomingWebhook({
  id:       'acme',
  provider: acmeProvider,
  payload:  Schema.Struct({ id: Schema.String }),
  handler:  async (ctx) => { /* ... */ },
})

#Failure semantics

• Signature mismatch → 401, handler never runs.
• Schema validation failure → 422, handler never runs.
• Duplicate idempotency key (within TTL) → 200 + { duplicate: true }, handler never runs.
• Concurrent in-flight delivery with the same idempotency key → 409, the provider retries later.
• Handler exception → 500, the idempotency claim is released so the provider's retry can re-process. Common providers (Stripe, GitHub, Slack) retry 5xx automatically.

#Anti-patterns

• Inventing an events.emit accessor or a plugin-list entry to enable webhooks. Neither exists. Emit via useWebhooks(ctx).emit(...); declare events and endpoints in *.webhook.tsx files (the framework discovers them).
• Trusting webhook input without signature verification. The framework rejects unsigned incoming requests by default. If you genuinely need to accept unsigned traffic, set signature: undefined explicitly and gate the query at the network boundary (IP allow-list, VPC peering).
• Mutating domain state synchronously in the outgoing emit path. Outbound delivery is async by design. Mutate state in your mutation, THEN emit.
• Using the same idempotency-key TTL as the provider's retry window. Pick a TTL of at least 2× the provider's max retry window so the dedup catches the slowest retry.