TASK 5: Persist Raw CDC Batches And Create Durable Ingestion Job State

Summary

Implement the first durable handoff for KDS CDC after intake validation and tenant resolution: write accepted CDC envelopes to tenant- and entity-scoped S3 raw storage, then create durable internal ETL batch or job state that lets downstream stages resume from raw without depending on SQS retention.

This task freezes raw artifact boundaries, raw payload fidelity, durable stage-batch creation rules, raw-to-batch metadata shape, and SQS acknowledgement gating. It establishes the extract_raw success point for KDS CDC.

This task does not yet implement normalization to load Parquet, Redshift COPY, curated merges, curated export, or final stage checkpoint policy across the full pipeline.

Current Codebase Findings

Repository analysis shows the codebase already has raw S3 storage and stage-batch lifecycle seams, but KDS CDC needs to use them with stricter replay metadata and tenant/entity slice boundaries.

• backend/Kds.Ai.Warehouse/Models/Storage/RawObjectWriteRequest.cs already supports tenant-scoped raw writes with CompanyId, TenantId, SourceType, BatchId, StreamName, BusinessDate, CapturedAtUtc, FileName, ContentType, and metadata.
• backend/Kds.Ai.Warehouse/Services/Storage/S3EtlStorage.cs writes raw objects through WriteRawAsync(...) and builds keys from RawObjectKey using the existing ETL S3 bucket and storage rules.
• backend/Kds.Ai.Warehouse/Services/Storage/EtlS3KeyBuilder.cs already builds raw keys in the required company/tenant/source/date/batch/entity shape: raw/company_id=.../tenant_id=.../source_type=.../{yyyy}/{MM}/{dd}/batch_id=.../entity=.../....
• backend/Kds.Ai.Data/Models/Ingestion/CreateIngestionBatchRequest.cs, backend/Kds.Ai.Data/Repositories/IIngestionBatchRepository.cs, and backend/Kds.Ai.Data/Services/IngestionBatchLifecycleService.cs already support stage-scoped batch creation, completion, failure, details JSON, upstream linkage, and raw S3 path persistence.
• backend/Kds.Ai.Data/Entities/IngestionBatch.cs stores CompanyId, TenantId, TenantDatabase, BusinessDate, PipelineStage, DetailsJson, RawS3Path, LoadS3Path, and UpstreamBatchId, which is enough for one raw artifact per stage batch but not for collapsing many tenant/entity raw artifacts into one row.
• backend/Kds.Ai.Integrations/BrinkPos/BrinkPosExtractJobs.cs is the current reference for extract_raw: start stage batch, write raw artifact, persist RawS3Path, serialize diagnostics into DetailsJson, and mark the batch complete or failed.
• backend/Kds.Ai.Integrations/Etl/StageDiagnosticsEnvelope.cs and StageDiagnosticsJson.cs already provide the repo pattern for storing structured per-batch diagnostics in DetailsJson.
• backend/Kds.Ai.Data/Models/Ingestion/UpsertIngestionCheckpointRequest.cs and backend/Kds.Ai.Integrations/Etl/StageExecutionSupport.cs show current checkpoint seams are stage-based and tenant-scoped. They do not yet persist CDC coordinate ranges as first-class checkpoint values, so Task 5 should preserve those ranges in durable batch metadata for Task 8 rather than treating SQS as replay state.

These findings create two important Task 5 constraints:

1. a KDS CDC extract_raw batch should represent one bounded tenant/entity slice, not one whole mixed poll window, because current IngestionBatch.RawS3Path stores one raw artifact path,
2. raw-write success alone is not enough; durable batch metadata must also capture coordinate and SQS ordering ranges before messages can be deleted.

Dependencies

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

Source References

• ../../docs/features/ETL/kds-cdc/implementation-breakdown.md#task-checklist
• ../../docs/features/ETL/kds-cdc/index.md#stage-4-store-to-s3-raw-extract_raw
• ../../docs/features/ETL/kds-cdc/index.md#stage-5-durable-internal-job-creation
• ../../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/kds-cdc/index.md#selected-behavior-summary

Scope From Breakdown

This task covers only the fifth implementation-breakdown item:

• write accepted CDC envelopes to tenant/entity-scoped S3 raw storage
• preserve replay-safe source metadata in raw artifacts
• create durable internal batch/job metadata after raw write succeeds
• acknowledge SQS messages only after raw persistence and durable resume state exist

Files to Create

1. KDS CDC raw persistence orchestration seam

Create a dedicated seam for taking resolved KDS CDC tenant/entity slices and persisting them as extract_raw artifacts.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/Raw/

Recommended logical types:

• IKdsCdcRawBatchWriter
• KdsCdcRawBatchWriter
• IKdsCdcExtractJobs or IKdsCdcExtractRawJobs
• KdsCdcExtractJobs or KdsCdcExtractRawJobs
• KdsCdcRawBatchWriteRequest
• KdsCdcRawBatchWriteResult
• KdsCdcRawArtifactDetails

Minimum responsibilities:

• accept only records that already passed Task 2 validation and Task 4 tenant resolution,
• split or consume input as one resolved tenant/entity slice per durable extract_raw batch,
• serialize raw CDC envelopes in replay-safe format,
• write raw object to ETL S3 storage,
• create or complete durable stage-batch metadata only after raw write succeeds,
• return enough information for later SQS acknowledgement and Task 6 normalization handoff.

2. Raw artifact serialization contract

Create a contract for the raw object contents, not only the storage path.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/Raw/Contracts/

Recommended logical types:

• KdsCdcRawRecord
• KdsCdcRawBatchPayload
• KdsCdcRawMetadataRange

Task 5 should explicitly choose the raw payload shape, for example:

1. NDJSON with one source-faithful envelope per line, or
2. JSON array or wrapped batch document containing ordered records plus batch metadata.

Preferred direction:

• use NDJSON or another streaming-friendly JSON form that preserves each original CDC envelope without forcing full-file materialization during replay.

Minimum raw content requirements:

• original CDC payload,
• operation code,
• source coordinates,
• normalized production ordering token derived from source coordinates when contract profile is production,
• source table/entity and primary key,
• resolved company_id and tenant_id,
• SQS message id, group id, and ordering metadata,
• durable batch id,
• worker identity or worker-attempt metadata.

Production ordering token rule:

• Task 5 must define a durable, comparable ordering token extracted from Debezium/source coordinates for production traffic.
• This token should preserve enough fields to compare order on replay without reopening SQS state, for example connector identity plus commit/LSN/binlog coordinates plus any row-within-transaction ordering field present in the source envelope.
• Raw persistence must store both the original source coordinates and the normalized comparable token; later stages may use the normalized token for ordering proof while still retaining the original envelope for replay/debugging.
• SQS SequenceNumber is transport metadata only for production. It may help diagnose redelivery or FIFO behavior, but it must not be treated as authoritative proof of production CDC order.

3. Durable KDS CDC extract diagnostics contract

Create a KDS CDC-specific details contract for extract_raw batch metadata persisted in DetailsJson and stage outputs.

Recommended logical location:

• backend/Kds.Ai.Integrations/KdsCdc/Models/

Recommended logical types:

• KdsCdcExtractBatchDetails
• KdsCdcRawRangeDetails
• KdsCdcSqsAcknowledgementDetails

This details payload should be durable enough for:

• later normalization from raw S3,
• later checkpoint advancement in Task 8,
• replay after raw-write success but before later stages succeed,
• proving which SQS messages became safe to delete.

For production ordering safety, these durable details must also preserve the comparable ordering-token contract chosen above so Task 6 can validate apply order from stored facts rather than from transient worker memory or SQS-only sequence values.

Files to Modify

4. KDS CDC worker handoff from Task 2 and Task 4

Modify likely KDS CDC intake and resolution seams so closed, resolved micro-batches hand off into extract_raw persistence rather than directly to later stages or immediate acknowledgement.

Expected areas:

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

Required flow:

• Task 2 worker closes ordered micro-batch,
• Task 4 resolution splits good records into resolved tenant/entity slices,
• Task 5 creates durable extract_raw batch per resolved slice,
• only after raw write plus durable metadata succeed can SQS delete be attempted for those messages.

5. KDS CDC adapter stage wiring from Task 1

Modify or implement the KDS CDC extract_raw stage wrapper introduced conceptually in Task 1.

Expected areas:

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

Task 5 should define whether the worker:

1. invokes IEtlPipelineRunner once per resolved tenant/entity slice using the KDS CDC adapter, or
2. invokes a thinner raw-write orchestration seam that itself produces the extract_raw stage result and then hands downstream work into the pipeline.

Preferred direction:

• keep KDS CDC aligned with framework-native stage execution and preserve extract_raw as a real stage batch, not an out-of-band worker-only write.

6. Ingestion batch persistence and metadata seams

Review and likely modify:

• backend/Kds.Ai.Data/Entities/IngestionBatch.cs
• backend/Kds.Ai.Data/Configurations/IngestionBatchConfiguration.cs
• backend/Kds.Ai.Data/Repositories/IIngestionBatchRepository.cs
• backend/Kds.Ai.Data/Repositories/EfIngestionBatchRepository.cs
• backend/Kds.Ai.Data/Models/Ingestion/CreateIngestionBatchRequest.cs

Task 5 should decide whether current batch storage is sufficient with RawS3Path + DetailsJson, or whether KDS CDC needs an additive child metadata table for richer raw-artifact or acknowledgement ranges.

Strong repo-driven guidance:

• do not overload one IngestionBatch row to represent many raw artifact paths for many entities,
• prefer one extract_raw batch per tenant/entity raw object if that keeps current schema sufficient,
• add a new child metadata table only if required for multiple acknowledgement ranges or artifact references that cannot fit cleanly into structured DetailsJson.

7. S3 storage metadata usage

Review and likely modify KDS CDC raw persistence code to use existing ETL storage cleanly.

Relevant areas:

• backend/Kds.Ai.Warehouse/Models/Storage/RawObjectWriteRequest.cs
• backend/Kds.Ai.Warehouse/Services/Storage/S3EtlStorage.cs
• backend/Kds.Ai.Warehouse/Services/Storage/EtlS3KeyBuilder.cs

Task 5 should specify KDS CDC raw file naming, content type, optional compression, and object metadata fields. Do not invent a separate KDS CDC bucket layout unless later docs change.

8. SQS acknowledgement seam

Modify likely queue orchestration from Task 2 so delete acknowledgement happens only after Task 5 durable boundary is complete.

Expected areas:

• Task 2 queue abstraction and worker types under backend/Kds.Ai.ApiService/Jobs/KdsCdc/
• queue client abstraction under backend/Kds.Ai.Integrations/KdsCdc/

Important rule:

• Task 5 should not allow message delete immediately after successful S3 upload if durable batch metadata creation failed.

9. Tests

Add or modify tests under:

• backend/tests/Kds.Ai.Integrations.Tests/KdsCdc/
• backend/tests/Kds.Ai.Data.Tests/
• backend/tests/Kds.Ai.ApiService.Tests/
• existing storage tests only if shared ETL S3 behavior needs additive assertions

Required Behavior

Raw storage boundary

extract_raw persists accepted CDC records to S3 raw using the existing ETL company/tenant path layout.

Fixed raw key intent from feature doc:

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

Task 5 must preserve tenant safety and replay isolation by ensuring each durable raw artifact is scoped to one resolved tenant and one entity/table.

Important consequence from current repo model:

• one extract_raw batch should correspond to one tenant/entity raw artifact or one bounded artifact set that can still be represented unambiguously by current batch metadata,
• Task 5 must not mix many entities or tenants into one IngestionBatch.RawS3Path without introducing a deliberate richer metadata model.

Raw payload fidelity

Raw object contents are replay source of truth after acknowledgement. They must preserve the CDC envelope faithfully rather than only storing transformed row DTOs.

Minimum raw contents:

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

Task 5 must explicitly forbid producer-side or extract-stage reshaping into only staging-row fields. Replay after code or mapping fixes depends on original source envelope fidelity.

Business date rule

For KDS CDC raw batches, BusinessDate remains the CDC processing date rule established earlier in the KDS CDC spec:

DateOnly.FromDateTime(DateTime.UtcNow)

Task 5 must not inherit Brink business-date semantics from source data.

Durable internal job or batch state

After raw S3 succeeds, the system must create durable internal metadata for downstream ETL work before SQS delete.

At minimum, durable metadata must connect:

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

Current repo pattern strongly suggests this metadata can start as structured DetailsJson on a stage-scoped IngestionBatch, provided Task 5 keeps one batch per tenant/entity slice. If richer durability is needed, Task 5 should add an explicit child metadata record rather than storing opaque unqueryable blobs elsewhere.

SQS acknowledgement gate

SQS messages are deleted only after both conditions are true:

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

Acknowledgement is per message or per receipt handle, not per original poll window.

• When a poll window splits into multiple tenant/entity slices, only messages belonging to slices that completed raw write plus durable metadata are eligible for delete.
• Messages from failed slices remain unacknowledged unless they were durably quarantined under an explicit policy.
• Durable extract details must record or summarize which message ids/receipt handles became safe to acknowledge.
• Deleting every receipt handle from a mixed poll window after partial success is forbidden.

If raw S3 write fails:

• do not delete SQS messages,
• rely on visibility timeout and retry,
• preserve failure diagnostics.

If raw S3 succeeds but durable metadata creation fails:

• do not delete SQS messages,
• allow retry,
• assume duplicate raw objects may be written on retry,
• require downstream dedupe to rely on source coordinates plus batch metadata rather than object uniqueness.

If a production message cannot yield a valid comparable ordering token from its Debezium/source coordinates:

• do not treat FIFO SequenceNumber as fallback ordering proof,
• fail or quarantine according to the established batch-versus-record safety rule before delete eligibility is granted,
• do not acknowledge the affected message until durable handling exists.

Ordering and range preservation

Task 5 must preserve enough ordered-range data to support later replay and checkpoint advancement.

Minimum range metadata to persist per durable raw batch:

• first and last Debezium source coordinates in batch,
• first and last normalized production ordering tokens in batch,
• first and last SQS sequence or ordering values in batch,
• message count,
• duplicate or redelivery notes when safe duplicates were collapsed,
• source connector or source identity if available.

This does not replace full Task 8 checkpointing, but it freezes what raw-stage durability must preserve now so Task 6 can validate production order and Task 8 can advance checkpoints from stored facts rather than from transient worker memory.

Failure boundary

Task 5 must preserve feature-doc boundary between record-local quarantine and batch-level failure.

• malformed or unmapped records should already be quarantined before raw write,
• raw S3 write failure is batch-level failure for the resolved tenant/entity slice,
• durable job creation failure is batch-level failure for the resolved tenant/entity slice,
• good slices may still succeed even if another slice from same poll window fails.

Task 5 must not reintroduce all-or-nothing failure across unrelated tenants or entities from same worker batch.

Fit with pipeline lineage

Task 5 establishes the first durable KDS CDC stage batch in the framework lifecycle.

Task 5 should preserve these expectations for later stages:

• extract_raw produces an EtlStageResult,
• downstream normalize_load uses upstream raw batch metadata and RawS3Path,
• later stages chain through UpstreamBatchId,
• Task 8 can advance checkpoints after documented success points.

Proposed Implementation Shape

Recommended high-level flow:

Task 2 worker closes ordered micro-batch
  -> Task 4 resolution filters to resolved records
  -> split resolved records by company_id + tenant_id + entity
  -> for each slice:
       start KDS CDC extract_raw stage batch
       serialize source-faithful raw payload
       write raw object to ETL S3
       persist RawS3Path + coordinate/sequence ranges + retry metadata
       complete stage batch
        mark only that slice's records safe for SQS delete
   -> delete only receipt handles for records whose slice raw write and durable metadata both succeeded

Recommended raw writer contract shape:

public interface IKdsCdcRawBatchWriter
{
    Task<KdsCdcRawBatchWriteResult> WriteAsync(
        KdsCdcRawBatchWriteRequest request,
        CancellationToken ct = default);
}

Recommended request shape:

KdsCdcRawBatchWriteRequest
  CompanyId
  TenantId
  TenantDatabase
  Entity
  BusinessDate
  Records[]
  FirstSourceCoordinate
  LastSourceCoordinate
  FirstProductionOrderingToken
  LastProductionOrderingToken
  FirstSqsSequence
  LastSqsSequence
  WorkerInstanceId

Recommended durable details shape:

KdsCdcExtractBatchDetails
  CompanyId
  TenantId
  TenantDatabase
  Entity
  RawS3Path
  RecordCount
  SourceCoordinateRange
  ProductionOrderingTokenRange
  SqsSequenceRange
  MessageIds or acknowledgement token summary
  WorkerAttempt
  SerializationFormat

Recommended stage identity:

new CreateIngestionBatchRequest(
    tenantId,
    tenantDatabase,
    IngestionMode.Cdc,
    IngestionSourceType.KdsCdc,
    CompanyId: companyId,
    BusinessDate: businessDate,
    PipelineStage: EtlStageNames.ExtractRaw)

Important design notes:

• prefer one durable extract_raw batch per resolved tenant/entity slice,
• use existing ETL S3 layout and storage abstraction,
• for production traffic, derive and persist comparable ordering tokens from Debezium/source coordinates before any delete acknowledgement decision; never promote SQS FIFO sequence into source-of-truth ordering proof,
• keep receipt handles or delete tokens outside raw object body if they are transport-only, but persist enough acknowledgement summary in durable metadata to prove safe delete set,
• keep DetailsJson structured and queryable enough for tests and future replay tooling,
• avoid relying on transient in-memory ordering state after S3 write finishes.

Testing

Unit tests

Add tests for:

• raw serializer preserves original CDC envelope and required replay metadata,
• production raw serializer preserves both original Debezium coordinates and normalized comparable ordering tokens,
• raw writer builds S3 raw keys under company_id, tenant_id, source_type=kds_cdc, batch_id, and entity,
• raw writer uses processing-date business date,
• durable details include raw path, entity, record count, coordinate range, and SQS range,
• durable details include production ordering-token range and never rely on SQS SequenceNumber alone for production ordering proof,
• one resolved tenant/entity slice produces one stage batch and one raw artifact path under current schema assumptions,
• mixed-entity or mixed-tenant input is split before durable batch persistence,
• successful raw write plus durable metadata marks messages safe for delete,
• raw-write failure leaves messages unacknowledged,
• production message with missing or unparseable ordering-token fields is not acknowledged,
• durable-metadata failure after raw write also leaves messages unacknowledged.

Data/repository tests

Add tests for:

• IngestionBatch persistence stores RawS3Path and KDS CDC structured DetailsJson correctly,
• any additive child metadata entity persists coordinate ranges and acknowledgement summary correctly,
• stage-batch creation uses IngestionMode.Cdc, IngestionSourceType.KdsCdc, PipelineStage = extract_raw, and processing-date BusinessDate.

Integration/composition tests

Add tests proving:

• resolved KDS CDC slice flows through ETL raw storage and stage-batch lifecycle successfully,
• IEtlS3Storage.WriteRawAsync(...) is invoked before SQS delete,
• SetRawS3PathAsync(...) or equivalent durable raw-path persistence happens before acknowledgement,
• worker does not delete messages when durable batch creation or completion fails,
• downstream handoff can reopen raw object using existing IRawObjectReader path style,
• current Brink raw path behavior remains unchanged.

Not required in task 5

This task does not require:

• normalization to load Parquet,
• Redshift COPY,
• final checkpoint advancement behavior,
• curated merge/export,
• full replay orchestration UI or CLI.

Acceptance Criteria

• [ ] Accepted KDS CDC records are written to tenant/entity-scoped ETL raw S3 paths
• [ ] Raw object contents preserve original CDC envelopes and replay-safe source metadata
• [ ] KDS CDC extract_raw uses IngestionMode.Cdc, IngestionSourceType.KdsCdc, and shared EtlStageNames.ExtractRaw
• [ ] Durable internal batch or job metadata is created only after raw write succeeds
• [ ] Durable metadata includes raw path, tenant identity, tenant database, entity, and source/SQS ordering ranges
• [ ] Production durable metadata includes comparable ordering-token ranges derived from Debezium/source coordinates
• [ ] Current schema is used safely with one durable raw batch per tenant/entity slice, or additive metadata storage is introduced deliberately where needed
• [ ] SQS messages are deleted only after both raw persistence and durable resume state succeed
• [ ] Production ordering proof never relies on SQS SequenceNumber alone when Debezium/source-coordinate ordering cannot be derived
• [ ] Partial success deletes only receipt handles for successfully persisted/quarantined records, never an entire mixed poll window by accident
• [ ] Raw-write failure leaves SQS messages unacknowledged
• [ ] Durable-metadata failure after raw write also leaves SQS messages unacknowledged
• [ ] Good tenant/entity slices can succeed even when another slice from same poll window fails
• [ ] Tests cover raw payload fidelity, durable metadata, split boundaries, and acknowledgement gating

Notes

• Follow docs/features/ETL/kds-cdc/index.md#stage-4-store-to-s3-raw-extract_raw and #stage-5-durable-internal-job-creation as source of truth.
• Keep S3 raw as replay source of truth after acknowledgement; never depend on SQS retention for replay.
• Current repo storage and batch model strongly favor one KDS CDC extract_raw batch per tenant/entity slice.
• Use Brink extract_raw flow as structural reference only; do not inherit Brink source semantics or business-date rules.
• Leave final checkpoint advancement and replay controls to Task 8, but persist enough raw-batch facts now that Task 8 can build on durable state rather than worker memory.