Ingestion Service

┌──────────────────────────────────────────────────────────┐
│                    Data Sources                          │
│  FTP/SFTP  │  REST API  │  Cloud Storage  │  File Drop  │
└───────────┬──────────────────────────────────────────────┘
            │
            ▼
┌───────────────────────────────────────────────────────────┐
│                   Ingestion Pipeline                      │
│                                                           │
│  1. Fetch    → Pull/Push data from source                │
│  2. Convert  → Parse to JSONL (CSV/XML/JSON → JSONL)     │
│  3. Map      → Transform fields using mapper specs        │
│  4. Enrich   → Execute custom business logic              │
│  5. Stage    → Prepare database operations                │
│  6. Persist  → Write to database in batches               │
│                                                           │
│  Each step: Chunks → Temp Storage → Error Tracking       │
└───────────┬───────────────────────────────────────────────┘
            │
            ▼
┌───────────────────────────────────────────────────────────┐
│                   Output Targets                          │
│         Database Collections  │  Cloud Storage            │
└───────────────────────────────────────────────────────────┘

Fetcher implementations:
  - SFtpFetcher: Secure FTP with SSH key or password auth
  - FtpFetcher: Standard FTP connections
  - HttpFetcher: REST API polling with token auth
  - HttpStreamFetcher: Streaming API consumption
  - LocalBucketFetcher: Cloud storage file copying
  - BucketDrop: Automatic file detection via storage triggers
 
Automatic handling:
  - Connection pooling and retry logic
  - File pattern filtering (regex-based)
  - Most-recent-only mode for incremental imports
  - Source file deletion after successful fetch
  - Destination path construction with customerKey/sourceId

Converter implementations:
  - CSVConverter: Stream parsing with configurable separators
  - JSONConverter: Array and object unwrapping
  - JSONLConverter: Pass-through with validation
  - XMLConverter: Tag-based data extraction
  - EDI support via format type enum
 
Streaming architecture:
  - Chain: FileStream → Parser → Transformer → Output
  - Memory-efficient chunked processing (default 500 records)
  - Automatic error file generation (error-{filename}.jsonl)
  - Line-by-line processing prevents memory overflow
  - PipelineWritable handles automatic file chunking

JsonMapper integration:
  - Field renaming: source → destination
  - Nested object flattening/expansion
  - Type coercion (string → number, date parsing)
  - Default value injection
  - Conditional transformations
  - Array transformations
 
Mapper processing:
  - Streams JSONL input line-by-line
  - Applies IMapperSpec transformations
  - Separate error stream for failed mappings
  - Tracks source vs. processed item counts
  - Stores error items with original data + error message

EnrichmentStep base class:
  - handleRecord(record, ingestionLedger): Transform individual records
  - Optional parallel batch processing
  - Automatic error isolation per record
  - Supports both task and job execution modes
 
Enricher orchestration:
  - Streams through JSONL chunk files
  - Executes custom step logic on each record
  - Parallel batch mode for performance (configurable batch size)
  - Promise.allSettled ensures partial batch success
  - Error records written to separate error stream
  - Updates ledger with source/processed counts
 
Multiple enrichment steps per pipeline:
  - Sequential execution with output chaining
  - Each step isolated in separate temp files
  - Custom step IDs for tracking in ledger

Staging phase:
  - StagingStep.handleRecord(): Business logic determines operations
  - Output.upsert() / Output.delete(): Declares intended operations
  - Operations written to temp files (not database)
  - Multiple outputs per staging step
  - FirestoreOutput and CloudStorageOutput implementations
 
Persist phase:
  - Reads staged operations from temp files
  - BaseBatchSink processes in configurable batches (default 1,000)
  - FirestoreSink: Parallel upsert/delete with error tracking
  - CloudStorageSink: Bulk object creation
  - Operation counts tracked by type (Create/Update/Delete/Error)
  - Source vs. processed chunk tracking
 
Transaction safety:
  - Staging failures don't corrupt database
  - Can retry persist step independently
  - Failed operations logged with error details

IIngestionLedger tracks:
  - Pipeline state machine (Start → Fetch → Convert → Map → Enrich → Stage → Persist → Complete)
  - Per-step ledgers with timing, counts, and outcomes
  - Source files, destination files, completed files, error files
  - Source items vs. processed items at each step
  - Error counts and error messages
  - Start and end timestamps
 
Per-step ledgers include:
  - IFetchLedger: Source files fetched, download errors
  - IConversionLedger: Parse errors with line context
  - IMappingLedger: Transformation failures
  - IEnrichmentLedger[]: One ledger per enrichment step
  - IStagingLedger: Output logs with operation counts
  - IPersistLedger: Operation counts by type, chunk progress
 
Automatic ledger updates:
  - Each pipeline step updates its ledger section
  - transitionLedger() advances state machine
  - Database update triggers publish events
  - Real-time visibility into pipeline progress

PipelineWritable stream:
  - Configurable chunk size (records per file)
  - Automatic file rotation when chunk limit reached
  - Sequential file naming: {basename}-1.jsonl, {basename}-2.jsonl
  - Files written to temp directories by pipeline state
  - Returns ITempFileOutput[] with bucket, path, and name
  - Supports streaming writes to cloud storage
 
Chunking throughout pipeline:
  - Convert step: 500 records/chunk
  - Map step: 500 records/chunk
  - Enrich step: 500 records/chunk
  - Persist step: 1,000 operations/batch
 
Benefits:
  - Large files (millions of records) processed without memory limits
  - Failed chunks don't invalidate entire import
  - Parallel processing potential (not yet implemented)
  - Storage-efficient temp file management

Scheduled function (hourly):
  - Queries all pull-method feed configs
  - Parses cron expressions for each config
  - Compares lastRun time with cron schedule
  - Dispatches tasks for configs due to run
  - Updates lastRun timestamp on execution
 
FeedConfig includes:
  - method: Pull vs. Push
  - interval: Cron expression (e.g., "0 2 * * *" for 2 AM daily)
  - lastRun: Timestamp of most recent execution
  - Automatic tracking prevents duplicate runs

Storage event trigger:
  - onObjectFinalized monitors storage bucket
  - FILE_DROP_REGEX: ^[^/]+/{customerKey}/{sourceId}/{fileName}$
  - Matches customer/source from path
  - Dispatches to task queue for processing
 
BeginFromFileDropHandler:
  - Extracts customerKey/sourceId from file path
  - Looks up IBucketDropFeedConfig
  - Validates file type matches expected format
  - Creates ingestion ledger
  - Skips Fetch step (already have file)
  - Transitions directly to Convert step
  - Copies file to ingestion bucket if needed

IngestionState enum:
  Start → Fetch → Convert → Map → Enrich → Stage → Persist → Complete
  Also: Paused, Cancelled
 
transitionLedger() logic:
  - Automatically advances to next state after step completion
  - Fetch step skipped for Push-method configs
  - Enrich step waits for all enrichment ledgers to complete
  - Updates database with new state
 
ExecutePipelineStepHandler:
  - Checks for Paused state before executing
  - Returns early if paused (doesn't queue next step)
  - Checks for Complete state (idempotency)
  - Executes current step based on ingestionState
  - Transitions to next state
  - Queues next step execution via Cloud Tasks
 
Resume capability:
  - Unpause ingestion ledger (set state from Paused to previous state)
  - Re-run ExecutePipelineStepCommand with last known inputFiles
  - Pipeline continues from where it stopped

Error tracking per step:
  - errorCount: Total errors encountered
  - errors: Array of error messages
  - errorFiles: ITempFileOutput[] pointing to error JSONL files
  - completedFiles: Successfully processed files
  - incompleteFiles: Partially processed files
 
Error file format:
  {
    item: {original record data},
    error: "error message"
  }
 
Error isolation:
  - Fetch errors: Connection failures, auth errors
  - Conversion errors: Parse failures with line context
  - Mapping errors: Transformation failures with original record
  - Enrichment errors: Custom logic exceptions per record
  - Staging errors: Business rule violations
  - Persist errors: Database write failures
 
Error files stored separately:
  - error-{filename}.jsonl alongside successful output
  - Allows reprocessing of failed records independently
  - Full original record preserved for debugging

Cloud Functions publish events:
  - INGESTION_LEDGER_UPDATED: On every ledger update
  - PIPELINE_COMPLETED: When pipeline reaches Complete state
  - Events routed to pub/sub topics
  - Downstream services subscribe to events
 
Topic handlers:
  - Execute cleanup tasks on completion
  - Trigger dependent pipelines
  - Send notifications on success/failure
  - Update dashboards with progress

BaseEnrichJob for long-running logic:
  - Runs in background job service (isolated from main pipeline)
  - Receives chunk file reference as input
  - Downloads and processes records
  - Uploads results to new chunk file
  - Returns IEnrichmentJobResult with ledger + output file
 
Job orchestration:
  - EnrichmentCommand dispatches jobs for each chunk
  - Waits for all jobs to complete via orchestration
  - Aggregates results back into pipeline
  - Updates enrichment ledger with combined metrics
 
Benefits over task execution:
  - No 10-minute timeout limit
  - Isolated compute resources
  - Retry and failure handling
  - Progress tracking per chunk