tangled.org
Monorepo for Tangled
tangled.org
1package xrpc
2
3import (
4 "encoding/json"
5 "fmt"
6 "net/http"
7 "strings"
8
9 "github.com/bluesky-social/indigo/atproto/syntax"
10 "tangled.org/core/api/tangled"
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.CiPipelineCancelPipeline_Input
30 if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
31 fail(xrpcerr.GenericError(err))
32 return
33 }
34
35 aturi := syntax.ATURI(input.Pipeline)
36 pipelineId := models.PipelineId{
37 Knot: strings.TrimPrefix(aturi.Authority().String(), "did:web:"),
38 Rkey: aturi.RecordKey().String(),
39 }
40
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 }
54 }
55
56 if _, xerr, ok := x.resolveOwnedRepo(r.Context(), actorDid, input.Repo); !ok {
57 fail(xerr)
58 return
59 }
60
61 for _, wName := range workflows {
62 wid := models.WorkflowId{
63 PipelineId: pipelineId,
64 Name: wName,
65 }
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 }
80 }
81 }
82
83 w.WriteHeader(http.StatusOK)
84}