feat: soft delete a workflow, integration tests, fctl integration

components/fctl/cmd/orchestration/workflows/delete.go

@@ -0,0 +1,94 @@
+package workflows
+
+import (
+	"fmt"
+
+	fctl "github.com/formancehq/fctl/pkg"
+	"github.com/formancehq/formance-sdk-go/pkg/models/operations"
+	"github.com/pkg/errors"
+	"github.com/pterm/pterm"
+	"github.com/spf13/cobra"
+)
+
+type WorkflowsDeleteStore struct {
+	WorkflowId string `json:"workflowId"`
+	Success    bool   `json:"success"`
+}
+type WorkflowsDeleteController struct {
+	store *WorkflowsDeleteStore
+}
+
+var _ fctl.Controller[*WorkflowsDeleteStore] = (*WorkflowsDeleteController)(nil)
+
+func NewDefaultWorkflowsDeleteStore() *WorkflowsDeleteStore {
+	return &WorkflowsDeleteStore{}
+}
+
+func NewWorkflowsDeleteController() *WorkflowsDeleteController {
+	return &WorkflowsDeleteController{
+		store: NewDefaultWorkflowsDeleteStore(),
+	}
+}
+func NewDeleteCommand() *cobra.Command {
+	return fctl.NewCommand("delete <workflow-id>",
+		fctl.WithAliases("del", "d"),
+		fctl.WithShortDescription("Soft delete a workflow"),
+		fctl.WithArgs(cobra.ExactArgs(1)),
+		fctl.WithController[*WorkflowsDeleteStore](NewWorkflowsDeleteController()),
+	)
+}
+
+func (c *WorkflowsDeleteController) GetStore() *WorkflowsDeleteStore {
+	return c.store
+}
+
+func (c *WorkflowsDeleteController) Run(cmd *cobra.Command, args []string) (fctl.Renderable, error) {
+	cfg, err := fctl.GetConfig(cmd)
+	if err != nil {
+		return nil, errors.Wrap(err, "retrieving config")
+	}
+
+	organizationID, err := fctl.ResolveOrganizationID(cmd, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	stack, err := fctl.ResolveStack(cmd, cfg, organizationID)
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := fctl.NewStackClient(cmd, cfg, stack)
+	if err != nil {
+		return nil, errors.Wrap(err, "creating stack client")
+	}
+
+	response, err := client.Orchestration.DeleteWorkflow(
+		cmd.Context(),
+		operations.DeleteWorkflowRequest{
+			FlowID: args[0],
+		},
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, fmt.Errorf("%s: %s", response.Error.ErrorCode, response.Error.ErrorMessage)
+	}
+
+	if response.StatusCode >= 300 {
+		return nil, fmt.Errorf("unexpected status code: %d", response.StatusCode)
+	}
+
+	c.store.WorkflowId = args[0]
+	c.store.Success = true
+
+	return c, nil
+}
+
+func (c *WorkflowsDeleteController) Render(cmd *cobra.Command, args []string) error {
+	pterm.Success.WithShowLineNumber().Printfln("Workflow %s Deleted!", c.store.WorkflowId)
+	return nil
+}

components/fctl/cmd/orchestration/workflows/root.go

@@ -14,6 +14,7 @@ func NewCommand() *cobra.Command {
 			NewCreateCommand(),
 			NewRunCommand(),
 			NewShowCommand(),
+			NewDeleteCommand(),
 		),
 	)
 }

components/orchestration/internal/api/handler_delete_workflow.go

@@ -0,0 +1,40 @@
+package api
+
+import (
+	"errors"
+	"net/http"
+
+	"github.com/formancehq/orchestration/internal/workflow"
+	"github.com/formancehq/stack/libs/go-libs/api"
+	"github.com/go-playground/validator/v10"
+)
+
+var (
+	ErrEmptyID = errors.New("ID is empty")
+)
+
+func deleteWorkflow(m *workflow.Manager) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		id := workflowID(r)
+
+		err := validator.New().Var(id, "required,uuid")
+		if err != nil {
+			api.BadRequest(w, "VALIDATION", err)
+			return
+		}
+
+		err = m.DeleteWorkflow(r.Context(), workflowID(r))
+
+		if errors.Is(err, workflow.ErrWorkflowNotFound) {
+			api.NotFound(w)
+			return
+		}
+
+		if err != nil {
+			api.InternalServerError(w, r, err)
+			return
+		}
+
+		api.NoContent(w)
+	}
+}

components/orchestration/internal/api/handler_delete_workflow_test.go

@@ -0,0 +1,41 @@
+package api
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/formancehq/orchestration/internal/workflow"
+	"github.com/formancehq/stack/libs/go-libs/api/apitesting"
+	"github.com/go-chi/chi/v5"
+	"github.com/stretchr/testify/require"
+
+	"github.com/uptrace/bun"
+)
+
+func TestDeleteWorkflow(t *testing.T) {
+	test(t, func(router *chi.Mux, m *workflow.Manager, db *bun.DB) {
+		// Create a workflow
+		req := httptest.NewRequest(http.MethodPost, "/workflows", bytes.NewBufferString(`{"stages": []}`))
+		rec := httptest.NewRecorder()
+
+		router.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusCreated, rec.Result().StatusCode)
+
+		workflow := workflow.Workflow{}
+		apitesting.ReadResponse(t, rec, &workflow)
+
+		require.NotEmpty(t, workflow.ID)
+
+		// 	Delete the workflow
+		req = httptest.NewRequest(http.MethodDelete, fmt.Sprintf("/workflows/%s/", workflow.ID), nil)
+		rec = httptest.NewRecorder()
+
+		router.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusNoContent, rec.Result().StatusCode)
+	})
+}

components/orchestration/internal/api/handler_list_instances_test.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -16,7 +17,7 @@ import (
 
 func TestListInstances(t *testing.T) {
 	test(t, func(router *chi.Mux, m *workflow.Manager, db *bun.DB) {
-
+		// Create a workflow with 10 instances
 		w := workflow.New(workflow.Config{})
 		_, err := db.NewInsert().Model(&w).Exec(context.TODO())
 		require.NoError(t, err)
@@ -34,8 +35,9 @@ func TestListInstances(t *testing.T) {
 		rec := httptest.NewRecorder()
 
 		router.ServeHTTP(rec, req)
-
 		require.Equal(t, http.StatusOK, rec.Result().StatusCode)
+
+		// Retrieve only running instances
 		instances := make([]workflow.Instance, 0)
 		apitesting.ReadResponse(t, rec, &instances)
 		require.Len(t, instances, 10)
@@ -48,5 +50,25 @@ func TestListInstances(t *testing.T) {
 		require.Equal(t, http.StatusOK, rec.Result().StatusCode)
 		apitesting.ReadResponse(t, rec, &instances)
 		require.Len(t, instances, 6)
+
+		// Delete the workflow
+		req = httptest.NewRequest(http.MethodDelete, fmt.Sprintf("/workflows/%s/", w.ID), nil)
+		rec = httptest.NewRecorder()
+
+		router.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusNoContent, rec.Result().StatusCode)
+
+		// Try to retrieve instances for the deleted workflow
+		req = httptest.NewRequest(http.MethodGet, "/instances", nil)
+		rec = httptest.NewRecorder()
+
+		router.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusOK, rec.Result().StatusCode)
+		instances = make([]workflow.Instance, 0)
+		apitesting.ReadResponse(t, rec, &instances)
+		require.Len(t, instances, 0)
+
 	})
 }

components/orchestration/internal/api/router.go

@@ -28,6 +28,7 @@ func newRouter(m *workflow.Manager, info ServiceInfo, healthController *health.H
 			r.Get("/", listWorkflows(m))
 			r.Post("/", createWorkflow(m))
 			r.Route("/{workflowId}", func(r chi.Router) {
+				r.Delete("/", deleteWorkflow(m))
 				r.Get("/", readWorkflow(m))
 				r.Route("/instances", func(r chi.Router) {
 					r.Post("/", runWorkflow(m))

components/orchestration/internal/storage/migrations.go

@@ -69,5 +69,15 @@ func registerMigrations(migrator *migrations.Migrator) {
 				return nil
 			},
 		},
+		migrations.Migration{
+			Up: func(tx bun.Tx) error {
+				if _, err := tx.Exec(`
+					alter table "workflows" add column deleted_at timestamp default null;
+				`); err != nil {
+					return err
+				}
+				return nil
+			},
+		},
 	)
 }

components/orchestration/internal/storage/migrations/00004_add_deleted_on_workflowsgo

@@ -0,0 +1,29 @@
+package migrations
+
+import (
+	"database/sql"
+
+	"github.com/pressly/goose/v3"
+)
+
+func init() {
+	goose.AddMigration(upAddColumnDeletedAt, downRemoveColumnDeletedAt)
+}
+
+func upAddColumnDeletedAt(tx *sql.Tx) error {
+	if _, err := tx.Exec(`
+		alter table "workflows" add column deleted_at timestamp default null;
+	`); err != nil {
+		return err
+	}
+	return nil
+}
+
+func downRemoveColumnDeletedAt(tx *sql.Tx) error {
+	if _, err := tx.Exec(`
+		alter table "workflows" drop column deleted_at;
+	`); err != nil {
+		return err
+	}
+	return nil
+}

components/orchestration/internal/workflow/manager.go

@@ -17,6 +17,7 @@ import (
 
 var (
 	ErrInstanceNotFound = errors.New("Instance not found")
+	ErrWorkflowNotFound = errors.New("Workflow not found")
 )
 
 const (
@@ -51,6 +52,27 @@ func (m *Manager) Create(ctx context.Context, config Config) (*Workflow, error)
 	return &workflow, nil
 }
 
+func (m *Manager) DeleteWorkflow(ctx context.Context, id string) error {
+
+	var workflow Workflow
+
+	res, err := m.db.NewUpdate().Model(&workflow).Where("id = ?", id).Set("deleted_at = ?", time.Now()).Exec(ctx)
+
+	if err != nil {
+		return err
+	}
+
+	r, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if r == 0 {
+		return ErrWorkflowNotFound
+	}
+
+	return nil
+}
+
 func (m *Manager) RunWorkflow(ctx context.Context, id string, variables map[string]string) (Instance, error) {
 
 	workflow := Workflow{}
@@ -101,6 +123,7 @@ func (m *Manager) ListWorkflows(ctx context.Context) ([]Workflow, error) {
 	workflows := make([]Workflow, 0)
 	if err := m.db.NewSelect().
 		Model(&workflows).
+		Where("deleted_at IS NULL").
 		Scan(ctx); err != nil {
 		return nil, err
 	}
@@ -152,12 +175,16 @@ func (m *Manager) AbortRun(ctx context.Context, instanceID string) error {
 func (m *Manager) ListInstances(ctx context.Context, workflowID string, running bool) ([]Instance, error) {
 	instances := make([]Instance, 0)
 	query := m.db.NewSelect().Model(&instances)
+
+	query.Join("JOIN workflows ON workflows.id = u.workflow_id").Where("workflows.deleted_at IS NULL")
+
 	if workflowID != "" {
-		query = query.Where("workflow_id = ?", workflowID)
+		query = query.Where("workflows.workflow_id = ?", workflowID)
 	}
 	if running {
-		query = query.Where("terminated = false")
+		query = query.Where("u.terminated = false")
 	}
+
 	if err := query.Scan(ctx); err != nil {
 		return nil, errors.Wrap(err, "retrieving workflow")
 	}

components/orchestration/internal/workflow/workflow.go

@@ -9,10 +9,11 @@ import (
 
 type Workflow struct {
 	bun.BaseModel `bun:"table:workflows"`
-	ID            string    `json:"id" bun:",pk"`
-	Config        Config    `json:"config"`
-	CreatedAt     time.Time `json:"createdAt"`
-	UpdatedAt     time.Time `json:"updatedAt"`
+	ID            string     `json:"id" bun:",pk"`
+	Config        Config     `json:"config"`
+	CreatedAt     time.Time  `json:"createdAt"`
+	UpdatedAt     time.Time  `json:"updatedAt"`
+	DeletedAt     *time.Time `json:"deletedAt"`
 }
 
 func New(config Config) Workflow {

components/orchestration/openapi.yaml

@@ -81,7 +81,18 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/GetWorkflowResponse'
-
+    delete:
+      summary: Delete a flow by id
+      tags:
+      - Orchestration
+      description: Delete a flow by id
+      operationId: deleteWorkflow
+      responses:
+        default:
+          $ref: '#/components/responses/ErrorResponse'
+        204:
+          description: No content
+
   /workflows/{workflowID}/instances:
     parameters:
     - in: path