# KDS CDC Branch Direction And Process Flow ## Purpose This document records the assumptions, process flow, design decisions, and implementation direction taken on this branch for KDS CDC ingestion and ETL. It is intended as historical context for later analysis and maintenance. It describes what this branch is trying to achieve, how the current implementation approaches it, which assumptions it relies on, and which decisions were made while moving from simulator-shaped CDC intake to Debezium-native intake. ## What We Wanted To Achieve The branch goal is to ingest Logic Control KDS change data capture events into the KDSAI ETL and warehouse pipeline using real Debezium payloads. Primary outcomes: - Treat Debezium payloads as the only supported KDS CDC intake contract. - Stop depending on simulator-specific message shape, mock contract versioning, and producer-generated ordering semantics. - Keep the existing KDSAI ETL stage model instead of building a completely separate ingestion system. - Preserve source payload and source coordinates in raw storage so replay and audit remain possible. - Normalize KDS CDC rows into source-shaped staging tables for orders, items, item bumps, and customers. - Build curated warehouse state from staging state, not from a fragile assumption that CDC events arrive parent-before-child. - Make SQS a transport buffer, not the long-term source of truth. The desired long-term shape is capture-first and interpret-later: ```text SQS Debezium message -> durable raw capture -> tenant resolution and promotion -> normalize_load -> load_to_staging -> staging_to_curated -> curated_to_s3 ``` ## Branch-Level Direction This branch changes KDS CDC from a mock/simulator contract toward a production Debezium contract. Important direction decisions: - Debezium envelope shape is authoritative. - Top-level Debezium fields `schema` and `payload` are expected. - Semantic data comes from `payload.source`, `payload.op`, `payload.after`, and `payload.before`. - Row data is not read from the envelope root. - `payload.after` is used for create, update, and snapshot/read events. - `payload.before` is used for delete events. - Primary key field is `guid` for supported KDS tables. - Supported tables are `orders`, `items`, `item_bumps`, and `customers`. - Unsupported tables are rejected before tenant-scoped ETL work. - SQS FIFO metadata can be preserved but is not required for correctness. - `MessageGroupId` is not a partitioning or ordering dependency for KDS CDC. - `SequenceNumber` is nullable and optional. - Debezium source coordinates are the source identity and ordering basis. - `store_guid` is the tenant identity for this integration. - `source_account_id` is no longer part of required routing identity. ## Current Branch Implementation Summary The branch implements Debezium-native parsing and removes the simulator/mock CDC producer path from the KDS CDC intake surface. Implemented now: - Parser accepts Debezium envelopes and rejects the old mock/custom contract path. - Parser accepts SQS messages with missing `MessageGroupId` and missing `SequenceNumber`. - Parser extracts source database, schema, table, operation, selected row image, primary key, store guid, and source coordinates. - Duplicate detection keys on Debezium source identity instead of SQS FIFO identity. - Ordering token parsing understands Debezium coordinates in the form `binlog-file:position:row`. - Batch building no longer splits or closes batches on SQS `MessageGroupId` boundaries. - Tenant resolution maps by `store_guid` and no longer requires `source_account_id`. - Child messages without direct `store_guid` become retryable tenant-unresolved messages instead of permanent quarantine. - Raw records preserve nullable FIFO metadata. - Normalization reads the selected Debezium row image from `payload.after` or `payload.before`. - Mock CDC producer, mock envelope factory, mock queue publisher, and simulator CDC publishing registration were removed. - Tests were converted toward real Debezium payload fixtures. Known implementation boundary: - The desired separate landing RAW plus explicit tenant RAW promotion state is documented as target architecture, but this branch does not fully introduce a new durable landing RAW store and promotion-state table. - Current branch still primarily follows the existing tenant-scoped ETL raw path once tenant resolution succeeds. - Missing child tenant relation state is handled safely by retrying unresolved messages, but a durable `order_guid -> store_guid` and `item_guid -> store_guid` relation index is not yet fully introduced as a new persistence model. ## Assumptions ### Source Payload Assumptions KDS CDC messages are Debezium messages from Logic Control KDS source tables. Assumed Debezium shape: ```json { "schema": {}, "payload": { "before": {}, "after": {}, "source": { "db": "...", "schema": "...", "table": "...", "file": "mysql-bin-changelog.237363", "pos": 123, "row": 0 }, "op": "c" } } ``` Supported operation codes: - `c` means create and uses `payload.after`. - `u` means update and uses `payload.after`. - `r` means snapshot/read and uses `payload.after`. - `d` means delete and uses `payload.before`. Supported table names: - `orders` - `items` - `item_bumps` - `customers` Assumed row identity: - each supported row image has a primary key field named `guid`. - orders and customers normally carry direct `store_guid`. - items may not carry direct `store_guid` and may need resolution through `order_guid`. - item bumps may not carry direct `store_guid` and may need resolution through `item_guid`. ### Tenant Assumptions For this KDS CDC integration, `store_guid` is treated as the tenant identifier. Routing assumptions: - `store_guid == tenant_id` for KDS CDC routing purposes. - `store_guid` is stable and unique enough to resolve tenant ownership. - A mapping row still connects source type and store guid to platform company/tenant warehouse metadata. - `source_account_id` is optional historical metadata and not required for KDS CDC tenant resolution. Child resolution target model: - direct `store_guid` should be used when present. - items without `store_guid` should resolve through `order_guid -> store_guid`. - item bumps without `store_guid` should resolve through `item_guid -> store_guid`. - unresolved child records should not be treated as bad data immediately. - unresolved child records should be retried after parent relation state is available. ### Transport Assumptions SQS remains the transport mechanism. It may be a FIFO queue, but application correctness must not rely on FIFO-only attributes. Transport assumptions: - SQS `MessageId` and `ReceiptHandle` are required to process and acknowledge a transport message. - SQS `MessageGroupId` is optional for KDS CDC logic. - SQS `SequenceNumber` is optional for KDS CDC logic. - Approximate receive count is useful diagnostics but not business identity. - Source identity must come from Debezium source metadata, table, primary key, and source coordinates. ### Ordering Assumptions Ordering is best represented by Debezium source coordinates, not SQS FIFO sequence. Ordering token assumptions: - coordinate format is `file:pos:row`. - binlog file ordinal is parsed from the numeric suffix of the file name. - position and row are numeric components. - a comparable token can be built from file ordinal, position, and row. - missing or unparsable coordinates should not block raw capture. - comparable decreasing coordinates indicate unsafe ordering for a stream. This branch treats missing comparable source coordinates as acceptable for intake so durable capture is not blocked. Later ETL stages can still use preserved metadata to decide whether a batch is safe to apply. ## Process Flow ### 1. SQS Receive The KDS CDC worker receives transport messages from SQS. Current behavior: - request SQS message body and attributes when present. - preserve `MessageGroupId` and `SequenceNumber` when present. - allow `MessageGroupId` and `SequenceNumber` to be null. - keep body length for payload-size validation. The worker does not treat one SQS message as one full ETL batch. It builds bounded micro-batches using existing batching rules such as count, age, and buffered byte limits. ### 2. Parse Debezium Envelope The parser validates Debezium structure and extracts KDS CDC metadata. Validation outcomes: - malformed JSON is rejected. - missing Debezium payload is rejected. - missing `payload.source.db` or `payload.source.table` is rejected. - unsupported table is rejected. - unsupported op code is rejected. - missing selected row image is rejected. - missing `guid` primary key is rejected. Accepted message fields: - source database from `payload.source.db`. - source schema from `payload.source.schema` when present. - source table from `payload.source.table`. - operation from `payload.op`. - selected row image from `after` or `before`. - primary key name `guid` and primary key value from selected row image. - direct `store_guid` when present and non-blank. - source coordinates from `file`, `pos`, and `row` when all are present. - body hash for duplicate diagnostics. ### 3. Duplicate Detection Duplicates are detected using source-level Debezium identity rather than FIFO metadata. Branch dedupe identity: ```text source database + source table + primary key value + source coordinates ``` Fingerprint data includes operation, SQS message id, and body hash. FIFO group and sequence are not part of correctness identity. Decision: - same Debezium identity with same fingerprint can be safely treated as duplicate delivery. - same Debezium identity with different fingerprint is suspicious and should not be silently applied. ### 4. Ordering Validation Ordering validation uses structured Debezium coordinates when available. Coordinate comparison model: ```text mysql-bin-changelog.237363:1615:0 -> file ordinal 237363 -> position 1615 -> row 0 ``` Comparable token shape: ```text fileOrdinal:position:row ``` Decision: - decreasing comparable tokens are unsafe and fail the batch path. - duplicate comparable tokens are quarantined as suspicious. - missing or unparsable comparable tokens are accepted at intake because capture should not be blocked. ### 5. Micro-Batch Construction The batch builder groups accepted messages into micro-batches. Branch decision: - batches are not partitioned by SQS `MessageGroupId`. - a changed `MessageGroupId` does not force batch close. - source accounts are optional and omitted when empty. - stores and entities are collected from accepted messages. This keeps batching aligned with Debezium semantics and avoids coupling the ETL pipeline to transport-level FIFO grouping. ### 6. Tenant Resolution Tenant resolution happens before tenant-scoped raw storage and warehouse loading. Branch behavior: - direct `store_guid` is used as routing identity. - mapping repository lookup uses source type and store guid. - `source_account_id` is no longer required. - missing mapping is still a failure for direct-tenant messages. - ambiguous mapping is still a failure. - missing or inactive tenant warehouse is still a failure. - missing direct `store_guid` returns `TenantUnresolved` instead of permanent quarantine. Reason for this decision: - orders and customers can usually resolve directly. - child records can arrive before parent relation state is available. - unresolved children should not be destroyed or quarantined as bad data if they may become resolvable later. Current limitation: - durable relation indexes for `order_guid -> store_guid` and `item_guid -> store_guid` are target architecture but not fully implemented in this branch. - unresolved child messages currently remain retryable rather than being promoted through a completed landing RAW relation-resolution workflow. ### 7. Raw Capture And Handoff Target architecture says raw capture should happen as early as possible in a landing RAW area, before tenant resolution is fully reliable. Target flow: ```text valid Debezium message -> landing RAW object -> promotion state -> SQS ack -> tenant resolution and tenant RAW promotion ``` Current branch behavior is a smaller step toward that target: - raw writing remains integrated with the existing `extract_raw` stage after tenant resolution. - raw records now preserve Debezium envelope, selected metadata, source coordinates, nullable FIFO transport metadata, and original payload. - SQS ack behavior still depends on successful handoff for resolved slices. - unresolved messages are not acknowledged and get visibility extension. Decision record: - durable raw-first capture is the desired architecture. - branch implementation makes the code Debezium-safe and transport-decoupled first. - full landing RAW plus promotion-state persistence remains a follow-up if strict capture-before-resolution is required. ### 8. Normalize Load `normalize_load` reads raw NDJSON records and writes normalized load artifacts. Important branch change: - normalization maps from selected Debezium row image, not from the envelope root. Selected image rule: ```text op == d -> payload.before else -> payload.after ``` Normalization output entities: - orders to KDS order staging row shape. - items to KDS item staging row shape. - item_bumps to KDS item bump staging row shape. - customers to KDS customer staging row shape. Normalization preserves: - primary key. - operation. - delete marker. - source database, table, and coordinates. - nullable SQS metadata. - raw envelope for replay and audit. Unknown source columns are ignored by row mapping but remain preserved in raw payload. ### 9. Load To Staging `load_to_staging` loads normalized artifacts into tenant Redshift staging tables. The branch continues using existing staging-oriented ETL behavior. Intended role of staging: - hold source-shaped KDS state. - provide tenant-local relational state for downstream curated rebuild. - support idempotent reprocessing from durable raw or load artifacts. ### 10. Staging To Curated Curated state should be rebuilt from staging truth, not from the order in which SQS delivered messages. Decision: - CDC arrival order should not be the only mechanism that makes parent and child rows consistent. - if an item arrives before an order, curated correctness should come from staged tenant state after both are present. - curated jobs should operate from source-shaped staging tables and rebuild affected targets. Curated targets covered by this branch family include: - kitchen orders. - kitchen order items. - kitchen item status events. - customers. ### 11. Curated To S3 `curated_to_s3` remains part of the KDS CDC ETL flow. Purpose: - export curated tenant state to S3. - preserve the same externalized analytics pattern used by other ETL adapters. - keep Redshift curated state and S3 curated artifacts aligned through stage metadata. ## Quarantine And Retry Decisions The branch separates bad data from temporarily unresolved routing. Rejected or quarantined: - malformed JSON. - missing Debezium payload. - missing required Debezium source metadata. - unsupported table. - unsupported operation. - missing selected row image. - missing primary key. - ambiguous tenant mapping. - missing tenant warehouse. - inactive tenant warehouse. - suspicious duplicate. - unsafe decreasing ordering token. Retryable unresolved: - child message without direct `store_guid` where tenant may later resolve through relation state. This distinction is important because unresolved child routing is expected in CDC streams and should not be treated the same as invalid payload format. ## Removed Simulator And Mock CDC Path The branch removes the KDS CDC mock producer path because it conflicted with the Debezium-only direction. Removed concepts: - mock contract version branch in parser. - mock envelope factory. - mock CDC producer. - mock sequence allocator. - mock SQS queue publisher. - simulator publishing KDS CDC mock messages. - tests that validate mock message generation rather than Debezium intake. Decision: - simulator may still create synthetic warehouse rows for demo traffic. - simulator no longer defines or publishes the KDS CDC intake contract. - KDS CDC tests should use real Debezium-style fixture payloads. ## Data And Persistence Decisions Source tenant mapping changed from source-account keyed routing to store-guid keyed routing. Mapping direction: ```text source_type + store_guid -> company_id + tenant_id + tenant warehouse ``` Entity changes: - `KdsCdcSourceTenantMapping.SourceAccountId` is nullable. - repository lookup no longer requires source account. - active mapping index is based on source type, store guid, and active state. - quarantine transport fields allow nullable message group and sequence. Reason: - Debezium payload samples do not establish `source_account_id` as required routing identity. - KDS CDC tenant identity is the store guid. - SQS FIFO metadata should not be forced into durable data as required business metadata. ## What This Branch Does Not Claim To Solve Completely This branch is a major contract and direction change, not a complete redesign of every ETL persistence boundary. Not fully complete in this branch: - separate immutable landing RAW area before tenant resolution. - explicit durable promotion-state entity for landing RAW to tenant RAW. - persistent relation index tables for child tenant resolution. - full ack-after-landing behavior independent of tenant resolution. - replacement synthetic data generator that emits real Debezium envelopes. - admin UI or operations tooling for unresolved landing records. These are still valid future work if the system needs the fully documented capture-first architecture. ## Why These Decisions Were Made The implementation favored the fastest safe movement toward production compatibility. Reasons: - real Debezium samples showed the previous simulator contract was not representative. - relying on `MessageGroupId` and `SequenceNumber` would make transport metadata part of business correctness. - source coordinates are better replay, dedupe, audit, and ordering identity than SQS attributes. - `store_guid` is the stable KDS tenant identity and avoids carrying a synthetic source-account requirement. - normalizing from the wrong JSON level would silently produce bad data once real Debezium messages arrive. - removing mock producer code prevents old simulator behavior from continuing to look like a supported intake contract. ## Maintenance Notes Future maintainers should treat this branch as the point where KDS CDC changed from simulator-compatible to Debezium-compatible. When extending this area: - preserve Debezium payload as the supported contract. - avoid adding new logic that requires SQS `MessageGroupId` or `SequenceNumber` for correctness. - keep source coordinates in raw, diagnostics, checkpoints, and replay identity. - route by `store_guid` and relation indexes, not by source account. - make child tenant-unresolved behavior retryable unless the payload itself is invalid. - keep normalization tied to `payload.after` or `payload.before` based on operation. - prefer durable capture and replayable state over doing all semantic work inside the SQS polling loop. ## Summary This branch establishes the Debezium-native direction for KDS CDC. The biggest conceptual shift is that KDS CDC intake is no longer a simulator-shaped stream with FIFO-group ordering and source-account tenant routing. It is a Debezium stream where source metadata, row images, source coordinates, and `store_guid` drive correctness. The branch implements the core compatibility changes needed for real payloads and removes mock CDC generation. It also records the next architectural boundary: durable landing RAW plus explicit tenant RAW promotion, which remains the intended long-term shape for safest capture-first operation.