package spindle
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"net/http/cgi"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/gorilla/websocket"
"github.com/klauspost/compress/zstd"
"tangled.org/core/api/org_tangled"
"tangled.org/core/jetstream"
"tangled.org/core/log"
"tangled.org/core/rbac"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/secrets"
"tangled.org/core/tapc"
)
const faultOtherRepoDid = syntax.DID("did:plc:repotwo")
const testRetryTick = 50 * time.Millisecond
type jsEvent struct {
timeUS int64
wire []byte
}
type fakeJetstream struct {
srv *httptest.Server
mu sync.Mutex
events []jsEvent
oldest int
last int64
cursors []int64
conns map[*websocket.Conn]struct{}
}
func newFakeJetstream(t *testing.T) *fakeJetstream {
t.Helper()
j := &fakeJetstream{conns: map[*websocket.Conn]struct{}{}, last: time.Now().Add(-time.Hour).UnixMicro()}
j.srv = httptest.NewServer(http.HandlerFunc(j.serve))
t.Cleanup(func() {
j.dropConnections()
j.srv.Close()
})
return j
}
func (j *fakeJetstream) url() string {
return "ws" + strings.TrimPrefix(j.srv.URL, "http") + "/subscribe"
}
// 5s apart, so the 1s rewind on resume only replays the last event
func (j *fakeJetstream) add(e jetstream.Event) int64 {
j.mu.Lock()
defer j.mu.Unlock()
j.last += 5_000_000
e.TimeUS = j.last
raw, err := json.Marshal(e)
if err != nil {
panic(err)
}
enc, _ := zstd.NewWriter(nil)
j.events = append(j.events, jsEvent{timeUS: e.TimeUS, wire: enc.EncodeAll(raw, nil)})
return e.TimeUS
}
func (j *fakeJetstream) commit(did syntax.DID, collection, rkey, op string, record any) int64 {
commit := &jetstream.Commit{Rev: "rev", Operation: op, Collection: collection, RKey: rkey, CID: "bafy"}
if record != nil {
raw, err := json.Marshal(record)
if err != nil {
panic(err)
}
commit.Record = raw
}
return j.add(jetstream.Event{Did: did.String(), Kind: jetstream.EventKindCommit, Commit: commit})
}
// the ingester ignores identity events, but the cursor still moves past them
func (j *fakeJetstream) filler() int64 {
return j.add(jetstream.Event{Did: testOwner.String(), Kind: jetstream.EventKindIdentity})
}
// like retention, a resume can't go back before this point
func (j *fakeJetstream) forget() {
j.mu.Lock()
defer j.mu.Unlock()
j.oldest = len(j.events)
}
func (j *fakeJetstream) dropConnections() {
j.mu.Lock()
defer j.mu.Unlock()
for c := range j.conns {
c.Close()
}
}
func (j *fakeJetstream) connects() int {
j.mu.Lock()
defer j.mu.Unlock()
return len(j.cursors)
}
func (j *fakeJetstream) serve(w http.ResponseWriter, r *http.Request) {
conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
if err != nil {
return
}
defer conn.Close()
j.mu.Lock()
j.conns[conn] = struct{}{}
next, cursor := len(j.events), int64(-1)
if raw := r.URL.Query().Get("cursor"); raw != "" {
cursor, _ = strconv.ParseInt(raw, 10, 64)
next = j.oldest
for next < len(j.events) && j.events[next].timeUS < cursor {
next++
}
}
j.cursors = append(j.cursors, cursor)
j.mu.Unlock()
defer func() {
j.mu.Lock()
delete(j.conns, conn)
j.mu.Unlock()
}()
gone := closedWhenDone(conn)
for {
j.mu.Lock()
var batch [][]byte
for ; next < len(j.events); next++ {
batch = append(batch, j.events[next].wire)
}
j.mu.Unlock()
for _, wire := range batch {
if err := conn.WriteMessage(websocket.BinaryMessage, wire); err != nil {
return
}
}
select {
case <-gone:
return
case <-time.After(5 * time.Millisecond):
}
}
}
// serves did docs so spindle's real resolver runs unchanged
type fakePLC struct {
srv *httptest.Server
mu sync.Mutex
knots map[syntax.DID]string
pdses map[syntax.DID]string
}
func newFakePLC(t *testing.T) *fakePLC {
t.Helper()
p := &fakePLC{knots: map[syntax.DID]string{}, pdses: map[syntax.DID]string{}}
p.srv = httptest.NewServer(http.HandlerFunc(p.serve))
t.Cleanup(p.srv.Close)
return p
}
func (p *fakePLC) setKnot(repo syntax.DID, url string) {
p.mu.Lock()
defer p.mu.Unlock()
p.knots[repo] = url
}
func (p *fakePLC) setPDS(owner syntax.DID, url string) {
p.mu.Lock()
defer p.mu.Unlock()
p.pdses[owner] = url
}
func (p *fakePLC) serve(w http.ResponseWriter, r *http.Request) {
did := syntax.DID(strings.TrimPrefix(r.URL.Path, "/"))
p.mu.Lock()
knot, isRepo := p.knots[did]
pds, isOwner := p.pdses[did]
p.mu.Unlock()
if !isRepo && !isOwner {
http.NotFound(w, r)
return
}
type service struct {
ID string `json:"id"`
Type string `json:"type"`
Endpoint string `json:"serviceEndpoint"`
}
var services []service
if isRepo {
services = append(services, service{"#tangled_knot", "TangledKnot", knot})
}
if isOwner {
services = append(services, service{"#atproto_pds", "AtprotoPersonalDataServer", pds})
}
json.NewEncoder(w).Encode(map[string]any{
"@context": []string{"https://www.w3.org/ns/did/v1"},
"id": did,
"alsoKnownAs": []string{},
"verificationMethod": []any{},
"service": services,
})
}
// requests wait on it until released
type hold struct {
release chan struct{}
fail bool
}
func (h *hold) wait(r *http.Request) bool {
select {
case <-h.release:
return true
case <-r.Context().Done():
return false
}
}
type fakeRepo struct {
work string
bare string
n int
}
func gitExecPath(t *testing.T) string {
t.Helper()
out, err := exec.Command("git", "--exec-path").Output()
if err != nil {
t.Skipf("git not available: %v", err)
}
backend := filepath.Join(strings.TrimSpace(string(out)), "git-http-backend")
if _, err := os.Stat(backend); err != nil {
t.Skipf("git http-backend not available: %v", err)
}
return backend
}
func (r *fakeRepo) git(t *testing.T, dir string, args ...string) string {
t.Helper()
cmd := exec.Command("git", append([]string{"-C", dir}, args...)...)
out, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("git %v: %v\n%s", args, err, out)
}
return strings.TrimSpace(string(out))
}
func (r *fakeRepo) commit(t *testing.T) string {
t.Helper()
return r.commitWorkflows(t, map[string]string{
"ci.yaml": "engine: dummy\nwhen:\n - event: push\n branch: main\nsteps:\n - name: hi\n command: echo hi\n",
})
}
// pushed to its own branch so the sha stays reachable
func (r *fakeRepo) commitWorkflows(t *testing.T, files map[string]string) string {
t.Helper()
r.n++
workflows := filepath.Join(r.work, ".tangled", "workflows")
if err := os.MkdirAll(workflows, 0o755); err != nil {
t.Fatal(err)
}
for name, contents := range files {
if err := os.WriteFile(filepath.Join(workflows, name), []byte(contents), 0o644); err != nil {
t.Fatal(err)
}
}
if err := os.WriteFile(filepath.Join(r.work, "n"), []byte(strconv.Itoa(r.n)), 0o644); err != nil {
t.Fatal(err)
}
r.git(t, r.work, "add", "-A")
r.git(t, r.work, "commit", "-q", "-m", "c"+strconv.Itoa(r.n))
sha := r.git(t, r.work, "rev-parse", "HEAD")
r.git(t, r.work, "push", "-q", r.bare, sha+":refs/heads/c"+strconv.Itoa(r.n))
return sha
}
type world struct {
t *testing.T
dir string
hydDir string
js *fakeJetstream
plc *fakePLC
knot *fakeKnot
other *fakeKnot
hyd *fakeHydrant
skew atomic.Int64
cgi *cgi.Handler
// empty means open admission and the usual spindle owner
admission config.Admission
owner syntax.DID
mu sync.Mutex
hydFailing map[syntax.DID]bool
gitHold *hold
manifestHold *hold
gitFetches atomic.Int32
gitFailing atomic.Bool
heldReads atomic.Int32
}
func newWorld(t *testing.T) *world {
t.Helper()
t.Parallel()
w := &world{t: t, dir: t.TempDir(), hydDir: shortTempDir(t), hydFailing: map[syntax.DID]bool{}}
w.js = newFakeJetstream(t)
w.plc = newFakePLC(t)
w.other = newFakeKnot(t)
w.knot = &fakeKnot{records: map[string]any{}}
w.knot.srv = httptest.NewServer(http.HandlerFunc(w.serveKnot))
t.Cleanup(w.knot.srv.Close)
w.plc.setPDS(testOwner, w.knot.srv.URL)
w.plc.setKnot(testRepoDid, w.knot.srv.URL)
w.plc.setKnot(faultOtherRepoDid, w.knot.srv.URL)
w.hyd = &fakeHydrant{}
w.hyd.onTrack = func(did syntax.DID) {
if w.hyd.repoStatus(did) != synced(did).Status {
w.backfillOwner(did)
}
}
w.hyd.onResync = func(did syntax.DID) {
if w.hyd.repoStatus(did) != "" {
w.backfillOwner(did)
}
}
return w
}
func (w *world) dbPath() string { return filepath.Join(w.dir, "spindle.db") }
func (w *world) spindleDid() string { return config.Server{Hostname: faultHostname}.Did().String() }
const faultHostname = "spindle.example.com"
func (w *world) knotRef() knotclient.Knot {
return knotclient.Knot{Did: syntax.DID("did:web:" + w.knot.host()), Host: w.knot.host(), Scheme: "http"}
}
func (w *world) serveKnot(rw http.ResponseWriter, r *http.Request) {
if strings.HasPrefix(r.URL.Path, "/did:") {
w.serveGit(rw, r)
return
}
w.mu.Lock()
held := w.manifestHold
w.mu.Unlock()
if held != nil && r.URL.Query().Get("collection") == org_tangled.RepoManifestNSID {
w.heldReads.Add(1)
if !held.wait(r) {
return
}
}
w.knot.serve(rw, r)
}
func (w *world) serveGit(rw http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/info/refs") {
w.gitFetches.Add(1)
}
w.mu.Lock()
held, handler := w.gitHold, w.cgi
w.mu.Unlock()
if held != nil {
if !held.wait(r) {
return
}
if held.fail {
http.Error(rw, "boom", http.StatusInternalServerError)
return
}
}
if w.gitFailing.Load() {
http.Error(rw, "boom", http.StatusInternalServerError)
return
}
if handler == nil {
http.Error(rw, "no git", http.StatusNotFound)
return
}
handler.ServeHTTP(rw, r)
}
func (w *world) serveGitRepo(repoDid syntax.DID) *fakeRepo {
t := w.t
t.Helper()
backend := gitExecPath(t)
repo := &fakeRepo{work: filepath.Join(w.dir, "work"), bare: filepath.Join(w.dir, "git", repoDid.String())}
for _, dir := range []string{repo.work, repo.bare} {
if err := os.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
}
repo.git(t, repo.work, "init", "-q", "-b", "main")
repo.git(t, repo.bare, "init", "-q", "--bare")
repo.git(t, repo.bare, "config", "uploadpack.allowFilter", "true")
repo.git(t, repo.bare, "config", "uploadpack.allowAnySHA1InWant", "true")
w.mu.Lock()
defer w.mu.Unlock()
w.cgi = &cgi.Handler{
Path: backend,
Root: "/",
Env: []string{"GIT_PROJECT_ROOT=" + filepath.Join(w.dir, "git"), "GIT_HTTP_EXPORT_ALL=1"},
InheritEnv: []string{"PATH", "HOME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM"},
}
return repo
}
func (w *world) hold(slot **hold) (release func(fail bool)) {
h := &hold{release: make(chan struct{})}
w.mu.Lock()
*slot = h
w.mu.Unlock()
var once sync.Once
return func(fail bool) {
once.Do(func() {
h.fail = fail
w.mu.Lock()
*slot = nil
w.mu.Unlock()
close(h.release)
})
}
}
func (w *world) holdGit() func(fail bool) { return w.hold(&w.gitHold) }
func (w *world) holdManifest() func(fail bool) { return w.hold(&w.manifestHold) }
func (w *world) setManifest(repo syntax.DID, name, spindle string) {
w.knot.put(repo, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(testOwner, name, spindle))
}
func manifestRecord(owner syntax.DID, name, spindle string) org_tangled.RepoManifest {
m := org_tangled.RepoManifest{
LexiconTypeID: org_tangled.RepoManifestNSID,
CreatedAt: "2026-01-01T00:00:00Z",
Declaration: &org_tangled.RepoManifest_Declaration{Owner: owner.String(), Name: name},
}
if spindle != "" {
m.Spindle = &spindle
}
return m
}
func (w *world) declare(name string, repo syntax.DID) {
w.knot.declare(testOwner, strings.ToLower(name), repo)
}
func (w *world) undeclare(name string) { w.knot.undeclare(testOwner, strings.ToLower(name)) }
func (w *world) addCollaborator(repo, who syntax.DID) {
w.knot.put(repo, org_tangled.GraphCollaboratorNSID, who.String(), map[string]string{"createdAt": "2026-01-01T00:00:00Z"})
}
func (w *world) removeCollaborator(repo, who syntax.DID) {
w.knot.erase(repo, org_tangled.GraphCollaboratorNSID, who.String())
}
func (w *world) serve(repo syntax.DID, name string, collaborators ...syntax.DID) {
w.setManifest(repo, name, w.spindleDid())
w.declare(name, repo)
for _, c := range collaborators {
w.addCollaborator(repo, c)
}
}
func (w *world) emitManifest(repo syntax.DID) int64 {
return w.js.manifest(repo, w.knot.record(repo, org_tangled.RepoManifestNSID, knotclient.ManifestRkey))
}
func (w *world) emitCollaborator(repo, who syntax.DID, op string) int64 {
return w.js.collaborator(repo, who, op)
}
func (w *world) emitDeclaration(name string, repo syntax.DID) int64 {
return w.js.declaration(testOwner, name, repo)
}
func (j *fakeJetstream) manifest(repo syntax.DID, value any) int64 {
return j.commit(repo, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, jetstream.CommitOperationUpdate, value)
}
func (j *fakeJetstream) collaborator(repo, who syntax.DID, op string) int64 {
var record any
if op != jetstream.CommitOperationDelete {
record = map[string]string{"createdAt": "2026-01-01T00:00:00Z"}
}
return j.commit(repo, org_tangled.GraphCollaboratorNSID, who.String(), op, record)
}
func (j *fakeJetstream) declaration(owner syntax.DID, name string, repo syntax.DID) int64 {
record := org_tangled.RepoDeclaration{LexiconTypeID: org_tangled.RepoDeclarationNSID, Repo: repo.String()}
return j.commit(owner, org_tangled.RepoDeclarationNSID, strings.ToLower(name), opCreate, record)
}
func (k *fakeKnot) record(repo syntax.DID, collection, rkey string) any {
k.mu.Lock()
defer k.mu.Unlock()
return k.records[repo.String()+"|"+collection+"|"+rkey]
}
func (w *world) backfillOwner(did syntax.DID) {
w.mu.Lock()
failing := w.hydFailing[did]
w.mu.Unlock()
if failing {
w.hyd.setRepo(failedWith(did, "transport", 0))
return
}
w.hyd.backfill(did, w.knot.recordsOf(did))
}
// set the repo first, or a listing can call it backfilled before the events land
func (f *fakeHydrant) backfill(did syntax.DID, events []tapc.RecordEventData) {
f.setRepo(retrying(did))
for _, rec := range events {
f.addEvent(rec)
}
f.setRepo(synced(did))
}
func (w *world) setHydrantFailing(did syntax.DID, on bool) {
w.mu.Lock()
defer w.mu.Unlock()
w.hydFailing[did] = on
}
// backfill only sees current state, so every record is a create
func (k *fakeKnot) recordsOf(did syntax.DID) []tapc.RecordEventData {
var out []tapc.RecordEventData
k.mu.Lock()
defer k.mu.Unlock()
prefix := did.String() + "|"
for key, value := range k.records {
rest, ok := strings.CutPrefix(key, prefix)
if !ok {
continue
}
collection, rkey, _ := strings.Cut(rest, "|")
raw, err := json.Marshal(value)
if err != nil {
panic(err)
}
out = append(out, tapc.RecordEventData{Did: did, Collection: syntax.NSID(collection), Rkey: syntax.RecordKey(rkey), Rev: "rev", Action: tapc.RecordCreateAction, Record: raw})
}
return out
}
type node struct {
t *testing.T
s *Spindle
cancel context.CancelFunc
done chan error
}
func (w *world) boot() *node {
t := w.t
t.Helper()
connects := w.js.connects()
ctx := log.IntoContext(context.Background(), slog.Default().With("test", t.Name()))
d, err := db.Make(ctx, w.dbPath())
if err != nil {
t.Fatal(err)
}
admission, owner := w.admission, w.owner
if admission == "" {
admission = config.AdmissionOpen
}
if owner == "" {
owner = testOwner
}
cfg := &config.Config{Role: config.RoleStandalone}
cfg.Server = config.Server{
ListenAddr: "127.0.0.1:0",
DBPath: w.dbPath(),
RepoDir: filepath.Join(w.dir, "repos"),
LogDir: filepath.Join(w.dir, "logs"),
Hostname: faultHostname,
JetstreamEndpoint: w.js.url(),
PlcUrl: w.plc.srv.URL,
Hydrant: config.Hydrant{Dir: w.hydDir},
Dev: true,
Owner: owner.String(),
Admission: admission,
}
cfg.ArtifactStores.Disk.Dir = filepath.Join(w.dir, "artifacts")
s, err := New(ctx, cfg, d, map[string]models.Engine{}, nil)
if err != nil {
t.Fatalf("New: %v", err)
}
s.backfill.start = w.hyd.start
s.backfill.poll = 10 * time.Millisecond
s.backfill.retryBase = time.Millisecond
s.retries.now = func() time.Time { return time.Now().Add(time.Duration(w.skew.Load())) }
s.retries.tick = testRetryTick
runCtx, cancel := context.WithCancel(ctx)
n := &node{t: t, s: s, cancel: cancel, done: make(chan error, 1)}
go func() { n.done <- s.Start(runCtx) }()
t.Cleanup(n.stop)
// a live connection only gets events sent after it connects, so wait for it
eventually(t, 10*time.Second, "jetstream to connect", func() bool { return w.js.connects() > connects })
return n
}
func (n *node) stop() {
if n.cancel == nil {
return
}
n.cancel()
n.cancel = nil
select {
case err := <-n.done:
if err != nil {
n.t.Errorf("Start: %v", err)
}
case <-time.After(60 * time.Second):
n.t.Fatal("spindle did not stop")
}
}
func repoRows(t *testing.T, s *Spindle, repo syntax.DID) []db.Repo {
t.Helper()
rows, err := s.db.ReposByDid(repo)
if err != nil {
t.Fatal(err)
}
return rows
}
func isCollaborator(t *testing.T, s *Spindle, repo, who syntax.DID) bool {
t.Helper()
ok, err := s.e.IsRepoCollaborator(who.String(), rbac.ThisServer, repo.String())
if err != nil {
t.Fatal(err)
}
return ok
}
func secretCount(t *testing.T, s *Spindle, repo syntax.DID) int {
t.Helper()
got, err := s.vault.GetSecretsUnlocked(context.Background(), secrets.RepoIdentifier(repo.String()))
if err != nil {
t.Fatal(err)
}
return len(got)
}
func addTestSecret(t *testing.T, s *Spindle, repo syntax.DID) {
t.Helper()
mustAddSecret(t, s.vault, repo.String(), "api_key", "v", time.Now(), testOwner.String())
}
func (n *node) pipelines(repo syntax.DID) int {
n.t.Helper()
records, _, _, err := n.s.db.QueryPipelines(context.Background(), repo.String(), nil, "", nil, 100)
if err != nil {
n.t.Fatal(err)
}
return len(records)
}
// a restarted spindle works its backlog in the background, so poll
func eventually(t *testing.T, within time.Duration, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(within)
for !cond() {
if time.Now().After(deadline) {
t.Fatalf("timed out after %v waiting for %s", within, what)
}
time.Sleep(20 * time.Millisecond)
}
}
func (w *world) quiet() {
w.t.Helper()
last := -1
stable := 0
for stable < 5 {
time.Sleep(40 * time.Millisecond)
count := w.traffic()
if count == last {
stable++
} else {
stable, last = 0, count
}
}
}
func (w *world) traffic() int { return w.knot.count() + w.other.count() }
func (w *world) setKnotDown(on bool) {
w.knot.mu.Lock()
defer w.knot.mu.Unlock()
w.knot.failing = on
}
func (w *world) setBare404(on bool) {
w.knot.mu.Lock()
defer w.knot.mu.Unlock()
w.knot.bare404 = on
}
func (w *world) setPDSFailing(on bool) {
w.knot.mu.Lock()
defer w.knot.mu.Unlock()
w.knot.failCollection = ""
if on {
w.knot.failCollection = org_tangled.RepoDeclarationNSID
}
}
func describeRows(rows []db.Repo) string {
var out []string
for _, r := range rows {
out = append(out, fmt.Sprintf("%s/%s@%s", r.Owner, r.Rkey, r.Knot))
}
return "[" + strings.Join(out, " ") + "]"
}
// wait on this before a stop, or the cursor stays behind
func (n *node) consumed(us int64) {
n.t.Helper()
eventually(n.t, 15*time.Second, "spindle to finish the jetstream events", func() bool { return n.s.jc.LastSeenUs() > us })
}
// spindle reads an event before it acts on it. this waits until it has acted, or failed and scheduled a retry
func (n *node) applied(repo syntax.DID, us int64) {
n.t.Helper()
n.consumed(us)
eventually(n.t, 15*time.Second, "the repo job to try its marks", func() bool {
now := n.s.retries.clock()
_, due, err := n.s.db.RepoRetryDue(repo, now)
if err != nil || due {
return false
}
marks, err := n.s.db.DueRecordMarksOf(repo, now)
return err == nil && len(marks) == 0
})
}
func (n *node) cursor() int64 {
n.t.Helper()
us, err := n.s.db.GetLastTimeUs()
if errors.Is(err, sql.ErrNoRows) {
return 0
}
if err != nil {
n.t.Fatal(err)
}
return us
}