package main
import (
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"path/filepath"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/oltp"
)
// A completed run raises ci.completed on the event plane, carrying the
// runId — the one place a run's terminal transition is learned, so
// every live page can wait on the event instead of asking dagnats on
// its own timer (RFC 0005).
func TestPollCIRaisesCICompletedWithRunID(t *testing.T) {
dagnats := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
json.NewEncoder(w).Encode(ci.Run{
RunID: "r1",
Status: "completed",
CompletedAt: "2026-08-29T12:00:00Z",
})
}))
defer dagnats.Close()
engine := ci.New(dagnats.URL)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err := db.AddCIRun(oltp.CIRun{RunID: "r1", Repo: "alpha", Checkin: "c1",
Workflow: "w", Pipeline: "ci/p.json", Status: "running"}); err != nil {
t.Fatal(err)
}
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
sub, err := ev.Conn().SubscribeSync("quarry.evt.alpha.ci.completed")
if err != nil {
t.Fatal(err)
}
defer sub.Unsubscribe()
ev.Conn().Flush()
pollCI(engine, db, nil, false, t.TempDir(), ev)
msg, err := sub.NextMsg(3 * time.Second)
if err != nil {
t.Fatalf("ci.completed not raised: %v", err)
}
var got struct {
Kind string `json:"kind"`
Repo string `json:"repo"`
Payload map[string]any `json:"payload"`
}
if err := json.Unmarshal(msg.Data, &got); err != nil {
t.Fatal(err)
}
if got.Kind != "ci.completed" || got.Repo != "alpha" {
t.Fatalf("event = %+v, want kind ci.completed repo alpha", got)
}
if got.Payload["runId"] != "r1" {
t.Fatalf("payload = %+v, want runId r1", got.Payload)
}
rows, err := db.CIRuns("alpha", 10)
if err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0].Status != "completed" || rows[0].Completed == "" {
t.Fatalf("run not recorded terminal: %+v", rows)
}
}
// A verdict that could not be written is written on the next pass.
//
// The run used to be marked complete first, so a failed tag write left
// a finished row the poller never asks about again, and the landing
// waited on a result that would never arrive.
func TestAVerdictThatFailedToWriteIsWrittenOnTheNextPass(t *testing.T) {
dagnats := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
json.NewEncoder(w).Encode(ci.Run{RunID: "r1", Status: "completed",
CompletedAt: "2026-08-29T12:00:00Z"})
}))
defer dagnats.Close()
engine := ci.New(dagnats.URL)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err := db.AddCIRun(oltp.CIRun{RunID: "r1", Repo: "alpha", Checkin: "c1",
Workflow: "w", Pipeline: "ci/p.json", Status: "running"}); err != nil {
t.Fatal(err)
}
// The lease holder is the seam every tag write passes through: the
// first attempt fails, the next one lands.
attempts, written := 0, 0
repos := api.NewRepos(map[string]*api.Repo{"alpha": {
Guard: func(func() error) error {
attempts++
if attempts == 1 {
return errors.New("lease lost")
}
written++
return nil
},
}})
unfinished := func() bool {
runs, err := db.UnfinishedCIRuns()
if err != nil {
t.Fatal(err)
}
return len(runs) == 1 && runs[0].RunID == "r1"
}
pollCI(engine, db, repos, true, t.TempDir(), nil)
if attempts == 1 && written == 0 && unfinished() {
t.Log("ok: a failed verdict write leaves the run on the poller's work list")
} else {
t.Fatalf("FAIL: after a failed write attempts=%d written=%d unfinished=%v",
attempts, written, unfinished())
}
pollCI(engine, db, repos, true, t.TempDir(), nil)
if written == 1 && !unfinished() {
t.Log("ok: the next pass writes the verdict and only then completes the run")
} else {
t.Fatalf("FAIL: after the retry attempts=%d written=%d unfinished=%v",
attempts, written, unfinished())
}
}