Artifact [c574963bc9]

Artifact c574963bc90596605ffc691aa5009c80354bb7c8cf505f7d9fc39aee1a2b74df:

• File internal/oltp/webhook.go — part of check-in [550b7af81c] at 2026-08-16 19:38:20 on branch trunk — webhooks: the standard outbound surface (0001 Principle 6)

  Activity events reach third parties over signed HTTP, so an integrator never has to speak NATS. Subscriptions live in forge OLTP; the signing secret is printed once at creation and never readable again.

  Delivery is durable rather than a fire-and-forget goroutine: each (event, subscription) is one JetStream message, acked on 2xx, NAKed with growing backoff, MaxDeliver as the dead-letter boundary. An endpoint down during a deploy still gets its events — proven live by writing with the receiver stopped and seeing the delivery land when it returned.

  0003 §3 names dagnats for this class of work; it runs on the JetStream quarry already embeds, so this uses that directly and leaves dagnats as the upgrade path when v3 brings the engine in for CI.

  Mentions are not forwarded: they address a person, not a repository. (user: dmestas, size: 4613) [annotate] [blame] [check-ins using]

// 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"
	"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);`

// 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
	Created 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 string) (Webhook, error) {
	if repo == "" {
		return Webhook{}, errors.New("oltp: webhook needs a repository")
	}
	// 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)
	}
	var buf [32]byte
	if _, err := rand.Read(buf[:]); err != nil {
		return Webhook{}, err
	}
	secret := hex.EncodeToString(buf[:])
	created := time.Now().UTC().Format(time.RFC3339)
	res, err := d.db.Exec(`INSERT INTO webhook (repo, url, secret, active, created)
		VALUES (?, ?, ?, 1, ?)`, repo, endpoint, secret, created)
	if err != nil {
		return Webhook{}, err
	}
	id, _ := res.LastInsertId()
	return Webhook{ID: id, Repo: repo, URL: endpoint, Secret: secret, Active: true, Created: created}, 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, 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.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, created 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.Created); 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, created FROM webhook WHERE id = ?`,
		id).Scan(&w.ID, &w.Repo, &w.URL, &w.Secret, &active, &w.Created)
	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
}

// 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))
}