Reintroduce fixity

Author: sevein

config.json.example

@@ -84,5 +84,16 @@
         }
       ]
     }
+  },
+
+  // ===========================================================================
+  // WORKFLOW BEHAVIOUR
+  // ---------------------------------------------------------------------------
+  // Toggle additional workflow steps. When unset, all options default to false.
+  "workflows": {
+    "move": {
+      // Run a fixity check via the Storage Service before moving an AIP.
+      "check_fixity": false
+    }
   }
 }

internal/application/config.go

@@ -174,6 +174,9 @@ type Config struct {
 
 	// Storage Service configuration.
 	StorageService StorageServiceConfig `json:"storage_service"`
+
+	// Workflow-level configuration toggles.
+	Workflows WorkflowConfig `json:"workflows"`
 }
 
 type TemporalConfig struct {
@@ -245,6 +248,16 @@ type ReplicationTarget struct {
 	Name string `json:"name"`
 }
 
+// WorkflowConfig holds configuration for individual workflows.
+type WorkflowConfig struct {
+	Move WorkflowMoveConfig `json:"move"`
+}
+
+// WorkflowMoveConfig controls behaviour specific to the move workflow.
+type WorkflowMoveConfig struct {
+	CheckFixity bool `json:"check_fixity"`
+}
+
 type DatabaseConfig struct {
 	Engine string       `json:"engine"`
 	SQLite SQLiteConfig `json:"sqlite"`

internal/application/config_test.go

@@ -55,4 +55,6 @@ func TestLoadConfig(t *testing.T) {
 
 	assert.Equal(t, cfg.Database.Engine, "sqlite")
 	assert.Equal(t, cfg.Database.SQLite.Path, DefaultSQLitePath())
+
+	assert.Assert(t, !cfg.Workflows.Move.CheckFixity)
 }

internal/application/events.go

@@ -21,6 +21,7 @@ func (a Action) String() string {
 
 var (
 	ActionFind      = Action{"find"}
+	ActionFixity    = Action{"fixity"}
 	ActionMove      = Action{"move"}
 	ActionReplicate = Action{"Replicate"}
 	ActionIndex     = Action{"index"}
@@ -57,6 +58,7 @@ func StartEvent(a Action) Event {
 	}
 }
 
+// EndEvent marks the AIP with the provided status and stores the event.
 func EndEvent(ctx context.Context, s AIPStatus, a *App, e Event, aip *models.Aip) error {
 	e.End = time.Now()
 	if err := a.UpdateAIPStatus(ctx, aip.ID, s); err != nil {
@@ -69,13 +71,15 @@ func EndEvent(ctx context.Context, s AIPStatus, a *App, e Event, aip *models.Aip
 	return aip.InsertEvents(ctx, a.DB, setter)
 }
 
+// EndEventNoChange stores the event without changing the AIP status.
 func EndEventNoChange(ctx context.Context, a *App, e Event, aip *models.Aip) error {
 	if err := aip.Reload(ctx, a.DB); err != nil {
 		return err
 	}
 	return EndEvent(ctx, AIPStatus(aip.Status), a, e, aip)
 }
 
+// EndEventErr records the error, marks the AIP as failed, and stores the event.
 func EndEventErr(ctx context.Context, a *App, e Event, aip *models.Aip, eventErr string) error {
 	e.End = time.Now()
 	a.AddAIPError(ctx, aip, eventErr)
@@ -89,6 +93,17 @@ func EndEventErr(ctx context.Context, a *App, e Event, aip *models.Aip, eventErr
 	return aip.InsertEvents(ctx, a.DB, setter)
 }
 
+// EndEventErrNoFailure records the error and stores the event without changing the AIP status.
+func EndEventErrNoFailure(ctx context.Context, a *App, e Event, aip *models.Aip, eventErr string) error {
+	e.End = time.Now()
+	a.AddAIPError(ctx, aip, eventErr)
+	setter, err := EventToSetter(e)
+	if err != nil {
+		return err
+	}
+	return aip.InsertEvents(ctx, a.DB, setter)
+}
+
 func EventToSetter(e Event) (*models.EventSetter, error) {
 	formatDetails, err := e.FormatDetails()
 	if err != nil {

internal/application/fixity.go

@@ -0,0 +1,92 @@
+package application
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/artefactual-labs/migrate/internal/database/gen/models"
+	"github.com/artefactual-labs/migrate/internal/storage_service"
+)
+
+const FixityActivityName = "fixity-activity"
+
+type FixityActivityParams struct {
+	UUID string `json:"uuid"`
+}
+
+type FixityActivityResult struct {
+	Status string
+}
+
+func (a *App) FixityA(ctx context.Context, params FixityActivityParams) (*FixityActivityResult, error) {
+	aip, err := a.GetAIPByID(ctx, params.UUID)
+	if err != nil {
+		return nil, err
+	}
+
+	if aip.FixityRun {
+		return &FixityActivityResult{Status: aip.Status}, nil
+	}
+
+	if err := checkFixity(ctx, a, a.StorageClient, aip); err != nil {
+		return nil, err
+	}
+
+	if err := aip.Reload(ctx, a.DB); err != nil {
+		return nil, err
+	}
+
+	if aip.Status != string(AIPStatusFixityChecked) {
+		return &FixityActivityResult{Status: aip.Status}, fmt.Errorf("fixity did not complete successfully, status: %s", aip.Status)
+	}
+
+	return &FixityActivityResult{Status: aip.Status}, nil
+}
+
+func checkFixity(ctx context.Context, a *App, storageClient *storage_service.API, aip *models.Aip) error {
+	if aip.FixityRun {
+		return nil
+	}
+
+	e := StartEvent(ActionFixity)
+	e.AddDetail(fmt.Sprintf("Running fixity for: %s", aip.UUID))
+
+	res, err := storageClient.Packages.CheckFixity(ctx, aip.UUID)
+	if err != nil {
+		var ssErr storage_service.SSError
+		if errors.As(err, &ssErr) && ssErr.StatusCode >= 500 {
+			if eventErr := EndEventErrNoFailure(ctx, a, e, aip, err.Error()); eventErr != nil {
+				return eventErr
+			}
+			return fmt.Errorf("storage service fixity call failed: %w", err)
+		}
+		if eventErr := EndEventErrNoFailure(ctx, a, e, aip, err.Error()); eventErr != nil {
+			return eventErr
+		}
+		return nil
+	}
+
+	if res.Success {
+		if err := EndEvent(ctx, AIPStatusFixityChecked, a, e, aip); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	a.AddAIPError(ctx, aip, res.Message)
+	for _, f := range res.Failures.Files.Changed {
+		e.AddDetail("file changed: " + f)
+	}
+	for _, f := range res.Failures.Files.Untracked {
+		e.AddDetail("file untracked: " + f)
+	}
+	for _, f := range res.Failures.Files.Missing {
+		e.AddDetail("file missing: " + f)
+	}
+	if err := EndEventErr(ctx, a, e, aip, res.Message); err != nil {
+		return err
+	}
+
+	return nil
+}

internal/application/move_workflow.go

@@ -65,6 +65,16 @@ func (w *MoveWorkflow) Run(ctx workflow.Context, params MoveWorkflowParams) (*Mo
 		return result, nil
 	}
 
+	if w.App.Config.Workflows.Move.CheckFixity {
+		fixityParams := FixityActivityParams{UUID: params.UUID.String()}
+		fixityResult := FixityActivityResult{}
+		err = workflow.ExecuteActivity(ctx, FixityActivityName, fixityParams).Get(ctx, &fixityResult)
+		if err != nil {
+			return nil, err
+		}
+		result.MoveDetails = append(result.MoveDetails, "Fixity status: "+fixityResult.Status)
+	}
+
 	moveParams := MoveActivityParams{UUID: params.UUID.String()}
 	moveResult := MoveActivityResult{}
 	err = workflow.ExecuteActivity(ctx, MoveActivityName, moveParams).Get(ctx, &moveResult)

internal/cmd/workercmd/workercmd.go

@@ -78,6 +78,7 @@ func registerWorker(app *application.App) worker.Worker {
 	w.RegisterActivityWithOptions(app.ReplicateA, activity.RegisterOptions{Name: application.ReplicateAName})
 	w.RegisterActivityWithOptions(app.FindA, activity.RegisterOptions{Name: application.FindAName})
 	w.RegisterActivityWithOptions(app.CheckReplicationStatus, activity.RegisterOptions{Name: application.CheckReplicationStatusName})
+	w.RegisterActivityWithOptions(app.FixityA, activity.RegisterOptions{Name: application.FixityActivityName})
 	w.RegisterActivityWithOptions(app.MoveA, activity.RegisterOptions{Name: application.MoveActivityName})
 
 	return w

internal/ssmock/server.go

@@ -215,6 +215,19 @@ func (s *Server) handleFile(w http.ResponseWriter, r *http.Request) {
 		http.NotFound(w, r)
 		return
 	}
+	if strings.HasSuffix(remainder, "/check_fixity/") {
+		if r.Method != http.MethodGet {
+			methodNotAllowed(w, http.MethodGet)
+			return
+		}
+		id := strings.TrimSuffix(remainder, "/check_fixity/")
+		if id == "" {
+			http.NotFound(w, r)
+			return
+		}
+		s.handleCheckFixity(w, r, id)
+		return
+	}
 	if strings.HasSuffix(remainder, "/move/") {
 		if r.Method != http.MethodPost {
 			methodNotAllowed(w, http.MethodPost)
@@ -248,6 +261,23 @@ func (s *Server) handleFile(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, pkg)
 }
 
+func (s *Server) handleCheckFixity(w http.ResponseWriter, r *http.Request, id string) {
+	s.mu.RLock()
+	pkg, ok := s.state.clonePackage(id)
+	s.mu.RUnlock()
+	if !ok {
+		http.NotFound(w, r)
+		return
+	}
+
+	resp := storage_service.FixityResponse{
+		Success:   true,
+		Message:   fmt.Sprintf("Fixity check completed for %s", pkg.UUID),
+		Timestamp: time.Now().UTC().Format(time.RFC3339),
+	}
+	writeJSON(w, &resp)
+}
+
 func (s *Server) handleMove(w http.ResponseWriter, r *http.Request, id string) {
 	if err := r.ParseForm(); err != nil {
 		http.Error(w, fmt.Sprintf("invalid form: %v", err), http.StatusBadRequest)

internal/storage_service/package.go

@@ -41,3 +41,25 @@ func (s *PackageService) Move(ctx context.Context, packageID, locationID string)
 	p.Set("location_uuid", locationID)
 	return s.client.Call(ctx, http.MethodPost, path, p.Encode(), nil)
 }
+
+type FixityResponse struct {
+	Success   bool           `json:"success"`
+	Message   string         `json:"message"`
+	Timestamp string         `json:"timestamp"`
+	Failures  FixityFailures `json:"failures"`
+}
+
+type FixityFailures struct {
+	Files struct {
+		Missing   []string `json:"missing"`
+		Changed   []string `json:"changed"`
+		Untracked []string `json:"untracked"`
+	} `json:"files"`
+}
+
+func (s *PackageService) CheckFixity(ctx context.Context, id string) (*FixityResponse, error) {
+	res := &FixityResponse{}
+	path := fmt.Sprintf("/api/v2/file/%s/check_fixity/", id)
+	err := s.client.Call(ctx, http.MethodGet, path, nil, res)
+	return res, err
+}

testdata/move.txtar

@@ -26,6 +26,11 @@ cmp stdout ssmock-snapshot.toml
       "move_target_location_id": "71cb2196-5629-4225-aaf7-d8431b0895c4",
       "replication_targets": []
     }
+  },
+  "workflows": {
+    "move": {
+      "check_fixity": true
+    }
   }
 }
 -- ssmock.toml --
@@ -47,7 +52,7 @@ id = '2faa61dc-ed33-49f4-8b36-954f203bab4a'
 status = 'UPLOADED'
 -- move-report.expected.csv --
 UUID,AIPStatus,Duration,fixity-run,moved,cleaned,replicated,re-indexed,size,Duration Nanoseconds,New Path,Old Path,Replica UUID,Local copy Path,Staged Copy Path,Errors
-2faa61dc-ed33-49f4-8b36-954f203bab4a,moved,Not Done,Done,Not Done,Not Done,Not Done,0 B,
+2faa61dc-ed33-49f4-8b36-954f203bab4a,moved,Done,Done,Not Done,Not Done,Not Done,0 B,
 -- db.aips.expected.csv --
 id,uuid,status,found,fixity_run,moved,cleaned,replicated,re_indexed,current_location,size,location_uuid
-1,2faa61dc-ed33-49f4-8b36-954f203bab4a,moved,1,0,1,0,0,0,/api/v2/location/71cb2196-5629-4225-aaf7-d8431b0895c4/|/api/v2/location/72a9c518-2747-4cb5-aeba-e6309d946e79/,0,
+1,2faa61dc-ed33-49f4-8b36-954f203bab4a,moved,1,1,1,0,0,0,/api/v2/location/71cb2196-5629-4225-aaf7-d8431b0895c4/|/api/v2/location/72a9c518-2747-4cb5-aeba-e6309d946e79/,0,