package webhook
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"path/filepath"
"testing"
"time"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/evt"
)
// The whole chain: something raises a vocabulary event and a third-party
// endpoint receives it, without the raiser knowing an endpoint exists.
//
// The raiser here is internal/events directly, which is what BOTH the
// fold and the ops layer call — so this proves the property the change
// was for: a delivery no longer depends on how the change arrived.
func TestVocabularyEventReachesAnEndpoint(t *testing.T) {
bodies := make(chan []byte, 4)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
b, _ := io.ReadAll(r.Body)
bodies <- b
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
subs := &fakeSubs{subs: map[int64]Sub{
1: {ID: 1, URL: srv.URL, Secret: "k", Active: true},
}}
d, err := New(ev.JS(), subs)
if err != nil {
t.Fatal(err)
}
d.Now = func() string { return "2026-08-24T00:00:00Z" }
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
if err := d.Start(ctx); err != nil {
t.Fatal(err)
}
if err := d.Fanout(ctx); err != nil {
t.Fatal(err)
}
// The fan-out consumer starts at the tip, so a raise must not race
// its subscription.
time.Sleep(300 * time.Millisecond)
if err := ev.Raise(evt.TicketClosed, "alpha", map[string]any{
"ticket": "abcdef0123456789aa", "actor": "dan",
"hash": "ffee0011223344556677", "date": "2026-08-24T10:00:00Z",
}); err != nil {
t.Fatal(err)
}
select {
case body := <-bodies:
var e Event
if err := json.Unmarshal(body, &e); err != nil {
t.Fatal(err)
}
if e.Kind != "ticket.closed" {
t.Errorf("kind = %q, want ticket.closed — the subject carries it now", e.Kind)
}
if e.Repo != "alpha" || e.Actor != "dan" {
t.Errorf("repo/actor = %q/%q", e.Repo, e.Actor)
}
if e.Link != "/r/alpha/ticket/abcdef0123456789" {
t.Errorf("link = %q", e.Link)
}
if e.At != "2026-08-24T10:00:00Z" {
t.Errorf("at = %q, want the event's own date", e.At)
}
if e.ID != "ticket.closed:ffee0011223344556677" {
t.Errorf("id = %q; it must be stable across redeliveries", e.ID)
}
case <-time.After(20 * time.Second):
t.Fatal("a raised event never reached the endpoint")
}
}
func TestLinkAndSummaryPerNoun(t *testing.T) {
for _, tc := range []struct {
kind string
payload map[string]any
link string
summary string
}{
{"checkin.created", map[string]any{"hash": "aabbccddeeff00112233", "summary": "fix it"},
"/r/alpha/checkin/aabbccddeeff0011", "fix it"},
{"branch.moved", map[string]any{"hash": "aabbccddeeff00112233", "branch": "feature"},
"/r/alpha/checkin/aabbccddeeff0011", "branch feature moved"},
{"wiki.created", map[string]any{"page": "Home"},
"/r/alpha/wiki/Home", "wiki page Home created"},
{"tag.created", map[string]any{"tag": "release", "target": "aabbccddeeff00112233"},
"/r/alpha/checkin/aabbccddeeff0011", "tag release created"},
{"landing.accepted", map[string]any{"landing": "1122334455667788aa"},
"/r/alpha/landing/1122334455667788", "landing.accepted"},
// A noun with no page gets no link rather than a broken one.
{"clone.cloned", map[string]any{}, "", "clone.cloned"},
} {
e := eventOf(raised{Kind: tc.kind, Repo: "alpha", Payload: tc.payload}, "now", "")
if e.Link != tc.link {
t.Errorf("%s link = %q, want %q", tc.kind, e.Link, tc.link)
}
if e.Summary != tc.summary {
t.Errorf("%s summary = %q, want %q", tc.kind, e.Summary, tc.summary)
}
}
}
// A subscription that named patterns gets those and nothing else, and a
// subscription that named none gets everything — including events added
// to the vocabulary after it was written.
func TestSubscriptionSelectsItsEvents(t *testing.T) {
for _, tc := range []struct {
events string
kind string
want bool
}{
{"", "ticket.closed", true}, // asked for nothing = everything
{"*", "landing.accepted", true}, // asked for everything
{"ticket", "ticket.closed", true}, // a bare noun takes every action
{"ticket", "ticket.reopened", true}, // including ones added later
{"ticket", "landing.accepted", false}, // and nothing else
{"landing.accepted", "landing.accepted", true},
{"landing.accepted", "landing.opened", false}, // an action is exact
{"ticket landing.accepted", "landing.accepted", true},
{"ticket landing.accepted", "wiki.edited", false},
// A selective endpoint must not receive a surprise it cannot name.
{"ticket", "something.new", false},
{"*", "something.new", true},
} {
s := Sub{Events: tc.events}
if got := s.wants(tc.kind); got != tc.want {
t.Errorf("Sub{Events: %q}.wants(%q) = %v, want %v",
tc.events, tc.kind, got, tc.want)
}
}
}
// The filter has to hold at the fan-out, not only in the predicate: an
// endpoint that asked for tickets must see no delivery at all for a
// landing, rather than one it has to ignore.
func TestUnselectedEventIsNeverQueued(t *testing.T) {
hit := make(chan string, 4)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hit <- r.Header.Get("X-Quarry-Event")
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
subs := &fakeSubs{subs: map[int64]Sub{
1: {ID: 1, URL: srv.URL, Secret: "k", Active: true, Events: "ticket"},
}}
d := testDispatcher(t, subs)
if err := d.Raise(Event{ID: "a", Repo: "alpha", Kind: "landing.accepted"}); err != nil {
t.Fatal(err)
}
if err := d.Raise(Event{ID: "b", Repo: "alpha", Kind: "ticket.closed"}); err != nil {
t.Fatal(err)
}
select {
case got := <-hit:
if got != "ticket.closed" {
t.Fatalf("delivered %q; the landing was not selected", got)
}
case <-time.After(15 * time.Second):
t.Fatal("the selected event was not delivered")
}
select {
case got := <-hit:
t.Errorf("a second delivery arrived: %q", got)
case <-time.After(500 * time.Millisecond):
}
}
// Two CI runs finishing inside JetStream's duplicate window are two
// deliveries. CI events carry no hash, so their id used to be the bare
// kind, and the second ci.completed to an endpoint was dropped as a
// duplicate of the first.
func TestTwoCIRunsFinishingTogetherBothReachAnEndpoint(t *testing.T) {
bodies := make(chan []byte, 4)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
b, _ := io.ReadAll(r.Body)
bodies <- b
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
subs := &fakeSubs{subs: map[int64]Sub{
1: {ID: 1, URL: srv.URL, Secret: "k", Active: true},
}}
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)
}
if err := d.Fanout(ctx); err != nil {
t.Fatal(err)
}
time.Sleep(300 * time.Millisecond)
for _, run := range []string{"run-one", "run-two"} {
if err := ev.Raise(evt.CICompleted, "alpha", map[string]any{
"runId": run, "checkin": "aabbccddeeff00112233",
"pipeline": "ci/p.json", "status": "completed",
}); err != nil {
t.Fatal(err)
}
}
ids := map[string]bool{}
deadline := time.After(20 * time.Second)
for len(ids) < 2 {
select {
case body := <-bodies:
var e Event
if err := json.Unmarshal(body, &e); err != nil {
t.Fatal(err)
}
ids[e.ID] = true
case <-deadline:
t.Fatalf("FAIL: %d of 2 ci.completed deliveries arrived; ids %v", len(ids), ids)
}
}
t.Logf("ok: both runs delivered with distinct ids %v", ids)
}
// An event with no natural key still gets an id of its own, so it is
// never mistaken for another event of the same kind.
func TestAnEventWithNoNaturalKeyIsNeverADuplicateOfAnother(t *testing.T) {
a := eventOf(raised{Kind: "clone.cloned", Repo: "alpha", Payload: map[string]any{}}, "now", "")
b := eventOf(raised{Kind: "clone.cloned", Repo: "alpha", Payload: map[string]any{}}, "now", "")
if a.ID == b.ID || a.ID == "clone.cloned:" {
t.Fatalf("FAIL: keyless events share id %q / %q", a.ID, b.ID)
}
t.Logf("ok: keyless events carry distinct ids %q and %q", a.ID, b.ID)
// Redelivery of one vocabulary message is the same event: the
// stream position it was read from keys it.
c := eventOf(raised{Kind: "clone.cloned", Repo: "alpha", Payload: map[string]any{}}, "now", "seq-7")
e := eventOf(raised{Kind: "clone.cloned", Repo: "alpha", Payload: map[string]any{}}, "now", "seq-7")
if c.ID != e.ID {
t.Fatalf("FAIL: one message read twice got ids %q and %q", c.ID, e.ID)
}
t.Logf("ok: a redelivered message keeps its id %q", c.ID)
}