diff --git a/internal/app/dependencies.go b/internal/app/dependencies.go index 81515a2..2e2cbd5 100644 --- a/internal/app/dependencies.go +++ b/internal/app/dependencies.go @@ -69,10 +69,13 @@ type knotClientFactory interface { type pipelineClient interface { QueryPipelines(context.Context, string, string) (*spindle.QueryPipelinesOutput, error) QueryLatestPipeline(context.Context, string) (*spindle.QueryPipelinesOutput, error) + GetPipeline(context.Context, string) (*spindle.Pipeline, error) + CancelPipeline(context.Context, spindle.CancelPipelineInput) error } type spindleClientFactory interface { New(string) (pipelineClient, error) + NewWithToken(string, string) (pipelineClient, error) } type knotOwnershipVerifier interface { @@ -135,6 +138,10 @@ func (f productionSpindleFactory) New(host string) (pipelineClient, error) { return spindle.New(host, f.httpClient) } +func (f productionSpindleFactory) NewWithToken(host, token string) (pipelineClient, error) { + return spindle.NewWithToken(host, token, f.httpClient) +} + func (f productionKnotFactory) New(host, token string) knotClient { return knot.NewWithClient(host, token, f.httpClient) } diff --git a/internal/app/pipelines.go b/internal/app/pipelines.go index a0dc376..be0cd40 100644 --- a/internal/app/pipelines.go +++ b/internal/app/pipelines.go @@ -18,6 +18,71 @@ func (s *Service) ListPipelines(ctx context.Context, target Target) ([]Pipeline, return listPipelinePages(ctx, client, repoDID) } +// CancelPipeline cancels every workflow in a pipeline, or only the selected workflows. +func (s *Service) CancelPipeline(ctx context.Context, target Target, pipelineID string, workflows []string) (*PipelineCancelResult, error) { + spindleHost, repoDID, err := s.pipelineTarget(ctx, target) + if err != nil { + return nil, err + } + client, err := s.spindle.New(spindleHost) + if err != nil { + return nil, fmt.Errorf("connect to pipeline spindle: %w", err) + } + pipeline, err := client.GetPipeline(ctx, pipelineID) + if err != nil { + return nil, err + } + cancellableWorkflows := selectCancellableWorkflows(pipeline.Workflows, workflows) + if len(cancellableWorkflows) == 0 { + return &PipelineCancelResult{Pipeline: pipelineID}, nil + } + + pds, _, err := s.authenticatedPDS(ctx) + if err != nil { + return nil, err + } + audience, err := spindle.ServiceDID(spindleHost) + if err != nil { + return nil, err + } + token, err := pds.GetServiceAuth(ctx, audience, "sh.tangled.ci.cancelPipeline") + if err != nil { + return nil, fmt.Errorf("mint pipeline cancel token: %w", err) + } + authenticatedClient, err := s.spindle.NewWithToken(spindleHost, token) + if err != nil { + return nil, fmt.Errorf("connect to pipeline spindle: %w", err) + } + workflowsForRequest := cancellableWorkflows + if len(workflows) == 0 { + workflowsForRequest = nil + } + if err := authenticatedClient.CancelPipeline(ctx, spindle.CancelPipelineInput{ + Pipeline: pipelineID, Repo: repoDID, Workflows: workflowsForRequest, + }); err != nil { + return nil, err + } + return &PipelineCancelResult{Pipeline: pipelineID, Workflows: cancellableWorkflows, CancellationRequested: true}, nil +} + +func selectCancellableWorkflows(workflows []spindle.Workflow, selected []string) []string { + selectedWorkflows := make(map[string]bool, len(selected)) + for _, workflow := range selected { + selectedWorkflows[workflow] = true + } + + cancellable := make([]string, 0, len(workflows)) + for _, workflow := range workflows { + if workflow.Status != "pending" && workflow.Status != "running" { + continue + } + if len(selectedWorkflows) == 0 || selectedWorkflows[workflow.Name] { + cancellable = append(cancellable, workflow.Name) + } + } + return cancellable +} + // PipelineStatus returns the most recent pipeline for a repository. func (s *Service) PipelineStatus(ctx context.Context, target Target) (*PipelineStatusResult, error) { client, repoDID, err := s.pipelineClient(ctx, target) @@ -37,23 +102,31 @@ func (s *Service) PipelineStatus(ctx context.Context, target Target) (*PipelineS } func (s *Service) pipelineClient(ctx context.Context, target Target) (pipelineClient, string, error) { - repo, err := s.resolveRepo(ctx, target) + spindleHost, repoDID, err := s.pipelineTarget(ctx, target) if err != nil { return nil, "", err } + client, err := s.spindle.New(spindleHost) + if err != nil { + return nil, "", fmt.Errorf("connect to pipeline spindle: %w", err) + } + return client, repoDID, nil +} + +func (s *Service) pipelineTarget(ctx context.Context, target Target) (string, string, error) { + repo, err := s.resolveRepo(ctx, target) + if err != nil { + return "", "", err + } spindleHost := stringValue(repo.Value.Spindle) if spindleHost == "" { - return nil, "", fmt.Errorf("pipelines are not configured for repository %q", target.String()) + return "", "", fmt.Errorf("pipelines are not configured for repository %q", target.String()) } repoDID := stringValue(repo.Value.RepoDid) if repoDID == "" { - return nil, "", fmt.Errorf("repository %q has no repository DID", target.String()) - } - client, err := s.spindle.New(spindleHost) - if err != nil { - return nil, "", fmt.Errorf("connect to pipeline spindle: %w", err) + return "", "", fmt.Errorf("repository %q has no repository DID", target.String()) } - return client, repoDID, nil + return spindleHost, repoDID, nil } // ViewPipeline finds a pipeline by its spindle-local ID. diff --git a/internal/app/pipelines_test.go b/internal/app/pipelines_test.go index 8a2ea2a..8a09d49 100644 --- a/internal/app/pipelines_test.go +++ b/internal/app/pipelines_test.go @@ -3,6 +3,7 @@ package app import ( "context" "errors" + "slices" "testing" "github.com/alyraffauf/tg/internal/tangledlex" @@ -67,6 +68,20 @@ func TestPipelineHasFailures(t *testing.T) { } } +func TestSelectCancellableWorkflows(t *testing.T) { + workflows := []spindle.Workflow{ + {Name: "pending.yml", Status: "pending"}, + {Name: "running.yml", Status: "running"}, + {Name: "done.yml", Status: "success"}, + } + if got := selectCancellableWorkflows(workflows, nil); !slices.Equal(got, []string{"pending.yml", "running.yml"}) { + t.Fatalf("all cancellable workflows = %q", got) + } + if got := selectCancellableWorkflows(workflows, []string{"running.yml", "done.yml"}); !slices.Equal(got, []string{"running.yml"}) { + t.Fatalf("selected cancellable workflows = %q", got) + } +} + func TestPipelineStatusReturnsLatestPipeline(t *testing.T) { client := &testPipelineClient{responses: []*spindle.QueryPipelinesOutput{{ Pipelines: []spindle.Pipeline{{ @@ -88,16 +103,54 @@ func TestPipelineStatusReturnsLatestPipeline(t *testing.T) { } } +func TestCancelPipelineMintsSpindleToken(t *testing.T) { + client := &testPipelineClient{pipeline: &spindle.Pipeline{Workflows: []spindle.Workflow{ + {Name: "test.yml", Status: "pending"}, + {Name: "done.yml", Status: "success"}, + }}} + pds := &testPDS{} + service := testService(pds, &testGit{}, &testKnot{}) + service.appview = testAppview{repo: &tangled.Repo{Value: tangledlex.Repo{ + Knot: "knot.example", Spindle: optionalString("spindle.example"), RepoDid: optionalString("did:plc:repo"), + }}} + service.spindle = testSpindleFactory{client: client} + + result, err := service.CancelPipeline(context.Background(), Target{Handle: "owner.test", Repo: "example"}, "3mrvk5dbnep22", []string{"test.yml", "done.yml", "unknown.yml"}) + if err != nil { + t.Fatalf("CancelPipeline() error = %v", err) + } + if result.Pipeline != "3mrvk5dbnep22" || len(result.Workflows) != 1 { + t.Fatalf("CancelPipeline() = %+v", result) + } + if pds.serviceAuthAudiences[0] != "did:web:spindle.example" || pds.serviceAuthLexiconMethods[0] != "sh.tangled.ci.cancelPipeline" { + t.Fatalf("service auth = audience %q, method %q", pds.serviceAuthAudiences[0], pds.serviceAuthLexiconMethods[0]) + } + if client.cancelInput.Pipeline != "3mrvk5dbnep22" || client.cancelInput.Repo != "did:plc:repo" || !slices.Equal(client.cancelInput.Workflows, []string{"test.yml"}) { + t.Fatalf("cancel input = %+v", client.cancelInput) + } +} + type testPipelineClient struct { - responses []*spindle.QueryPipelinesOutput - cursors []string - err error + responses []*spindle.QueryPipelinesOutput + cursors []string + err error + cancelInput spindle.CancelPipelineInput + pipeline *spindle.Pipeline } func (c *testPipelineClient) QueryLatestPipeline(_ context.Context, _ string) (*spindle.QueryPipelinesOutput, error) { return c.responses[0], nil } +func (c *testPipelineClient) GetPipeline(context.Context, string) (*spindle.Pipeline, error) { + return c.pipeline, c.err +} + +func (c *testPipelineClient) CancelPipeline(_ context.Context, input spindle.CancelPipelineInput) error { + c.cancelInput = input + return c.err +} + type testSpindleFactory struct { client pipelineClient } @@ -106,6 +159,10 @@ func (f testSpindleFactory) New(string) (pipelineClient, error) { return f.client, nil } +func (f testSpindleFactory) NewWithToken(string, string) (pipelineClient, error) { + return f.client, nil +} + func (c *testPipelineClient) QueryPipelines(_ context.Context, _ string, cursor string) (*spindle.QueryPipelinesOutput, error) { c.cursors = append(c.cursors, cursor) if c.err != nil { diff --git a/internal/app/types.go b/internal/app/types.go index 9cdf864..54e90a4 100644 --- a/internal/app/types.go +++ b/internal/app/types.go @@ -60,6 +60,13 @@ type PipelineStatusResult struct { HasFailures bool `json:"hasFailures"` } +// PipelineCancelResult describes a cancelled pipeline or its selected workflows. +type PipelineCancelResult struct { + Pipeline string `json:"pipeline"` + Workflows []string `json:"workflows,omitempty"` + CancellationRequested bool `json:"cancellationRequested"` +} + // SSHKeyItem is one SSH public key in a listing. type SSHKeyItem struct { Name string `json:"name"` diff --git a/internal/cli/pipeline_cancel.go b/internal/cli/pipeline_cancel.go new file mode 100644 index 0000000..2a5c6fd --- /dev/null +++ b/internal/cli/pipeline_cancel.go @@ -0,0 +1,48 @@ +package cli + +import ( + "fmt" + "strings" + + "github.com/alyraffauf/tg/internal/app" + "github.com/spf13/cobra" +) + +func newPipelineCancelCommand(service *app.Service) *cobra.Command { + var repository string + var workflows []string + + command := &cobra.Command{ + Use: "cancel ", + Short: "Cancel a pipeline or selected workflows", + Long: `Cancel every workflow in a pipeline, or only the workflows selected with --workflow. + +If --repo is not set, the repository is detected from the current directory's +git origin remote.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + target, err := resolveTargetFlag(cmd.Context(), repository, service) + if err != nil { + return err + } + result, err := service.CancelPipeline(cmd.Context(), target, args[0], workflows) + if err != nil { + return err + } + return output(cmd, result, func(result *app.PipelineCancelResult) { + if !result.CancellationRequested { + fmt.Fprintf(cmd.OutOrStdout(), "Pipeline %s has no pending or running workflows.\n", result.Pipeline) + return + } + if len(workflows) == 0 { + fmt.Fprintf(cmd.OutOrStdout(), "Cancellation requested for pipeline %s.\n", result.Pipeline) + return + } + fmt.Fprintf(cmd.OutOrStdout(), "Cancellation requested for workflows %s in pipeline %s.\n", strings.Join(result.Workflows, ", "), result.Pipeline) + }) + }, + } + command.Flags().StringVarP(&repository, "repo", "R", "", "Target repository as handle/repo") + command.Flags().StringSliceVarP(&workflows, "workflow", "w", nil, "Workflow name to cancel (repeatable)") + return command +} diff --git a/internal/cli/pipeline_cancel_test.go b/internal/cli/pipeline_cancel_test.go new file mode 100644 index 0000000..55b1165 --- /dev/null +++ b/internal/cli/pipeline_cancel_test.go @@ -0,0 +1,19 @@ +package cli + +import ( + "testing" + + "github.com/alyraffauf/tg/internal/app" +) + +func TestPipelineCancelCommandFlags(t *testing.T) { + command := newPipelineCancelCommand(&app.Service{}) + if command.Use != "cancel " { + t.Fatalf("command use = %q", command.Use) + } + for _, name := range []string{"repo", "workflow"} { + if command.Flags().Lookup(name) == nil { + t.Errorf("pipeline cancel has no --%s flag", name) + } + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 10e948c..4e40864 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -49,7 +49,7 @@ func newRoot(service *app.Service, defaultKnot, defaultSSHPort, defaultProtocol rootCmd.AddCommand(repo) pipeline := newPipelineCommand(service) - pipeline.AddCommand(newPipelineListCommand(service), newPipelineViewCommand(service), newPipelineStatusCommand(service)) + pipeline.AddCommand(newPipelineListCommand(service), newPipelineViewCommand(service), newPipelineStatusCommand(service), newPipelineCancelCommand(service)) rootCmd.AddCommand(pipeline) keys := newSSHKeyCommand(service) diff --git a/spindle/client.go b/spindle/client.go index 5d3d115..a25c538 100644 --- a/spindle/client.go +++ b/spindle/client.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "net/http" + "net/url" "strings" "github.com/bluesky-social/indigo/atproto/atclient" @@ -19,6 +20,32 @@ type Client struct { // New creates a spindle client. A spindle record may contain either a host // name or a complete HTTP URL. func New(host string, httpClient *http.Client) (*Client, error) { + return newClient(host, httpClient, nil) +} + +// NewWithToken creates a spindle client authenticated with a service-auth JWT. +func NewWithToken(host, token string, httpClient *http.Client) (*Client, error) { + return newClient(host, httpClient, bearerAuth(token)) +} + +func newClient(host string, httpClient *http.Client, auth atclient.AuthMethod) (*Client, error) { + serviceURL, err := parseServiceURL(host) + if err != nil { + return nil, err + } + return &Client{APIClient: &atclient.APIClient{Client: httpClient, Host: serviceURL.String(), Auth: auth}}, nil +} + +// ServiceDID returns the did:web identifier used as a spindle's service-auth audience. +func ServiceDID(host string) (string, error) { + serviceURL, err := parseServiceURL(host) + if err != nil { + return "", err + } + return "did:web:" + strings.ReplaceAll(serviceURL.Host, ":", "%3A"), nil +} + +func parseServiceURL(host string) (*url.URL, error) { host = strings.TrimSpace(host) if host == "" { return nil, fmt.Errorf("spindle host is empty") @@ -26,7 +53,18 @@ func New(host string, httpClient *http.Client) (*Client, error) { if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") { host = "https://" + host } - return &Client{APIClient: &atclient.APIClient{Client: httpClient, Host: host}}, nil + serviceURL, err := url.Parse(host) + if err != nil || serviceURL.Host == "" { + return nil, fmt.Errorf("invalid spindle host %q", host) + } + return serviceURL, nil +} + +type bearerAuth string + +func (b bearerAuth) DoWithAuth(client *http.Client, request *http.Request, _ syntax.NSID) (*http.Response, error) { + request.Header.Set("Authorization", "Bearer "+string(b)) + return client.Do(request) } // Workflow is one workflow executed by a pipeline. @@ -57,6 +95,30 @@ type QueryPipelinesOutput struct { Total int `json:"total"` } +// CancelPipelineInput is the argument to sh.tangled.ci.cancelPipeline. +type CancelPipelineInput struct { + Pipeline string `json:"pipeline"` + Repo string `json:"repo"` + Workflows []string `json:"workflows,omitempty"` +} + +// GetPipeline fetches one pipeline by its spindle-local ID. +func (c *Client) GetPipeline(ctx context.Context, pipelineID string) (*Pipeline, error) { + var pipeline Pipeline + if err := c.Get(ctx, syntax.NSID("sh.tangled.ci.getPipeline"), map[string]any{"pipeline": pipelineID}, &pipeline); err != nil { + return nil, fmt.Errorf("get pipeline %q: %w", pipelineID, err) + } + return &pipeline, nil +} + +// CancelPipeline cancels every workflow in a pipeline, or only Workflows when supplied. +func (c *Client) CancelPipeline(ctx context.Context, input CancelPipelineInput) error { + if err := c.Post(ctx, syntax.NSID("sh.tangled.ci.cancelPipeline"), input, nil); err != nil { + return fmt.Errorf("cancel pipeline %q: %w", input.Pipeline, err) + } + return nil +} + // QueryPipelines fetches one page of pipelines for repoDID. func (c *Client) QueryPipelines(ctx context.Context, repoDID, cursor string) (*QueryPipelinesOutput, error) { params := map[string]any{"repo": repoDID, "limit": 250} diff --git a/spindle/client_test.go b/spindle/client_test.go index 41b86d1..32e5fba 100644 --- a/spindle/client_test.go +++ b/spindle/client_test.go @@ -2,6 +2,7 @@ package spindle import ( "context" + "encoding/json" "net/http" "net/http/httptest" "testing" @@ -33,6 +34,36 @@ func TestQueryPipelinesUsesSpindleEndpoint(t *testing.T) { } } +func TestCancelPipelineAuthenticatesAndPostsInput(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/xrpc/sh.tangled.ci.cancelPipeline" || request.Method != http.MethodPost { + t.Fatalf("request = %s %s", request.Method, request.URL.Path) + } + if request.Header.Get("Authorization") != "Bearer token" { + t.Fatalf("authorization = %q", request.Header.Get("Authorization")) + } + var input CancelPipelineInput + if err := json.NewDecoder(request.Body).Decode(&input); err != nil { + t.Fatalf("decode input: %v", err) + } + if input.Pipeline != "3mrvk5dbnep22" || input.Repo != "did:plc:repo" || len(input.Workflows) != 1 { + t.Fatalf("input = %+v", input) + } + writer.WriteHeader(http.StatusOK) + })) + defer server.Close() + + client, err := NewWithToken(server.URL, "token", server.Client()) + if err != nil { + t.Fatalf("NewWithToken() error = %v", err) + } + if err := client.CancelPipeline(context.Background(), CancelPipelineInput{ + Pipeline: "3mrvk5dbnep22", Repo: "did:plc:repo", Workflows: []string{"test.yml"}, + }); err != nil { + t.Fatalf("CancelPipeline() error = %v", err) + } +} + func TestQueryLatestPipelineLimitsResults(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { if request.URL.Query().Get("repo") != "did:plc:repo" {