Skip to content
quarry Artifact [f157a223c0]
Login

Artifact [f157a223c0]

Check-ins Using Download Hex

Artifact f157a223c007ddec8d20bcd35aa2550599aeac40a56198d03233d8e7236ddfa1:

// 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"
	"time"

	"github.com/nats-io/nats.go"
)

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

// Sub is one endpoint the dispatcher can deliver to.
type Sub struct {
	ID     int64
	URL    string
	Secret string
	Active bool
}

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

// 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 {
		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.post(ctx, sub, dl.Event); 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 {
	body, err := json.Marshal(e)
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, "POST", s.URL, bytes.NewReader(body))
	if err != nil {
		return 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))
	resp, err := d.http.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		return fmt.Errorf("endpoint answered %d", resp.StatusCode)
	}
	return 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))
}