Receive and deliver signed events

When to use this guide

Use these tasks for a signed webhook endpoint or an application outbox. Install Echo Pro first. Run examples without a backend using:

go test -run Example ./webhook/...

The webhook overview explains acceptance, delivery uncertainty and replay windows. The reference lists body bounds, routing and retention settings.

Mount a dedicated receiver

Construct a verifier from trusted source and tenant configuration. Pass it and an inbox to ReceiveConfig. Attach the returned middleware to a dedicated POST route. It is terminal, so it does not forward accepted requests to your handler. This executable fixture exercises acceptance and repeated input:

func ExampleReceiveConfig_ToMiddleware() {
  secret := []byte("example-signing-key")
  sign := func(body string) string {
    m := hmac.New(sha256.New, secret)
    m.Write([]byte(body))
    return hex.EncodeToString(m.Sum(nil))
  }
  mw, err := webhook.ReceiveConfig{
    Inbox:  &inbox{seen: map[string][32]byte{}},
    Source: "billing",
    // Verify authenticates the original bytes, then maps a trusted tenant and
    // returns the minimized processing representation.
    Verify: func(_ context.Context, r *http.Request, body []byte) (
      webhook.Incoming,
      error,
    ) {
      if !hmac.Equal([]byte(
        r.Header.Get("X-Signature"),
      ), []byte(sign(string(body)))) {
        return webhook.Incoming{}, errors.New("invalid signature")
      }
      id, _, _ := strings.Cut(string(body), ":")
      return webhook.Incoming{
        Tenant:  "t1",
        Source:  "billing",
        Account: "acct",
        EventID: id,
        Type:    "invoice.paid",
        Format:  "text",
        Payload: body,
      }, nil
    },
  }.ToMiddleware()
  if err != nil {
    fmt.Println(err)
    return
  }
  e := echo.New()
  // The middleware is terminal: this handler is never called.
  e.POST("/webhooks/billing", func(c *echo.Context) error {
    return c.NoContent(http.StatusInternalServerError)
  }, mw)
  send := func(body, signature string) {
    req := httptest.NewRequestWithContext(
      context.Background(),
      http.MethodPost,
      "/webhooks/billing",
      strings.NewReader(body),
    )
    req.Header.Set("X-Signature", signature)
    rec := httptest.NewRecorder()
    e.ServeHTTP(rec, req)
    fmt.Println(rec.Code)
  }
  send("evt_1:paid", sign("evt_1:paid")) // committed
  send("evt_1:paid", sign("evt_1:paid")) // identical duplicate
  send("evt_1:void", sign("evt_1:void")) // same event ID, changed content
  send("evt_2:paid", "forged")           // never persisted
  // Output:
  // 204
  // 204
  // 409
  // 400
}

Set MaxBodyBytes for the provider’s documented payload size and Timeout for verification plus acceptance. Keep all body-transforming middleware after verification or outside this route. Retain inbox deduplication for the provider’s full replay window. A valid signature shows the sender holds the secret. It does not create an Echo principal.

Queue an event from your handler

The combined example adds a POST handler to its protected Echo group. id comes from tenant.From(c.Request().Context()); invoice is the application’s generated invoice ID. outbox is a configured memory store. The getting-started program constructs it and owns its worker.

api := e.Group("/v1/tenants/:tenant", guard, quotaMW)
api.GET("/invoices", func(c *echo.Context) error {
  id, _ := tenant.From(c.Request().Context())
  return c.JSON(http.StatusOK, map[string]any{
    "tenant":   id,
    "invoices": []string{"inv-1"},
  })
})

The POST handler prepares and enqueues this event with the request context:

prepared, err := webhook.PrepareOutgoing(webhook.Outgoing{
  Tenant: string(
    id,
  ), Destination: "invoices", Version: "v1", Key: invoice + "/created",
  Type: "invoice.created", Format: "json", Payload: []byte(fmt.Sprintf(
    `{"invoice":%q}`,
    invoice,
  )),
})
if err != nil {
  return err
}
receipt, err := outbox.Enqueue(c.Request().Context(), prepared)
if err != nil {
  return err
}
return c.JSON(http.StatusAccepted, map[string]string{
  "invoice":     invoice,
  "webhook_ref": receipt.Ref,
})

With HTTPSenderConfig, the outgoing Key is the Standard Webhooks message ID: 1–128 printable ASCII bytes, without spaces or dots (.). PrepareOutgoing accepts any 1–128-byte UTF-8 key without control characters, because custom SendFunc senders may allow more. Check the key with webhook.StandardID before you enqueue; otherwise the HTTP sender rejects the delivery with invalid_id and never sends it.

Exercise a memory queue and worker

The example prepares a fixed business key and inspects ready state. It starts one worker and verifies the signed request. It waits for the worker to stop after cancellation. It prints ready, verified credit.add invoice_42/grant and done.

package webhookmem_test

import (
  "context"
  "errors"
  "fmt"
  "io"
  "net/http"
  "net/http/httptest"
  "time"

  "echo.labstack.com/pro/webhook"
  "echo.labstack.com/pro/webhook/webhookmem"
)

func Example() {
  // BEGIN quickstart
  ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  defer cancel()
  // Public test fixture only. Load a crypto/rand-generated key from your
  // secret manager for real endpoints; never reuse keys between customers.
  secret := "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"
  verify, err := (webhook.StandardVerifierConfig{
    Secrets: []string{secret}, Source: "billing", Tenant: "t1",
    Type: "credit.add", Format: "json",
  }).ToVerifyFunc()
  if err != nil {
    panic(err)
  }
  verified := make(chan webhook.Incoming, 1)
  receiver := httptest.NewServer(http.HandlerFunc(func(
    w http.ResponseWriter,
    r *http.Request,
  ) {
    body, err := io.ReadAll(io.LimitReader(r.Body, webhook.MaxPayloadBytes+1))
    if err != nil {
      w.WriteHeader(http.StatusBadRequest)
      return
    }
    in, err := verify(r.Context(), r, body)
    if err != nil {
      w.WriteHeader(http.StatusBadRequest)
      return
    }
    select {
    case verified <- in:
    default:
    }
    w.WriteHeader(http.StatusNoContent)
  }))
  defer receiver.Close()
  send, err := (webhook.HTTPSenderConfig{Endpoints: map[string]webhook.Endpoint{
    "credits": {URL: receiver.URL, Secrets: []string{secret}},
  }}).ToSendFunc()
  if err != nil {
    panic(err)
  }
  store, err := webhookmem.New(webhookmem.StoreConfig{MaxMessages: 100,
    Routing: webhookmem.RoutingConfig{
      Destinations: map[string]webhookmem.DestinationConfig{
        "credits": {Version: "v1", Send: send, ReplayWindow: 24 * time.Hour},
      },
    },
  })
  if err != nil {
    panic(err)
  }
  p, err := webhook.PrepareOutgoing(webhook.Outgoing{
    Tenant: "t1", Destination: "credits", Version: "v1", Key: "invoice_42/grant",
    Type: "credit.add", Format: "json", Payload: []byte(`{"amount":100}`),
  })
  if err != nil {
    panic(err)
  }
  receipt, err := store.Enqueue(ctx, p)
  if err != nil {
    panic(err)
  }
  record, err := store.InspectRecord(ctx, receipt.Ref)
  if err != nil {
    panic(err)
  }
  fmt.Println(record.State) // ready: no worker has started
  worker, err := webhookmem.NewWorker(store, webhookmem.WorkerConfig{Concurrency: 1})
  if err != nil {
    panic(err)
  }
  done := make(chan error, 1)
  go func() { done <- worker.Run(ctx) }()
  select {
  case in := <-verified:
    fmt.Println("verified", in.Type, in.EventID)
  case <-ctx.Done():
    panic(ctx.Err())
  }
  // Receiver verification precedes the sender's recorded completion.
  poll := time.NewTicker(10 * time.Millisecond)
  defer poll.Stop()
  for {
    record, err = store.InspectRecord(ctx, receipt.Ref)
    if err != nil {
      panic(err)
    }
    if record.State == "done" {
      break
    }
    select {
    case <-poll.C:
    case <-ctx.Done():
      panic(ctx.Err())
    }
  }
  fmt.Println(record.State) // done: the sender reported acceptance
  cancel()
  // Join the caller-managed worker before releasing its dependencies.
  if err := <-done; !errors.Is(err, context.Canceled) {
    panic(err)
  }
  // END quickstart
  // Output:
  // ready
  // verified credit.add invoice_42/grant
  // done
}

The response from this fixture acknowledges signature verification only. It has no durable inbox or business ledger. For durable acceptance, use the receive middleware with a PostgreSQL inbox.

Couple an outgoing command to a business transaction

Use caller-owned deployment and runtime pools connected to the same database. Run Migrate during deployment, then apply the shipped runtime-role grants before New. Create the application’s permanent ledger separately. The recipe uses app.entitlements with primary key (tenant, invoice, action) and a stored amount.

The following compiled recipe inserts the ledger row and enqueues the frozen outgoing command in one READ COMMITTED transaction. It sets synchronous_commit=on. A repeated business key reads and compares the original amount. An existing identical grant does not enqueue again, even after queue deduplication expires.

// BEGIN production
// runHTTPOutbox uses caller-owned pools with deployment/runtime roles. Apply
// grants.sql after Migrate and before invoking this function with runtimePool.
// The application ledger is deployed separately, with a permanent UNIQUE key:
// CREATE TABLE app.entitlements (
//
//  tenant text NOT NULL, invoice text NOT NULL, action text NOT NULL,
//  amount bigint NOT NULL, PRIMARY KEY (tenant, invoice, action));
func runHTTPOutbox(
  ctx context.Context,
  ownerPool, runtimePool *pgxpool.Pool,
  endpoint webhook.Endpoint,
) error {
  send, err := (webhook.HTTPSenderConfig{Endpoints: map[string]webhook.Endpoint{
    "credits": endpoint,
  }}).ToSendFunc()
  if err != nil {
    return err
  }
  cfg := webhookpg.StoreConfig{Schema: "webhooks", Namespace: "billing",
    Routing: webhookpg.RoutingConfig{
      Destinations: map[string]webhookpg.DestinationConfig{
        "credits": {Version: "v1", Send: send, ReplayWindow: 7 * 24 * time.Hour},
      },
    },
  }
  if err := webhookpg.Migrate(ctx, ownerPool, cfg); err != nil {
    return err
  }
  store, err := webhookpg.New(ctx, runtimePool, cfg)
  if err != nil {
    return err
  }
  p, err := webhook.PrepareOutgoing(webhook.Outgoing{
    Tenant:      "t1",
    Destination: "credits",
    Version:     "v1",
    Key:         "invoice_42/grant",
    Type:        "credit.add",
    Format:      "json",
    Payload: []byte(
      `{"amount":100}`,
    ),
  })
  if err != nil {
    return err
  }
  if err := enqueueHTTPGrant(ctx, runtimePool, store, p); err != nil {
    return err
  }
  worker, err := webhookpg.NewWorker(store, webhookpg.WorkerConfig{Concurrency: 8})
  if err != nil {
    return err
  }
  // Run blocks; the application cancels ctx on shutdown and waits for Run.
  // Investigate non-cancellation stop errors before starting a new worker.
  err = worker.Run(ctx)
  if errors.Is(err, context.Canceled) {
    return nil
  }
  return err
}

func enqueueHTTPGrant(
  ctx context.Context,
  pool *pgxpool.Pool,
  store *webhookpg.Store,
  p webhook.PreparedOutgoing,
) (
  err error,
) {
  tx, err := pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
  if err != nil {
    return err
  }
  defer func() {
    cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
    defer cancel()
    if rollbackErr := tx.Rollback(cleanup); rollbackErr != nil && !errors.Is(
      rollbackErr,
      pgx.ErrTxClosed,
    ) {
      err = errors.Join(err, rollbackErr)
    }
  }()
  if _, err = tx.Exec(ctx, "SET LOCAL synchronous_commit = on"); err != nil {
    return err
  }
  tag, err := tx.Exec(
    ctx,
    `INSERT INTO app.entitlements(tenant, invoice, action, amount)
VALUES ($1, $2, 'grant', $3) ON CONFLICT DO NOTHING`,
    "t1",
    "invoice_42",
    int64(100),
  )
  if err != nil {
    return err
  }
  if tag.RowsAffected() == 0 {
    var amount int64
    if err := tx.QueryRow(
      ctx,
      `SELECT amount FROM app.entitlements
WHERE tenant=$1 AND invoice=$2 AND action='grant'`,
      "t1",
      "invoice_42",
    ).Scan(&amount); err != nil {
      return err
    }
    if amount != 100 {
      return errors.New("conflicting grant for the business key")
    }
    // Original ledger and command committed together. Never re-enqueue an
    // existing grant, even after outbox/receiver deduplication expires.
  } else {
    receipt, err := store.Enqueue(ctx, tx, p)
    if err != nil {
      return err
    }
    _ = receipt // Provisional until this transaction commits.
  }
  // On a transient failure or unknown COMMIT outcome, retry this ENTIRE
  // function with the same prepared p, never just Enqueue. The permanent
  // business-key ledger handles even a lost reply followed by outbox pruning.
  return tx.Commit(ctx)
}

// END production

This function requires real pools and an endpoint; its unit example does not run database I/O. Test it against your assigned PostgreSQL backend. Retry the entire transaction with the same prepared message after a transient error or unknown COMMIT. A lost commit reply cannot prove rollback. Keep the permanent ledger after webhook payload compaction and deduplication retention expire.

Process incoming events

For an incoming event that grants a business entitlement, register a transactional ProcessFunc. This compiled processor inserts an application ledger row and prepares an outgoing credit command only on its first business-key insertion:

func grantInvoice(ctx context.Context, tx pgx.Tx, in webhook.Incoming) (
  []webhook.PreparedOutgoing,
  error,
) {
  invoice, amount := string(in.Payload), int64(100) // parse the verified payload
  tag, err := tx.Exec(
    ctx,
    `INSERT INTO app.entitlements(tenant, invoice, action, amount)
VALUES ($1, $2, 'grant', $3) ON CONFLICT DO NOTHING`,
    in.Tenant,
    invoice,
    amount,
  )
  if err != nil {
    return nil, err
  }
  if tag.RowsAffected() == 0 {
    var recorded int64
    if err := tx.QueryRow(
      ctx,
      "SELECT amount FROM app.entitlements WHERE tenant = $1 AND "+
        "invoice = $2 AND action = 'grant'",
      in.Tenant,
      invoice,
    ).Scan(&recorded); err != nil {
      return nil, err
    }
    if recorded != amount {
      return nil, errors.New("conflicting grant")
    }
    return nil, nil // the ledger and original command committed together
  }
  // Emit only with the first ledger insertion: outbox and receiver keys expire.
  p, err := webhook.PrepareOutgoing(webhook.Outgoing{
    Tenant:      in.Tenant,
    Destination: "credits",
    Version:     "v1",
    Key: strconv.Quote(in.Tenant) + "/" + strconv.Quote(
      invoice,
    ) + "/grant",
    Type:    "credit.add",
    Format:  "json",
    Payload: []byte(`{"amount":100}`),
  })
  return []webhook.PreparedOutgoing{p}, err
}

The fixture treats the verified payload as an invoice identifier and fixes the amount at 100. Parse and validate your provider’s signed payload in the real processor. The unique (tenant, invoice, action) ledger key permanently prevents a duplicate grant.

Configure delivery and shutdown

Register each destination name in both store routing and sender endpoints. Set an immutable version and the receiver’s actual deduplication horizon as ReplayWindow. Start the worker only after New returns without error. Run stops claiming on cancellation, drains dispatched work within ShutdownTimeout and returns. Join it before closing either pool.

Choose Concurrency for available database and receiver capacity. Schedule Maintain separately with a bounded batch. Retain source processors and destination versions while ready or sending records reference them; removing live routing makes New refuse startup.

Handle inspection and uncertain sends

Inspect attempt and replay history before deciding another business effect is safe. HTTP 2xx records receiver acceptance. HTTP timeouts and network failures can leave an outgoing delivery uncertain. A different webhook ID would defeat receiver deduplication. Preserve the key and payload.

The admin console can expose authorized dead-record replay and explicit uncertain-delivery decisions through webhookadmin. It requires PostgreSQL. The full demo exercises those panels with disposable data. Keep operator scope, confirmations and durable audit enabled for recovery actions.