# 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:

```text
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:

```csharp
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:

```text
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:

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

Recommended request shape:

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

Recommended durable details shape:

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

Recommended stage identity:

```csharp
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.