webhook

package
v0.37.0 Latest Latest
Warning

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

 Go to latest Published: Jul 20, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package webhook is the outbound-webhook battery for GoFastr.

Apps publish events; subscribers are HTTP endpoints that receive signed POST requests. Failed deliveries retry with exponential backoff and are parked in a dead-letter list once the attempt budget is exhausted.

The core package (Manager, Store, SignWithTimestamp/VerifyTimestamped) is dependency-free beyond the standard library. The optional Bridge helpers in bridge.go pull in framework/event so internal Emit/EmitAsync calls can auto-fan out to subscribers; if you don't use the bridge, that dependency is dead code.

See docs/webhooks.md for the wiring guide.

Index

Constants

View Source 
const (
	InboundStatusReceived   = "received"
	InboundStatusProcessing = "processing"
	InboundStatusProcessed  = "processed"
	InboundStatusFailed     = "failed"
)

Inbound lifecycle status values for an InboundEnvelope. Prefix avoids colliding with the outbound DeliveryStatus constants (StatusFailed etc.).

View Source 
const DefaultMaxResponseBodyBytes int64 = 64 << 10 // 64 KiB

DefaultMaxResponseBodyBytes is the per-attempt response-body cap when Options.MaxResponseBodyBytes is unset. A malicious receiver returning gigabytes of body would otherwise exhaust manager memory.

View Source 
const DefaultTimestampTolerance = 5 * time.Minute

DefaultTimestampTolerance is the suggested replay window for VerifyTimestamped — wide enough to cover modest clock skew, narrow enough that a captured signature decays quickly.

View Source 
const MaxPayloadBytes = 1 << 20 // 1 MiB

MaxPayloadBytes caps webhook payload size. Larger payloads are rejected at Publish time to prevent unbounded queue/memory growth.

View Source 
const SignatureHeader = "X-GoFastr-Signature"

SignatureHeader is the response/request header that carries the payload signature.

Variables

This section is empty.

Functions

func Bridge 

func Bridge(bus *event.EventBus, mgr *Manager, events ...string) (cancel func())

Bridge subscribes the Manager to every named event on the bus so that an Emit/EmitAsync automatically fans out to matching webhook subscribers.

The returned cancel function detaches every subscription at once — call it from your shutdown path before stopping the Manager so no in-flight Emit lands after the workers have exited.

If events is empty (the variadic), the bridge defaults to the three entity lifecycle events (entity.created / entity.updated / entity.deleted). Pass the explicit list when your app emits a custom taxonomy.

The event's Data field is marshalled to JSON for the webhook payload. Marshal and publish failures are silently dropped by default; install callbacks via WithBridgeMarshalError / WithBridgePublishError to observe them.

func BridgeWithOptions 

func BridgeWithOptions(bus *event.EventBus, mgr *Manager, events []string, opts ...BridgeOption) (cancel func())

BridgeWithOptions is the option-aware variant of Bridge. The events list slot is explicit (use nil/empty for the lifecycle default) so the trailing variadic stays available for BridgeOption.

func IngestHandler added in v0.15.0 

func IngestHandler(cfg IngestConfig) (http.Handler, error)

IngestHandler builds the inbound ingestion HTTP handler.

Flow: non-POST → 405; oversize body → 413; verification failure → 401 with a generic body (the unverified payload is NEVER persisted); a seen dedupe key → 200 (idempotent redelivery ack); otherwise the verified envelope is persisted as "received", optionally enqueued, and the request is acked with 202 + {"id": "<envelope id>"}.

With a Queue wired, the dedupe key is registered on the envelope only AFTER the enqueue succeeds: a never-enqueued envelope therefore can never dedupe-ack a redelivery, no matter which cleanup step fails. On enqueue failure the handler responds 500 (the sender will retry) and best-effort marks the just-persisted envelope "failed" with LastError as a forensic record; the retry re-persists and re-enqueues a fresh copy. Without a Queue, persistence alone is durable acceptance, so the key is stored with the envelope up front.

func ProcessInbound added in v0.15.0 

func ProcessInbound(store InboundStore, fn func(ctx context.Context, e InboundEnvelope) error) queue.Handler

ProcessInbound adapts a business handler into a queue.Handler that loads the envelope referenced by the job payload and drives its lifecycle:

  • received → processing (Attempts++), then the handler runs
  • success → processed
  • failure → failed (LastError set), and fn's error is returned so the queue's retry/backoff machinery can reschedule it

A job whose payload won't decode or whose envelope is missing returns an error (non-retryable data corruption — the queue will retry per its own dead-letter policy).

func SignWithTimestamp 

func SignWithTimestamp(secret string, unixTimestamp int64, body []byte) string

SignWithTimestamp returns a `t=<unix>,v1=<hex-hmac>` signature header where the HMAC covers `<unix>.<body>`. Binding the timestamp into the signed material lets receivers reject replays via VerifyTimestamped.

Pass time.Now().Unix() for fresh signatures; tests may pin to a specific instant.

func VerifyTimestamped 

func VerifyTimestamped(secret, header string, body []byte, tolerance time.Duration) bool

VerifyTimestamped accepts a header in the `t=<unix>,v1=<hex>` form and a tolerance window. Returns true iff:

  • the header parses cleanly
  • |now - ts| <= tolerance
  • the HMAC over <ts>.<body> equals v1

`now` is captured at call time so receivers don't need a clock arg.

tolerance MUST be positive. A non-positive tolerance is treated as a usage error and always rejects — otherwise a caller that accidentally passed 0 (zero value of a forgotten config field) would silently disable the replay check. Use DefaultTimestampTolerance for the suggested default.

An empty secret always rejects.

Types

type BridgeOption 

type BridgeOption func(*bridgeOpts)

BridgeOption configures the auto-bridge from a framework event bus to a webhook Manager.

func WithBridgeMarshalError 

func WithBridgeMarshalError(fn func(eventType string, err error)) BridgeOption

WithBridgeMarshalError installs a callback invoked when json.Marshal fails on an event payload. The default is a silent drop (so a buggy emitter cannot take the bus down); supply a callback to surface the failure to logs, metrics, or an alerting hook.

func WithBridgePublishError 

func WithBridgePublishError(fn func(eventType string, err error)) BridgeOption

WithBridgePublishError installs a callback invoked when Manager.Publish itself returns an error (typically a store write failure). Same default-silent rationale as marshal error.

type Delivery 

type Delivery struct {
	ID            string
	SubscriberID  string
	Event         string
	Payload       []byte
	Attempts      int
	Status        DeliveryStatus
	LastError     string
	NextAttemptAt time.Time
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

Delivery is one attempt-set against a Subscriber for a single event.

type DeliveryStatus 

type DeliveryStatus string

DeliveryStatus is the lifecycle state of a single delivery attempt. 

const (
	StatusPending DeliveryStatus = "pending"
	StatusSuccess DeliveryStatus = "success"
	StatusFailed  DeliveryStatus = "failed"
	StatusDead    DeliveryStatus = "dead"
)

type InboundEnvelope added in v0.15.0 

type InboundEnvelope struct {
	ID         string
	Source     string            // config-supplied source label, e.g. "github"
	DedupeKey  string            // empty = no dedupe for this request
	Headers    map[string]string // allowlisted subset only
	Payload    []byte
	Status     string
	Attempts   int
	LastError  string
	ReceivedAt time.Time
	UpdatedAt  time.Time
}

InboundEnvelope is a single persisted inbound webhook request.

Status walks received → processing → processed (or → failed). Payload is the raw verified request body; Headers is ONLY the allowlisted subset (see IngestConfig.KeepHeaders), never the full request header set.

type InboundSQLOption added in v0.15.0 

type InboundSQLOption func(*SQLInboundStore)

InboundSQLOption configures SQLInboundStore.

func WithInboundTable added in v0.15.0 

func WithInboundTable(name string) InboundSQLOption

WithInboundTable overrides the default "webhook_inbound" table name.

type InboundStore added in v0.15.0 

type InboundStore interface {
	AddEnvelope(ctx context.Context, e InboundEnvelope) error
	GetEnvelope(ctx context.Context, id string) (*InboundEnvelope, error)
	UpdateEnvelope(ctx context.Context, e InboundEnvelope) error
	ListEnvelopes(ctx context.Context, status string, limit int) ([]InboundEnvelope, error)
	// SeenDedupeKey reports whether a key was already persisted for this
	// source. Empty key must return (false, nil).
	SeenDedupeKey(ctx context.Context, source, key string) (bool, error)
}

InboundStore persists inbound envelopes. Implementations return (nil, nil) from GetEnvelope for an unknown ID — matching the outbound Store.GetSubscriber convention so callers can branch on a nil pointer without inspecting an error.

type InboundVerifier added in v0.15.0 

type InboundVerifier func(r *http.Request, body []byte) error

InboundVerifier authenticates a request. Return a non-nil error to reject with 401. Implementations MUST be constant-time on secrets — the built-in TimestampedVerifier and HMACSHA256Verifier use hmac.Equal.

func HMACSHA256Verifier added in v0.15.0 

func HMACSHA256Verifier(header, prefix, secret string) InboundVerifier

HMACSHA256Verifier authenticates GitHub-style requests: the header carries "<prefix><hex-hmac-sha256-of-body>" (e.g. header "X-Hub-Signature-256", prefix "sha256="). Comparison is hmac.Equal.

NOTE: the body alone is signed — there is no timestamp binding, so this offers no replay defense. Use TimestampedVerifier when the sender supports it. A missing header also rejects (returns errVerifyFailed).

func TimestampedVerifier added in v0.15.0 

func TimestampedVerifier(secret string, tolerance time.Duration) InboundVerifier

TimestampedVerifier wraps VerifyTimestamped over the X-GoFastr-Signature header (SignatureHeader). It binds the timestamp into the signed material, so a captured request cannot replay past `tolerance`. Prefer this over HMACSHA256Verifier whenever the sender supports it.

type IngestConfig added in v0.15.0 

type IngestConfig struct {
	Source       string          // required, non-empty
	Verifier     InboundVerifier // required
	Store        InboundStore    // required
	Queue        queue.Queue     // optional; nil = persist only
	JobType      string          // required if Queue != nil, e.g. "webhook.inbound"
	MaxBodyBytes int64           // default 1 MiB
	KeepHeaders  []string        // header allowlist to persist; default none
	// DedupeKeyFunc extracts a provider idempotency key (e.g.
	// X-GitHub-Delivery). Empty return = no dedupe for that request.
	DedupeKeyFunc func(r *http.Request, body []byte) string
	// Logger receives operational warnings the response can't carry (e.g. a
	// best-effort store update failing after an enqueue error). Default:
	// log.Printf. Set to a no-op func to silence.
	Logger func(format string, args ...any)
}

IngestConfig configures IngestHandler. Source, Verifier, and Store are required; JobType is required when Queue is non-nil.

type LeasedStore 

type LeasedStore interface {
	Store
	ClaimDueDeliveries(ctx context.Context, now time.Time, limit int, leasePeriod time.Duration) ([]Delivery, error)
}

LeasedStore is the optional Store upgrade for multi-instance deployments. Implementations atomically claim up to `limit` pending rows that are due at `now` and push their next_attempt_at to `now + leasePeriod`, so concurrent workers see those rows as not yet due and skip them.

If the worker crashes mid-attempt, the lease expires and another worker re-claims. Pick a leasePeriod larger than your worst-case handler latency.

The Manager probes for this interface at tick time; stores that don't implement it fall back to the plain DueDeliveries + best-effort behaviour, which is fine for single-instance setups.

type Manager 

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

Manager owns the worker goroutine and is the entry point for Publish and subscriber management.

func New 

func New(s Store, opts Options) *Manager

New constructs a Manager with defaults applied.

func (*Manager) DeadDeliveries 

func (m *Manager) DeadDeliveries(ctx context.Context, limit int) ([]Delivery, error)

DeadDeliveries lists terminally-failed deliveries when the store supports it.

func (*Manager) Publish 

func (m *Manager) Publish(ctx context.Context, event string, payload []byte) (int, error)

Publish fans the event out to every active matching subscriber.

func (*Manager) Replay 

func (m *Manager) Replay(ctx context.Context, id string) error

Replay re-queues a dead delivery when the store supports it. The worker (Start) picks it up on the next tick.

func (*Manager) Start 

func (m *Manager) Start()

Start launches the worker goroutine. Safe to call once per Manager; subsequent calls are no-ops.

func (*Manager) Stop 

func (m *Manager) Stop(ctx context.Context) error

Stop signals the worker to exit, cancels any in-flight HTTP attempt via the worker's derived context, then waits for the worker to return. If ctx fires first, returns ctx.Err — but the cancellation of in-flight HTTP requests has already been signaled. Safe to call more than once.

func (*Manager) Subscribe 

func (m *Manager) Subscribe(ctx context.Context, s Subscriber) (Subscriber, error)

Subscribe registers (or replaces) a Subscriber. The ID is assigned when empty. The URL must point at a public endpoint unless Options.AllowPrivateNetworks is set; otherwise the call returns an SSRF error.

Active defaults to true for the zero-value struct (the common case "register and start delivering"). To register a paused subscriber, set Paused=true and Active will be forced false.

func (*Manager) Subscribers 

func (m *Manager) Subscribers(ctx context.Context) ([]Subscriber, error)

Subscribers returns every registered Subscriber.

func (*Manager) Unsubscribe 

func (m *Manager) Unsubscribe(ctx context.Context, id string) error

Unsubscribe removes a Subscriber. Missing IDs are not an error.

type MemoryInboundStore added in v0.15.0 

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

MemoryInboundStore is the in-process InboundStore. Envelopes live in a map guarded by a single mutex — suitable for single-instance apps and tests; nothing is persistent. Mirrors MemoryStore's conventions: a mutex, a map, upsert-on-add, nil-pointer-on-miss.

func NewMemoryInboundStore added in v0.15.0 

func NewMemoryInboundStore() *MemoryInboundStore

NewMemoryInboundStore creates an empty store.

func (*MemoryInboundStore) AddEnvelope added in v0.15.0 

func (m *MemoryInboundStore) AddEnvelope(_ context.Context, e InboundEnvelope) error

AddEnvelope stores e, replacing any existing record with the same ID (upsert semantics, matching MemoryStore.AddSubscriber).

func (*MemoryInboundStore) GetEnvelope added in v0.15.0 

func (m *MemoryInboundStore) GetEnvelope(_ context.Context, id string) (*InboundEnvelope, error)

GetEnvelope returns (nil, nil) when the ID is unknown.

func (*MemoryInboundStore) ListEnvelopes added in v0.15.0 

func (m *MemoryInboundStore) ListEnvelopes(_ context.Context, status string, limit int) ([]InboundEnvelope, error)

ListEnvelopes returns envelopes filtered by status (empty = all), newest-received first, capped at limit (0 = no cap).

func (*MemoryInboundStore) SeenDedupeKey added in v0.15.0 

func (m *MemoryInboundStore) SeenDedupeKey(_ context.Context, source, key string) (bool, error)

SeenDedupeKey reports whether a key was already persisted for source. Empty key returns (false, nil) without scanning.

NOTE: linear scan with no DB constraint — this is the race window the SQL store documents too: a concurrent request between SeenDedupeKey and AddEnvelope can still double-insert. Acceptable for single-instance use; the SQL store's index makes the lookup cheap but the check-then-insert race is inherent without a unique constraint.

func (*MemoryInboundStore) UpdateEnvelope added in v0.15.0 

func (m *MemoryInboundStore) UpdateEnvelope(_ context.Context, e InboundEnvelope) error

UpdateEnvelope overwrites the stored envelope for e.ID.

type MemoryStore 

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

MemoryStore is the bundled in-process Store. It keeps subscribers and deliveries in maps protected by a single mutex. Suitable for single-instance apps and tests; nothing is persistent.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates an empty store.

func (*MemoryStore) AddDelivery 

func (m *MemoryStore) AddDelivery(_ context.Context, d Delivery) error

func (*MemoryStore) AddSubscriber 

func (m *MemoryStore) AddSubscriber(_ context.Context, s Subscriber) error

AddSubscriber stores s, replacing any existing record with the same ID.

func (*MemoryStore) ClaimDueDeliveries 

func (m *MemoryStore) ClaimDueDeliveries(_ context.Context, now time.Time, limit int, leasePeriod time.Duration) ([]Delivery, error)

ClaimDueDeliveries reserves rows under the store's write lock and pushes their NextAttemptAt to now+leasePeriod so a concurrent claimer sees them as not-yet-due. Single-process by design — the memory store can't span instances, but exposing the same interface keeps Manager wiring uniform across store backends.

func (*MemoryStore) DeleteSubscriber 

func (m *MemoryStore) DeleteSubscriber(_ context.Context, id string) error

func (*MemoryStore) DueDeliveries 

func (m *MemoryStore) DueDeliveries(_ context.Context, now time.Time, limit int) ([]Delivery, error)

func (*MemoryStore) GetSubscriber 

func (m *MemoryStore) GetSubscriber(_ context.Context, id string) (*Subscriber, error)

GetSubscriber returns (nil, nil) when the ID is unknown.

func (*MemoryStore) ListDeadDeliveries 

func (m *MemoryStore) ListDeadDeliveries(_ context.Context, limit int) ([]Delivery, error)

ListDeadDeliveries implements ReplayableStore: terminally-failed (StatusDead) deliveries, newest-first.

func (*MemoryStore) ListDeliveries 

func (m *MemoryStore) ListDeliveries(_ context.Context, subscriberID string, limit int) ([]Delivery, error)

func (*MemoryStore) ListSubscribers 

func (m *MemoryStore) ListSubscribers(_ context.Context) ([]Subscriber, error)

func (*MemoryStore) ResetDelivery 

func (m *MemoryStore) ResetDelivery(_ context.Context, id string) error

ResetDelivery implements ReplayableStore: returns a dead delivery to pending (attempts + error cleared, due now). Only StatusDead rows are touched, so resetting a non-dead/unknown delivery is a no-op.

func (*MemoryStore) UpdateDelivery 

func (m *MemoryStore) UpdateDelivery(_ context.Context, d Delivery) error

type NoopSecretCodec 

type NoopSecretCodec struct{}

NoopSecretCodec is the default — it stores secrets as-is. Use it only when the database is itself encrypted at rest or the threat model doesn't include a DB-snapshot attacker.

func (NoopSecretCodec) Decode 

func (NoopSecretCodec) Decode(e string) (string, error)

Decode returns encoded unchanged.

func (NoopSecretCodec) Encode 

func (NoopSecretCodec) Encode(p string) (string, error)

Encode returns plaintext unchanged.

type Options 

type Options struct {
	MaxAttempts          int
	Backoff              []time.Duration
	HTTPClient           *http.Client
	PollInterval         time.Duration
	MaxResponseBodyBytes int64
	AllowPrivateNetworks bool
	SignatureTolerance   time.Duration
	LeasePeriod          time.Duration
}

Options configures the Manager.

MaxAttempts caps how many times a delivery will be retried before it becomes "dead". Default 6 (≈3h with the default backoff).

Backoff is the wait between attempts. Position [i-1] selects the wait after attempt i. The last value is reused for any further attempts. Default: 30s, 1m, 5m, 15m, 1h, 3h.

HTTPClient is the client used for outbound POSTs. Default has a 10s per-request timeout, no redirect following, and a dial-time SSRF guard (net.Dialer.Control re-checks the resolved IP at connect time to defeat DNS rebinding) unless AllowPrivateNetworks is set. A caller-supplied client is used as-is — it gets no dial-time guard.

PollInterval controls how often the worker looks for due deliveries. Default 1 second.

MaxResponseBodyBytes caps bytes read from each subscriber response. Default 64 KiB. Set < 0 to disable the cap (not recommended).

AllowPrivateNetworks opts-out of the SSRF guard that rejects subscriber URLs targeting RFC1918, loopback, link-local, or cloud metadata endpoints. Default false — required production posture. Tests and explicit dev wiring may set true.

SignatureTolerance bounds how far receivers may consider the signed timestamp drifting from "now" when verifying. The sender embeds the timestamp into the signature; this field is purely documentation here — receivers pass it to VerifyTimestamped.

LeasePeriod controls how long a claimed delivery is hidden from other workers when the store implements LeasedStore. Must be longer than the worst-case handler latency; default 30s.

type ReplayableStore 

type ReplayableStore interface {
	Store
	// ListDeadDeliveries returns up to limit terminally-failed (StatusDead)
	// deliveries, newest-first.
	ListDeadDeliveries(ctx context.Context, limit int) ([]Delivery, error)
	// ResetDelivery returns a dead delivery to pending so the worker retries
	// it (attempts + last error cleared, due immediately). Idempotent: it MUST
	// only touch StatusDead rows, so resetting a non-dead/unknown delivery is a
	// no-op — never resurrecting an in-flight or already-delivered one.
	ResetDelivery(ctx context.Context, id string) error
}

ReplayableStore is the optional Store capability for dead-letter inspection and replay. Both shipped stores (SQLStore, MemoryStore) implement it; Manager.DeadDeliveries / Manager.Replay probe for it so admin tooling can surface a replay action only when the backend supports it.

type SQLInboundStore added in v0.15.0 

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

SQLInboundStore is a SQL-backed InboundStore. It mirrors SQLStore's conventions: dialect detected at construction (sqlite vs postgres), table created on first use, headers persisted as a JSON TEXT column.

Schema: 

webhook_inbound(
    id          TEXT PRIMARY KEY,
    source      TEXT NOT NULL,
    dedupe_key  TEXT NOT NULL DEFAULT '',  -- '' means no dedupe
    headers     TEXT NOT NULL DEFAULT '{}', -- JSON object, allowlisted subset
    payload     BLOB NOT NULL,
    status      TEXT NOT NULL,
    attempts    INTEGER NOT NULL DEFAULT 0,
    last_error  TEXT NOT NULL DEFAULT '',
    received_at TIMESTAMPTZ NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL
)

Dedupe constraint: there is intentionally NO unique constraint on (source, dedupe_key). A portable partial index (Postgres) vs the NULL-coalescing trick (SQLite) can't be expressed in one DDL, and a plain unique index would forbid two legitimately-undedupe'd requests (empty key) from coexisting. Dedupe is enforced application-side via SeenDedupeKey; the (source, dedupe_key) index makes that lookup cheap. The check-then-insert race window is documented in SeenDedupeKey.

func NewSQLInboundStore added in v0.15.0 

func NewSQLInboundStore(db *sql.DB, opts ...InboundSQLOption) (*SQLInboundStore, error)

NewSQLInboundStore constructs a SQL-backed InboundStore and ensures the table exists. Dialect is detected exactly like NewSQLStore: a Postgres version() string flips the dialect; everything else defaults to sqlite.

func (*SQLInboundStore) AddEnvelope added in v0.15.0 

func (s *SQLInboundStore) AddEnvelope(ctx context.Context, e InboundEnvelope) error

AddEnvelope inserts e. Headers are JSON-encoded into the headers column.

func (*SQLInboundStore) GetEnvelope added in v0.15.0 

func (s *SQLInboundStore) GetEnvelope(ctx context.Context, id string) (*InboundEnvelope, error)

GetEnvelope returns (nil, nil) for an unknown id.

func (*SQLInboundStore) ListEnvelopes added in v0.15.0 

func (s *SQLInboundStore) ListEnvelopes(ctx context.Context, status string, limit int) ([]InboundEnvelope, error)

ListEnvelopes returns envelopes filtered by status (empty = all), newest-received first, capped at limit (0 = no cap).

func (*SQLInboundStore) SeenDedupeKey added in v0.15.0 

func (s *SQLInboundStore) SeenDedupeKey(ctx context.Context, source, key string) (bool, error)

SeenDedupeKey reports whether a (source, key) pair already exists. Empty key returns (false, nil) without hitting the DB.

As noted on the type, there is no unique constraint, so two concurrent requests with the same key can both pass this check before either inserts — the same check-then-insert race as the memory store. A unique index would close it but can't be expressed portably across sqlite and postgres without dialect branches; callers that need strict exactly-once should serialize at the source or accept the duplicate-suppression done downstream (idempotent ProcessInbound handlers).

func (*SQLInboundStore) UpdateEnvelope added in v0.15.0 

func (s *SQLInboundStore) UpdateEnvelope(ctx context.Context, e InboundEnvelope) error

UpdateEnvelope overwrites the mutable columns (status, attempts, last_error, headers, updated_at) for e.ID. Source/payload/received_at are immutable.

type SQLOption 

type SQLOption func(*SQLStore)

SQLOption configures SQLStore.

func WithSQLAllowPlaintext 

func WithSQLAllowPlaintext() SQLOption

WithSQLAllowPlaintext opts the SQLStore into plaintext secret storage using NoopSecretCodec. Without this (or an explicit WithSQLSecretCodec) NewSQLStore returns an error rather than silently storing subscriber secrets in cleartext.

func WithSQLDeliveriesTable 

func WithSQLDeliveriesTable(name string) SQLOption

WithSQLDeliveriesTable overrides the default "webhook_deliveries" table name.

func WithSQLSecretCodec 

func WithSQLSecretCodec(c SecretCodec) SQLOption

WithSQLSecretCodec installs a SecretCodec that encrypts subscriber secrets at write time and decrypts them at read time.

Use NewAESGCMSecretCodec to wrap a 16/24/32-byte key. Callers that genuinely want plaintext storage (DB encrypted at rest, threat model excludes a DB-snapshot attacker) must use WithSQLAllowPlaintext explicitly — there is no plaintext default.

func WithSQLSubscribersTable 

func WithSQLSubscribersTable(name string) SQLOption

WithSQLSubscribersTable overrides the default "webhook_subscribers" table name.

type SQLStore 

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

SQLStore is a SQL-backed Store. Subscribers and deliveries each live in a single table; allow lists and payloads are persisted in place rather than normalised, since fan-out cardinality is small and replay needs the body intact.

Dialect is detected at construction (sqlite vs postgres).

Schemas: 

webhook_subscribers(
    id       TEXT PRIMARY KEY,
    url      TEXT NOT NULL,
    secret   TEXT NOT NULL,
    events   TEXT NOT NULL,   -- JSON array
    active   INTEGER NOT NULL,
    created  TIMESTAMP NOT NULL
)

webhook_deliveries(
    id              TEXT PRIMARY KEY,
    subscriber_id   TEXT NOT NULL,
    event           TEXT NOT NULL,
    payload         BLOB NOT NULL,
    attempts        INTEGER NOT NULL,
    status          TEXT NOT NULL,
    last_error      TEXT NOT NULL,
    next_attempt_at TIMESTAMP,
    created_at      TIMESTAMP NOT NULL,
    updated_at      TIMESTAMP NOT NULL
)

func NewSQLStore 

func NewSQLStore(db *sql.DB, opts ...SQLOption) (*SQLStore, error)

NewSQLStore constructs a SQL-backed Store and ensures both tables exist.

func (*SQLStore) AddDelivery 

func (s *SQLStore) AddDelivery(ctx context.Context, d Delivery) error

func (*SQLStore) AddSubscriber 

func (s *SQLStore) AddSubscriber(ctx context.Context, sub Subscriber) error

func (*SQLStore) ClaimDueDeliveries 

func (s *SQLStore) ClaimDueDeliveries(ctx context.Context, now time.Time, limit int, leasePeriod time.Duration) ([]Delivery, error)

ClaimDueDeliveries atomically reserves up to `limit` pending rows whose next_attempt_at <= now, pushing them to now+leasePeriod so concurrent workers skip them.

Postgres uses FOR UPDATE SKIP LOCKED inside a CTE so the inner SELECT only sees uncontested rows; SQLite serializes writers via BEGIN IMMEDIATE so the SELECT-then-UPDATE sequence inside one tx is safely exclusive.

func (*SQLStore) DeleteSubscriber 

func (s *SQLStore) DeleteSubscriber(ctx context.Context, id string) error

func (*SQLStore) DueDeliveries 

func (s *SQLStore) DueDeliveries(ctx context.Context, now time.Time, limit int) ([]Delivery, error)

func (*SQLStore) GetSubscriber 

func (s *SQLStore) GetSubscriber(ctx context.Context, id string) (*Subscriber, error)

func (*SQLStore) ListDeadDeliveries 

func (s *SQLStore) ListDeadDeliveries(ctx context.Context, limit int) ([]Delivery, error)

ListDeadDeliveries implements ReplayableStore: terminally-failed (StatusDead) deliveries, newest-first.

func (*SQLStore) ListDeliveries 

func (s *SQLStore) ListDeliveries(ctx context.Context, subscriberID string, limit int) ([]Delivery, error)

func (*SQLStore) ListSubscribers 

func (s *SQLStore) ListSubscribers(ctx context.Context) ([]Subscriber, error)

func (*SQLStore) ResetDelivery 

func (s *SQLStore) ResetDelivery(ctx context.Context, id string) error

ResetDelivery implements ReplayableStore: returns a dead delivery to pending (attempts + error cleared, due now). The `AND status = dead` clause makes it idempotent — resetting a non-dead/unknown delivery matches no row and is a no-op, so it can never resurrect an in-flight or delivered one.

func (*SQLStore) UpdateDelivery 

func (s *SQLStore) UpdateDelivery(ctx context.Context, d Delivery) error

type SecretCodec 

type SecretCodec interface {
	Encode(plaintext string) (string, error)
	Decode(encoded string) (string, error)
}

SecretCodec encodes and decodes subscriber secrets when they're persisted by a Store. Implementations must be safe for concurrent use.

Encode is called at write time (AddSubscriber); Decode is called at read time (GetSubscriber / ListSubscribers). The codec is purely a storage concern — the Manager only ever sees plaintext.

func NewAESGCMSecretCodec 

func NewAESGCMSecretCodec(key []byte) (SecretCodec, error)

NewAESGCMSecretCodec constructs a SecretCodec backed by AES-GCM using the supplied key. Key length must be 16, 24, or 32 bytes (AES-128 / AES-192 / AES-256).

Treat the key as a critical secret: rotate it by re-encrypting subscribers through a transitional codec (decode-from-old, encode-to-new); the package does not bundle a key-ring abstraction to keep this primitive small.

type Store 

type Store interface {
	AddSubscriber(ctx context.Context, s Subscriber) error
	GetSubscriber(ctx context.Context, id string) (*Subscriber, error)
	ListSubscribers(ctx context.Context) ([]Subscriber, error)
	DeleteSubscriber(ctx context.Context, id string) error

	AddDelivery(ctx context.Context, d Delivery) error
	UpdateDelivery(ctx context.Context, d Delivery) error
	ListDeliveries(ctx context.Context, subscriberID string, limit int) ([]Delivery, error)
	DueDeliveries(ctx context.Context, now time.Time, limit int) ([]Delivery, error)
}

Store is the persistence interface.

type Subscriber 

type Subscriber struct {
	ID      string
	URL     string
	Secret  string
	Events  []string
	Active  bool
	Paused  bool // explicit opt-out; honored by Subscribe so callers can register paused
	Created time.Time
}

Subscriber is a registered HTTP endpoint that wants to receive signed POSTs for the listed events.

Events accept simple glob patterns: `*` matches a single segment, `**` matches one-or-more segments. An entry of `*` alone (or `**` alone) matches every event.

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.