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
// gate, when set, holds every delivery-time lookup until closed —
// how a test wins an ordering race on purpose instead of by sleep.
gate chan struct{}
looked atomic.Int32
}
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()
gate := f.gate
f.mu.Unlock()
if gate != nil {
<-gate
}
f.looked.Add(1)
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},
}}
// The race under test is queue-then-pause: the pause must beat the
// delivery-time lookup. Won by construction — the gate holds every
// lookup until the pause is in — not by hoping a sleep is long
// enough, which is how this failed on a loaded CI host (run
// 31dea4ff: the worker looked before the pause landed).
subs.gate = make(chan struct{})
d := testDispatcher(t, subs)
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})
close(subs.gate)
deadline := time.Now().Add(10 * time.Second)
for subs.looked.Load() == 0 && time.Now().Before(deadline) {
time.Sleep(20 * time.Millisecond)
}
if subs.looked.Load() == 0 {
t.Fatal("the delivery worker never looked at the subscription")
}
time.Sleep(200 * time.Millisecond) // a post, had one been sent, would land well within this
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")
}
}