package spindle
import (
"context"
"encoding/json"
"os"
"path/filepath"
"tangled.org/core/api/org_tangled"
"testing"
"time"
comatproto "github.com/bluesky-social/indigo/api/atproto"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/jetstream"
"tangled.org/core/log"
"tangled.org/core/notifier"
"tangled.org/core/rbac"
"tangled.org/core/spindle/artifactstore"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/secrets"
)
func newWipeTestSpindle(t *testing.T) *Spindle {
t.Helper()
tmp := t.TempDir()
ctx := context.Background()
d, err := db.Make(ctx, filepath.Join(tmp, "spindle.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { d.Close() })
e, err := rbac.NewEnforcer(filepath.Join(tmp, "rbac.db"))
if err != nil {
t.Fatal(err)
}
vault, err := secrets.NewSQLiteManager(filepath.Join(tmp, "secrets.db"))
if err != nil {
t.Fatal(err)
}
stores, err := artifactstore.NewStores(config.ArtifactStores{
Disk: config.ArtifactStoreDisk{Dir: filepath.Join(tmp, "artifacts")},
}, "", "")
if err != nil {
t.Fatal(err)
}
jc, err := jetstream.NewJetstreamClient("", nil, nil, log.New("test"), d, false, false)
if err != nil {
t.Fatal(err)
}
return &Spindle{
l: log.New("test"),
db: d,
e: e,
vault: vault,
stores: stores,
jc: jc,
n: ptr(notifier.New()),
engs: map[string]models.Engine{},
cfg: &config.Config{
Server: config.Server{
Hostname: "spindle.example.com",
RepoDir: filepath.Join(tmp, "repos"),
LogDir: filepath.Join(tmp, "logs"),
},
},
}
}
func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) {
t.Helper()
ctx := context.Background()
sfx := repoDid.String()[len(repoDid.String())-1:]
if err := s.db.AddRepo(db.Repo{Knot: "knot.example.com", Owner: owner, Rkey: syntax.RecordKey("r" + sfx), RepoDid: repoDid}); err != nil {
t.Fatal(err)
}
if err := s.db.AddRepoCollaborator(db.RepoCollaborator{OwnerDid: owner, Rkey: syntax.RecordKey("c" + sfx), Subject: "did:plc:collab", RepoDid: repoDid}); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into jobs (repo_did, pipeline_id, tpl, created_at_ns) values (?, ?, '{}', 0)`, repoDid.String(), "p"+sfx); err != nil {
t.Fatal(err)
}
if err := s.db.SaveMillLease(db.MillLease{LeaseID: "l" + sfx, NodeID: "n1", Epoch: "e1", Engine: "microvm", PipelineID: "p" + sfx, Workflow: "w" + sfx, State: "active", RepoDID: repoDid.String()}); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into mill_artifacts (lease_id, repo_did, pipeline_id, workflow, ref, hash) values (?, ?, ?, ?, ?, 'h')`, "l"+sfx, repoDid.String(), "p"+sfx, "w"+sfx, "out/l"+sfx+".bin"); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into quota_allocations (repo_did, resource, kind, key, amount) values (?, 'compute', 'generic', 'k', 1)`, repoDid.String()); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into webhooks (repo_did, url, events) values (?, 'https://example.com/hook', 'push')`, repoDid.String()); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into webhook_deliveries (webhook_id, event, delivery_id, url, success) values ((select max(id) from webhooks), 'push', ?, 'https://example.com/hook', 1)`, "d"+sfx); err != nil {
t.Fatal(err)
}
if err := s.db.AddPullTicket(repoDid, "pull1", db.PullTicket{Author: owner, State: "open"}); err != nil {
t.Fatal(err)
}
if err := s.db.AddPullPatch(repoDid, db.PullPatch{Rkey: "patch1", Ticket: "pull1", Head: "abc", CreatedAt: "2026-01-01T00:00:00Z"}, db.PullEvent{RepoDid: repoDid, Ticket: "pull1", Pull: []byte("{}")}); err != nil {
t.Fatal(err)
}
if err := s.db.PutRecordRetry(db.RecordRetry{RepoDid: repoDid, Collection: "org.tangled.track.ticket", Rkey: "pull1"}, 1, time.Now()); err != nil {
t.Fatal(err)
}
if err := s.vault.AddSecret(ctx, secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: owner}); err != nil {
t.Fatal(err)
}
pipeline := &org_tangled.CiDefs_Pipeline{
Id: "p" + sfx,
Repo: repoDid.String(),
Workflows: []*org_tangled.CiDefs_WorkflowRun{{Id: "w" + sfx, Name: ptr("w" + sfx), Status: "pending"}},
}
payload, err := json.Marshal(pipeline)
if err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into pipelines (pipeline_id, repo_did, commit_sha, kind, payload) values (?, ?, '', '', ?)`, pipeline.Id, repoDid.String(), string(payload)); err != nil {
t.Fatal(err)
}
if _, err := s.db.Exec(`insert into workflow_statuses (pipeline_id, workflow, status, created_at) values (?, ?, 'pending', 'now')`, pipeline.Id, "w"+sfx); err != nil {
t.Fatal(err)
}
if err := os.MkdirAll(s.newRepoPath(repoDid), 0755); err != nil {
t.Fatal(err)
}
for _, ref := range []string{"out/l" + sfx + ".bin", "logs/l" + sfx + ".log"} {
for _, put := range s.stores.PutFile(ctx, ref, writeTempFile(t, "data")) {
if put.Err != nil {
t.Fatal(put.Err)
}
}
}
}
func writeTempFile(t *testing.T, content string) string {
t.Helper()
p := filepath.Join(t.TempDir(), "src")
if err := os.WriteFile(p, []byte(content), 0644); err != nil {
t.Fatal(err)
}
return p
}
func TestWipeRepoRemovesAllState(t *testing.T) {
s := newWipeTestSpindle(t)
ctx := context.Background()
repoDid := syntax.DID("did:plc:repo1")
owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner)
if err := s.WipeRepo(ctx, repoDid, "test"); err != nil {
t.Fatalf("WipeRepo: %v", err)
}
var n int
for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "pipelines", "workflow_statuses", "webhooks", "webhook_deliveries", "pull_tickets", "pull_patches", "pull_events", "record_retries"} {
if err := s.db.QueryRow(`select count(*) from ` + table).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Fatalf("%s: expected empty, got %d rows", table, n)
}
}
got, err := s.vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid.String()))
if err != nil {
t.Fatal(err)
}
if len(got) != 0 {
t.Fatalf("expected secrets gone, got %d", len(got))
}
if _, err := os.Stat(s.newRepoPath(repoDid)); !os.IsNotExist(err) {
t.Fatalf("expected clone dir gone, stat err: %v", err)
}
if _, err := s.stores.Open(ctx, "out/l1.bin"); err == nil {
t.Fatal("expected artifact gone")
}
if _, err := s.stores.Open(ctx, "logs/l1.log"); err == nil {
t.Fatal("expected log artifact gone")
}
if err := s.WipeRepo(ctx, repoDid, "test again"); err != nil {
t.Fatalf("second WipeRepo: %v", err)
}
}
func TestWipeShouldKeepAllowListedOwners(t *testing.T) {
s := newWipeTestSpindle(t)
ctx := context.Background()
repoDid := syntax.DID("did:plc:repo1")
owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner)
if err := db.AddToAllowlist(s.db, owner); err != nil {
t.Fatal(err)
}
if err := s.WipeRepo(ctx, repoDid, "test"); err != nil {
t.Fatalf("WipeRepo: %v", err)
}
onList, err := db.AllowlistContains(s.db, owner)
if err != nil {
t.Fatal(err)
}
if !onList {
t.Fatal("A wipe mustn't take the owner off the allow-list.")
}
}
func TestWipeRepoRemovesLocalEngineArtifacts(t *testing.T) {
s := newWipeTestSpindle(t)
ctx := context.Background()
repoDid := syntax.DID("did:plc:repo1")
owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner)
wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"}
localRef := "logs/" + wid.String() + ".log"
if err := s.db.SaveArtifactRef(wid.String(), repoDid.String(), wid, localRef, "h"); err != nil {
t.Fatal(err)
}
for _, put := range s.stores.PutFile(ctx, localRef, writeTempFile(t, "log")) {
if put.Err != nil {
t.Fatal(put.Err)
}
}
if err := s.WipeRepo(ctx, repoDid, "test"); err != nil {
t.Fatalf("WipeRepo: %v", err)
}
if _, err := s.stores.Open(ctx, localRef); err == nil {
t.Fatal("expected local engine artifact gone")
}
var n int
if err := s.db.QueryRow(`select count(*) from mill_artifacts`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Fatalf("expected mill_artifacts empty, got %d rows", n)
}
}
func TestWipeOwnerWipesAllOwnedRepos(t *testing.T) {
s := newWipeTestSpindle(t)
ctx := context.Background()
owner := syntax.DID("did:plc:owner1")
other := syntax.DID("did:plc:owner2")
seedWipeRepo(t, s, "did:plc:repo1", owner)
seedWipeRepo(t, s, "did:plc:repo2", owner)
seedWipeRepo(t, s, "did:plc:repo3", other)
if err := s.WipeOwner(ctx, owner, "test"); err != nil {
t.Fatalf("WipeOwner: %v", err)
}
var n int
if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 1 {
t.Fatalf("expected only the other owner's repo left, got %d", n)
}
repos, err := s.db.ReposByDid("did:plc:repo3")
if err != nil || len(repos) != 1 {
t.Fatalf("expected repo3 intact, got %v err %v", repos, err)
}
}
type destroyRecorder struct {
models.Engine
destroyed []models.WorkflowId
}
func (e *destroyRecorder) DestroyWorkflow(_ context.Context, wid models.WorkflowId) error {
e.destroyed = append(e.destroyed, wid)
return nil
}
func TestWipeDestroysFinishedWorkflowStillHoldingItsVM(t *testing.T) {
s := newWipeTestSpindle(t)
repoDid := syntax.DID("did:plc:repo1")
seedWipeRepo(t, s, repoDid, "did:plc:owner1")
eng := &destroyRecorder{}
s.engs["microvm"] = eng
wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"}
if err := s.db.StatusFailed(wid, "User step error: exited with code 1", 1, s.n); err != nil {
t.Fatal(err)
}
if _, err := s.cancelRepoWorkflows(context.Background(), repoDid, "banned"); err != nil {
t.Fatal(err)
}
if len(eng.destroyed) != 1 || eng.destroyed[0] != wid {
t.Fatalf("destroyed = %v, want the held workflow", eng.destroyed)
}
if st, err := s.db.GetStatus(wid); err != nil || st != models.StatusKindFailed {
t.Fatalf("status = %q (err %v), want the failure kept", st, err)
}
}
func ptr[T any](v T) *T { return &v }
func TestIngestAccountTakedownWipes(t *testing.T) {
s := newWipeTestSpindle(t)
owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, "did:plc:repo1", owner)
status := "takendown"
err := s.ingest()(context.Background(), &jetstream.Event{
Kind: jetstream.EventKindAccount,
Did: owner.String(),
Account: &comatproto.SyncSubscribeRepos_Account{Did: owner.String(), Active: false, Status: &status},
})
if err != nil {
t.Fatalf("ingest account event: %v", err)
}
var n int
if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Fatalf("expected repos wiped after takedown, got %d", n)
}
seedWipeRepo(t, s, "did:plc:repo1", owner)
err = s.ingest()(context.Background(), &jetstream.Event{
Kind: jetstream.EventKindAccount,
Did: owner.String(),
Account: &comatproto.SyncSubscribeRepos_Account{Did: owner.String(), Active: true},
})
if err != nil {
t.Fatal(err)
}
if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 1 {
t.Fatalf("expected active account untouched, got %d", n)
}
}