lexicons,appview,spindle: add workflow cancel

Signed-off-by: Seongmin Lee <[e-mail]>

api/tangled/pipelinecancelPipeline.go

···
+// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT.
+
+package tangled
+
+// schema: sh.tangled.pipeline.cancelPipeline
+
+import (
+	"context"
+
+	"github.com/bluesky-social/indigo/lex/util"
+)
+
+const (
+	PipelineCancelPipelineNSID = "sh.tangled.pipeline.cancelPipeline"
+)
+
+// PipelineCancelPipeline_Input is the input argument to a sh.tangled.pipeline.cancelPipeline call.
+type PipelineCancelPipeline_Input struct {
+	// pipeline: pipeline at-uri
+	Pipeline string `json:"pipeline" cborgen:"pipeline"`
+	// repo: repo at-uri, spindle can't resolve repo from pipeline at-uri yet
+	Repo string `json:"repo" cborgen:"repo"`
+	// workflow: workflow name
+	Workflow string `json:"workflow" cborgen:"workflow"`
+}
+
+// PipelineCancelPipeline calls the XRPC method "sh.tangled.pipeline.cancelPipeline".
+func PipelineCancelPipeline(ctx context.Context, c util.LexClient, input *PipelineCancelPipeline_Input) error {
+	if err := c.LexDo(ctx, util.Procedure, "application/json", "sh.tangled.pipeline.cancelPipeline", nil, input, nil); err != nil {
+		return err
+	}
+
+	return nil
+}

appview/pages/templates/repo/pipelines/workflow.html

···
     {{ block "sidebar" . }} {{ end }}
   </div>
   <div class="col-span-1 md:col-span-3">
+    <div class="flex justify-end mb-2">
+      <button
+        class="btn"
+        hx-post="/{{ $.RepoInfo.FullName }}/pipelines/{{ .Pipeline.Id }}/workflow/{{ .Workflow }}/cancel"
+        hx-swap="none"
+        {{ if (index .Pipeline.Statuses .Workflow).Latest.Status.IsFinish -}}
+          disabled
+        {{- end }}
+      >Cancel</button>
+    </div>
     {{ block "logs" . }} {{ end }}
   </div>
 </section>

appview/pipelines/pipelines.go

···
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"log/slog"
 	"net/http"
 	"strings"
 	"time"
 
+	"tangled.org/core/api/tangled"
 	"tangled.org/core/appview/config"
 	"tangled.org/core/appview/db"
+	"tangled.org/core/appview/models"
 	"tangled.org/core/appview/oauth"
 	"tangled.org/core/appview/pages"
 	"tangled.org/core/appview/reporesolver"
···
 	r.Get("/", p.Index)
 	r.Get("/{pipeline}/workflow/{workflow}", p.Workflow)
 	r.Get("/{pipeline}/workflow/{workflow}/logs", p.Logs)
+	r.Post("/{pipeline}/workflow/{workflow}/cancel", p.Cancel)
 
 	return r
 }
···
 			}
 		}
 	}
+}
+
+func (p *Pipelines) Cancel(w http.ResponseWriter, r *http.Request) {
+	l := p.logger.With("handler", "Cancel")
+
+	var (
+		pipelineId = chi.URLParam(r, "pipeline")
+		workflow   = chi.URLParam(r, "workflow")
+	)
+	if pipelineId == "" || workflow == "" {
+		http.Error(w, "missing pipeline ID or workflow", http.StatusBadRequest)
+		return
+	}
+
+	f, err := p.repoResolver.Resolve(r)
+	if err != nil {
+		l.Error("failed to get repo and knot", "err", err)
+		http.Error(w, "bad repo/knot", http.StatusBadRequest)
+		return
+	}
+
+	pipeline, err := func() (models.Pipeline, error) {
+		ps, err := db.GetPipelineStatuses(
+			p.db,
+			1,
+			orm.FilterEq("repo_owner", f.Did),
+			orm.FilterEq("repo_name", f.Name),
+			orm.FilterEq("knot", f.Knot),
+			orm.FilterEq("id", pipelineId),
+		)
+		if err != nil {
+			return models.Pipeline{}, err
+		}
+		if len(ps) != 1 {
+			return models.Pipeline{}, fmt.Errorf("wrong pipeline count %d", len(ps))
+		}
+		return ps[0], nil
+	}()
+	if err != nil {
+		l.Error("pipeline query failed", "err", err)
+		http.Error(w, "pipeline not found", http.StatusNotFound)
+	}
+	var (
+		spindle = f.Spindle
+		knot    = f.Knot
+		rkey    = pipeline.Rkey
+	)
+
+	if spindle == "" || knot == "" || rkey == "" {
+		http.Error(w, "invalid repo info", http.StatusBadRequest)
+		return
+	}
+
+	spindleClient, err := p.oauth.ServiceClient(
+		r,
+		oauth.WithService(f.Spindle),
+		oauth.WithLxm(tangled.PipelineCancelPipelineNSID),
+		oauth.WithExp(60),
+		oauth.WithDev(p.config.Core.Dev),
+		oauth.WithTimeout(time.Second*30), // workflow cleanup usually takes time
+	)
+
+	err = tangled.PipelineCancelPipeline(
+		r.Context(),
+		spindleClient,
+		&tangled.PipelineCancelPipeline_Input{
+			Repo:     string(f.RepoAt()),
+			Pipeline: pipeline.AtUri().String(),
+			Workflow: workflow,
+		},
+	)
+	errorId := "pipeline-action"
+	if err != nil {
+		l.Error("failed to cancel pipeline", "err", err)
+		p.pages.Notice(w, errorId, "Failed to add secret.")
+		return
+	}
+	l.Debug("canceled pipeline", "uri", pipeline.AtUri())
 }
 
 // either a message or an error

lexicons/pipeline/cancelPipeline.json

···
+{
+  "lexicon": 1,
+  "id": "sh.tangled.pipeline.cancelPipeline",
+  "defs": {
+    "main": {
+      "type": "procedure",
+      "description": "Cancel a running pipeline",
+      "input": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "object",
+          "required": ["repo", "pipeline", "workflow"],
+          "properties": {
+            "repo": {
+              "type": "string",
+              "format": "at-uri",
+              "description": "repo at-uri, spindle can't resolve repo from pipeline at-uri yet"
+            },
+            "pipeline": {
+              "type": "string",
+              "format": "at-uri",
+              "description": "pipeline at-uri"
+            },
+            "workflow": {
+              "type": "string",
+              "description": "workflow name"
+            }
+          }
+        }
+      }
+    }
+  }
+}

spindle/db/events.go

···
 	return d.createStatusEvent(workflowId, models.StatusKindFailed, &workflowError, &exitCode, n)
 }
 
+func (d *DB) StatusCancelled(workflowId models.WorkflowId, workflowError string, exitCode int64, n *notifier.Notifier) error {
+	return d.createStatusEvent(workflowId, models.StatusKindCancelled, &workflowError, &exitCode, n)
+}
+
 func (d *DB) StatusSuccess(workflowId models.WorkflowId, n *notifier.Notifier) error {
 	return d.createStatusEvent(workflowId, models.StatusKindSuccess, nil, nil, n)
 }

spindle/server.go

···
 		Config:      s.cfg,
 		Resolver:    s.res,
 		Vault:       s.vault,
+		Notifier:    s.Notifier(),
 		ServiceAuth: serviceAuth,
 	}
 

spindle/xrpc/pipeline_cancelPipeline.go

···
+package xrpc
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/xrpc"
+	securejoin "github.com/cyphar/filepath-securejoin"
+	"tangled.org/core/api/tangled"
+	"tangled.org/core/rbac"
+	"tangled.org/core/spindle/models"
+	xrpcerr "tangled.org/core/xrpc/errors"
+)
+
+func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) {
+	l := x.Logger
+	fail := func(e xrpcerr.XrpcError) {
+		l.Error("failed", "kind", e.Tag, "error", e.Message)
+		writeError(w, e, http.StatusBadRequest)
+	}
+	l.Debug("cancel pipeline")
+
+	actorDid, ok := r.Context().Value(ActorDid).(syntax.DID)
+	if !ok {
+		fail(xrpcerr.MissingActorDidError)
+		return
+	}
+
+	var input tangled.PipelineCancelPipeline_Input
+	if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+		fail(xrpcerr.GenericError(err))
+		return
+	}
+
+	aturi := syntax.ATURI(input.Pipeline)
+	wid := models.WorkflowId{
+		PipelineId: models.PipelineId{
+			Knot: strings.TrimPrefix(aturi.Authority().String(), "did:web:"),
+			Rkey: aturi.RecordKey().String(),
+		},
+		Name: input.Workflow,
+	}
+	l.Debug("cancel pipeline", "wid", wid)
+
+	// unfortunately we have to resolve repo-at here
+	repoAt, err := syntax.ParseATURI(input.Repo)
+	if err != nil {
+		fail(xrpcerr.InvalidRepoError(input.Repo))
+		return
+	}
+
+	ident, err := x.Resolver.ResolveIdent(r.Context(), repoAt.Authority().String())
+	if err != nil || ident.Handle.IsInvalidHandle() {
+		fail(xrpcerr.GenericError(fmt.Errorf("failed to resolve handle: %w", err)))
+		return
+	}
+
+	xrpcc := xrpc.Client{Host: ident.PDSEndpoint()}
+	resp, err := atproto.RepoGetRecord(r.Context(), &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String())
+	if err != nil {
+		fail(xrpcerr.GenericError(err))
+		return
+	}
+
+	repo := resp.Value.Val.(*tangled.Repo)
+	didSlashRepo, err := securejoin.SecureJoin(ident.DID.String(), repo.Name)
+	if err != nil {
+		fail(xrpcerr.GenericError(err))
+		return
+	}
+
+	// TODO: fine-grained role based control
+	isRepoOwner, err := x.Enforcer.IsRepoOwner(actorDid.String(), rbac.ThisServer, didSlashRepo)
+	if err != nil || !isRepoOwner {
+		fail(xrpcerr.AccessControlError(actorDid.String()))
+		return
+	}
+	for _, engine := range x.Engines {
+		l.Debug("destorying workflow", "wid", wid)
+		err = engine.DestroyWorkflow(r.Context(), wid)
+		if err != nil {
+			fail(xrpcerr.GenericError(fmt.Errorf("dailed to destroy workflow: %w", err)))
+			return
+		}
+		err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier)
+		if err != nil {
+			fail(xrpcerr.GenericError(fmt.Errorf("dailed to emit status failed: %w", err)))
+			return
+		}
+	}
+
+	w.WriteHeader(http.StatusOK)
+}

spindle/xrpc/xrpc.go

···
 
 	"tangled.org/core/api/tangled"
 	"tangled.org/core/idresolver"
+	"tangled.org/core/notifier"
 	"tangled.org/core/rbac"
 	"tangled.org/core/spindle/config"
 	"tangled.org/core/spindle/db"
···
 	Config      *config.Config
 	Resolver    *idresolver.Resolver
 	Vault       secrets.Manager
+	Notifier    *notifier.Notifier
 	ServiceAuth *serviceauth.ServiceAuth
 }
 
···
 		r.Post("/"+tangled.RepoAddSecretNSID, x.AddSecret)
 		r.Post("/"+tangled.RepoRemoveSecretNSID, x.RemoveSecret)
 		r.Get("/"+tangled.RepoListSecretsNSID, x.ListSecrets)
+		r.Post("/"+tangled.PipelineCancelPipelineNSID, x.CancelPipeline)
 	})
 
 	// service query endpoints (no auth required)