Add workflow skeleton and request from dashboard

- Create and register basic deletion workflow.
- Trigger workflow from API endpoint.
- Add button and API call to dashboard.

Author: jraddaoui

cmd/enduro/main.go

@@ -453,6 +453,10 @@ func main() {
 			temporalsdk_activity.RegisterOptions{Name: activities.DisposeOriginalActivityName},
 		)
 
+		w.RegisterWorkflowWithOptions(
+			storage_workflows.NewStorageDeleteWorkflow(storagesvc).Execute,
+			temporalsdk_workflow.RegisterOptions{Name: storage.StorageDeleteWorkflowName},
+		)
 		w.RegisterWorkflowWithOptions(
 			storage_workflows.NewStorageUploadWorkflow().Execute,
 			temporalsdk_workflow.RegisterOptions{Name: storage.StorageUploadWorkflowName},

dashboard/src/components/AipLocationCard.vue

@@ -80,6 +80,17 @@ watch(aipStore.ui.download, () => download());
           </template>
           <template v-else>Move</template>
         </button>
+        <button
+          v-if="
+            aipStore.isStored &&
+            authStore.checkAttributes(['storage:aips:deletion:request'])
+          "
+          type="button"
+          class="btn btn-primary btn-sm"
+          @click="aipStore.requestDeletion"
+        >
+          Delete
+        </button>
       </div>
     </div>
   </div>

dashboard/src/stores/aip.ts

@@ -173,6 +173,27 @@ export const useAipStore = defineStore("aip", {
       }
       this.locationChanging = !resp.done;
     },
+    async requestDeletion() {
+      if (!this.current) return;
+      // TODO:
+      // - Add dialog and get reason from it.
+      // - Improve error reporting.
+      try {
+        await client.storage.storageRequestAipDeletion({
+          uuid: this.current.uuid,
+          requestAipDeletionRequestBody: {
+            reason: "Requesting deletion",
+          },
+        });
+      } catch (error) {
+        return error;
+      }
+      this.$patch((state) => {
+        // TODO: Use proper status.
+        if (state.current)
+          state.current.status = api.EnduroStorageAipStatusEnum.InReview;
+      });
+    },
     nextPage() {
       if (this.hasNextPage) {
         this.fetchAips(this.pager.current + 1);

internal/storage/deletion_request.go

@@ -2,14 +2,41 @@ package storage
 
 import (
 	"context"
+	"errors"
+
+	"github.com/google/uuid"
 
 	goastorage "github.com/artefactual-sdps/enduro/internal/api/gen/storage"
 )
 
-func (svc *serviceImpl) RequestAipDeletion(ctx context.Context, payload *goastorage.RequestAipDeletionPayload) error {
+func (s *serviceImpl) RequestAipDeletion(ctx context.Context, payload *goastorage.RequestAipDeletionPayload) error {
+	aipID, err := uuid.Parse(payload.UUID)
+	if err != nil {
+		return goastorage.MakeNotValid(errors.New("invalid UUID"))
+	}
+	if payload.Reason == "" {
+		return goastorage.MakeNotValid(errors.New("invalid reason"))
+	}
+
+	s.logger.Info("HERE WE GO!!", "UUID", payload.UUID, "Reason", payload.Reason)
+
+	// TODO:
+	// - Check AIP existence and status, same as in workflow.
+	// - Get user details from context claim and include them in the request.
+
+	_, err = InitStorageDeleteWorkflow(ctx, s.tc, &StorageDeleteWorkflowRequest{
+		AIPID:     aipID,
+		Reason:    payload.Reason,
+		TaskQueue: s.config.TaskQueue,
+	})
+	if err != nil {
+		s.logger.Error(err, "error initializing delete workflow")
+		return goastorage.MakeNotAvailable(errors.New("cannot perform operation"))
+	}
+
 	return nil
 }
 
-func (svc *serviceImpl) ReviewAipDeletion(ctx context.Context, payload *goastorage.ReviewAipDeletionPayload) error {
+func (s *serviceImpl) ReviewAipDeletion(ctx context.Context, payload *goastorage.ReviewAipDeletionPayload) error {
 	return nil
 }

internal/storage/workflow.go

@@ -13,11 +13,21 @@ import (
 const (
 	CopyToPermanentLocationActivityName = "copy-to-permanent-location-activity"
 	DeleteFromLocationActivityName      = "delete-from-location-activity"
+	StorageDeleteWorkflowName           = "storage-delete-workflow"
 	StorageUploadWorkflowName           = "storage-upload-workflow"
 	StorageMoveWorkflowName             = "storage-move-workflow"
 	UploadDoneSignalName                = "upload-done-signal"
 )
 
+type StorageDeleteWorkflowRequest struct {
+	AIPID     uuid.UUID
+	Reason    string
+	UserEmail string
+	UserSub   string
+	UserISS   string
+	TaskQueue string
+}
+
 type StorageUploadWorkflowRequest struct {
 	AIPID     uuid.UUID
 	TaskQueue string
@@ -36,6 +46,22 @@ type CopyToPermanentLocationActivityParams struct {
 
 type UploadDoneSignal struct{}
 
+func InitStorageDeleteWorkflow(
+	ctx context.Context,
+	tc temporalsdk_client.Client,
+	req *StorageDeleteWorkflowRequest,
+) (temporalsdk_client.WorkflowRun, error) {
+	ctx, cancel := context.WithTimeout(ctx, time.Second*5)
+	defer cancel()
+
+	opts := temporalsdk_client.StartWorkflowOptions{
+		ID:                    fmt.Sprintf("%s-%s", StorageDeleteWorkflowName, req.AIPID),
+		TaskQueue:             req.TaskQueue,
+		WorkflowIDReusePolicy: temporalsdk_api_enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY,
+	}
+	return tc.ExecuteWorkflow(ctx, opts, StorageDeleteWorkflowName, req)
+}
+
 func InitStorageUploadWorkflow(
 	ctx context.Context,
 	tc temporalsdk_client.Client,

internal/storage/workflows/delete.go

@@ -0,0 +1,88 @@
+package workflows
+
+import (
+	"errors"
+
+	temporalsdk_workflow "go.temporal.io/sdk/workflow"
+
+	"github.com/artefactual-sdps/enduro/internal/storage"
+	"github.com/artefactual-sdps/enduro/internal/storage/enums"
+)
+
+type StorageDeleteWorkflow struct {
+	storagesvc storage.Service
+}
+
+func NewStorageDeleteWorkflow(storagesvc storage.Service) *StorageDeleteWorkflow {
+	return &StorageDeleteWorkflow{storagesvc: storagesvc}
+}
+
+func (w *StorageDeleteWorkflow) Execute(
+	ctx temporalsdk_workflow.Context,
+	req storage.StorageDeleteWorkflowRequest,
+) (e error) {
+	// TODO: Check AIP existence and status, fail workflow.
+
+	// Set AIP status to processing.
+	// TODO: Update AIP status enum and use proper status.
+	if err := updateAIPStatus(ctx, w.storagesvc, req.AIPID, enums.AIPStatusInReview); err != nil {
+		return err
+	}
+
+	// Create persistence workflow.
+	workflowDBID, err := createWorkflow(ctx, w.storagesvc, req.AIPID, enums.WorkflowTypeDeleteAip)
+	if err != nil {
+		return err
+	}
+
+	// Complete persistence workflow.
+	defer func() {
+		// TODO: Consider rejected/cancelled case.
+		status := enums.WorkflowStatusDone
+		if e != nil {
+			status = enums.WorkflowStatusError
+		}
+
+		err := completeWorkflow(ctx, w.storagesvc, workflowDBID, status)
+		if err != nil {
+			e = errors.Join(e, err)
+		}
+	}()
+
+	// Create review task.
+	reviewTaskID, err := createTask(
+		ctx,
+		w.storagesvc,
+		workflowDBID,
+		"Review AIP deletion request",
+		"Awaiting user decision",
+	)
+	if err != nil {
+		return err
+	}
+
+	// TODO:
+	// - Create DeletionRequest.
+	// - Add signal channel, etc.
+	// - Update DeletionRequest.
+
+	// Complete review task.
+	taskNote := "AIP deletion request approved"
+	if false {
+		taskNote = "AIP deletion request rejected"
+	}
+	taskErr := completeTask(ctx, w.storagesvc, reviewTaskID, enums.TaskStatusDone, taskNote)
+	if err = errors.Join(err, taskErr); err != nil {
+		return err
+	}
+
+	// TODO: Delete the AIP from AMSS or MinIO location source.
+
+	// Set AIP status to deleted.
+	// TODO: Update AIP status enum and use proper status.
+	if err := updateAIPStatus(ctx, w.storagesvc, req.AIPID, enums.AIPStatusRejected); err != nil {
+		return err
+	}
+
+	return nil
+}

internal/storage/workflows/move.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/google/uuid"
 	temporalsdk_temporal "go.temporal.io/sdk/temporal"
 	temporalsdk_workflow "go.temporal.io/sdk/workflow"
 
@@ -29,7 +28,7 @@ func (w *StorageMoveWorkflow) Execute(
 ) (e error) {
 	// Set AIP status to moving.
 	{
-		if err := w.updateAIPStatus(ctx, enums.AIPStatusMoving, req.AIPID); err != nil {
+		if err := updateAIPStatus(ctx, w.storagesvc, req.AIPID, enums.AIPStatusMoving); err != nil {
 			return err
 		}
 	}
@@ -145,34 +144,10 @@ func (w *StorageMoveWorkflow) Execute(
 
 	// Set AIP status to stored.
 	{
-		if err := w.updateAIPStatus(ctx, enums.AIPStatusStored, req.AIPID); err != nil {
+		if err := updateAIPStatus(ctx, w.storagesvc, req.AIPID, enums.AIPStatusStored); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
-
-func (w *StorageMoveWorkflow) updateAIPStatus(
-	ctx temporalsdk_workflow.Context,
-	st enums.AIPStatus,
-	aipID uuid.UUID,
-) error {
-	activityOpts := temporalsdk_workflow.WithLocalActivityOptions(ctx, temporalsdk_workflow.LocalActivityOptions{
-		ScheduleToCloseTimeout: 5 * time.Second,
-		RetryPolicy: &temporalsdk_temporal.RetryPolicy{
-			InitialInterval:    time.Second,
-			BackoffCoefficient: 2,
-			MaximumInterval:    time.Minute,
-			MaximumAttempts:    3,
-		},
-	})
-
-	params := &storage.UpdateAIPStatusLocalActivityParams{
-		AIPID:  aipID,
-		Status: st,
-	}
-
-	return temporalsdk_workflow.ExecuteLocalActivity(activityOpts, storage.UpdateAIPStatusLocalActivity, w.storagesvc, params).
-		Get(activityOpts, nil)
-}

internal/storage/workflows/workflows.go

@@ -23,6 +23,24 @@ func localActivityOptions(ctx temporalsdk_workflow.Context) temporalsdk_workflow
 	})
 }
 
+func updateAIPStatus(
+	ctx temporalsdk_workflow.Context,
+	storagesvc storage.Service,
+	aipID uuid.UUID,
+	s enums.AIPStatus,
+) error {
+	activityOpts := localActivityOptions(ctx)
+	return temporalsdk_workflow.ExecuteLocalActivity(
+		activityOpts,
+		storage.UpdateAIPStatusLocalActivity,
+		storagesvc,
+		&storage.UpdateAIPStatusLocalActivityParams{
+			AIPID:  aipID,
+			Status: s,
+		},
+	).Get(activityOpts, nil)
+}
+
 func createWorkflow(
 	ctx temporalsdk_workflow.Context,
 	storagesvc storage.Service,