willow.sh
forked from
tangled.org/core
Monorepo for Tangled
2.7 kB
107 lines
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 l = l.With("input.pipeline", pipelineTid, "input.workflows", input.Workflows)
68
69 workflows := input.Workflows
70 if len(workflows) == 0 {
71 // cancel every workflow when none are specified
72 for _, w := range p.Workflows {
73 workflows = append(workflows, w.Name)
74 }
75 }
76
77 canceled := false
78 defer func() {
79 l.Debug("canceled pipeline", "canceled", canceled)
80 }()
81
82 for _, wName := range workflows {
83 wid := models.WorkflowId{
84 PipelineId: pipelineId,
85 Name: wName,
86 }
87 l.Debug("cancel pipeline", "wid", wid)
88
89 for _, engine := range x.Engines {
90 l.Debug("destroying workflow", "wid", wid)
91 err := engine.DestroyWorkflow(r.Context(), wid)
92 if err != nil {
93 fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err)))
94 return
95 }
96 err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier)
97 if err != nil {
98 fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err)))
99 return
100 }
101 }
102 }
103
104 canceled = true
105
106 w.WriteHeader(http.StatusOK)
107}