Add batch option to determine depth of SIPs

Co-authored-by: camlyall <[email protected]>

internal/api/design/batch.go

@@ -20,6 +20,10 @@ var _ = Service("batch", func() {
 			Attribute("reject_duplicates", Boolean, func() { Default(false) })
 			Attribute("transfer_type", String)
 			Attribute("process_name_metadata", Boolean, func() { Default(false) })
+			Attribute("depth", Int, func() {
+				Default(0)
+				Minimum(0)
+			})
 			Required("path")
 		})
 		Result(BatchResult)

internal/batch/service.go

@@ -77,6 +77,7 @@ func (s *batchImpl) Submit(ctx context.Context, payload *goabatch.SubmitPayload)
 		input.TransferType = *payload.TransferType
 	}
 	input.MetadataConfig.ProcessNameMetadata = payload.ProcessNameMetadata
+	input.Depth = int32(payload.Depth)
 	opts := temporalsdk_client.StartWorkflowOptions{
 		ID:                       BatchWorkflowID,
 		WorkflowIDReusePolicy:    temporalapi_enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,

internal/batch/workflow.go

@@ -2,7 +2,9 @@ package batch
 
 import (
 	"context"
-	"os"
+	"io/fs"
+	"path/filepath"
+	"strings"
 	"time"
 
 	temporalsdk_temporal "go.temporal.io/sdk/temporal"
@@ -32,6 +34,7 @@ type BatchWorkflowInput struct {
 	RejectDuplicates bool
 	TransferType     string
 	MetadataConfig   metadata.Config
+	Depth            int32
 }
 
 func BatchWorkflow(ctx temporalsdk_workflow.Context, params BatchWorkflowInput) error {
@@ -56,19 +59,39 @@ func NewBatchActivity(batchsvc Service) *BatchActivity {
 }
 
 func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput) error {
-	files, err := os.ReadDir(params.Path)
-	if err != nil {
-		return temporal.NewNonRetryableError(err)
-	}
 	pipelines := []string{}
 	if params.PipelineName != "" {
 		pipelines = append(pipelines, params.PipelineName)
 	}
-	for _, file := range files {
+
+	if params.Depth < 0 {
+		params.Depth = 0
+	}
+
+	root := params.Path
+	err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+
+		rel, err := filepath.Rel(root, path)
+		if err != nil {
+			return err
+		}
+
+		if rel == "." {
+			return nil // Ignore root.
+		}
+
+		depth := len(strings.Split(rel, string(filepath.Separator))) - 1
+		if depth != int(params.Depth) {
+			return nil // Keep walking.
+		}
+
 		req := collection.ProcessingWorkflowRequest{
-			BatchDir:         params.Path,
-			Key:              file.Name(),
-			IsDir:            file.IsDir(),
+			BatchDir:         filepath.Dir(path),
+			Key:              entry.Name(),
+			IsDir:            entry.IsDir(),
 			PipelineNames:    pipelines,
 			ProcessingConfig: params.ProcessingConfig,
 			CompletedDir:     params.CompletedDir,
@@ -77,7 +100,14 @@ func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput)
 			TransferType:     params.TransferType,
 			MetadataConfig:   params.MetadataConfig,
 		}
+
 		_ = a.batchsvc.InitProcessingWorkflow(ctx, &req)
+
+		return fs.SkipDir
+	})
+	if err != nil {
+		return temporal.NewNonRetryableError(err)
 	}
+
 	return nil
 }

internal/batch/workflow_test.go

@@ -69,5 +69,5 @@ func TestBatchActivityFailsWithBogusBatchPath(t *testing.T) {
 		Path:         "/non/existent/path",
 		PipelineName: "am",
 	})
-	assert.Error(t, err, "open /non/existent/path: no such file or directory")
+	assert.ErrorContains(t, err, "no such file or directory")
 }