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/:
2. IKdsCdcStagingToCuratedJobs
3. KdsCdcStagingToCuratedJobs
4. IKdsCdcCuratedToS3Jobs
5. KdsCdcCuratedToS3Jobs
6. KdsCdcStagingToCuratedDetails
7. KdsCdcCuratedTargetDiagnostic
8. KdsCdcCuratedToS3Details
9. 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.

1. KDS CDC curated target catalog under backend/Kds.Ai.Integrations/KdsCdc/:
2. KdsCdcCuratedTargetCatalog
3. 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.

1. KDS CDC curated export catalog under backend/Kds.Ai.Integrations/KdsCdc/:
2. KdsCdcCuratedExportCatalog
3. 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.

1. Optional KDS CDC curated SQL helpers under backend/Kds.Ai.Integrations/KdsCdc/Sql/ if query text becomes too large for one catalog file:
2. keep helper scope minimal,
3. 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:

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:

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:

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:

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.