TASK 2: Implement SQS FIFO Intake, Validation, And Ordered Micro-Batching

Summary

Implement the runtime intake seam that pulls KDS CDC messages from cross-account SQS FIFO, validates delivery and CDC metadata, quarantines record-local failures, and forms bounded ordered micro-batches for downstream ETL execution.

This task establishes the production worker boundary between temporary transport in SQS and durable downstream ETL work. It freezes polling shape, message validation rules, duplicate and ordering detection expectations, micro-batch close triggers, and the handoff contract into later tenant-resolution and raw-persistence tasks.

This task does not yet implement tenant mapping, raw S3 persistence, durable internal job creation, Redshift load, or curated processing.

Current Codebase Findings

Repository analysis shows no existing SQS polling runtime in the backend today.

• Kds.Ai.ApiService/Program.cs registers startup/background services directly with AddHostedService(...) and disables background services in integration tests through options.DisableBackgroundServices().
• Kds.Ai.ApiService/Jobs/BrinkCronSyncService.cs is current reference for a host-managed service that opens a scope, resolves services from DI, logs through ILogger<T>, and performs startup orchestration.
• backend/simulator/Kds.Ai.Simulator.Web/Services/SimulationService.cs is current reference for a long-running BackgroundService loop shape.
• Kds.Ai.ApiService/Jobs/DocumentIngestionJobs.cs is current reference for AWS client usage style, structured logging, failure recording, and scoped dependency composition.
• Kds.Ai.Integrations/DependencyInjectionExtensions.cs currently wires ETL services only for Brink POS. KDS CDC worker registration will need its own additive registration path rather than piggybacking on Brink scheduling jobs.
• Kds.Ai.AgentsCommon/Settings/AwsCredentials.cs already defines validated AWS credential options.
• Kds.Ai.AppHost/AppHostExtensions.cs currently wires AWS credentials and ETL S3 configuration, but nothing yet for SQS queue URL or CDC polling options.
• No project currently references Amazon.SQS or IAmazonSQS, so this task should explicitly reserve package and DI wiring work for a new SQS client seam.

These findings mean Task 2 should define a dedicated KDS CDC background worker/service layer and SQS abstraction rather than trying to retrofit Ticker jobs or reuse synthetic-feed orchestration.

Dependencies

• docs/features/ETL/kds-cdc/index.md
• docs/features/ETL/kds-cdc/implementation-breakdown.md
• specs/KdsCdcIngestion/TASK1.md

Source References

• ../../docs/features/ETL/kds-cdc/implementation-breakdown.md#task-checklist
• ../../docs/features/ETL/kds-cdc/index.md#stage-1-debezium-to-sqs-fifo
• ../../docs/features/ETL/kds-cdc/index.md#stage-2-worker-pull-and-micro-batch-creation
• ../../docs/features/ETL/kds-cdc/index.md#supported-corner-cases
• ../../docs/features/ETL/kds-cdc/index.md#selected-behavior-summary
• ../../docs/features/ETL/kds-cdc/index.md#non-goals

Scope From Breakdown

This task covers only the second implementation-breakdown item:

• poll Debezium CDC messages from cross-account SQS FIFO
• validate required message metadata and detect malformed, duplicate, missing, or out-of-order records
• build bounded micro-batches using count, age, and memory/visibility limits instead of one batch per record
• preserve explicitly scoped ordering guarantees before downstream load begins

Files to Create

1. KDS CDC worker and orchestration seam

Create a dedicated hosted worker path for KDS CDC intake.

Recommended logical location:

• backend/Kds.Ai.ApiService/Jobs/KdsCdc/

Recommended logical types:

• KdsCdcPollingService or KdsCdcWorkerService as BackgroundService
• IKdsCdcMessagePump / KdsCdcMessagePump
• IKdsCdcBatchBuilder / KdsCdcBatchBuilder
• IKdsCdcMessageOrderingValidator / KdsCdcMessageOrderingValidator
• IKdsCdcDuplicateDetector / KdsCdcDuplicateDetector
• IKdsCdcQuarantineWriter or equivalent quarantine handoff seam if record-local failures must be persisted later

Important rule:

• worker owns polling, validation, grouping, and close-trigger decisions,
• worker should not embed downstream ETL stage logic,
• keep orchestration seams narrow enough that later tasks can plug in tenant resolution and raw persistence without rewriting poll loop.

2. KDS CDC SQS contract and configuration models

Create dedicated configuration and transport contract models.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/Configuration/
• backend/Kds.Ai.Integrations/KdsCdc/Messages/

Recommended logical types:

• KdsCdcSqsOptions
• KdsCdcBatchingOptions
• KdsCdcEnvelope or similarly named intake DTO representing required Debezium plus SQS metadata
• KdsCdcReceivedMessage combining parsed body plus SQS receipt/ordering metadata
• KdsCdcMicroBatchCandidate or equivalent in-memory batch contract

Configuration should cover at least:

• queue URL,
• AWS region if it differs from existing ETL S3 region assumptions,
• max messages per poll,
• wait time / long-poll duration,
• visibility timeout target or renewal threshold,
• max records per batch,
• max batch age,
• soft byte or memory ceiling,
• optional dedupe cache horizon.

3. SQS client abstraction

Create a thin SQS abstraction so worker logic can be unit tested without AWS.

Recommended logical types:

• IKdsCdcQueueClient
• KdsCdcSqsQueueClient

Minimum responsibilities:

• receive messages,
• surface FIFO metadata needed for ordering,
• optionally extend visibility if batch hold time requires it,
• delete acknowledged messages only when later tasks say durable boundary is reached.

Do not hide ordering metadata behind a too-generic queue abstraction. FIFO sequencing details are first-class requirements in this feature.

Files to Modify

4. API host registration

Modify:

• backend/Kds.Ai.ApiService/Program.cs

Add the KDS CDC background worker registration in the host layer.

Important host rule:

• background service must respect current integration-test convention where background services do not auto-run,
• KDS CDC worker should not start in integration-test environment unless a test opts into it explicitly,
• registration should be additive and must not change Brink cron or monitor cron startup behavior.

5. Integration module wiring

Modify:

• backend/Kds.Ai.Integrations/DependencyInjectionExtensions.cs

Add KDS CDC-specific service registrations for:

• queue client,
• batching and validation services,
• options binding,
• any first worker-facing orchestration services.

Prefer a dedicated extension such as AddKdsCdcIntegration() or AddKdsCdcEtlIntakeModule() instead of bloating AddBrinkPosEtlModule().

6. AppHost and local configuration seams

Review and likely modify:

• backend/Kds.Ai.AppHost/AppHostExtensions.cs
• backend/Kds.Ai.AppHost/AppHost.cs

Task 2 should define how local/dev hosts provide:

• SQS queue URL,
• AWS region,
• any KDS CDC batching option overrides.

Do not hardcode queue URLs in worker code.

7. Project/package references

Review and likely modify the relevant project files under:

• backend/Kds.Ai.ApiService/
• backend/Kds.Ai.Integrations/

No SQS SDK reference exists yet. This task should account for adding the minimal AWS SQS package needed for IAmazonSQS or chosen equivalent.

8. Tests

Add or modify tests under:

• backend/tests/Kds.Ai.Integrations.Tests/KdsCdc/
• backend/tests/Kds.Ai.ApiService.Tests/Kds.Ai.ApiService.Tests/

Use integration-layer tests for pure validation/batching behavior and API host tests for worker registration/environment gating where appropriate.

Required Behavior

Transport source

Worker must poll Debezium CDC messages from cross-account SQS FIFO.

Task 2 should assume SQS is temporary transport only.

• SQS is not replay source,
• worker may hold messages in memory only until bounded micro-batch closes,
• delete acknowledgement stays deferred until later durable boundary tasks complete.

Contract profiles and required message metadata

Task 2 must support separate validation profiles for temporary mock traffic and future production Debezium traffic.

Mock profile:

• contract_version = kds-cdc-mock/v1,
• required top-level mock fields from mock-cdc-sqs-contracts.md,
• entity plus sequence.stream_key and sequence.stream_position are accepted as the mock source-order surrogate,
• source database/schema/table and real Debezium offsets are not required in mock mode unless the mock contract is later expanded.

Production profile:

• real Debezium/source database, schema, table, connector/source identity, and source coordinates are mandatory,
• unsupported or missing production coordinate metadata must not flow into downstream ETL steps.

Accepted messages must expose or derive all metadata required by the main feature doc:

• 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 for production, or mock sequence metadata for kds-cdc-mock/v1,
• SQS FIFO message group id,
• SQS FIFO ordering value such as sequence number,
• SQS message id,
• receipt handle needed for later delete acknowledgement.

Task 2 should freeze which fields are mandatory for each contract profile and must not apply production Debezium requirements blindly to mock messages.

Missing mandatory fields must not silently flow into later ETL steps.

Validation and quarantine boundary

Worker must validate message shape before record enters a micro-batch.

Task 2 should distinguish:

1. record-local failures that can be quarantined while good messages continue,
2. batch-level ordering failures that must stop downstream handoff.

Record-local quarantine candidates include:

• malformed or unreadable payload,
• missing source primary key,
• missing required source-order metadata for the active contract profile,
• missing FIFO sequencing metadata,
• unsupported source table when table allowlist is already known,
• duplicate message that cannot be safely collapsed.

Task 2 does not need final durable quarantine persistence yet, but it must define an explicit quarantine handoff contract so later tasks do not rework validation flow.

Duplicate detection

Worker must detect duplicates using production Debezium source coordinates plus source table and primary key, or mock sequence plus entity/table and primary key when the active contract profile is kds-cdc-mock/v1.

Task 2 should define duplicate outcomes clearly:

• exact duplicate that is safely ignorable should not create a new downstream micro-batch record,
• suspicious duplicate that conflicts on coordinates or ordering should become quarantine or batch failure,
• dedupe must be bounded in memory and safe for low- and high-volume tenants.

Do not rely on SQS message id alone for dedupe identity.

Ordering validation

Ordering scope must be explicit.

Mock contract scope:

• FIFO order is guaranteed only within MessageGroupId = source_account_id:store_guid,
• mock sequence.stream_position proves per-stream ordering, not total cross-store ordering,
• tests must not claim cross-store global order from per-store FIFO groups.

Production contract scope:

• production messages must preserve the Debezium/source coordinate facts needed to prove the intended global ordering scope before Redshift load,
• if a total cross-store/source order is required, it must be proven by source coordinates or by queue grouping strategy, not by multiple FIFO message groups alone.

Before downstream load begins, worker must preserve enough ordering evidence to prove batch order using:

• FIFO message group metadata,
• FIFO sequence metadata,
• Debezium source coordinates, or mock stream sequence in mock mode.

Task 2 should define how worker reacts to:

• out-of-order message inside active poll window,
• sequence gap,
• missing ordering metadata,
• redelivery of already-seen message,
• multi-poll accumulation while preserving overall order.

If the required ordering scope cannot be proven for the candidate batch, the batch must fail safely and not advance to later pipeline stages.

Message size validation

Task 2 must account for SQS FIFO message size limits.

• Producer-side oversize messages should fail before publish where possible.
• Consumer-side oversize or truncated/unreadable messages must become quarantine-ready failures and must not enter micro-batches.
• Tests should cover at least one oversize or unreadable body path if the queue seam exposes payload size.

Micro-batch close triggers

Worker must build bounded micro-batches rather than one ETL batch per CDC record.

Batch close triggers must include:

1. record count limit, initially in range 500 to 2,000 configurable records,
2. age limit, normally 1 to 5 minutes,
3. optional longer window up to 15 minutes for lower-freshness environments,
4. byte or memory pressure threshold,
5. visibility-timeout safety threshold so worker does not hold unacknowledged messages too long.

Task 2 should define precedence and interaction of these triggers. Example: batch closes on first trigger reached, not only count threshold.

Raw-batch shaping constraints

Although tenant resolution and raw persistence are later tasks, Task 2 must preserve future split requirements.

Micro-batch contract must carry enough information so later stages can split accepted records by:

• company,
• tenant,
• entity/table.

Task 2 must not bake in a design that assumes one batch equals one final raw object or one tenant only.

Visibility timeout safety

Task 2 must explicitly account for SQS visibility timeout.

Worker design should either:

• keep batch lifetimes below safe visibility threshold, or
• provide a seam to renew visibility while batch remains open.

Ignoring visibility timeout would violate selected micro-batch behavior in the feature doc.

Host lifecycle

Worker should run as a host-managed background service, not as a Ticker job.

Reason:

• CDC is continuous pull-based runtime,
• long polling and visibility management fit BackgroundService,
• Ticker jobs are current fit for scheduled/manual triggers, not continuous FIFO consumption.

Proposed Implementation Shape

Recommended worker split:

BackgroundService loop
  -> queue client receive
  -> payload parse + metadata validation
  -> duplicate/order validation
  -> add to current in-memory micro-batch
  -> close batch when count/age/size/visibility trigger hits
  -> hand off closed batch to downstream task-5 durability seam
  -> keep receipt handles pending until durable ack point exists

Recommended queue receive contract shape:

public interface IKdsCdcQueueClient
{
    Task<IReadOnlyList<KdsCdcReceivedMessage>> ReceiveAsync(CancellationToken ct);
    Task ChangeVisibilityAsync(IReadOnlyList<KdsCdcReceivedMessage> messages, TimeSpan visibilityTimeout, CancellationToken ct);
    Task DeleteAsync(IReadOnlyList<KdsCdcReceivedMessage> messages, CancellationToken ct);
}

Recommended batching seam:

public interface IKdsCdcBatchBuilder
{
    KdsCdcBatchBuildResult Add(KdsCdcReceivedMessage message, DateTimeOffset receivedAtUtc);
    bool ShouldClose(DateTimeOffset nowUtc);
    KdsCdcMicroBatchCandidate Close(DateTimeOffset closedAtUtc);
}

Recommended options shape:

KdsCdcSqsOptions
  QueueUrl
  Region
  MaxMessagesPerPoll
  WaitTimeSeconds
  VisibilityTimeoutSeconds

KdsCdcBatchingOptions
  MaxRecordCount
  MaxBatchAge
  MaxBufferedBytes
  VisibilitySafetyWindow
  DuplicateCacheWindow

Important note:

• use dedicated options types rather than scattering magic constants through worker code,
• keep queue receive/delete operations behind testable seam,
• keep downstream handoff abstract until task 5 creates durable raw/job boundary.

Testing

Unit tests

Add tests for:

• message parser rejects malformed body and surfaces quarantine-ready failure details,
• validator rejects missing mandatory metadata,
• duplicate detector ignores safe duplicates and flags unsafe duplicates,
• ordering validator detects out-of-order and missing-sequence conditions,
• batch builder closes on count threshold,
• batch builder closes on age threshold,
• batch builder closes on byte or memory threshold,
• batch builder closes before visibility safety limit is exceeded,
• batch builder does not create one closed batch per record under normal multi-record input,
• low-volume path closes by age without waiting for count threshold,
• high-volume path creates multiple ordered micro-batches.

Host/composition tests

Add tests proving:

• KDS CDC worker is registered only in runtime environments where background services are enabled,
• integration-test environment does not auto-run KDS CDC background worker,
• DI can resolve KDS CDC queue client and batching services,
• options binding fails fast when required queue configuration is missing.

Not required in task 2

This task does not require:

• live AWS SQS integration tests,
• tenant mapping tests,
• S3 raw persistence tests,
• Redshift tests,
• final delete-acknowledgement tests after durable write, because durable boundary arrives in Task 5.

Acceptance Criteria

• [ ] KDS CDC has dedicated continuous worker/poller seam for SQS FIFO intake
• [ ] Worker is designed as hosted background service, not Ticker job
• [ ] SQS client abstraction exists and exposes receive plus later ack operations
• [ ] Required Debezium and FIFO metadata fields are defined in intake contracts
• [ ] Mock and production validation profiles are distinct, with mock sequence accepted as coordinate surrogate only for kds-cdc-mock/v1
• [ ] Validation rejects malformed or metadata-incomplete messages before downstream ETL handoff
• [ ] Record-local validation failures can be quarantined without forcing every good record to fail
• [ ] Duplicate detection uses Debezium coordinates plus source table and primary key, not only SQS message id
• [ ] Ordering validation detects missing, duplicate, and out-of-order conditions
• [ ] Micro-batch builder closes on configurable count, age, and memory/visibility triggers
• [ ] Design does not create one ETL batch per CDC record
• [ ] Worker preserves ordering evidence required before downstream load begins
• [ ] Ordering scope is explicit, with mock per-stream ordering not treated as cross-store global ordering
• [ ] Oversize or unreadable SQS payloads are rejected or quarantined before batching
• [ ] API host/background-service registration follows current integration-test disable convention
• [ ] AppHost or configuration seams exist for queue URL and batching options
• [ ] Tests cover validation, batching triggers, ordering behavior, and runtime registration rules

Notes

• Follow docs/features/ETL/kds-cdc/index.md as source of truth for ordering and micro-batch behavior.
• Reuse existing host patterns for IHostedService and BackgroundService, but do not copy Brink scheduler semantics into CDC.
• Keep delete acknowledgement deferred until Task 5 establishes raw-write plus durable-job boundary.
• Prefer explicit contracts for quarantine-ready failures now; persistence details can land later.
• Avoid generic queue abstractions that erase FIFO sequence metadata, because ordering is core behavior here.