Skip to content
quarry Artifact [371d259da2]
Login

Artifact [371d259da2]

Check-ins Using Download Hex

Artifact 371d259da23ffdae1fdbd10ac6a2532a6165062f03cf1edfd1b107106cda9e46:

package webhook

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"craftdesign.group/quarry/internal/events"
)

// fakeSubs is an in-memory Subscriptions, so the dispatcher's tests do
// not need the OLTP database.
type fakeSubs struct {
	mu   sync.Mutex
	subs map[int64]Sub
}

func (f *fakeSubs) ActiveWebhooks(string) ([]Sub, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []Sub
	for _, s := range f.subs {
		if s.Active {
			out = append(out, s)
		}
	}
	return out, nil
}

func (f *fakeSubs) SubByID(id int64) (Sub, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	s, ok := f.subs[id]
	return s, ok
}

func (f *fakeSubs) set(s Sub) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.subs[s.ID] = s
}

func testDispatcher(t *testing.T, subs Subscriptions) *Dispatcher {
	t.Helper()
	ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	d, err := New(ev.JS(), subs)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	if err := d.Start(ctx); err != nil {
		t.Fatal(err)
	}
	return d
}

func TestSignAndVerify(t *testing.T) {
	body := []byte(`{"kind":"ticket.new"}`)
	sig := Sign("s3cret", body)
	if !Verify("s3cret", body, sig) {
		t.Fatal("signature does not verify with the right secret")
	}
	if Verify("wrong", body, sig) {
		t.Fatal("signature verified under the wrong secret")
	}
	// A signature must cover the exact bytes, or a receiver that
	// verifies-then-parses can be fed different content than was signed.
	if Verify("s3cret", []byte(`{"kind":"ticket.set"}`), sig) {
		t.Fatal("signature verified over different bytes")
	}
}

// The happy path: a raised event reaches the endpoint, signed, with the
// headers a receiver routes on.
func TestDeliversSignedEvent(t *testing.T) {
	got := make(chan *http.Request, 1)
	bodies := make(chan []byte, 1)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		b, _ := io.ReadAll(r.Body)
		bodies <- b
		got <- r
		w.WriteHeader(http.StatusOK)
	}))
	defer srv.Close()

	subs := &fakeSubs{subs: map[int64]Sub{
		1: {ID: 1, URL: srv.URL, Secret: "k", Active: true},
	}}
	d := testDispatcher(t, subs)

	want := Event{ID: "evt-1", Repo: "alpha", Kind: "ticket.new", Actor: "dan",
		Summary: "opened ticket", Link: "/r/alpha/ticket/abc", At: "2026-08-16T00:00:00Z"}
	if err := d.Raise(want); err != nil {
		t.Fatal(err)
	}

	select {
	case req := <-got:
		body := <-bodies
		if req.Header.Get("X-Quarry-Event") != "ticket.new" {
			t.Errorf("event header = %q", req.Header.Get("X-Quarry-Event"))
		}
		if req.Header.Get("X-Quarry-Repo") != "alpha" {
			t.Errorf("repo header = %q", req.Header.Get("X-Quarry-Repo"))
		}
		if req.Header.Get("X-Quarry-Delivery") != "evt-1" {
			t.Errorf("delivery header = %q", req.Header.Get("X-Quarry-Delivery"))
		}
		if !Verify("k", body, req.Header.Get("X-Quarry-Signature")) {
			t.Error("signature does not verify over the delivered body")
		}
		var e Event
		if err := json.Unmarshal(body, &e); err != nil {
			t.Fatal(err)
		}
		if e != want {
			t.Errorf("payload = %+v, want %+v", e, want)
		}
	case <-time.After(15 * time.Second):
		t.Fatal("no delivery")
	}
}

// A failing endpoint must be retried, not dropped — that is the whole
// reason delivery is durable rather than a fire-and-forget goroutine.
func TestRetriesUntilAccepted(t *testing.T) {
	var attempts atomic.Int32
	done := make(chan struct{})
	var once sync.Once
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		if attempts.Add(1) < 3 {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		w.WriteHeader(http.StatusOK)
		once.Do(func() { close(done) })
	}))
	defer srv.Close()

	subs := &fakeSubs{subs: map[int64]Sub{
		1: {ID: 1, URL: srv.URL, Secret: "k", Active: true},
	}}
	d := testDispatcher(t, subs)
	if err := d.Raise(Event{ID: "evt-retry", Repo: "alpha", Kind: "wiki.save"}); err != nil {
		t.Fatal(err)
	}
	select {
	case <-done:
		if n := attempts.Load(); n < 3 {
			t.Fatalf("accepted after %d attempts, expected retries", n)
		}
	case <-time.After(40 * time.Second):
		t.Fatalf("never accepted after %d attempts", attempts.Load())
	}
}

// Unsubscribing must actually stop delivery, including for events
// already queued — otherwise "delete" is advisory.
func TestPausedSubscriptionStopsDelivery(t *testing.T) {
	var hits atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		hits.Add(1)
		w.WriteHeader(http.StatusOK)
	}))
	defer srv.Close()

	subs := &fakeSubs{subs: map[int64]Sub{
		1: {ID: 1, URL: srv.URL, Secret: "k", Active: true},
	}}
	d := testDispatcher(t, subs)
	// Queue first, then pause before the worker can get to it.
	if err := d.Raise(Event{ID: "evt-paused", Repo: "alpha", Kind: "wiki.save"}); err != nil {
		t.Fatal(err)
	}
	subs.set(Sub{ID: 1, URL: srv.URL, Secret: "k", Active: false})

	time.Sleep(6 * time.Second)
	if n := hits.Load(); n != 0 {
		t.Fatalf("paused subscription received %d deliveries", n)
	}
}

// Two subscriptions on one repo each get their own delivery, and one
// failing endpoint must not starve the other.
func TestFansOutPerSubscription(t *testing.T) {
	var good, bad atomic.Int32
	okSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		good.Add(1)
		w.WriteHeader(http.StatusOK)
	}))
	defer okSrv.Close()
	badSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		bad.Add(1)
		w.WriteHeader(http.StatusBadGateway)
	}))
	defer badSrv.Close()

	subs := &fakeSubs{subs: map[int64]Sub{
		1: {ID: 1, URL: okSrv.URL, Secret: "k", Active: true},
		2: {ID: 2, URL: badSrv.URL, Secret: "k", Active: true},
	}}
	d := testDispatcher(t, subs)
	if err := d.Raise(Event{ID: "evt-fan", Repo: "alpha", Kind: "forum.post"}); err != nil {
		t.Fatal(err)
	}
	deadline := time.After(20 * time.Second)
	for good.Load() == 0 {
		select {
		case <-deadline:
			t.Fatal("healthy endpoint never received its delivery")
		case <-time.After(200 * time.Millisecond):
		}
	}
	if bad.Load() == 0 {
		t.Error("failing endpoint was never attempted")
	}
}