# TASK 4: Add Fleet-Specific Parquet Publication For Athena

## Summary

Export the committed Fleet source slice by affected date and entity to batch staging, validate the staged output, then idempotently activate one stable query-visible Parquet object per source/date/entity.

## Dependencies

- [TASK 0](./TASK0.md)
- [TASK 0](./TASK0.md) frozen replacement/publication contracts and active Athena key. Integrate with Task 3 after both implementations exist.

## Current Codebase Findings

- `CuratedExportExecutor` uses tenant Redshift `UNLOAD`, explicit projections supplied by catalogs, standard auth, and cleanup on failure.
- `BrinkPosCuratedExportCatalog` contains the current explicit POS order/item projection contracts and injects `tenant_id`.
- current `CuratedObjectKey` always includes `batch_id`, which would accumulate multiple query-visible Fleet generations.
- S3 does not provide a transaction across every date/entity object. Publication must be an idempotent journaled activation, not described as cross-system atomic.

## Files To Create

Recommended locations:

- `backend/Kds.Ai.Integrations/FleetS3/IFleetS3CuratedPublisher.cs`
- `backend/Kds.Ai.Integrations/FleetS3/FleetS3CuratedPublisher.cs`
- `backend/Kds.Ai.Integrations/FleetS3/FleetS3CuratedExportCatalog.cs`
- Fleet active/staging key implementation using the Task 0 contract.

Task 0 owns the shared publication request/result/journal records; Task 4 implements those frozen contracts.

Do not register Fleet catalog as the singleton `ICuratedExportCatalog`, because Brink currently owns that registration and Fleet has different filters/output semantics.

## Export Targets

Exactly two targets:

- `curated.pos_orders` -> entity `pos_orders`;
- `curated.pos_order_items` -> entity `pos_order_items`.

Copy the explicit column order from the current Brink contracts, including canonical `tenant_id` as the first column, but use Fleet filters:

```text
tenant_name = canonical tenantId
AND business_date = affected date
AND _etl_source_type = fleet_s3:{canonicalSourceFolder}
AND _etl_batch_id = current committed batchId
```

The batch predicate prevents publication of stale or mixed Fleet rows.

## Staged Export

For each affected date/entity:

1. count committed rows using the exact export filter;
2. if count is greater than zero, `UNLOAD` the explicit projection as Parquet to a batch-owned non-query-visible staging prefix with `MANIFEST VERBOSE`;
3. use `PARALLEL OFF` so one target/date produces one replaceable Parquet data object;
4. parse the verbose UNLOAD manifest and require its total row count to equal the pre-UNLOAD Redshift count;
5. list staged output and require exactly one nonempty Parquet data object plus its expected manifest;
6. record expected/manifest row counts, staged object identity/checksum, and URI in the publication journal before activation;
7. if count is zero, record an active-prefix removal action instead of exporting an empty object.

Staging must not occur below the stable `curated/` Fleet source boundary.

## Stable Active Boundary

Use this fixed source-scoped key:

```text
curated/company_id={customerId}/tenant_id={tenantId}/entity={entity}/
dt={yyyy-MM-dd}/source_type=fleet_s3/source_id={canonicalSourceFolder}/data.parquet
```

There is no `batch_id` in the active Fleet key. Batch lineage remains inside Parquet and in operational details. Task 0 verifies that deployed Glue entity/date locations recursively scan this descendant; Task 4 must not invent a different ordering.

## Activation And Retry

- Activate by server-side copying the validated staged object to the exact stable `data.parquet` key. S3 replacement of one key is atomic at object level.
- A replacement never creates a second active Fleet generation for the same entity/date/source.
- Delete the exact stable source prefix when the replacement has zero rows for an old affected date.
- Journal each entity/date activation after it succeeds. Retry skips entries whose stable object identity/checksum already matches the intended staged object.
- Before every activation or retry, verify Redshift still exposes this batch as the complete current tenant/source slice. If a newer import replaced it, mark the old publication attempt `superseded` and never overwrite newer Athena data.
- If activation fails midway, leave completed stable replacements in place and retry the remaining journal entries. Do not claim all-date atomic visibility.
- Batch success is withheld until every journal entry is active/removed and revalidated.

This design guarantees no duplicate active generation. It provides recoverable per-object activation rather than an impossible Redshift/S3 distributed transaction.

## Validation

For each active object verify:

- exact stable key boundary;
- object exists and is nonempty when expected count is nonzero;
- object checksum/ETag or version matches the activated staged artifact;
- verbose UNLOAD manifest row count equals the expected Redshift row count before activation;
- export SQL count remains the expected count for the current batch;
- both entity actions for every affected date are complete;
- no active data remains for removed dates;
- no Brink, KDS, another Fleet folder, customer, or tenant prefix is listed for deletion.

The live validation in Task 7 must query Athena and compare row counts/lineage with Redshift.

## Testing

Add tests for:

- exact explicit projections and absence of `SELECT *`;
- tenant/date/source/batch filters;
- `tenant_id` first-class column and `USD` data preservation;
- one staged Parquet object with `PARALLEL OFF`;
- `MANIFEST VERBOSE` total-row validation and mismatch rejection before activation;
- stable key generation and segment sanitization;
- overwrite of existing stable key without duplicate generation;
- removal of old dates;
- partial activation journal resume;
- newer-batch supersession protection;
- cleanup limited to staged/current Fleet prefixes;
- Brink export catalog/executor regression.

## Acceptance Criteria

- every affected date has complete order/item publication actions;
- active Fleet data has one stable object per entity/date/source;
- reruns replace rather than accumulate query-visible generations;
- publication can resume after Redshift commit without rerunning replacement;
- Athena can distinguish tenant and Fleet source lineage from projected columns;
- existing Brink/KDS exports are unchanged.