package db
import (
"context"
"errors"
"path/filepath"
"slices"
"strings"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/notifier"
"tangled.org/core/spindle/models"
pipelinecodec "tangled.org/core/spindle/pipeline"
"tangled.org/core/workflow"
)
const pushRepo = "did:plc:pushrepo"
func pushRecord(t *testing.T, id, repoDid, ref, oldSha, newSha string) (*models.PipelineRecord, workflow.PipelineSpec) {
t.Helper()
did := repoDid
raw := workflow.PipelineSpec{
TriggerMetadata: &workflow.TriggerMetadata{
Kind: "push",
Repo: &workflow.TriggerRepo{Knot: "knot.test", RepoDid: &did, Did: repoDid},
Push: &workflow.PushTriggerData{Ref: ref, OldSha: oldSha, NewSha: newSha},
},
Workflows: []*workflow.WorkflowSpec{{Name: "ci.yml"}},
}
record, err := pipelinecodec.FromTangled(models.PipelineId(id), time.Now(), raw)
if err != nil {
t.Fatal(err)
}
return record, raw
}
func seedPush(t *testing.T, d *DB, id, repoDid, ref, oldSha, newSha string) {
t.Helper()
record, _ := pushRecord(t, id, repoDid, ref, oldSha, newSha)
if err := d.CreatePipeline(record); err != nil {
t.Fatal(err)
}
}
func TestEnqueuePipelineStoresTheRowAndItsJobTogether(t *testing.T) {
d := newTestDB(t)
ctx := context.Background()
record, raw := pushRecord(t, "p1", pushRepo, "refs/heads/main", "aaa", "bbb")
n := notifier.New()
if err := d.EnqueuePipeline(ctx, pushRepo, record, nil, raw, false, "", "", &n); err != nil {
t.Fatal(err)
}
if id, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "aaa", "bbb"); err != nil || !found || id != "p1" {
t.Fatalf("PushPipeline = %q, %v, %v", id, found, err)
}
if job, err := d.DequeueJob(ctx, nil); err != nil || job == nil || job.PipelineId != "p1" {
t.Fatalf("dequeued %+v, %v, want the pipeline's job", job, err)
}
}
// a pipeline row alone would make the replayed push look handled forever
func TestEnqueuePipelineLeavesNoRowWhenTheJobCannotBeStored(t *testing.T) {
d := newTestDB(t)
if _, err := d.Exec(`create trigger no_jobs before insert on jobs begin select raise(abort, 'disk full'); end`); err != nil {
t.Fatal(err)
}
record, raw := pushRecord(t, "p1", pushRepo, "refs/heads/main", "aaa", "bbb")
if err := d.EnqueuePipeline(context.Background(), pushRepo, record, nil, raw, false, "", "", nil); err == nil {
t.Fatal("enqueue succeeded although the job insert failed")
}
if _, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "aaa", "bbb"); err != nil || found {
t.Fatalf("a failed enqueue left a pipeline row behind, found = %v, err = %v", found, err)
}
}
func TestEnqueuePipelineForABannedRepoLeavesNoRow(t *testing.T) {
d := newTestDB(t)
if _, err := d.Exec(`insert into bans (subject_did) values (?)`, pushRepo); err != nil {
t.Fatal(err)
}
record, raw := pushRecord(t, "p1", pushRepo, "refs/heads/main", "aaa", "bbb")
if err := d.EnqueuePipeline(context.Background(), pushRepo, record, nil, raw, false, "", "", nil); !errors.Is(err, ErrBannedSubject) {
t.Fatalf("err = %v, want ErrBannedSubject", err)
}
if _, found, _ := d.PushPipeline(pushRepo, "refs/heads/main", "aaa", "bbb"); found {
t.Fatal("a refused pipeline left a row a replay would be deduped against")
}
}
func TestPushPipelineMatchesOnRepoRefAndBothShas(t *testing.T) {
d := newTestDB(t)
seedPush(t, d, "p1", pushRepo, "refs/heads/main", "aaa", "bbb")
for name, tc := range map[string]struct {
repo, ref, old, new string
want bool
}{
"same push": {pushRepo, "refs/heads/main", "aaa", "bbb", true},
"another repo": {"did:plc:other", "refs/heads/main", "aaa", "bbb", false},
"another ref": {pushRepo, "refs/heads/dev", "aaa", "bbb", false},
"another old sha": {pushRepo, "refs/heads/main", "ccc", "bbb", false},
"another new sha": {pushRepo, "refs/heads/main", "aaa", "ccc", false},
"the push in reverse (A-B then B-A)": {pushRepo, "refs/heads/main", "bbb", "aaa", false},
} {
id, found, err := d.PushPipeline(tc.repo, tc.ref, tc.old, tc.new)
if err != nil || found != tc.want {
t.Errorf("%s: found = %v, err = %v, want found = %v", name, found, err, tc.want)
}
if found && id != "p1" {
t.Errorf("%s: id = %q, want p1", name, id)
}
}
}
func TestPushPipelineIgnoresOtherTriggerKinds(t *testing.T) {
d := newTestDB(t)
seedPipeline(t, d, "p-manual", pushRepo, "manual")
if _, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "", "sha-p-manual"); err != nil || found {
t.Fatalf("a manual pipeline matched a push: found = %v, err = %v", found, err)
}
}
func TestPushPipelineLookupUsesTheIndex(t *testing.T) {
d := newTestDB(t)
rows, err := d.Query(`explain query plan
select pipeline_id from pipelines
where kind = 'push' and repo_did = ? and commit_sha = ? and ref = ? and old_sha = ? limit 1`,
pushRepo, "b", "refs/heads/main", "a")
if err != nil {
t.Fatal(err)
}
defer rows.Close()
var plan []string
for rows.Next() {
var id, parent, unused int
var detail string
if err := rows.Scan(&id, &parent, &unused, &detail); err != nil {
t.Fatal(err)
}
plan = append(plan, detail)
}
if got := strings.Join(plan, "; "); !strings.Contains(got, "idx_pipelines_push_dedupe") {
t.Fatalf("plan = %q, want the push dedupe index", got)
}
}
func TestPushDedupeMigrationKeepsExistingPipelines(t *testing.T) {
path := filepath.Join(t.TempDir(), "spindle.db")
ctx := context.Background()
d, err := Make(ctx, path)
if err != nil {
t.Fatal(err)
}
seedPush(t, d, "old-run", pushRepo, "refs/heads/main", "aaa", "bbb")
if _, err := d.Exec(`
drop index idx_pipelines_push_dedupe;
alter table pipelines drop column ref;
alter table pipelines drop column old_sha;
delete from migrations where name = 'pipelines-push-dedupe';
`); err != nil {
t.Fatal(err)
}
d.Close()
d, err = Make(ctx, path)
if err != nil {
t.Fatalf("reopening an old db: %v", err)
}
defer d.Close()
if _, err := d.GetPipeline(ctx, "old-run"); err != nil {
t.Fatalf("pipeline lost in the migration: %v", err)
}
seedPush(t, d, "new-run", pushRepo, "refs/heads/main", "bbb", "ccc")
if id, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "bbb", "ccc"); err != nil || !found || id != "new-run" {
t.Fatalf("PushPipeline after migrating = %q, %v, %v", id, found, err)
}
}
func TestRepoRetriesSurviveReopen(t *testing.T) {
path := filepath.Join(t.TempDir(), "spindle.db")
d, err := Make(context.Background(), path)
if err != nil {
t.Fatal(err)
}
now := time.Unix(5000, 0)
if err := d.PutRepoRetry("did:plc:a", 2, now); err != nil {
t.Fatal(err)
}
if err := d.PutRepoRetry("did:plc:b", 1, now.Add(time.Hour)); err != nil {
t.Fatal(err)
}
d.Close()
d, err = Make(context.Background(), path)
if err != nil {
t.Fatal(err)
}
defer d.Close()
if due, err := d.DueRepoRetries(now); err != nil || !slices.Equal(due, []syntax.DID{"did:plc:a"}) {
t.Fatalf("due = %v, %v, want only a", due, err)
}
if n, err := d.RepoRetryAttempts("did:plc:a"); err != nil || n != 2 {
t.Fatalf("attempts = %d, %v, want 2", n, err)
}
if err := d.DeleteRepoRetry("did:plc:a"); err != nil {
t.Fatal(err)
}
if n, _ := d.RepoRetryAttempts("did:plc:a"); n != 0 {
t.Fatalf("attempts after delete = %d", n)
}
}
func TestPendingCatchUpKeepsTheOriginalTimeAndAgesOut(t *testing.T) {
d := newTestDB(t)
now := time.Unix(1_000_000, 0)
if err := d.AddPendingCatchUp([]syntax.DID{"did:plc:a", "did:plc:b"}, now.Add(-10*24*time.Hour)); err != nil {
t.Fatal(err)
}
// re-adding keeps the first timestamp, so did:plc:a stays stale
if err := d.AddPendingCatchUp([]syntax.DID{"did:plc:a", "did:plc:c"}, now); err != nil {
t.Fatal(err)
}
dropped, err := d.DropStalePendingCatchUp(now.Add(-7 * 24 * time.Hour))
slices.Sort(dropped)
if err != nil || !slices.Equal(dropped, []syntax.DID{"did:plc:a", "did:plc:b"}) {
t.Fatalf("dropped = %v, %v, want a and b", dropped, err)
}
if left, err := d.PendingCatchUp(); err != nil || !slices.Equal(left, []syntax.DID{"did:plc:c"}) {
t.Fatalf("left = %v, %v, want c", left, err)
}
if err := d.DeletePendingCatchUp("did:plc:c", "did:plc:nobody"); err != nil {
t.Fatal(err)
}
if left, _ := d.PendingCatchUp(); len(left) != 0 {
t.Fatalf("left after delete = %v", left)
}
}