tangled.org
Monorepo for Tangled
tangled.org
1package xrpc
2
3import (
4 "encoding/json"
5 "fmt"
6 "net/http"
7
8 "github.com/bluesky-social/indigo/atproto/syntax"
9 "tangled.org/core/api/tangled"
10 "tangled.org/core/spindle/models"
11 xrpcerr "tangled.org/core/xrpc/errors"
12)
13
14func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) {
15 l := x.Logger
16 fail := func(e xrpcerr.XrpcError) {
17 l.Error("failed", "kind", e.Tag, "error", e.Message)
18 writeError(w, e, http.StatusBadRequest)
19 }
20 l.Debug("cancel pipeline")
21
22 actorDid, ok := r.Context().Value(ActorDid).(syntax.DID)
23 if !ok {
24 fail(xrpcerr.MissingActorDidError)
25 return
26 }
27
28 var input tangled.CiCancelPipeline_Input
29 if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
30 fail(xrpcerr.GenericError(err))
31 return
32 }
33
34 pipelineTid, err := syntax.ParseTID(input.Pipeline)
35 if err != nil {
36 fail(xrpcerr.GenericError(fmt.Errorf("invalid pipeline TID %q: %w", input.Pipeline, err)))
37 return
38 }
39
40 repoDid, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo)
41 if !ok {
42 fail(xerr)
43 return
44 }
45 repo, err := x.Db.GetRepoByDid(repoDid)
46 if err != nil {
47 fail(xrpcerr.GenericError(fmt.Errorf("failed to get repo: %w", err)))
48 return
49 }
50
51 // the actor is only authorized against input.Repo, so make sure the
52 // pipeline actually belongs to it before cancelling anything
53 p, err := x.Db.GetPipeline(r.Context(), pipelineTid.String())
54 if err != nil {
55 fail(xrpcerr.GenericError(fmt.Errorf("failed to get pipeline: %w", err)))
56 return
57 }
58 if p.Repo == nil || *p.Repo != repoDid.String() {
59 fail(xrpcerr.AccessControlError(actorDid.String()))
60 return
61 }
62
63 pipelineId := models.PipelineId{
64 Knot: repo.Knot,
65 Rkey: pipelineTid.String(),
66 }
67
68 workflows := input.Workflows
69 if len(workflows) == 0 {
70 // cancel every workflow when none are specified
71 for _, w := range p.Workflows {
72 workflows = append(workflows, w.Name)
73 }
74 }
75
76 for _, wName := range workflows {
77 wid := models.WorkflowId{
78 PipelineId: pipelineId,
79 Name: wName,
80 }
81 l.Debug("cancel pipeline", "wid", wid)
82
83 for _, engine := range x.Engines {
84 l.Debug("destroying workflow", "wid", wid)
85 err := engine.DestroyWorkflow(r.Context(), wid)
86 if err != nil {
87 fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err)))
88 return
89 }
90 err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier)
91 if err != nil {
92 fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err)))
93 return
94 }
95 }
96 }
97
98 w.WriteHeader(http.StatusOK)
99}