# 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:

```text
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:

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

Recommended adapter shape:

```csharp
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:

```csharp
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.