package spindle
import (
"context"
"database/sql"
"encoding/json"
"errors"
"io"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/models"
)
const (
pullRkey = "3kpull0000001"
patchOne = "bafypatchone"
patchTwo = "bafypatchtwo"
patchSha1 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
patchSha2 = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
settleTime = 300 * time.Millisecond
)
type delivery struct {
event string
signature string
body []byte
}
// stands in for the user's endpoint
type sink struct {
srv *httptest.Server
mu sync.Mutex
got []delivery
}
func newSink(t *testing.T) *sink {
t.Helper()
s := &sink{}
s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
s.mu.Lock()
s.got = append(s.got, delivery{event: r.Header.Get("X-Tangled-Event"), signature: r.Header.Get("X-Tangled-Signature-256"), body: body})
s.mu.Unlock()
}))
t.Cleanup(s.srv.Close)
return s
}
func (s *sink) all() []delivery {
s.mu.Lock()
defer s.mu.Unlock()
return append([]delivery(nil), s.got...)
}
// deliveries are async, so wait for the count and then see that nothing else follows
func (s *sink) expect(t *testing.T, want int) []delivery {
t.Helper()
if want > 0 {
eventually(t, 10*time.Second, "the webhook deliveries", func() bool { return len(s.all()) >= want })
}
time.Sleep(settleTime)
got := s.all()
if len(got) != want {
t.Fatalf("got %d webhook deliveries, want %d: %s", len(got), want, describeDeliveries(got))
}
return got
}
func describeDeliveries(got []delivery) string {
out := ""
for _, d := range got {
out += "\n" + d.event + " " + string(d.body)
}
return out
}
func (w *world) subscribe(n *node, s *sink, events ...string) {
t := w.t
t.Helper()
hook := &models.Webhook{RepoDid: testRepoDid, Url: s.srv.URL, Secret: "shh", Active: true, Events: events}
if created, err := n.s.db.AddWebhook(context.Background(), hook, 0, 0); err != nil || !created {
t.Fatalf("add webhook: created=%v err=%v", created, err)
}
}
var pullEvents = []string{
string(models.WebhookEventPullRequestCreated),
string(models.WebhookEventPullRequestResubmitted),
string(models.WebhookEventPullRequestClosed),
string(models.WebhookEventPullRequestMerged),
string(models.WebhookEventPullRequestReopened),
}
func (w *world) putTicket(repo syntax.DID, rkey, state string, editor syntax.DID, branch bool) map[string]any {
record := map[string]any{
"$type": org_tangled.TrackTicketNSID,
"state": state,
"title": map[string]any{"$type": "org.tangled.markup.markdown#inline", "text": "fix the thing"},
"body": map[string]any{"$type": "org.tangled.markup.markdown", "text": "it was broken"},
"createdAt": "2026-02-01T00:00:00Z",
}
if branch {
record["targetBranch"] = "main"
}
w.knot.putVersion(repo, org_tangled.TrackTicketNSID, rkey, editor, record)
return record
}
func (w *world) putPatch(repo syntax.DID, rkey, ticket, head string, editor syntax.DID) map[string]any {
record := map[string]any{
"$type": org_tangled.ReviewPatchNSID,
"base": map[string]any{"oid": patchSha1},
"head": map[string]any{"oid": head},
"ticket": ticket,
"createdAt": "2026-02-01T00:00:00Z",
}
w.knot.putVersion(repo, org_tangled.ReviewPatchNSID, rkey, editor, record)
return record
}
// the knot already holds the record, the event only says it changed
func (w *world) emitTicket(repo syntax.DID, rkey, op string) int64 {
return w.js.commit(repo, org_tangled.TrackTicketNSID, rkey, op, map[string]any{"$type": org_tangled.TrackTicketNSID})
}
func (w *world) emitPatch(repo syntax.DID, rkey, op string) int64 {
return w.js.commit(repo, org_tangled.ReviewPatchNSID, rkey, op, map[string]any{"$type": org_tangled.ReviewPatchNSID})
}
type pullEnv struct {
*world
n *node
sink *sink
}
func newPullEnv(t *testing.T) *pullEnv {
t.Helper()
w := newWorld(t)
n := w.boot()
w.seed(n, testBob)
e := &pullEnv{world: w, n: n, sink: newSink(t)}
w.subscribe(n, e.sink, pullEvents...)
return e
}
func (e *pullEnv) ticket(state string, editor syntax.DID, op string) {
e.putTicket(testRepoDid, pullRkey, state, editor, true)
e.n.applied(testRepoDid, e.emitTicket(testRepoDid, pullRkey, op))
}
func (e *pullEnv) patch(rkey, head string, editor syntax.DID) {
e.putPatch(testRepoDid, rkey, pullRkey, head, editor)
e.n.applied(testRepoDid, e.emitPatch(testRepoDid, rkey, opCreate))
}
func pullPayload(t *testing.T, d delivery) models.WebhookPullRequestPayload {
t.Helper()
var p models.WebhookPullRequestPayload
if err := json.Unmarshal(d.body, &p); err != nil {
t.Fatalf("payload %s: %v", d.body, err)
}
return p
}
func wantPull(t *testing.T, d delivery, event models.WebhookEvent, action, state string, round int, author, sender syntax.DID, head string) {
t.Helper()
p := pullPayload(t, d)
if d.event != string(event) || p.Action != action || p.PullRequest.State != state {
t.Errorf("delivered %q action %q state %q, want %q %q %q", d.event, p.Action, p.PullRequest.State, event, action, state)
}
if p.PullRequest.RoundNumber != round {
t.Errorf("%s: round %d, want %d", action, p.PullRequest.RoundNumber, round)
}
if p.PullRequest.Owner.Did != author.String() {
t.Errorf("%s: pull author %s, want %s", action, p.PullRequest.Owner.Did, author)
}
if p.Sender.Did != sender.String() {
t.Errorf("%s: sender %s, want %s", action, p.Sender.Did, sender)
}
if p.PullRequest.Title != "fix the thing" || p.PullRequest.Body != "it was broken" || p.PullRequest.TargetBranch != "main" {
t.Errorf("%s: pull request %+v lost its record fields", action, p.PullRequest)
}
if p.PullRequest.Source == nil || p.PullRequest.Source.Sha != head || p.PullRequest.Source.Branch != "" {
t.Errorf("%s: source %+v, want only the sha %s", action, p.PullRequest.Source, head)
}
if p.Repository.FullName != testOwner.String()+"/repo" {
t.Errorf("%s: repository %q", action, p.Repository.FullName)
}
if d.signature == "" {
t.Errorf("%s: delivery isn't signed", action)
}
}
func TestPullRequestWebhooksFireOncePerTransition(t *testing.T) {
e := newPullEnv(t)
// the ticket alone is no pull request event, the first patch is
e.ticket("open", testBob, opCreate)
e.sink.expect(t, 0)
e.patch(patchOne, patchSha1, testBob)
got := e.sink.expect(t, 1)
wantPull(t, got[0], models.WebhookEventPullRequestCreated, "created", "open", 1, testBob, testBob, patchSha1)
e.patch(patchTwo, patchSha2, testBob)
got = e.sink.expect(t, 2)
wantPull(t, got[1], models.WebhookEventPullRequestResubmitted, "resubmitted", "open", 2, testBob, testBob, patchSha2)
// the maintainer closes it, so sender and author differ
e.ticket("closed", testCarol, "update")
got = e.sink.expect(t, 3)
wantPull(t, got[2], models.WebhookEventPullRequestClosed, "closed", "closed", 2, testBob, testCarol, patchSha2)
e.ticket("open", testBob, "update")
got = e.sink.expect(t, 4)
wantPull(t, got[3], models.WebhookEventPullRequestReopened, "reopened", "open", 2, testBob, testBob, patchSha2)
e.ticket("completed", testCarol, "update")
got = e.sink.expect(t, 5)
wantPull(t, got[4], models.WebhookEventPullRequestMerged, "merged", "merged", 2, testBob, testCarol, patchSha2)
}
func TestPullRequestReplaysFireNothing(t *testing.T) {
e := newPullEnv(t)
e.ticket("open", testBob, opCreate)
e.patch(patchOne, patchSha1, testBob)
e.patch(patchTwo, patchSha2, testBob)
e.ticket("completed", testCarol, "update")
e.sink.expect(t, 3)
// a stale event reads the knot's current state, which spindle already has
e.emitTicket(testRepoDid, pullRkey, opCreate)
e.emitTicket(testRepoDid, pullRkey, "update")
e.emitPatch(testRepoDid, patchOne, opCreate)
e.n.consumed(e.emitPatch(testRepoDid, patchTwo, opCreate))
e.sink.expect(t, 3)
}
func TestPullRequestPatchBeforeItsTicketEvent(t *testing.T) {
e := newPullEnv(t)
e.putTicket(testRepoDid, pullRkey, "open", testBob, true)
e.putPatch(testRepoDid, patchOne, pullRkey, patchSha1, testBob)
e.emitPatch(testRepoDid, patchOne, opCreate)
e.n.consumed(e.emitTicket(testRepoDid, pullRkey, opCreate))
got := e.sink.expect(t, 1)
wantPull(t, got[0], models.WebhookEventPullRequestCreated, "created", "open", 1, testBob, testBob, patchSha1)
}
func TestPullRequestTransitionSurvivesARestart(t *testing.T) {
e := newPullEnv(t)
e.ticket("open", testBob, opCreate)
e.patch(patchOne, patchSha1, testBob)
e.sink.expect(t, 1)
e.pushCursorPast(e.n)
e.putTicket(testRepoDid, pullRkey, "closed", testCarol, true)
e.emitTicket(testRepoDid, pullRkey, "update")
n2 := e.boot()
eventually(t, 15*time.Second, "the close to be seen after the restart", func() bool {
ticket, err := n2.s.db.GetPullTicket(testRepoDid, pullRkey)
return err == nil && ticket.State == "closed"
})
got := e.sink.expect(t, 2)
wantPull(t, got[1], models.WebhookEventPullRequestClosed, "closed", "closed", 1, testBob, testCarol, patchSha1)
}
func TestPullRequestIgnoresIssuesAndForeignRepos(t *testing.T) {
e := newPullEnv(t)
e.putTicket(testRepoDid, "3kissue0000001", "open", testBob, false)
e.putPatch(testRepoDid, "bafyissuepatch", "3kissue0000001", patchSha1, testBob)
e.emitTicket(testRepoDid, "3kissue0000001", opCreate)
e.n.consumed(e.emitPatch(testRepoDid, "bafyissuepatch", opCreate))
e.putTicket(faultOtherRepoDid, pullRkey, "open", testBob, true)
e.putPatch(faultOtherRepoDid, patchOne, pullRkey, patchSha1, testBob)
e.emitTicket(faultOtherRepoDid, pullRkey, opCreate)
e.n.consumed(e.emitPatch(faultOtherRepoDid, patchOne, opCreate))
e.sink.expect(t, 0)
if _, err := e.n.s.db.GetPullTicket(testRepoDid, "3kissue0000001"); !errors.Is(err, sql.ErrNoRows) {
t.Errorf("an issue got a pull row: %v", err)
}
if _, err := e.n.s.db.GetPullTicket(faultOtherRepoDid, pullRkey); !errors.Is(err, sql.ErrNoRows) {
t.Errorf("a repo this spindle doesn't serve got a pull row: %v", err)
}
}
func TestPullRequestEventsForRecordsTheKnotLacksFireNothing(t *testing.T) {
e := newPullEnv(t)
// nothing was ever put on the knot, so these are forged or already gone
e.emitTicket(testRepoDid, pullRkey, opCreate)
e.emitTicket(testRepoDid, pullRkey, "update")
e.n.consumed(e.emitPatch(testRepoDid, patchOne, opCreate))
e.sink.expect(t, 0)
// a patch whose ticket the knot lacks has no pull request to belong to
e.putPatch(testRepoDid, patchTwo, pullRkey, patchSha1, testBob)
e.n.consumed(e.emitPatch(testRepoDid, patchTwo, opCreate))
e.sink.expect(t, 0)
}
func TestPullRequestDeletesDropTheRowsWithoutAWebhook(t *testing.T) {
e := newPullEnv(t)
e.ticket("open", testBob, opCreate)
e.patch(patchOne, patchSha1, testBob)
e.sink.expect(t, 1)
e.knot.erase(testRepoDid, org_tangled.ReviewPatchNSID, patchOne)
e.emitPatch(testRepoDid, patchOne, opDelete)
e.knot.erase(testRepoDid, org_tangled.TrackTicketNSID, pullRkey)
e.n.consumed(e.emitTicket(testRepoDid, pullRkey, opDelete))
e.sink.expect(t, 1)
if _, err := e.n.s.db.GetPullTicket(testRepoDid, pullRkey); !errors.Is(err, sql.ErrNoRows) {
t.Errorf("deleted ticket still has a row: %v", err)
}
if rounds, _, err := e.n.s.db.PullRounds(testRepoDid, pullRkey); err != nil || rounds != 0 {
t.Errorf("deleted ticket still has %d rounds, err %v", rounds, err)
}
}
func TestPushWebhookIsDelivered(t *testing.T) {
w := newWorld(t)
n := w.boot()
w.seed(n)
hook := newSink(t)
w.subscribe(n, hook, string(models.WebhookEventPush))
// skip-ci keeps git out of it, the webhook fires either way
if _, err := n.s.NotifyEvent(context.Background(), w.knotRef(), testRepoDid, pushEvent("refs/heads/main", "skip-ci")); err != nil {
t.Fatalf("NotifyEvent: %v", err)
}
got := hook.expect(t, 1)
if got[0].event != "push" || got[0].signature == "" {
t.Errorf("delivered %q signed=%v, want a signed push", got[0].event, got[0].signature != "")
}
var p models.WebhookPayload
if err := json.Unmarshal(got[0].body, &p); err != nil {
t.Fatal(err)
}
if p.Ref != "refs/heads/main" || p.Before != testHexOld || p.After != testHexNew || p.Pusher.Did != testBob.String() {
t.Errorf("push payload %+v", p)
}
if p.Repository.FullName != testOwner.String()+"/repo" || p.Repository.Owner.Did != testOwner.String() {
t.Errorf("push repository %+v", p.Repository)
}
}
// web puts the manifest with the new name, then creates the new declaration and deletes the old one
func TestRenameWebhookFiresOnceWhateverTheEventOrder(t *testing.T) {
steps := map[string]func(w *world) int64{
"manifest": func(w *world) int64 {
w.setManifest(testRepoDid, "Renamed", w.spindleDid())
return w.emitManifest(testRepoDid)
},
"declaration created": func(w *world) int64 {
w.declare("Renamed", testRepoDid)
return w.emitDeclaration("Renamed", testRepoDid)
},
"declaration deleted": func(w *world) int64 {
w.undeclare("repo")
return w.js.commit(testOwner, org_tangled.RepoDeclarationNSID, "repo", opDelete, nil)
},
}
orders := [][]string{
{"manifest", "declaration created", "declaration deleted"},
{"manifest", "declaration deleted", "declaration created"},
{"declaration created", "manifest", "declaration deleted"},
{"declaration created", "declaration deleted", "manifest"},
{"declaration deleted", "manifest", "declaration created"},
{"declaration deleted", "declaration created", "manifest"},
}
for _, order := range orders {
t.Run(order[0]+" then "+order[1]+" then "+order[2], func(t *testing.T) {
w := newWorld(t)
n := w.boot()
w.seed(n, testBob)
addTestSecret(t, n.s, testRepoDid)
hook := newSink(t)
w.subscribe(n, hook, string(models.WebhookEventRepoRenamed))
for _, step := range order {
n.consumed(steps[step](w))
}
got := hook.expect(t, 1)
var p models.WebhookRenamePayload
if err := json.Unmarshal(got[0].body, &p); err != nil {
t.Fatal(err)
}
if got[0].event != "repository:renamed" || p.OldName != "repo" || p.NewName != "Renamed" || p.Sender.Did != testOwner.String() {
t.Errorf("delivered %q %+v", got[0].event, p)
}
rows := repoRows(t, n.s, testRepoDid)
if len(rows) != 1 || rows[0].Name != "Renamed" || rows[0].Rkey != "renamed" {
t.Errorf("repo rows after the rename: %s", describeRows(rows))
}
if secretCount(t, n.s, testRepoDid) != 1 || !isCollaborator(t, n.s, testRepoDid, testBob) {
t.Error("the rename wiped the repo's secrets or collaborators")
}
})
}
}