webhooks

Documentation

Overview

Package webhooks delivers signed event notifications to registered webhook endpoints. It renders payloads, signs them in the Standard Webhooks format, and runs the delivery worker that drains the webhook job queue with at-least-once semantics.

Index

• func AsyncResultPath(requestType schemas.RequestType, jobID string) (string, bool)
• func EventForJobStatus(status schemas.AsyncJobStatus) (tables.WebhookEvent, bool)
• func Sign(secret, webhookID string, timestamp time.Time, body []byte) (string, error)
• type ConfigStore
• type Dispatcher
  • func NewDispatcher(ctx context.Context, runnerID string, historyRetention time.Duration, ...) *Dispatcher
  • func (d *Dispatcher) DeliverTest(ctx context.Context, endpoint *tables.TableWebhookEndpoint, ...) (int, error)
  • func (d *Dispatcher) EnqueueJobEvent(ctx context.Context, job *logstore.AsyncJob)
  • func (d *Dispatcher) Start()
  • func (d *Dispatcher) Stop()
  • func (d *Dispatcher) Wake()
• type EndpointResolver
• type LogStore

Functions

func AsyncResultPath

func AsyncResultPath(requestType schemas.RequestType, jobID string) (string, bool)

AsyncResultPath returns the relative URL a caller can GET to fetch the result of the given job, or false when the request type has no async result route.

func EventForJobStatus

func EventForJobStatus(status schemas.AsyncJobStatus) (tables.WebhookEvent, bool)

EventForJobStatus maps a terminal async job status to the webhook event it fires; non-terminal statuses return false.

func Sign

func Sign(secret, webhookID string, timestamp time.Time, body []byte) (string, error)

Sign computes the Standard Webhooks signature for one delivery attempt: HMAC-SHA256 over "{webhookID}.{unix timestamp}.{body}" keyed with the base64-decoded endpoint secret, returned in the "v1,<base64>" wire form carried by the webhook-signature header.

Types

type ConfigStore

type ConfigStore interface {
	CreateWebhookJob(ctx context.Context, job *tables.TableWebhookJob) error
	ListDueWebhookJobs(ctx context.Context, limit int) ([]tables.TableWebhookJob, error)
	ClaimWebhookJob(ctx context.Context, id, runnerID string, leaseUntil time.Time) (bool, error)
	RescheduleWebhookJob(ctx context.Context, id, runnerID string, leaseUntil, nextAttemptAt time.Time) error
	DeleteWebhookJob(ctx context.Context, id, runnerID string, leaseUntil time.Time) error
	RecordWebhookEndpointSuccess(ctx context.Context, id string) error
	RecordWebhookEndpointFailure(ctx context.Context, id string) (int, error)
}

ConfigStore is the subset of the configstore the dispatcher needs: the webhook job queue plus the endpoint operational counters.

type Dispatcher

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

Dispatcher drains the webhook job queue: it claims due jobs, performs one signed delivery attempt per claim, records history, and reschedules or retires the job. One Dispatcher runs per node; the atomic claim decides a single owner per attempt, so running it everywhere needs no coordination.

func NewDispatcher

func NewDispatcher(ctx context.Context, runnerID string, historyRetention time.Duration, configStore ConfigStore, logStore LogStore, resolver EndpointResolver, logger schemas.Logger) *Dispatcher

NewDispatcher builds a stopped dispatcher; call Start to launch its worker. The dispatcher's lifetime is bounded by ctx as well as by Stop. runnerID fences queue claims — the node id in a cluster, empty on a single node. historyRetention sets expires_at on delivery history rows and must be positive. Delivery tuning (retries, backoff, timeouts, payload caps, concurrency) is per endpoint.

func (*Dispatcher) DeliverTest

func (d *Dispatcher) DeliverTest(ctx context.Context, endpoint *tables.TableWebhookEndpoint, event tables.WebhookEvent) (int, error)

DeliverTest sends one signed sample of the given event through the exact production delivery path — same rendering, signing, client, and redirect/TLS policy — without touching the queue, history, or failure counters. It returns the receiver's status code.

func (*Dispatcher) EnqueueJobEvent

func (d *Dispatcher) EnqueueJobEvent(ctx context.Context, job *logstore.AsyncJob)

EnqueueJobEvent queues the webhook delivery for a job that just reached a terminal state. Callers invoke it right after the terminal job update; it performs no receiver I/O — just the queue insert plus a worker wake-up.

func (*Dispatcher) Start

func (d *Dispatcher) Start()

Start launches the queue worker. The first scan runs immediately, which is also what recovers deliveries left claimed or due by a previous process.

func (*Dispatcher) Stop

func (d *Dispatcher) Stop()

Stop cancels the worker and waits for in-flight deliveries to finish.

func (*Dispatcher) Wake

func (d *Dispatcher) Wake()

Wake nudges the worker to scan the queue now instead of waiting for the next poll tick — non-blocking; a full signal buffer means a scan is already pending. Callers use it after inserting queue rows outside EnqueueJobEvent (e.g. redelivery).

type EndpointResolver

type EndpointResolver interface {
	WebhookEndpointByID(id string) (*tables.TableWebhookEndpoint, bool)
}

EndpointResolver serves webhook endpoint configuration from memory. It is consulted on every enqueue and every delivery attempt, so changes to an endpoint (disable, delete, secret rotation) take effect on the next attempt without a database read.

type LogStore

type LogStore interface {
	FindAsyncJobByID(ctx context.Context, id string) (*logstore.AsyncJob, error)
	CreateWebhookDelivery(ctx context.Context, delivery *logstore.WebhookDelivery) error
}

LogStore is the subset of the logstore the dispatcher needs: reading the async job at send time and appending per-attempt delivery history.