workflow: make retentionPeriod optional

With this commit, we support three different scenarios:

* `retentionPeriod` is undefined - original never removed,
* `retentionPeriod` is zero - original is removed immediately,
* `retentionPeriod` is >0 (e.g. "1s") - timer kicks in.

This fixes #10.

internal/watcher/config.go

@@ -14,7 +14,7 @@ type FilesystemConfig struct {
 	Inotify bool
 
 	Pipeline        string
-	RetentionPeriod time.Duration
+	RetentionPeriod *time.Duration
 }
 
 // See minio.go for more.
@@ -31,5 +31,5 @@ type MinioConfig struct {
 	Token        string
 
 	Pipeline        string
-	RetentionPeriod time.Duration
+	RetentionPeriod *time.Duration
 }

internal/watcher/event.go

@@ -23,7 +23,7 @@ type BlobEvent struct {
 	PipelineName string
 
 	// Retention period for this blob.
-	RetentionPeriod time.Duration
+	RetentionPeriod *time.Duration
 
 	// Key of the blob.
 	Key string

internal/watcher/watcher.go

@@ -23,15 +23,15 @@ type Watcher interface {
 	// Every watcher targets a pipeline.
 	Pipeline() string
 
-	RetentionPeriod() time.Duration
+	RetentionPeriod() *time.Duration
 
 	fmt.Stringer // It should return the name of the watcher.
 }
 
 type commonWatcherImpl struct {
 	name            string
 	pipeline        string
-	retentionPeriod time.Duration
+	retentionPeriod *time.Duration
 }
 
 func (w *commonWatcherImpl) String() string {
@@ -42,7 +42,7 @@ func (w *commonWatcherImpl) Pipeline() string {
 	return w.pipeline
 }
 
-func (w *commonWatcherImpl) RetentionPeriod() time.Duration {
+func (w *commonWatcherImpl) RetentionPeriod() *time.Duration {
 	return w.retentionPeriod
 }
 

internal/workflow/processing.go

@@ -176,8 +176,8 @@ func (w *ProcessingWorkflow) Execute(ctx workflow.Context, req *collection.Proce
 
 	// Schedule deletion of the original in the watched data source.
 	var deletionTimer workflow.Future
-	if tinfo.Status == collection.StatusDone {
-		deletionTimer = workflow.NewTimer(ctx, tinfo.Event.RetentionPeriod)
+	if tinfo.Status == collection.StatusDone && tinfo.Event.RetentionPeriod != nil {
+		deletionTimer = workflow.NewTimer(ctx, *tinfo.Event.RetentionPeriod)
 	}
 
 	// Activities that we want to run within the session regardless the