alice.pds.demo.boltless.dev
This repository has no description
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/engine"
11 "tangled.org/core/spindle/models"
12 xrpcerr "tangled.org/core/xrpc/errors"
13)
14
15func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) {
16 l := x.Logger
17 fail := func(e xrpcerr.XrpcError) {
18 l.Error("failed", "kind", e.Tag, "error", e.Message)
19 writeError(w, e, http.StatusBadRequest)
20 }
21 l.Debug("cancel pipeline")
22
23 actorDid, ok := r.Context().Value(ActorDid).(syntax.DID)
24 if !ok {
25 fail(xrpcerr.MissingActorDidError)
26 return
27 }
28
29 var input tangled.CiCancelPipeline_Input
30 if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
31 fail(xrpcerr.GenericError(err))
32 return
33 }
34
35 pipelineTid, err := syntax.ParseTID(input.Pipeline)
36 if err != nil {
37 fail(xrpcerr.GenericError(fmt.Errorf("invalid pipeline TID %q: %w", input.Pipeline, err)))
38 return
39 }
40
41 repoDid, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo)
42 if !ok {
43 fail(xerr)
44 return
45 }
46 repo, err := x.Db.GetRepoByDid(repoDid)
47 if err != nil {
48 fail(xrpcerr.GenericError(fmt.Errorf("failed to get repo: %w", err)))
49 return
50 }
51
52 // the actor is only authorized against input.Repo, so make sure the
53 // pipeline actually belongs to it before cancelling anything
54 p, err := x.Db.GetPipeline(r.Context(), pipelineTid.String())
55 if err != nil {
56 fail(xrpcerr.GenericError(fmt.Errorf("failed to get pipeline: %w", err)))
57 return
58 }
59 if p.Repo == nil || *p.Repo != repoDid.String() {
60 fail(xrpcerr.AccessControlError(actorDid.String()))
61 return
62 }
63
64 pipelineId := models.PipelineId{
65 Knot: repo.Knot,
66 Rkey: pipelineTid.String(),
67 }
68 l = l.With("input.pipeline", pipelineTid, "input.workflows", input.Workflows)
69
70 workflows := input.Workflows
71 if len(workflows) == 0 {
72 // cancel every workflow when none are specified
73 for _, w := range p.Workflows {
74 workflows = append(workflows, w.Name)
75 }
76 }
77
78 canceled := false
79 defer func() {
80 l.Debug("canceled pipeline", "canceled", canceled)
81 }()
82
83 for _, wName := range workflows {
84 wid := models.WorkflowId{
85 PipelineId: pipelineId,
86 Name: wName,
87 }
88 l.Debug("cancel pipeline", "wid", wid)
89
90 // dont cancel a workflow that already finished
91 st, err := x.Db.GetStatus(wid)
92 if err == nil && models.StatusKind(st.Status).IsFinish() {
93 continue
94 }
95
96 if err := x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier); err != nil {
97 fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status cancelled: %w", err)))
98 return
99 }
100
101 engine.CancelWorkflow(wid)
102
103 for _, eng := range x.Engines {
104 l.Debug("destroying workflow", "wid", wid)
105 if err := eng.DestroyWorkflow(r.Context(), wid); err != nil {
106 fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err)))
107 return
108 }
109 }
110 }
111
112 canceled = true
113
114 w.WriteHeader(http.StatusOK)
115}