TASK 1: Establish KDS CDC Adapter Identity And Pipeline Integration

Summary

Establish the production KDS CDC adapter as a first-class ETL framework source.

This task creates the implementation seam for all later KDS CDC work by freezing:

• source identity,
• ingestion mode,
• stage list,
• adapter registration,
• runtime adapter resolution pattern,
• separation from synthetic KDS bootstrap code.

This task does not implement CDC message intake, tenant mapping, raw persistence, or Redshift logic yet.

Current Codebase Findings

Repository analysis shows the existing ETL framework already has the core runtime seam KDS CDC should reuse:

• IEtlSourceAdapter exposes Descriptor plus ordered GetStages().
• EtlPipelineRunner executes adapter-declared stages in order and carries prior stage outputs through EtlStageContext.
• EtlStageNames already defines all five required stage names: extract_raw, normalize_load, load_to_staging, staging_to_curated, and curated_to_s3.
• BrinkPosSourceAdapter is current reference implementation for a full five-stage adapter.
• DependencyInjectionExtensions.AddBrinkPosEtlModule() registers stage wrappers plus IEtlSourceAdapter using DI.
• BrinkPosEtlJobs resolves the adapter by candidate.Descriptor.SourceType == "brink_pos" and passes it into IEtlPipelineRunner.

Repository analysis also shows why a new identity is required:

• IngestionMode.Cdc already exists and can be reused directly.
• IngestionSourceType currently contains only SyntheticKds and BrinkOrders; no production KDS CDC enum exists yet.
• SyntheticKdsFeedService is a separate direct warehouse-feed path with IngestionMode.SyntheticFeed and IngestionSourceType.SyntheticKds.
• synthetic KDS code writes directly through warehouse services and is not an ETL source adapter.
• tenant warehouse bootstrap currently ties KDS staging resources to Brink POS credential/service enablement in existing warehouse bootstrap code, so KDS CDC must introduce its own enablement path rather than relying on Brink onboarding.

This means Task 1 should add a new production identity and adapter registration rather than extend SyntheticKdsFeedService or reuse SyntheticKds enum/storage identity.

Dependencies

• docs/features/ETL/kds-cdc/index.md
• docs/features/ETL/kds-cdc/implementation-breakdown.md
• existing ETL framework from specs/EtlSourceAdapterFramework

Source References

• ../../docs/features/ETL/kds-cdc/implementation-breakdown.md#task-checklist
• ../../docs/features/ETL/kds-cdc/index.md#selected-behavior-summary
• ../../docs/features/ETL/kds-cdc/index.md#batch-lifecycle-and-lineage
• ../../docs/features/ETL/kds-cdc/index.md#non-goals
• ../../docs/features/ETL/etl-source-adapter-framework.md
• ../../docs/features/ETL/how-to-add-etl-source-adapter.md

Scope From Breakdown

This task covers only the first implementation-breakdown item:

• introduce production source identity for KDS CDC and keep it separate from SyntheticKds
• register adapter in existing ETL framework with ingestion mode Cdc
• define stage list and adapter wiring so KDS CDC follows framework-native stage execution

Files to Create

1. KDS CDC adapter package

Create a dedicated KDS CDC adapter package under the ETL integration layer.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/

Minimum logical types:

• KdsCdcSourceAdapter
• KdsCdcExtractStage
• KdsCdcNormalizeLoadStage
• KdsCdcLoadToStagingStage
• KdsCdcStagingToCuratedStage
• KdsCdcCuratedToS3Stage
• KdsCdcEtlStageConstants only if truly needed for source-local constants not already covered by EtlStageNames

Important rule:

• stage wrappers should use EtlStageNames for stage names,
• source-local constants should only hold KDS CDC identity values or contract-version details,
• do not duplicate shared stage-name literals in a new source-local constants class unless there is a concrete need beyond current framework contracts.

2. KDS CDC entrypoint seam

Create the first runtime entrypoint seam for KDS CDC pipeline invocation.

Exact host type may vary, but the code should make room for a CDC-native entrypoint such as:

• background worker,
• job host wrapper,
• poller orchestration service.

Recommended logical types:

• IKdsCdcPipelineLauncher
• KdsCdcPipelineLauncher
• or a KDS CDC worker/job class that owns adapter selection and IEtlPipelineRunner invocation.

This seam should be distinct from Brink scheduler/manual jobs because KDS CDC runtime trigger semantics are different.

3. KDS CDC pipeline request/result models if needed

If current EtlPipelineRequest is too Brink-shaped for CDC caller clarity, add thin KDS CDC-specific request models that translate into the shared pipeline request.

Do not redesign the framework request shape in this task unless an actual blocker appears.

Files to Modify

4. Source identity enum

Modify:

• backend/Kds.Ai.Data/Enums/IngestionSourceType.cs

Add a new production enum value for KDS CDC.

Expected value:

• KdsCdc

Do not rename or repurpose SyntheticKds.

5. DI registration

Modify:

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

Add KDS CDC stage wrappers, adapter registration, and any first entrypoint or worker services needed for runtime composition.

Important DI rule:

• KDS CDC should register as another IEtlSourceAdapter, alongside Brink,
• do not replace the existing Brink adapter registration,
• do not collapse both sources into one adapter.

6. Adapter selection call sites

Review and update any call sites that assume only Brink exists.

Confirmed current call site:

• backend/Kds.Ai.ApiService/Jobs/BrinkPosEtlJobs.cs

Current pattern is acceptable for Brink because it filters by Descriptor.SourceType == "brink_pos".

Task 1 should establish the same selection pattern for KDS CDC callers, typically by resolving sourceAdapters.Single(...) for "kds_cdc".

Do not force a premature generic source-dispatch registry unless a second caller already needs it in this task.

7. ETL framework tests

Modify or add tests under:

• backend/tests/Kds.Ai.Integrations.Tests/Etl/
• backend/tests/Kds.Ai.Integrations.Tests/KdsCdc/ if a new source-specific test folder is introduced

8. KDS CDC service/bootstrap enablement

Review warehouse/bootstrap enablement seams so KDS CDC can provision required KDS staging/curated resources without depending on CustomerCredentialService.BrinkPos.

Expected areas:

• customer credential/service enum or equivalent source enablement metadata,
• tenant warehouse bootstrap resource-prefix selection,
• AppHost/local configuration for enabling KDS CDC in development.

Preferred direction:

• add explicit KDS CDC service/source enablement, such as CustomerCredentialService.KdsCdc, or another narrowly scoped warehouse bootstrap flag if credential-service enum is not the right abstraction.

Required Behavior

Source identity

KDS CDC must have both:

• storage/source string identity: kds_cdc
• typed enum identity: IngestionSourceType.KdsCdc

These identities must flow consistently through:

• EtlSourceDescriptor,
• stage batch creation requests,
• checkpoint requests,
• failed package metadata,
• future curated _etl_source_type writes.

Ingestion mode

KDS CDC stage jobs must use IngestionMode.Cdc.

This is already supported by backend/Kds.Ai.Data/Enums/IngestionMode.cs and should not be emulated via ApiPull or SyntheticFeed.

Stage list

KDS CDC adapter must expose exactly five stages in framework order:

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

This ordering must be returned from IEtlSourceAdapter.GetStages().

The stage list must match the main KDS CDC feature doc and the current framework capability.

Framework-native runtime execution

KDS CDC must execute through:

• IEtlPipelineRunner
• IEtlSourceAdapter
• IEtlStage
• EtlStageContext
• EtlStageResult

Do not implement a custom internal loop for stage chaining.

Separation from synthetic KDS

The new KDS CDC implementation must not:

• reuse IngestionSourceType.SyntheticKds,
• reuse IngestionMode.SyntheticFeed,
• extend SyntheticKdsFeedService as its orchestration mechanism,
• identify production CDC rows or batches as synthetic KDS.

The synthetic feed remains a separate bootstrap path.

Stage wrapper behavior for task 1

Because downstream CDC logic is not implemented yet, stage wrapper behavior may begin as one of two acceptable shapes:

1. thin wrappers over not-yet-implemented KDS CDC job services that define final contracts and throw intentional NotImplementedException, or
2. thin wrappers over placeholder services that return task-specific temporary results used only to prove registration and ordering in tests.

Preferred rule:

• define the final wrapper and job-service seams now,
• keep placeholder behavior minimal,
• avoid fake business logic that later tasks must delete.

Batch lifecycle alignment

Task 1 should freeze the expectation that each KDS CDC stage will own its own IngestionBatch and chain through UpstreamBatchId like the main feature doc requires.

Task 1 does not need to implement all stage job details, but the public seams and acceptance criteria must reflect that per-stage lifecycle model.

Business date rule

KDS CDC has no source business date. The adapter must plan for:

BusinessDate = DateOnly.FromDateTime(DateTime.UtcNow)

This rule becomes authoritative for KDS CDC stage jobs.

Task 1 does not need full implementation, but it should not leave room for Brink-style business-date resolution semantics to leak into CDC.

Proposed Implementation Shape

Recommended descriptor:

public EtlSourceDescriptor Descriptor { get; } = new("kds_cdc", IngestionSourceType.KdsCdc);

Recommended adapter shape:

internal sealed class KdsCdcSourceAdapter(
    KdsCdcExtractStage extractStage,
    KdsCdcNormalizeLoadStage normalizeLoadStage,
    KdsCdcLoadToStagingStage loadToStagingStage,
    KdsCdcStagingToCuratedStage stagingToCuratedStage,
    KdsCdcCuratedToS3Stage curatedToS3Stage) : IEtlSourceAdapter
{
    private readonly IReadOnlyList<IEtlStage> _stages =
    [
        extractStage,
        normalizeLoadStage,
        loadToStagingStage,
        stagingToCuratedStage,
        curatedToS3Stage
    ];

    public EtlSourceDescriptor Descriptor { get; } = new("kds_cdc", IngestionSourceType.KdsCdc);

    public IReadOnlyList<IEtlStage> GetStages() => _stages;
}

Recommended entrypoint pattern:

var adapter = sourceAdapters.Single(static candidate =>
    string.Equals(candidate.Descriptor.SourceType, "kds_cdc", StringComparison.Ordinal));

await pipelineRunner.RunAsync(
    new EtlPipelineRequest(
        companyId,
        tenantId,
        businessDate: null,
        trigger,
        forceReset,
        requestedBy,
        adapter),
    ct);

Important note:

• businessDate: null here reflects that caller does not supply a business date.
• stage-job implementations should derive processing date internally in later tasks.
• do not change the shared pipeline contract only to force a CDC business date placeholder into task 1.

Testing

Unit tests

Add unit tests for:

• new IngestionSourceType.KdsCdc presence and KDS CDC descriptor usage where practical,
• KDS CDC adapter descriptor equals ("kds_cdc", IngestionSourceType.KdsCdc),
• KDS CDC adapter returns five stages in correct order,
• each KDS CDC wrapper exposes correct StageName from EtlStageNames,
• DI resolves both Brink and KDS CDC IEtlSourceAdapter registrations together,
• KDS CDC entrypoint resolves adapter by "kds_cdc" and invokes IEtlPipelineRunner with that adapter,
• no test path identifies KDS CDC as SyntheticKds.

Integration or composition tests

Add at least one composition-focused test proving:

• host or service provider can resolve IEnumerable<IEtlSourceAdapter>,
• the KDS CDC adapter is present,
• stage ordering is stable,
• existing Brink adapter remains present.

Not required in task 1

This task does not require:

• live SQS integration,
• live AWS tests,
• Redshift tests,
• stage lifecycle persistence tests beyond what adapter identity requires.

Acceptance Criteria

• [ ] IngestionSourceType.KdsCdc exists as new production source enum value
• [ ] KDS CDC uses source string kds_cdc
• [ ] KDS CDC uses IngestionMode.Cdc
• [ ] A dedicated KdsCdcSourceAdapter exists
• [ ] KDS CDC adapter implements IEtlSourceAdapter
• [ ] KDS CDC adapter returns five framework stages
• [ ] KDS CDC stage order is extract_raw -> normalize_load -> load_to_staging -> staging_to_curated -> curated_to_s3
• [ ] KDS CDC stage wrappers use shared EtlStageNames
• [ ] KDS CDC runtime composition is registered through DI
• [ ] DI resolves KDS CDC alongside Brink POS, not instead of Brink POS
• [ ] A KDS CDC pipeline entry seam exists for future worker/job invocation
• [ ] KDS CDC entry seam selects adapter by Descriptor.SourceType == "kds_cdc"
• [ ] KDS CDC warehouse/bootstrap enablement is separate from Brink POS credential/service enablement
• [ ] No production CDC path reuses SyntheticKdsFeedService as its adapter runtime
• [ ] No production CDC path uses IngestionSourceType.SyntheticKds
• [ ] No production CDC path uses IngestionMode.SyntheticFeed
• [ ] Tests cover adapter descriptor, stage ordering, DI registration, and source separation
• [ ] Solution builds without errors

Notes

• Follow docs/features/ETL/kds-cdc/index.md as source of truth for source identity and stage model.
• Use Brink POS adapter as structural reference, not as semantic template for scheduling or business-date behavior.
• Keep Task 1 minimal: identity and integration seams only.
• Prefer thin placeholder job services over speculative CDC logic.
• Preserve room for a future CDC-native worker that polls SQS and then invokes the shared pipeline.