# TASK 3: Add Redshift JSON Load, Validation, And Atomic Curated Replacement

## Summary

Implement the Fleet-specific Redshift executor that loads date-grouped JSON manifests into temporary `SUPER` data, transforms and validates all candidates, then atomically replaces the exact curated source slice on one tenant connection.

## Dependencies

- [TASK 0](./TASK0.md)
- [TASK 0](./TASK0.md) frozen snapshot, replacement-intent, and result contracts. Integrate with Task 2 after both implementations exist.

## Current Codebase Findings

- `ITenantWarehouseConnectionFactory` opens a connection to the resolved tenant database.
- `IRedshiftS3AuthClauseFactory` centralizes Redshift S3 authorization.
- `IRedshiftCopyCommandFactory` emits Parquet COPY and must remain unchanged.
- `CuratedTargetExecutor` demonstrates transaction and command-timeout handling, but its target/date parameter model does not fit a whole-folder Fleet import.
- target DDL is defined in `0612_create_curated_pos_orders.sql` and `0611_create_curated_pos_order_items.sql`.

## Files To Create

Recommended locations:

- `backend/Kds.Ai.Integrations/FleetS3/IFleetS3RedshiftImportExecutor.cs`
- `backend/Kds.Ai.Integrations/FleetS3/FleetS3RedshiftImportExecutor.cs`
- `backend/Kds.Ai.Integrations/FleetS3/FleetS3ImportSql.cs`

Task 0 owns the shared request/result and diagnostic record files; Task 3 implements those frozen contracts.

Keep SQL explicit and source-specific. Inline raw string constants are acceptable; embedded SQL is also acceptable if the project is configured consistently and tests can load it.

## One-Connection Execution

All temporary data and permanent replacement work must use one open tenant connection:

1. create reusable session-scoped raw `SUPER` table;
2. create aggregate temporary order/item candidate tables with target-compatible types plus validation lineage;
3. for each date manifest in deterministic ascending order:
   - truncate the raw temporary table;
   - execute dedicated `COPY ... MANIFEST FORMAT JSON 'noshred'`;
   - assert loaded raw count equals manifest object count;
   - transform rows into aggregate candidates while injecting the manifest path date as `path_business_date`;
4. validate the complete aggregate candidates;
5. replace permanent rows in one transaction;
6. return counts and old/new affected dates.

The executor must not call the Parquet `IRedshiftCopyCommandFactory`.

## Dedicated COPY Shape

The command uses fixed temporary identifiers, a storage-produced manifest URI, centralized S3 auth, and no caller-provided SQL:

```sql
COPY fleet_s3_raw
FROM 's3://.../dt=2026-07-08.manifest.json'
<central auth clause>
MANIFEST
FORMAT JSON 'noshred';
```

Apply the configured long-running command timeout to COPY, validation, and replacement commands.

## Candidate Mapping

Implement every mapping in [proposed-solution.md](./proposed-solution.md#curated-mapping), including:

- canonical tenant and source metadata on both targets;
- `USD` currency;
- validated JSON/path/local timestamp date as `business_date`;
- normalized status/channel values;
- one-based item array ordinal;
- compact modifier JSON and source-order modifier summary;
- deterministic positive 32-bit order `id` using a documented Redshift hash expression;
- deterministic `{transaction_id}:{ordinal}` source line identity.

Recommended order ID expression:

```sql
ABS(MOD(FNV_HASH(transaction_id), 2147483646)) + 1
```

The collision check is mandatory; the hash is not treated as globally collision-free.

Channel normalization is exact:

- apply the documented normalization function to `order.order_type` for `service_channel`;
- apply it to nonblank `order.destination` for `fulfillment_channel`, falling back to `order.order_type` only when destination is null/blank;
- never merge two conflicting values or choose precedence heuristically.

## Complete Validation

Validation occurs before permanent deletes and returns aggregate counts plus at most the configured sample limit. It must detect:

- raw count mismatch with snapshot manifest index;
- root JSON type and schema version;
- all required root, `location`, `order`, `items`, `kds`, `partition`, item, and modifier fields/types from schema v1.0;
- empty item arrays;
- transaction ID pattern, uniqueness, nonempty value, and `VARCHAR(50)` fit;
- valid UTC/local timestamps;
- path date equals JSON partition date and local timestamp calendar date;
- `location.terminal_id` exactly equals the case-sensitive canonical source folder;
- integer range/positivity rules;
- decimal range/scale compatibility with `DECIMAL(10,2)` without silent rounding beyond the source contract;
- all target string lengths, including generated source IDs and modifier fields;
- order hash collisions and generated line-ID collisions;
- candidate order/item count greater than zero;
- every item has exactly one candidate parent;
- unsupported/malformed values before casts that could turn them into null.

Do not skip, truncate, or coerce invalid rows. Full payloads and personal fields must not appear in diagnostics.

## Atomic Replacement

After validation:

1. begin a transaction;
2. lock `curated.pos_orders` first and `curated.pos_order_items` second, matching the existing POS target order, before any delete;
3. query old affected dates for the exact tenant/source discriminator;
4. recheck matching-row existence and `overrideExisting` inside the transaction;
5. construct a replacement intent containing batch ID, old/new/affected dates, candidate counts, source identity, and a deterministic intent checksum;
6. invoke the Task 0 checkpoint contract and require PostgreSQL to durably persist that intent before any permanent delete;
7. delete items, then orders, using both exact `tenant_name` and `_etl_source_type`;
8. insert orders, then items, from candidates using `INSERT ... SELECT`;
9. verify inserted counts, duplicate business keys, source lineage, and item-parent integrity;
10. commit;
11. return the persisted replacement intent plus commit result for publication.

If replacement-intent persistence fails, roll back Redshift without deleting permanent data. Persisting an intent before a Redshift rollback is safe: reconciliation sees no complete current slice for this batch and may restart pre-commit execution. This ordering ensures that a crash immediately after Redshift commit cannot lose old-only dates that Athena must remove.

Never delete by tenant, source system, batch, business date, or order ID alone.

## Crash Reconciliation

A process can die after Redshift commits but before PostgreSQL records the commit flag. The replacement intent already exists at that point. Before any retry repeats replacement, load that intent and probe the current Fleet source slice:

- if every row in both target tables is stamped with the current batch ID and counts/integrity match the durable snapshot expectations, return a reconciled committed result without deleting/reinserting;
- if no current rows use this batch ID, normal pre-commit execution may continue;
- if lineage is mixed or counts are inconsistent, fail with a non-destructive reconciliation error requiring investigation.

This check makes an interrupted same-batch delivery safe when Task 1's execution lease is reacquired. A batch already marked as a handled pre-commit failure is terminal and is not automatically resumed.

## Testing

Add unit/recording-connection tests for:

- dedicated JSON/noshred/manifest COPY and centralized auth;
- no use or regression of Parquet COPY factory;
- one COPY per populated path date, not per file;
- fixed identifier and parameter/literal escaping;
- transaction/lock/delete/insert ordering;
- orders-then-items lock order and items-then-orders delete order;
- exact tenant/source delete predicates;
- rollback on every injected failure point;
- `overrideExisting` true/false behavior;
- old/new affected date union;
- crash before intent persistence, after intent persistence/before commit, and immediately after commit;
- commit reconciliation states.

Add opt-in live Redshift tests for SUPER traversal, array ordinals, all mappings, validation queries, hash collision detection fixture, rollback, and sample one-order/four-item result.

## Acceptance Criteria

- no permanent curated mutation happens before complete validation;
- replacement of both tables is one Redshift transaction;
- previous rows survive all pre-commit failures;
- only the exact tenant/Fleet source slice changes;
- repeated override runs produce the same business rows with new lineage;
- no curated DDL or staging table is added.