webhook

package
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	OutboxPending    = "pending"
	OutboxDispatched = "dispatched"
	OutboxFailed     = "failed"
)

outbox row statuses.

View Source 
const MaxOutboxAttempts = 15

MaxOutboxAttempts is the DLQ threshold. After this many failed attempts a row becomes terminal ('failed') and is no longer retried automatically. Backoff ramp (see outboxBackoff) totals ~72h across 15 attempts.

View Source 
const SecretRotationGracePeriod = 72 * time.Hour

SecretRotationGracePeriod is how long a rotated-out secret keeps signing alongside its replacement. Velox uses 72h; Stripe's hosted equivalent caps at 24h (https://docs.stripe.com/webhooks/signature). The intentional deviation: self-hosted Velox deployments often have slower deploy cadences (manual rolls, no central CI fleet), so 24h would force a rushed cutover for tenants whose ops team checks the webhook receivers once a week. 72h covers a typical "find the change request → ship → verify" loop without the rushed-deploy footgun. Compromise case: a leaked secret stays usable for up to 72h after the operator rotates — bounded but not instant. Tighten to 24h if a tenant- configurable cap is justified by a real DP request.

Variables

View Source 
var ErrStaleDeliveryMark = errors.New("webhook: stale delivery mark (row no longer pending)")

ErrStaleDeliveryMark is returned when a delivery mark matched no pending row — the row was resolved by another worker after this worker's lease expired. The stale writer drops its result.

Functions

This section is empty.

Types

type AuditWriter 

type AuditWriter interface {
	Log(ctx context.Context, tenantID, action, resourceType, resourceID, resourceLabel string, metadata map[string]any) error
}

AuditWriter is the narrow audit surface webhook handler uses.

type CreateEndpointInput 

type CreateEndpointInput struct {
	URL         string   `json:"url"`
	Description string   `json:"description,omitempty"`
	Events      []string `json:"events"`
}

type CreateEndpointResult 

type CreateEndpointResult struct {
	Endpoint domain.WebhookEndpoint `json:"endpoint"`
	Secret   string                 `json:"secret"` // Shown once
}

type DeliveriesResponse 

type DeliveriesResponse struct {
	RootEventID string         `json:"root_event_id"`
	Deliveries  []DeliveryView `json:"deliveries"`
}

DeliveriesResponse wraps the timeline. We surface root_event_id so the dashboard can confirm it received the original-pivot's chain (matters when the operator clicked Replay from a clone — the chain is still rooted at the original).

type DeliveryView 

type DeliveryView struct {
	ID      string `json:"id"`
	EventID string `json:"event_id"`
	// EndpointID stays for machine use, but the timeline is an operator
	// surface: EndpointURL/EndpointDescription are what let a human
	// answer "which of my receivers is this?" without cross-referencing
	// an opaque id against a page that never displays ids. Resolved by
	// the store's historical join, so they survive endpoint deletion.
	EndpointID           string     `json:"endpoint_id"`
	EndpointURL          string     `json:"endpoint_url"`
	EndpointDescription  string     `json:"endpoint_description"`
	AttemptNo            int        `json:"attempt_no"`
	Status               string     `json:"status"`
	StatusCode           int        `json:"status_code"`
	ResponseBody         string     `json:"response_body"`
	Error                string     `json:"error"`
	RequestPayloadSHA256 string     `json:"request_payload_sha256"`
	AttemptedAt          time.Time  `json:"attempted_at"`
	CompletedAt          *time.Time `json:"completed_at"`
	NextRetryAt          *time.Time `json:"next_retry_at"`
	IsReplay             bool       `json:"is_replay"`
	ReplayEventID        string     `json:"replay_event_id"`
}

DeliveryView is the dashboard-facing delivery row: it carries the per-attempt facts the timeline needs (attempt number, status, response body, timestamps) plus the request_payload_sha256 the diff viewer uses to flag "payload identical between attempts" (the common case for Stripe-style replays).

Snake_case throughout — pinned by TestWireShape_WebhookEventDeliveries.

type DispatchLock 

type DispatchLock interface {
	Release()
}

DispatchLock is a held cluster-wide lock the dispatcher must release.

type DispatchLocker 

type DispatchLocker interface {
	TryDispatcherLock(ctx context.Context) (DispatchLock, bool, error)
}

DispatchLocker gates the dispatcher tick on a cluster-wide advisory lock. Row-level FOR UPDATE SKIP LOCKED already prevents double-delivery when two dispatchers race, but the lock avoids both replicas issuing the same claim query every 2s when only one drain worker is actually needed — less churn on the connection pool and on webhook_outbox's index scan. Nil Locker disables gating (single-replica / test mode).

type Dispatcher 

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

Dispatcher drains the webhook_outbox by invoking Service.Dispatch for each pending row. It is the bridge between the durable outbox (what producers enqueue) and the existing per-endpoint delivery pipeline (webhook_events + webhook_deliveries). Handler semantics: a row is marked 'dispatched' once Service.Dispatch returns nil, which means the event has been persisted and queued to all matching endpoints — per-endpoint HTTP retry is then owned by Service.StartRetryWorker, independent of the outbox.

func NewDispatcher 

func NewDispatcher(outbox *OutboxStore, svc *Service, cfg DispatcherConfig) *Dispatcher

func (*Dispatcher) Config 

func (d *Dispatcher) Config() DispatcherConfig

Start runs the dispatcher loop until ctx is cancelled. Intended to be launched as a goroutine from cmd/velox during boot, alongside the existing webhook retry worker. Config exposes the resolved configuration for the invariant test.

func (*Dispatcher) SetLocker 

func (d *Dispatcher) SetLocker(locker DispatchLocker)

SetLocker enables leader gating on the dispatcher tick.

func (*Dispatcher) Start 

func (d *Dispatcher) Start(ctx context.Context)

type DispatcherConfig 

type DispatcherConfig struct {
	// Interval is the poll cadence between ProcessBatch calls. Default 2s if zero.
	Interval time.Duration
	// BatchSize bounds how many rows are claimed per tick. Default 25 if zero.
	BatchSize int
	// BatchTimeout bounds how long a single batch is allowed to run before its
	// tx is cancelled (releasing row locks). Default 30s if zero.
	BatchTimeout time.Duration
}

DispatcherConfig controls the outbox dispatcher loop.

type EndpointDeliveryView added in v0.2.0 

type EndpointDeliveryView struct {
	ID        string `json:"id"`
	EventID   string `json:"event_id"`
	EventType string `json:"event_type"`
	// IsReplay marks rows born from an operator replay (event-wide or
	// single-receiver) — the drill-down badges them the same way the
	// event timeline does.
	IsReplay        bool       `json:"is_replay"`
	ReplayOfEventID string     `json:"replay_of_event_id,omitempty"`
	Status          string     `json:"status"`
	StatusCode      int        `json:"status_code"`
	AttemptCount    int        `json:"attempt_count"`
	ResponseBody    string     `json:"response_body"`
	Error           string     `json:"error"`
	CreatedAt       time.Time  `json:"created_at"`
	CompletedAt     *time.Time `json:"completed_at"`
	NextRetryAt     *time.Time `json:"next_retry_at"`
}

EndpointDeliveryView is the endpoint drill-down row: one receiver's history across events, so each row names the business fact it carried (event_type) rather than the receiver (which is the page's own subject). Snake_case pinned by TestWireShape_EndpointDeliveries.

type EndpointStats 

type EndpointStats struct {
	EndpointID      string  `json:"endpoint_id"`
	TotalDeliveries int     `json:"total_deliveries"`
	Succeeded       int     `json:"succeeded"`
	Failed          int     `json:"failed"`
	SuccessRate     float64 `json:"success_rate"`
}

EndpointStats holds delivery statistics for a single webhook endpoint.

type EventBus 

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

EventBus is the in-memory pub/sub the SSE handler uses to live-tail new events as they're dispatched. We deliberately do NOT poll the DB on a tick — that approach burns the partial-index cache on every poll and lags ~1s behind dispatch. Instead, the Service's Dispatch path (and the OutboxDispatcher's success callback) calls EventBus.Publish synchronously; subscribers receive frames within goroutine-scheduling latency.

Slow subscribers do NOT block the publisher — Publish does a non-blocking send and drops frames for any subscriber whose buffer is full. Dropped frames are not retried; the dashboard's snapshot-at- connect compensates by re-fetching recent events on reconnect, so a disconnected client picks back up cleanly.

Per-tenant fan-out keeps the bus a no-op for any tenant with zero dashboards open. We index subscribers by tenant_id at registration so Publish doesn't iterate every tenant's subscriber list.

func NewEventBus 

func NewEventBus() *EventBus

NewEventBus returns an empty in-memory bus. Single instance per process is sufficient — the API server is the only producer (replicas use the leader-elected outbox dispatcher; replicas with no HTTP traffic produce no live events to fan out).

func (*EventBus) Publish 

func (b *EventBus) Publish(tenantID string, frame StreamFrame)

Publish fans a frame out to every subscriber for the frame's tenant. Non-blocking by construction: a slow subscriber drops the frame silently rather than back-pressuring the dispatcher. The bus is hot- path adjacent (called from Service.Dispatch) so we keep the lock hold tight and read-only.

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(tenantID string) (<-chan StreamFrame, func())

Subscribe registers a new subscriber for the given tenant. Returns the receive channel and an unsubscribe func the caller MUST call (typically deferred at request-handler scope) so the goroutine scheduler can collect the closed connection.

The buffer size of 32 is empirically picked: at ~5 frames/s steady- state per tenant (a busy production workload), the buffer absorbs ~6s of producer bursts before drops. A subscriber that can't drain at that rate is almost certainly a dead connection — better to drop than block the dispatcher.

type HTTPClient 

type HTTPClient interface {
	Do(req *http.Request) (*http.Response, error)
}

HTTPClient is the interface for making HTTP requests (mockable in tests).

type Handler 

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

func NewHandler 

func NewHandler(svc *Service) *Handler

func (*Handler) EventRoutes 

func (h *Handler) EventRoutes() chi.Router

EventRoutes is the Week 6 real-time event surface, mounted at /v1/webhook_events. Lives here (rather than alongside Routes) so the route table stays tightly scoped to "the live-tail dashboard" and router.go can mount it under its own auth scope without dragging in the endpoint-management permissions.

Critical: chi/v5 dispatches by registration order. We register /stream BEFORE /{id} so a literal "stream" path doesn't get captured as an ID and route to the deliveries handler. See docs/design-create-preview.md for the canonical write-up.

Note: the SSE handler (streamEvents) IS exposed here for tests and dev-time mounting, but the production router mounts /stream SEPARATELY outside the /v1 block so it can skip the global 30s middleware.Timeout (which would kill any long-lived stream). See internal/api/router.go for the production mount.

func (*Handler) Routes 

func (h *Handler) Routes() chi.Router

func (*Handler) SetAuditLogger 

func (h *Handler) SetAuditLogger(a AuditWriter)

SetAuditLogger wires audit on webhook endpoint lifecycle + manual event replays. Endpoint URL changes + secret rotations are integration-security events; manual replays change downstream state and need forensic trail.

func (*Handler) StreamHandler 

func (h *Handler) StreamHandler() http.HandlerFunc

StreamHandler is the chi-compatible handler for the live-tail SSE stream. Exported so router.go can mount it on a route block that skips the 30s timeout middleware applied to /v1/* routes.

type OutboxDispatcher 

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

OutboxDispatcher satisfies domain.EventDispatcher by persisting each event as a pending row in webhook_outbox. The Dispatcher worker drains it.

This is the durable drop-in replacement for calling Service.Dispatch directly: producers get persist-before-return semantics, and the actual HTTP delivery is done by the worker asynchronously — same as today, just with a durable queue in between.

func NewOutboxDispatcher 

func NewOutboxDispatcher(outbox *OutboxStore) *OutboxDispatcher

func (*OutboxDispatcher) Dispatch 

func (d *OutboxDispatcher) Dispatch(ctx context.Context, tenantID, eventType string, payload map[string]any) error

Dispatch enqueues a pending outbox row. Returns an error if the insert fails — callers should treat that as a fatal enqueue failure, since the event would otherwise be lost. The current fireEvent call sites ignore the error because under the pre-outbox scheme nothing could be done; tightening that is a follow-up once the outbox becomes the default.

type OutboxHandler 

type OutboxHandler func(ctx context.Context, row OutboxRow) error

OutboxHandler is called once per claimed row by ProcessBatch. Returning nil means the row is dispatched successfully; returning an error schedules a retry (or DLQ once MaxOutboxAttempts is reached).

type OutboxRow 

type OutboxRow struct {
	ID            string
	TenantID      string
	Livemode      bool // Carries the producer tx's mode; dispatcher propagates this into ctx so delivery hits only same-mode endpoints.
	EventType     string
	Payload       map[string]any
	Status        string
	Attempts      int
	NextAttemptAt time.Time
	LastError     string
	CreatedAt     time.Time
	DispatchedAt  *time.Time
}

OutboxRow is a single queued outbound-event intent. Produced inside the business-op transaction, drained by the dispatcher.

type OutboxStore 

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

OutboxStore persists webhook-event emission intents.

func NewOutboxStore 

func NewOutboxStore(db *postgres.DB) *OutboxStore

func (*OutboxStore) Enqueue 

func (s *OutboxStore) Enqueue(ctx context.Context, tx *sql.Tx, tenantID, eventType string, payload map[string]any) (string, error)

Enqueue inserts a pending outbox row inside the caller's tx. Use this from a business-op store method so the event is persisted atomically with the state change — if the tx rolls back, no event is emitted; if it commits, the dispatcher will eventually deliver.

func (*OutboxStore) EnqueueStandalone 

func (s *OutboxStore) EnqueueStandalone(ctx context.Context, tenantID, eventType string, payload map[string]any) (string, error)

EnqueueStandalone opens its own tenant-scoped tx to insert the outbox row. Use when the caller has no tx already in scope — still durable because the insert commits before return, but not atomic with the preceding business op. Prefer Enqueue whenever a tx is available.

func (*OutboxStore) FailedCount 

func (s *OutboxStore) FailedCount(ctx context.Context) (int64, error)

FailedCount returns rows currently in the DLQ — used for alerting. If this grows, an endpoint is persistently broken or a producer is emitting malformed events.

func (*OutboxStore) PendingCount 

func (s *OutboxStore) PendingCount(ctx context.Context) (int64, error)

PendingCount returns the current number of rows awaiting dispatch. Intended for metrics (operator gauge) — not on the hot path.

func (*OutboxStore) ProcessBatch 

func (s *OutboxStore) ProcessBatch(ctx context.Context, limit int, handler OutboxHandler) (int, error)

ProcessBatch locks up to `limit` due pending rows across all tenants, hands them to `handler`, and marks each row based on the handler's result — all within a single tx. Row locks held for the tx's duration prevent concurrent dispatchers from double-delivering (`FOR UPDATE SKIP LOCKED`).

Returns the number of rows processed (attempted, regardless of outcome). Callers should set a sensible query timeout on ctx so a stuck handler can't hold locks indefinitely.

func (*OutboxStore) TryDispatcherLock 

func (s *OutboxStore) TryDispatcherLock(ctx context.Context) (DispatchLock, bool, error)

TryDispatcherLock tries to acquire the cluster-wide advisory lock that gates the outbox dispatcher tick. Returns (lock, true, nil) on success; caller defers lock.Release. Returns (nil, false, nil) if another replica holds it. Implements webhook.DispatchLocker.

type PostgresStore 

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

func NewPostgresStore 

func NewPostgresStore(db *postgres.DB) *PostgresStore

func (*PostgresStore) CreateDeliveriesForEvent 

func (s *PostgresStore) CreateDeliveriesForEvent(ctx context.Context, tenantID, eventID string, endpointIDs []string, birthLease time.Duration) ([]domain.WebhookDelivery, error)

CreateDeliveriesForEvent batch-creates born-leased delivery rows for an event that already exists (Replay: the clone event commits in CreateReplayEvent first). One tx — a replay whose fan-out partially failed would otherwise deliver to a subset with no record of the rest.

func (*PostgresStore) CreateDelivery 

func (s *PostgresStore) CreateDelivery(ctx context.Context, tenantID string, d domain.WebhookDelivery) (domain.WebhookDelivery, error)

func (*PostgresStore) CreateEndpoint 

func (s *PostgresStore) CreateEndpoint(ctx context.Context, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)

func (*PostgresStore) CreateEndpointTx 

func (s *PostgresStore) CreateEndpointTx(ctx context.Context, tx *sql.Tx, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)

CreateEndpointTx inserts a webhook endpoint inside an existing tx. Used by recipe.Service.Instantiate so a recipe that wants a default outbound receiver lands its endpoint atomically with the rest of the recipe.

func (*PostgresStore) CreateEvent 

func (s *PostgresStore) CreateEvent(ctx context.Context, tenantID string, event domain.WebhookEvent) (domain.WebhookEvent, error)

func (*PostgresStore) CreateEventWithDeliveries 

func (s *PostgresStore) CreateEventWithDeliveries(ctx context.Context, tenantID string, event domain.WebhookEvent, endpointIDs []string, birthLease time.Duration, outboxRowID string) (domain.WebhookEvent, []domain.WebhookDelivery, error)

CreateEventWithDeliveries creates the event AND all its delivery rows in ONE transaction (P5, CreateEndpointTx precedent) — the old shape committed the event, then each fan-out goroutine inserted its own delivery row: a crash in between minted an event no delivery row referenced, so no endpoint ever received it and nothing retried. Delivery rows are BORN LEASED (next_retry_at = now()+birthLease): the in-process goroutine owns the first attempt inside the lease, and the retry worker is the crash backstop that picks the row up at lease expiry — never a NULL next_retry_at double-POST race.

outboxRowID, when non-empty, marks the webhook_outbox row dispatched INSIDE the same tx (handler-owns-mark): a crash after this commit but before the old separate outbox mark used to re-run the handler and mint a DUPLICATE event with a fresh id that receivers cannot dedupe.

func (*PostgresStore) CreateReplayEvent 

func (s *PostgresStore) CreateReplayEvent(ctx context.Context, tenantID, originalEventID string) (domain.WebhookEvent, error)

CreateReplayEvent clones an existing event row into a new event whose replay_of_event_id points at the original. The clone reuses the original's event_type and payload byte-for-byte (so the diff viewer can verify "payload identical, only timestamps differ" — Stripe's "Resend" semantics).

func (*PostgresStore) DeleteEndpoint 

func (s *PostgresStore) DeleteEndpoint(ctx context.Context, tenantID, id string) error

func (*PostgresStore) EventDeliveryStatuses added in v0.2.0 

func (s *PostgresStore) EventDeliveryStatuses(ctx context.Context, tenantID string, eventIDs []string) (map[string]string, error)

EventDeliveryStatuses rolls each event's deliveries up to one status. Precedence: any pending → "pending" (work in flight), else any failed → "failed" (something needs attention — a partial success must not read as done), else "delivered". Replaces the age heuristic the SSE snapshot used to guess with ("<24h old ⇒ pending"), which showed a settled event as pending all day and would have shown a permanently-failed delivery as delivered once it aged past the ladder.

func (*PostgresStore) GetDelivery added in v0.2.0 

func (s *PostgresStore) GetDelivery(ctx context.Context, tenantID, id string) (domain.WebhookDelivery, error)

func (*PostgresStore) GetEndpoint 

func (s *PostgresStore) GetEndpoint(ctx context.Context, tenantID, id string) (domain.WebhookEndpoint, error)

func (*PostgresStore) GetEndpointStats 

func (s *PostgresStore) GetEndpointStats(ctx context.Context, tenantID string) ([]EndpointStats, error)

func (*PostgresStore) GetEvent 

func (s *PostgresStore) GetEvent(ctx context.Context, tenantID, id string) (domain.WebhookEvent, error)

GetEvent returns a single event by id (tenant-scoped via RLS).

func (*PostgresStore) ListDeliveries 

func (s *PostgresStore) ListDeliveries(ctx context.Context, tenantID, eventID string) ([]domain.WebhookDelivery, error)

func (*PostgresStore) ListDeliveriesByEndpoint added in v0.2.0 

func (s *PostgresStore) ListDeliveriesByEndpoint(ctx context.Context, tenantID, endpointID string, limit int) ([]domain.WebhookDelivery, error)

ListDeliveriesByEndpoint is the endpoint drill-down: one receiver's delivery history, newest first. The events join hydrates each row with the business fact it carried (event_type) and its replay pivot — this surface lists deliveries ACROSS events, so a row must be readable without a second fetch. INNER JOIN is safe here: a delivery row is created in the same tx as its event (or referencing an event that already committed, on the replay path), so the FK target always exists.

func (*PostgresStore) ListEndpoints 

func (s *PostgresStore) ListEndpoints(ctx context.Context, tenantID string) ([]domain.WebhookEndpoint, error)

func (*PostgresStore) ListEvents 

func (s *PostgresStore) ListEvents(ctx context.Context, tenantID string, limit int) ([]domain.WebhookEvent, error)

func (*PostgresStore) ListPendingDeliveries 

func (s *PostgresStore) ListPendingDeliveries(ctx context.Context, limit int) ([]domain.WebhookDelivery, error)

func (*PostgresStore) RotateEndpointSecret 

func (s *PostgresStore) RotateEndpointSecret(ctx context.Context, tenantID, id, newSecret string, gracePeriod time.Duration) (domain.WebhookEndpoint, error)

func (*PostgresStore) SetEncryptor 

func (s *PostgresStore) SetEncryptor(enc *crypto.Encryptor)

SetEncryptor configures AES-256-GCM encryption for webhook signing secrets at rest. When set (non-noop), Create/UpdateEndpointSecret encrypt the raw whsec_ secret before INSERT; Get/ListEndpoints decrypt it after SELECT so the Dispatch path can sign with the plaintext key. Without this, the raw signing key is stored in plaintext — a DB dump yields webhook-forging capability against every tenant's receivers.

func (*PostgresStore) TryRetryLock 

func (s *PostgresStore) TryRetryLock(ctx context.Context) (DispatchLock, bool, error)

ListPendingDeliveries atomically CLAIMS up to `limit` due deliveries and returns them. It is not a pure read: each returned row is leased by pushing next_retry_at one claim-lease into the future, and rows already locked by a concurrent worker are skipped (FOR UPDATE SKIP LOCKED). This makes the retry worker safe to run on multiple replicas — two workers never claim the same delivery, so a webhook isn't delivered twice per due-tick. The lease also provides crash recovery: if the claiming worker dies before overwriting the row's status, the lease (retryClaimLease — sized to the claim batch, P5) expires and another worker re-claims it. The window must exceed the per-attempt HTTP timeout (10s) by a wide margin so an in-flight delivery is never re-claimed underneath itself. TryRetryLock leader-gates the retry worker tick (LockKeyWebhookRetry).

func (*PostgresStore) UpdateDelivery 

func (s *PostgresStore) UpdateDelivery(ctx context.Context, tenantID string, d domain.WebhookDelivery) (domain.WebhookDelivery, error)

func (*PostgresStore) UpdateEndpoint 

func (s *PostgresStore) UpdateEndpoint(ctx context.Context, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)

UpdateEndpoint mutates url / description / events / active WITHOUT touching the signing secret (Stripe's update shape: subscribers change the target or the event set without a receiver redeploy — pre-2026-07-05 the only mutation path was delete+recreate, which minted a new secret, and recipe-created endpoints were permanently dead: Active:false with a placeholder URL and no way to fix either). The caller (service) has already validated URL + event names.

type ReplayDeliveryResult added in v0.2.0 

type ReplayDeliveryResult struct {
	// EventID is the freshly-minted replay clone (replay_of set), same
	// shape as an event-wide replay — the delivery timeline folds it
	// into the original's chain.
	EventID string `json:"event_id"`
	// ReplayOf is the root original event of the audit chain.
	ReplayOf string `json:"replay_of"`
	// DeliveryID is the new delivery row created for the target
	// endpoint — the drill-down highlights it on refetch.
	DeliveryID string `json:"delivery_id"`
	Status     string `json:"status"`
}

ReplayDeliveryResult is the response for a single-receiver replay.

type ReplayResult 

type ReplayResult struct {
	// EventID is the freshly-minted webhook_events row that's been
	// queued for delivery. The dashboard's SSE tail will pick it up
	// within a tick.
	EventID string `json:"event_id"`
	// ReplayOf is the ID of the original event whose payload was
	// cloned. The dashboard groups the original + all replays under
	// this pivot so the Deliveries timeline shows the full audit
	// chain.
	ReplayOf string `json:"replay_of"`
	// Status is "queued" — the deliver fan-out runs asynchronously, so
	// at response time we can only confirm the clone landed and
	// matched at least one endpoint.
	Status string `json:"status"`
}

ReplayResult is the response shape for POST /v1/webhook_events/{id}/replay. Returned to the dashboard so the live tail can highlight the new row and the Toast confirms what the operator just queued.

type RetryLocker 

type RetryLocker interface {
	TryRetryLock(ctx context.Context) (DispatchLock, bool, error)
}

RetryLocker is the optional leader-gate the retry worker uses when the store provides it (PostgresStore does; unit-test fakes usually don't).

type RotateSecretResult 

type RotateSecretResult struct {
	Secret             string     `json:"secret"`
	SecondaryValidTill *time.Time `json:"secondary_valid_until,omitempty"`
}

RotateSecretResult carries the new secret and the expiry of the grace-period sibling. Exposed on the handler response so the dashboard can show "old secret valid until <time>" copy.

type Service 

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

func NewService 

func NewService(store Store, client HTTPClient) *Service

func NewTestService 

func NewTestService(store Store, client HTTPClient) *Service

NewTestService creates a service with synchronous delivery (no goroutines).

func (*Service) CreateEndpoint 

func (s *Service) CreateEndpoint(ctx context.Context, tenantID string, input CreateEndpointInput) (CreateEndpointResult, error)

func (*Service) CreateEndpointTx 

func (s *Service) CreateEndpointTx(ctx context.Context, tx *sql.Tx, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)

CreateEndpointTx forwards to the store's tx-aware insert. Used by recipe.Service.Instantiate so a recipe with a default outbound endpoint commits atomically with the rest of the recipe.

func (*Service) DeleteEndpoint 

func (s *Service) DeleteEndpoint(ctx context.Context, tenantID, id string) error

func (*Service) Dispatch 

func (s *Service) Dispatch(ctx context.Context, tenantID, eventType string, payload map[string]any) error

Dispatch creates a webhook event and delivers it to all matching endpoints.

Mode scoping: ListEndpoints runs under the caller's ctx livemode, which RLS already filters on. The explicit ep.Livemode == event.Livemode check below is defense-in-depth — if a future call path opens a bypass tx or the RLS predicate is relaxed, test-mode events must still never cross into a live endpoint (and vice versa). Cross-mode delivery would leak synthetic data into production monitoring.

func (*Service) DispatchFromOutbox 

func (s *Service) DispatchFromOutbox(ctx context.Context, outboxRowID, tenantID, eventType string, payload map[string]any) error

DispatchFromOutbox is Dispatch carrying the webhook_outbox row that produced this event, so the row's dispatched-mark commits ATOMICALLY with the event + delivery rows (handler-owns-mark, P5): the old separate mark meant a crash between the event commit and the outbox mark re-ran the handler next tick and minted a duplicate event with a fresh id — receivers cannot dedupe those. Empty outboxRowID = direct dispatch (no outbox row).

The endpoint fan-out is resolved FIRST (the event's livemode is the ctx's — the 0021 trigger stamps the row from the same session GUC) and the event + every matching delivery row are created in ONE store tx, rows born leased (birthLeaseWindow): the in-process goroutines own the first attempt; a crash any time after the commit leaves rows the retry worker picks up at lease expiry — never an event no delivery row references, never a NULL next_retry_at.

func (*Service) EventBus 

func (s *Service) EventBus() *EventBus

EventBus exposes the in-memory pub/sub backing the SSE stream. The handler subscribes per-request; the service publishes frames from Dispatch and the deliver path. Returns the same instance for the lifetime of the service so multiple handlers / cmd-side workers share one fan-out.

func (*Service) EventDeliveryStatuses added in v0.2.0 

func (s *Service) EventDeliveryStatuses(ctx context.Context, tenantID string, eventIDs []string) (map[string]string, error)

EventDeliveryStatuses rolls each event's deliveries up to one dashboard status — see the Store method for the precedence contract.

func (*Service) GetEndpointStats 

func (s *Service) GetEndpointStats(ctx context.Context, tenantID string) ([]EndpointStats, error)

GetEndpointStats returns delivery success/failure stats per endpoint.

func (*Service) GetEvent 

func (s *Service) GetEvent(ctx context.Context, tenantID, id string) (domain.WebhookEvent, error)

GetEvent fetches a single event by id (tenant-scoped via RLS at the store layer). Surfaced for the SSE handler's deliveries-list path so it can resolve the replay root before walking the timeline.

func (*Service) ListDeliveries 

func (s *Service) ListDeliveries(ctx context.Context, tenantID, eventID string) ([]domain.WebhookDelivery, error)

ListDeliveries returns deliveries for a specific event.

func (*Service) ListEndpointDeliveries added in v0.2.0 

func (s *Service) ListEndpointDeliveries(ctx context.Context, tenantID, endpointID string, limit int) (domain.WebhookEndpoint, []domain.WebhookDelivery, error)

ListEndpointDeliveries returns one endpoint plus its delivery history (newest first). The endpoint lookup runs first so a bad id 404s rather than returning an empty list that reads as "healthy receiver, no traffic".

func (*Service) ListEndpoints 

func (s *Service) ListEndpoints(ctx context.Context, tenantID string) ([]domain.WebhookEndpoint, error)

func (*Service) ListEvents 

func (s *Service) ListEvents(ctx context.Context, tenantID string, limit int) ([]domain.WebhookEvent, error)

ListEvents returns recent webhook events for a tenant.

func (*Service) Replay 

func (s *Service) Replay(ctx context.Context, tenantID, eventID string) (ReplayResult, error)

Replay clones an existing webhook event into a fresh row (with replay_of_event_id pointing at the original) and dispatches it to every matching active endpoint. The clone is what gets delivered, so the original's deliveries are never mutated — every replay produces a brand-new row in the timeline. A second replay of the same original event is therefore not idempotent in the DB-row sense (it creates another clone), but is idempotent in the audit-trail sense: the original delivery history is preserved and the operator sees N distinct replay attempts on the timeline.

func (*Service) ReplayDelivery added in v0.2.0 

func (s *Service) ReplayDelivery(ctx context.Context, tenantID, endpointID, deliveryID string) (ReplayDeliveryResult, error)

ReplayDelivery re-sends one delivery's event to THAT delivery's endpoint only — the industry replay unit (Stripe "Resend"/"Retry now", GitHub "Redeliver", Svix per-attempt resend). Event-wide Replay fans the clone out to every matching endpoint, which re-delivers to receivers that already succeeded; this path exists so fixing ONE broken receiver doesn't spray duplicates at the healthy ones.

Same audit chain as Replay: a clone event (replay_of = root original) plus a born-leased delivery row — the original delivery is history and is never mutated.

func (*Service) RetryPendingDeliveries 

func (s *Service) RetryPendingDeliveries(ctx context.Context) error

RetryPendingDeliveries picks up deliveries due for retry and re-attempts them.

func (*Service) RotateSecret 

func (s *Service) RotateSecret(ctx context.Context, tenantID, endpointID string) (RotateSecretResult, error)

RotateSecret generates a new signing secret for an endpoint and returns it alongside the grace-period expiry. The previous secret stays valid for SecretRotationGracePeriod — dispatcher signs outbound events with BOTH secrets during the window (two v1= entries in Velox-Signature, Stripe multi-signature style) so receivers can stage a verifier update without breaking production traffic. After the window, only the new secret is used. The new secret is returned once to the caller (dashboard shows it, then it's no longer retrievable).

func (*Service) StartRetryWorker 

func (s *Service) StartRetryWorker(ctx context.Context, interval time.Duration)

StartRetryWorker runs a background loop that retries pending deliveries on the given interval. It blocks until the context is cancelled. When the store exposes an advisory lock (production Postgres), each tick is leader-gated — the claim lease alone is a correct multi-replica guard, but gating makes its sizing non-critical (P5; same posture as both outbox dispatchers).

func (*Service) UpdateEndpoint 

func (s *Service) UpdateEndpoint(ctx context.Context, tenantID, id string, input UpdateEndpointInput) (domain.WebhookEndpoint, error)

UpdateEndpoint mutates url/description/events/active without rotating the signing secret (Stripe's update shape). This is also what makes recipe-created endpoints usable: they are created inactive with a placeholder URL, and this is the "point it at a real URL and activate" surface that previously didn't exist.

type Store 

type Store interface {
	// Endpoints
	CreateEndpoint(ctx context.Context, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)
	CreateEndpointTx(ctx context.Context, tx *sql.Tx, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)
	GetEndpoint(ctx context.Context, tenantID, id string) (domain.WebhookEndpoint, error)
	// UpdateEndpoint mutates url/description/events/active without touching
	// the signing secret (the PATCH shape; delete+recreate was the only
	// mutation path before and it rotated the secret).
	UpdateEndpoint(ctx context.Context, tenantID string, ep domain.WebhookEndpoint) (domain.WebhookEndpoint, error)
	ListEndpoints(ctx context.Context, tenantID string) ([]domain.WebhookEndpoint, error)
	DeleteEndpoint(ctx context.Context, tenantID, id string) error
	// RotateEndpointSecret atomically moves the current signing secret into
	// the secondary slot with `gracePeriod` left on its clock, and replaces
	// the primary with `newSecret`. Runs under a single transaction with
	// FOR UPDATE so a concurrent rotate can't lose the old secret. If
	// gracePeriod is zero, no secondary is written (hard-replace).
	RotateEndpointSecret(ctx context.Context, tenantID, id, newSecret string, gracePeriod time.Duration) (domain.WebhookEndpoint, error)

	// Events
	CreateEvent(ctx context.Context, tenantID string, event domain.WebhookEvent) (domain.WebhookEvent, error)
	// CreateEventWithDeliveries creates the event and every fan-out
	// delivery row (born leased) in ONE tx; outboxRowID, when set, marks
	// the producing webhook_outbox row dispatched in the same tx (P5
	// handler-owns-mark — see PostgresStore for the crash analysis).
	CreateEventWithDeliveries(ctx context.Context, tenantID string, event domain.WebhookEvent, endpointIDs []string, birthLease time.Duration, outboxRowID string) (domain.WebhookEvent, []domain.WebhookDelivery, error)
	// CreateDeliveriesForEvent batch-creates born-leased delivery rows
	// for an EXISTING event (the replay path — the clone commits first).
	CreateDeliveriesForEvent(ctx context.Context, tenantID, eventID string, endpointIDs []string, birthLease time.Duration) ([]domain.WebhookDelivery, error)
	ListEvents(ctx context.Context, tenantID string, limit int) ([]domain.WebhookEvent, error)
	// EventDeliveryStatuses rolls each event's deliveries up to one status:
	// any pending → "pending", else any failed → "failed", else "delivered".
	// Events with no deliveries are absent from the map (no endpoint matched).
	EventDeliveryStatuses(ctx context.Context, tenantID string, eventIDs []string) (map[string]string, error)
	GetEvent(ctx context.Context, tenantID, id string) (domain.WebhookEvent, error)
	// CreateReplayEvent clones an existing event into a fresh event row
	// with replay_of_event_id pointing back at the original. The clone
	// goes through the normal Dispatch fan-out so each subscribed
	// endpoint produces its own delivery row — visible alongside the
	// original's deliveries on the dashboard timeline.
	CreateReplayEvent(ctx context.Context, tenantID, originalEventID string) (domain.WebhookEvent, error)

	// Deliveries
	CreateDelivery(ctx context.Context, tenantID string, d domain.WebhookDelivery) (domain.WebhookDelivery, error)
	UpdateDelivery(ctx context.Context, tenantID string, d domain.WebhookDelivery) (domain.WebhookDelivery, error)
	ListDeliveries(ctx context.Context, tenantID, eventID string) ([]domain.WebhookDelivery, error)
	GetDelivery(ctx context.Context, tenantID, id string) (domain.WebhookDelivery, error)
	// ListDeliveriesByEndpoint is the endpoint drill-down query: every
	// delivery sent to ONE receiver, newest first, each row hydrated with
	// its event's type + replay pivot (the surface lists deliveries across
	// events, so rows must be self-describing).
	ListDeliveriesByEndpoint(ctx context.Context, tenantID, endpointID string, limit int) ([]domain.WebhookDelivery, error)

	// Stats
	GetEndpointStats(ctx context.Context, tenantID string) ([]EndpointStats, error)

	// Retry support (cross-tenant, system-level)
	ListPendingDeliveries(ctx context.Context, limit int) ([]domain.WebhookDelivery, error)
}

type StreamFrame 

type StreamFrame struct {
	EventID         string     `json:"event_id"`
	EventType       string     `json:"event_type"`
	CustomerID      string     `json:"customer_id"`
	Status          string     `json:"status"`
	LastAttemptAt   *time.Time `json:"last_attempt_at"`
	CreatedAt       time.Time  `json:"created_at"`
	Livemode        bool       `json:"livemode"`
	ReplayOfEventID *string    `json:"replay_of_event_id"`
}

StreamFrame is the SSE-shaped projection of a webhook event. We use a flat snake_case struct (not the raw domain.WebhookEvent) because the dashboard contract is stable — adding a column to webhook_events shouldn't silently widen the wire shape.

The frame carries the latest delivery rollup we know about at emit time:

  • status — the pub/sub-side aggregate ("dispatched" / "pending" / "failed"). The dashboard renders this directly; clicking the row fetches the per-attempt timeline via /v1/webhook_events/{id}/deliveries.
  • last_attempt_at — best-effort time of the most recent dispatch attempt seen by the bus. NULL on the initial "snapshot" frames sent at connect time when the event hasn't been dispatched yet.

func FrameFromEvent 

func FrameFromEvent(e domain.WebhookEvent, status string, lastAttemptAt *time.Time) StreamFrame

FrameFromEvent builds a StreamFrame from a domain.WebhookEvent. status defaults to "pending" if not yet observed by the dispatcher. The frame inspects payload.customer_id (the convention all Velox internal Dispatchers honor) so the dashboard can group by customer without a JOIN.

type UpdateEndpointInput 

type UpdateEndpointInput struct {
	URL         *string   `json:"url,omitempty"`
	Description *string   `json:"description,omitempty"`
	Events      *[]string `json:"events,omitempty"`
	Active      *bool     `json:"active,omitempty"`
}

UpdateEndpointInput is the PATCH shape: nil = leave unchanged.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.