# TASK 7: Implement Curated Transformation And Curated-To-S3 Export

## Summary

Implement KDS CDC `staging_to_curated` and `curated_to_s3` as framework-native downstream stages after successful source-shaped staging load. This task defines curated merge boundaries, source-identity and soft-delete handling, curated target and export catalogs, and Redshift `UNLOAD` export behavior for initial delivery. It excludes final checkpoint/replay orchestration beyond the durable details these stages must persist.

## Current Codebase Findings

- `docs/features/ETL/kds-cdc/index.md` Stage 8 requires idempotent staging-to-curated merges into `curated.kitchen_orders`, `curated.kitchen_order_items`, `curated.kitchen_item_status_events`, and `curated.customers` where applicable.
- Stage 8 also fixes behavior: preserve production KDS CDC identity, apply source `is_deleted` as soft-delete semantics, keep transaction sizes bounded by batch size, and do not advance checkpoint if curated merge fails.
- Stage 9 requires `curated_to_s3` from first delivery, using Redshift `UNLOAD` into tenant/entity-partitioned curated Parquet under the existing ETL S3 layout.
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosStagingToCuratedJobs.cs` is current framework-native reference for curated-stage job orchestration:
  - resolve upstream successful `load_to_staging` batch,
  - start a stage-scoped `IngestionBatch`,
  - execute curated targets through `ICuratedTargetExecutor`,
  - persist structured stage details,
  - complete batch, then advance checkpoint,
  - fail the batch through `StageExecutionSupport.FailStageBatchAsync` on exception.
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosCuratedToS3Jobs.cs` is current framework-native reference for export-stage orchestration:
  - resolve upstream successful `staging_to_curated` batch,
  - start a stage-scoped `IngestionBatch`,
  - execute export targets through `ICuratedExportExecutor`,
  - persist structured export details,
  - complete batch, then advance checkpoint,
  - fail the batch on exception using warehouse-curated failure context.
- `backend/Kds.Ai.Integrations/Etl/CuratedTargetExecutor.cs` provides a generic transactional executor that runs each target's `DeleteSql` then `WriteSql` inside one Redshift transaction using `tenant_id`, `business_date`, and `snapshot_date` parameters.
- `backend/Kds.Ai.Integrations/Etl/CuratedExportExecutor.cs` provides a generic Redshift `UNLOAD` executor that:
  - builds curated S3 prefix URIs through `IEtlS3Storage.BuildCuratedUri(...)`,
  - uses explicit `SELECT` SQL per target,
  - writes `PARQUET ALLOWOVERWRITE`,
  - cleans up already-exported prefixes if a later export target fails.
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosCuratedTargetCatalog.cs` shows repo pattern for curated-target catalogs: explicit target names, required staging tables, SQL template ids, delete SQL, insert SQL, and source-type filtering.
- `backend/Kds.Ai.Integrations/BrinkPos/BrinkPosCuratedExportCatalog.cs` shows repo pattern for curated export catalogs: explicit `SELECT` projection, export entity name, file-prefix naming, and source-type/date filters.
- `backend/Kds.Ai.Warehouse/Services/Synthetic/RedshiftCuratedOrderWriter.cs` is current KDS-specific curated data reference, but it is a legacy synthetic runtime path. It directly mutates `curated.kitchen_orders`, `curated.kitchen_order_items`, and `curated.kitchen_item_status_events` using `SourceSystem = "synthetic_kds"`, id-based delete/reinsert logic, and post-write orphan validation. It is useful for KDS table relationships and field parity only, not for production CDC runtime shape.
- `backend/Kds.Ai.Warehouse/Models/Curated/CuratedTableNames.cs` already includes `KitchenOrders`, `KitchenOrderItems`, `KitchenItemStatusEvents`, and `Customers`.
- Repo evidence currently shows bootstrap/runtime coverage for `curated.kitchen_orders`, `curated.kitchen_order_items`, and `curated.kitchen_item_status_events`, but no `CREATE TABLE curated.customers` SQL or corresponding runtime/catalog/export/test path was found.

Current repo constraints this task must respect:

1. Curated KDS production runtime must move through framework-native `staging_to_curated` and `curated_to_s3` stages, not the old synthetic direct-writer path.
2. Existing generic curated executors are reusable, but KDS CDC needs its own catalogs, details models, and SQL semantics because CDC row deltas and soft deletes are not Brink snapshot semantics.
3. `curated.customers` is part of the documented target set, but repo does not show a finished schema/runtime/export path for it today, so Task 7 must make an explicit implementation-or-conditional-scope decision instead of assuming it already exists.

## Dependencies

- `../../docs/features/ETL/kds-cdc/index.md`
- `../../docs/features/ETL/kds-cdc/implementation-breakdown.md`
- `../../docs/features/ETL/etl-source-adapter-framework.md`
- `./TASK6.md`

## Source References

- `../../docs/features/ETL/kds-cdc/implementation-breakdown.md#task-checklist`
- `../../docs/features/ETL/kds-cdc/index.md#stage-8-staging-to-curated-staging_to_curated`
- `../../docs/features/ETL/kds-cdc/index.md#stage-9-curated-to-s3-curated_to_s3`
- `../../docs/features/ETL/kds-cdc/index.md#selected-behavior-summary`
- `../../docs/features/ETL/kds-cdc/index.md#checkpointing-and-replay`
- `../../docs/features/ETL/kds-cdc/index.md#quarantine-and-failed-packages`
- `../../docs/features/ETL/kds-cdc/index.md#supported-corner-cases`
- `../../docs/features/ETL/etl-source-adapter-framework.md`

## Scope From Breakdown

- apply idempotent staging-to-curated merges for KDS warehouse targets
- preserve production KDS CDC source identity and soft-delete semantics in curated layer
- export curated data to tenant/entity-partitioned S3 Parquet with Redshift `UNLOAD`
- treat curated export as part of first delivery, not later follow-up work

## Files to Create

1. KDS CDC curated-stage jobs and details under `backend/Kds.Ai.Integrations/KdsCdc/` and `/Models/`:
   - `IKdsCdcStagingToCuratedJobs`
   - `KdsCdcStagingToCuratedJobs`
   - `IKdsCdcCuratedToS3Jobs`
   - `KdsCdcCuratedToS3Jobs`
   - `KdsCdcStagingToCuratedDetails`
   - `KdsCdcCuratedTargetDiagnostic`
   - `KdsCdcCuratedToS3Details`
   - `KdsCdcCuratedExportDiagnostic`

   Responsibilities:
   - resolve successful upstream batches from Task 6,
   - select executable curated targets,
   - run framework-native curated merge and export executors,
   - persist structured details for replay and diagnostics,
   - complete stage batches only after merge/export success.

2. KDS CDC curated target catalog under `backend/Kds.Ai.Integrations/KdsCdc/`:
   - `KdsCdcCuratedTargetCatalog`
   - optional `KdsCdcCuratedTargetDefinition`

   Responsibilities:
   - define curated targets for `kitchen_orders`, `kitchen_order_items`, `kitchen_item_status_events`, and `customers` where applicable,
   - declare required staging tables,
   - define SQL template ids,
   - define idempotent delete/upsert SQL that preserves production source identity and soft-delete behavior.

3. KDS CDC curated export catalog under `backend/Kds.Ai.Integrations/KdsCdc/`:
   - `KdsCdcCuratedExportCatalog`
   - optional `KdsCdcCuratedExportDefinition`

   Responsibilities:
   - define explicit export targets and `SELECT` projections,
   - freeze export entity names and output-prefix naming,
   - filter exported rows to production KDS CDC source identity,
   - preserve tenant/entity partitioning for S3 curated exports.

4. Optional KDS CDC curated SQL helpers under `backend/Kds.Ai.Integrations/KdsCdc/Sql/` if query text becomes too large for one catalog file:
   - keep helper scope minimal,
   - prefer explicit target-owned SQL over generic abstraction if only KDS CDC uses it.

## Files to Modify

- KDS CDC adapter stage wrappers from Task 1 in `backend/Kds.Ai.Integrations/KdsCdc/` and `backend/Kds.Ai.Integrations/DependencyInjectionExtensions.cs` so `staging_to_curated` and `curated_to_s3` are real framework stages using `EtlStageNames.StagingToCurated` and `EtlStageNames.CuratedToS3`.
- Task 6 load-to-staging details/contracts under `backend/Kds.Ai.Integrations/KdsCdc/` so curated jobs can deterministically select executable targets from durable upstream staging results.
- Shared curated executor/catalog seams only if additive capability is truly required:
  - `backend/Kds.Ai.Integrations/Etl/CuratedTargetExecutor.cs`
  - `backend/Kds.Ai.Integrations/Etl/CuratedExportExecutor.cs`
  - `backend/Kds.Ai.Integrations/Etl/ICuratedTargetCatalog.cs`
  - `backend/Kds.Ai.Integrations/Etl/ICuratedExportCatalog.cs`

  Guidance: prefer KDS CDC-specific catalogs and details rather than broadening shared abstractions unless a missing shared feature is clearly generic.
- Warehouse bootstrap/schema/runtime seams under `backend/Kds.Ai.Warehouse/Services/`, `backend/Kds.Ai.WarehouseDataModel/`, and related tests if `curated.customers` or KDS CDC-specific curated columns do not already exist or are not yet asserted.
- KDS CDC service/bootstrap enablement so curated dependencies for KDS CDC tenants are provisioned independently of Brink POS credentials.
- Tests under:
  - `backend/tests/Kds.Ai.Integrations.Tests/KdsCdc/`
  - `backend/tests/Kds.Ai.Data.Tests/`
  - additive warehouse/bootstrap/export tests where needed.

## Required Behavior

### Staging-To-Curated Boundary

- `staging_to_curated` transforms source-shaped KDS staging rows into curated warehouse tables after successful Task 6 load-to-staging.
- Current curated target set for this task is:
  - `curated.kitchen_orders`
  - `curated.kitchen_order_items`
  - `curated.kitchen_item_status_events`
  - `curated.customers` where applicable
- Stage execution must remain framework-native:
  - resolve upstream `load_to_staging` batch,
  - start a new `IngestionBatch` for `staging_to_curated`,
  - run curated SQL through `ICuratedTargetExecutor` or a clearly justified additive extension,
  - persist structured details,
  - complete batch,
  - advance checkpoint only after completion.

### Curated Merge Semantics

- Curated merges must be idempotent for retry from staging/load/raw after failures.
- Curated rows must preserve production source identity as `kds_cdc` / `KdsCdc`, not synthetic KDS.
- Source `is_deleted` must flow into curated soft-delete behavior.
- Normal KDS CDC path must not physically delete curated rows as business behavior.
- Transaction sizes must remain bounded by micro-batch size and tenant/entity slice boundaries.
- If curated merge fails, checkpoint must not advance and `curated_to_s3` must not run for that slice.

### KDS Curated SQL Shape

- SQL should be explicit per curated target, following existing catalog style.
- Preferred direction is idempotent delete/reinsert or merge-like behavior scoped to keys present in the current CDC slice, not broad tenant-day replacement.
- Any delete step used for idempotence must preserve soft-delete semantics in final curated state.
- Relationship integrity between orders, items, and item status events must remain safe under partial row updates and retries.
- `RedshiftCuratedOrderWriter.cs` may be used as field-parity and dependency-order reference for kitchen tables, especially parent/child delete ordering and orphan validation, but production KDS CDC runtime must not call that synthetic writer.

### Customers Gap

- `curated.customers` is part of documented target scope and must remain explicit in this task.
- Since current repo evidence did not show `CREATE TABLE curated.customers`, Task 7 must choose and document one of two outcomes:
  - implement `curated.customers` DDL, target catalog, merge SQL, export projection, and tests now,
  - or mark KDS customer curated output as explicitly conditional with a documented blocker and ensure runtime skips only missing customer artifacts, not all customer support silently.
- If repo bootstrap/runtime support for `curated.customers` is incomplete and no blocker is accepted, Task 7 must close that gap with additive schema/catalog/export work rather than silently excluding customers.
- If customer events are absent for some tenants or early mock flows, the runtime may skip export/merge for missing upstream customer staging artifacts, but target definitions and stage behavior must still be designed now.

### Curated-To-S3 Boundary

- `curated_to_s3` is part of first delivery, not follow-up work.
- Export must happen only after curated merge succeeds for the same tenant/entity slice lineage.
- Use Redshift `UNLOAD` to export curated data to S3 curated Parquet.
- Preserve tenant/entity partitioning with conceptual key shape:

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

- Stage success is recorded only after export succeeds.
- On export failure, curated Redshift data remains committed, incomplete curated S3 artifacts should be cleaned up when possible, and retry should be possible from curated-stage output without re-reading SQS.

### Source Filtering And Export Contract

- Curated merge and export queries must filter or stamp rows so production CDC rows are identifiable as KDS CDC, not synthetic KDS.
- Export projections should be explicit, not `SELECT *`, matching current catalog style and keeping Parquet schema stable.
- Export targets should include tenant identity as explicit projected data where repo export conventions already do so.

### Durable Stage Details

- `staging_to_curated` details should include at minimum:
  - upstream `load_to_staging` batch id,
  - executed target names,
  - required staging tables,
  - SQL template ids,
  - status per target,
  - deleted/upserted row counts or equivalent diagnostics,
  - start/completion timestamps,
  - source identity marker used.
- `curated_to_s3` details should include at minimum:
  - upstream `staging_to_curated` batch id,
  - exported target names,
  - entity names,
  - output prefix URIs,
  - status per target,
  - start/completion timestamps.
- Details must be durable enough for Task 8 replay/checkpoint work to retry curated merge/export from staging/load/curated durable state.

### Business Date Rule

- Preserve CDC processing-date rule established earlier:

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

- Do not switch KDS CDC curated stages to synthetic feed or Brink snapshot semantics.

## Proposed Implementation Shape

High-level flow:

```text
Task 6 load_to_staging succeeds for tenant/entity slice
  -> resolve upstream successful load_to_staging batch
  -> select KDS curated targets whose required staging tables succeeded
  -> start staging_to_curated batch
  -> execute idempotent curated target SQL inside Redshift transaction(s)
  -> persist curated target diagnostics
  -> complete staging_to_curated batch
  -> start curated_to_s3 batch
  -> UNLOAD explicit curated projections to S3 curated prefix per entity
  -> persist export diagnostics
  -> complete curated_to_s3 batch
```

Recommended stage job shape:

```csharp
public interface IKdsCdcStagingToCuratedJobs
{
    Task<KdsCdcStageJobResult> ExecuteAsync(
        KdsCdcStagingToCuratedJobInput input,
        CancellationToken ct = default);
}

public interface IKdsCdcCuratedToS3Jobs
{
    Task<KdsCdcStageJobResult> ExecuteAsync(
        KdsCdcCuratedToS3JobInput input,
        CancellationToken ct = default);
}
```

Recommended curated target mapping direction:

- `staging.kds_orders` -> `curated.kitchen_orders`
- `staging.kds_items` -> `curated.kitchen_order_items`
- `staging.kds_item_bumps` -> `curated.kitchen_item_status_events`
- `staging.kds_customers` -> `curated.customers`

Recommended design notes:

- Prefer one `staging_to_curated` batch and one `curated_to_s3` batch per tenant/entity slice lineage, matching earlier KDS CDC boundaries.
- Use `ICuratedTargetExecutor` and `ICuratedExportExecutor` unless a concrete KDS CDC need requires small additive extension.
- Keep catalog SQL explicit and source-scoped by KDS CDC identity.
- Keep export projections explicit and stable.
- Keep synthetic KDS writer as reference only for dependency ordering, key-based idempotence, and soft-delete/orphan safety checks.

## Testing

- Unit tests for curated target catalog definitions:
  - required staging-table dependencies per target,
  - target SQL/source-type identity uses KDS CDC rather than synthetic KDS,
  - soft-delete semantics are preserved,
  - customer target exists or is explicitly conditionally executable by upstream success.
- Unit tests for curated export catalog definitions:
  - explicit target/entity naming,
  - explicit projections instead of `SELECT *`,
  - KDS CDC source filtering,
  - expected curated S3 entity naming and output-prefix rules.
- Job orchestration tests proving:
  - `staging_to_curated` resolves upstream successful `load_to_staging` batch,
  - only executable targets run when required staging tables succeeded,
  - stage batch completes only after merge/export success,
  - checkpoint advancement happens after completion,
  - curated merge failure stops export and marks batch failure,
  - curated export failure leaves upstream curated merge committed but marks export batch failure.
- Data/warehouse tests for:
  - idempotent re-run of same curated slice,
  - soft-delete propagation into curated rows,
  - safe order/item/status-event relationship handling under retry,
  - bootstrap/schema coverage for `curated.customers` if currently missing.
  - KDS CDC curated/bootstrap readiness does not rely on Brink POS service enablement.
- Export executor integration tests or additive tests proving:
  - KDS CDC curated exports use Redshift `UNLOAD`,
  - cleanup occurs for already-exported prefixes when a later target fails,
  - exported prefixes use tenant/entity/date/batch partitioning.
- Regression tests ensuring Brink curated stages remain unchanged.

Not required in this task:

- full checkpoint/replay orchestration policy,
- new SQS behavior,
- raw/load normalization changes,
- downstream Athena/query-layer consumers.

## Acceptance Criteria

- KDS CDC has dedicated `staging_to_curated` and `curated_to_s3` job seams wired into framework-native ETL stages.
- Curated merge is idempotent and retry-safe.
- Curated rows preserve production KDS CDC identity, not `synthetic_kds`.
- Source `is_deleted` becomes curated soft-delete behavior, not normal-path physical delete.
- `curated_to_s3` is included now and uses Redshift `UNLOAD`.
- Curated exports are partitioned by company, tenant, entity, and date using existing ETL curated layout.
- Curated export stage succeeds only after export succeeds and cleans up incomplete prefixes when possible.
- Stage details durably record enough merge/export diagnostics for later replay and checkpoint work.
- Kitchen order, order item, and item status event targets are covered.
- `curated.customers` is explicitly addressed in schema/catalog/runtime scope rather than silently omitted.
- If `curated.customers` is not fully implemented, the conditional blocker is explicit and test-covered.
- No KDS CDC runtime path calls `RedshiftCuratedOrderWriter` or stamps curated rows with `SourceSystem = "synthetic_kds"`; production KDS CDC curated rows are identifiable as `kds_cdc` / `KdsCdc`.
- Tests cover target catalogs, export catalogs, stage orchestration, soft-delete behavior, retry behavior, and any required curated customer schema/runtime additions.

## Notes

- Use `docs/features/ETL/kds-cdc/index.md#stage-8-staging-to-curated-staging_to_curated` and `#stage-9-curated-to-s3-curated_to_s3` as source of truth.
- Use Brink curated jobs/catalogs as structural reference only; their day/snapshot-scoped SQL should not be copied blindly into CDC row-delta semantics.
- Use `RedshiftCuratedOrderWriter.cs` only as KDS field-parity and dependency-order reference. Do not reuse it as production CDC runtime path.
- Keep `curated.customers` visible in scope even if current repo support is incomplete.