tangled.org / core
Monorepo for Tangled tangled.org
1
Fork 31

Configure Feed

Select the types of activity you want to include in your feed.

spindle,lexicons: add ci.pipeline.triggerPipeline xrpc

Signed-off-by: dawn <dawn@tangled.org>

author
dawn committer
Tangled date (Jul 4, 2026, 8:12 AM +0300) commit 766f0114 parent cb4ed19f change-id wpsrlypn
+657 -169
+93 -2
api/tangled/cbor_gen.go
··· 7511 7511 } 7512 7512 7513 7513 cw := cbg.NewCborWriter(w) 7514 - fieldCount := 1 7514 + fieldCount := 3 7515 7515 7516 7516 if t.Inputs == nil { 7517 + fieldCount-- 7518 + } 7519 + 7520 + if t.Ref == nil { 7517 7521 fieldCount-- 7518 7522 } 7519 7523 ··· 7521 7525 return err 7522 7526 } 7523 7527 7528 + // t.Ref (string) (string) 7529 + if t.Ref != nil { 7530 + 7531 + if len("ref") > 1000000 { 7532 + return xerrors.Errorf("Value in field \"ref\" was too long") 7533 + } 7534 + 7535 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("ref"))); err != nil { 7536 + return err 7537 + } 7538 + if _, err := cw.WriteString(string("ref")); err != nil { 7539 + return err 7540 + } 7541 + 7542 + if t.Ref == nil { 7543 + if _, err := cw.Write(cbg.CborNull); err != nil { 7544 + return err 7545 + } 7546 + } else { 7547 + if len(*t.Ref) > 1000000 { 7548 + return xerrors.Errorf("Value in field t.Ref was too long") 7549 + } 7550 + 7551 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.Ref))); err != nil { 7552 + return err 7553 + } 7554 + if _, err := cw.WriteString(string(*t.Ref)); err != nil { 7555 + return err 7556 + } 7557 + } 7558 + } 7559 + 7560 + // t.Sha (string) (string) 7561 + if len("sha") > 1000000 { 7562 + return xerrors.Errorf("Value in field \"sha\" was too long") 7563 + } 7564 + 7565 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("sha"))); err != nil { 7566 + return err 7567 + } 7568 + if _, err := cw.WriteString(string("sha")); err != nil { 7569 + return err 7570 + } 7571 + 7572 + if len(t.Sha) > 1000000 { 7573 + return xerrors.Errorf("Value in field t.Sha was too long") 7574 + } 7575 + 7576 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Sha))); err != nil { 7577 + return err 7578 + } 7579 + if _, err := cw.WriteString(string(t.Sha)); err != nil { 7580 + return err 7581 + } 7582 + 7524 7583 // t.Inputs ([]*tangled.Pipeline_Pair) (slice) 7525 7584 if t.Inputs != nil { 7526 7585 ··· 7593 7652 } 7594 7653 7595 7654 switch string(nameBuf[:nameLen]) { 7596 - // t.Inputs ([]*tangled.Pipeline_Pair) (slice) 7655 + // t.Ref (string) (string) 7656 + case "ref": 7657 + 7658 + { 7659 + b, err := cr.ReadByte() 7660 + if err != nil { 7661 + return err 7662 + } 7663 + if b != cbg.CborNull[0] { 7664 + if err := cr.UnreadByte(); err != nil { 7665 + return err 7666 + } 7667 + 7668 + sval, err := cbg.ReadStringWithMax(cr, 1000000) 7669 + if err != nil { 7670 + return err 7671 + } 7672 + 7673 + t.Ref = (*string)(&sval) 7674 + } 7675 + } 7676 + // t.Sha (string) (string) 7677 + case "sha": 7678 + 7679 + { 7680 + sval, err := cbg.ReadStringWithMax(cr, 1000000) 7681 + if err != nil { 7682 + return err 7683 + } 7684 + 7685 + t.Sha = string(sval) 7686 + } 7687 + // t.Inputs ([]*tangled.Pipeline_Pair) (slice) 7597 7688 case "inputs": 7598 7689 7599 7690 maj, extra, err = cr.ReadHeader()
+11 -11
api/tangled/pipelinecancelPipeline.go
··· 2 2 3 3 package tangled 4 4 5 - // schema: sh.tangled.pipeline.cancelPipeline 5 + // schema: sh.tangled.ci.pipeline.cancelPipeline 6 6 7 7 import ( 8 8 "context" ··· 11 11 ) 12 12 13 13 const ( 14 - PipelineCancelPipelineNSID = "sh.tangled.pipeline.cancelPipeline" 14 + CiPipelineCancelPipelineNSID = "sh.tangled.ci.pipeline.cancelPipeline" 15 15 ) 16 16 17 - // PipelineCancelPipeline_Input is the input argument to a sh.tangled.pipeline.cancelPipeline call. 18 - type PipelineCancelPipeline_Input struct { 19 - // pipeline: pipeline at-uri 17 + // CiPipelineCancelPipeline_Input is the input argument to a sh.tangled.ci.pipeline.cancelPipeline call. 18 + type CiPipelineCancelPipeline_Input struct { 19 + // pipeline: pipeline TID 20 20 Pipeline string `json:"pipeline" cborgen:"pipeline"` 21 - // repo: repo at-uri, spindle can't resolve repo from pipeline at-uri yet 21 + // repo: git repository DID 22 22 Repo string `json:"repo" cborgen:"repo"` 23 - // workflow: workflow name 24 - Workflow string `json:"workflow" cborgen:"workflow"` 23 + // workflows: Workflow names to filter. When not provided, entire pipeline will be canceled. 24 + Workflows []string `json:"workflows,omitempty" cborgen:"workflows,omitempty"` 25 25 } 26 26 27 - // PipelineCancelPipeline calls the XRPC method "sh.tangled.pipeline.cancelPipeline". 28 - func PipelineCancelPipeline(ctx context.Context, c util.LexClient, input *PipelineCancelPipeline_Input) error { 29 - if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.pipeline.cancelPipeline", nil, input, nil); err != nil { 27 + // CiPipelineCancelPipeline calls the XRPC method "sh.tangled.ci.pipeline.cancelPipeline". 28 + func CiPipelineCancelPipeline(ctx context.Context, c util.LexClient, input *CiPipelineCancelPipeline_Input) error { 29 + if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.pipeline.cancelPipeline", nil, input, nil); err != nil { 30 30 return err 31 31 } 32 32
+43
api/tangled/pipelinetriggerPipeline.go
··· 1 + // Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. 2 + 3 + package tangled 4 + 5 + // schema: sh.tangled.ci.pipeline.triggerPipeline 6 + 7 + import ( 8 + "context" 9 + 10 + "github.com/bluesky-social/indigo/lex/util" 11 + ) 12 + 13 + const ( 14 + CiPipelineTriggerPipelineNSID = "sh.tangled.ci.pipeline.triggerPipeline" 15 + ) 16 + 17 + // CiPipelineTriggerPipeline_Input is the input argument to a sh.tangled.ci.pipeline.triggerPipeline call. 18 + type CiPipelineTriggerPipeline_Input struct { 19 + // ref: optional ref the SHA was resolved from, for display 20 + Ref *string `json:"ref,omitempty" cborgen:"ref,omitempty"` 21 + // repo: AT-URI of the sh.tangled.repo record 22 + Repo string `json:"repo" cborgen:"repo"` 23 + // sha: commit SHA to run the pipeline at 24 + Sha string `json:"sha" cborgen:"sha"` 25 + // workflows: Workflow names to run. When not provided, every dispatchable workflow is run. 26 + Workflows []string `json:"workflows,omitempty" cborgen:"workflows,omitempty"` 27 + } 28 + 29 + // CiPipelineTriggerPipeline_Output is the output of a sh.tangled.ci.pipeline.triggerPipeline call. 30 + type CiPipelineTriggerPipeline_Output struct { 31 + // pipeline: AT-URI of the created pipeline 32 + Pipeline string `json:"pipeline" cborgen:"pipeline"` 33 + } 34 + 35 + // CiPipelineTriggerPipeline calls the XRPC method "sh.tangled.ci.pipeline.triggerPipeline". 36 + func CiPipelineTriggerPipeline(ctx context.Context, c util.LexClient, input *CiPipelineTriggerPipeline_Input) (*CiPipelineTriggerPipeline_Output, error) { 37 + var out CiPipelineTriggerPipeline_Output 38 + if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.ci.pipeline.triggerPipeline", nil, input, &out); err != nil { 39 + return nil, err 40 + } 41 + 42 + return &out, nil 43 + }
+4
api/tangled/tangledpipeline.go
··· 35 35 // Pipeline_ManualTriggerData is a "manualTriggerData" in the sh.tangled.pipeline schema. 36 36 type Pipeline_ManualTriggerData struct { 37 37 Inputs []*Pipeline_Pair `json:"inputs,omitempty" cborgen:"inputs,omitempty"` 38 + // ref: optional ref the SHA was resolved from, for display and TANGLED_REF 39 + Ref *string `json:"ref,omitempty" cborgen:"ref,omitempty"` 40 + // sha: commit SHA the manual run targets 41 + Sha string `json:"sha" cborgen:"sha"` 38 42 } 39 43 40 44 // Pipeline_Pair is a "pair" in the sh.tangled.pipeline schema.
+6 -6
appview/pipelines/pipelines.go
··· 471 471 spindleClient, err := p.oauth.ServiceClient( 472 472 r, 473 473 oauth.WithService(hostname), 474 - oauth.WithLxm(tangled.PipelineCancelPipelineNSID), 474 + oauth.WithLxm(tangled.CiPipelineCancelPipelineNSID), 475 475 oauth.WithDev(noTLS), 476 476 oauth.WithTimeout(time.Second*30), // workflow cleanup usually takes time 477 477 ) 478 478 479 - if err := tangled.PipelineCancelPipeline( 479 + if err := tangled.CiPipelineCancelPipeline( 480 480 r.Context(), 481 481 spindleClient, 482 - &tangled.PipelineCancelPipeline_Input{ 483 - Repo: string(f.RepoAt()), 484 - Pipeline: pipelineId.String(), 485 - Workflow: workflowName, 482 + &tangled.CiPipelineCancelPipeline_Input{ 483 + Repo: string(f.RepoAt()), 484 + Pipeline: pipelineId.String(), 485 + Workflows: []string{workflowName}, 486 486 }, 487 487 ); err != nil { 488 488 l.Error("failed to cancel workflow", "err", err)
+62
lexicons/ci/pipeline/triggerPipeline.json
··· 1 + { 2 + "lexicon": 1, 3 + "id": "sh.tangled.ci.pipeline.triggerPipeline", 4 + "defs": { 5 + "main": { 6 + "type": "procedure", 7 + "description": "Manually trigger a pipeline at an explicit commit. Runs the named workflows, or every workflow defined in the repo when none are named.", 8 + "input": { 9 + "encoding": "application/json", 10 + "schema": { 11 + "type": "object", 12 + "required": ["repo", "sha"], 13 + "properties": { 14 + "repo": { 15 + "type": "string", 16 + "format": "at-uri", 17 + "description": "AT-URI of the sh.tangled.repo record" 18 + }, 19 + "sha": { 20 + "type": "string", 21 + "minLength": 40, 22 + "maxLength": 40, 23 + "description": "commit SHA to run the pipeline at" 24 + }, 25 + "ref": { 26 + "type": "string", 27 + "description": "optional ref the SHA was resolved from, for display" 28 + }, 29 + "workflows": { 30 + "type": "array", 31 + "items": { 32 + "type": "string", 33 + "description": "workflow name" 34 + }, 35 + "description": "Workflow names to run. When not provided, every dispatchable workflow is run." 36 + } 37 + } 38 + } 39 + }, 40 + "output": { 41 + "encoding": "application/json", 42 + "schema": { 43 + "type": "object", 44 + "required": ["pipeline"], 45 + "properties": { 46 + "pipeline": { 47 + "type": "string", 48 + "format": "at-uri", 49 + "description": "AT-URI of the created pipeline" 50 + } 51 + } 52 + } 53 + }, 54 + "errors": [ 55 + { 56 + "name": "InvalidRequest", 57 + "description": "Invalid request parameters" 58 + } 59 + ] 60 + } 61 + } 62 + }
+13
lexicons/pipeline/pipeline.json
··· 140 140 }, 141 141 "manualTriggerData": { 142 142 "type": "object", 143 + "required": [ 144 + "sha" 145 + ], 143 146 "properties": { 147 + "sha": { 148 + "type": "string", 149 + "description": "commit SHA the manual run targets", 150 + "minLength": 40, 151 + "maxLength": 40 152 + }, 153 + "ref": { 154 + "type": "string", 155 + "description": "optional ref the SHA was resolved from, for display and TANGLED_REF" 156 + }, 144 157 "inputs": { 145 158 "type": "array", 146 159 "items": {
+1
spindle/db/pipelines.go
··· 154 154 } 155 155 case "manual": 156 156 if raw.TriggerMetadata.Manual != nil { 157 + commitSha = raw.TriggerMetadata.Manual.Sha 157 158 trigger.CiTrigger_Manual = &tangled.CiTrigger_Manual{} 158 159 } 159 160 }
+58
spindle/filter_workflows_test.go
··· 1 + package spindle 2 + 3 + import ( 4 + "testing" 5 + 6 + "github.com/stretchr/testify/assert" 7 + 8 + "tangled.org/core/api/tangled" 9 + ) 10 + 11 + func TestFilterWorkflows(t *testing.T) { 12 + wf := func(name string) *tangled.Pipeline_Workflow { 13 + return &tangled.Pipeline_Workflow{Name: name} 14 + } 15 + 16 + tests := []struct { 17 + name string 18 + workflows []*tangled.Pipeline_Workflow 19 + only []string 20 + want []string 21 + }{ 22 + { 23 + name: "narrows to named workflows", 24 + workflows: []*tangled.Pipeline_Workflow{wf("ci"), wf("lint"), wf("deploy")}, 25 + only: []string{"ci", "deploy"}, 26 + want: []string{"ci", "deploy"}, 27 + }, 28 + { 29 + name: "names not present are dropped", 30 + workflows: []*tangled.Pipeline_Workflow{wf("ci")}, 31 + only: []string{"ci", "ghost"}, 32 + want: []string{"ci"}, 33 + }, 34 + { 35 + name: "no overlap yields nothing", 36 + workflows: []*tangled.Pipeline_Workflow{wf("ci")}, 37 + only: []string{"lint"}, 38 + want: nil, 39 + }, 40 + { 41 + name: "nil entries are skipped", 42 + workflows: []*tangled.Pipeline_Workflow{nil, wf("ci")}, 43 + only: []string{"ci"}, 44 + want: []string{"ci"}, 45 + }, 46 + } 47 + 48 + for _, tt := range tests { 49 + t.Run(tt.name, func(t *testing.T) { 50 + got := filterWorkflows(tt.workflows, tt.only) 51 + var names []string 52 + for _, w := range got { 53 + names = append(names, w.Name) 54 + } 55 + assert.Equal(t, tt.want, names) 56 + }) 57 + } 58 + }
+4 -4
spindle/models/clone.go
··· 99 99 return tr.PullRequest.SourceSha, nil 100 100 101 101 case workflow.TriggerKindManual: 102 - // Manual triggers don't have an explicit SHA in the metadata 103 - // For now, return empty string - could be enhanced to fetch from default branch 104 - // TODO: Implement manual trigger SHA resolution (fetch default branch HEAD) 105 - return "", nil 102 + if tr.Manual == nil { 103 + return "", fmt.Errorf("manual trigger metadata is nil") 104 + } 105 + return tr.Manual.Sha, nil 106 106 107 107 default: 108 108 return "", fmt.Errorf("unknown trigger kind: %s", tr.Kind)
+5 -1
spindle/models/clone_test.go
··· 111 111 tr := tangled.Pipeline_TriggerMetadata{ 112 112 Kind: string(workflow.TriggerKindManual), 113 113 Manual: &tangled.Pipeline_ManualTriggerData{ 114 + Sha: "manualsha456", 114 115 Inputs: nil, 115 116 }, 116 117 Repo: &tangled.Pipeline_TriggerRepo{ ··· 123 124 124 125 step := BuildCloneStep(twf, tr, false) 125 126 126 - // Manual triggers don't have a SHA yet (TODO), so git fetch won't include a SHA 127 127 allCmds := strings.Join(step.Commands(), " ") 128 128 // Should still have basic git commands 129 129 if !strings.Contains(allCmds, "git init") { ··· 131 131 } 132 132 if !strings.Contains(allCmds, "git fetch") { 133 133 t.Error("Commands should contain 'git fetch'") 134 + } 135 + // Manual triggers now carry an explicit SHA, which the fetch targets 136 + if !strings.Contains(allCmds, "manualsha456") { 137 + t.Error("Commands should contain the manual trigger SHA") 134 138 } 135 139 } 136 140
+18 -8
spindle/models/pipeline_env.go
··· 8 8 "tangled.org/core/workflow" 9 9 ) 10 10 11 - // PipelineEnvVars extracts environment variables from pipeline trigger metadata. 12 - // These are framework-provided variables that are injected into workflow steps. 11 + // PipelineEnvVars builds the standard CI environment variables for a pipeline 13 12 func PipelineEnvVars(tr *tangled.Pipeline_TriggerMetadata, pipelineId PipelineId) map[string]string { 14 13 if tr == nil { 15 14 return nil ··· 17 16 18 17 env := make(map[string]string) 19 18 20 - // Standard CI environment variable 19 + // standard CI env vars 21 20 env["CI"] = "true" 22 21 23 22 env["TANGLED_PIPELINE_ID"] = pipelineId.AtUri().String() 24 23 env["TANGLED_PIPELINE_KIND"] = tr.Kind 25 24 26 - // Repo info 25 + // repo info 27 26 if tr.Repo != nil { 28 27 env["TANGLED_REPO_KNOT"] = tr.Repo.Knot 29 28 env["TANGLED_REPO_DID"] = tr.Repo.Did ··· 55 54 56 55 case workflow.TriggerKindPullRequest: 57 56 if tr.PullRequest != nil { 58 - // For PRs, the "ref" is the source branch 57 + // for PRs, ref is the source branch 59 58 env["TANGLED_REF"] = "refs/heads/" + tr.PullRequest.SourceBranch 60 59 env["TANGLED_REF_NAME"] = tr.PullRequest.SourceBranch 61 60 env["TANGLED_REF_TYPE"] = "branch" 62 61 env["TANGLED_SHA"] = tr.PullRequest.SourceSha 63 62 env["TANGLED_COMMIT_SHA"] = tr.PullRequest.SourceSha 64 63 65 - // PR-specific variables 64 + // PR-specific env vars 66 65 env["TANGLED_PR_SOURCE_BRANCH"] = tr.PullRequest.SourceBranch 67 66 env["TANGLED_PR_TARGET_BRANCH"] = tr.PullRequest.TargetBranch 68 67 env["TANGLED_PR_SOURCE_SHA"] = tr.PullRequest.SourceSha ··· 70 69 } 71 70 72 71 case workflow.TriggerKindManual: 73 - // Manual triggers may not have ref/sha info 74 - // Include any manual inputs if present 75 72 if tr.Manual != nil { 73 + env["TANGLED_SHA"] = tr.Manual.Sha 74 + env["TANGLED_COMMIT_SHA"] = tr.Manual.Sha 75 + if tr.Manual.Ref != nil && *tr.Manual.Ref != "" { 76 + refName := plumbing.ReferenceName(*tr.Manual.Ref) 77 + refType := "branch" 78 + if refName.IsTag() { 79 + refType = "tag" 80 + } 81 + env["TANGLED_REF"] = *tr.Manual.Ref 82 + env["TANGLED_REF_NAME"] = refName.Short() 83 + env["TANGLED_REF_TYPE"] = refType 84 + } 85 + // include manual inputs if present 76 86 for _, pair := range tr.Manual.Inputs { 77 87 env["TANGLED_INPUT_"+strings.ToUpper(pair.Key)] = pair.Value 78 88 }
+10 -5
spindle/models/pipeline_env_test.go
··· 163 163 tr := &tangled.Pipeline_TriggerMetadata{ 164 164 Kind: string(workflow.TriggerKindManual), 165 165 Manual: &tangled.Pipeline_ManualTriggerData{ 166 + Sha: "manualsha789", 166 167 Inputs: []*tangled.Pipeline_Pair{ 167 168 {Key: "version", Value: "1.0.0"}, 168 169 {Key: "environment", Value: "production"}, ··· 189 190 t.Errorf("Expected TANGLED_INPUT_ENVIRONMENT='production', got '%s'", env["TANGLED_INPUT_ENVIRONMENT"]) 190 191 } 191 192 192 - // Manual triggers shouldn't have ref/sha variables 193 - if _, ok := env["TANGLED_REF"]; ok { 194 - t.Error("Manual trigger should not have TANGLED_REF") 193 + // Manual triggers carry the explicit SHA 194 + if env["TANGLED_SHA"] != "manualsha789" { 195 + t.Errorf("Expected TANGLED_SHA='manualsha789', got '%s'", env["TANGLED_SHA"]) 195 196 } 196 - if _, ok := env["TANGLED_SHA"]; ok { 197 - t.Error("Manual trigger should not have TANGLED_SHA") 197 + if env["TANGLED_COMMIT_SHA"] != "manualsha789" { 198 + t.Errorf("Expected TANGLED_COMMIT_SHA='manualsha789', got '%s'", env["TANGLED_COMMIT_SHA"]) 199 + } 200 + // No ref was supplied, so ref vars stay unset 201 + if _, ok := env["TANGLED_REF"]; ok { 202 + t.Error("Manual trigger without a ref should not have TANGLED_REF") 198 203 } 199 204 } 200 205
+144 -68
spindle/server.go
··· 396 396 Vault: s.vault, 397 397 Notifier: s.Notifier(), 398 398 ServiceAuth: serviceAuth, 399 + Trigger: s, 399 400 } 400 401 401 402 return x.Router() ··· 431 432 } 432 433 l.Info("synced git repo") 433 434 434 - scheme := "https" 435 - if s.cfg.Server.Dev { 436 - scheme = "http" 435 + triggerRepo, err := s.buildTriggerRepo(ctx, repo) 436 + if err != nil { 437 + return fmt.Errorf("building trigger repo: %w", err) 437 438 } 438 - client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} 439 439 440 - // HACK: fetch current default branch 441 - // TODO: this should be included in refUpdate event 442 - defaultBranch, _ := func(repo syntax.DID) (string, error) { 443 - defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) 444 - if err != nil { 445 - return "", err 446 - } 447 - return defaultBranchOut.Name, nil 448 - }(repoDid) 449 - 450 - compiler := workflow.Compiler{ 451 - ChangedFiles: event.ChangedFiles, 452 - Trigger: tangled.Pipeline_TriggerMetadata{ 453 - Kind: string(workflow.TriggerKindPush), 454 - Push: &tangled.Pipeline_PushTriggerData{ 455 - Ref: event.Ref, 456 - OldSha: event.OldSha, 457 - NewSha: event.NewSha, 458 - }, 459 - Repo: &tangled.Pipeline_TriggerRepo{ 460 - Did: repo.Owner.String(), 461 - Knot: repo.Knot, 462 - Repo: (*string)(&repo.Rkey), 463 - RepoDid: (*string)(&repoDid), 464 - DefaultBranch: defaultBranch, 465 - }, 440 + trigger := tangled.Pipeline_TriggerMetadata{ 441 + Kind: string(workflow.TriggerKindPush), 442 + Push: &tangled.Pipeline_PushTriggerData{ 443 + Ref: event.Ref, 444 + OldSha: event.OldSha, 445 + NewSha: event.NewSha, 466 446 }, 447 + Repo: triggerRepo, 467 448 } 468 449 469 - // load workflow definitions from rev (without spindle context) 470 - rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, event.NewSha) 450 + pipelineId, err := s.runPipeline(ctx, repoDid, trigger, event.ChangedFiles, repoCloneUri, repoPath, event.NewSha, nil) 471 451 if err != nil { 472 - return fmt.Errorf("loading pipeline: %w", err) 473 - } 474 - if len(rawPipeline) == 0 { 475 - l.Info("no workflow definition find for the repo. skipping the event") 476 - return nil 477 - } 478 - tpl := compiler.Compile(compiler.Parse(rawPipeline)) 479 - // TODO: pass compile error to workflow log 480 - for _, w := range compiler.Diagnostics.Errors { 481 - l.Error(w.String()) 452 + return err 482 453 } 483 - for _, w := range compiler.Diagnostics.Warnings { 484 - l.Warn(w.String()) 485 - } 486 - if len(tpl.Workflows) == 0 { 487 - l.Info("no workflow matching trigger 'push'. skipping the event") 454 + if pipelineId.Rkey == "" { 455 + l.Info("no workflow matched 'push' trigger, skipping the event") 488 456 return nil 489 457 } 458 + l.Info("pipeline triggered", "pipeline", pipelineId.AtUri()) 459 + } 460 + 461 + return nil 462 + } 490 463 491 - pipelineId := models.PipelineId{ 492 - Knot: tpl.TriggerMetadata.Repo.Knot, 493 - Rkey: tid.TID(), 494 - } 495 - if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { 496 - l.Error("failed to create pipeline event", "err", err) 497 - return nil 464 + // buildTriggerRepo gathers trigger metadata, resolving default branch from the knot 465 + func (s *Spindle) buildTriggerRepo(ctx context.Context, repo *db.Repo) (*tangled.Pipeline_TriggerRepo, error) { 466 + scheme := "https" 467 + if s.cfg.Server.Dev { 468 + scheme = "http" 469 + } 470 + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} 471 + 472 + // todo(dawn): this should be in the refUpdate event itself to save a roundtrip 473 + defaultBranch := "" 474 + if out, err := tangled.RepoGetDefaultBranch(ctx, client, repo.RepoDid.String()); err == nil { 475 + defaultBranch = out.Name 476 + } 477 + 478 + rkey := string(repo.Rkey) 479 + repoDid := repo.RepoDid.String() 480 + return &tangled.Pipeline_TriggerRepo{ 481 + Did: repo.Owner.String(), 482 + Knot: repo.Knot, 483 + Repo: &rkey, 484 + RepoDid: &repoDid, 485 + DefaultBranch: defaultBranch, 486 + }, nil 487 + } 488 + 489 + // runPipeline compiles and enqueues the pipeline for the given revision 490 + 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) { 491 + l := log.FromContext(ctx) 492 + 493 + compiler := workflow.Compiler{ 494 + ChangedFiles: changedFiles, 495 + Trigger: trigger, 496 + } 497 + 498 + rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, rev) 499 + if err != nil { 500 + return models.PipelineId{}, fmt.Errorf("loading pipeline: %w", err) 501 + } 502 + if len(rawPipeline) == 0 { 503 + return models.PipelineId{}, nil 504 + } 505 + 506 + tpl := compiler.Compile(compiler.Parse(rawPipeline)) 507 + // todo(dawn): pass compile error to workflow log 508 + for _, w := range compiler.Diagnostics.Errors { 509 + l.Error(w.String()) 510 + } 511 + for _, w := range compiler.Diagnostics.Warnings { 512 + l.Warn(w.String()) 513 + } 514 + 515 + if len(only) > 0 { 516 + tpl.Workflows = filterWorkflows(tpl.Workflows, only) 517 + } 518 + if len(tpl.Workflows) == 0 { 519 + return models.PipelineId{}, nil 520 + } 521 + 522 + pipelineId := models.PipelineId{ 523 + Knot: trigger.Repo.Knot, 524 + Rkey: tid.TID(), 525 + } 526 + if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { 527 + return models.PipelineId{}, fmt.Errorf("creating pipeline event: %w", err) 528 + } 529 + err = s.processPipeline(repoDid, tpl, pipelineId) 530 + return pipelineId, err 531 + } 532 + 533 + // filterWorkflows filters workflows to the requested names 534 + func filterWorkflows(workflows []*tangled.Pipeline_Workflow, only []string) []*tangled.Pipeline_Workflow { 535 + allowed := make(map[string]struct{}, len(only)) 536 + for _, n := range only { 537 + allowed[n] = struct{}{} 538 + } 539 + var filtered []*tangled.Pipeline_Workflow 540 + for _, w := range workflows { 541 + if w == nil { 542 + continue 498 543 } 499 - err = s.processPipeline(ctx, repoDid, tpl, pipelineId) 500 - if err != nil { 501 - return err 544 + if _, ok := allowed[w.Name]; ok { 545 + filtered = append(filtered, w) 502 546 } 503 547 } 548 + return filtered 549 + } 504 550 505 - return nil 551 + // TriggerManual dispatches a pipeline manually at sha 552 + func (s *Spindle) TriggerManual(ctx context.Context, repoDid syntax.DID, sha, ref string, workflows []string) (syntax.ATURI, error) { 553 + repo, err := s.db.GetRepoByDid(repoDid) 554 + if err != nil { 555 + return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) 556 + } 557 + 558 + triggerRepo, err := s.buildTriggerRepo(ctx, repo) 559 + if err != nil { 560 + return "", fmt.Errorf("building trigger repo: %w", err) 561 + } 562 + 563 + var refPtr *string 564 + if ref != "" { 565 + refPtr = &ref 566 + } 567 + trigger := tangled.Pipeline_TriggerMetadata{ 568 + Kind: string(workflow.TriggerKindManual), 569 + Manual: &tangled.Pipeline_ManualTriggerData{ 570 + Sha: sha, 571 + Ref: refPtr, 572 + }, 573 + Repo: triggerRepo, 574 + } 575 + 576 + repoCloneUri := s.newRepoCloneUrl(repo.Knot, repoDid) 577 + repoPath := s.newRepoPath(repoDid) 578 + 579 + pipelineId, err := s.runPipeline(ctx, repoDid, trigger, nil, repoCloneUri, repoPath, sha, workflows) 580 + if err != nil { 581 + return "", err 582 + } 583 + if pipelineId.Rkey == "" { 584 + return "", xrpc.ErrNoMatchingWorkflows 585 + } 586 + return pipelineId.AtUri(), nil 506 587 } 507 588 508 589 func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, error) { ··· 543 624 return rawPipeline, nil 544 625 } 545 626 546 - func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId) error { 627 + func (s *Spindle) processPipeline(repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId) error { 547 628 // Build pipeline environment variables once for all workflows 548 629 pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId) 549 630 ··· 553 634 if w == nil { 554 635 continue 555 636 } 556 - if _, ok := s.engs[w.Engine]; !ok { 637 + eng, ok := s.engs[w.Engine] 638 + if !ok { 557 639 err := s.db.StatusFailed(models.WorkflowId{ 558 640 PipelineId: pipelineId, 559 641 Name: w.Name, ··· 565 647 continue 566 648 } 567 649 568 - eng := s.engs[w.Engine] 569 - 570 - if _, ok := workflows[eng]; !ok { 571 - workflows[eng] = []models.Workflow{} 572 - } 573 - 574 - ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) 650 + ewf, err := eng.InitWorkflow(*w, tpl) 575 651 if err != nil { 576 652 err = s.db.StatusFailed(models.WorkflowId{ 577 653 PipelineId: pipelineId, ··· 597 673 // enqueue pipeline 598 674 ok := s.jq.Enqueue(repoDid, queue.Job{ 599 675 Run: func() error { 600 - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{ 676 + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{ 601 677 RepoDid: repoDid, 602 678 Workflows: workflows, 603 679 }, pipelineId)
-2
spindle/stream.go
··· 91 91 92 92 filePath := models.LogFilePath(s.cfg.Server.LogDir, wid) 93 93 94 - 95 - 96 94 config := tail.Config{ 97 95 Follow: !isFinished, 98 96 ReOpen: !isFinished,
+1 -1
spindle/tapclient.go
··· 433 433 l.Error("failed to create pipeline event", "err", err) 434 434 return nil 435 435 } 436 - err = t.spindle.processPipeline(ctx, repo.RepoDid, tpl, pipelineId) 436 + err = t.spindle.processPipeline(repo.RepoDid, tpl, pipelineId) 437 437 if err != nil { 438 438 // don't retry 439 439 l.Error("failed processing pipeline", "err", err)
-2
spindle/xrpc/ci_pipeline_subscribe_logs.go
··· 168 168 169 169 filePath := models.LogFilePath(x.Config.Server.LogDir, wid) 170 170 171 - 172 - 173 171 tailConfig := tail.Config{ 174 172 Follow: !isFinished, 175 173 ReOpen: !isFinished,
+105
spindle/xrpc/ci_pipeline_trigger_pipeline.go
··· 1 + package xrpc 2 + 3 + import ( 4 + "context" 5 + "encoding/json" 6 + "errors" 7 + "fmt" 8 + "net/http" 9 + 10 + "github.com/bluesky-social/indigo/api/atproto" 11 + "github.com/bluesky-social/indigo/atproto/syntax" 12 + "github.com/bluesky-social/indigo/xrpc" 13 + 14 + "tangled.org/core/api/tangled" 15 + "tangled.org/core/rbac" 16 + xrpcerr "tangled.org/core/xrpc/errors" 17 + ) 18 + 19 + func (x *Xrpc) TriggerPipeline(w http.ResponseWriter, r *http.Request) { 20 + l := x.Logger 21 + fail := func(e xrpcerr.XrpcError) { 22 + l.Error("failed", "kind", e.Tag, "error", e.Message) 23 + writeError(w, e, http.StatusBadRequest) 24 + } 25 + l.Debug("trigger pipeline") 26 + 27 + actorDid, ok := r.Context().Value(ActorDid).(syntax.DID) 28 + if !ok { 29 + fail(xrpcerr.MissingActorDidError) 30 + return 31 + } 32 + 33 + var input tangled.CiPipelineTriggerPipeline_Input 34 + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { 35 + fail(xrpcerr.GenericError(err)) 36 + return 37 + } 38 + 39 + if len(input.Sha) != 40 { 40 + fail(xrpcerr.GenericError(fmt.Errorf("sha must be a 40-character commit hash"))) 41 + return 42 + } 43 + 44 + repoDid, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo) 45 + if !ok { 46 + fail(xerr) 47 + return 48 + } 49 + 50 + ref := "" 51 + if input.Ref != nil { 52 + ref = *input.Ref 53 + } 54 + 55 + pipelineAt, err := x.Trigger.TriggerManual(r.Context(), repoDid, input.Sha, ref, input.Workflows) 56 + if errors.Is(err, ErrNoMatchingWorkflows) { 57 + fail(xrpcerr.GenericError(err)) 58 + return 59 + } 60 + if err != nil { 61 + fail(xrpcerr.GenericError(fmt.Errorf("failed to trigger pipeline: %w", err))) 62 + return 63 + } 64 + 65 + if err := writeJson(w, http.StatusOK, tangled.CiPipelineTriggerPipeline_Output{ 66 + Pipeline: pipelineAt.String(), 67 + }); err != nil { 68 + l.Error("failed to write response", "err", err) 69 + } 70 + } 71 + 72 + // resolveOwnedRepo resolves a repo AT-URI to DID and checks owner auth 73 + func (x *Xrpc) resolveOwnedRepo(ctx context.Context, actorDid syntax.DID, repoAtUri string) (syntax.DID, xrpcerr.XrpcError, bool) { 74 + repoAt, err := syntax.ParseATURI(repoAtUri) 75 + if err != nil { 76 + return "", xrpcerr.InvalidRepoError(repoAtUri), false 77 + } 78 + 79 + ident, err := x.Resolver.ResolveIdent(ctx, repoAt.Authority().String()) 80 + if err != nil || ident.Handle.IsInvalidHandle() { 81 + return "", xrpcerr.GenericError(fmt.Errorf("failed to resolve handle: %w", err)), false 82 + } 83 + 84 + xrpcc := xrpc.Client{Host: ident.PDSEndpoint()} 85 + resp, err := atproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String()) 86 + if err != nil { 87 + return "", xrpcerr.GenericError(err), false 88 + } 89 + 90 + repoRec, ok := resp.Value.Val.(*tangled.Repo) 91 + if !ok { 92 + return "", xrpcerr.RepoNotFoundError, false 93 + } 94 + if repoRec.RepoDid == nil || *repoRec.RepoDid == "" { 95 + return "", xrpcerr.GenericError(fmt.Errorf("repo record %s has no repoDid", repoAt)), false 96 + } 97 + repoDid := *repoRec.RepoDid 98 + 99 + isPushAllowed, err := x.Enforcer.IsPushAllowed(actorDid.String(), rbac.ThisServer, repoDid) 100 + if err != nil || !isPushAllowed { 101 + return "", xrpcerr.AccessControlError(actorDid.String()), false 102 + } 103 + 104 + return syntax.DID(repoDid), xrpcerr.XrpcError{}, true 105 + }
+37 -53
spindle/xrpc/pipeline_cancel_pipeline.go
··· 6 6 "net/http" 7 7 "strings" 8 8 9 - "github.com/bluesky-social/indigo/api/atproto" 10 9 "github.com/bluesky-social/indigo/atproto/syntax" 11 - "github.com/bluesky-social/indigo/xrpc" 12 10 "tangled.org/core/api/tangled" 13 - "tangled.org/core/rbac" 14 11 "tangled.org/core/spindle/models" 15 12 xrpcerr "tangled.org/core/xrpc/errors" 16 13 ) ··· 29 26 return 30 27 } 31 28 32 - var input tangled.PipelineCancelPipeline_Input 29 + var input tangled.CiPipelineCancelPipeline_Input 33 30 if err := json.NewDecoder(r.Body).Decode(&input); err != nil { 34 31 fail(xrpcerr.GenericError(err)) 35 32 return 36 33 } 37 34 38 35 aturi := syntax.ATURI(input.Pipeline) 39 - wid := models.WorkflowId{ 40 - PipelineId: models.PipelineId{ 41 - Knot: strings.TrimPrefix(aturi.Authority().String(), "did:web:"), 42 - Rkey: aturi.RecordKey().String(), 43 - }, 44 - Name: input.Workflow, 36 + pipelineId := models.PipelineId{ 37 + Knot: strings.TrimPrefix(aturi.Authority().String(), "did:web:"), 38 + Rkey: aturi.RecordKey().String(), 45 39 } 46 - l.Debug("cancel pipeline", "wid", wid) 47 40 48 - // unfortunately we have to resolve repo-at here 49 - repoAt, err := syntax.ParseATURI(input.Repo) 50 - if err != nil { 51 - fail(xrpcerr.InvalidRepoError(input.Repo)) 52 - return 53 - } 54 - 55 - ident, err := x.Resolver.ResolveIdent(r.Context(), repoAt.Authority().String()) 56 - if err != nil || ident.Handle.IsInvalidHandle() { 57 - fail(xrpcerr.GenericError(fmt.Errorf("failed to resolve handle: %w", err))) 58 - return 59 - } 60 - 61 - xrpcc := xrpc.Client{Host: ident.PDSEndpoint()} 62 - resp, err := atproto.RepoGetRecord(r.Context(), &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String()) 63 - if err != nil { 64 - fail(xrpcerr.GenericError(err)) 65 - return 41 + var workflows []string 42 + if len(input.Workflows) > 0 { 43 + workflows = input.Workflows 44 + } else { 45 + // fetch workflows from db if none are specified 46 + p, err := x.Db.GetPipeline(r.Context(), pipelineId.Rkey) 47 + if err != nil { 48 + fail(xrpcerr.GenericError(fmt.Errorf("failed to get pipeline: %w", err))) 49 + return 50 + } 51 + for _, w := range p.Workflows { 52 + workflows = append(workflows, w.Name) 53 + } 66 54 } 67 55 68 - repoRec, ok := resp.Value.Val.(*tangled.Repo) 69 - if !ok { 70 - fail(xrpcerr.RepoNotFoundError) 71 - return 72 - } 73 - if repoRec.RepoDid == nil || *repoRec.RepoDid == "" { 74 - fail(xrpcerr.GenericError(fmt.Errorf("repo record %s has no repoDid", repoAt))) 56 + if _, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo); !ok { 57 + fail(xerr) 75 58 return 76 59 } 77 - repoDid := *repoRec.RepoDid 78 60 79 - // TODO: fine-grained role based control 80 - isRepoOwner, err := x.Enforcer.IsRepoOwner(actorDid.String(), rbac.ThisServer, repoDid) 81 - if err != nil || !isRepoOwner { 82 - fail(xrpcerr.AccessControlError(actorDid.String())) 83 - return 84 - } 85 - for _, engine := range x.Engines { 86 - l.Debug("destroying workflow", "wid", wid) 87 - err = engine.DestroyWorkflow(r.Context(), wid) 88 - if err != nil { 89 - fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err))) 90 - return 61 + for _, wName := range workflows { 62 + wid := models.WorkflowId{ 63 + PipelineId: pipelineId, 64 + Name: wName, 91 65 } 92 - err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier) 93 - if err != nil { 94 - fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err))) 95 - return 66 + l.Debug("cancel pipeline", "wid", wid) 67 + 68 + for _, engine := range x.Engines { 69 + l.Debug("destroying workflow", "wid", wid) 70 + err := engine.DestroyWorkflow(r.Context(), wid) 71 + if err != nil { 72 + fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err))) 73 + return 74 + } 75 + err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier) 76 + if err != nil { 77 + fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err))) 78 + return 79 + } 96 80 } 97 81 } 98 82
+19 -5
spindle/xrpc/xrpc.go
··· 1 1 package xrpc 2 2 3 3 import ( 4 + "context" 4 5 _ "embed" 5 6 "encoding/json" 7 + "errors" 6 8 "log/slog" 7 9 "net/http" 8 10 11 + "github.com/bluesky-social/indigo/atproto/syntax" 9 12 "github.com/go-chi/chi/v5" 10 13 11 14 "tangled.org/core/api/tangled" ··· 22 25 23 26 const ActorDid = serviceauth.ActorDid 24 27 28 + // ErrNoMatchingWorkflows is returned when a manual dispatch resolves to no 29 + // workflows to run: the repo defines none at the requested commit, or none of 30 + // the requested workflow names exist. 31 + var ErrNoMatchingWorkflows = errors.New("no workflows to run") 32 + 33 + // PipelineTrigger builds and enqueues a manually-dispatched pipeline. It is 34 + // implemented by *spindle.Spindle, which owns the queue and engines; the xrpc 35 + // handler only does auth and input validation before delegating here. 36 + type PipelineTrigger interface { 37 + TriggerManual(ctx context.Context, repoDid syntax.DID, sha, ref string, workflows []string) (syntax.ATURI, error) 38 + } 39 + 25 40 type Xrpc struct { 26 41 Logger *slog.Logger 27 42 Db *db.DB ··· 32 47 Vault secrets.Manager 33 48 Notifier *notifier.Notifier 34 49 ServiceAuth *serviceauth.ServiceAuth 50 + Trigger PipelineTrigger 35 51 } 36 52 37 53 func (x *Xrpc) Router() http.Handler { ··· 43 59 r.Post("/"+tangled.RepoAddSecretNSID, x.AddSecret) 44 60 r.Post("/"+tangled.RepoRemoveSecretNSID, x.RemoveSecret) 45 61 r.Get("/"+tangled.RepoListSecretsNSID, x.ListSecrets) 46 - r.Post("/"+tangled.PipelineCancelPipelineNSID, x.CancelPipeline) 62 + r.Post("/"+tangled.CiPipelineCancelPipelineNSID, x.CancelPipeline) 63 + r.Post("/"+tangled.CiPipelineTriggerPipelineNSID, x.TriggerPipeline) 47 64 }) 48 65 49 66 // service query endpoints (no auth required) ··· 67 84 func writeJson(w http.ResponseWriter, status int, response any) error { 68 85 w.Header().Set("Content-Type", "application/json") 69 86 w.WriteHeader(status) 70 - if err := json.NewEncoder(w).Encode(response); err != nil { 71 - return err 72 - } 73 - return nil 87 + return json.NewEncoder(w).Encode(response) 74 88 }
+1 -1
workflow/def.go
··· 96 96 97 97 // if any of the constraints on a workflow is true, return true 98 98 func (w *Workflow) Match(trigger tangled.Pipeline_TriggerMetadata, changedFiles []string) (bool, error) { 99 - // manual triggers always run the workflow 99 + // manual dispatch skips matching constraints since selection is done by the caller 100 100 if trigger.Manual != nil { 101 101 return true, nil 102 102 }
+22
workflow/def_test.go
··· 496 496 }) 497 497 } 498 498 } 499 + 500 + func TestMatch_ManualDispatch(t *testing.T) { 501 + // manual dispatch is policy-free: every workflow matches regardless of its 502 + // declared event/branch/tag/path constraints. Selection is the caller's job. 503 + manualTrigger := tangled.Pipeline_TriggerMetadata{ 504 + Kind: string(TriggerKindManual), 505 + Manual: &tangled.Pipeline_ManualTriggerData{Sha: "deadbeef"}, 506 + } 507 + 508 + workflows := []Workflow{ 509 + {When: nil}, 510 + {When: []Constraint{{Event: []string{"push"}, Branch: []string{"main"}}}}, 511 + {When: []Constraint{{Event: []string{"pull_request"}, Paths: []string{"src/**"}}}}, 512 + {When: []Constraint{{Event: []string{"push"}, Tag: []string{"v*"}}}}, 513 + } 514 + 515 + for i, wf := range workflows { 516 + result, err := wf.Match(manualTrigger, nil) 517 + assert.NoError(t, err) 518 + assert.True(t, result, "workflow %d should match a manual dispatch", i) 519 + } 520 + }