Documentation
¶
Overview ¶
Package webhookdelivery is Layer 3 of the webhook pipeline (docs/design/webhook-delivery-river-migration.md): the River execution stage that POSTs a webhook_subscriber_deliveries (Layer 2) row to the customer endpoint and retries it. It replaces the hand-rolled SKIP-LOCKED claim + retry worker (internal/webhook/subscriber_retry.go) with a River Worker on the shared `webhook` queue; River owns claim/lease/retry/backoff, and Layer 2 becomes pure delivery state written here for the history API.
Delivery is at-least-once (the industry standard — Stripe/Svix/Postmark): River re-drives a crashed job, so an endpoint may receive a duplicate if the POST succeeds but the worker crashes before writing `delivered`. Consumers dedup on the event id in the signed payload.
Index ¶
- Constants
- type DeliverWorker
- type Deliverer
- type Jobs
- func (j *Jobs) EnqueueDelivery(ctx context.Context, pool *pgxpool.Pool, deliveryID string) error
- func (j *Jobs) EnqueueDeliveryTx(ctx context.Context, tx pgx.Tx, deliveryID string) (int64, error)
- func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error)
- func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
- func (j *Jobs) SetEnqueuer(e jobs.Enqueuer)
- type MaintenanceJobs
- type MaintenanceWorker
- type ReconcileWorker
- type Sweeper
- type WebhookDeliverArgs
- type WebhookMaintenanceArgs
- type WebhookReader
- type WebhookReconcileArgs
Constants ¶
const MaxDeliveryAttempts = 8
MaxDeliveryAttempts is River's MaxAttempts for a delivery job — after this many failed attempts River discards and the worker marks Layer 2 'failed'.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeliverWorker ¶
type DeliverWorker struct {
river.WorkerDefaults[WebhookDeliverArgs]
// contains filtered or unexported fields
}
DeliverWorker delivers one Layer 2 row. Mirrors the legacy processOne: re-fetch the webhook per attempt (disabled → snooze, deleted → cancel), POST, write Layer 2 state. River owns retry/backoff via NextRetry below.
func NewDeliverWorker ¶
func NewDeliverWorker(subStore *webhook.SubscriberStore, deliverer Deliverer, webhooks WebhookReader) *DeliverWorker
NewDeliverWorker constructs the worker. Used by the Registrar and by tests.
func (*DeliverWorker) NextRetry ¶
func (w *DeliverWorker) NextRetry(job *river.Job[WebhookDeliverArgs]) time.Time
NextRetry overrides River's client-wide policy for webhook jobs only, returning the exact 29h21m envelope. job.Attempt is the just-failed attempt (1-based); River discards at MaxAttempts so this is called for attempts 1..MaxDeliveryAttempts-1.
func (*DeliverWorker) Work ¶
func (w *DeliverWorker) Work(ctx context.Context, job *river.Job[WebhookDeliverArgs]) error
Work intentionally has no Timeout() override — a single HTTP POST (bounded by the deliverer's own client timeout) fits River's 60s default JobTimeout.
type Deliverer ¶
type Deliverer interface {
Deliver(ctx context.Context, url string, body []byte, secret, secretPrev, eventType, schemaVersion string) webhook.DeliveryOutcome
}
Deliverer is the POST surface. *webhook.SubscriberDeliverer satisfies it.
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
Jobs is the webhook-delivery integration on the shared River client: a jobs.Registrar (contributes DeliverWorker + the reconcile periodic) plus the transactional enqueue entry point the outbox drain + redelivery API call. The shared client is injected via SetEnqueuer after jobs.New builds it (two-phase wiring, same as senderidentity).
func NewJobs ¶
func NewJobs(subStore *webhook.SubscriberStore, deliverer Deliverer, webhooks WebhookReader, pool *pgxpool.Pool) *Jobs
NewJobs builds the integration with its dependencies (no client yet). pool backs the periodic reconciler's scan.
func (*Jobs) EnqueueDelivery ¶
EnqueueDelivery enqueues a River delivery job for an ALREADY-INSERTED pending Layer 2 row, in its own transaction, and stamps job_id. This is for the direct- insert API surfaces that bypass the outbox drain — the /test webhook endpoint and the redelivery API — which create a subscriber_deliveries row targeting a single webhook. Without this, those rows have no River job and (post SubscriberRetryWorker deletion) would never deliver. Idempotent per row via the job_id IS NULL guard under a row lock.
func (*Jobs) EnqueueDeliveryTx ¶
EnqueueDeliveryTx enqueues a delivery job WITHIN the caller's transaction — the outbox pattern: the Layer 2 row insert and this job commit together, so a delivery record can never exist without a job (or vice versa). The caller only calls this when the Layer 2 insert actually inserted a row (dedup ON CONFLICT returned an id), so a deduped event enqueues nothing. Returns the river_job id for the caller to stamp on the Layer 2 row's job_id.
func (*Jobs) ReconcilePending ¶
ReconcilePending enqueues a River delivery job for every pending Layer 2 row that has no job yet (job_id IS NULL). It runs BOTH at startup (the one-shot cutover from the legacy queue) AND on a live schedule (ReconcileWorker) so a stranded row — from the separate-tx /test/redelivery enqueue paths or an outbox-drain crash window — is re-driven within reconcileInterval rather than only on the next restart. Idempotent: the per-row FOR UPDATE + job_id IS NULL guard means a re-run (or a concurrent replica) never double-enqueues. Returns the number of rows enqueued.
func (*Jobs) RegisterJobs ¶
func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the DeliverWorker + the reconcile worker, and returns the reconcile periodic. Implements jobs.Registrar.
func (*Jobs) SetEnqueuer ¶
SetEnqueuer injects the shared client so EnqueueDeliveryTx can insert jobs.
type MaintenanceJobs ¶
type MaintenanceJobs struct {
// contains filtered or unexported fields
}
MaintenanceJobs is the jobs.Registrar for the webhook janitor: it contributes the MaintenanceWorker and a periodic that fires it on QueueMaintenance. No enqueuer needed — the schedule is the only trigger.
func NewMaintenanceJobs ¶
func NewMaintenanceJobs(sweeper Sweeper) *MaintenanceJobs
NewMaintenanceJobs builds the registrar around a Sweeper (the AutoDisableWorker).
func (*MaintenanceJobs) RegisterJobs ¶
func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the maintenance worker + its periodic schedule. Mirrors senderidentity's reaper: routed to QueueMaintenance, no UniqueOpts (River's periodic scheduler already inserts at most one per interval and a completed run must not dedup-block the next), RunOnStart:false (first sweep after one interval).
type MaintenanceWorker ¶
type MaintenanceWorker struct {
river.WorkerDefaults[WebhookMaintenanceArgs]
// contains filtered or unexported fields
}
MaintenanceWorker runs the janitor passes once per scheduled job. Errors are swallowed inside Tick (logged there); Work returns nil so a transient DB blip never spins River's retry machinery for a best-effort idempotent sweep — the next interval picks it up.
func NewMaintenanceWorker ¶
func NewMaintenanceWorker(sweeper Sweeper) *MaintenanceWorker
NewMaintenanceWorker builds the worker around a Sweeper. Exported so tests can drive Work directly (RegisterJobs builds an identical one for the client).
func (*MaintenanceWorker) Work ¶
func (w *MaintenanceWorker) Work(ctx context.Context, _ *river.Job[WebhookMaintenanceArgs]) error
type ReconcileWorker ¶
type ReconcileWorker struct {
river.WorkerDefaults[WebhookReconcileArgs]
// contains filtered or unexported fields
}
ReconcileWorker re-enqueues any pending delivery row with no River job. It is the LIVE backstop for the separate-tx enqueue paths (/test, redelivery) and any outbox-drain crash window — turning "recovered only on restart" into "recovered within reconcileInterval". Idempotent (ReconcilePending's job_id IS NULL guard).
func (*ReconcileWorker) Work ¶
func (w *ReconcileWorker) Work(ctx context.Context, _ *river.Job[WebhookReconcileArgs]) error
type Sweeper ¶
Sweeper runs the webhook maintenance passes (auto-disable chronically-failing webhooks + clear expired signing_secret_prev rows). Satisfied by *webhook.AutoDisableWorker — the River worker just drives its Tick on a schedule instead of a hand-rolled time.Ticker.
type WebhookDeliverArgs ¶
type WebhookDeliverArgs struct {
DeliveryID string `json:"delivery_id"`
}
WebhookDeliverArgs carries only the Layer 2 delivery id — the worker reads the payload + webhook from the DB (keeps river_job rows tiny; Layer 2 is source of truth).
func (WebhookDeliverArgs) Kind ¶
func (WebhookDeliverArgs) Kind() string
type WebhookMaintenanceArgs ¶
type WebhookMaintenanceArgs struct{}
WebhookMaintenanceArgs drives the periodic webhook janitor. No fields — the worker sweeps the whole table each run.
func (WebhookMaintenanceArgs) Kind ¶
func (WebhookMaintenanceArgs) Kind() string
type WebhookReader ¶
type WebhookReader interface {
GetWebhookByIDInternal(ctx context.Context, webhookID string) (*identity.Webhook, error)
}
WebhookReader is the narrow webhook-lookup surface the worker needs. *identity.Store satisfies it.
type WebhookReconcileArgs ¶
type WebhookReconcileArgs struct{}
WebhookReconcileArgs drives the periodic stranded-delivery reconciler.
func (WebhookReconcileArgs) Kind ¶
func (WebhookReconcileArgs) Kind() string
Source Files