Documentation
¶
Overview ¶
Package hookpost sends reliable, signed outgoing webhooks from Go applications backed by Postgres: transactional enqueue, Standard Webhooks signing, retries, dead letters and replay. No extra service, no extra queue.
Overview ¶
Create one Client per process, run its dispatcher in the background, and call Send inside the same transaction as your business writes:
wh, _ := hookpost.New(pool, hookpost.Config{})
_ = wh.Migrate(ctx)
go wh.Run(ctx)
tx, _ := pool.Begin(ctx)
// ... create the order on tx ...
_, _ = wh.Send(ctx, tx, hookpost.Message{
OwnerID: shopID, EventType: "order.created", Payload: order,
})
_ = tx.Commit(ctx)
Customers register endpoints (CreateEndpoint, or the JSON API from AdminHandler) and verify what they receive with the zero-dependency signature package, or any Standard Webhooks library in any language.
Guarantees ¶
- Transactional: Send only writes rows. If the transaction rolls back, nothing is sent; once it commits, the webhook survives crashes and restarts until it is delivered or dead.
- At-least-once: a delivery may arrive more than once (for example when a process dies mid-request). The webhook-id header is the same on every attempt and replay; receivers should dedupe on it.
- Horizontally scalable: Run on every instance. Deliveries are leased with FOR UPDATE SKIP LOCKED, so instances never send the same attempt concurrently and never block each other.
- Retries: failed attempts are retried on Config.Schedule (8 attempts over about 27 hours by default), then marked dead. Dead deliveries can be retried, recovered or resent.
- SSRF-safe: by default the HTTP client refuses loopback, private, link-local and other internal addresses, checked after DNS resolution, and never follows redirects.
Ordering ¶
Deliveries are not ordered. A retried message can arrive after a newer one; include a timestamp or version in your payload if order matters.
Migrations ¶
Migrate creates hookpost's tables (all prefixed "hookpost_"). Teams that manage DDL themselves can copy migrations/*.sql into their own tool instead and never call Migrate.
Index ¶
- Constants
- Variables
- type AdminOptions
- type Attempt
- type AttemptInfo
- type Client
- func (c *Client) AdminHandler(opts AdminOptions) http.Handler
- func (c *Client) CreateEndpoint(ctx context.Context, in EndpointInput) (Endpoint, error)
- func (c *Client) DeleteEndpoint(ctx context.Context, ownerID, id string) error
- func (c *Client) GetDelivery(ctx context.Context, ownerID string, id int64) (Delivery, error)
- func (c *Client) GetEndpoint(ctx context.Context, ownerID, id string) (Endpoint, error)
- func (c *Client) ListAttempts(ctx context.Context, ownerID string, deliveryID int64) ([]Attempt, error)
- func (c *Client) ListDeliveries(ctx context.Context, ownerID, messageID string) ([]Delivery, error)
- func (c *Client) ListEndpoints(ctx context.Context, ownerID string) ([]Endpoint, error)
- func (c *Client) ListMessages(ctx context.Context, ownerID string, q MessageQuery) ([]MessageRecord, error)
- func (c *Client) Migrate(ctx context.Context) error
- func (c *Client) Prune(ctx context.Context, olderThan time.Duration) (int64, error)
- func (c *Client) RecoverEndpoint(ctx context.Context, ownerID, endpointID string, since time.Time) (int64, error)
- func (c *Client) ResendMessage(ctx context.Context, ownerID, messageID, endpointID string) (Delivery, error)
- func (c *Client) RetryDelivery(ctx context.Context, ownerID string, deliveryID int64) error
- func (c *Client) RotateSecret(ctx context.Context, ownerID, id string, grace time.Duration) (string, error)
- func (c *Client) Run(ctx context.Context) error
- func (c *Client) Send(ctx context.Context, db DBTX, msg Message) (SendResult, error)
- func (c *Client) SetEndpointDisabled(ctx context.Context, ownerID, id string, disabled bool) error
- func (c *Client) UpdateEndpoint(ctx context.Context, ownerID, id string, up EndpointUpdate) (Endpoint, error)
- type Config
- type DBTX
- type Delivery
- type Endpoint
- type EndpointInput
- type EndpointUpdate
- type Message
- type MessageQuery
- type MessageRecord
- type SendResult
Examples ¶
Constants ¶
const DefaultGrace = 24 * time.Hour
DefaultGrace is how long the old secret keeps signing after a rotation when the admin API is not told otherwise.
const Version = "0.1.0"
Version is the library version, used in the default User-Agent.
Variables ¶
var ( ErrNotFound = errors.New("hookpost: not found") ErrInvalidURL = errors.New("hookpost: endpoint URL must be absolute https (or http when AllowPrivateNetworks)") )
Errors returned by endpoint operations.
var ( ErrNoOwner = errors.New("hookpost: OwnerID is required") ErrNoEventType = errors.New("hookpost: EventType is required") ErrBadPayload = errors.New("hookpost: payload is not valid JSON") )
Errors returned by Send.
var DefaultSchedule = []time.Duration{ 5 * time.Second, 5 * time.Minute, 30 * time.Minute, 2 * time.Hour, 5 * time.Hour, 10 * time.Hour, 10 * time.Hour, }
DefaultSchedule is the delay after attempt 1, 2, ... giving 8 attempts over about 27.5 hours. Slice it to build a shorter schedule:
cfg.Schedule = hookpost.DefaultSchedule[:3] // 4 attempts total
var ErrBlockedAddress = errors.New("hookpost: destination address is not allowed")
ErrBlockedAddress is the error (wrapped) of attempts whose destination resolves to a private, loopback or otherwise internal address.
var ErrNoPool = errors.New("hookpost: nil pool")
ErrNoPool is returned by New when pool is nil.
Functions ¶
This section is empty.
Types ¶
type AdminOptions ¶
type AdminOptions struct {
// Owner returns the owner ID for the request, e.g. from the session.
// Return "" to reject the request with 401. Required.
Owner func(r *http.Request) string
}
AdminOptions configures AdminHandler.
type Attempt ¶
type Attempt struct {
ID int64 `json:"id"`
Attempt int `json:"attempt"`
StatusCode *int `json:"status_code"`
Error *string `json:"error"`
ResponseBody *string `json:"response_body"`
DurationMS int `json:"duration_ms"`
CreatedAt time.Time `json:"created_at"`
}
Attempt is one logged HTTP attempt of a delivery.
type AttemptInfo ¶
type AttemptInfo struct {
DeliveryID int64
MessageID string
EndpointID string
EventType string
Attempt int
StatusCode int // 0 when no response was received
Err error
Duration time.Duration
// Dead is true when this attempt failed and no retries remain.
Dead bool
}
AttemptInfo describes one delivery attempt and is passed to Config.OnAttempt.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client sends webhooks. It is safe for concurrent use; create one per process.
func New ¶
New builds a Client. It does not start the dispatcher; call Run for that. The caller owns pool and is responsible for closing it.
Example ¶
package main
import (
"context"
"log"
"os"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/ahmadraza-khan/hookpost"
)
func main() {
ctx := context.Background()
pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
defer pool.Close()
wh, err := hookpost.New(pool, hookpost.Config{})
if err != nil {
log.Fatal(err)
}
if err := wh.Migrate(ctx); err != nil {
log.Fatal(err)
}
go func() { _ = wh.Run(ctx) }() // background dispatcher
}
Output:
func (*Client) AdminHandler ¶
func (c *Client) AdminHandler(opts AdminOptions) http.Handler
AdminHandler returns a JSON API for managing endpoints and inspecting, retrying and replaying deliveries. Mount it behind your own auth; every route is scoped to the owner returned by opts.Owner:
mux.Handle("/api/webhooks/", http.StripPrefix("/api/webhooks",
wh.AdminHandler(hookpost.AdminOptions{
Owner: func(r *http.Request) string { return auth.ShopID(r) },
})))
Routes:
GET /endpoints list endpoints
POST /endpoints create; the only response with the secret
GET /endpoints/{id} get
PATCH /endpoints/{id} update url, description, event_types
DELETE /endpoints/{id} delete
POST /endpoints/{id}/enable re-enable (e.g. after a 410)
POST /endpoints/{id}/disable disable
POST /endpoints/{id}/rotate-secret {"grace_seconds": 86400} -> {"secret": ...}
POST /endpoints/{id}/recover re-queue dead deliveries (?since=RFC3339)
GET /messages ?event_type=&limit=&before=<message id>
GET /messages/{id}/deliveries delivery status per endpoint
POST /messages/{id}/resend {"endpoint_id": "..."} replay to one endpoint
GET /deliveries/{id} get
GET /deliveries/{id}/attempts attempt log
POST /deliveries/{id}/retry re-queue now (works on dead deliveries)
List responses are {"data": [...]}; /messages adds "next_before" when there may be more. Errors are {"error": "..."}.
Example ¶
package main
import (
"log"
"net/http"
"github.com/ahmadraza-khan/hookpost"
)
func main() {
var wh *hookpost.Client // from hookpost.New
shopID := func(r *http.Request) string {
return r.Header.Get("X-Shop-ID") // use your real session/auth here
}
mux := http.NewServeMux()
mux.Handle("/api/webhooks/", http.StripPrefix("/api/webhooks",
wh.AdminHandler(hookpost.AdminOptions{Owner: shopID})))
log.Fatal(http.ListenAndServe(":8080", mux))
}
Output:
func (*Client) CreateEndpoint ¶
CreateEndpoint registers a new endpoint with a fresh signing secret. The returned Endpoint.Secret is the only time the secret is exposed; hand it to the customer.
Example ¶
package main
import (
"context"
"fmt"
"log"
"github.com/ahmadraza-khan/hookpost"
)
func main() {
var wh *hookpost.Client // from hookpost.New
ep, err := wh.CreateEndpoint(context.Background(), hookpost.EndpointInput{
OwnerID: "shop_42",
URL: "https://example.com/webhooks",
EventTypes: []string{"order.created", "order.paid"}, // empty = all events
})
if err != nil {
log.Fatal(err)
}
// The secret is only returned here; give it to the customer.
fmt.Println(ep.ID, ep.Secret)
}
Output:
func (*Client) DeleteEndpoint ¶
DeleteEndpoint removes an endpoint along with its deliveries.
func (*Client) GetDelivery ¶
GetDelivery returns one of ownerID's deliveries.
func (*Client) GetEndpoint ¶
GetEndpoint returns one of ownerID's endpoints, without its secret.
func (*Client) ListAttempts ¶
func (c *Client) ListAttempts(ctx context.Context, ownerID string, deliveryID int64) ([]Attempt, error)
ListAttempts returns the attempt log of one of ownerID's deliveries, or ErrNotFound if the delivery doesn't exist.
func (*Client) ListDeliveries ¶
ListDeliveries returns the per-endpoint deliveries of one of ownerID's messages, or ErrNotFound if the message doesn't exist.
func (*Client) ListEndpoints ¶
ListEndpoints returns all of ownerID's endpoints, oldest first, without their secrets.
func (*Client) ListMessages ¶
func (c *Client) ListMessages(ctx context.Context, ownerID string, q MessageQuery) ([]MessageRecord, error)
ListMessages returns ownerID's messages, newest first, using keyset pagination on (created_at, id).
func (*Client) Migrate ¶
Migrate creates or upgrades hookpost's tables. It is safe to call on every startup and from several instances at once (it takes an advisory lock).
Teams that manage DDL with their own tool can instead copy migrations/*.sql into it and never call Migrate.
func (*Client) Prune ¶
Prune deletes messages older than olderThan that have no pending deliveries, along with their deliveries and attempts. Run calls it hourly when Config.Retention > 0. It returns the number of messages deleted.
func (*Client) RecoverEndpoint ¶
func (c *Client) RecoverEndpoint(ctx context.Context, ownerID, endpointID string, since time.Time) (int64, error)
RecoverEndpoint re-queues every dead delivery of an endpoint created at or after since (zero = all), each with a fresh retry schedule, e.g. after the customer fixed an outage. It returns how many deliveries were re-queued.
func (*Client) ResendMessage ¶
func (c *Client) ResendMessage(ctx context.Context, ownerID, messageID, endpointID string) (Delivery, error)
ResendMessage (re)queues a message to one endpoint with a fresh retry schedule, e.g. to replay a past event to a newly added endpoint. Both must belong to ownerID.
func (*Client) RetryDelivery ¶
RetryDelivery re-queues a delivery for immediate sending, including dead ones. The attempt counter is kept, so a dead delivery gets exactly one more try rather than a whole new retry schedule; use RecoverEndpoint or ResendMessage for a fresh schedule.
func (*Client) RotateSecret ¶
func (c *Client) RotateSecret(ctx context.Context, ownerID, id string, grace time.Duration) (string, error)
RotateSecret issues and returns a new secret for an endpoint. For the grace period, requests are signed with both the new and the old secret, so the customer's existing verifier keeps working until they deploy the new one.
func (*Client) Run ¶
Run delivers webhooks until ctx is cancelled, then waits for in-flight requests to finish and be recorded before returning. Call it once per process; running it on several instances against one database is safe, since deliveries are leased with FOR UPDATE SKIP LOCKED.
Run holds one pool connection for LISTEN (unless Config.DisableListen) plus up to Config.Workers connections while recording results.
Delivery is at-least-once: if a process dies mid-request, the delivery is sent again once its lease (Timeout + 30s) expires, with the same webhook-id. Receivers should dedupe on that header.
func (*Client) Send ¶
Send records msg and queues one delivery per matching, enabled endpoint. It never does HTTP; the dispatcher started by Run delivers later.
Pass a pgx.Tx to make the webhook part of your transaction: it is only sent if the transaction commits. Pass the pool for fire-and-forget.
An endpoint matches when it belongs to OwnerID, is enabled, and lists EventType (exact match) or no event types at all. A message with no matching endpoints is still stored (Endpoints == 0) so it can be replayed to an endpoint added later.
Example ¶
package main
import (
"context"
"log"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/ahmadraza-khan/hookpost"
)
func main() {
var (
ctx = context.Background()
pool *pgxpool.Pool // your pool
wh *hookpost.Client // from hookpost.New
)
tx, err := pool.Begin(ctx)
if err != nil {
log.Fatal(err)
}
defer func() { _ = tx.Rollback(ctx) }()
// ... your business writes on tx, e.g. INSERT INTO orders ...
// Only delivered if the transaction commits.
_, err = wh.Send(ctx, tx, hookpost.Message{
ID: "order-123-created", // optional: makes Send idempotent
OwnerID: "shop_42",
EventType: "order.created",
Payload: map[string]any{"order_id": 123, "total": "9.99"},
})
if err != nil {
log.Fatal(err)
}
if err := tx.Commit(ctx); err != nil {
log.Fatal(err)
}
}
Output:
func (*Client) SetEndpointDisabled ¶
SetEndpointDisabled disables or re-enables an endpoint. A disabled endpoint gets no new deliveries; ones already queued stay pending and are sent once it is enabled again.
func (*Client) UpdateEndpoint ¶
func (c *Client) UpdateEndpoint(ctx context.Context, ownerID, id string, up EndpointUpdate) (Endpoint, error)
UpdateEndpoint changes the non-nil fields of up.
type Config ¶
type Config struct {
// Workers is the number of concurrent deliveries per process. Default 10.
Workers int
// PollInterval is how often idle workers check for due deliveries. Default 1s.
PollInterval time.Duration
// BatchSize is the most deliveries leased per poll. Default = Workers.
BatchSize int
// Timeout is the per-request HTTP timeout. Default 15s.
Timeout time.Duration
// Schedule is the retry delays after each failed attempt, so
// len(Schedule)+1 is the maximum number of attempts. Default DefaultSchedule.
// An empty, non-nil slice means a single attempt with no retries.
Schedule []time.Duration
// MaxResponseBytes is how much of each response body is stored. Default 4096.
MaxResponseBytes int64
// UserAgent header sent with every request. Default "hookpost/<Version>".
UserAgent string
// RawPayload sends the payload as-is instead of wrapping it in the
// {"type","timestamp","data"} envelope.
RawPayload bool
// AllowPrivateNetworks disables the SSRF guard and allows http:// URLs.
// Only for local development and tests.
AllowPrivateNetworks bool
// HTTPClient replaces the built-in client entirely; you then own SSRF safety.
HTTPClient *http.Client
// DisableListen turns off LISTEN/NOTIFY wake-ups (e.g. behind PgBouncer in
// transaction mode). Deliveries are then picked up on the next poll.
DisableListen bool
// Retention is how long finished messages are kept before Run prunes
// them. Default 30 days; negative disables pruning.
Retention time.Duration
// Logger for dispatcher errors. Default slog.Default().
Logger *slog.Logger
// OnAttempt is called after every delivery attempt (e.g. for metrics).
// It runs on the delivery goroutine, so keep it fast.
OnAttempt func(AttemptInfo)
}
Config controls delivery behaviour. Zero values get sensible defaults. Config is copied by New; changing it afterwards has no effect.
type DBTX ¶
type DBTX interface {
Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}
DBTX is satisfied by *pgxpool.Pool, *pgx.Conn and pgx.Tx, so Send can run inside the caller's own transaction.
type Delivery ¶
type Delivery struct {
ID int64 `json:"id"`
MessageID string `json:"message_id"`
EndpointID string `json:"endpoint_id"`
Status string `json:"status"` // pending, succeeded or dead
Attempts int `json:"attempts"`
NextAttemptAt time.Time `json:"next_attempt_at"`
LastStatusCode *int `json:"last_status_code"`
LastError *string `json:"last_error"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
Delivery is the state of one message for one endpoint.
type Endpoint ¶
type Endpoint struct {
ID string `json:"id"`
OwnerID string `json:"owner_id"`
URL string `json:"url"`
Description string `json:"description"`
EventTypes []string `json:"event_types"` // empty = all events
Disabled bool `json:"disabled"`
DisabledReason string `json:"disabled_reason,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
// Secret is only populated by CreateEndpoint; use RotateSecret to issue
// a new one.
Secret string `json:"secret,omitempty"`
}
Endpoint is a URL an owner receives webhooks at.
type EndpointInput ¶
type EndpointInput struct {
OwnerID string
URL string
Description string
EventTypes []string // empty = all events (exact match only)
}
EndpointInput is the input to CreateEndpoint.
type EndpointUpdate ¶
EndpointUpdate is the input to UpdateEndpoint. Nil fields are unchanged.
type Message ¶
type Message struct {
// ID is optional. Set it to make Send idempotent (e.g. "order-123-created"):
// sending the same ID twice is a no-op the second time. It becomes the
// webhook-id header, so it must be unique across all owners.
ID string
// OwnerID selects whose endpoints receive the message. Required.
OwnerID string
// EventType, e.g. "order.created". Required.
EventType string
// Payload is JSON-encoded unless it is already []byte or json.RawMessage,
// in which case it must be valid JSON and is sent byte for byte.
Payload any
}
Message is one event to deliver to every matching endpoint of OwnerID.
type MessageQuery ¶
type MessageQuery struct {
EventType string // optional exact match
// Before is a message ID; only messages older than it are returned.
// Pass the last ID of the previous page to get the next one.
Before string
Limit int // default 50, max 250
}
MessageQuery filters ListMessages.
type MessageRecord ¶
type MessageRecord struct {
ID string `json:"id"`
EventType string `json:"event_type"`
Payload json.RawMessage `json:"payload"` // the exact body sent to receivers
CreatedAt time.Time `json:"created_at"`
}
MessageRecord is a stored message, as returned by ListMessages.
type SendResult ¶
type SendResult struct {
MessageID string
// Endpoints is how many deliveries were queued.
Endpoints int
// Duplicate is true when a message with this ID already existed and
// nothing new was queued.
Duplicate bool
}
SendResult reports what Send did.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
examples
|
|
|
receiver
command
Command receiver is a minimal webhook receiver that verifies Standard Webhooks signatures with hookpost/signature.
|
Command receiver is a minimal webhook receiver that verifies Standard Webhooks signatures with hookpost/signature. |
|
sender
command
Command sender is a runnable hookpost demo: it registers an endpoint for a local receiver and sends one "order.created" event per second.
|
Command sender is a runnable hookpost demo: it registers an endpoint for a local receiver and sends one "order.created" event per second. |
|
internal
|
|
|
ids
Package ids generates prefixed, time-sortable, URL-safe identifiers.
|
Package ids generates prefixed, time-sortable, URL-safe identifiers. |
|
Package signature signs and verifies webhooks per the Standard Webhooks specification (https://www.standardwebhooks.com).
|
Package signature signs and verifies webhooks per the Standard Webhooks specification (https://www.standardwebhooks.com). |