From a7fe4409c2299ae951cb11ee11a51f9a2acda101 Mon Sep 17 00:00:00 2001 From: dawn Date: Tue, 30 Jun 2026 12:37:39 +0000 Subject: [PATCH] spindle,lexicons: add ci.pipeline.triggerPipeline xrpc Signed-off-by: dawn --- spindle/filter_workflows_test.go | 58 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/server.go | 220 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------ spindle/stream.go | 2 -- spindle/tapclient.go | 2 +- workflow/def.go | 2 +- workflow/def_test.go | 22 ++++++++++++++++++++++ api/tangled/cbor_gen.go | 95 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-- api/tangled/pipelinecancelPipeline.go | 22 +++++++++++----------- api/tangled/pipelinetriggerPipeline.go | 43 +++++++++++++++++++++++++++++++++++++++++++ api/tangled/tangledpipeline.go | 4 ++++ appview/pipelines/pipelines.go | 12 ++++++------ lexicons/pipeline/pipeline.json | 13 +++++++++++++ spindle/db/pipelines.go | 1 + spindle/models/clone.go | 8 ++++---- spindle/models/clone_test.go | 6 +++++- spindle/models/pipeline_env.go | 26 ++++++++++++++++++-------- spindle/models/pipeline_env_test.go | 15 ++++++++++----- spindle/xrpc/ci_pipeline_subscribe_logs.go | 2 -- spindle/xrpc/ci_pipeline_trigger_pipeline.go | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/xrpc/pipeline_cancel_pipeline.go | 96 ++++++++++++++++++++++++++++++++++++++++-------------------------------------------------------- spindle/xrpc/xrpc.go | 24 +++++++++++++++++++----- lexicons/ci/pipeline/triggerPipeline.json | 62 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 22 file(s) changed, 664 insertion(s)(+), 176 deletion(s)(-) diff --git a/spindle/filter_workflows_test.go b/spindle/filter_workflows_test.go new file mode 100644 --- /dev/null +++ b/spindle/filter_workflows_test.go @@ -0,0 +1,58 @@ +package spindle + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "tangled.org/core/api/tangled" +) + +func TestFilterWorkflows(t *testing.T) { + wf := func(name string) *tangled.Pipeline_Workflow { + return &tangled.Pipeline_Workflow{Name: name} + } + + tests := []struct { + name string + workflows []*tangled.Pipeline_Workflow + only []string + want []string + }{ + { + name: "narrows to named workflows", + workflows: []*tangled.Pipeline_Workflow{wf("ci"), wf("lint"), wf("deploy")}, + only: []string{"ci", "deploy"}, + want: []string{"ci", "deploy"}, + }, + { + name: "names not present are dropped", + workflows: []*tangled.Pipeline_Workflow{wf("ci")}, + only: []string{"ci", "ghost"}, + want: []string{"ci"}, + }, + { + name: "no overlap yields nothing", + workflows: []*tangled.Pipeline_Workflow{wf("ci")}, + only: []string{"lint"}, + want: nil, + }, + { + name: "nil entries are skipped", + workflows: []*tangled.Pipeline_Workflow{nil, wf("ci")}, + only: []string{"ci"}, + want: []string{"ci"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := filterWorkflows(tt.workflows, tt.only) + var names []string + for _, w := range got { + names = append(names, w.Name) + } + assert.Equal(t, tt.want, names) + }) + } +} diff --git a/spindle/server.go b/spindle/server.go --- a/spindle/server.go +++ b/spindle/server.go @@ -396,6 +396,7 @@ Vault: s.vault, Notifier: s.Notifier(), ServiceAuth: serviceAuth, + Trigger: s, } return x.Router() @@ -431,78 +432,158 @@ } l.Info("synced git repo") - scheme := "https" - if s.cfg.Server.Dev { - scheme = "http" - } - client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} - - // HACK: fetch current default branch - // TODO: this should be included in refUpdate event - defaultBranch, _ := func(repo syntax.DID) (string, error) { - defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) - if err != nil { - return "", err - } - return defaultBranchOut.Name, nil - }(repoDid) - - compiler := workflow.Compiler{ - ChangedFiles: event.ChangedFiles, - Trigger: tangled.Pipeline_TriggerMetadata{ - Kind: string(workflow.TriggerKindPush), - Push: &tangled.Pipeline_PushTriggerData{ - Ref: event.Ref, - OldSha: event.OldSha, - NewSha: event.NewSha, - }, - Repo: &tangled.Pipeline_TriggerRepo{ - Did: repo.Owner.String(), - Knot: repo.Knot, - Repo: (*string)(&repo.Rkey), - RepoDid: (*string)(&repoDid), - DefaultBranch: defaultBranch, - }, - }, - } - - // load workflow definitions from rev (without spindle context) - rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, event.NewSha) + triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { - return fmt.Errorf("loading pipeline: %w", err) - } - if len(rawPipeline) == 0 { - l.Info("no workflow definition find for the repo. skipping the event") - return nil - } - tpl := compiler.Compile(compiler.Parse(rawPipeline)) - // TODO: pass compile error to workflow log - for _, w := range compiler.Diagnostics.Errors { - l.Error(w.String()) - } - for _, w := range compiler.Diagnostics.Warnings { - l.Warn(w.String()) - } - if len(tpl.Workflows) == 0 { - l.Info("no workflow matching trigger 'push'. skipping the event") - return nil + return fmt.Errorf("building trigger repo: %w", err) } - pipelineId := models.PipelineId{ - Knot: tpl.TriggerMetadata.Repo.Knot, - Rkey: tid.TID(), + trigger := tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindPush), + Push: &tangled.Pipeline_PushTriggerData{ + Ref: event.Ref, + OldSha: event.OldSha, + NewSha: event.NewSha, + }, + Repo: triggerRepo, } - if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { - l.Error("failed to create pipeline event", "err", err) - return nil - } - err = s.processPipeline(ctx, repoDid, tpl, pipelineId) + + pipelineId, err := s.runPipeline(ctx, repoDid, trigger, event.ChangedFiles, repoCloneUri, repoPath, event.NewSha, nil) if err != nil { return err } + if pipelineId.Rkey == "" { + l.Info("no workflow matched 'push' trigger, skipping the event") + return nil + } + l.Info("pipeline triggered", "pipeline", pipelineId.AtUri()) } return nil +} + +// buildTriggerRepo gathers trigger metadata, resolving default branch from the knot +func (s *Spindle) buildTriggerRepo(ctx context.Context, repo *db.Repo) (*tangled.Pipeline_TriggerRepo, error) { + scheme := "https" + if s.cfg.Server.Dev { + scheme = "http" + } + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} + + // todo(dawn): this should be in the refUpdate event itself to save a roundtrip + defaultBranch := "" + if out, err := tangled.RepoGetDefaultBranch(ctx, client, repo.RepoDid.String()); err == nil { + defaultBranch = out.Name + } + + rkey := string(repo.Rkey) + repoDid := repo.RepoDid.String() + return &tangled.Pipeline_TriggerRepo{ + Did: repo.Owner.String(), + Knot: repo.Knot, + Repo: &rkey, + RepoDid: &repoDid, + DefaultBranch: defaultBranch, + }, nil +} + +// runPipeline compiles and enqueues the pipeline for the given revision +func (s *Spindle) runPipeline(ctx context.Context, repoDid syntax.DID, trigger tangled.Pipeline_TriggerMetadata, changedFiles []string, repoCloneUri, repoPath, rev string, only []string) (models.PipelineId, error) { + l := log.FromContext(ctx) + + compiler := workflow.Compiler{ + ChangedFiles: changedFiles, + Trigger: trigger, + } + + rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, rev) + if err != nil { + return models.PipelineId{}, fmt.Errorf("loading pipeline: %w", err) + } + if len(rawPipeline) == 0 { + return models.PipelineId{}, nil + } + + tpl := compiler.Compile(compiler.Parse(rawPipeline)) + // todo(dawn): pass compile error to workflow log + for _, w := range compiler.Diagnostics.Errors { + l.Error(w.String()) + } + for _, w := range compiler.Diagnostics.Warnings { + l.Warn(w.String()) + } + + if len(only) > 0 { + tpl.Workflows = filterWorkflows(tpl.Workflows, only) + } + if len(tpl.Workflows) == 0 { + return models.PipelineId{}, nil + } + + pipelineId := models.PipelineId{ + Knot: trigger.Repo.Knot, + Rkey: tid.TID(), + } + if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { + return models.PipelineId{}, fmt.Errorf("creating pipeline event: %w", err) + } + err = s.processPipeline(repoDid, tpl, pipelineId) + return pipelineId, err +} + +// filterWorkflows filters workflows to the requested names +func filterWorkflows(workflows []*tangled.Pipeline_Workflow, only []string) []*tangled.Pipeline_Workflow { + allowed := make(map[string]struct{}, len(only)) + for _, n := range only { + allowed[n] = struct{}{} + } + var filtered []*tangled.Pipeline_Workflow + for _, w := range workflows { + if w == nil { + continue + } + if _, ok := allowed[w.Name]; ok { + filtered = append(filtered, w) + } + } + return filtered +} + +// TriggerManual dispatches a pipeline manually at sha +func (s *Spindle) TriggerManual(ctx context.Context, repoDid syntax.DID, sha, ref string, workflows []string) (syntax.ATURI, error) { + repo, err := s.db.GetRepoByDid(repoDid) + if err != nil { + return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) + } + + triggerRepo, err := s.buildTriggerRepo(ctx, repo) + if err != nil { + return "", fmt.Errorf("building trigger repo: %w", err) + } + + var refPtr *string + if ref != "" { + refPtr = &ref + } + trigger := tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindManual), + Manual: &tangled.Pipeline_ManualTriggerData{ + Sha: sha, + Ref: refPtr, + }, + Repo: triggerRepo, + } + + repoCloneUri := s.newRepoCloneUrl(repo.Knot, repoDid) + repoPath := s.newRepoPath(repoDid) + + pipelineId, err := s.runPipeline(ctx, repoDid, trigger, nil, repoCloneUri, repoPath, sha, workflows) + if err != nil { + return "", err + } + if pipelineId.Rkey == "" { + return "", xrpc.ErrNoMatchingWorkflows + } + return pipelineId.AtUri(), nil } func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, error) { @@ -543,7 +624,7 @@ return rawPipeline, nil } -func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId) error { +func (s *Spindle) processPipeline(repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId) error { // Build pipeline environment variables once for all workflows pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId) @@ -553,7 +634,8 @@ if w == nil { continue } - if _, ok := s.engs[w.Engine]; !ok { + eng, ok := s.engs[w.Engine] + if !ok { err := s.db.StatusFailed(models.WorkflowId{ PipelineId: pipelineId, Name: w.Name, @@ -565,13 +647,7 @@ continue } - eng := s.engs[w.Engine] - - if _, ok := workflows[eng]; !ok { - workflows[eng] = []models.Workflow{} - } - - ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) + ewf, err := eng.InitWorkflow(*w, tpl) if err != nil { err = s.db.StatusFailed(models.WorkflowId{ PipelineId: pipelineId, @@ -597,7 +673,7 @@ // enqueue pipeline ok := s.jq.Enqueue(repoDid, queue.Job{ Run: func() error { - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{ + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{ RepoDid: repoDid, Workflows: workflows, }, pipelineId) diff --git a/spindle/stream.go b/spindle/stream.go --- a/spindle/stream.go +++ b/spindle/stream.go @@ -91,8 +91,6 @@ filePath := models.LogFilePath(s.cfg.Server.LogDir, wid) - - config := tail.Config{ Follow: !isFinished, ReOpen: !isFinished, diff --git a/spindle/tapclient.go b/spindle/tapclient.go --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -433,7 +433,7 @@ l.Error("failed to create pipeline event", "err", err) return nil } - err = t.spindle.processPipeline(ctx, repo.RepoDid, tpl, pipelineId) + err = t.spindle.processPipeline(repo.RepoDid, tpl, pipelineId) if err != nil { // don't retry l.Error("failed processing pipeline", "err", err) diff --git a/workflow/def.go b/workflow/def.go --- a/workflow/def.go +++ b/workflow/def.go @@ -96,7 +96,7 @@ // if any of the constraints on a workflow is true, return true func (w *Workflow) Match(trigger tangled.Pipeline_TriggerMetadata, changedFiles []string) (bool, error) { - // manual triggers always run the workflow + // manual dispatch skips matching constraints since selection is done by the caller if trigger.Manual != nil { return true, nil } diff --git a/workflow/def_test.go b/workflow/def_test.go --- a/workflow/def_test.go +++ b/workflow/def_test.go @@ -496,3 +496,25 @@ }) } } + +func TestMatch_ManualDispatch(t *testing.T) { + // manual dispatch is policy-free: every workflow matches regardless of its + // declared event/branch/tag/path constraints. Selection is the caller's job. + manualTrigger := tangled.Pipeline_TriggerMetadata{ + Kind: string(TriggerKindManual), + Manual: &tangled.Pipeline_ManualTriggerData{Sha: "deadbeef"}, + } + + workflows := []Workflow{ + {When: nil}, + {When: []Constraint{{Event: []string{"push"}, Branch: []string{"main"}}}}, + {When: []Constraint{{Event: []string{"pull_request"}, Paths: []string{"src/**"}}}}, + {When: []Constraint{{Event: []string{"push"}, Tag: []string{"v*"}}}}, + } + + for i, wf := range workflows { + result, err := wf.Match(manualTrigger, nil) + assert.NoError(t, err) + assert.True(t, result, "workflow %d should match a manual dispatch", i) + } +} diff --git a/api/tangled/cbor_gen.go b/api/tangled/cbor_gen.go --- a/api/tangled/cbor_gen.go +++ b/api/tangled/cbor_gen.go @@ -7511,13 +7511,72 @@ } cw := cbg.NewCborWriter(w) - fieldCount := 1 + fieldCount := 3 if t.Inputs == nil { fieldCount-- } + if t.Ref == nil { + fieldCount-- + } + if _, err := cw.Write(cbg.CborEncodeMajorType(cbg.MajMap, uint64(fieldCount))); err != nil { + return err + } + + // t.Ref (string) (string) + if t.Ref != nil { + + if len("ref") > 1000000 { + return xerrors.Errorf("Value in field \"ref\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("ref"))); err != nil { + return err + } + if _, err := cw.WriteString(string("ref")); err != nil { + return err + } + + if t.Ref == nil { + if _, err := cw.Write(cbg.CborNull); err != nil { + return err + } + } else { + if len(*t.Ref) > 1000000 { + return xerrors.Errorf("Value in field t.Ref was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.Ref))); err != nil { + return err + } + if _, err := cw.WriteString(string(*t.Ref)); err != nil { + return err + } + } + } + + // t.Sha (string) (string) + if len("sha") > 1000000 { + return xerrors.Errorf("Value in field \"sha\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("sha"))); err != nil { + return err + } + if _, err := cw.WriteString(string("sha")); err != nil { + return err + } + + if len(t.Sha) > 1000000 { + return xerrors.Errorf("Value in field t.Sha was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Sha))); err != nil { + return err + } + if _, err := cw.WriteString(string(t.Sha)); err != nil { return err } @@ -7593,7 +7652,39 @@ } switch string(nameBuf[:nameLen]) { - // t.Inputs ([]*tangled.Pipeline_Pair) (slice) + // t.Ref (string) (string) + case "ref": + + { + b, err := cr.ReadByte() + if err != nil { + return err + } + if b != cbg.CborNull[0] { + if err := cr.UnreadByte(); err != nil { + return err + } + + sval, err := cbg.ReadStringWithMax(cr, 1000000) + if err != nil { + return err + } + + t.Ref = (*string)(&sval) + } + } + // t.Sha (string) (string) + case "sha": + + { + sval, err := cbg.ReadStringWithMax(cr, 1000000) + if err != nil { + return err + } + + t.Sha = string(sval) + } + // t.Inputs ([]*tangled.Pipeline_Pair) (slice) case "inputs": maj, extra, err = cr.ReadHeader() diff --git a/api/tangled/pipelinecancelPipeline.go b/api/tangled/pipelinecancelPipeline.go --- a/api/tangled/pipelinecancelPipeline.go +++ b/api/tangled/pipelinecancelPipeline.go @@ -2,7 +2,7 @@ package tangled -// schema: sh.tangled.pipeline.cancelPipeline +// schema: sh.tangled.ci.pipeline.cancelPipeline import ( "context" @@ -11,22 +11,22 @@ ) const ( - PipelineCancelPipelineNSID = "sh.tangled.pipeline.cancelPipeline" + CiPipelineCancelPipelineNSID = "sh.tangled.ci.pipeline.cancelPipeline" ) -// PipelineCancelPipeline_Input is the input argument to a sh.tangled.pipeline.cancelPipeline call. -type PipelineCancelPipeline_Input struct { - // pipeline: pipeline at-uri +// CiPipelineCancelPipeline_Input is the input argument to a sh.tangled.ci.pipeline.cancelPipeline call. +type CiPipelineCancelPipeline_Input struct { + // pipeline: pipeline TID Pipeline string `json:"pipeline" cborgen:"pipeline"` - // repo: repo at-uri, spindle can't resolve repo from pipeline at-uri yet + // repo: git repository DID Repo string `json:"repo" cborgen:"repo"` - // workflow: workflow name - Workflow string `json:"workflow" cborgen:"workflow"` + // workflows: Workflow names to filter. When not provided, entire pipeline will be canceled. + Workflows []string `json:"workflows,omitempty" cborgen:"workflows,omitempty"` } -// PipelineCancelPipeline calls the XRPC method "sh.tangled.pipeline.cancelPipeline". -func PipelineCancelPipeline(ctx context.Context, c util.LexClient, input *PipelineCancelPipeline_Input) error { - if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.pipeline.cancelPipeline", nil, input, nil); err != nil { +// CiPipelineCancelPipeline calls the XRPC method "sh.tangled.ci.pipeline.cancelPipeline". +func CiPipelineCancelPipeline(ctx context.Context, c util.LexClient, input *CiPipelineCancelPipeline_Input) error { + if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.pipeline.cancelPipeline", nil, input, nil); err != nil { return err } diff --git a/api/tangled/pipelinetriggerPipeline.go b/api/tangled/pipelinetriggerPipeline.go new file mode 100644 --- /dev/null +++ b/api/tangled/pipelinetriggerPipeline.go @@ -0,0 +1,43 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +package tangled + +// schema: sh.tangled.ci.pipeline.triggerPipeline + +import ( + "context" + + "github.com/bluesky-social/indigo/lex/util" +) + +const ( + CiPipelineTriggerPipelineNSID = "sh.tangled.ci.pipeline.triggerPipeline" +) + +// CiPipelineTriggerPipeline_Input is the input argument to a sh.tangled.ci.pipeline.triggerPipeline call. +type CiPipelineTriggerPipeline_Input struct { + // ref: optional ref the SHA was resolved from, for display + Ref *string `json:"ref,omitempty" cborgen:"ref,omitempty"` + // repo: AT-URI of the sh.tangled.repo record + Repo string `json:"repo" cborgen:"repo"` + // sha: commit SHA to run the pipeline at + Sha string `json:"sha" cborgen:"sha"` + // workflows: Workflow names to run. When not provided, every dispatchable workflow is run. + Workflows []string `json:"workflows,omitempty" cborgen:"workflows,omitempty"` +} + +// CiPipelineTriggerPipeline_Output is the output of a sh.tangled.ci.pipeline.triggerPipeline call. +type CiPipelineTriggerPipeline_Output struct { + // pipeline: AT-URI of the created pipeline + Pipeline string `json:"pipeline" cborgen:"pipeline"` +} + +// CiPipelineTriggerPipeline calls the XRPC method "sh.tangled.ci.pipeline.triggerPipeline". +func CiPipelineTriggerPipeline(ctx context.Context, c util.LexClient, input *CiPipelineTriggerPipeline_Input) (*CiPipelineTriggerPipeline_Output, error) { + var out CiPipelineTriggerPipeline_Output + if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.pipeline.triggerPipeline", nil, input, &out); err != nil { + return nil, err + } + + return &out, nil +} diff --git a/api/tangled/tangledpipeline.go b/api/tangled/tangledpipeline.go --- a/api/tangled/tangledpipeline.go +++ b/api/tangled/tangledpipeline.go @@ -35,6 +35,10 @@ // Pipeline_ManualTriggerData is a "manualTriggerData" in the sh.tangled.pipeline schema. type Pipeline_ManualTriggerData struct { Inputs []*Pipeline_Pair `json:"inputs,omitempty" cborgen:"inputs,omitempty"` + // ref: optional ref the SHA was resolved from, for display and TANGLED_REF + Ref *string `json:"ref,omitempty" cborgen:"ref,omitempty"` + // sha: commit SHA the manual run targets + Sha string `json:"sha" cborgen:"sha"` } // Pipeline_Pair is a "pair" in the sh.tangled.pipeline schema. diff --git a/appview/pipelines/pipelines.go b/appview/pipelines/pipelines.go --- a/appview/pipelines/pipelines.go +++ b/appview/pipelines/pipelines.go @@ -471,18 +471,18 @@ spindleClient, err := p.oauth.ServiceClient( r, oauth.WithService(hostname), - oauth.WithLxm(tangled.PipelineCancelPipelineNSID), + oauth.WithLxm(tangled.CiPipelineCancelPipelineNSID), oauth.WithDev(noTLS), oauth.WithTimeout(time.Second*30), // workflow cleanup usually takes time ) - if err := tangled.PipelineCancelPipeline( + if err := tangled.CiPipelineCancelPipeline( r.Context(), spindleClient, - &tangled.PipelineCancelPipeline_Input{ - Repo: string(f.RepoAt()), - Pipeline: pipelineId.String(), - Workflow: workflowName, + &tangled.CiPipelineCancelPipeline_Input{ + Repo: string(f.RepoAt()), + Pipeline: pipelineId.String(), + Workflows: []string{workflowName}, }, ); err != nil { l.Error("failed to cancel workflow", "err", err) diff --git a/lexicons/pipeline/pipeline.json b/lexicons/pipeline/pipeline.json --- a/lexicons/pipeline/pipeline.json +++ b/lexicons/pipeline/pipeline.json @@ -140,7 +140,20 @@ }, "manualTriggerData": { "type": "object", + "required": [ + "sha" + ], "properties": { + "sha": { + "type": "string", + "description": "commit SHA the manual run targets", + "minLength": 40, + "maxLength": 40 + }, + "ref": { + "type": "string", + "description": "optional ref the SHA was resolved from, for display and TANGLED_REF" + }, "inputs": { "type": "array", "items": { diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go --- a/spindle/db/pipelines.go +++ b/spindle/db/pipelines.go @@ -154,6 +154,7 @@ } case "manual": if raw.TriggerMetadata.Manual != nil { + commitSha = raw.TriggerMetadata.Manual.Sha trigger.CiTrigger_Manual = &tangled.CiTrigger_Manual{} } } diff --git a/spindle/models/clone.go b/spindle/models/clone.go --- a/spindle/models/clone.go +++ b/spindle/models/clone.go @@ -99,10 +99,10 @@ return tr.PullRequest.SourceSha, nil case workflow.TriggerKindManual: - // Manual triggers don't have an explicit SHA in the metadata - // For now, return empty string - could be enhanced to fetch from default branch - // TODO: Implement manual trigger SHA resolution (fetch default branch HEAD) - return "", nil + if tr.Manual == nil { + return "", fmt.Errorf("manual trigger metadata is nil") + } + return tr.Manual.Sha, nil default: return "", fmt.Errorf("unknown trigger kind: %s", tr.Kind) diff --git a/spindle/models/clone_test.go b/spindle/models/clone_test.go --- a/spindle/models/clone_test.go +++ b/spindle/models/clone_test.go @@ -111,6 +111,7 @@ tr := tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindManual), Manual: &tangled.Pipeline_ManualTriggerData{ + Sha: "manualsha456", Inputs: nil, }, Repo: &tangled.Pipeline_TriggerRepo{ @@ -123,7 +124,6 @@ step := BuildCloneStep(twf, tr, false) - // Manual triggers don't have a SHA yet (TODO), so git fetch won't include a SHA allCmds := strings.Join(step.Commands(), " ") // Should still have basic git commands if !strings.Contains(allCmds, "git init") { @@ -131,6 +131,10 @@ } if !strings.Contains(allCmds, "git fetch") { t.Error("Commands should contain 'git fetch'") + } + // Manual triggers now carry an explicit SHA, which the fetch targets + if !strings.Contains(allCmds, "manualsha456") { + t.Error("Commands should contain the manual trigger SHA") } } diff --git a/spindle/models/pipeline_env.go b/spindle/models/pipeline_env.go --- a/spindle/models/pipeline_env.go +++ b/spindle/models/pipeline_env.go @@ -8,8 +8,7 @@ "tangled.org/core/workflow" ) -// PipelineEnvVars extracts environment variables from pipeline trigger metadata. -// These are framework-provided variables that are injected into workflow steps. +// PipelineEnvVars builds the standard CI environment variables for a pipeline func PipelineEnvVars(tr *tangled.Pipeline_TriggerMetadata, pipelineId PipelineId) map[string]string { if tr == nil { return nil @@ -17,13 +16,13 @@ env := make(map[string]string) - // Standard CI environment variable + // standard CI env vars env["CI"] = "true" env["TANGLED_PIPELINE_ID"] = pipelineId.AtUri().String() env["TANGLED_PIPELINE_KIND"] = tr.Kind - // Repo info + // repo info if tr.Repo != nil { env["TANGLED_REPO_KNOT"] = tr.Repo.Knot env["TANGLED_REPO_DID"] = tr.Repo.Did @@ -55,14 +54,14 @@ case workflow.TriggerKindPullRequest: if tr.PullRequest != nil { - // For PRs, the "ref" is the source branch + // for PRs, ref is the source branch env["TANGLED_REF"] = "refs/heads/" + tr.PullRequest.SourceBranch env["TANGLED_REF_NAME"] = tr.PullRequest.SourceBranch env["TANGLED_REF_TYPE"] = "branch" env["TANGLED_SHA"] = tr.PullRequest.SourceSha env["TANGLED_COMMIT_SHA"] = tr.PullRequest.SourceSha - // PR-specific variables + // PR-specific env vars env["TANGLED_PR_SOURCE_BRANCH"] = tr.PullRequest.SourceBranch env["TANGLED_PR_TARGET_BRANCH"] = tr.PullRequest.TargetBranch env["TANGLED_PR_SOURCE_SHA"] = tr.PullRequest.SourceSha @@ -70,9 +69,20 @@ } case workflow.TriggerKindManual: - // Manual triggers may not have ref/sha info - // Include any manual inputs if present if tr.Manual != nil { + env["TANGLED_SHA"] = tr.Manual.Sha + env["TANGLED_COMMIT_SHA"] = tr.Manual.Sha + if tr.Manual.Ref != nil && *tr.Manual.Ref != "" { + refName := plumbing.ReferenceName(*tr.Manual.Ref) + refType := "branch" + if refName.IsTag() { + refType = "tag" + } + env["TANGLED_REF"] = *tr.Manual.Ref + env["TANGLED_REF_NAME"] = refName.Short() + env["TANGLED_REF_TYPE"] = refType + } + // include manual inputs if present for _, pair := range tr.Manual.Inputs { env["TANGLED_INPUT_"+strings.ToUpper(pair.Key)] = pair.Value } diff --git a/spindle/models/pipeline_env_test.go b/spindle/models/pipeline_env_test.go --- a/spindle/models/pipeline_env_test.go +++ b/spindle/models/pipeline_env_test.go @@ -163,6 +163,7 @@ tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindManual), Manual: &tangled.Pipeline_ManualTriggerData{ + Sha: "manualsha789", Inputs: []*tangled.Pipeline_Pair{ {Key: "version", Value: "1.0.0"}, {Key: "environment", Value: "production"}, @@ -189,12 +190,16 @@ t.Errorf("Expected TANGLED_INPUT_ENVIRONMENT='production', got '%s'", env["TANGLED_INPUT_ENVIRONMENT"]) } - // Manual triggers shouldn't have ref/sha variables - if _, ok := env["TANGLED_REF"]; ok { - t.Error("Manual trigger should not have TANGLED_REF") + // Manual triggers carry the explicit SHA + if env["TANGLED_SHA"] != "manualsha789" { + t.Errorf("Expected TANGLED_SHA='manualsha789', got '%s'", env["TANGLED_SHA"]) } - if _, ok := env["TANGLED_SHA"]; ok { - t.Error("Manual trigger should not have TANGLED_SHA") + if env["TANGLED_COMMIT_SHA"] != "manualsha789" { + t.Errorf("Expected TANGLED_COMMIT_SHA='manualsha789', got '%s'", env["TANGLED_COMMIT_SHA"]) + } + // No ref was supplied, so ref vars stay unset + if _, ok := env["TANGLED_REF"]; ok { + t.Error("Manual trigger without a ref should not have TANGLED_REF") } } diff --git a/spindle/xrpc/ci_pipeline_subscribe_logs.go b/spindle/xrpc/ci_pipeline_subscribe_logs.go --- a/spindle/xrpc/ci_pipeline_subscribe_logs.go +++ b/spindle/xrpc/ci_pipeline_subscribe_logs.go @@ -168,8 +168,6 @@ filePath := models.LogFilePath(x.Config.Server.LogDir, wid) - - tailConfig := tail.Config{ Follow: !isFinished, ReOpen: !isFinished, diff --git a/spindle/xrpc/ci_pipeline_trigger_pipeline.go b/spindle/xrpc/ci_pipeline_trigger_pipeline.go new file mode 100644 --- /dev/null +++ b/spindle/xrpc/ci_pipeline_trigger_pipeline.go @@ -0,0 +1,105 @@ +package xrpc + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + + "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/bluesky-social/indigo/xrpc" + + "tangled.org/core/api/tangled" + "tangled.org/core/rbac" + xrpcerr "tangled.org/core/xrpc/errors" +) + +func (x *Xrpc) TriggerPipeline(w http.ResponseWriter, r *http.Request) { + l := x.Logger + fail := func(e xrpcerr.XrpcError) { + l.Error("failed", "kind", e.Tag, "error", e.Message) + writeError(w, e, http.StatusBadRequest) + } + l.Debug("trigger pipeline") + + actorDid, ok := r.Context().Value(ActorDid).(syntax.DID) + if !ok { + fail(xrpcerr.MissingActorDidError) + return + } + + var input tangled.CiPipelineTriggerPipeline_Input + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { + fail(xrpcerr.GenericError(err)) + return + } + + if len(input.Sha) != 40 { + fail(xrpcerr.GenericError(fmt.Errorf("sha must be a 40-character commit hash"))) + return + } + + repoDid, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo) + if !ok { + fail(xerr) + return + } + + ref := "" + if input.Ref != nil { + ref = *input.Ref + } + + pipelineAt, err := x.Trigger.TriggerManual(r.Context(), repoDid, input.Sha, ref, input.Workflows) + 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.CiPipelineTriggerPipeline_Output{ + Pipeline: pipelineAt.String(), + }); err != nil { + l.Error("failed to write response", "err", err) + } +} + +// resolveOwnedRepo resolves a repo AT-URI to DID and checks owner auth +func (x *Xrpc) resolveOwnedRepo(ctx context.Context, actorDid syntax.DID, repoAtUri string) (syntax.DID, xrpcerr.XrpcError, bool) { + repoAt, err := syntax.ParseATURI(repoAtUri) + if err != nil { + return "", xrpcerr.InvalidRepoError(repoAtUri), false + } + + ident, err := x.Resolver.ResolveIdent(ctx, repoAt.Authority().String()) + if err != nil || ident.Handle.IsInvalidHandle() { + return "", xrpcerr.GenericError(fmt.Errorf("failed to resolve handle: %w", err)), false + } + + xrpcc := xrpc.Client{Host: ident.PDSEndpoint()} + resp, err := atproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String()) + if err != nil { + return "", xrpcerr.GenericError(err), false + } + + repoRec, ok := resp.Value.Val.(*tangled.Repo) + if !ok { + return "", xrpcerr.RepoNotFoundError, false + } + if repoRec.RepoDid == nil || *repoRec.RepoDid == "" { + return "", xrpcerr.GenericError(fmt.Errorf("repo record %s has no repoDid", repoAt)), false + } + repoDid := *repoRec.RepoDid + + isPushAllowed, err := x.Enforcer.IsPushAllowed(actorDid.String(), rbac.ThisServer, repoDid) + if err != nil || !isPushAllowed { + return "", xrpcerr.AccessControlError(actorDid.String()), false + } + + return syntax.DID(repoDid), xrpcerr.XrpcError{}, true +} diff --git a/spindle/xrpc/pipeline_cancel_pipeline.go b/spindle/xrpc/pipeline_cancel_pipeline.go --- a/spindle/xrpc/pipeline_cancel_pipeline.go +++ b/spindle/xrpc/pipeline_cancel_pipeline.go @@ -6,11 +6,8 @@ "net/http" "strings" - "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" - "github.com/bluesky-social/indigo/xrpc" "tangled.org/core/api/tangled" - "tangled.org/core/rbac" "tangled.org/core/spindle/models" xrpcerr "tangled.org/core/xrpc/errors" ) @@ -29,70 +26,57 @@ return } - var input tangled.PipelineCancelPipeline_Input + var input tangled.CiPipelineCancelPipeline_Input if err := json.NewDecoder(r.Body).Decode(&input); err != nil { fail(xrpcerr.GenericError(err)) return } aturi := syntax.ATURI(input.Pipeline) - wid := models.WorkflowId{ - PipelineId: models.PipelineId{ - Knot: strings.TrimPrefix(aturi.Authority().String(), "did:web:"), - Rkey: aturi.RecordKey().String(), - }, - Name: input.Workflow, - } - l.Debug("cancel pipeline", "wid", wid) - - // unfortunately we have to resolve repo-at here - repoAt, err := syntax.ParseATURI(input.Repo) - if err != nil { - fail(xrpcerr.InvalidRepoError(input.Repo)) - return + pipelineId := models.PipelineId{ + Knot: strings.TrimPrefix(aturi.Authority().String(), "did:web:"), + Rkey: aturi.RecordKey().String(), } - ident, err := x.Resolver.ResolveIdent(r.Context(), repoAt.Authority().String()) - if err != nil || ident.Handle.IsInvalidHandle() { - fail(xrpcerr.GenericError(fmt.Errorf("failed to resolve handle: %w", err))) - return - } - - xrpcc := xrpc.Client{Host: ident.PDSEndpoint()} - resp, err := atproto.RepoGetRecord(r.Context(), &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String()) - if err != nil { - fail(xrpcerr.GenericError(err)) - return - } - - repoRec, ok := resp.Value.Val.(*tangled.Repo) - if !ok { - fail(xrpcerr.RepoNotFoundError) - return - } - if repoRec.RepoDid == nil || *repoRec.RepoDid == "" { - fail(xrpcerr.GenericError(fmt.Errorf("repo record %s has no repoDid", repoAt))) - return - } - repoDid := *repoRec.RepoDid - - // TODO: fine-grained role based control - isRepoOwner, err := x.Enforcer.IsRepoOwner(actorDid.String(), rbac.ThisServer, repoDid) - if err != nil || !isRepoOwner { - fail(xrpcerr.AccessControlError(actorDid.String())) - return - } - for _, engine := range x.Engines { - l.Debug("destroying workflow", "wid", wid) - err = engine.DestroyWorkflow(r.Context(), wid) + var workflows []string + if len(input.Workflows) > 0 { + workflows = input.Workflows + } else { + // fetch workflows from db if none are specified + p, err := x.Db.GetPipeline(r.Context(), pipelineId.Rkey) if err != nil { - fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err))) + fail(xrpcerr.GenericError(fmt.Errorf("failed to get pipeline: %w", err))) return } - err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier) - if err != nil { - fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err))) - return + for _, w := range p.Workflows { + workflows = append(workflows, w.Name) + } + } + + if _, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo); !ok { + fail(xerr) + return + } + + for _, wName := range workflows { + wid := models.WorkflowId{ + PipelineId: pipelineId, + Name: wName, + } + l.Debug("cancel pipeline", "wid", wid) + + for _, engine := range x.Engines { + l.Debug("destroying workflow", "wid", wid) + err := engine.DestroyWorkflow(r.Context(), wid) + if err != nil { + fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err))) + return + } + err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier) + if err != nil { + fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err))) + return + } } } diff --git a/spindle/xrpc/xrpc.go b/spindle/xrpc/xrpc.go --- a/spindle/xrpc/xrpc.go +++ b/spindle/xrpc/xrpc.go @@ -1,11 +1,14 @@ package xrpc import ( + "context" _ "embed" "encoding/json" + "errors" "log/slog" "net/http" + "github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-chi/chi/v5" "tangled.org/core/api/tangled" @@ -22,6 +25,18 @@ const ActorDid = serviceauth.ActorDid +// ErrNoMatchingWorkflows is returned when a manual dispatch resolves to no +// workflows to run: the repo defines none at the requested commit, or none of +// the requested workflow names exist. +var ErrNoMatchingWorkflows = errors.New("no workflows to run") + +// PipelineTrigger builds and enqueues a manually-dispatched pipeline. It is +// implemented by *spindle.Spindle, which owns the queue and engines; the xrpc +// handler only does auth and input validation before delegating here. +type PipelineTrigger interface { + TriggerManual(ctx context.Context, repoDid syntax.DID, sha, ref string, workflows []string) (syntax.ATURI, error) +} + type Xrpc struct { Logger *slog.Logger Db *db.DB @@ -32,6 +47,7 @@ Vault secrets.Manager Notifier *notifier.Notifier ServiceAuth *serviceauth.ServiceAuth + Trigger PipelineTrigger } func (x *Xrpc) Router() http.Handler { @@ -43,7 +59,8 @@ r.Post("/"+tangled.RepoAddSecretNSID, x.AddSecret) r.Post("/"+tangled.RepoRemoveSecretNSID, x.RemoveSecret) r.Get("/"+tangled.RepoListSecretsNSID, x.ListSecrets) - r.Post("/"+tangled.PipelineCancelPipelineNSID, x.CancelPipeline) + r.Post("/"+tangled.CiPipelineCancelPipelineNSID, x.CancelPipeline) + r.Post("/"+tangled.CiPipelineTriggerPipelineNSID, x.TriggerPipeline) }) // service query endpoints (no auth required) @@ -67,8 +84,5 @@ func writeJson(w http.ResponseWriter, status int, response any) error { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) - if err := json.NewEncoder(w).Encode(response); err != nil { - return err - } - return nil + return json.NewEncoder(w).Encode(response) } diff --git a/lexicons/ci/pipeline/triggerPipeline.json b/lexicons/ci/pipeline/triggerPipeline.json new file mode 100644 --- /dev/null +++ b/lexicons/ci/pipeline/triggerPipeline.json @@ -0,0 +1,62 @@ +{ + "lexicon": 1, + "id": "sh.tangled.ci.pipeline.triggerPipeline", + "defs": { + "main": { + "type": "procedure", + "description": "Manually trigger a pipeline at an explicit commit. Runs the named workflows, or every workflow defined in the repo when none are named.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repo", "sha"], + "properties": { + "repo": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the sh.tangled.repo record" + }, + "sha": { + "type": "string", + "minLength": 40, + "maxLength": 40, + "description": "commit SHA to run the pipeline at" + }, + "ref": { + "type": "string", + "description": "optional ref the SHA was resolved from, for display" + }, + "workflows": { + "type": "array", + "items": { + "type": "string", + "description": "workflow name" + }, + "description": "Workflow names to run. When not provided, every dispatchable workflow is run." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["pipeline"], + "properties": { + "pipeline": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the created pipeline" + } + } + } + }, + "errors": [ + { + "name": "InvalidRequest", + "description": "Invalid request parameters" + } + ] + } + } +} -- tangled.sh