webhookpub

type DeliveryEnqueuer ¶

type DeliveryEnqueuer interface {
	EnqueueDeliveryTx(ctx context.Context, tx pgx.Tx, deliveryID string) (int64, error)
}

DeliveryEnqueuer enqueues a webhook delivery job transactionally. *webhookdelivery.Jobs satisfies it. Injected so webhookpub stays decoupled from the River-delivery package and the choice is a wiring decision.

type Envelope ¶

type Envelope struct {
	Type string `json:"type"`
	ID   string `json:"id"`
	// SchemaVersion is the envelope schema version (currently "1"), carried on the
	// wire so consumers can branch on it before parsing `data`. Sourced from the
	// SchemaVersion constant — the same value the DB column default uses.
	SchemaVersion string    `json:"schema_version"`
	CreatedAt     time.Time `json:"created_at"`
	Data          any       `json:"data"`
}

Envelope is the wire shape sent in the HTTP body to webhook subscribers. Stable across retries — the event_payload JSON column stores the envelope verbatim and the delivery worker POSTs it unchanged. Type is the event-type discriminator. Named `type` on the wire to match the /v1/events REST resource (EventView.type) and the Stripe-style SDK `construct_event` helper — a webhook delivery body is the same shape as a GET /v1/events/{id} object, so consumers handle both identically.

type Event ¶

type Event struct {
	// ID is a unique identifier for this event firing. Stable across
	// retries — receivers dedup on it.
	ID string

	// Type is one of the EventEmail* constants.
	Type string

	// CreatedAt is the time the event was published. Embedded in
	// the wire envelope so receivers can reason about staleness.
	CreatedAt time.Time

	// UserID is the owner — used to find matching webhooks. Routing
	// is strictly bounded to the owning user's subscribers; cross-
	// user delivery is impossible by construction.
	UserID string

	// AgentID, ConversationID, Labels are filter-matching keys. Each
	// is matched against the corresponding key in
	// WebhookFilters. Empty / nil here means "the event has no value
	// for this attribute" — see Publisher's null/empty semantics
	// (filters that REQUIRE a value while the event has none → skip).
	AgentID        string
	ConversationID string
	Labels         []string

	// MessageID is the originating message row, if any. May be empty
	// for events without a direct message backing (e.g.
	// email.review_requested before the held message gets promoted).
	MessageID string

	// Data is the event-specific payload. Wrapped in the envelope
	// {event, id, created_at, data} and serialized into the delivery
	// row's event_payload column.
	Data any
}

Event is the input to Publisher.Publish. Carries the routing keys (UserID, AgentID, ConversationID, Labels) needed to apply filters plus the Data payload that's serialized into the delivery row's event_payload JSONB.

MessageID is optional — set when the event has an originating message row. Persisted on the delivery row with ON DELETE SET NULL so explicit message deletion or trash purge doesn't orphan the delivery.

func NewEvent(eventType, userID string, data any) Event

func (e Event) AsEnvelope() Envelope

type FanOutArgs ¶

type FanOutArgs struct {
	EventID string `json:"event_id"`
}

FanOutArgs drives one webhook fan-out: match the event user's enabled webhooks, insert the matching webhook_subscriber_deliveries rows, and enqueue their delivery jobs. Carries only the event id — the worker re-reads the durable webhook_events row (the three-layer pattern: the job is the trigger, the row is the source of truth), so args stay tiny and a schema change to the event never invalidates enqueued jobs.

This is the River replacement for the legacy webhookpub.OutboxWorker drain (LISTEN/NOTIFY + poll + SKIP-LOCKED lease). See docs/design/webhook-fanout-river-migration.md.

func (FanOutArgs) Kind() string

type FanOutEnqueuer ¶

type FanOutEnqueuer interface {
	EnqueueFanOutTx(ctx context.Context, tx pgx.Tx, eventID string) (int64, error)
}

FanOutEnqueuer enqueues a webhook fan-out job within the caller's transaction and returns the river_job id. *FanOutJobs satisfies it. Injected (not imported) so the outbox writer stays decoupled from the River wiring — mirrors DeliveryEnqueuer.

type FanOutJobs ¶

type FanOutJobs struct {
	// contains filtered or unexported fields
}

FanOutJobs is the webhook fan-out integration on the shared River client: a jobs.Registrar (contributes FanOutWorker + the reconcile periodic) plus the transactional enqueue entry point (EnqueueFanOutTx) that PublishTx / PublishBestEffortTx call in the event's own tx. The shared client is injected via SetEnqueuer after jobs.New builds it (two-phase wiring, same as webhookdelivery).

func NewFanOutJobs(pool *pgxpool.Pool, identityStore identityReader, deliveryEnq DeliveryEnqueuer, metrics telemetry.Metrics) *FanOutJobs

func (j *FanOutJobs) EnqueueFanOutTx(ctx context.Context, tx pgx.Tx, eventID string) (int64, error)

func (j *FanOutJobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error)

func (j *FanOutJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob

func (j *FanOutJobs) SetEnqueuer(e jobs.Enqueuer)

type FanOutReconcileArgs ¶

type FanOutReconcileArgs struct{}

FanOutReconcileArgs drives the periodic stranded-event reconciler.

func (FanOutReconcileArgs) Kind() string

type FanOutReconcileWorker ¶

type FanOutReconcileWorker struct {
	river.WorkerDefaults[FanOutReconcileArgs]
	// contains filtered or unexported fields
}

FanOutReconcileWorker re-enqueues any pending event with no fan-out job. It is the LIVE backstop for the best-effort publish path (PublishBestEffortTx must not fail the caller's tx, so an event can commit with its enqueue lost) and any crash window between the event commit and the job insert — turning "recovered only on restart" into "recovered within fanOutReconcileInterval". Idempotent (fanout_job_id IS NULL guard). Mirrors webhookdelivery.ReconcileWorker.

func (w *FanOutReconcileWorker) Work(ctx context.Context, _ *river.Job[FanOutReconcileArgs]) error

type FanOutWorker ¶

type FanOutWorker struct {
	river.WorkerDefaults[FanOutArgs]
	// contains filtered or unexported fields
}

FanOutWorker fans out one webhook_events row on River, replacing the legacy OutboxWorker drain. It re-reads the event by id, skips it if it is gone (30d GC) or no longer 'pending' (a duplicate at-least-once job — already fanned out), and otherwise runs the shared fanOutEventCore. Idempotent: the (event_id, webhook_id) unique index dedups delivery-row inserts and the status='pending' guard on the terminal UPDATE makes a re-run a no-op.

func NewFanOutWorker(pool *pgxpool.Pool, identityStore identityReader, deliveryEnq DeliveryEnqueuer, metrics telemetry.Metrics) *FanOutWorker

func (w *FanOutWorker) Timeout(*river.Job[FanOutArgs]) time.Duration

func (w *FanOutWorker) Work(ctx context.Context, job *river.Job[FanOutArgs]) error

type FeatureFlag ¶

type FeatureFlag interface {
	Enabled() bool
}

FeatureFlag is a boolean gate. Retained because the outbox drain (OutboxWorker) is constructed with one — StaticFlag(true) in production, since the outbox is now unconditional.

type Outbox ¶

type Outbox interface {
	// PublishTx writes the event to webhook_events inside the
	// caller's transaction. Returns error so the caller can roll back
	// its business state if the outbox write fails.
	//
	// Used for PRE-side-effect triggers (email.received, future
	// email.bounced from SNS, email.review_requested, email.review_rejected).
	// If the outbox write fails the caller's tx rolls back; on
	// retry, the deterministic event id makes the second outbox
	// INSERT a no-op via ON CONFLICT (id) DO NOTHING.
	PublishTx(ctx context.Context, tx pgx.Tx, e Event) error

	// PublishBestEffortTx attempts the outbox write inside the
	// caller's transaction but never returns an error. On failure,
	// logs and lets the caller's tx commit anyway (a future slice may pipe these
	// to a webhook_publish_failures table — not yet built).
	//
	// Returns `wrote=true` iff the row was actually written (flag
	// enabled AND writeOutboxRow succeeded). The wrote signal is retained for
	// observability; with the outbox unconditional there is no legacy fallback
	// path left to gate.
	//
	// Used for POST-side-effect triggers (email.sent, email.review_approved)
	// where the irreversible action (SES.Send) has already happened
	// and rolling back the business state would orphan an SES
	// delivery.
	PublishBestEffortTx(ctx context.Context, tx pgx.Tx, e Event) (wrote bool)

	// DeleteExpiredWebhookEvents drops rows past their 30-day retention.
	// Called from the hourly cleanup loop in cmd/e2a/main.go.
	DeleteExpiredWebhookEvents(ctx context.Context) (int, error)

	// Enabled reports whether the outbox writes durable event rows for this
	// deployment. Now unconditional (StaticFlag(true) in production) — the events
	// API gates its list/get/redeliver endpoints on this, and the outbox writer
	// no-ops when false. Retained as a seam; there is no longer a legacy
	// fan-out path to fall back to.
	Enabled() bool

	// SetFanOutEnqueuer wires the River fan-out enqueuer (two-phase, called after the
	// shared client exists). When set (E2A_WEBHOOK_FANOUT_MODE=river), PublishTx /
	// PublishBestEffortTx enqueue a webhook_fanout job in the event's own tx and stamp
	// fanout_job_id — the River FanOutWorker then does the fan-out, replacing the
	// legacy in-process OutboxWorker drain. nil (default/legacy) ⇒ the event row is
	// written as before and the OutboxWorker fans it out via LISTEN/NOTIFY + poll.
	SetFanOutEnqueuer(e FanOutEnqueuer)
}

Outbox is the Stripe-tier publisher entry point. Triggers write the webhook_events row inside the same transaction as their business state (e.g. the messages row for email.received). A separate publisher worker (slice 2) consumes pending rows and fans out into webhook_subscriber_deliveries.

Outbox.PublishTx writes ONE row to webhook_events inside the trigger's transaction and arranges NOTIFY; the OutboxWorker (slice 2) fans it out into webhook_subscriber_deliveries and enqueues a River delivery job per row. This is now the sole event path — the legacy in-process fan-out publisher is retired and River is the sole delivery engine.

See docs/design/2026-06-01-stripe-tier-webhooks.md §4.2 and Appendix A.

func NewOutbox(pool *pgxpool.Pool, flag FeatureFlag) Outbox

type OutboxPublisher ¶

type OutboxPublisher struct {
	// contains filtered or unexported fields
}

OutboxPublisher is the sole Publisher: it writes events to the durable outbox (webhook_events) in their own transaction. Used for post-side-effect standalone events that have no business tx to join — domain.sending_* (senderidentity), email.delivered/bounced/complained + domain.suppression_added (SNS delivery feedback), and hitlworker TTL-resolution events. Routing these through the outbox is what closes the webhook-delivery→River gap: those sources used to bypass webhook_events via the legacy publisher, so their delivery rows got no River job and were never delivered. Now they fan out + enqueue like every other event (via the outbox drain). Best-effort by the Publisher contract (fire-after-the-fact): a write failure is logged.

func NewOutboxPublisher(outbox Outbox, pool *pgxpool.Pool) *OutboxPublisher

func (p *OutboxPublisher) Publish(ctx context.Context, e Event)

type OutboxWorker ¶

type OutboxWorker struct {
	// contains filtered or unexported fields
}

OutboxWorker drains webhook_events. For each pending row it reads enabled webhooks for the user, applies filter matching in Go, and inserts one webhook_subscriber_deliveries row per match, enqueuing a River delivery job for each in the same transaction. The River DeliverWorker then POSTs each row to the customer endpoint. See design §4.4 for the full architecture.

Wakeup paths:

1. LISTEN webhook_events_new — dedicated connection, sub-50ms latency from trigger COMMIT to fan-out.
2. 1s fallback poll — catches notifications missed during deploy or LISTEN reconnect.

Multi-replica safety:

• FOR UPDATE SKIP LOCKED + next_poll_at bump in GetPending leases rows so concurrent replicas don't fan out the same event.
• The per-row partial unique index on (event_id, webhook_id) in webhook_subscriber_deliveries is the backstop for lease-expiry races: a slow worker B can finish after a fast worker A's lease has expired and a new worker C has taken over; per-row ON CONFLICT DO NOTHING swallows the duplicates rather than aborting B's transaction.
• The final UPDATE outbox row uses WHERE status='pending' so a stale-and-late worker can't overwrite a fast-finisher's matched_webhook_ids snapshot.

func NewOutboxWorker(pool *pgxpool.Pool, store identityReader) *OutboxWorker

func (w *OutboxWorker) Start(ctx context.Context)

func (w *OutboxWorker) Tick(ctx context.Context)

func (w *OutboxWorker) WithDeliveryEnqueuer(e DeliveryEnqueuer) *OutboxWorker

func (w *OutboxWorker) WithMetrics(m telemetry.Metrics) *OutboxWorker

type Publisher ¶

type Publisher interface {
	Publish(ctx context.Context, e Event)
}

Publisher fires an Event at the webhook pipeline. There is exactly one implementation left — OutboxPublisher — which writes the event to the durable outbox (webhook_events) in its own transaction; the OutboxWorker drain then fans it out to matching subscribers (webhook_subscriber_deliveries) and the River DeliverWorker POSTs each row. The legacy in-process fan-out publisher (direct delivery-row inserts, drained by the hand-rolled SubscriberRetryWorker) is retired: River is the sole delivery engine and every event flows webhook_events → drain → River delivery.

Best-effort by contract (fire-after-the-fact): a write failure is logged, not returned, so a trigger's primary state change is never rolled back on a publish error. Callers that need atomicity join the message transaction via Outbox.PublishTx directly instead of going through this interface.

type StaticFlag ¶

type StaticFlag bool

StaticFlag is a trivially-implemented FeatureFlag for tests and for production wiring that reads an env var at startup.

func (f StaticFlag) Enabled() bool