Documentation
¶
Overview ¶
Package webhook registers outbound endpoints, fans application events out to the subscribers of each event name, and records every delivery attempt. Deliveries are signed (HMAC-SHA256 over a timestamped canonical body) and retried by the queue, so the HTTP path never blocks on a slow receiver.
The module owns the webhook surface and its delivery records.
Index ¶
- Constants
- Variables
- func CanonicalPayload(payload map[string]any) ([]byte, error)
- func SignatureHeaderValue(timestamp string, body []byte, secret string) string
- func VerifySignature(header string, body []byte, secret string, now time.Time) error
- type AttemptRecord
- type AttemptResult
- type CreateParams
- type Delivery
- type DeliveryID
- type Executor
- type ListParams
- type Module
- type Option
- type OutboundDelivery
- type Page
- type PendingDelivery
- type PostgresStore
- func (s *PostgresStore) Create(ctx context.Context, params CreateParams, encryptedSecret string) (Webhook, error)
- func (s *PostgresStore) Delete(ctx context.Context, id WebhookID) error
- func (s *PostgresStore) DeliveryForSend(ctx context.Context, id DeliveryID) (PendingDelivery, error)
- func (s *PostgresStore) Get(ctx context.Context, id WebhookID) (Webhook, error)
- func (s *PostgresStore) InsertDelivery(ctx context.Context, exec Executor, delivery *Delivery, body []byte) error
- func (s *PostgresStore) List(ctx context.Context, params ListParams) ([]Webhook, int, error)
- func (s *PostgresStore) ListAttempts(ctx context.Context, id DeliveryID) ([]AttemptRecord, error)
- func (s *PostgresStore) ListDeliveries(ctx context.Context, params ListParams, webhookID *WebhookID) ([]Delivery, int, error)
- func (s *PostgresStore) PruneDeliveries(ctx context.Context, before time.Time) (int64, error)
- func (s *PostgresStore) RecordAttempt(ctx context.Context, id DeliveryID, result AttemptResult) error
- func (s *PostgresStore) Secret(ctx context.Context, id WebhookID) (string, error)
- func (s *PostgresStore) SetSecret(ctx context.Context, id WebhookID, encryptedSecret string) error
- func (s *PostgresStore) Subscribers(ctx context.Context, event string) ([]Webhook, error)
- func (s *PostgresStore) Update(ctx context.Context, id WebhookID, params UpdateParams) (Webhook, error)
- type Sender
- type Service
- func (s *Service) Create(ctx context.Context, params CreateParams) (Webhook, error)
- func (s *Service) Delete(ctx context.Context, id WebhookID) error
- func (s *Service) Deliver(ctx context.Context, task WebhookDeliveryTask) error
- func (s *Service) DeliverTo(ctx context.Context, id WebhookID, event string, payload map[string]any) (DeliveryID, error)
- func (s *Service) Emit(ctx context.Context, event string, payload map[string]any) error
- func (s *Service) Get(ctx context.Context, id WebhookID) (Webhook, error)
- func (s *Service) List(ctx context.Context, params ListParams) ([]Webhook, int, error)
- func (s *Service) ListDeliveries(ctx context.Context, id *WebhookID, params ListParams) ([]Delivery, int, error)
- func (*Service) Name() string
- func (s *Service) RegisterQueue(client *queue.Client)
- func (s *Service) RotateSecret(ctx context.Context, id WebhookID) (string, error)
- func (s *Service) Update(ctx context.Context, id WebhookID, params UpdateParams) (Webhook, error)
- type ServiceOption
- type Store
- type UpdateParams
- type Webhook
- type WebhookDeliveryTask
- type WebhookID
Constants ¶
const ( // WebhookQueue is the registry key of the delivery queue; it must // stay stable across releases. WebhookQueue = "webhook" // WebhookMaxAttempts is the retry budget for one signed delivery. WebhookMaxAttempts = 5 // WebhookTimeout bounds the outbound request; the spec fixes 30s, // the extra 10s absorbs connection setup and TLS. WebhookTimeout = 40 * time.Second // WebhookRequestTimeout is the receiver-facing deadline. WebhookRequestTimeout = 30 * time.Second // WebhookBackoff is the wait between delivery attempts. WebhookBackoff = 30 * time.Second // WebhookRetention keeps completed delivery tasks for a week so // failed payloads stay inspectable; the delivery row is the // primary record and is pruned separately. WebhookRetention = 7 * 24 * time.Hour )
Delivery queue tuning. Webhook attempts are bounded by the queue's own backoff, so the values here stay deliberately small: a failing receiver must not stall the worker pool for minutes.
const ( // SignatureHeader carries the signed delivery timestamp + digest. SignatureHeader = "X-Signature" // EventHeader names the event that triggered the delivery. EventHeader = "X-Webhook-Event" // DeliveryHeader identifies the endpoint the body was sent for. DeliveryHeader = "X-Webhook-Id" // SignatureVersion prefixes the digest inside the header. SignatureVersion = "v1" )
Signature header shape, mirroring the widely deployed "t=timestamp,v1=hexdigest" scheme: the timestamp binds the signature to a moment so a captured body cannot be replayed later.
const AllEvents = "*"
Subscription wildcard: accepting it means "every event".
const ModuleName = "webhook"
ModuleName identifies the webhook module in the registry.
Variables ¶
var ( // ErrNotFound covers unknown endpoints and delivery logs. ErrNotFound = errors.New("webhook: not found") // ErrDuplicateName is a unique-constraint violation on name. ErrDuplicateName = errors.New("webhook: name already exists") // ErrDisabled reports a delivery against a disabled endpoint. ErrDisabled = errors.New("webhook: endpoint is disabled") // ErrTooLarge rejects an oversized event payload. ErrTooLarge = errors.New("webhook: payload exceeds the size limit") )
Errors surfaced to handlers; statuses live on the sentinels.
Functions ¶
func CanonicalPayload ¶
CanonicalPayload encodes an event payload for signing. Deterministic ordering makes the bytes reproducible for the same logical payload — the receiver can verify over exactly what it received. encoding/json/v2 does not sort map keys by default, so the option is required here.
func SignatureHeaderValue ¶
SignatureHeaderValue builds "t=<unix>,v1=<hex hmac>". The signed message is "<unix>.<body>": the timestamp is authenticated without being re-serialized, and short hex avoids a signed-digit attack.
Types ¶
type AttemptRecord ¶
type AttemptRecord struct {
Number int `json:"attempt_number"`
HTTPStatus *int `json:"response_status,omitzero"`
Error *string `json:"error,omitzero"`
DurationMs *int `json:"duration_ms,omitzero"`
Response map[string]any `json:"response,omitzero"`
Succeeded bool `json:"succeeded"`
CreatedAt time.Time `json:"created_at"`
}
AttemptRecord is one webhook_delivery_attempts row as exposed to the listing: the outcome of a single delivery try.
type AttemptResult ¶
type AttemptResult struct {
HTTPStatus int
Request map[string]any
Response map[string]any
Duration time.Duration
Err error
Succeeded bool
}
AttemptResult carries one delivery outcome back into the delivery row and its attempt record: what was sent (Request) and what came back (Response).
type CreateParams ¶
type CreateParams struct {
Name string `json:"name"`
Description *string `json:"description,omitzero"`
Endpoint string `json:"endpoint"`
Method string `json:"method,omitzero"`
Headers map[string]string `json:"headers,omitzero"`
Enabled *bool `json:"enabled,omitzero"`
EventTypes []string `json:"event_types,omitzero"`
}
CreateParams carries the POST body fields.
func (*CreateParams) Validate ¶
func (p *CreateParams) Validate() error
Validate checks the endpoint name and shape.
type Delivery ¶
type Delivery struct {
ID DeliveryID `json:"id"`
WebhookID *WebhookID `json:"webhook_id,omitzero"`
Event *string `json:"event,omitzero"`
HTTPStatus *int `json:"http_status,omitzero"`
Response map[string]any `json:"response,omitzero"`
Attempts int `json:"attempts"`
Succeeded bool `json:"succeeded"`
Error *string `json:"error,omitzero"`
CreatedAt time.Time `json:"created_at"`
DeliveredAt *time.Time `json:"delivered_at,omitzero"`
}
Delivery is one webhook_deliveries row: the immutable canonical body plus the delivery state. The latest attempt's outcome (status, error, redacted response) rides along for the listing view.
type DeliveryID ¶
DeliveryID identifies a webhook_deliveries row (one delivery).
type Executor ¶
Executor is the query surface accepted for outbox writes: the pool (autocommit) or an open transaction, so the log row and the enqueued delivery records commit together.
type ListParams ¶
ListParams narrows and pages the endpoint listing.
type Module ¶
type Module struct {
// contains filtered or unexported fields
}
Module is the webhook feature: endpoint CRUD, delivery logs, and the outbox event sink. The whole surface serves ConnectRPC exclusively (see handler_rpc.go) — the composition root wraps the mount with the admin guard.
func (*Module) Emit ¶
Emit fans an application event out to its subscribers. It is the event sink the composition root hands to other modules, so a domain package never imports this one directly.
func (*Module) RPCService ¶ added in v0.0.12
RPCService returns the Connect registration for the webhook surface.
type OutboundDelivery ¶
type OutboundDelivery struct {
// URL is the registered endpoint.
URL string
// Method is the registered HTTP verb.
Method string
// Headers are the registered custom headers plus the signature set.
Headers map[string]string
// Body is the canonical JSON encoding of the event payload.
Body []byte
}
Delivery is one signed webhook request, already rendered.
type Page ¶
Page is the store-level paging window: plain ints with no HTTP dependency. The handler converts the request query into it.
type PendingDelivery ¶
type PendingDelivery struct {
ID DeliveryID
Event string
Body []byte
}
PendingDelivery is the send-ready view of one delivery row: the committed event name and the exact canonical body bytes.
type PostgresStore ¶
type PostgresStore struct {
// contains filtered or unexported fields
}
PostgresStore persists endpoints, deliveries, and attempts.
func NewPostgresStore ¶
func NewPostgresStore(exec datastore.Executor) *PostgresStore
NewPostgresStore builds the production store over the shared pool.
func (*PostgresStore) Create ¶
func (s *PostgresStore) Create(ctx context.Context, params CreateParams, encryptedSecret string) (Webhook, error)
Create inserts an endpoint with its already-encrypted secret.
func (*PostgresStore) Delete ¶
func (s *PostgresStore) Delete(ctx context.Context, id WebhookID) error
Delete removes an endpoint; its delivery logs stay (FK sets the webhook_id to NULL) so the audit trail survives.
func (*PostgresStore) DeliveryForSend ¶
func (s *PostgresStore) DeliveryForSend(ctx context.Context, id DeliveryID) (PendingDelivery, error)
DeliveryForSend loads the immutable delivery bytes and event name for one attempt. The body is returned exactly as committed.
func (*PostgresStore) InsertDelivery ¶
func (s *PostgresStore) InsertDelivery(ctx context.Context, exec Executor, delivery *Delivery, body []byte) error
InsertDelivery writes the outbox record through exec (pool or caller tx). The canonical body bytes are stored verbatim: every attempt signs and sends exactly these bytes.
func (*PostgresStore) List ¶
func (s *PostgresStore) List(ctx context.Context, params ListParams) ([]Webhook, int, error)
List pages the endpoints, newest first.
func (*PostgresStore) ListAttempts ¶
func (s *PostgresStore) ListAttempts(ctx context.Context, id DeliveryID) ([]AttemptRecord, error)
ListAttempts returns the attempt history of one delivery in attempt order.
func (*PostgresStore) ListDeliveries ¶
func (s *PostgresStore) ListDeliveries(ctx context.Context, params ListParams, webhookID *WebhookID) ([]Delivery, int, error)
ListDeliveries pages the delivery logs, newest first, optionally scoped to one endpoint.
func (*PostgresStore) PruneDeliveries ¶
PruneDeliveries deletes deliveries recorded before the cutoff; their attempt rows cascade.
func (*PostgresStore) RecordAttempt ¶
func (s *PostgresStore) RecordAttempt(ctx context.Context, id DeliveryID, result AttemptResult) error
RecordAttempt appends one attempt row (response status, error, duration, redacted response) and folds the outcome into the delivery row.
func (*PostgresStore) Subscribers ¶
Subscribers returns the enabled endpoints for an event: an empty subscription list or the wildcard receives everything. The match is an array overlap (&&), the operator the GIN index on event_types serves.
func (*PostgresStore) Update ¶
func (s *PostgresStore) Update(ctx context.Context, id WebhookID, params UpdateParams) (Webhook, error)
Update patches the endpoint; nil params fields keep current values.
type Sender ¶
type Sender interface {
Send(ctx context.Context, delivery OutboundDelivery) (status int, body []byte, err error)
}
Sender delivers one signed request and reports the outcome.
func NewFetcherSender ¶
NewFetcherSender adapts the shared fetcher to the delivery contract.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service holds the webhook business rules: endpoint CRUD, event fan out (the outbox write), and signed delivery.
func NewService ¶
func NewService(store Store, db datastore.Store, queue *queue.Client, sealer cipher, log logger.Logger, opts ...ServiceOption) *Service
NewService builds the feature over its dependencies. db is used for the outbox transaction; queue carries the delivery tasks.
func (*Service) Create ¶
Create registers an endpoint and returns it with the freshly minted signing secret — the only time the plaintext is exposed.
func (*Service) Deliver ¶
func (s *Service) Deliver(ctx context.Context, task WebhookDeliveryTask) error
Deliver executes one queued delivery and records the attempt. The error is returned so the queue applies its retry schedule.
func (*Service) DeliverTo ¶
func (s *Service) DeliverTo(ctx context.Context, id WebhookID, event string, payload map[string]any) (DeliveryID, error)
DeliverTo queues one delivery to a named endpoint, bypassing the subscription filter. It backs the manual test route: a receiver must be reachable even when its subscription list does not name the synthetic event.
func (*Service) Emit ¶
Emit fans an event out to its subscribers. The pending log row and the queued delivery are written in one transaction, so an event that never commits never delivers. A nil queue (tests) or a module without subscribers degrades to a no-op.
func (*Service) ListDeliveries ¶
func (s *Service) ListDeliveries(ctx context.Context, id *WebhookID, params ListParams) ([]Delivery, int, error)
ListDeliveries pages delivery records; a nil id lists across endpoints.
func (*Service) RegisterQueue ¶
RegisterQueue registers the webhook delivery queue on the shared client. Called at build time by the composition root, before the dispatcher starts; a second call panics inside the queue (duplicate queue name), which is the intended fail-fast for double wiring.
func (*Service) RotateSecret ¶
RotateSecret issues a new signing secret and returns it once.
type ServiceOption ¶
type ServiceOption func(*Service)
ServiceOption configures the service.
func WithClock ¶
func WithClock(now func() time.Time) ServiceOption
WithClock overrides the signing clock (tests).
func WithSender ¶
func WithSender(sender Sender) ServiceOption
WithSender overrides the HTTP sender (tests).
type Store ¶
type Store interface {
List(ctx context.Context, params ListParams) ([]Webhook, int, error)
Get(ctx context.Context, id WebhookID) (Webhook, error)
Create(ctx context.Context, params CreateParams, encryptedSecret string) (Webhook, error)
Update(ctx context.Context, id WebhookID, params UpdateParams) (Webhook, error)
Delete(ctx context.Context, id WebhookID) error
SetSecret(ctx context.Context, id WebhookID, encryptedSecret string) error
// Subscribers returns the enabled endpoints for an event, with
// their decrypted-at-the-service-layer secret still encrypted.
Subscribers(ctx context.Context, event string) ([]Webhook, error)
// Secret returns the stored ciphertext for one endpoint.
Secret(ctx context.Context, id WebhookID) (string, error)
// InsertDelivery writes the outbox record (pending delivery).
InsertDelivery(ctx context.Context, exec Executor, delivery *Delivery, body []byte) error
// DeliveryForSend loads the immutable body bytes and event name
// for one attempt.
DeliveryForSend(ctx context.Context, id DeliveryID) (PendingDelivery, error)
// RecordAttempt appends one attempt row and folds its outcome
// into the delivery.
RecordAttempt(ctx context.Context, id DeliveryID, result AttemptResult) error
// ListAttempts returns the attempt history of one delivery in
// attempt order.
ListAttempts(ctx context.Context, id DeliveryID) ([]AttemptRecord, error)
ListDeliveries(ctx context.Context, params ListParams, webhookID *WebhookID) ([]Delivery, int, error)
// PruneDeliveries deletes deliveries recorded before the cutoff.
PruneDeliveries(ctx context.Context, before time.Time) (int64, error)
}
Store persists endpoints, deliveries, and attempts in public.webhook_endpoints / public.webhook_deliveries / public.webhook_delivery_attempts.
type UpdateParams ¶
type UpdateParams struct {
Name *string `json:"name,omitzero"`
Description *string `json:"description,omitzero"`
Endpoint *string `json:"endpoint,omitzero"`
Method *string `json:"method,omitzero"`
Headers map[string]string `json:"headers,omitzero"`
Enabled *bool `json:"enabled,omitzero"`
EventTypes []string `json:"event_types,omitzero"`
}
UpdateParams carries the PUT body fields. Nil pointers keep the current value; a rotated secret is issued by the rotate route.
func (*UpdateParams) Validate ¶
func (p *UpdateParams) Validate() error
Validate enforces the same rules as create on the fields present.
type Webhook ¶
type Webhook struct {
ID WebhookID `json:"id"`
Name string `json:"name"`
Description *string `json:"description,omitzero"`
Endpoint string `json:"endpoint"`
Method string `json:"method"`
Headers map[string]string `json:"headers,omitzero"`
Enabled bool `json:"enabled"`
EventTypes []string `json:"event_types"`
// Secret is the plaintext signing secret. It is populated on
// create/rotate only; listings and reads never carry it.
Secret *string `json:"secret,omitzero"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt *time.Time `json:"updated_at,omitzero"`
}
Webhook is one registered endpoint.
func (Webhook) SubscribedTo ¶
SubscribedTo reports whether the endpoint receives the event.
type WebhookDeliveryTask ¶
type WebhookDeliveryTask struct {
// DeliveryID identifies the webhook_deliveries row to update.
DeliveryID string `json:"delivery_id"`
// WebhookID identifies the endpoint row.
WebhookID string `json:"webhook_id"`
}
WebhookDeliveryTask delivers one recorded delivery to one endpoint. The delivery row is the outbox record: it is written in the same transaction that enqueues the task, and it holds the immutable body bytes every attempt (including retries) signs and sends.
func (WebhookDeliveryTask) Config ¶
func (WebhookDeliveryTask) Config() queue.QueueConfig
Config defines the queue settings.