From 5c76e96a88fb244299afaef4cb375aafa6b9e2a5 Mon Sep 17 00:00:00 2001 From: dawn Date: Thu, 22 Oct 2026 06:41:22 +0300 Subject: [PATCH] spindle: serve the ci routes under org.tangled only Signed-off-by: dawn --- spindle/db/db.go | 22 +- spindle/db/delete_test.go | 4 +- spindle/db/identity_migration_test.go | 15 +- spindle/db/pipelines_test.go | 16 +- spindle/describe_source_test.go | 47 ---- spindle/engines/nixery/engine.go | 2 +- spindle/engines/nixery/envs.go | 2 - spindle/engines/nixery/seats.go | 13 +- spindle/engines/nixery/seats_test.go | 17 ++ spindle/faults_test.go | 7 +- spindle/fingerprint_test.go | 34 --- spindle/hydrant_test.go | 6 +- spindle/ingester_test.go | 3 +- spindle/notify_event.go | 30 ++- spindle/notify_event_test.go | 14 +- spindle/pipeline/codec.go | 59 ----- spindle/server.go | 63 +----- spindle/source_repo_test.go | 2 +- spindle/wipe_test.go | 6 +- spindle/workflow_defs.go | 18 +- spindle/workflow_read_test.go | 17 -- ...i_pipeline_describe_workflow_definition.go | 50 ----- spindle/xrpc/ci_pipeline_subscribe_logs.go | 26 +-- spindle/xrpc/ci_pipeline_trigger_pipeline.go | 201 ------------------ spindle/xrpc/ci_query_pipelines.go | 87 -------- spindle/xrpc/ci_query_pipelines_test.go | 43 ---- .../xrpc/ci_subscribe_logs_workflows_test.go | 4 +- spindle/xrpc/cors_test.go | 14 +- spindle/xrpc/definitions_busy_test.go | 10 +- spindle/xrpc/legacy_routes_test.go | 34 +++ ...e.go => org_tangled_ci_cancel_pipeline.go} | 6 +- .../xrpc/org_tangled_ci_trigger_pipeline.go | 146 +++++++++++++ .../xrpc/org_tangled_event_notify_event.go | 4 +- .../org_tangled_event_notify_event_test.go | 9 +- spindle/xrpc/owner.go | 31 --- spindle/xrpc/repo_auth_test.go | 18 +- spindle/xrpc/repo_lookup.go | 53 +++++ spindle/xrpc/trigger_paused_test.go | 6 +- spindle/xrpc/trigger_sha_test.go | 8 +- spindle/xrpc/xrpc.go | 22 +- spindle/xrpc/xrpc_test.go | 34 ++- 41 files changed, 394 insertions(+), 809 deletions(-) delete mode 100644 spindle/describe_source_test.go delete mode 100644 spindle/fingerprint_test.go delete mode 100644 spindle/xrpc/ci_pipeline_describe_workflow_definition.go delete mode 100644 spindle/xrpc/ci_pipeline_trigger_pipeline.go delete mode 100644 spindle/xrpc/ci_query_pipelines.go delete mode 100644 spindle/xrpc/ci_query_pipelines_test.go create mode 100644 spindle/xrpc/legacy_routes_test.go rename spindle/xrpc/{pipeline_cancel_pipeline.go => org_tangled_ci_cancel_pipeline.go} (94%) create mode 100644 spindle/xrpc/org_tangled_ci_trigger_pipeline.go delete mode 100644 spindle/xrpc/owner.go diff --git a/spindle/db/db.go b/spindle/db/db.go index 77887c412..5322b027b 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -10,7 +10,6 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" _ "github.com/mattn/go-sqlite3" - "tangled.org/core/api/tangled" "tangled.org/core/log" "tangled.org/core/orm" "tangled.org/core/spindle/models" @@ -1368,12 +1367,27 @@ func (d *DB) GetRepoOwnerAndDid(knot, rkey string) (owner string, repoDid string return } +// the event log from before pipelines had their own table +const ( + legacyPipelineNSID = "sh.tangled.pipeline" + legacyPipelineStatusNSID = "sh.tangled.pipeline.status" +) + +type legacyPipelineStatus struct { + CreatedAt string `json:"createdAt"` + Error *string `json:"error,omitempty"` + ExitCode *int64 `json:"exitCode,omitempty"` + Pipeline string `json:"pipeline"` + Status string `json:"status"` + Workflow string `json:"workflow"` +} + // migratePipelines orders converted rows by their original creation time so // pagination keeps the historical order. func migratePipelines(tx *sql.Tx, logger *slog.Logger) error { rows, err := tx.Query( `select rkey, event, created from events where nsid = ? order by created asc`, - tangled.PipelineNSID, + legacyPipelineNSID, ) if err != nil { return err @@ -1441,7 +1455,7 @@ func migratePipelines(tx *sql.Tx, logger *slog.Logger) error { func migrateWorkflowStatuses(tx *sql.Tx, logger *slog.Logger) error { rows, err := tx.Query( `select event from events where nsid = ? order by created asc`, - tangled.PipelineStatusNSID, + legacyPipelineStatusNSID, ) if err != nil { return err @@ -1462,7 +1476,7 @@ func migrateWorkflowStatuses(tx *sql.Tx, logger *slog.Logger) error { return err } - var st tangled.PipelineStatus + var st legacyPipelineStatus if err := json.Unmarshal([]byte(eventJson), &st); err != nil { skipped++ continue diff --git a/spindle/db/delete_test.go b/spindle/db/delete_test.go index 0022cad4f..b96118e08 100644 --- a/spindle/db/delete_test.go +++ b/spindle/db/delete_test.go @@ -4,10 +4,10 @@ import ( "context" "encoding/json" "fmt" + "tangled.org/core/api/org_tangled" "testing" "github.com/bluesky-social/indigo/atproto/syntax" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/models" ) @@ -168,7 +168,7 @@ func TestDeletePipelinesByRepo(t *testing.T) { other := "did:plc:repo2" insertPipeline := func(rkey, repoDid string) { - p := &tangled.CiPipeline{Id: rkey, Repo: repoDid} + p := &org_tangled.CiPipeline{Id: rkey, Repo: repoDid} payload, err := json.Marshal(p) if err != nil { t.Fatal(err) diff --git a/spindle/db/identity_migration_test.go b/spindle/db/identity_migration_test.go index 4979d3e53..2c8772fe3 100644 --- a/spindle/db/identity_migration_test.go +++ b/spindle/db/identity_migration_test.go @@ -9,7 +9,6 @@ import ( "slices" "testing" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/models" "tangled.org/core/workflow" ) @@ -109,20 +108,20 @@ func TestPipelineMigrationSkipsMalformedLegacyRows(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "3mu2xwiorc2xm", tangled.PipelineNSID, validJSON, int64(2)); err != nil { + if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "3mu2xwiorc2xm", legacyPipelineNSID, validJSON, int64(2)); err != nil { t.Fatal(err) } - statusJSON, err := json.Marshal(tangled.PipelineStatus{ + statusJSON, err := json.Marshal(legacyPipelineStatus{ Pipeline: "at://did:plc:validrepo/sh.tangled.pipeline/3mu2xwiorc2xm", Workflow: "ci.yml", Status: "success", CreatedAt: "2026-08-31T08:00:00Z", }) if err != nil { t.Fatal(err) } - if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "status-valid", tangled.PipelineStatusNSID, statusJSON, int64(3)); err != nil { + if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "status-valid", legacyPipelineStatusNSID, statusJSON, int64(3)); err != nil { t.Fatal(err) } - if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "status-broken", tangled.PipelineStatusNSID, "{", int64(4)); err != nil { + if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "status-broken", legacyPipelineStatusNSID, "{", int64(4)); err != nil { t.Fatal(err) } invalidDefinitionPipeline := validPipeline @@ -134,18 +133,18 @@ func TestPipelineMigrationSkipsMalformedLegacyRows(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "3mu2xwiorc2xn", tangled.PipelineNSID, invalidDefinitionJSON, int64(5)); err != nil { + if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "3mu2xwiorc2xn", legacyPipelineNSID, invalidDefinitionJSON, int64(5)); err != nil { t.Fatal(err) } - orphanStatusJSON, err := json.Marshal(tangled.PipelineStatus{ + orphanStatusJSON, err := json.Marshal(legacyPipelineStatus{ Pipeline: "at://did:plc:validrepo/sh.tangled.pipeline/3mu2xwiorc2xn", Workflow: "broken.yml", Status: "failed", CreatedAt: "2026-08-31T08:01:00Z", }) if err != nil { t.Fatal(err) } - if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "status-orphan", tangled.PipelineStatusNSID, orphanStatusJSON, int64(6)); err != nil { + if _, err := rawDB.Exec(`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, "status-orphan", legacyPipelineStatusNSID, orphanStatusJSON, int64(6)); err != nil { t.Fatal(err) } if err := rawDB.Close(); err != nil { diff --git a/spindle/db/pipelines_test.go b/spindle/db/pipelines_test.go index e42f5a54e..d06d0ddf1 100644 --- a/spindle/db/pipelines_test.go +++ b/spindle/db/pipelines_test.go @@ -7,7 +7,6 @@ import ( "testing" "time" - "tangled.org/core/api/tangled" "tangled.org/core/notifier" "tangled.org/core/spindle/models" pipelinecodec "tangled.org/core/spindle/pipeline" @@ -187,21 +186,19 @@ func derefStr(s *string) string { return *s } -func TestToCiPipeline_TriggerKinds(t *testing.T) { +func TestFromTangledTriggerKinds(t *testing.T) { repo := "did:plc:boltless" sha := "1111111111111111111111111111111111111111" tests := []struct { - kind workflow.TriggerKind - meta func(*workflow.TriggerMetadata) - wantUnion func(*tangled.CiPipeline_Trigger) bool + kind workflow.TriggerKind + meta func(*workflow.TriggerMetadata) }{ { kind: workflow.TriggerKindPush, meta: func(m *workflow.TriggerMetadata) { m.Push = &workflow.PushTriggerData{NewSha: sha, Ref: "refs/heads/main"} }, - wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_Push != nil }, }, { kind: workflow.TriggerKindPullRequest, @@ -210,14 +207,12 @@ func TestToCiPipeline_TriggerKinds(t *testing.T) { SourceSha: sha, SourceBranch: "feature", TargetBranch: "main", } }, - wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_PullRequest != nil }, }, { kind: workflow.TriggerKindManual, meta: func(m *workflow.TriggerMetadata) { m.Manual = &workflow.ManualTriggerData{Sha: sha} }, - wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_Manual != nil }, }, { kind: workflow.TriggerKindSchedule, @@ -228,7 +223,6 @@ func TestToCiPipeline_TriggerKinds(t *testing.T) { ScheduledAt: "2026-08-10T09:00:00Z", } }, - wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_Schedule != nil }, }, } @@ -256,10 +250,6 @@ func TestToCiPipeline_TriggerKinds(t *testing.T) { if record.Commit != sha { t.Errorf("record.Commit = %q, want %q", record.Commit, sha) } - wire := pipelinecodec.ToTangled(record) - if !tt.wantUnion(wire.Trigger) { - t.Errorf("wrong trigger union populated for kind %q", tt.kind) - } if _, err := json.Marshal(record); err != nil { t.Errorf("marshal payload: %v", err) } diff --git a/spindle/describe_source_test.go b/spindle/describe_source_test.go deleted file mode 100644 index 953eb88b2..000000000 --- a/spindle/describe_source_test.go +++ /dev/null @@ -1,47 +0,0 @@ -package spindle - -import ( - "context" - "errors" - "net/http" - "path/filepath" - "strings" - "testing" - - "github.com/bluesky-social/indigo/atproto/identity" - "github.com/bluesky-social/indigo/atproto/syntax" - - "tangled.org/core/spindle/db" - "tangled.org/core/spindle/knotclient" - "tangled.org/core/spindle/xrpc" -) - -// anyone can call describe, and an unserved source must never get cloned -func TestDescribeRefusesASourceRepoThisSpindleDoesNotServe(t *testing.T) { - ctx := context.Background() - d, err := db.Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { d.Close() }) - repo := syntax.DID("did:plc:describetarget") - if err := d.AddRepo(db.Repo{Knot: "knot.test", Owner: "did:plc:owner", Rkey: "repo", RepoDid: repo}); err != nil { - t.Fatal(err) - } - source := syntax.DID("did:plc:strangersource") - knot := newFakeKnot(t) - dir := identity.NewMockDirectory() - dir.Insert(identity.Identity{ - DID: source, - Services: map[string]identity.ServiceEndpoint{"tangled_knot": {Type: "TangledKnot", URL: knot.srv.URL}}, - }) - s := &Spindle{db: d, knots: knotclient.New(dir, http.DefaultClient, true)} - - _, err = s.DescribeWorkflowDefinition(ctx, repo, strings.Repeat("a", 40), source) - if got := knot.count(); got != 0 { - t.Fatalf("the unserved source repo cost its knot %d requests, want 0", got) - } - if !errors.Is(err, xrpc.ErrSourceRepoNotServed) { - t.Fatalf("describe with an unserved source repo = %v, want ErrSourceRepoNotServed", err) - } -} diff --git a/spindle/engines/nixery/engine.go b/spindle/engines/nixery/engine.go index 97c453a7f..a12ee0215 100644 --- a/spindle/engines/nixery/engine.go +++ b/spindle/engines/nixery/engine.go @@ -334,7 +334,7 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m Hostname: "spindle", WorkingDir: workspaceDir, Labels: map[string]string{ - "sh.tangled.pipeline/workflow_id": wid.String(), + "org.tangled.pipeline/workflow_id": wid.String(), }, // TODO(winter): investigate whether environment variables passed here // get propagated to ContainerExec processes diff --git a/spindle/engines/nixery/envs.go b/spindle/engines/nixery/envs.go index 9aafc3443..d6a19aed4 100644 --- a/spindle/engines/nixery/envs.go +++ b/spindle/engines/nixery/envs.go @@ -6,8 +6,6 @@ import ( type EnvVars []string -// ConstructEnvs converts a tangled.Pipeline_Step_Environment_Elem.{Key,Value} -// representation into a docker-friendly []string{"KEY=value", ...} slice. func ConstructEnvs(envs map[string]string) EnvVars { var dockerEnvs EnvVars for k, v := range envs { diff --git a/spindle/engines/nixery/seats.go b/spindle/engines/nixery/seats.go index 7aebb591c..ad52d1bad 100644 --- a/spindle/engines/nixery/seats.go +++ b/spindle/engines/nixery/seats.go @@ -17,7 +17,9 @@ const ( seatNetworkPrefix = "spindle-wf" // seats are host-global, so a seat records which engine made it and we only // adopt our own - seatOwnerLabel = "sh.tangled.pipeline/engine" + seatOwnerLabel = "org.tangled.pipeline/engine" + // networks outlive the process, so seats made before the rename still carry this + legacySeatOwnerLabel = "sh.tangled.pipeline/engine" ) // seat networks are recycled, so an engine owns at most as many as its peak @@ -76,6 +78,13 @@ func (s *seatNetworks) name(seat int) string { return fmt.Sprintf("%s-%s-%d", seatNetworkPrefix, s.owner, seat) } +func seatOwner(labels map[string]string) string { + if owner, ok := labels[seatOwnerLabel]; ok { + return owner + } + return labels[legacySeatOwnerLabel] +} + func (e *Engine) ensureSeatNetwork(ctx context.Context, name, owner string) error { docker, err := e.ensureDocker() if err != nil { @@ -85,7 +94,7 @@ func (e *Engine) ensureSeatNetwork(ctx context.Context, name, owner string) erro existing, err := docker.NetworkInspect(ctx, name, network.InspectOptions{}) switch { case err == nil: - if held := existing.Labels[seatOwnerLabel]; held != owner { + if held := seatOwner(existing.Labels); held != owner { return fmt.Errorf("network %s is owned by %q, want %q", name, held, owner) } return nil diff --git a/spindle/engines/nixery/seats_test.go b/spindle/engines/nixery/seats_test.go index 192b44f80..a53818f1b 100644 --- a/spindle/engines/nixery/seats_test.go +++ b/spindle/engines/nixery/seats_test.go @@ -145,3 +145,20 @@ func TestNetworkNamespaceNamesThePoolAfterTheEngineProcess(t *testing.T) { }) } } + +func TestSeatOwnerRecognisesNetworksLabelledBeforeTheRename(t *testing.T) { + for name, labels := range map[string]map[string]string{ + "current": {seatOwnerLabel: "spindle"}, + "legacy": {legacySeatOwnerLabel: "spindle"}, + } { + if got := seatOwner(labels); got != "spindle" { + t.Errorf("%s: seatOwner = %q, want spindle", name, got) + } + } + if got := seatOwner(map[string]string{seatOwnerLabel: "new", legacySeatOwnerLabel: "old"}); got != "new" { + t.Errorf("both labels: seatOwner = %q, want the current one to win", got) + } + if got := seatOwner(nil); got != "" { + t.Errorf("unlabelled network: seatOwner = %q, want empty", got) + } +} diff --git a/spindle/faults_test.go b/spindle/faults_test.go index d0088ce47..5aa6ecac0 100644 --- a/spindle/faults_test.go +++ b/spindle/faults_test.go @@ -15,6 +15,7 @@ import ( "tangled.org/core/api/org_tangled" "tangled.org/core/rbac" "tangled.org/core/spindle/knotclient" + "tangled.org/core/spindle/models" ) var ( @@ -377,7 +378,7 @@ func faultPush(ref, oldSha, newSha string) *org_tangled.EventNotifyEvent_Input_E var zeroSha = strings.Repeat("0", 40) -func (e *pushEnv) notify(s *Spindle, event *org_tangled.EventNotifyEvent_Input_Event) (syntax.ATURI, error) { +func (e *pushEnv) notify(s *Spindle, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) { return s.NotifyEvent(context.Background(), e.w.knotRef(), testRepoDid, event) } @@ -398,7 +399,7 @@ func (e *pushEnv) wantPipelines(t *testing.T, want int, when string) { } } -func (e *pushEnv) mustNotify(t *testing.T, event *org_tangled.EventNotifyEvent_Input_Event) syntax.ATURI { +func (e *pushEnv) mustNotify(t *testing.T, event *org_tangled.EventNotifyEvent_Input_Event) models.PipelineId { t.Helper() uri, err := e.notify(e.n.s, event) if err != nil { @@ -408,7 +409,7 @@ func (e *pushEnv) mustNotify(t *testing.T, event *org_tangled.EventNotifyEvent_I } type notifyAnswer struct { - uri syntax.ATURI + uri models.PipelineId err error } diff --git a/spindle/fingerprint_test.go b/spindle/fingerprint_test.go deleted file mode 100644 index efa393c30..000000000 --- a/spindle/fingerprint_test.go +++ /dev/null @@ -1,34 +0,0 @@ -package spindle - -import ( - "testing" - - "github.com/stretchr/testify/assert" - - "tangled.org/core/workflow" -) - -func TestFingerprintWorkflowDefinition(t *testing.T) { - a := workflow.RawWorkflow{Name: "a.yml", Contents: []byte("engine: dummy\n")} - b := workflow.RawWorkflow{Name: "b.yml", Contents: []byte("engine: nixery\n")} - - // iteration order must not affect the fingerprint - assert.Equal(t, - fingerprintWorkflowDefinition(workflow.RawPipeline{a, b}), - fingerprintWorkflowDefinition(workflow.RawPipeline{b, a}), - ) - - // content and name changes must both change the fingerprint - base := fingerprintWorkflowDefinition(workflow.RawPipeline{a, b}) - assert.NotEqual(t, base, fingerprintWorkflowDefinition(workflow.RawPipeline{ - {Name: "a.yml", Contents: []byte("engine: nixery\n")}, b, - })) - assert.NotEqual(t, base, fingerprintWorkflowDefinition(workflow.RawPipeline{ - {Name: "c.yml", Contents: a.Contents}, b, - })) - - assert.NotEqual(t, - fingerprintWorkflowDefinition(nil), - fingerprintWorkflowDefinition(workflow.RawPipeline{a}), - ) -} diff --git a/spindle/hydrant_test.go b/spindle/hydrant_test.go index dd8b55f53..bee48c83f 100644 --- a/spindle/hydrant_test.go +++ b/spindle/hydrant_test.go @@ -22,7 +22,7 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/gorilla/websocket" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/tapc" @@ -84,7 +84,7 @@ func (f *fakeHydrant) addRecord(id int64, did, rkey string) { Type: tapc.EvtRecord, Record: &tapc.RecordEventData{ Did: syntax.DID(did), - Collection: syntax.NSID(tangled.RepoNSID), + Collection: syntax.NSID(org_tangled.RepoDeclarationNSID), Rkey: syntax.RecordKey(rkey), Action: tapc.RecordCreateAction, Record: json.RawMessage(`{}`), @@ -644,7 +644,7 @@ func TestBackfillParksARecordThatKeepsFailing(t *testing.T) { t.Fatalf("saved cursor = %d, %v, want 2", last, err) } parked := dueParked(t, b) - if len(parked) != 1 || parked[0].URI != "at://did:plc:a/sh.tangled.repo/bad" || parked[0].Err != "database is locked" { + if len(parked) != 1 || parked[0].URI != "at://did:plc:a/"+org_tangled.RepoDeclarationNSID+"/bad" || parked[0].Err != "database is locked" { t.Fatalf("parked = %+v, want the bad record with its error", parked) } } diff --git a/spindle/ingester_test.go b/spindle/ingester_test.go index 1af68df1a..b726c7645 100644 --- a/spindle/ingester_test.go +++ b/spindle/ingester_test.go @@ -9,7 +9,6 @@ import ( "github.com/bluesky-social/jetstream/pkg/models" "tangled.org/core/api/org_tangled" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/tapc" @@ -43,7 +42,7 @@ func processOK(t *testing.T, env *ensureEnv, evt tapc.Event) { func TestProcessEventIgnoresOtherCollections(t *testing.T) { env := newEnsureEnv(t) - for _, collection := range []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID, tangled.RepoPullStatusNSID} { + for _, collection := range []string{org_tangled.FeedStarNSID, org_tangled.FeedCommentNSID, org_tangled.FeedReactionNSID, org_tangled.FeedSubscriptionNSID} { evt := recordEvent("did:plc:someone", collection, "3mrhpypucbsg4", tapc.RecordCreateAction, map[string]string{"title": "x"}) if err := env.s.processEvent(context.Background(), evt); err != nil { t.Fatalf("processEvent(%s): %v", collection, err) diff --git a/spindle/notify_event.go b/spindle/notify_event.go index f1a656945..9c47c4d29 100644 --- a/spindle/notify_event.go +++ b/spindle/notify_event.go @@ -14,7 +14,6 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/hashicorp/golang-lru/v2/expirable" "tangled.org/core/api/org_tangled" - "tangled.org/core/api/tangled" "tangled.org/core/gitutil" "tangled.org/core/knotfeed" kgit "tangled.org/core/knotserver/git" @@ -25,7 +24,6 @@ import ( "tangled.org/core/spindle/xrpc" "tangled.org/core/tid" "tangled.org/core/workflow" - "tangled.org/core/xrpc/serviceauth" ) const ( @@ -241,8 +239,8 @@ func (a *notifyAdmission) miss(did syntax.DID) { a.memory().Add(did, struct{}{}) } -// an error makes the knot retry, an empty uri means nothing ran -func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (syntax.ATURI, error) { +// an error makes the knot retry, an empty id means nothing ran +func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) { l := log.FromContext(ctx).With("repo", repoDid) push := event.EventPush if push == nil { @@ -265,7 +263,7 @@ func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid sy }) select { case r := <-res: - pipeline, _ := r.Val.(syntax.ATURI) + pipeline, _ := r.Val.(models.PipelineId) return pipeline, r.Err case <-ctx.Done(): return "", ctx.Err() @@ -296,7 +294,7 @@ func (s *Spindle) notifyLimits() slotLimits { return slotLimits{pool: pool, reserve: slotReserve(pool), dedicated: dedicated, priority: s.notifyPriorityHosts} } -func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, push *org_tangled.EventPush) (pipeline syntax.ATURI, err error) { +func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, push *org_tangled.EventPush) (pipeline models.PipelineId, err error) { defer func() { if err != nil && !errors.Is(err, xrpc.ErrNotifyBusy) { log.FromContext(ctx).Error("notify event failed", "repo", repoDid, "err", err) @@ -353,7 +351,7 @@ func refusedAsStatus(err error) error { return err } -func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Repo, push *org_tangled.EventPush) (syntax.ATURI, error) { +func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Repo, push *org_tangled.EventPush) (models.PipelineId, error) { l := log.FromContext(ctx).With("repo", repo.RepoDid, "ref", push.Ref) if !isMaterializedRef(push.Ref) { @@ -388,8 +386,8 @@ func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Re fire := func() { s.wh.FirePush(context.WithoutCancel(ctx), repo, *push.Actor, push.Ref, oldSha.String(), newSha.String()) } - if uri, done, err := s.admitPush(l, k, repo, push, oldSha, newSha, fire); done || err != nil { - return uri, err + if id, done, err := s.admitPush(l, repo, push, oldSha, newSha, fire); done || err != nil { + return id, err } repoCloneUri := s.knots.CloneURL(k, repo.RepoDid) @@ -425,11 +423,11 @@ func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Re return "", nil } l.Info("pipeline triggered", "pipeline", pipelineId) - return pipelineURI(k.Host, pipelineId), nil + return pipelineId, nil } // the lock covers only db reads and the webhook since a fetch can hang on a knot -func (s *Spindle) admitPush(l *slog.Logger, k knotclient.Knot, repo *db.Repo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID, fire func()) (uri syntax.ATURI, done bool, err error) { +func (s *Spindle) admitPush(l *slog.Logger, repo *db.Repo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID, fire func()) (id models.PipelineId, done bool, err error) { defer s.lockRepo(repo.RepoDid).Unlock() if served, err := s.servedRepo(repo.RepoDid); err != nil { @@ -444,11 +442,11 @@ func (s *Spindle) admitPush(l *slog.Logger, k knotclient.Knot, repo *db.Repo, pu } else if first { fire() } - if id, found, err := s.db.PushPipeline(repo.RepoDid.String(), push.Ref, oldSha.String(), newSha.String()); err != nil { + if existing, found, err := s.db.PushPipeline(repo.RepoDid.String(), push.Ref, oldSha.String(), newSha.String()); err != nil { return "", true, fmt.Errorf("looking up the push: %w", err) } else if found { - l.Info("push already has a pipeline, skipping the replay", "pipeline", id) - return pipelineURI(k.Host, id), true, nil + l.Info("push already has a pipeline, skipping the replay", "pipeline", existing) + return existing, true, nil } // skip ci only skips the pipeline, the webhook already fired if kgit.HasSkipCIPushOption(push.PushOptions) { @@ -471,10 +469,6 @@ func pushTrigger(repo *workflow.TriggerRepo, push *org_tangled.EventPush, oldSha } } -func pipelineURI(knotHost string, id models.PipelineId) syntax.ATURI { - return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", serviceauth.DidWeb(knotHost), tangled.PipelineNSID, id)) -} - func isMaterializedRef(refname string) bool { return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/") } diff --git a/spindle/notify_event_test.go b/spindle/notify_event_test.go index 011035cfd..52d3ddb09 100644 --- a/spindle/notify_event_test.go +++ b/spindle/notify_event_test.go @@ -14,7 +14,6 @@ import ( "testing" "time" - "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/org_tangled" "tangled.org/core/gitutil" "tangled.org/core/knotfeed" @@ -77,7 +76,7 @@ func newHookEnv(t *testing.T) *hookEnv { return &hookEnv{ensureEnv: env, pushes: pushes} } -func (h *hookEnv) notifyErr(event *org_tangled.EventNotifyEvent_Input_Event) (syntax.ATURI, error) { +func (h *hookEnv) notifyErr(event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) { return h.s.NotifyEvent(context.Background(), h.k, testRepoDid, event) } @@ -166,7 +165,7 @@ func (h *hookEnv) expectNoPush(t *testing.T, why string) { } type notifyResult struct { - pipeline syntax.ATURI + pipeline models.PipelineId err error } @@ -201,11 +200,14 @@ func TestNotifyEventReplayOfAPipelinePushAnswersWithItAndRunsNothing(t *testing. h := newHookEnv(t) h.seedPushPipeline(t, "first", "refs/heads/main", testHexOld, testHexNew) - // skip-ci, so only the replay lookup can make the uri + // skip-ci makes no new pipeline, the id can only come from the replay lookup for range 2 { got := h.notify(t, pushEvent("refs/heads/main", "skip-ci")) - if want := string(pipelineURI(h.k.Host, "first")); got != want { - t.Fatalf("replay answered %q, want the first delivery's pipeline %q", got, want) + if got != "first" { + t.Fatalf("replay answered %q, want the first delivery's pipeline id", got) + } + if _, err := h.s.db.GetPipeline(context.Background(), models.PipelineId(got)); err != nil { + t.Fatalf("answer %q is not a stored pipeline id: %v", got, err) } } h.expectNoPush(t, "a replayed push fired a webhook") diff --git a/spindle/pipeline/codec.go b/spindle/pipeline/codec.go index fdbebd3b3..91e6a835f 100644 --- a/spindle/pipeline/codec.go +++ b/spindle/pipeline/codec.go @@ -4,7 +4,6 @@ import ( "fmt" "time" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/models" "tangled.org/core/workflow" ) @@ -156,61 +155,3 @@ func inputPairs(inputs []*workflow.TriggerPair) []*models.PipelineInputPair { } return out } - -func ToTangled(record *models.PipelineRecord) *tangled.CiPipeline { - if record == nil { - return nil - } - trigger := &tangled.CiPipeline_Trigger{} - switch { - case record.Trigger.Push != nil: - push := record.Trigger.Push - trigger.CiTrigger_Push = &tangled.CiTrigger_Push{ - Ref: push.Ref, NewSha: push.NewCommit, OldSha: push.OldCommit, - } - case record.Trigger.PullRequest != nil: - pullRequest := record.Trigger.PullRequest - trigger.CiTrigger_PullRequest = &tangled.CiTrigger_PullRequest{ - Action: pullRequest.Action, SourceRepo: pullRequest.SourceRepo, - SourceBranch: pullRequest.SourceBranch, SourceSha: pullRequest.SourceCommit, - TargetBranch: pullRequest.TargetBranch, Pull: pullRequest.Pull, - } - case record.Trigger.Manual != nil: - manual := record.Trigger.Manual - trigger.CiTrigger_Manual = &tangled.CiTrigger_Manual{ - Ref: manual.Ref, Sha: record.Commit, SourceRepo: manual.SourceRepo, - Inputs: tangledInputPairs(manual.Inputs), - } - case record.Trigger.Schedule != nil: - schedule := record.Trigger.Schedule - trigger.CiTrigger_Schedule = &tangled.CiTrigger_Schedule{ - Ref: schedule.Ref, ScheduledAt: schedule.ScheduledAt, Sha: schedule.Sha, - } - } - - createdAt := record.CreatedAt - workflows := make([]*tangled.CiPipeline_Workflow, 0, len(record.Workflows)) - for _, workflow := range record.Workflows { - if workflow == nil { - continue - } - workflows = append(workflows, &tangled.CiPipeline_Workflow{ - Id: workflow.ID, Name: workflow.Name, Status: workflow.Status, - Error: workflow.Error, StartedAt: workflow.StartedAt, FinishedAt: workflow.FinishedAt, - }) - } - return &tangled.CiPipeline{ - Id: string(record.ID), Repo: record.RepoDID, SourceRepo: record.SourceRepo, - Commit: record.Commit, CreatedAt: &createdAt, Trigger: trigger, Workflows: workflows, - } -} - -func tangledInputPairs(inputs []*models.PipelineInputPair) []*tangled.CiTrigger_Pair { - out := make([]*tangled.CiTrigger_Pair, 0, len(inputs)) - for _, input := range inputs { - if input != nil { - out = append(out, &tangled.CiTrigger_Pair{Key: input.Key, Value: input.Value}) - } - } - return out -} diff --git a/spindle/server.go b/spindle/server.go index 26086cf63..5c9c829cd 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -2,10 +2,7 @@ package spindle import ( "context" - "crypto/sha256" - "database/sql" _ "embed" - "encoding/binary" "errors" "fmt" "io" @@ -29,7 +26,6 @@ import ( "go.opentelemetry.io/otel/codes" "golang.org/x/sync/singleflight" "tangled.org/core/api/org_tangled" - "tangled.org/core/api/tangled" "tangled.org/core/gitutil" "tangled.org/core/idresolver" "tangled.org/core/jetstream" @@ -1026,7 +1022,7 @@ func filterWorkflows(workflows []*workflow.WorkflowSpec, only []string) []*workf // TriggerManual dispatches a pipeline at sha, authorized against and recorded // under repoDid. sourceRepo, pull, and inputs are optional trigger payload. -func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull xrpc.PullContext, inputs []*workflow.TriggerPair) (syntax.ATURI, error) { +func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull xrpc.PullContext, inputs []*workflow.TriggerPair) (models.PipelineId, error) { repo, err := s.db.GetRepoByDid(repoDid) if err != nil { return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) @@ -1114,7 +1110,7 @@ func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, if pipelineId == "" { return "", xrpc.ErrNoMatchingWorkflows } - return pipelineURI(repo.Knot, pipelineId), nil + return pipelineId, nil } // sourceInfo is nil when the checkout comes from the target repo. @@ -1164,61 +1160,6 @@ func (s *Spindle) loadWorkflowDefinitions(ctx context.Context, repoDid syntax.DI return out, nil } -func (s *Spindle) DescribeWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - // anyone can ask, the source has to be a repo we already serve - if sourceRepo != "" && sourceRepo != repoDid { - if _, err := s.db.GetRepoByDid(sourceRepo); errors.Is(err, sql.ErrNoRows) { - return nil, xrpc.ErrSourceRepoNotServed - } else if err != nil { - return nil, err - } - } - return s.workflowDefs.describe(repoDid, sourceRepo, sha, func() (*tangled.CiDescribeWorkflowDefinition_Output, error) { - return s.loadWorkflowDefinition(ctx, repoDid, sha, sourceRepo) - }) -} - -func (s *Spindle) loadWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - readFrom := repoDid - if sourceRepo != "" { - readFrom = sourceRepo - } - rawPipeline, _, err := s.readWorkflowDir(ctx, readFrom, sha) - if err != nil { - return nil, fmt.Errorf("loading pipeline: %w", err) - } - - hash := fingerprintWorkflowDefinition(rawPipeline) - workflows := make([]string, 0, len(rawPipeline)) - for _, w := range rawPipeline { - workflows = append(workflows, w.Name) - } - - return &tangled.CiDescribeWorkflowDefinition_Output{ - Derived: true, - Hash: &hash, - Workflows: workflows, - }, nil -} - -func fingerprintWorkflowDefinition(rawPipeline workflow.RawPipeline) string { - sorted := make([]workflow.RawWorkflow, len(rawPipeline)) - copy(sorted, rawPipeline) - sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) - - h := sha256.New() - var lenBuf [8]byte - for _, w := range sorted { - binary.LittleEndian.PutUint64(lenBuf[:], uint64(len(w.Contents))) - h.Write([]byte(w.Name)) - // terminate name to avoid ["fo", "o"] == ["f", "oo"] - h.Write([]byte{0}) - h.Write(lenBuf[:]) - h.Write(w.Contents) - } - return fmt.Sprintf("sha256:%x", h.Sum(nil)) -} - // the host comes from a did doc the repo owner controls and git dials it unguarded, so // every knot fetch comes through here and git's stderr only goes to the log func (s *Spindle) syncRepo(ctx context.Context, cloneUri, repoPath, rev string) error { diff --git a/spindle/source_repo_test.go b/spindle/source_repo_test.go index f9e8f42d6..0ce43c508 100644 --- a/spindle/source_repo_test.go +++ b/spindle/source_repo_test.go @@ -5,7 +5,7 @@ import ( "testing" ) -// knot2 says noRoute for sh.tangled routes, a fork has to be read the way ensureRepo reads a repo +// knot2 only serves records, so a fork has to be read the way ensureRepo reads a repo func TestSourceRepoOfAKnotThatOnlyServesRecordsResolves(t *testing.T) { env := newEnsureEnv(t) env.serve("fork") diff --git a/spindle/wipe_test.go b/spindle/wipe_test.go index bdb905d80..b82dc296c 100644 --- a/spindle/wipe_test.go +++ b/spindle/wipe_test.go @@ -5,13 +5,13 @@ import ( "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" jmodels "github.com/bluesky-social/jetstream/pkg/models" - "tangled.org/core/api/tangled" "tangled.org/core/jetstream" "tangled.org/core/log" "tangled.org/core/notifier" @@ -118,10 +118,10 @@ func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) { 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 := &tangled.CiPipeline{ + pipeline := &org_tangled.CiPipeline{ Id: "p" + sfx, Repo: repoDid.String(), - Workflows: []*tangled.CiPipeline_Workflow{{Id: "w" + sfx, Name: "w" + sfx, Status: "pending"}}, + Workflows: []*org_tangled.CiPipeline_Workflow{{Id: "w" + sfx, Name: ptr("w" + sfx), Status: "pending"}}, } payload, err := json.Marshal(pipeline) if err != nil { diff --git a/spindle/workflow_defs.go b/spindle/workflow_defs.go index 69a4e3b67..f8c47b5ab 100644 --- a/spindle/workflow_defs.go +++ b/spindle/workflow_defs.go @@ -8,14 +8,11 @@ import ( "github.com/hashicorp/golang-lru/v2/expirable" "golang.org/x/time/rate" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/models" "tangled.org/core/spindle/xrpc" ) const ( - // only refs get the short ttl because commits don't move but refs do - workflowDefCommitTTL = 10 * time.Minute workflowDefRefTTL = 10 * time.Second workflowDefCacheSize = 512 workflowDefLoadsPerSecond = 5 @@ -24,16 +21,14 @@ const ( // a miss costs the knot a tree read and a blob per workflow type workflowDefCache struct { - once sync.Once - mu sync.Mutex - byRef *expirable.LRU[string, []*models.WorkflowDefinition] - byCommit *expirable.LRU[string, *tangled.CiDescribeWorkflowDefinition_Output] - limits *expirable.LRU[syntax.DID, *rate.Limiter] + once sync.Once + mu sync.Mutex + byRef *expirable.LRU[string, []*models.WorkflowDefinition] + limits *expirable.LRU[syntax.DID, *rate.Limiter] } func (c *workflowDefCache) init() { c.byRef = expirable.NewLRU[string, []*models.WorkflowDefinition](workflowDefCacheSize, nil, workflowDefRefTTL) - c.byCommit = expirable.NewLRU[string, *tangled.CiDescribeWorkflowDefinition_Output](workflowDefCacheSize, nil, workflowDefCommitTTL) c.limits = expirable.NewLRU[syntax.DID, *rate.Limiter](workflowDefCacheSize, nil, time.Minute) } @@ -58,11 +53,6 @@ func (c *workflowDefCache) list(repo syntax.DID, ref string, load func() ([]*mod return cached(c.byRef, repo.String()+"\x00"+ref, load) } -func (c *workflowDefCache) describe(repo, source syntax.DID, sha string, load func() (*tangled.CiDescribeWorkflowDefinition_Output, error)) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - c.once.Do(c.init) - return cached(c.byCommit, repo.String()+"\x00"+source.String()+"\x00"+sha, load) -} - func cached[V any](lru *expirable.LRU[string, V], key string, load func() (V, error)) (V, error) { if v, ok := lru.Get(key); ok { return v, nil diff --git a/spindle/workflow_read_test.go b/spindle/workflow_read_test.go index 21a58e85f..6223eb3d1 100644 --- a/spindle/workflow_read_test.go +++ b/spindle/workflow_read_test.go @@ -336,23 +336,6 @@ func TestDefinitionBranchRefReadsLocallyAfterResolving(t *testing.T) { } } -func TestDescribeReadsTheSourceRepoNotTheTarget(t *testing.T) { - knot := newDefsKnot(t) - target, source := syntax.DID("did:plc:defstarget"), syntax.DID("did:plc:defssource") - s := newDefsSpindle(t, knot, target, source) - commit := strings.Repeat("a", 40) - knot.setTree(source, commit, regular("fork.yaml")) - t.Setenv("PATH", "") - - out, err := s.DescribeWorkflowDefinition(context.Background(), target, commit, source) - if err != nil { - t.Fatal(err) - } - if len(out.Workflows) != 1 || out.Workflows[0] != "fork.yaml" { - t.Fatalf("workflows = %v, want the fork's", out.Workflows) - } -} - func TestDefinitionTagRefsResolveToTheirCommitWithoutAClone(t *testing.T) { for _, ref := range []string{"v1.0", "refs/tags/v1.0"} { t.Run(ref, func(t *testing.T) { diff --git a/spindle/xrpc/ci_pipeline_describe_workflow_definition.go b/spindle/xrpc/ci_pipeline_describe_workflow_definition.go deleted file mode 100644 index 966afc74d..000000000 --- a/spindle/xrpc/ci_pipeline_describe_workflow_definition.go +++ /dev/null @@ -1,50 +0,0 @@ -package xrpc - -import ( - "errors" - "net/http" - - xrpcerr "tangled.org/core/xrpc/errors" -) - -func (x *Xrpc) DescribeWorkflowDefinition(w http.ResponseWriter, r *http.Request) { - l := x.Logger - fail := func(e xrpcerr.XrpcError) { - l.ErrorContext(r.Context(), "failed", "kind", e.Tag, "error", e.Message) - writeError(w, e, http.StatusBadRequest) - } - - q := r.URL.Query() - repoDid, xerr, ok := x.resolveKnownRepoDid(q.Get("repo")) - if !ok { - fail(xerr) - return - } - - sha := q.Get("sha") - if err := requireSha(sha); err != nil { - fail(invalidRequest(err)) - return - } - - sourceRepoParam := q.Get("sourceRepo") - sourceRepo, err := parseOptionalDID("sourceRepo", &sourceRepoParam) - if err != nil { - fail(invalidRequest(err)) - return - } - - out, err := x.Trigger.DescribeWorkflowDefinition(r.Context(), repoDid, sha, sourceRepo) - if errors.Is(err, ErrDefinitionsBusy) { - writeError(w, xrpcerr.GenericError(err), http.StatusTooManyRequests) - return - } - if err != nil { - fail(xrpcerr.GenericError(err)) - return - } - - if err := writeJson(w, http.StatusOK, out); err != nil { - l.ErrorContext(r.Context(), "failed to write response", "err", err) - } -} diff --git a/spindle/xrpc/ci_pipeline_subscribe_logs.go b/spindle/xrpc/ci_pipeline_subscribe_logs.go index e584cdef5..8133ba476 100644 --- a/spindle/xrpc/ci_pipeline_subscribe_logs.go +++ b/spindle/xrpc/ci_pipeline_subscribe_logs.go @@ -12,27 +12,11 @@ import ( "github.com/bluesky-social/indigo/atproto/atclient" "github.com/gorilla/websocket" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" "tangled.org/core/spindle/logview" "tangled.org/core/spindle/models" ) -func (x *Xrpc) HandleCiSubscribePipelineLogs(w http.ResponseWriter, r *http.Request) { - var ( - pipelineQuery = r.URL.Query().Get("pipeline") - workflows = r.URL.Query()["workflows"] - ) - - // pipeline ids are opaque, any format. WorkflowId.String() normalizes before - // it reaches a filesystem path, so only emptiness is rejected here. - if pipelineQuery == "" { - writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "pipeline parameter is required"}) - return - } - - x.handleSubscribeLogs(w, r, models.PipelineId(pipelineQuery), workflows) -} - var logWriteTimeout = time.Minute // clients only send pings and closes, and a big frame from one means something is off @@ -165,7 +149,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel } }() - eventsChan := make(chan tangled.CiSubscribePipelineLogs_Event, 128) + eventsChan := make(chan org_tangled.CiSubscribePipelineLogs_Event, 128) wg := sync.WaitGroup{} // 4. start a tail reader goroutine for each workflow @@ -237,7 +221,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel logLine = models.NewDataLogLine(0, line.Text, "stdout") } - var ev tangled.CiSubscribePipelineLogs_Event + var ev org_tangled.CiSubscribePipelineLogs_Event timeStr := logLine.Time.Format(time.RFC3339) if logLine.Time.IsZero() { timeStr = time.Now().Format(time.RFC3339) @@ -248,7 +232,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel if logLine.StepKind == models.StepKindSystem { stepKindStr = "system" } - ev = tangled.CiSubscribePipelineLogs_Event{Control: &tangled.CiSubscribePipelineLogs_Control{ + ev = org_tangled.CiSubscribePipelineLogs_Event{Control: &org_tangled.CiSubscribePipelineLogs_Control{ Time: timeStr, Workflow: wfName, Step: int64(logLine.StepId), @@ -263,7 +247,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel if streamType != "stdout" && streamType != "stderr" { streamType = "stdout" } - ev = tangled.CiSubscribePipelineLogs_Event{Data: &tangled.CiSubscribePipelineLogs_Data{ + ev = org_tangled.CiSubscribePipelineLogs_Event{Data: &org_tangled.CiSubscribePipelineLogs_Data{ Time: timeStr, Workflow: wfName, Step: int64(logLine.StepId), diff --git a/spindle/xrpc/ci_pipeline_trigger_pipeline.go b/spindle/xrpc/ci_pipeline_trigger_pipeline.go deleted file mode 100644 index ff982ed75..000000000 --- a/spindle/xrpc/ci_pipeline_trigger_pipeline.go +++ /dev/null @@ -1,201 +0,0 @@ -package xrpc - -import ( - "database/sql" - "encoding/json" - "errors" - "fmt" - "net/http" - - "github.com/bluesky-social/indigo/atproto/syntax" - - "tangled.org/core/api/tangled" - "tangled.org/core/workflow" - xrpcerr "tangled.org/core/xrpc/errors" - "tangled.org/core/xrpc/serviceauth" -) - -func (x *Xrpc) TriggerPipeline(w http.ResponseWriter, r *http.Request) { - l := x.Logger - fail := func(e xrpcerr.XrpcError) { - l.ErrorContext(r.Context(), "failed", "kind", e.Tag, "error", e.Message) - writeRepoRequestError(w, e) - } - l.DebugContext(r.Context(), "trigger pipeline") - - var input tangled.CiTriggerPipeline_Input - if err := json.NewDecoder(r.Body).Decode(&input); err != nil { - fail(xrpcerr.GenericError(err)) - return - } - - repoDid, xerr, ok := x.resolveOwnedRepo(r, input.Repo) - if !ok { - fail(xerr) - return - } - - var sha string - ref := "" - var sourceRepo syntax.DID - var pull PullContext - var inputs []*workflow.TriggerPair - - switch { - case input.Trigger == nil: - fail(xrpcerr.GenericError(fmt.Errorf("trigger is required"))) - return - - case input.Trigger.CiTrigger_Manual != nil: - manual := input.Trigger.CiTrigger_Manual - sha = manual.Sha - if manual.Ref != nil { - ref = *manual.Ref - } - parsed, err := parseOptionalDID("sourceRepo", manual.SourceRepo) - if err != nil { - fail(xrpcerr.GenericError(err)) - return - } - sourceRepo = parsed - inputs = ciTriggerPairsToPipelinePairs(manual.Inputs) - - case input.Trigger.CiTrigger_PullRequest != nil: - pr := input.Trigger.CiTrigger_PullRequest - sha = pr.SourceSha - parsed, err := parseOptionalDID("sourceRepo", pr.SourceRepo) - if err != nil { - fail(xrpcerr.GenericError(err)) - return - } - sourceRepo = parsed - - if pr.TargetBranch == "" { - fail(xrpcerr.GenericError(fmt.Errorf("pull request trigger targetBranch is required"))) - return - } - - var pullAt syntax.ATURI - if pr.Pull != nil { - var err error - pullAt, err = syntax.ParseATURI(*pr.Pull) - if err != nil { - fail(xrpcerr.InvalidRepoError(*pr.Pull)) - return - } - } - sourceBranch := "" - if pr.SourceBranch != nil { - sourceBranch = *pr.SourceBranch - } - pull = PullContext{ - IsPullRequest: true, - Pull: pullAt, - SourceBranch: sourceBranch, - TargetBranch: pr.TargetBranch, - } - - default: - fail(xrpcerr.GenericError(fmt.Errorf("unsupported trigger variant"))) - return - } - - if err := requireSha(sha); err != nil { - fail(invalidRequest(err)) - return - } - - attribution, ok := serviceauth.Attribution(r, repoDid) - if !ok { - fail(xrpcerr.MissingActorDidError) - return - } - - pipelineAt, err := x.Trigger.TriggerManual(r.Context(), attribution, repoDid, sha, ref, input.Workflows, sourceRepo, pull, inputs) - if errors.Is(err, ErrOwnerPaused) { - writeOwnerPaused(w) - return - } - if errors.Is(err, ErrNoMatchingWorkflows) { - fail(xrpcerr.GenericError(err)) - return - } - if err != nil { - fail(xrpcerr.GenericError(fmt.Errorf("failed to trigger pipeline: %w", err))) - return - } - - if err := writeJson(w, http.StatusOK, tangled.CiTriggerPipeline_Output{ - Pipeline: pipelineAt.String(), - }); err != nil { - l.ErrorContext(r.Context(), "failed to write response", "err", err) - } -} - -func parseOptionalDID(field string, value *string) (syntax.DID, error) { - if value == nil || *value == "" { - return "", nil - } - did, err := syntax.ParseDID(*value) - if err != nil { - return "", fmt.Errorf("invalid %s DID %q: %w", field, *value, err) - } - return did, nil -} - -func ciTriggerPairsToPipelinePairs(inputs []*tangled.CiTrigger_Pair) []*workflow.TriggerPair { - if len(inputs) == 0 { - return nil - } - pairs := make([]*workflow.TriggerPair, 0, len(inputs)) - for _, input := range inputs { - if input == nil { - continue - } - pairs = append(pairs, &workflow.TriggerPair{ - Key: input.Key, - Value: input.Value, - }) - } - return pairs -} - -func (x *Xrpc) resolveOwnedRepo(r *http.Request, repoDidStr string) (syntax.DID, xrpcerr.XrpcError, bool) { - repoDid, xerr, ok := repoDidOf(repoDidStr) - if !ok { - return "", xerr, false - } - - if xerr, ok := x.authorizeRepoPrincipal(r, repoDid, x.pushCheck); !ok { - return "", xerr, false - } - - return repoDid, xrpcerr.XrpcError{}, true -} - -func writeKnownRepoError(w http.ResponseWriter, xerr xrpcerr.XrpcError) { - status := http.StatusInternalServerError - switch xerr.Tag { - case "InvalidRequest": - status = http.StatusBadRequest - case "RepoNotFound": - status = http.StatusNotFound - } - writeError(w, xerr, status) -} - -func (x *Xrpc) resolveKnownRepoDid(repoDidStr string) (syntax.DID, xrpcerr.XrpcError, bool) { - repoDid, xerr, ok := repoDidOf(repoDidStr) - if !ok { - return "", xerr, false - } - - if _, err := x.Db.GetRepoByDid(repoDid); err != nil { - if errors.Is(err, sql.ErrNoRows) { - return "", xrpcerr.RepoNotFoundError, false - } - return "", xrpcerr.GenericError(err), false - } - - return repoDid, xrpcerr.XrpcError{}, true -} diff --git a/spindle/xrpc/ci_query_pipelines.go b/spindle/xrpc/ci_query_pipelines.go deleted file mode 100644 index 691086a1a..000000000 --- a/spindle/xrpc/ci_query_pipelines.go +++ /dev/null @@ -1,87 +0,0 @@ -package xrpc - -import ( - "fmt" - "net/http" - "strconv" - - "tangled.org/core/api/tangled" - "tangled.org/core/spindle/models" - pipelinecodec "tangled.org/core/spindle/pipeline" - xrpcerr "tangled.org/core/xrpc/errors" -) - -// a bigger limit gets capped here, and the cursor fetches the rest -const maxPipelinePage = 250 - -func (x *Xrpc) HandleCiQueryPipelines(w http.ResponseWriter, r *http.Request) { - l := x.Logger - fail := func(e xrpcerr.XrpcError, status int) { - l.ErrorContext(r.Context(), "failed", "kind", e.Tag, "error", e.Message) - writeError(w, e, status) - } - - repo := r.URL.Query().Get("repo") - if repo == "" { - fail(xrpcerr.GenericError(fmt.Errorf("missing repo parameter")), http.StatusBadRequest) - return - } - - commits := r.URL.Query()["commits"] - cursor := r.URL.Query().Get("cursor") - kinds := r.URL.Query()["kinds"] - limitStr := r.URL.Query().Get("limit") - - limit := 30 - if limitStr != "" { - if val, err := strconv.Atoi(limitStr); err == nil && val > 0 { - limit = min(val, maxPipelinePage) - } - } - - records, nextCursor, total, err := x.Db.QueryPipelines(r.Context(), repo, commits, cursor, kinds, limit) - if err != nil { - fail(xrpcerr.GenericError(err), http.StatusInternalServerError) - return - } - pipelines := make([]*tangled.CiPipeline, 0, len(records)) - for _, record := range records { - pipelines = append(pipelines, pipelinecodec.ToTangled(record)) - } - - output := tangled.CiQueryPipelines_Output{ - Pipelines: pipelines, - Total: total, - } - if nextCursor != "" { - output.Cursor = &nextCursor - } - - if err := writeJson(w, http.StatusOK, output); err != nil { - fail(xrpcerr.GenericError(err), http.StatusInternalServerError) - } -} - -func (x *Xrpc) HandleCiGetPipeline(w http.ResponseWriter, r *http.Request) { - l := x.Logger - fail := func(e xrpcerr.XrpcError, status int) { - l.ErrorContext(r.Context(), "failed", "kind", e.Tag, "error", e.Message) - writeError(w, e, status) - } - - pipeline := r.URL.Query().Get("pipeline") - if pipeline == "" { - fail(xrpcerr.GenericError(fmt.Errorf("missing pipeline parameter")), http.StatusBadRequest) - return - } - - record, err := x.Db.GetPipeline(r.Context(), models.PipelineId(pipeline)) - if err != nil { - fail(xrpcerr.GenericError(err), http.StatusInternalServerError) - return - } - - if err := writeJson(w, http.StatusOK, pipelinecodec.ToTangled(record)); err != nil { - fail(xrpcerr.GenericError(err), http.StatusInternalServerError) - } -} diff --git a/spindle/xrpc/ci_query_pipelines_test.go b/spindle/xrpc/ci_query_pipelines_test.go deleted file mode 100644 index 45fade94c..000000000 --- a/spindle/xrpc/ci_query_pipelines_test.go +++ /dev/null @@ -1,43 +0,0 @@ -package xrpc - -import ( - "encoding/json" - "fmt" - "net/http" - "net/http/httptest" - "testing" - - "tangled.org/core/api/tangled" - "tangled.org/core/spindle/models" -) - -func TestLegacyQueryPipelinesCapsThePageSize(t *testing.T) { - x, _, repo := newOrgPipelineXrpc(t) - const total = 300 - for i := range total { - sha := fmt.Sprintf("%040d", i) - err := x.Db.CreatePipeline(&models.PipelineRecord{ - ID: models.PipelineId(fmt.Sprintf("legacy%04d", i)), - RepoDID: repo, - Commit: sha, - CreatedAt: "2026-01-01T00:00:00Z", - Trigger: models.PipelineTrigger{Kind: "push", Push: &models.PipelinePushTrigger{Ref: "refs/heads/main", NewCommit: sha}}, - }) - if err != nil { - t.Fatal(err) - } - } - - w := httptest.NewRecorder() - x.HandleCiQueryPipelines(w, httptest.NewRequest(http.MethodGet, "/?repo="+repo+"&limit=100000", nil)) - var out tangled.CiQueryPipelines_Output - if w.Code != http.StatusOK || json.Unmarshal(w.Body.Bytes(), &out) != nil { - t.Fatalf("response = %d: %.200s", w.Code, w.Body.String()) - } - if len(out.Pipelines) > 250 { - t.Fatalf("limit=100000 returned %d pipelines, want at most 250", len(out.Pipelines)) - } - if out.Cursor == nil { - t.Fatalf("a capped page of %d out of %d pipelines came with no cursor", len(out.Pipelines), total+1) - } -} diff --git a/spindle/xrpc/ci_subscribe_logs_workflows_test.go b/spindle/xrpc/ci_subscribe_logs_workflows_test.go index 8ecaf99d8..830b67b3d 100644 --- a/spindle/xrpc/ci_subscribe_logs_workflows_test.go +++ b/spindle/xrpc/ci_subscribe_logs_workflows_test.go @@ -9,11 +9,11 @@ import ( "time" ) -func TestLegacyLogRouteRejectsAWorkflowThePipelineLacks(t *testing.T) { +func TestLogRouteRejectsAWorkflowThePipelineLacks(t *testing.T) { x, id, _ := newOrgPipelineXrpc(t) w := httptest.NewRecorder() r := httptest.NewRequest(http.MethodGet, "/?pipeline="+id.String()+"&workflows=ghost.yml", nil) - x.HandleCiSubscribePipelineLogs(w, r) + x.handleOrgTangledCiSubscribePipelineLogs(w, r) if w.Code != http.StatusNotFound { t.Fatalf("status = %d (%s), want 404", w.Code, strings.TrimSpace(w.Body.String())) } diff --git a/spindle/xrpc/cors_test.go b/spindle/xrpc/cors_test.go index 75a888751..6b61b35e1 100644 --- a/spindle/xrpc/cors_test.go +++ b/spindle/xrpc/cors_test.go @@ -6,7 +6,7 @@ import ( "net/http/httptest" "testing" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" "tangled.org/core/spindle/config" "tangled.org/core/xrpc/serviceauth" ) @@ -32,7 +32,7 @@ func preflight(router http.Handler, origin, nsid, method string) *httptest.Respo } func TestCorsOffByDefault(t *testing.T) { - rec := preflight(corsRouter(), "https://tangled.org", tangled.RepoListSecretsNSID, http.MethodGet) + rec := preflight(corsRouter(), "https://tangled.org", org_tangled.SecretListSecretsNSID, http.MethodGet) if got := rec.Header().Get("Access-Control-Allow-Origin"); got != "" { t.Errorf("allow-origin %q with no origins configured", got) } @@ -42,8 +42,8 @@ func TestCorsAnyOrigin(t *testing.T) { router := corsRouter("*") for _, tc := range []struct{ nsid, method string }{ - {tangled.RepoListSecretsNSID, http.MethodGet}, - {tangled.RepoAddSecretNSID, http.MethodPost}, + {org_tangled.SecretListSecretsNSID, http.MethodGet}, + {org_tangled.SecretAddSecretNSID, http.MethodPost}, } { rec := preflight(router, "https://tangled.org", tc.nsid, tc.method) if rec.Code >= 300 { @@ -61,19 +61,19 @@ func TestCorsAnyOrigin(t *testing.T) { func TestCorsListedOrigins(t *testing.T) { router := corsRouter("https://tangled.org") - rec := preflight(router, "https://tangled.org", tangled.RepoListSecretsNSID, http.MethodGet) + rec := preflight(router, "https://tangled.org", org_tangled.SecretListSecretsNSID, http.MethodGet) if got := rec.Header().Get("Access-Control-Allow-Origin"); got != "https://tangled.org" { t.Errorf("listed origin: allow-origin %q", got) } - rec = preflight(router, "https://evil.example", tangled.RepoListSecretsNSID, http.MethodGet) + rec = preflight(router, "https://evil.example", org_tangled.SecretListSecretsNSID, http.MethodGet) if got := rec.Header().Get("Access-Control-Allow-Origin"); got != "" { t.Errorf("unlisted origin: allow-origin %q", got) } } func TestCorsHeadersOnRejectedRequests(t *testing.T) { - req := httptest.NewRequest(http.MethodGet, "/"+tangled.RepoListSecretsNSID, nil) + req := httptest.NewRequest(http.MethodGet, "/"+org_tangled.SecretListSecretsNSID, nil) req.Header.Set("Origin", "https://tangled.org") req.Header.Set("Authorization", "Bearer not.a.jwt") rec := httptest.NewRecorder() diff --git a/spindle/xrpc/definitions_busy_test.go b/spindle/xrpc/definitions_busy_test.go index 91017d889..9a4e764f8 100644 --- a/spindle/xrpc/definitions_busy_test.go +++ b/spindle/xrpc/definitions_busy_test.go @@ -9,7 +9,6 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/models" ) @@ -19,18 +18,13 @@ func (*busyTrigger) ListWorkflowDefinitions(context.Context, syntax.DID, string) return nil, ErrDefinitionsBusy } -func (*busyTrigger) DescribeWorkflowDefinition(context.Context, syntax.DID, string, syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - return nil, ErrDefinitionsBusy -} - func TestBusyDefinitionReadsAnswer429(t *testing.T) { x, _, repo := newOrgPipelineXrpc(t) x.Trigger = &busyTrigger{} for name, call := range map[string]func(http.ResponseWriter, *http.Request){ - "list": x.handleOrgTangledCiListWorkflowDefinitions, - "get": x.handleOrgTangledCiGetWorkflowDefinition, - "describe": x.DescribeWorkflowDefinition, + "list": x.handleOrgTangledCiListWorkflowDefinitions, + "get": x.handleOrgTangledCiGetWorkflowDefinition, } { w := httptest.NewRecorder() call(w, httptest.NewRequest(http.MethodGet, "/?repo="+repo+"&workflow=ci.yml&sha="+strings.Repeat("a", 40), nil)) diff --git a/spindle/xrpc/legacy_routes_test.go b/spindle/xrpc/legacy_routes_test.go new file mode 100644 index 000000000..e5a1d8e64 --- /dev/null +++ b/spindle/xrpc/legacy_routes_test.go @@ -0,0 +1,34 @@ +package xrpc + +import ( + "net/http" + "net/http/httptest" + "testing" + + "tangled.org/core/api/org_tangled" +) + +func TestOnlyTheOrgPipelineRoutesAreServed(t *testing.T) { + router := corsRouter() + legacy := "sh" + ".tangled." + for _, name := range []string{ + "ci.triggerPipeline", "ci.cancelPipeline", "ci.describeWorkflowDefinition", + "ci.subscribePipelineLogs", "ci.queryPipelines", "ci.getPipeline", "owner", + } { + for _, method := range []string{http.MethodGet, http.MethodPost} { + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(method, "/"+legacy+name, nil)) + if rec.Code != http.StatusNotFound { + t.Errorf("%s %s answered %d, want 404", method, legacy+name, rec.Code) + } + } + } + + for _, nsid := range []string{org_tangled.CiTriggerPipelineNSID, org_tangled.CiCancelPipelineNSID} { + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/"+nsid, nil)) + if rec.Code != http.StatusUnauthorized && rec.Code != http.StatusForbidden { + t.Errorf("unsigned POST %s answered %d, want the auth middleware to refuse it", nsid, rec.Code) + } + } +} diff --git a/spindle/xrpc/pipeline_cancel_pipeline.go b/spindle/xrpc/org_tangled_ci_cancel_pipeline.go similarity index 94% rename from spindle/xrpc/pipeline_cancel_pipeline.go rename to spindle/xrpc/org_tangled_ci_cancel_pipeline.go index 8c5f9472b..6b720d5ac 100644 --- a/spindle/xrpc/pipeline_cancel_pipeline.go +++ b/spindle/xrpc/org_tangled_ci_cancel_pipeline.go @@ -6,13 +6,13 @@ import ( "net/http" "github.com/bluesky-social/indigo/atproto/syntax" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" xrpcerr "tangled.org/core/xrpc/errors" ) -func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) { +func (x *Xrpc) handleOrgTangledCiCancelPipeline(w http.ResponseWriter, r *http.Request) { l := x.Logger fail := func(e xrpcerr.XrpcError) { l.ErrorContext(r.Context(), "failed", "kind", e.Tag, "error", e.Message) @@ -26,7 +26,7 @@ func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) { return } - var input tangled.CiCancelPipeline_Input + var input org_tangled.CiCancelPipeline_Input if err := json.NewDecoder(r.Body).Decode(&input); err != nil { fail(xrpcerr.GenericError(err)) return diff --git a/spindle/xrpc/org_tangled_ci_trigger_pipeline.go b/spindle/xrpc/org_tangled_ci_trigger_pipeline.go new file mode 100644 index 000000000..de764c83c --- /dev/null +++ b/spindle/xrpc/org_tangled_ci_trigger_pipeline.go @@ -0,0 +1,146 @@ +package xrpc + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "tangled.org/core/api/org_tangled" + "tangled.org/core/workflow" + xrpcerr "tangled.org/core/xrpc/errors" + "tangled.org/core/xrpc/serviceauth" +) + +func (x *Xrpc) handleOrgTangledCiTriggerPipeline(w http.ResponseWriter, r *http.Request) { + l := x.Logger + fail := func(e xrpcerr.XrpcError) { + l.ErrorContext(r.Context(), "failed", "kind", e.Tag, "error", e.Message) + writeRepoRequestError(w, e) + } + l.DebugContext(r.Context(), "trigger pipeline") + + var input org_tangled.CiTriggerPipeline_Input + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { + fail(xrpcerr.GenericError(err)) + return + } + + repoDid, xerr, ok := x.resolveOwnedRepo(r, input.Repo) + if !ok { + fail(xerr) + return + } + + var sha string + ref := "" + var sourceRepo syntax.DID + var pull PullContext + var inputs []*workflow.TriggerPair + + switch { + case input.Trigger == nil: + fail(xrpcerr.GenericError(fmt.Errorf("trigger is required"))) + return + + case input.Trigger.CiTriggerManual != nil: + manual := input.Trigger.CiTriggerManual + // the trigger names a repo of its own and a mismatch would run one repo under another's auth + if manual.Repo != input.Repo { + fail(invalidRequest(fmt.Errorf("manual trigger repo %q does not match repo %q", manual.Repo, input.Repo))) + return + } + sha = manual.Commit + if manual.Ref != nil { + ref = *manual.Ref + } + inputs = manualPairsToPipelinePairs(manual.Inputs) + + case input.Trigger.EventPullRequest != nil: + pr := input.Trigger.EventPullRequest + sha = deref(pr.SourceSha) + parsed, err := parseOptionalDID("sourceRepo", pr.SourceRepo) + if err != nil { + fail(xrpcerr.GenericError(err)) + return + } + sourceRepo = parsed + + if deref(pr.TargetBranch) == "" { + fail(xrpcerr.GenericError(fmt.Errorf("pull request trigger targetBranch is required"))) + return + } + + var pullAt syntax.ATURI + if pr.Pull != nil { + var err error + pullAt, err = syntax.ParseATURI(*pr.Pull) + if err != nil { + fail(xrpcerr.InvalidRepoError(*pr.Pull)) + return + } + } + pull = PullContext{ + IsPullRequest: true, + Pull: pullAt, + SourceBranch: deref(pr.SourceBranch), + TargetBranch: deref(pr.TargetBranch), + } + + default: + fail(xrpcerr.GenericError(fmt.Errorf("unsupported trigger variant"))) + return + } + + if err := requireSha(sha); err != nil { + fail(invalidRequest(err)) + return + } + + attribution, ok := serviceauth.Attribution(r, repoDid) + if !ok { + fail(xrpcerr.MissingActorDidError) + return + } + + pipelineId, err := x.Trigger.TriggerManual(r.Context(), attribution, repoDid, sha, ref, input.Workflows, sourceRepo, pull, inputs) + if errors.Is(err, ErrOwnerPaused) { + writeOwnerPaused(w) + return + } + if errors.Is(err, ErrNoMatchingWorkflows) { + fail(xrpcerr.GenericError(err)) + return + } + if err != nil { + fail(xrpcerr.GenericError(fmt.Errorf("failed to trigger pipeline: %w", err))) + return + } + + if err := writeJson(w, http.StatusOK, org_tangled.CiTriggerPipeline_Output{Pipeline: pipelineId.String()}); err != nil { + l.ErrorContext(r.Context(), "failed to write response", "err", err) + } +} + +func manualPairsToPipelinePairs(inputs []*org_tangled.CiTriggerManual_Pair) []*workflow.TriggerPair { + if len(inputs) == 0 { + return nil + } + pairs := make([]*workflow.TriggerPair, 0, len(inputs)) + for _, input := range inputs { + if input == nil { + continue + } + pairs = append(pairs, &workflow.TriggerPair{Key: input.Key, Value: input.Value}) + } + return pairs +} + +func deref(s *string) string { + if s == nil { + return "" + } + return *s +} diff --git a/spindle/xrpc/org_tangled_event_notify_event.go b/spindle/xrpc/org_tangled_event_notify_event.go index 1d918d862..46b386128 100644 --- a/spindle/xrpc/org_tangled_event_notify_event.go +++ b/spindle/xrpc/org_tangled_event_notify_event.go @@ -141,8 +141,8 @@ func (x *Xrpc) handleOrgTangledEventNotifyEvent(w http.ResponseWriter, r *http.R } var out org_tangled.EventNotifyEvent_Output if pipeline != "" { - uri := pipeline.String() - out.Pipeline = &uri + id := pipeline.String() + out.Pipeline = &id } x.writeResponse(w, r, http.StatusOK, out) } diff --git a/spindle/xrpc/org_tangled_event_notify_event_test.go b/spindle/xrpc/org_tangled_event_notify_event_test.go index a092dec7b..fd4e315f1 100644 --- a/spindle/xrpc/org_tangled_event_notify_event_test.go +++ b/spindle/xrpc/org_tangled_event_notify_event_test.go @@ -19,6 +19,7 @@ import ( "github.com/golang-jwt/jwt/v5" "tangled.org/core/api/org_tangled" "tangled.org/core/spindle/knotclient" + "tangled.org/core/spindle/models" "tangled.org/core/xrpc/serviceauth" ) @@ -39,11 +40,11 @@ type notifyCall struct { type fakeNotifier struct { calls []notifyCall - pipeline syntax.ATURI + pipeline models.PipelineId err error } -func (f *fakeNotifier) NotifyEvent(_ context.Context, knot knotclient.Knot, repo syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (syntax.ATURI, error) { +func (f *fakeNotifier) NotifyEvent(_ context.Context, knot knotclient.Knot, repo syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) { f.calls = append(f.calls, notifyCall{knot, repo, event}) return f.pipeline, f.err } @@ -146,7 +147,7 @@ func pushBody(repo syntax.DID) org_tangled.EventNotifyEvent_Input { func TestNotifyEventPassesPushToTheSpindle(t *testing.T) { f := newNotifyFixture(t) - f.ev.pipeline = "at://did:web:knot.test/sh.tangled.pipeline/3kabc" + f.ev.pipeline = "3kabc" w := f.post(t, f.goodToken(t), pushBody(notifyRepo)) if w.Code != http.StatusOK { @@ -154,7 +155,7 @@ func TestNotifyEventPassesPushToTheSpindle(t *testing.T) { } var out org_tangled.EventNotifyEvent_Output if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || out.Pipeline == nil || *out.Pipeline != string(f.ev.pipeline) { - t.Fatalf("body = %s (err %v), want the pipeline uri", w.Body.String(), err) + t.Fatalf("body = %s (err %v), want the pipeline id", w.Body.String(), err) } if len(f.ev.calls) != 1 { diff --git a/spindle/xrpc/owner.go b/spindle/xrpc/owner.go deleted file mode 100644 index 9362a747a..000000000 --- a/spindle/xrpc/owner.go +++ /dev/null @@ -1,31 +0,0 @@ -package xrpc - -import ( - "encoding/json" - "net/http" - - "tangled.org/core/api/tangled" - xrpcerr "tangled.org/core/xrpc/errors" -) - -func (x *Xrpc) Owner(w http.ResponseWriter, r *http.Request) { - owner := x.Config.Server.Owner - if owner == "" { - writeError(w, xrpcerr.OwnerNotFoundError, http.StatusInternalServerError) - return - } - - response := tangled.Owner_Output{ - Owner: owner, - } - - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(response); err != nil { - x.Logger.ErrorContext(r.Context(), "failed to encode response", "error", err) - writeError(w, xrpcerr.NewXrpcError( - xrpcerr.WithTag("InternalServerError"), - xrpcerr.WithMessage("failed to encode response"), - ), http.StatusInternalServerError) - return - } -} diff --git a/spindle/xrpc/repo_auth_test.go b/spindle/xrpc/repo_auth_test.go index 5a7689ab1..d1267a7bc 100644 --- a/spindle/xrpc/repo_auth_test.go +++ b/spindle/xrpc/repo_auth_test.go @@ -15,7 +15,6 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/golang-jwt/jwt/v5" "tangled.org/core/api/org_tangled" - "tangled.org/core/api/tangled" "tangled.org/core/rbac" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" @@ -169,12 +168,13 @@ func (f *repoPrincipalFixture) orgListSecrets(actor, repoDid syntax.DID) *httpte func triggerInput(repoDid syntax.DID) []byte { ref := "refs/heads/main" - body, _ := json.Marshal(tangled.CiTriggerPipeline_Input{ + body, _ := json.Marshal(org_tangled.CiTriggerPipeline_Input{ Repo: repoDid.String(), - Trigger: &tangled.CiTriggerPipeline_Input_Trigger{ - CiTrigger_Manual: &tangled.CiTrigger_Manual{ - Sha: "0123456789abcdef0123456789abcdef01234567", - Ref: &ref, + Trigger: &org_tangled.CiTriggerPipeline_Input_Trigger{ + CiTriggerManual: &org_tangled.CiTriggerManual{ + Repo: repoDid.String(), + Commit: "0123456789abcdef0123456789abcdef01234567", + Ref: &ref, }, }, }) @@ -190,7 +190,7 @@ func TestRepoTokenPrincipalShouldAuthorizeRepoRequests(t *testing.T) { if w := f.orgListSecrets(f.repoDid, f.repoDid); w.Code != http.StatusOK { t.Fatalf("repo-token listSecrets = %d: %s", w.Code, w.Body.String()) } - if w := f.call(f.x.TriggerPipeline, http.MethodPost, "/sh.tangled.ci.triggerPipeline", f.repoDid, triggerInput(f.repoDid)); w.Code != http.StatusOK { + if w := f.call(f.x.handleOrgTangledCiTriggerPipeline, http.MethodPost, "/"+org_tangled.CiTriggerPipelineNSID, f.repoDid, triggerInput(f.repoDid)); w.Code != http.StatusOK { t.Fatalf("repo-token trigger = %d: %s", w.Code, w.Body.String()) } if w := f.call(f.x.WebhookList, http.MethodGet, "/org.tangled.temp.webhook.listWebhooks?repoDid="+f.repoDid.String(), f.repoDid, nil); w.Code != http.StatusOK { @@ -336,7 +336,7 @@ func TestActClaimMustNeverAuthorize(t *testing.T) { t.Fatalf("a token issued for one repo added a secret to another: %s", w.Body.String()) } - w := routerRequest(http.MethodPost, tangled.CiTriggerPipelineNSID, triggerInput(f.repoDid), "", signed(f.repoDid, f.pusherDid, tangled.CiTriggerPipelineNSID)) + w := routerRequest(http.MethodPost, org_tangled.CiTriggerPipelineNSID, triggerInput(f.repoDid), "", signed(f.repoDid, f.pusherDid, org_tangled.CiTriggerPipelineNSID)) if w.Code != http.StatusOK { t.Fatalf("trigger with the repo's own token = %d: %s", w.Code, w.Body.String()) } @@ -344,7 +344,7 @@ func TestActClaimMustNeverAuthorize(t *testing.T) { t.Fatalf("a pipeline was triggered as %s on the repo's own token naming %s", got, f.pusherDid) } - w = routerRequest(http.MethodPost, tangled.CiTriggerPipelineNSID, triggerInput(f.repoDid), "", signed(f.pusherDid, f.ownerDid, tangled.CiTriggerPipelineNSID)) + w = routerRequest(http.MethodPost, org_tangled.CiTriggerPipelineNSID, triggerInput(f.repoDid), "", signed(f.pusherDid, f.ownerDid, org_tangled.CiTriggerPipelineNSID)) if w.Code != http.StatusOK { t.Fatalf("trigger with a forged act claim = %d: %s", w.Code, w.Body.String()) } diff --git a/spindle/xrpc/repo_lookup.go b/spindle/xrpc/repo_lookup.go index eacf2e216..daf4c96f9 100644 --- a/spindle/xrpc/repo_lookup.go +++ b/spindle/xrpc/repo_lookup.go @@ -5,12 +5,14 @@ import ( "database/sql" "errors" "fmt" + "net/http" "time" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/hashicorp/golang-lru/v2/expirable" "tangled.org/core/spindle/db" "tangled.org/core/spindle/knotclient" + xrpcerr "tangled.org/core/xrpc/errors" ) const ( @@ -68,3 +70,54 @@ func (x *Xrpc) lookupRepo(ctx context.Context, repoDid syntax.DID) (*db.Repo, er } return repo, err } + +func parseOptionalDID(field string, value *string) (syntax.DID, error) { + if value == nil || *value == "" { + return "", nil + } + did, err := syntax.ParseDID(*value) + if err != nil { + return "", fmt.Errorf("invalid %s DID %q: %w", field, *value, err) + } + return did, nil +} + +func (x *Xrpc) resolveOwnedRepo(r *http.Request, repoDidStr string) (syntax.DID, xrpcerr.XrpcError, bool) { + repoDid, xerr, ok := repoDidOf(repoDidStr) + if !ok { + return "", xerr, false + } + + if xerr, ok := x.authorizeRepoPrincipal(r, repoDid, x.pushCheck); !ok { + return "", xerr, false + } + + return repoDid, xrpcerr.XrpcError{}, true +} + +func writeKnownRepoError(w http.ResponseWriter, xerr xrpcerr.XrpcError) { + status := http.StatusInternalServerError + switch xerr.Tag { + case "InvalidRequest": + status = http.StatusBadRequest + case "RepoNotFound": + status = http.StatusNotFound + } + writeError(w, xerr, status) +} + +func (x *Xrpc) resolveKnownRepoDid(repoDidStr string) (syntax.DID, xrpcerr.XrpcError, bool) { + repoDid, xerr, ok := repoDidOf(repoDidStr) + if !ok { + return "", xerr, false + } + + if _, err := x.Db.GetRepoByDid(repoDid); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", xrpcerr.RepoNotFoundError, false + } + return "", xrpcerr.GenericError(err), false + } + + return repoDid, xrpcerr.XrpcError{}, true +} diff --git a/spindle/xrpc/trigger_paused_test.go b/spindle/xrpc/trigger_paused_test.go index 30b624e74..1a7b30688 100644 --- a/spindle/xrpc/trigger_paused_test.go +++ b/spindle/xrpc/trigger_paused_test.go @@ -8,12 +8,14 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/org_tangled" + "tangled.org/core/spindle/models" "tangled.org/core/workflow" ) type pausedTrigger struct{ mockTrigger } -func (pausedTrigger) TriggerManual(context.Context, syntax.DID, syntax.DID, string, string, []string, syntax.DID, PullContext, []*workflow.TriggerPair) (syntax.ATURI, error) { +func (pausedTrigger) TriggerManual(context.Context, syntax.DID, syntax.DID, string, string, []string, syntax.DID, PullContext, []*workflow.TriggerPair) (models.PipelineId, error) { return "", ErrOwnerPaused } @@ -21,7 +23,7 @@ func TestTriggerOfAPausedOwnerIs403OwnerPaused(t *testing.T) { f := newRepoPrincipalFixture(t) f.x.Trigger = &pausedTrigger{} - w := f.call(f.x.TriggerPipeline, http.MethodPost, "/sh.tangled.ci.triggerPipeline", f.repoDid, triggerInput(f.repoDid)) + w := f.call(f.x.handleOrgTangledCiTriggerPipeline, http.MethodPost, "/"+org_tangled.CiTriggerPipelineNSID, f.repoDid, triggerInput(f.repoDid)) if w.Code != http.StatusForbidden || !strings.Contains(w.Body.String(), "OwnerPaused") { t.Fatalf("status = %d, want 403 OwnerPaused: %s", w.Code, w.Body.String()) } diff --git a/spindle/xrpc/trigger_sha_test.go b/spindle/xrpc/trigger_sha_test.go index c0fbc8c24..3ecd855b8 100644 --- a/spindle/xrpc/trigger_sha_test.go +++ b/spindle/xrpc/trigger_sha_test.go @@ -6,18 +6,18 @@ import ( "strings" "testing" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" xrpcerr "tangled.org/core/xrpc/errors" ) func TestManualTriggerShaRules(t *testing.T) { f := newRepoPrincipalFixture(t) trigger := func(sha string) (int, xrpcerr.XrpcError) { - body, _ := json.Marshal(tangled.CiTriggerPipeline_Input{ + body, _ := json.Marshal(org_tangled.CiTriggerPipeline_Input{ Repo: f.repoDid.String(), - Trigger: &tangled.CiTriggerPipeline_Input_Trigger{CiTrigger_Manual: &tangled.CiTrigger_Manual{Sha: sha}}, + Trigger: &org_tangled.CiTriggerPipeline_Input_Trigger{CiTriggerManual: &org_tangled.CiTriggerManual{Repo: f.repoDid.String(), Commit: sha}}, }) - w := f.call(f.x.TriggerPipeline, http.MethodPost, "/sh.tangled.ci.triggerPipeline", f.ownerDid, body) + w := f.call(f.x.handleOrgTangledCiTriggerPipeline, http.MethodPost, "/"+org_tangled.CiTriggerPipelineNSID, f.ownerDid, body) var out xrpcerr.XrpcError json.Unmarshal(w.Body.Bytes(), &out) return w.Code, out diff --git a/spindle/xrpc/xrpc.go b/spindle/xrpc/xrpc.go index e93cfd244..a350c4061 100644 --- a/spindle/xrpc/xrpc.go +++ b/spindle/xrpc/xrpc.go @@ -17,7 +17,6 @@ import ( "github.com/hashicorp/golang-lru/v2/expirable" "tangled.org/core/api/org_tangled" - "tangled.org/core/api/tangled" "tangled.org/core/gitutil" "tangled.org/core/idresolver" "tangled.org/core/notifier" @@ -41,9 +40,8 @@ var ( ErrNoMatchingWorkflows = errors.New("no workflows to run") ErrNotifyBusy = errors.New("too many events in flight") // anonymous reads can reach the knot, they're limited per repo for that - ErrDefinitionsBusy = errors.New("too many workflow definition reads for this repo, try again shortly") - ErrSourceRepoNotServed = errors.New("the source repo isn't served by this spindle") - ErrOwnerPaused = errors.New("this owner isn't admitted on this spindle right now, so nothing runs or is sent for their repos") + ErrDefinitionsBusy = errors.New("too many workflow definition reads for this repo, try again shortly") + ErrOwnerPaused = errors.New("this owner isn't admitted on this spindle right now, so nothing runs or is sent for their repos") ) func writeOwnerPaused(w http.ResponseWriter) { @@ -88,13 +86,12 @@ func requireSha(sha string) error { // this is to break an import cycle. spindle imports this package for Xrpc, // so this package can't import *spindle.Spindle back. type PipelineTrigger interface { - TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull PullContext, inputs []*workflow.TriggerPair) (syntax.ATURI, error) - DescribeWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) + TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull PullContext, inputs []*workflow.TriggerPair) (models.PipelineId, error) ListWorkflowDefinitions(ctx context.Context, repoDid syntax.DID, ref string) ([]*models.WorkflowDefinition, error) } type EventNotifier interface { - NotifyEvent(ctx context.Context, knot knotclient.Knot, repoDid syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (syntax.ATURI, error) + NotifyEvent(ctx context.Context, knot knotclient.Knot, repoDid syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) } type RepoWiper interface { @@ -174,8 +171,8 @@ func (x *Xrpc) Router() http.Handler { r.Post("/"+org_tangled.SecretAddSecretNSID, x.handleOrgTangledSecretAddSecret) r.Post("/"+org_tangled.SecretRemoveSecretNSID, x.handleOrgTangledSecretRemoveSecret) r.Get("/"+org_tangled.SecretListSecretsNSID, x.handleOrgTangledSecretListSecrets) - r.Post("/"+tangled.CiCancelPipelineNSID, x.CancelPipeline) - r.Post("/"+tangled.CiTriggerPipelineNSID, x.TriggerPipeline) + r.Post("/"+org_tangled.CiCancelPipelineNSID, x.handleOrgTangledCiCancelPipeline) + r.Post("/"+org_tangled.CiTriggerPipelineNSID, x.handleOrgTangledCiTriggerPipeline) r.Post("/"+org_tangled.TempSpindleQuotaSetNSID, x.SetLimit) r.Post("/"+org_tangled.TempSpindleQuotaUnsetNSID, x.UnsetLimit) r.Get("/"+org_tangled.TempSpindleQuotaGetNSID, x.GetLimit) @@ -202,13 +199,6 @@ func (x *Xrpc) Router() http.Handler { }) }) - // service query endpoints (no auth required) - r.Get("/"+tangled.OwnerNSID, x.Owner) - r.Get("/"+tangled.CiDescribeWorkflowDefinitionNSID, x.DescribeWorkflowDefinition) - r.Get("/"+tangled.CiSubscribePipelineLogsNSID, x.HandleCiSubscribePipelineLogs) - r.Get("/"+tangled.CiQueryPipelinesNSID, x.HandleCiQueryPipelines) - r.Get("/"+tangled.CiGetPipelineNSID, x.HandleCiGetPipeline) - return r } diff --git a/spindle/xrpc/xrpc_test.go b/spindle/xrpc/xrpc_test.go index f01b9ffae..11df3a2e8 100644 --- a/spindle/xrpc/xrpc_test.go +++ b/spindle/xrpc/xrpc_test.go @@ -17,7 +17,6 @@ import ( "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/org_tangled" - "tangled.org/core/api/tangled" "tangled.org/core/idresolver" "tangled.org/core/rbac" "tangled.org/core/spindle/config" @@ -35,14 +34,10 @@ type mockTrigger struct { workflowDefinitions []*models.WorkflowDefinition } -func (m *mockTrigger) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull PullContext, inputs []*workflow.TriggerPair) (syntax.ATURI, error) { +func (m *mockTrigger) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, sourceRepo syntax.DID, pull PullContext, inputs []*workflow.TriggerPair) (models.PipelineId, error) { m.triggered = true m.actor = actor - return "at://did:web:knot.test/sh.tangled.pipeline/testrkey", nil -} - -func (m *mockTrigger) DescribeWorkflowDefinition(context.Context, syntax.DID, string, syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - return &tangled.CiDescribeWorkflowDefinition_Output{}, nil + return "testrkey", nil } func (m *mockTrigger) ListWorkflowDefinitions(context.Context, syntax.DID, string) ([]*models.WorkflowDefinition, error) { @@ -113,26 +108,27 @@ func TestTriggerPipeline_RBAC(t *testing.T) { Trigger: trigger, } - sendReq := func(actor syntax.DID, input tangled.CiTriggerPipeline_Input) (*httptest.ResponseRecorder, int) { + sendReq := func(actor syntax.DID, input org_tangled.CiTriggerPipeline_Input) (*httptest.ResponseRecorder, int) { body, _ := json.Marshal(input) req := httptest.NewRequest(http.MethodPost, "/com.atproto.repo.createRecord", bytes.NewReader(body)) ctx := context.WithValue(req.Context(), ActorDid, actor) req = req.WithContext(ctx) w := httptest.NewRecorder() - x.TriggerPipeline(w, req) + x.handleOrgTangledCiTriggerPipeline(w, req) return w, w.Code } sha := "0123456789abcdef0123456789abcdef01234567" ref := "refs/heads/main" - input := tangled.CiTriggerPipeline_Input{ + input := org_tangled.CiTriggerPipeline_Input{ Repo: repoDid.String(), - Trigger: &tangled.CiTriggerPipeline_Input_Trigger{ - CiTrigger_Manual: &tangled.CiTrigger_Manual{ - Sha: sha, - Ref: &ref, + Trigger: &org_tangled.CiTriggerPipeline_Input_Trigger{ + CiTriggerManual: &org_tangled.CiTriggerManual{ + Repo: repoDid.String(), + Commit: sha, + Ref: &ref, }, }, } @@ -147,11 +143,11 @@ func TestTriggerPipeline_RBAC(t *testing.T) { if trigger.actor != pusherDid { t.Errorf("trigger actor = %q, want %q", trigger.actor, pusherDid) } - var out tangled.CiTriggerPipeline_Output + var out org_tangled.CiTriggerPipeline_Output if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil { t.Fatalf("decode trigger response: %v", err) } - if want := "at://did:web:knot.test/sh.tangled.pipeline/testrkey"; out.Pipeline != want { + if want := "testrkey"; out.Pipeline != want { t.Errorf("pipeline = %q, want %q", out.Pipeline, want) } @@ -235,18 +231,18 @@ func TestCancelPipeline_RBAC(t *testing.T) { Engines: make(map[string]models.Engine), } - sendReq := func(actor syntax.DID, input tangled.CiCancelPipeline_Input) (*httptest.ResponseRecorder, int) { + sendReq := func(actor syntax.DID, input org_tangled.CiCancelPipeline_Input) (*httptest.ResponseRecorder, int) { body, _ := json.Marshal(input) req := httptest.NewRequest(http.MethodPost, "/com.atproto.repo.createRecord", bytes.NewReader(body)) ctx := context.WithValue(req.Context(), ActorDid, actor) req = req.WithContext(ctx) w := httptest.NewRecorder() - x.CancelPipeline(w, req) + x.handleOrgTangledCiCancelPipeline(w, req) return w, w.Code } - input := tangled.CiCancelPipeline_Input{ + input := org_tangled.CiCancelPipeline_Input{ Repo: repoDid.String(), Pipeline: pipelineTid, } -- 2.54.0