Add option to exclude hidden files from the transfer

Co-authored-by: Jesús García Crespo <[email protected]>

Makefile

@@ -154,6 +154,7 @@ gen-mock: $(MOCKGEN)
 	mockgen -typed -destination=./internal/collection/fake/mock_collection.go -package=fake github.com/artefactual-labs/enduro/internal/collection Service
 	mockgen -typed -destination=./internal/pipeline/fake/mock_pipeline.go -package=fake github.com/artefactual-labs/enduro/internal/pipeline Service
 	mockgen -typed -destination=./internal/watcher/fake/mock_watcher.go -package=fake github.com/artefactual-labs/enduro/internal/watcher Service
+	mockgen -typed -destination=./internal/watcher/fake/mock_watcher_unit.go -package=fake github.com/artefactual-labs/enduro/internal/watcher Watcher
 
 temporal: # @HELP Runs a development instance of Temporal.
 temporal: PORT := 55555

enduro.toml

@@ -35,6 +35,7 @@ pipeline = "am"
 retentionPeriod = "10s"
 stripTopLevelDir = true
 rejectDuplicates = false
+excludeHiddenFiles = false
 transferType = "standard"
 
 [[watcher.filesystem]]
@@ -46,6 +47,7 @@ completedDir = "./hack/landfill"
 ignore = '(^\.gitkeep)|(^*\.mft)$'
 stripTopLevelDir = true
 rejectDuplicates = false
+excludeHiddenFiles = false
 transferType = "standard"
 
 [[pipeline]]

internal/api/design/batch.go

@@ -18,6 +18,7 @@ var _ = Service("batch", func() {
 			Attribute("completed_dir", String)
 			Attribute("retention_period", String)
 			Attribute("reject_duplicates", Boolean, func() { Default(false) })
+			Attribute("exclude_hidden_files", Boolean, func() { Default(false) })
 			Attribute("transfer_type", String)
 			Attribute("process_name_metadata", Boolean, func() { Default(false) })
 			Attribute("depth", Int, func() {

internal/batch/service.go

@@ -72,10 +72,11 @@ func (s *batchImpl) Submit(ctx context.Context, payload *goabatch.SubmitPayload)
 		}
 		input.RetentionPeriod = &dur
 	}
-	input.RejectDuplicates = payload.RejectDuplicates
 	if payload.TransferType != nil {
 		input.TransferType = *payload.TransferType
 	}
+	input.RejectDuplicates = payload.RejectDuplicates
+	input.ExcludeHiddenFiles = payload.ExcludeHiddenFiles
 	input.MetadataConfig.ProcessNameMetadata = payload.ProcessNameMetadata
 	input.Depth = int32(payload.Depth)
 	opts := temporalsdk_client.StartWorkflowOptions{

internal/batch/workflow.go

@@ -26,15 +26,16 @@ type BatchProgress struct {
 }
 
 type BatchWorkflowInput struct {
-	Path             string
-	PipelineName     string
-	ProcessingConfig string
-	CompletedDir     string
-	RetentionPeriod  *time.Duration
-	RejectDuplicates bool
-	TransferType     string
-	MetadataConfig   metadata.Config
-	Depth            int32
+	Path               string
+	PipelineName       string
+	ProcessingConfig   string
+	CompletedDir       string
+	RetentionPeriod    *time.Duration
+	RejectDuplicates   bool
+	ExcludeHiddenFiles bool
+	TransferType       string
+	MetadataConfig     metadata.Config
+	Depth              int32
 }
 
 func BatchWorkflow(ctx temporalsdk_workflow.Context, params BatchWorkflowInput) error {
@@ -89,16 +90,17 @@ func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput)
 		}
 
 		req := collection.ProcessingWorkflowRequest{
-			BatchDir:         filepath.Dir(path),
-			Key:              entry.Name(),
-			IsDir:            entry.IsDir(),
-			PipelineNames:    pipelines,
-			ProcessingConfig: params.ProcessingConfig,
-			CompletedDir:     params.CompletedDir,
-			RetentionPeriod:  params.RetentionPeriod,
-			RejectDuplicates: params.RejectDuplicates,
-			TransferType:     params.TransferType,
-			MetadataConfig:   params.MetadataConfig,
+			BatchDir:           filepath.Dir(path),
+			Key:                entry.Name(),
+			IsDir:              entry.IsDir(),
+			PipelineNames:      pipelines,
+			ProcessingConfig:   params.ProcessingConfig,
+			CompletedDir:       params.CompletedDir,
+			RetentionPeriod:    params.RetentionPeriod,
+			RejectDuplicates:   params.RejectDuplicates,
+			ExcludeHiddenFiles: params.ExcludeHiddenFiles,
+			TransferType:       params.TransferType,
+			MetadataConfig:     params.MetadataConfig,
 		}
 
 		_ = a.batchsvc.InitProcessingWorkflow(ctx, &req)

internal/collection/workflow.go

@@ -62,6 +62,9 @@ type ProcessingWorkflowRequest struct {
 	// Whether we reject duplicates based on name (key).
 	RejectDuplicates bool
 
+	// Whether we exclude hidden files from submission.
+	ExcludeHiddenFiles bool
+
 	// Transfer type.
 	TransferType string
 

internal/watcher/config.go

@@ -33,12 +33,13 @@ type FilesystemConfig struct {
 	Inotify bool
 	Ignore  string
 
-	Pipeline         []string
-	RetentionPeriod  *time.Duration
-	CompletedDir     string
-	StripTopLevelDir bool
-	RejectDuplicates bool
-	TransferType     string
+	Pipeline           []string
+	RetentionPeriod    *time.Duration
+	CompletedDir       string
+	StripTopLevelDir   bool
+	RejectDuplicates   bool
+	ExcludeHiddenFiles bool
+	TransferType       string
 }
 
 // See minio.go for more.
@@ -55,9 +56,10 @@ type MinioConfig struct {
 	Token        string
 	Bucket       string
 
-	Pipeline         []string
-	RetentionPeriod  *time.Duration
-	StripTopLevelDir bool
-	RejectDuplicates bool
-	TransferType     string
+	Pipeline           []string
+	RetentionPeriod    *time.Duration
+	StripTopLevelDir   bool
+	RejectDuplicates   bool
+	ExcludeHiddenFiles bool
+	TransferType       string
 }

internal/watcher/event.go

@@ -32,6 +32,9 @@ type BlobEvent struct {
 	// Whether duplicates are rejected or not.
 	RejectDuplicates bool
 
+	// Whether hidden files are exluded or not.
+	ExcludeHiddenFiles bool
+
 	// Which transfer type to use in Archivemaitca.
 	TransferType string
 
@@ -47,15 +50,16 @@ type BlobEvent struct {
 
 func NewBlobEvent(w Watcher, key string, isDir bool) *BlobEvent {
 	return &BlobEvent{
-		WatcherName:      w.String(),
-		PipelineName:     w.Pipelines(),
-		RetentionPeriod:  w.RetentionPeriod(),
-		CompletedDir:     w.CompletedDir(),
-		StripTopLevelDir: w.StripTopLevelDir(),
-		RejectDuplicates: w.RejectDuplicates(),
-		TransferType:     w.TransferType(),
-		Key:              key,
-		IsDir:            isDir,
+		WatcherName:        w.String(),
+		PipelineName:       w.Pipelines(),
+		RetentionPeriod:    w.RetentionPeriod(),
+		CompletedDir:       w.CompletedDir(),
+		StripTopLevelDir:   w.StripTopLevelDir(),
+		RejectDuplicates:   w.RejectDuplicates(),
+		ExcludeHiddenFiles: w.ExcludeHiddenFiles(),
+		TransferType:       w.TransferType(),
+		Key:                key,
+		IsDir:              isDir,
 	}
 }