Add result struct to activity function signature

internal/am/delete_transfer.go

@@ -20,19 +20,21 @@ type DeleteTransferActivity struct {
 	logger logr.Logger
 }
 
+type DeleteTransferActivityResult struct{}
+
 func NewDeleteTransferActivity(logger logr.Logger, client sftp.Client) *DeleteTransferActivity {
 	return &DeleteTransferActivity{client: client, logger: logger}
 }
 
-func (a *DeleteTransferActivity) Execute(ctx context.Context, params *DeleteTransferActivityParams) error {
+func (a *DeleteTransferActivity) Execute(ctx context.Context, params *DeleteTransferActivityParams) (*DeleteTransferActivityResult, error) {
 	a.logger.V(1).Info("Execute DeleteTransferActivity",
 		"destination", params.Destination,
 	)
 
 	err := a.client.Delete(ctx, params.Destination)
 	if err != nil {
-		return fmt.Errorf("delete transfer: path: %q: %v", params.Destination, err)
+		return &DeleteTransferActivityResult{}, fmt.Errorf("delete transfer: path: %q: %v", params.Destination, err)
 	}
 
-	return nil
+	return &DeleteTransferActivityResult{}, nil
 }

internal/storage/activities/copy.go

@@ -11,47 +11,49 @@ type CopyToPermanentLocationActivity struct {
 	storagesvc storage.Service
 }
 
+type CopyToPermanentLocationActivityResult struct{}
+
 func NewCopyToPermanentLocationActivity(storagesvc storage.Service) *CopyToPermanentLocationActivity {
 	return &CopyToPermanentLocationActivity{storagesvc: storagesvc}
 }
 
-func (a *CopyToPermanentLocationActivity) Execute(ctx context.Context, params *storage.CopyToPermanentLocationActivityParams) error {
+func (a *CopyToPermanentLocationActivity) Execute(ctx context.Context, params *storage.CopyToPermanentLocationActivityParams) (*CopyToPermanentLocationActivityResult, error) {
 	p, err := a.storagesvc.ReadPackage(ctx, params.AIPID)
 	if err != nil {
-		return err
+		return &CopyToPermanentLocationActivityResult{}, err
 	}
 
 	reader, err := a.storagesvc.PackageReader(ctx, p)
 	if err != nil {
-		return err
+		return &CopyToPermanentLocationActivityResult{}, err
 	}
 	defer reader.Close()
 
 	l, err := a.storagesvc.Location(ctx, params.LocationID)
 	if err != nil {
-		return err
+		return &CopyToPermanentLocationActivityResult{}, err
 	}
 
 	bucket, err := l.OpenBucket(ctx)
 	if err != nil {
-		return err
+		return &CopyToPermanentLocationActivityResult{}, err
 	}
 	defer bucket.Close()
 
 	writer, err := bucket.NewWriter(ctx, params.AIPID.String(), nil)
 	if err != nil {
-		return err
+		return &CopyToPermanentLocationActivityResult{}, err
 	}
 
 	_, copyErr := io.Copy(writer, reader)
 	closeErr := writer.Close()
 
 	if copyErr != nil {
-		return copyErr
+		return &CopyToPermanentLocationActivityResult{}, copyErr
 	}
 	if closeErr != nil {
-		return closeErr
+		return &CopyToPermanentLocationActivityResult{}, closeErr
 	}
 
-	return nil
+	return &CopyToPermanentLocationActivityResult{}, nil
 }

internal/storage/workflows/move_test.go

@@ -33,7 +33,7 @@ func TestStorageMoveWorkflow(t *testing.T) {
 
 	// Worker activities
 	env.RegisterActivityWithOptions(activities.NewCopyToPermanentLocationActivity(storagesvc).Execute, temporalsdk_activity.RegisterOptions{Name: storage.CopyToPermanentLocationActivityName})
-	env.OnActivity(storage.CopyToPermanentLocationActivityName, mock.Anything, mock.Anything).Return(nil)
+	env.OnActivity(storage.CopyToPermanentLocationActivityName, mock.Anything, mock.Anything).Return(nil, nil)
 
 	env.ExecuteWorkflow(
 		NewStorageMoveWorkflow(storagesvc).Execute,

internal/workflow/activities/cleanup.go

@@ -17,14 +17,16 @@ type CleanUpActivityParams struct {
 	FullPath string
 }
 
-func (a *CleanUpActivity) Execute(ctx context.Context, params *CleanUpActivityParams) error {
+type CleanUpActivityResult struct{}
+
+func (a *CleanUpActivity) Execute(ctx context.Context, params *CleanUpActivityParams) (*CleanUpActivityResult, error) {
 	if params == nil || params.FullPath == "" {
-		return fmt.Errorf("error processing parameters: missing or empty")
+		return &CleanUpActivityResult{}, fmt.Errorf("error processing parameters: missing or empty")
 	}
 
 	if err := os.RemoveAll(params.FullPath); err != nil {
-		return fmt.Errorf("error removing transfer directory: %v", err)
+		return &CleanUpActivityResult{}, fmt.Errorf("error removing transfer directory: %v", err)
 	}
 
-	return nil
+	return &CleanUpActivityResult{}, nil
 }

internal/workflow/activities/delete_original.go

@@ -10,10 +10,12 @@ type DeleteOriginalActivity struct {
 	wsvc watcher.Service
 }
 
+type DeleteOriginalActivityResult struct{}
+
 func NewDeleteOriginalActivity(wsvc watcher.Service) *DeleteOriginalActivity {
 	return &DeleteOriginalActivity{wsvc: wsvc}
 }
 
-func (a *DeleteOriginalActivity) Execute(ctx context.Context, watcherName, key string) error {
-	return a.wsvc.Delete(ctx, watcherName, key)
+func (a *DeleteOriginalActivity) Execute(ctx context.Context, watcherName, key string) (*DeleteOriginalActivityResult, error) {
+	return &DeleteOriginalActivityResult{}, a.wsvc.Delete(ctx, watcherName, key)
 }

internal/workflow/activities/dispose_original.go

@@ -10,10 +10,12 @@ type DisposeOriginalActivity struct {
 	wsvc watcher.Service
 }
 
+type DisposeOriginalActivityResult struct{}
+
 func NewDisposeOriginalActivity(wsvc watcher.Service) *DisposeOriginalActivity {
 	return &DisposeOriginalActivity{wsvc: wsvc}
 }
 
-func (a *DisposeOriginalActivity) Execute(ctx context.Context, watcherName, completedDir, key string) error {
-	return a.wsvc.Dispose(ctx, watcherName, key)
+func (a *DisposeOriginalActivity) Execute(ctx context.Context, watcherName, completedDir, key string) (*DisposeOriginalActivityResult, error) {
+	return &DisposeOriginalActivityResult{}, a.wsvc.Dispose(ctx, watcherName, key)
 }

internal/workflow/activities/download.go

@@ -43,16 +43,17 @@ func (a *DownloadActivity) Execute(ctx context.Context, params *DownloadActivity
 
 	destDir, err := os.MkdirTemp("", "enduro")
 	if err != nil {
-		return nil, temporal_tools.NewNonRetryableError(fmt.Errorf("make temp dir: %v", err))
+		return &DownloadActivityResult{}, temporal_tools.NewNonRetryableError(fmt.Errorf("make temp dir: %v", err))
 	}
 
 	dest := filepath.Clean(filepath.Join(destDir, params.Key))
 
 	ctx, span := trace.SpanFromContext(ctx).TracerProvider().Tracer(DownloadActivityName).Start(ctx, "download")
 	if err := a.wsvc.Download(ctx, dest, params.WatcherName, params.Key); err != nil {
+
 		span.RecordError(err)
 		span.End()
-		return nil, temporal_tools.NewNonRetryableError(fmt.Errorf("download: %v", err))
+		return &DownloadActivityResult{}, temporal_tools.NewNonRetryableError(fmt.Errorf("download: %v", err))
 	}
 	span.End()
 

internal/workflow/activities/storage.go

@@ -16,6 +16,8 @@ type MoveToPermanentStorageActivityParams struct {
 	LocationID uuid.UUID
 }
 
+type MoveToPermanentStorageActivityResult struct{}
+
 type MoveToPermanentStorageActivity struct {
 	storageClient *goastorage.Client
 }
@@ -26,7 +28,7 @@ func NewMoveToPermanentStorageActivity(storageClient *goastorage.Client) *MoveTo
 	}
 }
 
-func (a *MoveToPermanentStorageActivity) Execute(ctx context.Context, params *MoveToPermanentStorageActivityParams) error {
+func (a *MoveToPermanentStorageActivity) Execute(ctx context.Context, params *MoveToPermanentStorageActivityParams) (*MoveToPermanentStorageActivityResult, error) {
 	childCtx, cancel := context.WithTimeout(ctx, time.Second*5)
 	defer cancel()
 
@@ -35,7 +37,7 @@ func (a *MoveToPermanentStorageActivity) Execute(ctx context.Context, params *Mo
 		LocationID: params.LocationID,
 	})
 
-	return err
+	return &MoveToPermanentStorageActivityResult{}, err
 }
 
 type PollMoveToPermanentStorageActivityParams struct {
@@ -46,13 +48,15 @@ type PollMoveToPermanentStorageActivity struct {
 	storageClient *goastorage.Client
 }
 
+type PollMoveToPermanentStorageActivityResult struct{}
+
 func NewPollMoveToPermanentStorageActivity(storageClient *goastorage.Client) *PollMoveToPermanentStorageActivity {
 	return &PollMoveToPermanentStorageActivity{
 		storageClient: storageClient,
 	}
 }
 
-func (a *PollMoveToPermanentStorageActivity) Execute(ctx context.Context, params *PollMoveToPermanentStorageActivityParams) error {
+func (a *PollMoveToPermanentStorageActivity) Execute(ctx context.Context, params *PollMoveToPermanentStorageActivityParams) (*PollMoveToPermanentStorageActivityResult, error) {
 	var g run.Group
 
 	{
@@ -105,7 +109,7 @@ func (a *PollMoveToPermanentStorageActivity) Execute(ctx context.Context, params
 	}
 
 	err := g.Run()
-	return err
+	return &PollMoveToPermanentStorageActivityResult{}, err
 }
 
 type RejectPackageActivityParams struct {
@@ -116,19 +120,21 @@ type RejectPackageActivity struct {
 	storageClient *goastorage.Client
 }
 
+type RejectPackageActivityResult struct{}
+
 func NewRejectPackageActivity(storageClient *goastorage.Client) *RejectPackageActivity {
 	return &RejectPackageActivity{
 		storageClient: storageClient,
 	}
 }
 
-func (a *RejectPackageActivity) Execute(ctx context.Context, params *RejectPackageActivityParams) error {
+func (a *RejectPackageActivity) Execute(ctx context.Context, params *RejectPackageActivityParams) (*RejectPackageActivityResult, error) {
 	childCtx, cancel := context.WithTimeout(ctx, time.Second*5)
 	defer cancel()
 
 	err := a.storageClient.Reject(childCtx, &goastorage.RejectPayload{
 		AipID: params.AIPID,
 	})
 
-	return err
+	return &RejectPackageActivityResult{}, err
 }

internal/workflow/activities/upload.go

@@ -20,57 +20,59 @@ type UploadActivity struct {
 	storageClient *goastorage.Client
 }
 
+type UploadActivityResult struct{}
+
 func NewUploadActivity(storageClient *goastorage.Client) *UploadActivity {
 	return &UploadActivity{
 		storageClient: storageClient,
 	}
 }
 
-func (a *UploadActivity) Execute(ctx context.Context, params *UploadActivityParams) error {
+func (a *UploadActivity) Execute(ctx context.Context, params *UploadActivityParams) (*UploadActivityResult, error) {
 	childCtx, cancel := context.WithTimeout(ctx, time.Second*5)
 	defer cancel()
 	res, err := a.storageClient.Submit(childCtx, &goastorage.SubmitPayload{
 		AipID: params.AIPID,
 		Name:  params.Name,
 	})
 	if err != nil {
-		return err
+		return &UploadActivityResult{}, err
 	}
 
 	// Upload to MinIO using the upload pre-signed URL.
 	{
 		f, err := os.Open(params.AIPPath)
 		if err != nil {
-			return err
+			return &UploadActivityResult{}, err
 		}
 		defer f.Close() //#nosec G307 -- Errors returned by Close() here do not require specific handling.
 
 		uploadReq, err := http.NewRequestWithContext(ctx, http.MethodPut, res.URL, f)
 		if err != nil {
-			return nil
+			return &UploadActivityResult{}, nil
 		}
 		fi, err := f.Stat()
 		if err != nil {
-			return err
+			return &UploadActivityResult{}, err
 		}
 		uploadReq.ContentLength = fi.Size()
 		if err != nil {
-			return err
+			return &UploadActivityResult{}, err
 		}
 
 		minioClient := &http.Client{}
 		resp, err := minioClient.Do(uploadReq)
 		if err != nil {
-			return err
+			return &UploadActivityResult{}, err
 		}
 		if resp.StatusCode != http.StatusOK {
-			return errors.New("unexpected status code returned")
+			return &UploadActivityResult{}, errors.New("unexpected status code returned")
 		}
 	}
 
 	childCtx, cancel = context.WithTimeout(ctx, time.Second*5)
 	defer cancel()
 	err = a.storageClient.Update(childCtx, &goastorage.UpdatePayload{AipID: params.AIPID})
 
-	return err
+	return &UploadActivityResult{}, err
 }

internal/workflow/activities/upload_test.go

@@ -128,7 +128,7 @@ func TestUploadActivity(t *testing.T) {
 
 		activity := NewUploadActivity(storageClient)
 
-		err := activity.Execute(context.Background(), &UploadActivityParams{
+		_, err := activity.Execute(context.Background(), &UploadActivityParams{
 			AIPPath: tmpDir.Join("aip.7z"),
 			AIPID:   uuid.New().String(),
 			Name:    "aip.7z",
@@ -173,7 +173,7 @@ func TestUploadActivity(t *testing.T) {
 
 		activity := NewUploadActivity(storageClient)
 
-		err := activity.Execute(context.Background(), &UploadActivityParams{
+		_, err := activity.Execute(context.Background(), &UploadActivityParams{
 			AIPPath: tmpDir.Join("aip.7z"),
 			AIPID:   uuid.New().String(),
 			Name:    "aip.7z",

internal/workflow/activities/zip.go

@@ -40,7 +40,7 @@ func (a *zipActivity) Execute(ctx context.Context, params *ZipActivityParams) (*
 	)
 
 	if params.SourceDir == "" {
-		return nil, fmt.Errorf("%s: missing source dir", ZipActivityName)
+		return &ZipActivityResult{}, fmt.Errorf("%s: missing source dir", ZipActivityName)
 	}
 
 	dest := params.DestPath
@@ -51,7 +51,7 @@ func (a *zipActivity) Execute(ctx context.Context, params *ZipActivityParams) (*
 
 	w, err := os.Create(dest) // #nosec G304 -- trusted path
 	if err != nil {
-		return nil, fmt.Errorf("%s: create: %v", ZipActivityName, err)
+		return &ZipActivityResult{}, fmt.Errorf("%s: create: %v", ZipActivityName, err)
 	}
 	defer w.Close()
 
@@ -91,7 +91,7 @@ func (a *zipActivity) Execute(ctx context.Context, params *ZipActivityParams) (*
 		return nil
 	})
 	if err != nil {
-		return nil, fmt.Errorf("%s: add files: %v", ZipActivityName, err)
+		return &ZipActivityResult{}, fmt.Errorf("%s: add files: %v", ZipActivityName, err)
 	}
 
 	return &ZipActivityResult{Path: dest}, nil