// webhook.go stores webhook subscriptions — Principle 6's standard
// outbound surface. A subscription is forge configuration, not DAG
// state, so it belongs in the OLTP database beside accounts (0001 §8).
package oltp
import (
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"net/url"
"strings"
"sync/atomic"
"time"
)
const webhookSchema = `
CREATE TABLE IF NOT EXISTS webhook (
id INTEGER PRIMARY KEY AUTOINCREMENT,
repo TEXT NOT NULL,
url TEXT NOT NULL,
secret TEXT NOT NULL, -- HMAC-SHA256 key; shown once, at creation
active INTEGER NOT NULL DEFAULT 1,
created TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS webhook_by_repo ON webhook (repo, active);`
// webhookMigrations are additive columns for stores written by an
// earlier schema.
//
// events: which of RFC 0005's vocabulary this endpoint wants, as
// space-separated patterns. `*` is every event, which is what every
// subscription written before this column got — the default preserves
// exactly the behaviour those owners configured, which was "send me
// everything" by not having been asked.
var webhookMigrations = []string{
`ALTER TABLE webhook ADD COLUMN events TEXT NOT NULL DEFAULT '*'`,
// Rotation with an overlap. Replacing a secret in one write means
// every delivery between the write and the receiver's redeploy is
// rejected — so the old key keeps working for an hour, and both
// signatures are sent until it expires.
`ALTER TABLE webhook ADD COLUMN old_secret TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE webhook ADD COLUMN old_until TEXT NOT NULL DEFAULT ''`,
}
// RotateWindow is how long a replaced secret keeps working. An hour is
// long enough to redeploy a receiver by hand and short enough that a
// leaked key is not accepted for a working day.
const RotateWindow = time.Hour
// DefaultEvents is what a subscription selects when its owner does not
// say. Everything, because a webhook exists to be told things.
const DefaultEvents = "*"
// Webhook is one subscription. Secret is only populated when the
// subscription is created — listing never returns it, because a
// signing key that can be read back is a signing key that leaks.
type Webhook struct {
ID int64
Repo string
URL string
Secret string
Active bool
// Events is the space-separated vocabulary patterns this endpoint
// selects (RFC 0005). `*` is everything; `ticket` is every action on
// tickets; `landing.accepted` is exactly that.
Events string
Created string
// OldSecret is the key a rotation retired, honoured until OldUntil
// so a receiver can be updated without dropping deliveries.
OldSecret string
OldUntil string
}
// AddWebhook registers an endpoint and mints its signing secret. The
// secret is returned once and never retrievable afterwards.
func (d *DB) AddWebhook(repo, endpoint, events string) (Webhook, error) {
if repo == "" {
return Webhook{}, errors.New("oltp: webhook needs a repository")
}
if strings.TrimSpace(events) == "" {
events = DefaultEvents
}
// Only http(s): a webhook URL is fetched by the server, so any other
// scheme is a request to make quarry speak a protocol on someone
// else's behalf.
u, err := url.Parse(endpoint)
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
return Webhook{}, fmt.Errorf("oltp: webhook url must be http or https, got %q", endpoint)
}
secret, err := newSecret()
if err != nil {
return Webhook{}, err
}
created := time.Now().UTC().Format(time.RFC3339)
res, err := d.db.Exec(`INSERT INTO webhook (repo, url, secret, active, events, created)
VALUES (?, ?, ?, 1, ?, ?)`, repo, endpoint, secret, events, created)
if err != nil {
return Webhook{}, err
}
id, _ := res.LastInsertId()
return Webhook{ID: id, Repo: repo, URL: endpoint, Secret: secret, Active: true,
Events: events, Created: created}, nil
}
// newSecret mints a signing key: 32 random bytes, hex. One generator,
// so an added secret and a rotated one have the same strength.
func newSecret() (string, error) {
var buf [32]byte
if _, err := rand.Read(buf[:]); err != nil {
return "", err
}
return hex.EncodeToString(buf[:]), nil
}
// Webhooks lists a repository's subscriptions. Secret is deliberately
// left empty; ActiveWebhooks is the dispatcher's accessor.
func (d *DB) Webhooks(repo string) ([]Webhook, error) {
rows, err := d.db.Query(`SELECT id, repo, url, active, events, created FROM webhook
WHERE repo = ? ORDER BY id`, repo)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Webhook
for rows.Next() {
var w Webhook
var active int
if err := rows.Scan(&w.ID, &w.Repo, &w.URL, &active, &w.Events, &w.Created); err != nil {
return nil, err
}
w.Active = active != 0
out = append(out, w)
}
return out, rows.Err()
}
// ActiveWebhooks returns the subscriptions to deliver to, secrets
// included — the dispatcher needs them to sign.
func (d *DB) ActiveWebhooks(repo string) ([]Webhook, error) {
rows, err := d.db.Query(`SELECT id, repo, url, secret, events, created, old_secret, old_until
FROM webhook WHERE repo = ? AND active = 1 ORDER BY id`, repo)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Webhook
for rows.Next() {
var w Webhook
if err := rows.Scan(&w.ID, &w.Repo, &w.URL, &w.Secret, &w.Events, &w.Created,
&w.OldSecret, &w.OldUntil); err != nil {
return nil, err
}
w.Active = true
out = append(out, w)
}
return out, rows.Err()
}
// WebhookByID returns one subscription with its secret, for redelivery.
func (d *DB) WebhookByID(id int64) (Webhook, bool) {
var w Webhook
var active int
err := d.db.QueryRow(`SELECT id, repo, url, secret, active, events, created, old_secret, old_until
FROM webhook WHERE id = ?`,
id).Scan(&w.ID, &w.Repo, &w.URL, &w.Secret, &active, &w.Events, &w.Created,
&w.OldSecret, &w.OldUntil)
if err != nil {
return Webhook{}, false
}
w.Active = active != 0
return w, true
}
// SetWebhookActive pauses or resumes a subscription. Pausing is kept
// distinct from deleting so an endpoint that is down can be silenced
// without losing its identity and secret.
func (d *DB) SetWebhookActive(id int64, active bool) error {
res, err := d.db.Exec(`UPDATE webhook SET active = ? WHERE id = ?`, boolInt(active), id)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return fmt.Errorf("oltp: no webhook %d", id)
}
return nil
}
// SetWebhookEvents changes which events an endpoint selects. Kept
// separate from pausing: narrowing a subscription is a routine edit,
// where pausing is a statement about the endpoint's health.
func (d *DB) SetWebhookEvents(id int64, events string) error {
if strings.TrimSpace(events) == "" {
events = DefaultEvents
}
res, err := d.db.Exec(`UPDATE webhook SET events = ? WHERE id = ?`, events, id)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return fmt.Errorf("oltp: no webhook %d", id)
}
return nil
}
// DeleteWebhook removes a subscription.
func (d *DB) DeleteWebhook(id int64) error {
_, err := d.db.Exec(`DELETE FROM webhook WHERE id = ?`, id)
return err
}
// redacted keeps a secret out of logs and errors if one is ever printed.
func (w Webhook) String() string {
return fmt.Sprintf("webhook %d %s -> %s (secret %s)", w.ID, w.Repo, w.URL,
strings.Repeat("*", 8))
}
// RotateWebhookSecret mints a new signing key and returns it — the one
// time it exists in the clear. The previous key stays valid for
// RotateWindow, and every delivery in that window carries BOTH
// signatures, so a receiver can be updated without dropping events.
func (d *DB) RotateWebhookSecret(id int64) (string, error) {
w, ok := d.WebhookByID(id)
if !ok {
return "", fmt.Errorf("no webhook %d", id)
}
secret, err := newSecret()
if err != nil {
return "", err
}
_, err = d.db.Exec(`UPDATE webhook SET secret = ?, old_secret = ?, old_until = ?
WHERE id = ?`, secret, w.Secret,
time.Now().Add(RotateWindow).UTC().Format(time.RFC3339), id)
if err != nil {
return "", err
}
return secret, nil
}
// PreviousSecret is the retired key while it is still honoured, empty
// once the window has passed. Read at send time rather than cleared by
// a sweep: a forge that was down for the whole window must not come
// back still signing with a key it promised to retire.
func (w Webhook) PreviousSecret() string {
if w.OldSecret == "" || w.OldUntil == "" {
return ""
}
until, err := time.Parse(time.RFC3339, w.OldUntil)
if err != nil || time.Now().After(until) {
return ""
}
return w.OldSecret
}
// Deliveries (RFC 0008 §Integrations): what actually happened when the
// forge tried to tell an endpoint something.
//
// "Did my hook fire?" is the question a webhook owner asks, and until
// this table the only answer was the global events page — which says
// what quarry RAISED, not what any endpoint RECEIVED. The two differ
// exactly when something is broken, which is the only time anyone asks.
const deliverySchema = `
CREATE TABLE IF NOT EXISTS webhook_delivery (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sub INTEGER NOT NULL,
repo TEXT NOT NULL,
event TEXT NOT NULL, -- the event's id, for redelivery
kind TEXT NOT NULL,
at TEXT NOT NULL,
status INTEGER NOT NULL, -- HTTP code, or 0 when the request never completed
ms INTEGER NOT NULL,
err TEXT NOT NULL DEFAULT '',
attempt INTEGER NOT NULL DEFAULT 1,
-- The exact bytes sent. Kept so a redelivery is a REPLAY rather than
-- a fresh event with the same name: an endpoint debugging a
-- signature needs the body that produced it.
body TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS webhook_delivery_by_sub ON webhook_delivery (sub, id DESC);`
// deliveryKeep is how long an attempt is worth keeping. Long enough to
// debug an endpoint that broke over a weekend, short enough that a busy
// forge does not accumulate a log nobody reads.
const deliveryKeep = 30 * 24 * time.Hour
// deliveryBodyMax bounds one stored body. An event is a summary, not a
// payload, so this is generous; the cap exists because an unbounded
// TEXT column in a table written on every delivery is how a store
// grows without anybody choosing to.
const deliveryBodyMax = 16 << 10
// Delivery is one attempt, as the settings page lists it.
type Delivery struct {
ID int64 `json:"id"`
Sub int64 `json:"sub"`
Event string `json:"event"`
Kind string `json:"kind"`
At string `json:"at"`
Status int `json:"status"`
Ms int `json:"ms"`
Err string `json:"err,omitempty"`
Attempt int `json:"attempt"`
Body string `json:"body,omitempty"`
// Repo attributes the attempt, so a fleet can ask "which store's
// hooks are failing" without joining back through the subscription.
Repo string `json:"repo,omitempty"`
}
// OK reports whether the endpoint accepted this attempt.
func (d Delivery) OK() bool { return d.Status >= 200 && d.Status <= 299 }
var deliveriesSincePrune atomic.Int64
// RecordDelivery appends one attempt. Best-effort: a forge that cannot
// write its delivery log must still deliver.
func (d *DB) RecordDelivery(rec Delivery) {
if rec.Sub == 0 {
return
}
body := rec.Body
if len(body) > deliveryBodyMax {
body = body[:deliveryBodyMax]
}
if rec.At == "" {
rec.At = time.Now().UTC().Format(time.RFC3339Nano)
}
_, _ = d.db.Exec(`INSERT INTO webhook_delivery
(sub, repo, event, kind, at, status, ms, err, attempt, body)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
rec.Sub, rec.Repo, rec.Event, rec.Kind, rec.At, rec.Status, rec.Ms,
rec.Err, rec.Attempt, body)
// Pruned occasionally rather than per write: the scan belongs
// nowhere near the path a delivery storm runs down.
if deliveriesSincePrune.Add(1) < 200 {
return
}
deliveriesSincePrune.Store(0)
_, _ = d.db.Exec(`DELETE FROM webhook_delivery WHERE at < ?`,
time.Now().Add(-deliveryKeep).UTC().Format(time.RFC3339Nano))
}
// Deliveries lists one subscription's recent attempts, newest first.
func (d *DB) Deliveries(sub int64, limit int) ([]Delivery, error) {
if limit <= 0 || limit > 100 {
limit = 10
}
rows, err := d.db.Query(`SELECT id, sub, repo, event, kind, at, status, ms, err, attempt, body
FROM webhook_delivery WHERE sub = ? ORDER BY id DESC LIMIT ?`, sub, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Delivery
for rows.Next() {
var v Delivery
if err := rows.Scan(&v.ID, &v.Sub, &v.Repo, &v.Event, &v.Kind, &v.At,
&v.Status, &v.Ms, &v.Err, &v.Attempt, &v.Body); err != nil {
return nil, err
}
out = append(out, v)
}
return out, rows.Err()
}
// DeliveryByID is one attempt, for replaying it.
//
// It reads repo. The redeliver path refuses a delivery whose repo is
// not the one in the URL — the id names a row the whole forge shares —
// and this query used to leave Repo empty, so that check refused EVERY
// delivery: redeliver answered "no delivery N" for ids the deliveries
// list had just returned (b82291eb, measured by jva on 291..295).
func (d *DB) DeliveryByID(id int64) (Delivery, bool) {
var v Delivery
err := d.db.QueryRow(`SELECT id, sub, repo, event, kind, at, status, ms, err, attempt, body
FROM webhook_delivery WHERE id = ?`, id).Scan(&v.ID, &v.Sub, &v.Repo, &v.Event,
&v.Kind, &v.At, &v.Status, &v.Ms, &v.Err, &v.Attempt, &v.Body)
return v, err == nil
}