# ETL Source Adapter Framework

## Purpose

This document describes the ETL source adapter framework as it exists in the codebase today.

It replaces the earlier proposal document. The goal is to help a new developer understand:

- what the framework owns,
- what a source adapter owns,
- how one ETL run flows through the system,
- where Brink POS fits in,
- how to extend the system with another source later.

## Current Scope

The current implemented framework supports the active Brink POS pipeline with these stages:

1. `extract_raw`
2. `normalize_load`
3. `load_to_staging`
4. `staging_to_curated`

The framework also reserves a future stage name:

- `curated_to_s3`

That future stage is not implemented yet, but the runtime model already allows appending it without redesigning the pipeline runner.

## High-Level Intent

The framework is intentionally small.

It does not try to standardize source-specific transport, parsing, or business semantics. Instead it separates the system into two layers:

- framework runtime for shared ETL mechanics,
- source adapter for source-specific logic and metadata.

In practical terms this means:

- a new source should plug in by registering an adapter and its stages,
- shared concerns such as stage sequencing, checkpointing, failure packaging, and transactional curated execution should not be reimplemented per source,
- Brink-specific details stay in Brink code.

## Where The Code Lives

### Framework runtime

- `backend/Kds.Ai.Integrations/Etl/EtlPipelineRunner.cs`
- `backend/Kds.Ai.Integrations/Etl/IEtlPipelineRunner.cs`
- `backend/Kds.Ai.Integrations/Etl/IEtlSourceAdapter.cs`
- `backend/Kds.Ai.Integrations/Etl/IEtlStage.cs`
- `backend/Kds.Ai.Integrations/Etl/EtlPipelineRequest.cs`
- `backend/Kds.Ai.Integrations/Etl/EtlStageContext.cs`
- `backend/Kds.Ai.Integrations/Etl/EtlStageResult.cs`
- `backend/Kds.Ai.Integrations/Etl/EtlPipelineResult.cs`
- `backend/Kds.Ai.Integrations/Etl/EtlStageNames.cs`

### Shared stage mechanics

- `backend/Kds.Ai.Integrations/Etl/IStageExecutionSupport.cs`
- `backend/Kds.Ai.Integrations/Etl/StageExecutionSupport.cs`
- `backend/Kds.Ai.Integrations/Etl/StageCheckpointAdvanceRequest.cs`
- `backend/Kds.Ai.Integrations/Etl/StageBatchFailureRequest.cs`
- `backend/Kds.Ai.Integrations/Etl/EtlSourceDescriptor.cs`
- `backend/Kds.Ai.Integrations/Etl/StageDiagnosticsEnvelope.cs`

### Shared curated execution

- `backend/Kds.Ai.Integrations/Etl/ICuratedTargetExecutor.cs`
- `backend/Kds.Ai.Integrations/Etl/CuratedTargetExecutor.cs`
- `backend/Kds.Ai.Integrations/Etl/CuratedTargetDefinition.cs`
- `backend/Kds.Ai.Integrations/Etl/CuratedTargetExecutionRequest.cs`
- `backend/Kds.Ai.Integrations/Etl/CuratedTargetExecutionResult.cs`

### Shared catalog contracts

- `backend/Kds.Ai.Integrations/Etl/IExtractOperationCatalog.cs`
- `backend/Kds.Ai.Integrations/Etl/INormalizationContractCatalog.cs`
- `backend/Kds.Ai.Integrations/Etl/IStagingTableCatalog.cs`
- `backend/Kds.Ai.Integrations/Etl/ICuratedTargetCatalog.cs`

### Brink adapter implementation

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosSourceAdapter.cs`
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosExtractStage.cs`
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosNormalizeLoadStage.cs`
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosLoadToStagingStage.cs`
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosStagingToCuratedStage.cs`
- `backend/Kds.Ai.ApiService/Jobs/BrinkPosEtlJobs.cs`

## Responsibility Split

### Framework-owned responsibilities

- pipeline orchestration,
- stage ordering,
- carrying prior stage outputs between stages,
- checkpoint advancement,
- failed package writing contract,
- batch failure integration,
- transactional curated target execution,
- shared source descriptor shape,
- shared catalog contract shapes,
- reserved place for future stages.

### Adapter-owned responsibilities

- source credentials,
- transport and endpoint execution,
- business date rules,
- parser internals,
- normalization profile content,
- source contract definitions,
- staging table definitions,
- curated target SQL and mapping rules,
- stage-specific validation.

## Runtime Architecture

```mermaid
flowchart LR
    A[Scheduler or Manual Endpoint] --> B[BrinkPosEtlJobs]
    B --> C[IEtlPipelineRunner]
    C --> D[IEtlSourceAdapter]
    D --> E[Extract Stage]
    D --> F[Normalize Stage]
    D --> G[Load To Staging Stage]
    D --> H[Staging To Curated Stage]

    E --> I[Brink Extract Jobs]
    F --> J[Brink Normalize Jobs]
    G --> K[Brink Load To Staging Jobs]
    H --> L[Brink Staging To Curated Jobs]

    F --> M[StageExecutionSupport]
    G --> M
    H --> M

    H --> N[CuratedTargetExecutor]

    I --> O[S3 raw]
    J --> P[S3 load]
    K --> Q[Redshift staging]
    N --> R[Redshift curated]
```

## End-To-End Flow

### Entry point

`BrinkPosEtlJobs` is still the operational entry point for scheduled and manual Brink ETL runs.

- scheduled flow iterates active Brink tenant bindings,
- manual flow accepts a specific tenant and optional business date,
- both flows now invoke the framework runtime instead of a hardcoded four-call sequence.

### Pipeline invocation

The job builds `EtlPipelineRequest` with:

- `CompanyId`,
- `TenantId`,
- optional `BusinessDate`,
- `Trigger`,
- `ForceReset`,
- `RequestedBy`,
- selected `IEtlSourceAdapter`.

For Brink, the adapter is resolved by `SourceType == "brink_pos"`.

## Data Flow Diagram

```mermaid
sequenceDiagram
    participant Job as BrinkPosEtlJobs
    participant Runner as EtlPipelineRunner
    participant Adapter as BrinkPosSourceAdapter
    participant Extract as BrinkPosExtractStage
    participant Normalize as BrinkPosNormalizeLoadStage
    participant Staging as BrinkPosLoadToStagingStage
    participant Curated as BrinkPosStagingToCuratedStage
    participant Support as StageExecutionSupport
    participant Exec as CuratedTargetExecutor

    Job->>Runner: RunAsync(EtlPipelineRequest)
    Runner->>Adapter: GetStages()
    Adapter-->>Runner: extract, normalize, staging, curated

    Runner->>Extract: ExecuteAsync(context)
    Extract-->>Runner: EtlStageResult(raw extract output)

    Runner->>Normalize: ExecuteAsync(context + extract result)
    Normalize->>Support: AdvanceCheckpoint / FailStageBatch
    Normalize-->>Runner: EtlStageResult(normalize result)

    Runner->>Staging: ExecuteAsync(context + normalize result)
    Staging->>Support: AdvanceCheckpoint / FailStageBatch
    Staging-->>Runner: EtlStageResult(staging result)

    Runner->>Curated: ExecuteAsync(context + staging result)
    Curated->>Exec: ExecuteAsync(curated targets)
    Exec-->>Curated: target execution results
    Curated->>Support: AdvanceCheckpoint / FailStageBatch
    Curated-->>Runner: EtlStageResult(curated result)

    Runner-->>Job: EtlPipelineResult
```

## Stage-By-Stage Behavior

### 1. `extract_raw`

Wrapper:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosExtractStage.cs`

Delegates to:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosExtractJobs.cs`

Responsibility:

- resolve Brink extraction input,
- call external Brink source logic,
- persist raw payloads to S3,
- create the initial batch output containing the resolved business date.

Important note:

- extract stage remains source-specific and is not yet routed through `StageExecutionSupport`.

### 2. `normalize_load`

Wrapper:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosNormalizeLoadStage.cs`

Delegates to:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosNormalizeLoadJobs.cs`

Responsibility:

- read raw artifact metadata from the extract result,
- run normalization,
- write load-ready artifacts to S3,
- complete batch and advance checkpoint on success,
- write failed package and fail batch on exception.

### 3. `load_to_staging`

Wrapper:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosLoadToStagingStage.cs`

Delegates to:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosLoadToStagingJobs.cs`

Responsibility:

- read normalize details,
- validate contract-to-table compatibility,
- run `COPY`-based staging loads,
- record diagnostics per staging table,
- complete batch and advance checkpoint on success,
- create failed package on exception.

### 4. `staging_to_curated`

Wrapper:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosStagingToCuratedStage.cs`

Delegates to:

- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosStagingToCuratedJobs.cs`

Responsibility:

- validate required staging tables,
- map Brink curated targets into shared `CuratedTargetDefinition`,
- invoke shared transactional curated executor,
- convert executor results into Brink diagnostics,
- complete batch and advance checkpoint on success,
- write failed package on exception.

## Shared Failure And Checkpoint Handling

The shared helper is `StageExecutionSupport`.

It is used by the downstream stages that already share the same control-flow pattern.

### Success path

On success a stage calls `AdvanceCheckpointAsync` with:

- source descriptor,
- tenant,
- business date,
- stage name,
- batch id,
- completion time.

The helper then:

1. builds the checkpoint key from tenant + business date + stage,
2. stores the checkpoint through `IIngestionCheckpointService`,
3. keeps source identity framework-driven instead of hardcoded inside a Brink-only helper.

### Failure path

On exception a stage calls `FailStageBatchAsync` with:

- source descriptor,
- batch identity,
- stage identity,
- trigger metadata,
- original artifact reference,
- diagnostics payload.

The helper then:

1. serializes diagnostics,
2. writes a failed package through `IFailedPackageWriter`,
3. produces standard error details,
4. marks the batch failed through `IIngestionBatchLifecycleService`.

## Shared Transactional Curated Execution

`CuratedTargetExecutor` owns the transaction boundary for curated target execution.

Its algorithm is:

1. open tenant warehouse connection,
2. begin transaction,
3. for each target execute `DELETE`,
4. then execute target `WRITE` SQL,
5. collect diagnostics,
6. commit only if all targets succeed,
7. rollback the whole unit on any failure.

This keeps target definitions source-owned while moving transaction mechanics into shared code.

## Catalog Model

The framework defines shared catalog shapes, but Brink still owns the catalog content.

### Brink extract catalog

- `BrinkPosEndpointCatalog`
- shared shape: `IExtractOperationCatalog`

### Brink normalization catalog

- `BrinkPosNormalizationCatalog`
- shared shape: `INormalizationContractCatalog`

### Brink staging catalog

- `BrinkPosStagingTableCatalog`
- shared shape: `IStagingTableCatalog`

### Brink curated catalog

- `BrinkPosCuratedTargetCatalog`
- shared shape: `ICuratedTargetCatalog`

This means the framework can reason about completeness and structure, while adapter code still owns actual source metadata and SQL.

## How Stage Outputs Move Through The Runtime

`EtlPipelineRunner` stores completed stage results in a dictionary keyed by stage name.

Each later stage receives `EtlStageContext` and can access upstream results through:

```csharp
context.GetRequiredOutput<T>(stageName)
```

That is the key runtime seam that replaced the old hardcoded orchestration chain.

## Stage Interaction Diagram

```mermaid
flowchart TD
    A[EtlPipelineRequest] --> B[extract_raw]
    B --> C[EtlStageResult: BrinkPosExtractJobResult]
    C --> D[normalize_load]
    D --> E[EtlStageResult: BrinkPosStageJobResult]
    E --> F[load_to_staging]
    F --> G[EtlStageResult: BrinkPosStageJobResult]
    G --> H[staging_to_curated]
    H --> I[EtlStageResult: BrinkPosStageJobResult]
```

## Dependency Injection Shape

Runtime registration is in:

- `backend/Kds.Ai.Integrations/DependencyInjectionExtensions.cs`

The current registration includes:

- `IEtlPipelineRunner -> EtlPipelineRunner`
- `IStageExecutionSupport -> StageExecutionSupport`
- `ICuratedTargetExecutor -> CuratedTargetExecutor`
- Brink stage wrappers
- `IEtlSourceAdapter -> BrinkPosSourceAdapter`

This is the minimum composition required to make the framework seam real at runtime.

## What A New Source Needs To Implement

To add another source, the expected path is:

1. create a source adapter implementing `IEtlSourceAdapter`,
2. provide one or more stage implementations implementing `IEtlStage`,
3. keep source-specific extract logic inside the adapter package,
4. provide source catalogs through the shared catalog contracts,
5. reuse `StageExecutionSupport` and `CuratedTargetExecutor` where applicable,
6. register the adapter and stages in DI.

The new source should not reimplement:

- pipeline runner,
- stage result passing,
- checkpoint semantics,
- failed package writing shape,
- curated transaction management.

## Current Boundaries And Intentional Limits

The framework does not currently try to unify:

- source credential models,
- source clients,
- parser implementation style,
- business date semantics,
- curated SQL content,
- dormant or future Brink extraction paths.

That is deliberate. The current slice is an extraction from the active Brink flow, not a platform rewrite.

## Testing And Regression Coverage

Key regression tests live in:

- `backend/tests/Kds.Ai.Integrations.Tests/Etl/StageExecutionSupportTests.cs`
- `backend/tests/Kds.Ai.Integrations.Tests/Etl/CuratedTargetExecutorTests.cs`
- `backend/tests/Kds.Ai.Integrations.Tests/Etl/EtlPipelineRunnerTests.cs`
- `backend/tests/Kds.Ai.Integrations.Tests/BrinkPos/BrinkPosStagingToCuratedJobsTests.cs`
- `backend/tests/Kds.Ai.Integrations.Tests/BrinkPos/BrinkPosEtlCatalogTests.cs`
- `backend/tests/Kds.Ai.ApiService.Tests/Kds.Ai.ApiService.Tests/Endpoints/BrinkPosManualEtlEndpointTests.cs`

These tests cover:

- checkpoint advancement,
- failed package writing,
- curated transaction commit and rollback,
- stage ordering and prior-result visibility,
- Brink adapter stage list,
- endpoint-level manual trigger wiring.

## How To Read The System As A New Developer

Recommended reading order:

1. `backend/Kds.Ai.ApiService/Jobs/BrinkPosEtlJobs.cs`
2. `backend/Kds.Ai.Integrations/Etl/EtlPipelineRunner.cs`
3. `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosSourceAdapter.cs`
4. the four Brink stage wrappers,
5. the delegated Brink job classes,
6. `StageExecutionSupport` and `CuratedTargetExecutor`,
7. Brink catalogs,
8. the ETL regression tests.

That path mirrors the runtime flow and makes the ownership boundaries easiest to understand.

## Related Documents

- `docs/ETL.md`
- `docs/features/ETL/index.md`
- `docs/features/ETL/how-to-add-etl-source-adapter.md`
- `docs/features/ETL/warehouse-data/index.md`
- `docs/features/ETL/warehouse-data/curated-model/index.md`