webhook

package
v0.3.0 Latest Latest
Warning

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

Go to latest
Published: Aug 18, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package webhook queues, signs and delivers outbound event notifications.

Nothing is delivered on the request path. A consumer calls Emit, which renders one payload and fans it out to every subscribed webhook in the workspace with a single INSERT ... SELECT; the scheduler drains that table on its own clock. The table is the reason for the split, and it is the same reason the mail outbox exists (D23): a delivery that vanished because a deploy landed mid-retry would be invisible on both ends — nobody receives it, and nobody knows one was attempted.

**The queue is Postgres and Redis stays a cache.** Plan.md lists Redis Streams as an upgrade path for exactly this work. It is not taken, and this package is where that is settled: `webhook_deliveries` has shipped since 00600 with the shape a queue needs, the mail outbox already proved the claim-and-lease pattern on it, and a queue that lives in the cache is a queue that disappears when somebody flushes the cache. The upgrade path remains *unexercised* rather than adopted — nothing in the tree is written against Streams, and nothing here would have to be undone to move later.

**What makes this different from every other outbound call in the product** is that the target is chosen by whoever holds a workspace rather than by the operator. internal/feed dials one URL an operator named in configuration, and it refuses to follow redirects because *"a feed that answers 302 is a feed pointing this process somewhere nobody configured."* That sentence applies here with more force and less trust, which is why this package has a dialer of its own — see client.go, and decisions.md for why the two clients are not shared.

Index

Constants

View Source
const (
	HeaderEvent     = "X-LinkCtrl-Event"
	HeaderDelivery  = "X-LinkCtrl-Delivery"
	HeaderTimestamp = "X-LinkCtrl-Timestamp"
	HeaderSignature = "X-LinkCtrl-Signature"
)

Header names, fixed here because they are a published interface. Anything that renames one has changed what every receiver in the world verifies.

View Source
const (
	// MaxAttempts is how many deliveries one event gets before it is abandoned.
	// Counted at claim time, so a process that dies mid-delivery spends an
	// attempt — otherwise a crash loop would retry the same event forever.
	//
	// Seven rather than the mailer's five, and the extra two are the hour
	// boundary: with the backoff below, seven attempts span 61 minutes, which is
	// long enough to ride out a receiver's deploy and short enough that "we were
	// down this morning, did we lose events" has the answer "anything older than
	// an hour, yes" rather than a calculation.
	//
	// TestBackoffDoublesAndCaps holds the arithmetic to that sentence. It is
	// there because the first draft of this comment said six attempts and an
	// hour, and six is thirty-one minutes.
	MaxAttempts = 7
	// BackoffBase is the delay before the second attempt, doubling up to
	// BackoffMax. With these values the seven attempts span 61 minutes:
	// 1m, 2m, 4m, 8m, 16m, 30m.
	BackoffBase = time.Minute
	BackoffMax  = 30 * time.Minute

	// DrainBatch bounds one drain. Each row is a network round trip to somebody
	// else's server and the scheduler runs every half minute, so a backlog
	// drains over several runs instead of holding the job for minutes.
	DrainBatch = 20

	// DeliveryConcurrency is how many of a claimed batch are dialled at once,
	// and it is DrainBatch because that is what makes the sentence above true.
	//
	// There was no such constant before and the value it replaces was one, which
	// is what M42 was reopened for. A batch delivered one row after another costs
	// DrainBatch times the per-attempt timeout — twenty times ten seconds at the
	// shipped defaults, with no misconfiguration — and that time is spent on the
	// scheduler's single goroutine, where every other job's tick is dropped rather
	// than queued. The cost never landed on webhooks: it landed on invitation
	// mail, on automation's advertised clock, and on domain re-verification.
	// Delivered together, one drain costs one attempt, and the arithmetic no
	// longer has a batch size in it.
	//
	// Equal to DrainBatch rather than smaller, deliberately. Any value below it
	// puts the batch size back into the wall-clock cost — a limit of eight is
	// three waves, which at the default timeout is thirty seconds and back inside
	// the tick it is meant to fit under. What keeps the number small is the claim
	// itself: a drain never holds more than DrainBatch rows, so it can never dial
	// more than that at once.
	DeliveryConcurrency = DrainBatch

	// DefaultRetentionDays is how long a delivered or abandoned row is kept when
	// the operator has not said. The setting is WEBHOOK_RETENTION_DAYS; this is
	// the fallback, not a second policy.
	DefaultRetentionDays = 30
)

Retry policy. Bounded, and bounded on purpose: a receiver that has refused seven deliveries over an hour is not going to accept the eighth, and a queue that retries forever is one where a single dead endpoint is dialled on every tick until somebody notices.

View Source
const DefaultTimeout = 10 * time.Second

DefaultTimeout bounds one delivery attempt end to end.

Ten seconds: long enough for a receiver on another continent that does real work before answering, short enough that a batch of twenty slow ones fits well inside the job's own bound. The setting is WEBHOOK_TIMEOUT.

View Source
const SignatureVersion = "v1"

SignatureVersion prefixes the signature header value, so the scheme can change without a receiver having to guess which one it is looking at.

Variables

View Source
var ErrPrivateAddress = errors.New("webhook: refusing to connect to a private, " +
	"loopback or link-local address")

ErrPrivateAddress is what the dialer refuses with.

Its own error so the delivery row says why nothing was sent, and so a test can assert the refusal rather than assert that *some* error happened — a test that passes because the host was simply unreachable is a test that would keep passing after the guard was deleted.

Functions

func Backoff

func Backoff(attempts int) time.Duration

Backoff is the delay before attempt n+1, given that n attempts have been made. Doubling from BackoffBase, capped at BackoffMax.

No jitter: this is one leader draining one queue on a fixed tick, not N clients stampeding a service, so there is nothing to spread out. The same reasoning mail.Backoff records, and the same shape, because two spellings of one policy is one more thing to learn.

func Sign

func Sign(secret []byte, timestamp int64, payload []byte) string

Sign produces the payload signature a receiver verifies.

The scheme, for whoever is writing a receiver

signed  = "<timestamp>.<raw request body>"
digest  = HMAC-SHA256(key = the secret exactly as it was shown to you,
                      message = signed)
header  = "X-LinkCtrl-Signature: v1=" + lowercase hex of digest

Three things are worth stating because getting any of them wrong produces a signature that never matches and no way to tell why:

  • **The key is the secret string as displayed**, the 64 lowercase hex characters, used as-is. It is not hex-decoded first. This product stores 32 random bytes and shows you their hex; making the visible string the key means a receiver copies it out of the dashboard and uses it, with no encoding step to get wrong.
  • **The message is the raw body**, byte for byte as it arrived. Do not parse and re-serialize the JSON before verifying — key order and whitespace will not survive it.
  • **The timestamp is the header value**, seconds since the epoch, and it is part of what is signed. A receiver that wants replay protection rejects a timestamp too far from its own clock; a receiver that does not still has to include it in the message.

Compare with a constant-time comparison. hmac.Equal in Go, hash_equals in PHP, hmac.compare_digest in Python.

Types

type Client

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

Client delivers one signed payload to one receiver.

func NewClient

func NewClient(timeout time.Duration, rt GuardedTransport) *Client

NewClient builds the delivery client.

func (*Client) Deliver

func (c *Client) Deliver(ctx context.Context, url string, secret []byte, d Delivery) (int, error)

Deliver POSTs one signed payload and reports what the receiver said.

Returns the HTTP status code, or zero when there was no response at all — a refused connection, a timeout, or this instance declining to open the socket. A non-2xx status is an error carrying the code, so the caller records both.

type Config

type Config struct {
	// Timeout bounds one delivery attempt end to end — connect, write, read.
	// Zero takes DefaultTimeout.
	Timeout time.Duration
	// RetentionDays is how long a finished delivery is kept. Zero takes
	// DefaultRetentionDays; it is never "forever", because a table with one row
	// per link write per webhook and no window is the growth problem D5 exists
	// to stop repeating.
	RetentionDays int
	// Transport is for tests. Nil builds the guarded transport in client.go,
	// which is the only one production ever has.
	Transport GuardedTransport
	Logger    *slog.Logger
	Observer  Observer
}

Config is what a Service needs. Its own struct rather than config.Config, matching every other service in this tree: the package that does the work does not read the environment.

type Delivery

type Delivery struct {
	// ID is the delivery row, and it is the receiver's idempotency key: every
	// retry of one event carries the same value, and two events never share one.
	ID uuid.UUID
	// Event is the name from the vocabulary, also sent as a header so a receiver
	// can route without parsing the body.
	Event string
	// Payload is the rendered JSON, byte for byte as it was queued. Signed as
	// stored: re-encoding it here would produce a body whose signature a receiver
	// could not reproduce from what it received.
	Payload []byte
}

Delivery is one queued event as the client sends it.

type Emitter

type Emitter interface {
	Emit(ctx context.Context, workspaceID uuid.UUID, event string, data map[string]any)
}

Emitter is the writing half, as a consumer sees it.

internal/link holds this rather than *Service, so "no webhook delivery in this process" is a nil interface rather than a flag every call site has to remember to check — and so internal/link's tests need neither this package nor a delivery client. It is also what keeps the import graph one-way: internal/link never imports internal/webhook, and this package never imports internal/link except for the one address predicate in client.go.

type GuardedTransport

type GuardedTransport interface {
	http.RoundTripper
}

GuardedTransport is the seam tests substitute at.

An interface rather than http.RoundTripper directly, so a test cannot accidentally supply a transport that skips the guard without saying so: the name is the reminder, and the one production implementation is built below.

type Observer

type Observer interface {
	ObserveWebhookDelivery(outcome, status string)
}

Observer counts delivery outcomes. Nil counts nothing.

The label vocabulary is fixed by the implementation, not assembled here: M13's cardinality rule is that a bounded label is fine and an unbounded one is not, so an outcome and an HTTP status *class* are counted and a URL never is.

Called from several goroutines at once, because Drain delivers a batch together. A Prometheus counter already is; anything else here has to be.

type Service

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

Service is the outbox: Emit on one side, Drain on the other.

func NewService

func NewService(pool *pgxpool.Pool, cfg Config) *Service

func (*Service) Drain

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

Drain delivers everything due, and is what the scheduler calls.

One delivery's failure never stops the batch: errors are collected and the remaining rows are still attempted, because one dead receiver must not hold up everybody else's events.

**The batch goes out together, and Drain does not return until all of it has.** Both halves are load-bearing, and the second one is why this is written with a WaitGroup rather than fired and forgotten. The caller is `withLeadership` in cmd/linkctrl/jobs.go, which holds D77's advisory lock on a pooled connection it releases the moment the function it was given returns — so a goroutine that outlived this call would deliver *without* the lock, which is the duplicate delivery under split brain that D77 exists to prevent. Waiting here means the lock covers every dial, exactly as it did when they were sequential.

func (*Service) Emit

func (s *Service) Emit(ctx context.Context, workspaceID uuid.UUID, event string, data map[string]any)

Emit queues one event for every subscribed webhook in the workspace.

It runs inside a link write, so its cost where nobody has registered anything — which is every workspace on a default instance — is one indexed lookup that returns no rows. That is what the partial index `webhooks_workspace_idx ... WHERE enabled` (00600) makes it.

**It returns nothing, deliberately.** A caller that could fail because a notification could not be queued would be a caller whose link creation fails when the webhook table is unhappy. The event is a consequence of a change that has already been committed; losing the notification is worse than losing the change only if you think a webhook is the product. Logged at warn so the gap is visible to whoever goes looking, which is the same trade the audit writer makes.

func (*Service) Pending

func (s *Service) Pending(ctx context.Context) (int64, error)

Pending counts what is still queued, for tests and for anyone asking whether delivery is keeping up.

func (*Service) PurgeFinished

func (s *Service) PurgeFinished(ctx context.Context) (int64, error)

PurgeFinished deletes delivered and abandoned rows past the retention window, reporting how many went.

func (*Service) RetentionDays

func (s *Service) RetentionDays() int

RetentionDays is the configured window, for the docs endpoint and the page.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL