webhook

package
v1.0.2 Latest Latest
Warning

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

 Go to latest Published: Oct 28, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package webhook sends signed outgoing webhooks. It is a port of the former first-party webhooks plugin; the receiver contract (body, headers, signatures, secrets, retry schedule) is unchanged, see docs/WEBHOOKS.md.

Index

Constants

View Source 
const (
	EventEntityCreated   = "entity.created"
	EventEntityUpdated   = "entity.updated"
	EventEntityPublished = "entity.published"
	EventEntityDeleted   = "entity.deleted"
	EventFormSubmitted   = "form.submitted"
	EventImportCompleted = "import.completed"
	// EventTest is the synthetic event of "Send test"; it is never subscribable.
	EventTest = "webhook.test"
)

Event names a subscription may list.

View Source 
const (
	SecretPrefix     = "whsec_"
	MaxSubscriptions = 50
	// MaxResponseBody is how many bytes of a receiver's answer are stored.
	MaxResponseBody = 1024
	// MaxTimeoutSeconds is the longest request timeout a subscription may ask for.
	MaxTimeoutSeconds = 5
	MaxBackoff        = time.Hour
	// MaxPending stops queueing for a subscription whose backlog is this long.
	MaxPending = 5000
	// LogPerSubscription is how many finished deliveries each subscription keeps.
	LogPerSubscription = 1000
	// LeaseSeconds is how long a claimed delivery is left alone.
	LeaseSeconds = 60
)

Variables

View Source 
var ErrNotFound = errors.New("webhook not found")

ErrNotFound is returned when a subscription (or delivery) does not exist.

Functions

func Backoff 

func Backoff(attempt int, base time.Duration) time.Duration

Backoff is the wait before the next attempt after the attempt'th failed one: base * 2^attempt, at most MaxBackoff.

func NewEnvelope 

func NewEnvelope(id, event string, now time.Time, data any) ([]byte, error)

NewEnvelope returns the body stored for a delivery and re-sent verbatim on every attempt.

func NewSecret 

func NewSecret() (string, error)

NewSecret returns a signing secret: "whsec_" and 64 hex characters.

func Register 

func Register(h Registrar, q Enqueuer, log *slog.Logger)

Register subscribes the enqueue listeners to the entity, import and form hooks. A listener only inserts rows: it makes no network call, so a slow receiver can never make the engine drop or delay an event.

func SanitizeImported 

func SanitizeImported(s *Subscription)

SanitizeImported applies to an imported subscription the rules Validate applies to a new one: numbers are clamped, unknown and repeated events and headers the sender owns are dropped, the name falls back to the address and plain http is flagged (it worked before). A subscription that is still invalid afterwards (no events, an unusable address) is left paused.

func Sign 

func Sign(secret string, body []byte) string

Sign is the X-Webhook-Signature value: "sha256=" and the hex HMAC-SHA256 of the exact request body, keyed by the subscription's secret.

func SignV2 

func SignV2(secret string, unix int64, body []byte) string

SignV2 is the X-Webhook-Signature-V2 value: "t=<unix>,v1=<hex>" where the HMAC-SHA256 covers "<unix>.<body>", so a receiver can reject a request whose t is older than its tolerance (five minutes is usual).

func ValidEvents 

func ValidEvents() []string

ValidEvents returns the subscribable events in a stable order.

Types

type Deliverer 

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

Deliverer sends queued deliveries.

func NewDeliverer 

func NewDeliverer(repo *Repo, log *slog.Logger, o Options) *Deliverer

NewDeliverer returns a Deliverer on repo.

func (*Deliverer) Run 

func (d *Deliverer) Run(ctx context.Context) error

Run settles dead claims, then sends due deliveries one by one until none is due, runBatch were tried or the budget is nearly spent. A subscription whose request failed at the transport (timeout, refused) is left alone for the rest of the run, so one hung receiver cannot starve the others.

func (*Deliverer) SendTest 

func (d *Deliverer) SendTest(ctx context.Context, subscriptionID string) (string, error)

SendTest queues a webhook.test event for one subscription, whatever events it listens to, and attempts it now. It returns the delivery id.

type Delivery 

type Delivery struct {
	ID             string
	SubscriptionID string
	Event          string
	Payload        string
	Status         string
	Attempts       int
	MaxAttempts    int
}

Delivery is a claimed row: what an attempt needs.

type DeliveryRow 

type DeliveryRow struct {
	ID             string
	SubscriptionID string
	Event          string
	Status         string
	Attempts       int
	MaxAttempts    int
	NextAttemptAt  time.Time
	ResponseStatus *int
	ResponseBody   string
	Error          string
	DurationMS     *int
	CreatedAt      time.Time
	LastAttemptAt  *time.Time
	DeliveredAt    *time.Time
}

DeliveryRow is a row of the delivery log. It never carries the payload.

type Enqueuer 

type Enqueuer interface {
	Enqueue(ctx context.Context, event string, data any, now time.Time) (int, error)
}

Enqueuer queues one event for the matching subscriptions (*Repo).

type Envelope 

type Envelope struct {
	ID        string `json:"id"`
	Event     string `json:"event"`
	Timestamp string `json:"timestamp"` // RFC3339 UTC, enqueue time
	Data      any    `json:"data"`
}

Envelope is the JSON body of every webhook request. The field order is part of the receiver contract: id, event, timestamp, data.

type ListedSubscription 

type ListedSubscription struct {
	ID                string
	Name              string
	URL               string
	Events            []string
	Headers           map[string]string
	Active            bool
	MaxRetries        int
	RetryDelaySeconds int
	TimeoutSeconds    int
	AllowHTTP         bool
	CreatedAt         time.Time
	UpdatedAt         time.Time
	Stats
}

ListedSubscription is a subscription as shown in lists. It has no Secret field at all: the secret is shown once, on creation or rotation.

type Options 

type Options struct {
	// AllowPrivate lets the sender reach loopback/private addresses and any
	// port. For tests only: production wiring never sets it.
	AllowPrivate bool
	Lookup       safefetch.Lookup
	Now          func() time.Time
	// RunBudget is how long one Run spends sending; default 25s.
	RunBudget time.Duration
}

Options configures a Deliverer.

type PluginSource 

type PluginSource interface {
	Query(ctx context.Context, plugin, sql string, args ...any) (*pluginhost.QueryResult, error)
}

PluginSource reads the former webhooks plugin's own schema. pluginhost.PluginDB is one: it runs the statement as the plugin's database role.

type Registrar 

type Registrar interface {
	AddAction(name string, priority int, fn func(context.Context, any))
}

Registrar is the part of the hook engine Register needs.

type Repo 

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

Repo is the Postgres access to hatamoto_webhooks and hatamoto_webhook_deliveries.

func NewRepo 

func NewRepo(db *database.DB) *Repo

NewRepo returns a Repo on db.

func (*Repo) ClaimOne 

func (r *Repo) ClaimOne(ctx context.Context, runStart time.Time, skip []string) (*Delivery, error)

ClaimOne leases the oldest due delivery of a subscription not in skip, or returns nil. Only rows due at runStart qualify, so a delivery that failed with no backoff is not tried again in the same run. Claiming is atomic (FOR UPDATE SKIP LOCKED), so concurrent runs never send one delivery twice, and one row at a time, so a run that stops early leaves nothing leased.

func (*Repo) Count 

func (r *Repo) Count(ctx context.Context) (int, error)

Count returns the number of subscriptions.

func (*Repo) CountDeliveries 

func (r *Repo) CountDeliveries(ctx context.Context, subscriptionID string) (int, error)

CountDeliveries returns how many log rows ListDeliveries pages through, for one subscription or (empty id) all.

func (*Repo) CountFailedSince 

func (r *Repo) CountFailedSince(ctx context.Context, since time.Time) (int, error)

CountFailedSince returns how many deliveries failed for good and were queued at or after since, for the Overview's site health.

func (*Repo) DBNow 

func (r *Repo) DBNow(ctx context.Context) (time.Time, error)

DBNow is the database clock, which decides what is due (not the app's).

func (*Repo) Delete 

func (r *Repo) Delete(ctx context.Context, id string) error

Delete removes a subscription; its deliveries go with it (ON DELETE CASCADE).

func (*Repo) Enqueue 

func (r *Repo) Enqueue(ctx context.Context, event string, data any, now time.Time) (int, error)

Enqueue queues event for every active subscription listening to it, one delivery row (with its own envelope id) each, and returns how many rows it added. It only touches the database: delivery is the job's business. A subscription whose backlog has reached MaxPending gets no more.

func (*Repo) Get 

func (r *Repo) Get(ctx context.Context, id string) (*Subscription, error)

Get returns one subscription including its secret (the deliverer signs with it).

func (*Repo) GetDelivery 

func (r *Repo) GetDelivery(ctx context.Context, id string) (*DeliveryRow, error)

GetDelivery returns one delivery log row (never its payload), or ErrNotFound.

func (*Repo) ImportPluginSubscriptions 

func (r *Repo) ImportPluginSubscriptions(ctx context.Context, src PluginSource, paused bool) (copied int, err error)

ImportPluginSubscriptions copies the subscriptions of the former webhooks plugin into hatamoto_webhooks in one transaction, keeping ids and secrets, so receivers notice nothing. Each row is sanitised like the migration sanitises the hand-over rows. A row whose id exists already is kept as is, so running it twice copies nothing the second time. Delivery history is not copied. With paused set every row is imported paused, whatever its own flag says (the plugin was not sending, so core must not start). Nothing is written unless every row could be read.

func (*Repo) Insert 

func (r *Repo) Insert(ctx context.Context, s *Subscription) error

Insert stores a new subscription; s.ID and s.Secret must be set.

func (*Repo) InsertParked 

func (r *Repo) InsertParked(ctx context.Context, id, subscriptionID, event, payload string, maxAttempts int, park time.Duration) error

InsertParked stores a delivery that no job will pick up for the next park (Send test attempts its row inline, so a crash never leaves it due).

func (*Repo) List 

func (r *Repo) List(ctx context.Context) ([]ListedSubscription, error)

List returns every subscription, oldest first, with delivery counts. The secret is deliberately not selected.

func (*Repo) ListDeliveries 

func (r *Repo) ListDeliveries(ctx context.Context, subscriptionID string, limit, offset int) ([]DeliveryRow, error)

ListDeliveries returns the log, newest first, for one subscription or (empty id) all. limit defaults to 50 and is at most 200.

func (*Repo) MarkDelivered 

func (r *Repo) MarkDelivered(ctx context.Context, id string, status int, body string, dur time.Duration) error

MarkDelivered records a 2xx answer as the next attempt.

func (*Repo) MarkExhausted 

func (r *Repo) MarkExhausted(ctx context.Context, id string, status *int, body, errMsg string, dur time.Duration) error

MarkExhausted records the last failed attempt: the delivery is failed.

func (*Repo) MarkFailed 

func (r *Repo) MarkFailed(ctx context.Context, id, errMsg string) error

MarkFailed fails a delivery without counting an attempt (its subscription is gone or paused).

func (*Repo) MarkRetry 

func (r *Repo) MarkRetry(ctx context.Context, id string, status *int, body, errMsg string, dur, next time.Duration) error

MarkRetry records a failed attempt and schedules the next one after next. status is nil when the receiver gave no answer.

func (*Repo) Prune 

func (r *Repo) Prune(ctx context.Context) error

Prune keeps the delivery log bounded: delivered rows live 30 days, every other row 90 days (a pending row older than that is abandoned), and each subscription keeps at most LogPerSubscription finished rows.

func (*Repo) RetryFailed 

func (r *Repo) RetryFailed(ctx context.Context, id string) error

RetryFailed puts a failed delivery back in the queue for one more attempt; any other delivery is left alone.

func (*Repo) SetActive 

func (r *Repo) SetActive(ctx context.Context, id string, active bool) error

SetActive pauses or resumes a subscription.

func (*Repo) SetSecret 

func (r *Repo) SetSecret(ctx context.Context, id, secret string) error

SetSecret replaces the signing secret.

func (*Repo) SettleDead 

func (r *Repo) SettleDead(ctx context.Context) (int64, error)

SettleDead counts the attempt of every delivery whose lease ran out: the process that claimed it died, possibly after the receiver got the request, so the attempt is spent. A delivery that keeps killing its attempt ends up failed.

func (*Repo) Update 

func (r *Repo) Update(ctx context.Context, s *Subscription) error

Update writes the editable fields. It never writes the secret.

type Service 

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

Service is the management API of webhook subscriptions (the admin page calls it). It makes no network call.

func NewService 

func NewService(db *database.DB, log *slog.Logger) *Service

NewService returns a Service on db.

func (*Service) CountFailedSince 

func (s *Service) CountFailedSince(ctx context.Context, since time.Time) (int, error)

CountFailedSince returns how many deliveries queued at or after since have failed.

func (*Service) Create 

func (s *Service) Create(ctx context.Context, in Subscription) (Subscription, string, error)

Create validates and stores a subscription and returns it with its new secret, the only time the secret is shown. Callers start from Subscription.ApplyDefaults and fill the fields the user set.

func (*Service) Delete 

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

Delete removes a subscription and its deliveries.

func (*Service) Get 

func (s *Service) Get(ctx context.Context, id string) (*Subscription, error)

Get returns one subscription including its secret.

func (*Service) List 

func (s *Service) List(ctx context.Context) ([]ListedSubscription, error)

List returns every subscription with its delivery counts, never the secret.

func (*Service) RegenerateSecret 

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

RegenerateSecret replaces the secret and returns the new one.

func (*Service) Repo 

func (s *Service) Repo() *Repo

Repo is the service's repository, for the delivery job and the hooks.

func (*Service) RetryDelivery 

func (s *Service) RetryDelivery(ctx context.Context, id string) (bool, error)

RetryDelivery queues a failed delivery for one more attempt. It reports false, changing nothing, for a delivery that has not failed, and ErrNotFound for an unknown id.

func (*Service) SetActive 

func (s *Service) SetActive(ctx context.Context, id string, active bool) error

SetActive pauses or resumes a subscription.

func (*Service) Update 

func (s *Service) Update(ctx context.Context, id string, p SubscriptionPatch) error

Update merges patch onto the stored subscription and re-validates it. The id and the secret are not editable.

type Stats 

type Stats struct {
	Delivered int `json:"delivered"`
	Failed    int `json:"failed"`
	Pending   int `json:"pending"`
}

Stats are the delivery counts of one subscription.

type Subscription 

type Subscription struct {
	ID                string            `json:"id"`
	Name              string            `json:"name"`
	URL               string            `json:"url"`
	Secret            string            `json:"secret,omitempty"`
	Events            []string          `json:"events"`
	Headers           map[string]string `json:"headers"`
	Active            bool              `json:"active"`
	MaxRetries        int               `json:"max_retries"`
	RetryDelaySeconds int               `json:"retry_delay_seconds"`
	TimeoutSeconds    int               `json:"timeout_seconds"`
	AllowHTTP         bool              `json:"allow_http"`
	CreatedAt         time.Time         `json:"created_at"`
	UpdatedAt         time.Time         `json:"updated_at"`
}

Subscription is a row of hatamoto_webhooks.

func (*Subscription) ApplyDefaults 

func (s *Subscription) ApplyDefaults()

ApplyDefaults sets the values of a new subscription.

func (*Subscription) Validate 

func (s *Subscription) Validate() error

Validate normalises and checks a subscription. It returns a *ValidationError for the first problem.

type SubscriptionPatch 

type SubscriptionPatch struct {
	Name              *string
	URL               *string
	Events            *[]string
	Headers           *map[string]string
	Active            *bool
	MaxRetries        *int
	RetryDelaySeconds *int
	TimeoutSeconds    *int
	AllowHTTP         *bool
}

SubscriptionPatch is a partial update: nil fields keep their value.

type ValidationError 

type ValidationError struct{ Msg string }

ValidationError is a problem with a subscription's fields; handlers show it as a 422.

func (*ValidationError) Error 

func (e *ValidationError) Error() string

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.