feat(pipeline): add S3 send step for uploading pipeline output to S3-compatible storage

Implement a new pipeline step that uploads files from the previous step output
to S3-compatible storage (AWS S3, MinIO, Ceph, etc.).

Key features:
- S3Uploader interface with AWSS3Uploader implementation using AWS SDK v2
- Upload manifest tracking for retry resilience (skips already uploaded files)
- Environment variable expansion for credentials (${VAR} syntax)
- Error classification (transient vs non-transient) for retry logic
- Support for path-style addressing (required for MinIO)
- Default region set to eu-central-1

New files:
- internal/models/send.go: SendConfig, UploadManifest structs
- internal/services/s3_uploader.go: S3Uploader interface and implementation
- internal/pipeline/send.go: ExecuteSendStep function
- tests/unit/send_step_test.go: Unit tests for send step
- tests/unit/send_config_loading_test.go: Config loading tests

Closes #126

Author: trobanga

cmd/pipeline.go

@@ -230,6 +230,23 @@ func executeStep(job *models.PipelineJob, stepName models.StepName, config *mode
 		fmt.Printf("\n✓ Flattening completed\n")
 		return nil
 
+	case models.StepSend:
+		fmt.Println("Starting S3 send step...")
+		if err := pipeline.ExecuteSendStep(job, jobDir, logger); err != nil {
+			failedJob := pipeline.FailJob(job, err.Error())
+			if saveErr := pipeline.UpdateJob(config.JobsDir, failedJob); saveErr != nil {
+				logger.Error("Failed to save job state", "error", saveErr)
+			}
+			return fmt.Errorf("send step failed: %w", err)
+		}
+
+		if err := pipeline.UpdateJob(config.JobsDir, job); err != nil {
+			return fmt.Errorf("failed to save job state: %w", err)
+		}
+
+		fmt.Printf("\n✓ S3 send completed\n")
+		return nil
+
 	case models.StepWait:
 		// Execute wait step - creates empty wait directory and pauses pipeline
 		stepIndex := -1

config/aether.example.yaml

@@ -93,10 +93,42 @@ services:
     # Default: 30m (30 minutes)
     timeout: 30m
 
+  # S3 Send Service (optional)
+  # Uploads pipeline output files to S3-compatible storage (AWS S3, MinIO, Ceph, etc.)
+  # Leave empty to skip send step
+  send:
+    # S3 endpoint URL (optional)
+    # Leave empty for AWS S3, or set for S3-compatible services like MinIO
+    # Example for MinIO: "http://localhost:9000"
+    # Example for AWS S3: leave empty (uses default AWS endpoints)
+    endpoint: "${S3_ENDPOINT}"
+
+    # AWS region (default: eu-central-1)
+    region: "eu-central-1"
+
+    # S3 bucket name (required when send step is enabled)
+    # Supports environment variable expansion: "${S3_BUCKET}"
+    bucket: "${S3_BUCKET}"
+
+    # AWS credentials
+    # IMPORTANT: Use environment variables for sensitive credentials
+    # Never commit actual credentials to version control
+    access_key_id: "${AWS_ACCESS_KEY_ID}"
+    secret_access_key: "${AWS_SECRET_ACCESS_KEY}"
+
+    # Use path-style addressing instead of virtual-hosted-style
+    # Set to true for MinIO and other S3-compatible services
+    # Set to false (default) for AWS S3
+    use_path_style: false
+
+    # Upload timeout per operation
+    # Default: 30m (30 minutes)
+    timeout: 30m
+
 pipeline:
   # List of steps to execute in order
   # Import step options (must be first): torch, local_import, http_import
-  # Other step options: dimp, validation, csv_conversion, parquet_conversion, wait, flattening
+  # Other step options: dimp, validation, csv_conversion, parquet_conversion, wait, flattening, send
   # NOTE: At least one import step must be first in enabled_steps
   # NOTE: Enable all the import types you want to support - the system will automatically
   #       use the correct one based on your input (TORCH URL → torch, local dir → local_import, etc.)
@@ -107,6 +139,13 @@ pipeline:
   #   - Pipeline pauses until 'aether pipeline continue <job-id>' is run
   #   - Cannot be first step (needs previous output)
   #   - Consecutive waits are not allowed
+  #
+  # Send step:
+  #   Use 'send' to upload pipeline output to S3-compatible storage.
+  #   - Uploads files from the previous step output
+  #   - S3 key format: {job-id}/{filename}
+  #   - Tracks uploads for retry resilience (skips already uploaded files)
+  #   - Typically used as the last step in the pipeline
   enabled_steps:
     - torch           # TORCH import via CRTDL or direct TORCH URL
     - local_import    # Import from local directory
@@ -117,6 +156,7 @@ pipeline:
     - flattening      # Transform FHIR to CSV using fhir-flattener (CRTDL input only)
     - csv_conversion
     - parquet_conversion
+    # - send          # Uncomment to upload output to S3-compatible storage
 
 retry:
   # Maximum number of retry attempts for transient errors (network, 5xx)

go.mod

@@ -4,6 +4,7 @@ go 1.25
 
 require (
 	github.com/google/uuid v1.6.0
+	github.com/klauspost/compress v1.18.2
 	github.com/schollz/progressbar/v3 v3.19.0
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
@@ -12,11 +13,29 @@ require (
 )
 
 require (
+	github.com/aws/aws-sdk-go-v2 v1.41.1 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.32.7 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.7 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.95.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.30.9 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 // indirect
+	github.com/aws/smithy-go v1.24.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
-	github.com/klauspost/compress v1.18.2 // indirect
 	github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

go.sum

@@ -1,3 +1,41 @@
+github.com/aws/aws-sdk-go-v2 v1.41.1 h1:ABlyEARCDLN034NhxlRUSZr4l71mh+T5KAeGh6cerhU=
+github.com/aws/aws-sdk-go-v2 v1.41.1/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4/go.mod h1:IOAPF6oT9KCsceNTvvYMNHy0+kMF8akOjeDvPENWxp4=
+github.com/aws/aws-sdk-go-v2/config v1.32.7 h1:vxUyWGUwmkQ2g19n7JY/9YL8MfAIl7bTesIUykECXmY=
+github.com/aws/aws-sdk-go-v2/config v1.32.7/go.mod h1:2/Qm5vKUU/r7Y+zUk/Ptt2MDAEKAfUtKc1+3U1Mo3oY=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.7 h1:tHK47VqqtJxOymRrNtUXN5SP/zUTvZKeLx4tH6PGQc8=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.7/go.mod h1:qOZk8sPDrxhf+4Wf4oT2urYJrYt3RejHSzgAquYeppw=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 h1:I0GyV8wiYrP8XpA70g1HBcQO1JlQxCMTW9npl5UbDHY=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17/go.mod h1:tyw7BOl5bBe/oqvoIeECFJjMdzXoa/dfVz3QQ5lgHGA=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17 h1:xOLELNKGp2vsiteLsvLPwxC+mYmO6OZ8PYgiuPJzF8U=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17/go.mod h1:5M5CI3D12dNOtH3/mk6minaRwI2/37ifCURZISxA/IQ=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 h1:WWLqlh79iO48yLkj1v3ISRNiv+3KdQoZ6JWyfcsyQik=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17/go.mod h1:EhG22vHRrvF8oXSTYStZhJc1aUgKtnJe+aOiFEV90cM=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 h1:WKuaxf++XKWlHWu9ECbMlha8WOEGm0OUEZqm4K/Gcfk=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4/go.mod h1:ZWy7j6v1vWGmPReu0iSGvRiise4YI5SkR3OHKTZ6Wuc=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17 h1:JqcdRG//czea7Ppjb+g/n4o8i/R50aTBHkA7vu0lK+k=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17/go.mod h1:CO+WeGmIdj/MlPel2KwID9Gt7CNq4M65HUfBW97liM0=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 h1:0ryTNEdJbzUCEWkVXEXoqlXV72J5keC1GvILMOuD00E=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4/go.mod h1:HQ4qwNZh32C3CBeO6iJLQlgtMzqeG17ziAA/3KDJFow=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8 h1:Z5EiPIzXKewUQK0QTMkutjiaPVeVYXX7KIqhXu/0fXs=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8/go.mod h1:FsTpJtvC4U1fyDXk7c71XoDv3HlRm8V3NiYLeYLh5YE=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 h1:RuNSMoozM8oXlgLG/n6WLaFGoea7/CddrCfIiSA+xdY=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17/go.mod h1:F2xxQ9TZz5gDWsclCtPQscGpP0VUOc8RqgFM3vDENmU=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 h1:bGeHBsGZx0Dvu/eJC0Lh9adJa3M1xREcndxLNZlve2U=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17/go.mod h1:dcW24lbU0CzHusTE8LLHhRLI42ejmINN8Lcr22bwh/g=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.95.1 h1:C2dUPSnEpy4voWFIq3JNd8gN0Y5vYGDo44eUE58a/p8=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.95.1/go.mod h1:5jggDlZ2CLQhwJBiZJb4vfk4f0GxWdEDruWKEJ1xOdo=
+github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 h1:VrhDvQib/i0lxvr3zqlUwLwJP4fpmpyD9wYG1vfSu+Y=
+github.com/aws/aws-sdk-go-v2/service/signin v1.0.5/go.mod h1:k029+U8SY30/3/ras4G/Fnv/b88N4mAfliNn08Dem4M=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.9 h1:v6EiMvhEYBoHABfbGB4alOYmCIrcgyPPiBE1wZAEbqk=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.9/go.mod h1:yifAsgBxgJWn3ggx70A3urX2AN49Y5sJTD1UQFlfqBw=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13 h1:gd84Omyu9JLriJVCbGApcLzVR3XtmC4ZDPcAI6Ftvds=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13/go.mod h1:sTGThjphYE4Ohw8vJiRStAcu3rbjtXRsdNB0TvZ5wwo=
+github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 h1:5fFjR/ToSOzB2OQ/XqWpZBmNvmP/pJ1jOWYlFDJTjRQ=
+github.com/aws/aws-sdk-go-v2/service/sts v1.41.6/go.mod h1:qgFDZQSD/Kys7nJnVqYlWKnh0SSdMjAi0uSwON4wgYQ=
+github.com/aws/smithy-go v1.24.0 h1:LpilSUItNPFr1eY85RYgTIg5eIEPtvFbskaFcmmIUnk=
+github.com/aws/smithy-go v1.24.0/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0=
 github.com/chengxilo/virtualterm v1.0.4 h1:Z6IpERbRVlfB8WkOmtbHiDbBANU7cimRIof7mk9/PwM=
 github.com/chengxilo/virtualterm v1.0.4/go.mod h1:DyxxBZz/x1iqJjFxTFcr6/x+jSpqN0iwWCOK1q10rlY=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
@@ -37,8 +75,6 @@ github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/f
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
 github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
-github.com/schollz/progressbar/v3 v3.18.0 h1:uXdoHABRFmNIjUfte/Ex7WtuyVslrw2wVPQmCN62HpA=
-github.com/schollz/progressbar/v3 v3.18.0/go.mod h1:IsO3lpbaGuzh8zIMzgY3+J8l4C8GjO0Y9S69eFvNsec=
 github.com/schollz/progressbar/v3 v3.19.0 h1:Ea18xuIRQXLAUidVDox3AbwfUhD0/1IvohyTutOIFoc=
 github.com/schollz/progressbar/v3 v3.19.0/go.mod h1:IsO3lpbaGuzh8zIMzgY3+J8l4C8GjO0Y9S69eFvNsec=
 github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw=
@@ -47,8 +83,6 @@ github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I=
 github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg=
 github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY=
 github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
-github.com/spf13/cobra v1.10.1 h1:lJeBwCfmrnXthfAupyUTzJ/J4Nc1RsHC/mSRU2dll/s=
-github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4XaB0=
 github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU=
 github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4=
 github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=

internal/models/config.go

@@ -31,12 +31,13 @@ func DefaultCompressionConfig() CompressionConfig {
 
 // ServiceConfig contains connection details for external HTTP services
 type ServiceConfig struct {
-	DIMP               DIMPConfig                `yaml:"dimp" json:"dimp"`
-	CSVConversion      CSVConversionConfig       `yaml:"csv_conversion" json:"csv_conversion"`
-	ParquetConversion  ParquetConversionConfig   `yaml:"parquet_conversion" json:"parquet_conversion"`
-	TORCH              TORCHConfig               `yaml:"torch" json:"torch"`
-	Flattening         FlatteningConfig          `yaml:"flattening" json:"flattening"`
-	CRTDLPreprocessing CRTDLPreprocessingConfig  `yaml:"crtdl_preprocessing" json:"crtdl_preprocessing"`
+	DIMP               DIMPConfig               `yaml:"dimp" json:"dimp"`
+	CSVConversion      CSVConversionConfig      `yaml:"csv_conversion" json:"csv_conversion"`
+	ParquetConversion  ParquetConversionConfig  `yaml:"parquet_conversion" json:"parquet_conversion"`
+	TORCH              TORCHConfig              `yaml:"torch" json:"torch"`
+	Flattening         FlatteningConfig         `yaml:"flattening" json:"flattening"`
+	CRTDLPreprocessing CRTDLPreprocessingConfig `yaml:"crtdl_preprocessing" json:"crtdl_preprocessing"`
+	Send               SendConfig               `yaml:"send" json:"send"`
 }
 
 // DIMPConfig contains DIMP pseudonymization service settings
@@ -101,6 +102,7 @@ func DefaultConfig() ProjectConfig {
 			},
 			Flattening:         DefaultFlatteningConfig(),
 			CRTDLPreprocessing: DefaultCRTDLPreprocessingConfig(),
+			Send:               DefaultSendConfig(),
 		},
 		Pipeline: PipelineConfig{
 			EnabledSteps: []StepName{StepLocalImport, StepHttpImport},