How To Add An ETL Source Adapter

Purpose

This document is an onboarding guide for adding a new ETL source to the current framework.

It is written for a developer who needs a practical implementation path, not just architecture context.

Before You Start

Read these first:

1. docs/features/ETL/etl-source-adapter-framework.md
2. docs/ETL.md
3. the existing Brink implementation under backend/Kds.Ai.Integrations/BrinkPos/

The Brink adapter is the reference implementation for the current framework shape.

Design Rules

When adding a new source:

• keep framework mechanics in backend/Kds.Ai.Integrations/Etl/,
• keep source-specific code inside the source adapter package,
• reuse shared stage support and curated executor instead of cloning Brink patterns,
• do not redesign the runtime unless the new source exposes a real shared need.

What The Framework Expects

At minimum a new source should provide:

1. one IEtlSourceAdapter,
2. one or more IEtlStage implementations,
3. source-specific extraction logic,
4. source-specific catalogs through shared catalog contracts,
5. DI registration.

Implementation Checklist

Step 1. Define source identity

Create an EtlSourceDescriptor for the source.

It should define:

• stable SourceType string,
• matching IngestionSourceType.

Example shape:

public static readonly EtlSourceDescriptor Source = new(
    "my_source",
    IngestionSourceType.MySource);

Guidelines:

• SourceType should be stable and storage-safe,
• use one canonical string everywhere,
• avoid aliases unless there is a migration reason.

Step 2. Decide stage list

Choose which stages the source actually needs.

Today the framework supports ordered stage registration. The common baseline is:

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

If the source does not need one of these stages, decide that explicitly and document why.

Do not add a custom runner branch for one source.

Step 3. Implement stage wrappers

Create thin stage wrappers implementing IEtlStage.

Recommended pattern:

• wrapper reads upstream result from EtlStageContext,
• wrapper maps framework input to source-specific job input,
• wrapper delegates to source-specific service,
• wrapper returns EtlStageResult.

Example shape:

internal sealed class MySourceNormalizeStage(IMySourceNormalizeJobs jobs) : IEtlStage
{
    public string StageName => EtlStageNames.NormalizeLoad;

    public async Task<EtlStageResult> ExecuteAsync(EtlStageContext context, CancellationToken ct = default)
    {
        var extract = context.GetRequiredOutput<MyExtractResult>(EtlStageNames.ExtractRaw);
        var result = await jobs.ExecuteAsync(new MyNormalizeInput(...), ct);
        return new EtlStageResult(StageName, result.BusinessDate, result);
    }
}

Step 4. Keep source logic in source services

Do not move source-specific behavior into the framework.

Keep inside the adapter package:

• credentials,
• API clients,
• parsing,
• business date logic,
• source normalization logic,
• SQL target content.

The framework should call the source code, not absorb it.

Step 5. Reuse shared stage support

If a stage follows the same downstream pattern as Brink normalize_load, load_to_staging, or staging_to_curated, use IStageExecutionSupport.

Use it for:

• checkpoint advancement on success,
• failed package writing,
• batch failure integration.

This avoids per-source drift in operational behavior.

Step 6. Reuse shared curated executor where possible

If the source has a staging_to_curated phase with ordered SQL target execution, use ICuratedTargetExecutor.

The source should still own:

• target list,
• SQL text,
• dependency rules,
• target-specific diagnostics mapping.

The framework owns:

• transaction,
• parameter binding,
• commit / rollback unit.

Step 7. Implement catalogs

Expose source metadata through the shared contracts:

• IExtractOperationCatalog
• INormalizationContractCatalog
• IStagingTableCatalog
• ICuratedTargetCatalog

The content stays source-specific.

Recommended validations:

• every staging table maps to a known normalization contract,
• every curated dependency references a known staging table,
• every active extract operation has downstream ownership.

Step 8. Implement the adapter

Create one adapter class implementing IEtlSourceAdapter.

It should:

• expose Descriptor,
• return the active ordered stage list,
• keep stage composition source-local.

Example shape:

internal sealed class MySourceAdapter(
    MyExtractStage extract,
    MyNormalizeStage normalize,
    MyLoadStage load,
    MyCuratedStage curated) : IEtlSourceAdapter
{
    public EtlSourceDescriptor Descriptor { get; } = new("my_source", IngestionSourceType.MySource);

    public IReadOnlyList<IEtlStage> GetStages() => [extract, normalize, load, curated];
}

Step 9. Register in DI

Add registrations in:

• backend/Kds.Ai.Integrations/DependencyInjectionExtensions.cs

At minimum register:

• source services,
• stage wrappers,
• IEtlSourceAdapter implementation.

If the source uses shared helpers, it should consume existing registrations rather than creating new copies.

Step 10. Wire an entry point

If the source needs scheduled or manual execution, create an application job similar to BrinkPosEtlJobs.

That entry point should:

• gather runtime input,
• select the right adapter,
• call IEtlPipelineRunner.RunAsync(...).

Do not hardcode a new per-stage call chain in the job.

Recommended Folder Shape

Suggested source structure:

backend/Kds.Ai.Integrations/MySource/
  MySourceAdapter.cs
  MySourceExtractStage.cs
  MySourceNormalizeStage.cs
  MySourceLoadToStagingStage.cs
  MySourceStagingToCuratedStage.cs
  MySourceExtractJobs.cs
  MySourceNormalizeLoadJobs.cs
  MySourceLoadToStagingJobs.cs
  MySourceStagingToCuratedJobs.cs
  MySourceEndpointCatalog.cs
  MySourceNormalizationCatalog.cs
  MySourceStagingTableCatalog.cs
  MySourceCuratedTargetCatalog.cs
  Models/

Adjust the exact set only if the source genuinely needs less.

Data Flow Template

flowchart TD
    A[Entry Job] --> B[IEtlPipelineRunner]
    B --> C[MySourceAdapter]
    C --> D[extract_raw]
    D --> E[normalize_load]
    E --> F[load_to_staging]
    F --> G[staging_to_curated]

Delivery Checklist

Before considering a new source complete, verify:

• adapter returns the correct ordered active stage list,
• pipeline runs through IEtlPipelineRunner,
• stage outputs are passed through EtlStageContext,
• checkpoint advances only on success,
• failed packages are written on failure,
• curated SQL runs through shared transaction handling if applicable,
• catalogs are internally consistent,
• DI resolves the adapter successfully,
• scheduler or manual entrypoint resolves the adapter through framework seams.

Testing Checklist

Add tests for:

1. stage ordering,
2. upstream result usage,
3. checkpoint advancement,
4. failed package behavior,
5. curated executor success and rollback if used,
6. catalog consistency,
7. adapter stage list,
8. entrypoint wiring.

Use the Brink tests as the template:

• backend/tests/Kds.Ai.Integrations.Tests/Etl/EtlPipelineRunnerTests.cs
• backend/tests/Kds.Ai.Integrations.Tests/Etl/StageExecutionSupportTests.cs
• backend/tests/Kds.Ai.Integrations.Tests/Etl/CuratedTargetExecutorTests.cs
• backend/tests/Kds.Ai.Integrations.Tests/BrinkPos/BrinkPosEtlCatalogTests.cs

Common Mistakes To Avoid

• adding a new hardcoded pipeline job chain instead of using IEtlPipelineRunner,
• moving source-specific parsing or SQL into framework folders,
• duplicating failure/checkpoint logic inside the source job,
• bypassing EtlStageContext and coupling stages directly,
• creating source-specific transaction execution code when shared curated executor already fits,
• leaving catalog ownership unclear.

Minimal Success Definition

A new source is integrated correctly when:

1. the app can resolve the adapter from DI,
2. the entrypoint calls the pipeline runner,
3. stages execute in source-declared order,
4. shared mechanics are reused,
5. source-specific logic stays within the adapter boundary.