webhookd

package module
v0.3.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: 18 Imported by: 0

README

webhookd (Go)

Official Go SDK for NimbusNexus Webhooks — publish events, manage your endpoints / keys / deliveries, and verify the webhooks you receive. Zero dependencies (standard library only); Go ≥ 1.22.

go get github.com/NimbusNexus/webhookd-go

import webhookd "github.com/NimbusNexus/webhookd-go"

The import path ends in /go, but the package is named webhookd — import it with the explicit webhookd alias shown above.

Verify an incoming webhook (subscribers)

When webhookd delivers a webhook it signs the body with your endpoint's signing secret. Always verify the signature before trusting the payload — it proves the request really came from webhookd and wasn't tampered with or replayed. Pass the raw request body bytes (do not re-serialize the JSON, or the signature won't match).

package main

import (
	"io"
	"net/http"
	"strconv"

	webhookd "github.com/NimbusNexus/webhookd-go"
)

const signingSecret = "whsec_…" // the endpoint's signing secret (shown once on create)

func handleWebhook(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body) // RAW bytes — do not decode then re-encode before verifying
	if err != nil {
		http.Error(w, "bad body", http.StatusBadRequest)
		return
	}

	var opts *webhookd.VerifyOptions
	if raw := r.Header.Get("X-Webhook-Timestamp"); raw != "" {
		ts, perr := strconv.ParseInt(raw, 10, 64)
		if perr != nil {
			http.Error(w, "invalid signature", http.StatusBadRequest) // malformed timestamp
			return
		}
		opts = &webhookd.VerifyOptions{Timestamp: &ts} // enforces the 300s replay window
	}

	if !webhookd.Verify(signingSecret, body, r.Header.Get("X-Webhook-Signature"), opts) {
		http.Error(w, "invalid signature", http.StatusBadRequest) // forged, tampered, or replayed
		return
	}

	// ... trust the payload — do your work idempotently, then ack with any 2xx ...
	w.WriteHeader(http.StatusOK)
}

X-Webhook-Signature may carry a single token, or several comma-separated tokens during a signing-secret rotation — Verify accepts the request if any token matches, using a constant-time comparison. See examples/receiver for a full runnable server.

Sign(secret, body, timestamp) produces the same sha256=<hex> signature the server sends; it's mainly useful for tests — subscribers call Verify.

Publish an event (producers)

package main

import (
	"context"
	"errors"
	"fmt"

	webhookd "github.com/NimbusNexus/webhookd-go"
)

func main() {
	wh := webhookd.New("https://webhooks.example.com", "whsk_…")

	event, err := wh.Publish(context.Background(), "order.created",
		map[string]any{"order_id": "ord_123", "total": 4200},
		&webhookd.PublishOptions{IdempotencyKey: "order-123"}, // makes the publish safe to retry
	)
	if err != nil {
		var apiErr *webhookd.APIError
		if errors.As(err, &apiErr) {
			fmt.Println(apiErr.StatusCode, apiErr.Code, apiErr.Message) // the {error:{code,message}} envelope
		}
		return
	}
	fmt.Println(event.EventUID, event.DeliveriesCreated)
}

New accepts functional options: WithTimeout(time.Duration) (default 10s), WithMaxRetries(int) (default 2), and WithHTTPClient(*http.Client). Transient failures (connection errors, 429, 5xx) are retried with capped exponential backoff (a 429 honours Retry-After); other non-2xx responses return an *APIError carrying the stable {error:{code,message}} envelope. Every method takes a context.Context first.

Outbox / durable buffering (producers)

Publish calls webhookd synchronously — if webhookd is unreachable it returns an error and the event is lost. The write-first outbox decouples the two: Enqueue durably persists the event to a pluggable Store and returns IMMEDIATELY (no network); Drain (or a background drainer) ships the buffered events later. Every send carries Idempotency-Key = record.ID, so a re-drain after a crash or a lost response never double-publishes — webhookd dedupes. Delivery is at-least-once: nothing is lost while webhookd is down. The live Publish above is unchanged.

package main

import (
	"context"
	"log"
	"time"

	webhookd "github.com/NimbusNexus/webhookd-go"
	"github.com/NimbusNexus/webhookd-go/store/sqlite" // pulls in the SQLite driver
)

func main() {
	// 1. Configure a durable store (survives process restarts).
	store, err := sqlite.Open("outbox.db")
	if err != nil {
		log.Fatal(err)
	}
	defer store.Close()

	wh := webhookd.New("https://webhooks.example.com", "whsk_…",
		webhookd.WithStore(store),
		webhookd.WithMaxAttempts(10), // retry budget before a record is parked dead (default 10)
		webhookd.WithOnDead(func(r webhookd.Record) {
			log.Printf("dead-lettered %s: %v", r.ID, r.LastError)
		}),
	)
	ctx := context.Background()

	// 2. Enqueue instead of Publish — writes to the store and returns at once, NO network call.
	id, _ := wh.Enqueue(ctx, "order.created",
		map[string]any{"order_id": "ord_123", "total": 4200},
		&webhookd.EnqueueOptions{IdempotencyKey: "order-123"}) // id defaults to a fresh UUID v4
	_ = id

	// 3a. Drain on demand — returns DrainResult{Sent, Failed, Remaining}:
	wh.Drain(ctx, nil)

	// 3b. …or run a background drainer that calls Drain every 5s until StopDrainer.
	wh.StartDrainer(ctx, 5*time.Second, nil)
	// ... your app keeps enqueuing; the drainer ships in the background ...
	wh.StopDrainer() // signals the goroutine and waits for it to exit
}

Idempotency guarantee. The id returned by Enqueue is the IdempotencyKey you pass (or a generated UUID v4) and becomes the Idempotency-Key header on every delivery attempt for that record. If the process crashes after a send but before the response is recorded, the next Drain re-sends with the same key and webhookd returns the original event without re-fanning-out. A record that keeps failing is retried with capped exponential backoff up to WithMaxAttempts (default 10), then flagged dead (never retried again, kept buffered) and handed to the optional WithOnDead hook.

Built-in stores — pass one to webhookd.WithStore(...):

Store Durable? Extra needed
webhookd.NewMemoryStore() No (in-process) — (stdlib)
webhookd.NewFileStore(dir) Yes (per-record JSON, atomic write+rename) — (stdlib)
sqlite.Open(path) Yes (transactional) store/sqlite → modernc.org/sqlite (pure Go, no cgo)
redis.Open(redis.Options{URL}) Yes (sorted-set + hash under a key prefix) store/redis → github.com/redis/go-redis/v9
postgres.Open(ctx, postgres.Options{ConnString}) Yes (webhookd_outbox table, name configurable) store/postgres → github.com/jackc/pgx/v5

The core webhookd package stays stdlib-only. The three driver stores live in subpackages under store/ (store/sqlite, store/redis, store/postgres); their third-party driver dependencies are only compiled into your binary if you actually import the subpackage, so a consumer who only uses the core SDK never pulls them in.

Manage endpoints, keys & deliveries (operators)

The same Client wraps the control-plane API — register receivers, mint keys, and drain the dead-letter queue from code (needs an admin-scoped key). Management methods return typed structs (*Endpoint, *ApiKey, *Delivery); list methods return a *Page[T] (Items + NextOffset); DeleteEndpoint / RevokeAPIKey return just an error (a 204).

ctx := context.Background()
wh := webhookd.New("https://webhooks.example.com", "whsk_admin_…")

// --- Endpoints ---------------------------------------------------------------
// Create a receiver — its signing secret is in the response exactly once, so persist it now.
ep, _ := wh.CreateEndpoint(ctx, "https://your-app.example/webhooks", &webhookd.CreateEndpointOptions{
	Subscriptions: []webhookd.Subscription{{MatchKind: "prefix", Pattern: "order."}},
	Description:   ptr("orders service"),
})
endpointID, signingSecret := ep.Id, *ep.Secret

wh.ListEndpoints(ctx, &webhookd.ListEndpointsOptions{Environment: "prod"}) // *Page[Endpoint]
wh.GetEndpoint(ctx, endpointID)

// PATCH — send only the keys you want to change (an omitted key is unchanged, an explicit nil clears):
wh.UpdateEndpoint(ctx, endpointID, webhookd.Patch{"max_attempts": 10, "status": "disabled"})

wh.RotateEndpointSecret(ctx, endpointID) // returns the new secret, once
wh.EnableEndpoint(ctx, endpointID)       // recover an auto-disabled endpoint
wh.DeleteEndpoint(ctx, endpointID)       // -> error only (204)

// --- API keys ----------------------------------------------------------------
key, _ := wh.CreateAPIKey(ctx, &webhookd.CreateAPIKeyOptions{Name: "ci-publisher", Scope: "publish"})
fmt.Println(*key.Key)          // shown once
wh.RevokeAPIKey(ctx, key.Id)   // -> error only (204)

// --- Deliveries / dead-letter recovery ---------------------------------------
dead, _ := wh.ListDeliveries(ctx, &webhookd.ListDeliveriesOptions{Status: "dead"})
for _, d := range dead.Items {
	wh.Redeliver(ctx, d.Id)
}

Optional outbound scalar fields (e.g. Description, MaxAttempts) are pointers, so only the ones you set are serialized. ptr above is a tiny helper — func ptr[T any](v T) *T { return &v }.

CLI

The module ships a webhookd command that wraps this client and speaks the same v1 API.

go install github.com/NimbusNexus/webhookd-go/cmd/webhookd@latest

Configuration (base URL + API key) is resolved in order: the --url / --api-key flags, then the WEBHOOKD_URL / WEBHOOKD_API_KEY environment variables, then ~/.webhookd/config.json (written by webhookd configure).

webhookd configure                                   # save base URL + API key to ~/.webhookd/config.json
webhookd publish order.created --data '{"id":123}' --idempotency-key order-123

webhookd endpoints create --url https://your-app.example/webhooks --subscribe prefix:order.
webhookd endpoints list --env prod
webhookd endpoints get <id>
webhookd endpoints update <id> --set max_attempts=10 --set status=disabled  # values are JSON-coerced
webhookd endpoints rotate-secret <id>
webhookd endpoints enable <id>
webhookd endpoints delete <id>

webhookd keys create --name ci-publisher --scope publish --expires-in-days 90
webhookd keys revoke <id>

webhookd deliveries list --status dead
webhookd deliveries redeliver <id>

# Verify a webhook — the raw body is read from stdin; prints "ok"/"failed" and exits 0/1:
webhookd verify --secret whsec_… --signature "$SIG" --timestamp "$TS" < body.json

webhookd version

Successful results print as indented JSON to stdout; errors go to stderr and the process exits non-zero (an API error renders as code: message).

Examples

examples/receiver is a runnable net/http server that verifies incoming webhooks; examples/publish publishes one event. Both read their configuration from environment variables:

WEBHOOKD_SIGNING_SECRET=whsec_… go run ./examples/receiver           # a verifying receiver on :8080
WEBHOOKD_URL=… WEBHOOKD_API_KEY=whsk_… go run ./examples/publish     # publish one event

Develop

gofmt -l .        # must print nothing
go vet ./...
go build ./...
go test ./...     # includes the shared cross-language signature vectors in testdata/

The outbox stores share one contract test. Memory/File/SQLite run fully with no server. The Redis and Postgres store contracts (store/redis, store/postgres) run the same contract but skip cleanly unless you point them at a server: WEBHOOKD_TEST_REDIS_URL (e.g. redis://localhost:6379) and WEBHOOKD_TEST_PG_URL (e.g. postgres://postgres:postgres@localhost:5432/webhookd_test).

Documentation

Overview

Package webhookd is the official Go SDK for NimbusNexus Webhooks (webhookd).

  • Verify — verify an incoming webhook's HMAC signature (for subscribers).
  • Client — publish events to webhookd and manage endpoints/keys/deliveries (for producers).

webhookd signs every delivery as HMAC_SHA256(secret, "<timestamp>." + rawBody) (its default timestamped mode) and sends X-Webhook-Signature: sha256=<hex> plus X-Webhook-Timestamp (unix seconds). A subscriber MUST verify the signature to prove the request genuinely came from webhookd and was not tampered with. This mirrors delivery_core.webhook_outbox.{sign,verify} byte-for-byte.

Index

Constants

View Source 
const (

	// DefaultToleranceSeconds is webhookd's default replay window for signature verification.
	DefaultToleranceSeconds = 300
)
View Source 
const Version = "0.3.0"

Version tracks the release tag; keep in lockstep with the Python/TypeScript SDKs.

Variables

This section is empty.

Functions

func Sign 

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

Sign returns the "sha256=<hex>" signature webhookd would send for body, signing over "<timestamp>." + body. Mainly useful for tests; subscribers call Verify.

func Verify 

func Verify(secret string, body []byte, signature string, opts *VerifyOptions) bool

Verify reports whether signature is a valid webhookd signature for body.

Pass the EXACT bytes you received as body — do not re-serialize the JSON, or the signature won't match. When opts.Timestamp is set, the signature is rejected if it is older/newer than the tolerance window (replay protection). X-Webhook-Signature carries one token normally, or several comma-separated tokens during a signing-secret rotation (dual-sign overlap); Verify accepts the request if ANY token matches, using a constant-time comparison. A malformed X-Webhook-Timestamp returns false rather than panicking.

Types

type APIError 

type APIError struct {
	StatusCode int
	Code       string
	Message    string
}

APIError is a non-2xx response from the webhookd API.

It carries the M5a error envelope: a stable machine Code (e.g. "rate_limited", "not_found", "validation_error") and a human Message, plus the HTTP StatusCode.

func (*APIError) Error 

func (e *APIError) Error() string

type ApiKey 

type ApiKey struct {
	Id        string  `json:"id"`
	Name      string  `json:"name"`
	Scope     string  `json:"scope"`
	Key       *string `json:"key,omitempty"`
	CreatedBy *string `json:"created_by,omitempty"`
	CreatedAt *string `json:"created_at,omitempty"`
	ExpiresAt *string `json:"expires_at,omitempty"`
}

ApiKey is an API key, as returned by POST /v1/api-keys (webhookd's ApiKeyOut). Key is present ONLY on the create response (returned exactly once).

type Client 

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

Client publishes events to webhookd and manages endpoints, API keys and deliveries.

Authenticate with a per-tenant API key (whsk_…) or a service token. Transient failures (connection errors, 429, 5xx) are retried with capped exponential backoff; a 429 honours its Retry-After header. Other non-2xx responses return an *APIError carrying the server's {error: {code, message}} envelope.

func New 

func New(baseURL, apiKey string, opts ...Option) *Client

New creates a Client for the webhookd instance at baseURL authenticated with apiKey. Trailing slashes are trimmed from baseURL.

func (*Client) CreateAPIKey 

func (c *Client) CreateAPIKey(ctx context.Context, opts *CreateAPIKeyOptions) (*ApiKey, error)

CreateAPIKey creates an API key (POST /v1/api-keys). The response includes the raw key exactly once — persist it.

func (*Client) CreateEndpoint 

func (c *Client) CreateEndpoint(ctx context.Context, endpointURL string, opts *CreateEndpointOptions) (*Endpoint, error)

CreateEndpoint creates an endpoint (POST /v1/endpoints). The response includes the signing secret exactly once — persist it.

func (*Client) DeleteEndpoint 

func (c *Client) DeleteEndpoint(ctx context.Context, id string) error

DeleteEndpoint deletes an endpoint (DELETE /v1/endpoints/{id}, 204).

func (*Client) Drain 

func (c *Client) Drain(ctx context.Context, opts *DrainOptions) (DrainResult, error)

Drain sends buffered records to webhookd. It pulls a due batch (oldest first) and POSTs each to /v1/events with header Idempotency-Key = record.ID (so a re-drain after a crash never double-publishes — webhookd dedupes). On 2xx the record is marked sent; on error attempts is bumped and the record is rescheduled with capped exponential backoff, or parked dead once attempts reaches maxAttempts (the WithOnDead hook fires). Requires a store (WithStore).

func (*Client) EnableEndpoint 

func (c *Client) EnableEndpoint(ctx context.Context, id string) (*Endpoint, error)

EnableEndpoint re-enables an endpoint webhookd auto-disabled after repeated failures (POST /v1/endpoints/{id}/enable).

func (*Client) Enqueue 

func (c *Client) Enqueue(ctx context.Context, eventType string, payload map[string]any, opts *EnqueueOptions) (string, error)

Enqueue durably buffers one event and returns its id WITHOUT any network call.

The id is opts.IdempotencyKey (if non-empty) or a fresh UUID v4; it becomes the Idempotency-Key on every later Drain send, so re-draining after a crash never double-publishes. Requires a store (WithStore).

func (*Client) GetEndpoint 

func (c *Client) GetEndpoint(ctx context.Context, id string) (*Endpoint, error)

GetEndpoint fetches a single endpoint by id (GET /v1/endpoints/{id}).

func (*Client) ListDeliveries 

func (c *Client) ListDeliveries(ctx context.Context, opts *ListDeliveriesOptions) (*Page[Delivery], error)

ListDeliveries lists deliveries (GET /v1/deliveries). Only the filters you provide are forwarded.

func (*Client) ListEndpoints 

func (c *Client) ListEndpoints(ctx context.Context, opts *ListEndpointsOptions) (*Page[Endpoint], error)

ListEndpoints lists endpoints for an environment (default "prod") (GET /v1/endpoints).

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, eventType string, payload map[string]any, opts *PublishOptions) (*Event, error)

Publish publishes one event (POST /v1/events). idempotency makes the publish safe to retry — a replay returns the original event without re-fanning-out.

func (*Client) Redeliver 

func (c *Client) Redeliver(ctx context.Context, id string) (*Delivery, error)

Redeliver re-enqueues a delivery for another attempt (POST /v1/deliveries/{id}/redeliver).

func (*Client) RevokeAPIKey 

func (c *Client) RevokeAPIKey(ctx context.Context, id string) error

RevokeAPIKey revokes an API key (DELETE /v1/api-keys/{id}, 204).

func (*Client) RotateEndpointSecret 

func (c *Client) RotateEndpointSecret(ctx context.Context, id string) (*Endpoint, error)

RotateEndpointSecret rotates an endpoint's signing secret (POST /v1/endpoints/{id}/rotate-secret). The response includes the new secret exactly once.

func (*Client) StartDrainer 

func (c *Client) StartDrainer(ctx context.Context, interval time.Duration, opts *DrainOptions)

StartDrainer launches a background goroutine that calls Drain every interval until StopDrainer (or ctx cancellation). A per-drain error is swallowed so a transient outage never kills the loop. Idempotent — a second call while one is running is a no-op. Requires a store (WithStore).

func (*Client) StopDrainer 

func (c *Client) StopDrainer()

StopDrainer signals the background drainer to stop and waits for it to exit. Safe to call when none is running.

func (*Client) UpdateEndpoint 

func (c *Client) UpdateEndpoint(ctx context.Context, id string, patch Patch) (*Endpoint, error)

UpdateEndpoint partially updates an endpoint (PATCH /v1/endpoints/{id}). patch is sent verbatim: an omitted key is left unchanged, an explicit nil clears the field.

type CreateAPIKeyOptions 

type CreateAPIKeyOptions struct {
	Name          string
	Scope         string
	ExpiresInDays *int
}

CreateAPIKeyOptions carries the optional arguments to Client.CreateAPIKey. Scope defaults to "admin" when empty; ExpiresInDays is sent only when non-nil.

type CreateEndpointOptions 

type CreateEndpointOptions struct {
	Environment       string
	Application       string
	Subscriptions     []Subscription
	Secret            *string
	MaxAttempts       *int
	RetrySchedule     []int
	Description       *string
	CustomHeaders     map[string]string
	DeliveryTimeoutMs *int
}

CreateEndpointOptions carries the optional arguments to Client.CreateEndpoint. Environment defaults to "prod" and Application to "default" when empty. Every other field is sent only when non-nil so the server applies its own default.

type Delivery 

type Delivery struct {
	Id         string  `json:"id"`
	EndpointID string  `json:"endpoint_id"`
	EventID    *string `json:"event_id,omitempty"`
	EventType  *string `json:"event_type,omitempty"`
	Status     string  `json:"status"`
	Attempts   *int    `json:"attempts,omitempty"`
	CreatedAt  *string `json:"created_at,omitempty"`
	UpdatedAt  *string `json:"updated_at,omitempty"`
}

Delivery is a delivery attempt record, as returned by the deliveries endpoints (webhookd's DeliveryOut).

type DrainOptions 

type DrainOptions struct {
	BatchLimit  int
	MaxAttempts int
}

DrainOptions overrides the client defaults for a single Drain. A zero field falls back to the client's configured value.

type DrainResult 

type DrainResult struct {
	// Sent is the number of records delivered (2xx) and marked sent this drain.
	Sent int
	// Failed is the number of records that failed this drain (rescheduled or newly parked dead).
	Failed int
	// Remaining is the number of records still buffered afterwards (Store.Size).
	Remaining int
}

DrainResult summarises one Drain call.

type Endpoint 

type Endpoint struct {
	Id                string            `json:"id"`
	URL               string            `json:"url"`
	Environment       string            `json:"environment"`
	Application       string            `json:"application"`
	Status            string            `json:"status"`
	Subscriptions     []Subscription    `json:"subscriptions"`
	Secret            *string           `json:"secret,omitempty"`
	MaxAttempts       *int              `json:"max_attempts,omitempty"`
	RetrySchedule     []int             `json:"retry_schedule,omitempty"`
	Description       *string           `json:"description,omitempty"`
	CustomHeaders     map[string]string `json:"custom_headers,omitempty"`
	DeliveryTimeoutMs *int              `json:"delivery_timeout_ms,omitempty"`
	CreatedAt         *string           `json:"created_at,omitempty"`
	UpdatedAt         *string           `json:"updated_at,omitempty"`
}

Endpoint is an endpoint, as returned by the management endpoints (webhookd's EndpointOut). Fields mirror the server's snake_case wire shape verbatim. Secret is present ONLY on the create + rotate-secret responses (returned exactly once).

type EnqueueOptions 

type EnqueueOptions struct {
	Environment    string
	Application    string
	Source         *string
	IdempotencyKey string
}

EnqueueOptions carries the optional arguments to Client.Enqueue. Environment defaults to "prod" and Application to "default" when empty. Source is stored only when non-nil. IdempotencyKey, when non-empty, becomes the record id (and the Idempotency-Key on every later Drain send); otherwise a fresh UUID v4 is generated.

type Error 

type Error struct {
	// Message is the human-readable description of the failure.
	Message string
	// Err is the underlying cause, if any (e.g. a transport error).
	Err error
}

Error is the base error type for all webhookd SDK errors (network failures, etc.).

It wraps an underlying cause when present so callers can use errors.Is/errors.As.

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Unwrap 

func (e *Error) Unwrap() error

Unwrap exposes the underlying cause for errors.Is/errors.As.

type Event 

type Event struct {
	Id                string  `json:"id"`
	EventUID          string  `json:"event_uid"`
	EventType         string  `json:"event_type"`
	Application       string  `json:"application"`
	Environment       string  `json:"environment"`
	DeliveriesCreated int     `json:"deliveries_created"`
	Source            *string `json:"source"`
}

Event is the published event, as returned by POST /v1/events (webhookd's EventOut).

type FileStore 

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

FileStore is a durable Store backed by a directory of per-record JSON files, written atomically (temp file + rename) so a crash mid-write never corrupts a record. It survives process restarts — a fresh FileStore over the same directory sees every un-sent record. Suitable for a single process; it does not coordinate concurrent drainers across processes.

func NewFileStore 

func NewFileStore(dir string) (*FileStore, error)

NewFileStore returns a durable file-backed store rooted at dir, creating it if necessary.

func (*FileStore) Close 

func (s *FileStore) Close() error

func (*FileStore) ListPending 

func (s *FileStore) ListPending(_ context.Context, limit int) ([]Record, error)

func (*FileStore) MarkFailed 

func (s *FileStore) MarkFailed(_ context.Context, id string, lastError *string, attempts int, nextAttemptAt time.Time) error

func (*FileStore) MarkSent 

func (s *FileStore) MarkSent(_ context.Context, id string) error

func (*FileStore) Save 

func (s *FileStore) Save(_ context.Context, record Record) error

func (*FileStore) Size 

func (s *FileStore) Size(_ context.Context) (int, error)

type ListDeliveriesOptions 

type ListDeliveriesOptions struct {
	Status     string
	EndpointID string
	EventType  string
	Since      string
	Until      string
	Q          string
	Offset     *int
	Limit      *int
}

ListDeliveriesOptions carries the optional filters for Client.ListDeliveries. Every filter is forwarded only when provided (non-empty string, or non-nil pointer).

type ListEndpointsOptions 

type ListEndpointsOptions struct {
	Environment string
	Offset      int
	Limit       *int
}

ListEndpointsOptions carries the optional arguments to Client.ListEndpoints. Environment defaults to "prod" when empty; Offset is always sent; Limit is sent only when non-nil.

type MemoryStore 

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

MemoryStore is an in-process, non-durable Store. The default for tests and single-process best-effort buffering.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore returns an empty in-memory store.

func (*MemoryStore) Close 

func (s *MemoryStore) Close() error

func (*MemoryStore) ListPending 

func (s *MemoryStore) ListPending(_ context.Context, limit int) ([]Record, error)

func (*MemoryStore) MarkFailed 

func (s *MemoryStore) MarkFailed(_ context.Context, id string, lastError *string, attempts int, nextAttemptAt time.Time) error

func (*MemoryStore) MarkSent 

func (s *MemoryStore) MarkSent(_ context.Context, id string) error

func (*MemoryStore) Save 

func (s *MemoryStore) Save(_ context.Context, record Record) error

func (*MemoryStore) Size 

func (s *MemoryStore) Size(_ context.Context) (int, error)

type Option 

type Option func(*Client)

Option configures a Client in New.

func WithDrainBatchLimit 

func WithDrainBatchLimit(n int) Option

WithDrainBatchLimit sets how many records a single Drain pulls from the store (default 100).

func WithHTTPClient 

func WithHTTPClient(h *http.Client) Option

WithHTTPClient injects a custom *http.Client (used for tests, proxies, custom transports).

func WithMaxAttempts 

func WithMaxAttempts(n int) Option

WithMaxAttempts sets the delivery attempts a record gets before Drain parks it dead (default 10).

func WithMaxRetries 

func WithMaxRetries(n int) Option

WithMaxRetries sets the number of extra attempts for retryable failures (default 2).

func WithOnDead 

func WithOnDead(fn func(Record)) Option

WithOnDead registers a callback invoked once when Drain parks a record dead (retry budget exhausted). The record passed carries the final attempts count and last error.

func WithStore 

func WithStore(store Store) Option

WithStore configures the durable outbox store. When set, Enqueue / Drain / StartDrainer become usable; omit it to use only the live Publish.

func WithTimeout 

func WithTimeout(d time.Duration) Option

WithTimeout sets the per-request timeout (default 10s). Ignored when WithHTTPClient supplies a client with its own Timeout.

type Page 

type Page[T any] struct {
	Items      []T  `json:"items"`
	NextOffset *int `json:"next_offset"`
}

Page is a single page of a list endpoint — items plus the cursor for the next page (nil at the end).

type Patch 

type Patch = map[string]any

Patch is a raw PATCH mapping for UpdateEndpoint, keyed with the server's snake_case names. PATCH semantics: an OMITTED key is left unchanged; an explicit nil CLEARS the field (JSON null). The map is sent verbatim.

type PublishOptions 

type PublishOptions struct {
	Environment    string
	Application    string
	Source         *string
	IdempotencyKey string
}

PublishOptions carries the optional arguments to Client.Publish. Environment defaults to "prod" and Application to "default" when empty. Source is sent only when non-nil. IdempotencyKey, when non-empty, is sent as the Idempotency-Key header.

type Record 

type Record struct {
	ID            string         `json:"id"`
	EventType     string         `json:"event_type"`
	Payload       map[string]any `json:"payload"`
	Environment   string         `json:"environment"`
	Application   string         `json:"application"`
	Source        *string        `json:"source"`
	CreatedAt     time.Time      `json:"created_at"`
	Attempts      int            `json:"attempts"`
	LastError     *string        `json:"last_error"`
	NextAttemptAt time.Time      `json:"next_attempt_at"`
}

Record is one buffered event. ID doubles as the webhookd Idempotency-Key (caller-supplied or a generated UUID v4), so re-saving the same ID (an idempotent enqueue) overwrites, and re-draining after a crash never double-publishes.

type Store 

type Store interface {
	// Save inserts-or-UPDATEs by record.ID — the same ID overwrites, so enqueue is idempotent.
	Save(ctx context.Context, record Record) error
	// ListPending returns records not yet sent (and not dead) whose NextAttemptAt <= now, oldest
	// first (by CreatedAt), capped at limit.
	ListPending(ctx context.Context, limit int) ([]Record, error)
	// MarkSent removes (or flags sent) a record after a 2xx.
	MarkSent(ctx context.Context, id string) error
	// MarkFailed persists the failure and schedules the next retry (or parks the record dead via a
	// far-future nextAttemptAt).
	MarkFailed(ctx context.Context, id string, lastError *string, attempts int, nextAttemptAt time.Time) error
	// Size reports the count of records still buffered (i.e. not yet sent); dead records included.
	Size(ctx context.Context) (int, error)
	// Close releases any resources (file handles, DB connections). The zero-work stores no-op.
	Close() error
}

Store is a durable buffer of pending Records. Every method is safe to call from the background drainer goroutine and the caller goroutine concurrently. Built-ins MemoryStore and FileStore ship in this package (stdlib only); SqliteStore, RedisStore and PostgresStore live in subpackages under go/store so a consumer who never imports them never compiles their drivers.

type Subscription 

type Subscription struct {
	MatchKind string `json:"match_kind"`
	Pattern   string `json:"pattern"`
}

Subscription is a subscription filter attached to an endpoint. MatchKind is one of exact|prefix|suffix|all.

type VerifyOptions 

type VerifyOptions struct {
	// Timestamp is the X-Webhook-Timestamp header value (unix seconds). When non-nil the replay
	// window is enforced against it — always pass it. When nil, the signature is computed over the
	// body alone.
	Timestamp *int64
	// ToleranceSeconds is the replay window; 0 means the default (300).
	ToleranceSeconds int
	// Now overrides the current unix time (for tests); 0 means time.Now().Unix().
	Now int64
}

VerifyOptions configures Verify.

Source Files

View all Source files

Directories

Path Synopsis
cmd
webhookd command
Command webhookd is the official command-line interface for NimbusNexus Webhooks (webhookd).
 Command webhookd is the official command-line interface for NimbusNexus Webhooks (webhookd).
examples
publish command
Command publish is a runnable example: publish one event with the SDK.
 Command publish is a runnable example: publish one event with the SDK.
receiver command
Command receiver is a runnable net/http server that verifies incoming webhookd webhooks.
 Command receiver is a runnable net/http server that verifies incoming webhookd webhooks.
store
postgres
Package postgres provides a durable, transactional webhookd.Store backed by PostgreSQL via github.com/jackc/pgx/v5 (pgxpool).
 Package postgres provides a durable, transactional webhookd.Store backed by PostgreSQL via github.com/jackc/pgx/v5 (pgxpool).
redis
Package redis provides a durable webhookd.Store backed by Redis via github.com/redis/go-redis/v9.
 Package redis provides a durable webhookd.Store backed by Redis via github.com/redis/go-redis/v9.
sqlite
Package sqlite provides a durable, transactional webhookd.Store backed by SQLite via the pure-Go modernc.org/sqlite driver (no cgo).
 Package sqlite provides a durable, transactional webhookd.Store backed by SQLite via the pure-Go modernc.org/sqlite driver (no cgo).
storetest
Package storetest provides a shared Store contract test, parametrized over any webhookd.Store implementation.
 Package storetest provides a shared Store contract test, parametrized over any webhookd.Store implementation.

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.