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

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

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:

context.GetRequiredOutput<T>(stageName)

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

Stage Interaction Diagram

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