// Package webhook is Principle 6's standard outbound surface: quarry's
// activity, delivered to third-party endpoints over plain signed HTTP,
// so an integrator never has to speak NATS to react to a forge event.
//
// Delivery is durable by way of JetStream rather than an in-process
// retry loop: an endpoint that is down during a deploy must not cost
// its owner the events, and a quarry restart must not either. Each
// (event, subscription) pair is one stream message; the consumer acks
// on success, NAKs with backoff otherwise, and JetStream's MaxDeliver
// is the dead-letter boundary. 0003 §3 names dagnats as the durable
// mechanism for this class of work; dagnats runs on the same JetStream
// quarry already embeds, and using it directly keeps the dependency
// surface unchanged until v3 brings the engine in for CI.
//
// This package is a data plane; the arch gate restricts its importers.
package webhook
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"net/http"
"strings"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/evt"
)
const (
// StreamName carries pending deliveries.
StreamName = "QUARRYHOOK"
subjectAll = "quarry.hook.>"
durable = "hook-dispatcher"
// maxDeliver bounds attempts per delivery. Past it JetStream stops
// redelivering and the message is logged as dead-lettered — an
// endpoint that has failed this many times is broken, not flaky.
maxDeliver = 6
// deliverTimeout is per attempt. A receiver that cannot answer in
// this long is treated as failed and retried.
deliverTimeout = 10 * time.Second
)
// Event is the payload a subscriber receives. It is deliberately the
// same shape quarry raises internally for activity, so the webhook
// surface and the notify surface never drift into two vocabularies.
type Event struct {
ID string `json:"id"` // idempotency key; stable across redeliveries
Repo string `json:"repo"` //
Kind string `json:"kind"` // wiki.save, ticket.new, forum.post, …
Actor string `json:"actor"` // authenticated author, or "anonymous"
Summary string `json:"summary"` // one line, human-readable
Link string `json:"link"` // site-relative path to the thing
At string `json:"at"` // RFC3339, when quarry raised it
}
// delivery is one queued attempt: the event plus which subscription it
// is for. The secret is NOT carried on the stream — it is looked up at
// send time, so rotating a secret takes effect on pending deliveries
// and the stream never stores a key.
type delivery struct {
SubID int64 `json:"sub"`
Event Event `json:"event"`
}
// Subscriptions is the lookup the dispatcher needs; the OLTP database
// satisfies it.
type Subscriptions interface {
ActiveWebhooks(repo string) ([]Sub, error)
SubByID(id int64) (Sub, bool)
}
// Attempt is what happened on one try, for the delivery log. Recorded
// on failure AND on success: "it worked" is the answer to "did my hook
// fire", and a log that holds only failures cannot give it.
type Attempt struct {
Sub int64
Repo string
Event string
Kind string
Status int
Ms int
Err string
Attempt int
Body string
}
// Recorder receives every attempt. Optional: a forge with no store
// still delivers, it just cannot say afterwards what happened.
type Recorder interface{ RecordDelivery(Attempt) }
// Sub is one endpoint the dispatcher can deliver to.
type Sub struct {
ID int64
URL string
Secret string
Active bool
// Events is the space-separated vocabulary patterns this endpoint
// asked for. Empty means everything, which is what a subscription
// created before the field existed asked for by not being asked.
Events string
// Prev is a secret a rotation retired but has not yet expired. While
// it is set the delivery carries BOTH signatures, so a receiver can
// be redeployed without a window in which every event is rejected.
Prev string
}
// wants reports whether this subscription selected an event kind.
//
// An unparseable or unknown kind is DELIVERED to a subscription that
// asked for everything and withheld from one that named patterns. A
// selective endpoint asked for specific things and must not receive a
// surprise; a `*` endpoint asked for everything and a new event is
// still everything.
func (s Sub) wants(kind string) bool {
pats := strings.Fields(s.Events)
if len(pats) == 0 {
return true
}
return evt.Kind(kind).MatchesAny(pats)
}
// Dispatcher queues and delivers events.
type Dispatcher struct {
js nats.JetStreamContext
subs Subscriptions
http *http.Client
sub *nats.Subscription
// Now stamps a delivery that carries no date of its own. Injectable
// so a test can assert the field without racing a clock.
Now func() string
// rec logs what happened on each attempt. Nil is ordinary: a forge
// with no store still delivers, it just cannot say afterwards.
rec Recorder
}
// Record attaches the delivery log. Set once at startup, before Start.
func (d *Dispatcher) Record(rec Recorder) {
if d != nil {
d.rec = rec
}
}
// New prepares the delivery stream and returns a dispatcher. Start must
// be called to begin delivering.
func New(js nats.JetStreamContext, subs Subscriptions) (*Dispatcher, error) {
// Work-queue retention is right here and wrong for ZITEVT: a
// delivery has exactly one consumer and is finished once acked,
// where an artifact event must be folded independently by every
// node.
cfg := &nats.StreamConfig{
Name: StreamName,
Subjects: []string{subjectAll},
Retention: nats.WorkQueuePolicy,
MaxAge: 24 * time.Hour, // an undeliverable event stops being interesting
}
if _, err := js.AddStream(cfg); err != nil && err != nats.ErrStreamNameAlreadyInUse {
return nil, fmt.Errorf("webhook: stream: %w", err)
}
return &Dispatcher{js: js, subs: subs, http: &http.Client{Timeout: deliverTimeout}}, nil
}
// Raise queues one event for every active subscription on its repo.
// Queueing is a publish, so it is fast enough to call from an op; the
// actual HTTP happens in the worker. No subscriptions means no work.
func (d *Dispatcher) Raise(e Event) error {
if d == nil {
return nil
}
subs, err := d.subs.ActiveWebhooks(e.Repo)
if err != nil {
return err
}
for _, s := range subs {
if !s.wants(e.Kind) {
continue
}
body, err := json.Marshal(delivery{SubID: s.ID, Event: e})
if err != nil {
return err
}
// The idempotency key covers (event, subscription): a redelivered
// stream message must not become a second event for the same
// endpoint, and two endpoints must each get their own.
msgID := fmt.Sprintf("%s:%d", e.ID, s.ID)
if _, err := d.js.Publish("quarry.hook."+e.Repo, body,
nats.MsgId(msgID)); err != nil {
return err
}
}
return nil
}
// Start begins the delivery worker. It returns once subscribed; the
// worker runs until ctx is cancelled.
func (d *Dispatcher) Start(ctx context.Context) error {
if d == nil {
return nil
}
sub, err := d.js.PullSubscribe(subjectAll, durable,
nats.MaxDeliver(maxDeliver), nats.AckWait(deliverTimeout+5*time.Second))
if err != nil {
return fmt.Errorf("webhook: subscribe: %w", err)
}
d.sub = sub
go d.run(ctx)
return nil
}
func (d *Dispatcher) run(ctx context.Context) {
for {
if ctx.Err() != nil {
return
}
msgs, err := d.sub.Fetch(16, nats.MaxWait(2*time.Second))
if err != nil {
continue // timeout is the common case: nothing to deliver
}
for _, m := range msgs {
d.deliver(ctx, m)
}
}
}
func (d *Dispatcher) deliver(ctx context.Context, m *nats.Msg) {
var dl delivery
if err := json.Unmarshal(m.Data, &dl); err != nil {
// Undecodable: retrying cannot help, so drop it rather than
// occupying a redelivery slot forever.
log.Printf("webhook: undecodable delivery, dropping: %v", err)
m.Term()
return
}
sub, ok := d.subs.SubByID(dl.SubID)
if !ok || !sub.Active {
// Deleted or paused between queueing and delivery. Not an error:
// the subscriber asked to stop hearing from us.
m.Term()
return
}
if err := d.send(ctx, sub, dl.Event, deliveredCount(m)); err != nil {
meta, _ := m.Metadata()
if meta != nil && meta.NumDelivered >= maxDeliver {
log.Printf("webhook: giving up on subscription %d (%s) after %d attempts: %v",
sub.ID, sub.URL, meta.NumDelivered, err)
m.Term()
return
}
// Backoff grows with attempts so a flapping endpoint is not
// hammered; JetStream redelivers after the delay.
delay := time.Duration(1<<min(uint(deliveredCount(m)), 5)) * time.Second
m.NakWithDelay(delay)
return
}
m.Ack()
}
func deliveredCount(m *nats.Msg) int {
meta, err := m.Metadata()
if err != nil || meta == nil {
return 1
}
return int(meta.NumDelivered)
}
// post sends one event and reports whether the endpoint accepted it.
// Any 2xx is success; everything else is retried.
func (d *Dispatcher) post(ctx context.Context, s Sub, e Event) error {
return d.send(ctx, s, e, 1)
}
// send is post with the attempt number, so the log can say whether an
// endpoint accepted the first try or the fourth.
func (d *Dispatcher) send(ctx context.Context, s Sub, e Event, attempt int) error {
body, err := json.Marshal(e)
if err != nil {
return err
}
started := time.Now()
status, serr := d.deliverOnce(ctx, s, e, body)
if d.rec != nil {
msg := ""
if serr != nil {
msg = serr.Error()
}
d.rec.RecordDelivery(Attempt{Sub: s.ID, Repo: e.Repo, Event: e.ID, Kind: e.Kind,
Status: status, Ms: int(time.Since(started).Milliseconds()),
Err: msg, Attempt: attempt, Body: string(body)})
}
return serr
}
// Replay sends an exact body again, to the same endpoint, and records
// the outcome as its own attempt. The body is REPLAYED rather than
// rebuilt: an endpoint debugging a signature needs the bytes that
// produced the one it rejected.
//
// The ENDPOINT's answer comes back as an Outcome, not as an error. The
// forge's job here is to send and to record, and it has done both by
// the time the endpoint answers; a dial failure or a 503 from the far
// side is the result of the test, not a fault in running it. Returning
// it as an error made every unreachable endpoint a 500 from the op —
// after the delivery row had been written — so a client that trusted
// the status retried and double-delivered (b82291eb, G-102).
func (d *Dispatcher) Replay(ctx context.Context, s Sub, e Event, body []byte) Outcome {
if d == nil {
return Outcome{}
}
started := time.Now()
status, serr := d.deliverOnce(ctx, s, e, body)
out := Outcome{Status: status, Ms: int(time.Since(started).Milliseconds())}
if serr != nil {
out.Err = serr.Error()
}
if d.rec != nil {
d.rec.RecordDelivery(Attempt{Sub: s.ID, Repo: e.Repo, Event: e.ID, Kind: e.Kind,
Status: status, Ms: out.Ms, Err: out.Err, Attempt: 1, Body: string(body)})
}
return out
}
// Outcome is what one delivery attempt came back with: the endpoint's
// status (0 when it never answered), how long it took, and the error
// text when there was one. Sent says the endpoint accepted it.
type Outcome struct {
Status int
Ms int
Err string
}
// Sent is true when the endpoint answered 2xx.
func (o Outcome) Sent() bool { return o.Status >= 200 && o.Status < 300 }
// deliverOnce is the request itself, returning the status alongside the
// error so the log can distinguish "answered 500" from "never answered".
func (d *Dispatcher) deliverOnce(ctx context.Context, s Sub, e Event, body []byte) (int, error) {
req, err := http.NewRequestWithContext(ctx, "POST", s.URL, bytes.NewReader(body))
if err != nil {
return 0, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "quarry-webhook")
req.Header.Set("X-Quarry-Event", e.Kind)
req.Header.Set("X-Quarry-Repo", e.Repo)
req.Header.Set("X-Quarry-Delivery", e.ID)
req.Header.Set("X-Quarry-Signature", Sign(s.Secret, body))
// During a rotation the previous key signs a second header rather
// than replacing the first: a receiver checking either one keeps
// working, and one that checks both sees the new key arrive before
// it has to accept it.
if s.Prev != "" {
req.Header.Set("X-Quarry-Signature-Previous", Sign(s.Prev, body))
}
resp, err := d.http.Do(req)
if err != nil {
return 0, err
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode > 299 {
return resp.StatusCode, fmt.Errorf("endpoint answered %d", resp.StatusCode)
}
return resp.StatusCode, nil
}
// Sign returns the signature header value for a body: HMAC-SHA256 over
// the EXACT bytes sent, so a receiver verifies before parsing and never
// has to reproduce quarry's JSON formatting.
func Sign(secret string, body []byte) string {
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(body)
return "sha256=" + hex.EncodeToString(mac.Sum(nil))
}
// Verify is the check a receiver performs; exported so quarry's own
// tests and any Go integrator share one implementation.
func Verify(secret string, body []byte, signature string) bool {
return hmac.Equal([]byte(Sign(secret, body)), []byte(signature))
}