# KDS CDC Ingestion Solution

## Purpose

This document describes the chosen production approach for ingesting Logic Control KDS CDC data into KDSAI warehouse storage.

Related notes:

- [Performance And Scaling Notes](./performance-and-scaling-notes.md)
- [Debezium Payload Compatibility Notes](./debezium-payload-compatibility-notes.md)

The chosen approach is a framework-native SQS micro-batch CDC adapter. It follows the same ETL stage model used by the existing Brink POS adapter while replacing source-specific extraction with Debezium CDC messages delivered through cross-account SQS FIFO.

The design goals are:

- preserve the current ETL adapter abstraction,
- keep the existing S3 company/tenant key structure,
- avoid one ingestion batch per CDC record,
- persist accepted CDC data in S3 raw for replay,
- load Redshift through S3 Parquet and `COPY`,
- quarantine bad records without blocking good records,
- support global CDC ordering,
- include curated-to-S3 export from the start.

## Selected Behavior Summary

| Area | Decision |
|---|---|
| Source adapter | New KDS CDC adapter using existing ETL framework stages. |
| Source type | New production source identity, e.g. `kds_cdc` / `KdsCdc`. Do not reuse `SyntheticKds`. |
| Ingestion mode | `Cdc`. |
| Source tenant mapping | Source customer/account maps to `company_id`; source `store_guid` maps to `tenant_id`. |
| S3 layout | Use current S3 layout with company and tenant breakdown. |
| SQS acknowledgement | Delete SQS messages only after raw S3 write and durable internal job creation. |
| Batch size | Configurable, initial target 500 to 2,000 CDC records. |
| Batch latency | Configurable, target 1 to 5 minutes. Longer windows such as 15 minutes are allowed where freshness is less important. |
| Raw batch isolation | Split raw batches by tenant and entity. |
| Load target | Direct source-shaped staging tables. |
| Deletes | Preserve source `is_deleted` and soft-delete rows. |
| Initial snapshot | Ongoing CDC only. Baseline/seed is handled outside this feature. |
| Ordering | Global order. |
| Failed records | Quarantine bad records and load good records. |
| Replay source | S3 raw. |
| Curated export | Include `curated_to_s3` now. |

## Integration Flow

```mermaid
flowchart TD
    A[Logic Control KDS DB] --> B[Debezium connector<br/>source AWS account]
    B --> C[SQS FIFO queue<br/>cross-account]
    C --> D[KDS CDC background worker]
    D --> E[Poll and validate messages]
    E --> F[Resolve source customer/account<br/>and store_guid]
    F --> G{Mapping valid?}
    G -- no --> Q[Quarantine bad record<br/>with raw/source diagnostics]
    G -- yes --> H[Build ordered micro-batch<br/>500-2,000 records or age limit]
    H --> I[extract_raw<br/>write S3 raw by company/tenant/entity]
    I --> J[Create durable internal job/batch]
    J --> K[Delete SQS messages]
    J --> L[normalize_load<br/>write S3 load Parquet]
    L --> M[load_to_staging<br/>Redshift COPY]
    M --> N[staging_to_curated<br/>soft-delete/upsert]
    N --> O[curated_to_s3<br/>Redshift UNLOAD]
    O --> P[S3 curated Parquet]
```

## Stage 1: Debezium To SQS FIFO

Debezium runs in the source AWS account against the Logic Control KDS database. It emits CDC envelopes to SQS FIFO.

The KDSAI side treats SQS as the delivery buffer, not the long-term replay system. SQS messages are temporary transport records. S3 raw is the replay source after acknowledgement.

Expected message content:

- source database/schema/table,
- source operation code,
- source primary key,
- source customer/account identifier,
- source `store_guid` where available,
- source `is_deleted` where available,
- Debezium source coordinates or offset metadata,
- SQS FIFO message group and sequence metadata.

Global order is required. The worker must use FIFO sequence metadata and Debezium source coordinates to detect gaps, duplicates, and out-of-order messages. If global order cannot be proven for a batch, the batch must fail safely before Redshift load.

## Stage 2: Worker Pull And Micro-Batch Creation

The KDS CDC background worker polls SQS FIFO and builds micro-batches.

Batch close triggers:

| Trigger | Selected behavior |
|---|---|
| Record count | Close at configurable limit, initially 500 to 2,000 CDC records. |
| Age | Close at configurable latency target, normally 1 to 5 minutes. |
| Longer windows | A 15-minute window is allowed for lower-freshness environments. |
| Byte/memory pressure | Close before visibility timeout or worker memory pressure. |
| Tenant/entity split | Raw and load artifacts are split by company, tenant, and entity. |

The worker must not create one ingestion batch per CDC record. One ETL batch represents a bounded set of ordered CDC records for a tenant/entity slice.

Bad records are isolated during pull/validation when possible. A malformed record or unmapped tenant record is quarantined with diagnostics, while good records continue into the batch.

## Stage 3: Tenant Resolution

Tenant resolution happens before any tenant-scoped S3 path or Redshift load is created.

Mapping rule:

```text
source customer/account -> company_id
source store_guid -> tenant_id within company_id
company_id + tenant_id -> tenant Redshift database
```

The source field names do not need to be named `company_id` or `tenant_id`. They must map explicitly to platform values through a registered mapping table or mapping service.

No implicit tenant inference is allowed. If mapping is missing or ambiguous, the record is quarantined and not loaded to Redshift.

Required quarantine metadata for mapping failures:

- source customer/account value,
- source `store_guid`,
- source table,
- source primary key,
- Debezium source coordinates,
- SQS message id and sequence metadata,
- raw S3 path when available,
- reason mapping failed.

## Stage 4: Store To S3 Raw (`extract_raw`)

`extract_raw` persists accepted CDC records to S3 raw.

Raw storage keeps the existing ETL S3 company/tenant breakdown. Raw objects are split by tenant and entity so replay, auditing, and failure isolation stay tenant-safe.

Conceptual raw key shape:

```text
raw/company_id={company_id}/tenant_id={tenant_id}/source_type=kds_cdc/{yyyy}/{MM}/{dd}/batch_id={batch_id}/entity={entity}/...
```

Raw object format should preserve the Debezium envelope faithfully, such as NDJSON or compressed JSON. The raw object is not just transformed row data; it must contain enough source metadata to replay normalization after a code or mapping fix.

Raw object contents should include:

- original CDC payload,
- operation code,
- source coordinates,
- source table and primary key,
- resolved `company_id` and `tenant_id`,
- SQS message id/group/sequence metadata,
- batch id and worker metadata.

SQS messages are not deleted immediately after polling. They are deleted only after:

1. raw S3 write succeeds,
2. durable internal job or batch metadata is created.

After that point, S3 raw plus internal job metadata is the replay source of truth.

## Stage 5: Durable Internal Job Creation

After raw S3 succeeds, the worker creates durable internal metadata for downstream ETL work.

This metadata should connect:

- raw batch id,
- company id,
- tenant id,
- tenant database,
- entity/table,
- source type `kds_cdc`,
- ingestion mode `Cdc`,
- pipeline stage,
- S3 raw path,
- Debezium coordinate range,
- SQS sequence range,
- status and retry information.

The durable job boundary is important because SQS messages can be deleted only after the system has enough internal state to resume from S3 raw.

## Stage 6: Normalize To S3 Load (`normalize_load`)

`normalize_load` reads S3 raw and writes load-ready Parquet files to S3 load.

The stage maps CDC envelopes into rows shaped for the existing KDS staging tables:

- `staging.kds_orders`,
- `staging.kds_items`,
- `staging.kds_item_bumps`,
- `staging.kds_customers`.

Conceptual load key shape:

```text
load/company_id={company_id}/tenant_id={tenant_id}/entity={entity}/batch_id={batch_id}/created_at={timestamp}/...
```

Normalization behavior:

- split output by tenant and entity,
- preserve source `is_deleted`,
- preserve CDC source coordinates in metadata columns or batch diagnostics,
- deduplicate exact duplicate CDC messages by source coordinates and primary key,
- keep global order metadata for downstream validation,
- quarantine record-local parse/mapping errors and continue with good records,
- fail the tenant/entity batch only when a safe load artifact cannot be produced.

The load artifact is an optimization checkpoint. S3 raw remains the source of truth.

## Stage 7: Load To Source-Shaped Staging (`load_to_staging`)

`load_to_staging` resolves the tenant Redshift database and runs Redshift `COPY` from S3 load Parquet into source-shaped staging tables.

The selected approach does not introduce separate CDC landing tables in the first version. It loads directly into the current KDS staging model.

Load behavior:

- use Redshift `COPY`, not row-by-row inserts,
- load one tenant/entity artifact at a time,
- apply idempotent delete/insert or merge behavior around the copied batch,
- preserve `is_deleted` for soft-delete handling,
- validate global ordering before applying the batch,
- fail safely if a gap or out-of-order sequence is detected,
- retry idempotently from the same S3 load artifact if `COPY` fails.

This stage must never load records whose tenant mapping failed or was ambiguous.

## Stage 8: Staging To Curated (`staging_to_curated`)

`staging_to_curated` transforms source-shaped KDS staging rows into curated warehouse tables.

Current curated targets include:

- `curated.kitchen_orders`,
- `curated.kitchen_order_items`,
- `curated.kitchen_item_status_events`,
- `curated.customers` where applicable.

Curated behavior:

- apply idempotent upsert semantics,
- preserve source system identity as production KDS CDC, not synthetic KDS,
- apply soft deletes from source `is_deleted`,
- keep transaction sizes bounded by batch size,
- do not advance checkpoint if curated merge fails,
- allow retry from staging/load/raw depending on failure point.

## Stage 9: Curated To S3 (`curated_to_s3`)

KDS CDC includes curated-to-S3 export from the start.

This stage exports updated curated Redshift tables to S3 curated Parquet, matching the current ETL architecture and enabling cross-tenant/Athena query patterns.

Conceptual curated key shape:

```text
curated/company_id={company_id}/tenant_id={tenant_id}/entity={entity}/dt={date}/batch_id={batch_id}/...
```

Export behavior:

- use Redshift `UNLOAD`,
- export after curated merge succeeds,
- preserve tenant/entity partitioning,
- mark stage success only after export succeeds,
- clean up incomplete curated artifacts on failure when possible,
- retry from curated stage output without re-reading SQS.

## Batch Lifecycle And Lineage

Each of the 5 framework stages (`extract_raw`, `normalize_load`, `load_to_staging`, `staging_to_curated`, `curated_to_s3`) has its own `IngestionBatch` lifecycle. The worker does not create one batch for the whole pipeline.

### Per-Stage Batch Creation

Every stage calls `IIngestionBatchLifecycleService.StartStageBatchAsync` before doing work and `CompleteStageBatchAsync` after success. On failure, `StageExecutionSupport.FailStageBatchAsync` marks the batch as `Failed` and writes a failed package.

Each stage batch carries:

- `BatchId` (new per stage),
- `CompanyId`, `TenantId`, `TenantDatabase`,
- `IngestionMode` = `Cdc`,
- `SourceType` = `KdsCdc`,
- `PipelineStage` = the current stage name,
- `BusinessDate`,
- `RawS3Path` / `LoadS3Path` where applicable,
- `Status` = `Pending` -> `Running` -> `Succeeded` or `Failed`.

### Upstream Batch Chaining

`IngestionBatch.UpstreamBatchId` links each stage to its predecessor:

```text
extract_raw batch
  └── normalize_load batch (UpstreamBatchId = extract_raw BatchId)
        └── load_to_staging batch (UpstreamBatchId = normalize_load BatchId)
              └── staging_to_curated batch (UpstreamBatchId = load_to_staging BatchId)
                    └── curated_to_s3 batch (UpstreamBatchId = staging_to_curated BatchId)
```

This lineage enables replay from any stage by following the chain back to the raw batch. It also enables audit queries like "which raw batch produced this curated row."

### Explicit Batch Completion

After each stage succeeds, the worker calls `CompleteStageBatchAsync` to mark the batch as `Succeeded`. Checkpoint advancement via `StageExecutionSupport.AdvanceCheckpointAsync` happens after completion, not before. If a stage fails, the batch is marked `Failed` and subsequent stages do not run.

### BusinessDate Determination

CDC has no natural business date — it processes ongoing changes. The adapter uses the processing date:

```text
BusinessDate = DateOnly.FromDateTime(DateTime.UtcNow)
```

This matches the framework requirement that `EtlStageResult` returns a `DateOnly BusinessDate`. The actual CDC progress is tracked in `CheckpointValue` (Debezium offset/source coordinates), not in `BusinessDate`. Daily rotation of `BusinessDate` is acceptable because:

- `IngestionBatch.ReceivedAt` already captures the exact processing timestamp,
- `IngestionCheckpoint.CheckpointValue` holds the real Debezium offset for replay,
- checkpoint keys rotate daily per tenant/source/stage, matching the Brink POS pattern.

## Checkpointing And Replay

Checkpointing must use source coordinates, not wall-clock time alone.

Checkpoint metadata should include:

- Debezium connector/source identity,
- source database/schema/table,
- Debezium offset/source coordinates,
- SQS message group id,
- SQS sequence number or message id,
- global order range,
- last successful raw batch id,
- last successful load batch id,
- last successful curated batch id,
- last successful curated export batch id.

S3 raw is the replay source of truth. Replay must support:

| Replay level | Use case |
|---|---|
| Raw batch replay | Re-run one failed raw batch through normalize/load/curated/export. |
| Tenant/entity replay | Reprocess a bounded tenant/entity slice from raw S3. |
| Coordinate replay | Reprocess from a Debezium source coordinate range when available. |
| Load artifact retry | Retry Redshift `COPY` when normalization already succeeded. |
| Curated retry | Retry curated merge/export when staging load already succeeded. |

Checkpoints advance only after the configured durable success point for the stage. If SQS messages have already been deleted, replay must never depend on SQS retention.

## Quarantine And Failed Packages

Bad records are quarantined; good records continue when safe.

Quarantine applies to record-local failures such as:

- malformed Debezium envelope,
- missing source primary key,
- missing or ambiguous tenant mapping,
- unsupported source table,
- invalid value that prevents row mapping,
- duplicate message that cannot be reconciled safely.

Batch-level failure applies when safe partial processing is not possible, such as:

- raw S3 write failure,
- internal job metadata creation failure,
- ordering gap or out-of-order condition,
- Parquet artifact cannot be produced safely,
- Redshift `COPY` fails,
- curated merge fails,
- curated export fails.

Quarantine/failed package metadata should include:

- failure reason,
- source payload or pointer to raw payload,
- source table and primary key,
- Debezium operation and coordinates,
- SQS message id/group/sequence,
- resolved or unresolved company/tenant mapping,
- raw S3 path,
- load S3 path when available,
- pipeline stage,
- batch id,
- retry/replay guidance.

## Supported Corner Cases

### Missing Tenant Mapping

If source customer/account or `store_guid` cannot be mapped, the record is quarantined. It is not written to any tenant Redshift database. Other valid records continue.

### Ambiguous Tenant Mapping

If one source identity maps to multiple possible tenants, the record is quarantined. No best-effort tenant selection is allowed.

### Bad Record In Good Batch

A malformed or unmapped record is quarantined when the failure is record-local. Good records in the same poll window continue through raw, load, staging, curated, and export.

### Raw S3 Write Failure

If raw S3 write fails, SQS messages are not deleted. SQS visibility timeout and retry handle redelivery.

### Internal Job Creation Failure

If raw S3 succeeds but durable internal job creation fails, SQS messages are not deleted. A retry may write duplicate raw objects; downstream dedupe must use source coordinates and batch metadata.

### Duplicate CDC Message

Duplicate messages are detected by Debezium source coordinates plus source table and primary key. Replays must be idempotent.

### Out-Of-Order Or Missing Sequence

Global order is required. If the worker detects a sequence gap or out-of-order batch, it must fail safely before Redshift load and preserve diagnostics for replay.

### Source Delete Or Soft Delete

Source `is_deleted` is preserved. Curated rows are soft-deleted rather than physically deleted in the normal path.

### Redshift COPY Failure

Raw and load artifacts remain durable. Retry uses the same S3 load artifact and must be idempotent.

### Curated Merge Failure

Checkpoint does not advance. The system retries after SQL/data correction from staging/load/raw depending on the failure.

### Curated Export Failure

Curated Redshift data remains the committed warehouse state. The export stage can retry `UNLOAD`; incomplete S3 curated artifacts should be cleaned up when possible.

### SQS Message Redelivery After Raw Write

If SQS redelivers messages after raw write, dedupe uses source coordinates and SQS metadata. Duplicate raw batches must not corrupt staging or curated state.

### Low Volume Tenant

The batch age limit closes a batch even when record count is below target, so quiet tenants still receive data within the configured latency window.

### High Volume Tenant

The record count and byte limits bound transaction size and worker memory. Large streams become multiple ordered micro-batches.

### Ongoing CDC Without Snapshot

This feature only handles ongoing CDC. Production use assumes a baseline seed or backfill exists before CDC is considered complete history.

## Non-Goals

- Do not reuse the simulator publisher as the production CDC ingestion path.
- Do not identify production CDC as `SyntheticKds`.
- Do not create one ingestion batch per CDC record.
- Do not rely on SQS as long-term replay after messages are acknowledged.
- Do not introduce CDC landing tables in the first version.
- Do not physically delete curated rows in the normal KDS CDC path.