# Fleet S3 Temporary Import - Implementation Tasks ## Overview Implement the approved temporary, admin-triggered import from customer Fleet JSON objects in S3 into the existing tenant Redshift curated order tables, followed by source-scoped Parquet publication for Athena. Behavioral source of truth: - [proposed-solution.md](./proposed-solution.md) - [transaction_schema.json](./transaction_schema.json) - [ORD-015D3A48C0.json](./ORD-015D3A48C0.json) This is an explicit temporary exception to the normal source-adapter pipeline. It must reuse shared platform infrastructure where it fits, but it must not add Fleet staging tables or force Fleet-specific semantics into Brink/KDS catalogs. The implementation outcome is: - an admin reserves and enqueues one import for an exact customer, tenant, and `sourceFolder`; - the application creates a batch-owned immutable S3 snapshot without downloading all payloads; - Redshift bulk-loads and validates the snapshot through temporary `SUPER` data; - one transaction replaces only the exact tenant/Fleet-folder slice in `curated.pos_orders` and `curated.pos_order_items`; - every affected business date is published as explicit-projection Parquet under a stable Fleet source boundary for Athena; - post-commit publication can resume without repeating the Redshift replacement; - operational batches expose bounded, non-sensitive diagnostics; - no curated Redshift DDL changes are introduced. ## Current Repository State - `BrinkPosEtlEndpoints` and `BrinkPosEtlJobs` are the closest manual API/TickerQ references. - `CustomerEtlProvisioningService` resolves stored customer identity, but the Fleet flow additionally requires an explicit customer/tenant match and a successfully provisioned active warehouse. - `IngestionBatch` supports nullable `BusinessDate`, details JSON, and generic status updates. It does not currently expose a durable source scope for race-safe per-folder reservation. - `AmazonS3ObjectStoreClient` can upload, copy, read, and delete objects, but its abstraction does not list object metadata or perform conditional snapshot copies. - `RedshiftCopyCommandFactory` is intentionally Parquet-only. Fleet JSON `noshred` COPY must use a dedicated command path. - `CuratedExportExecutor` and `BrinkPosCuratedExportCatalog` establish explicit projection and `UNLOAD` conventions, but their date, source filter, and batch-key output semantics do not satisfy Fleet replacement publication. - The existing curated target schemas already contain all columns required by this temporary mapping. ## Spec Name This implementation spec should be referred to as **Fleet S3 Temporary Import**. ## Fixed Decisions - API route: `POST /etl/fleet-s3/import`, protected by the existing admin policy. - Required identity: caller supplies both `customerId` and `tenantId`; they must match exactly. - `sourceFolder` is one safe immediate path segment and is the source/store identity. - `overrideExisting` defaults to `true`; `false` is a create-only guard. - Source bucket/root are server-owned configuration, initially `s3://kds-warehouse-ai-test/fleet/`. - Source payload contract is schema version `1.0`. - Target tables are only `curated.pos_orders` and `curated.pos_order_items`. - Curated source identity is `source_system = 'fleet_s3'` and `_etl_source_type = 'fleet_s3:{canonicalSourceFolder}'`. - Curated `business_date` comes from JSON partition values and must match both the path date and local timestamp date. - Currency is `USD`. - Item array order is stable and defines the one-based line ordinal. - Import semantics are full snapshot replacement, not append/upsert by file. - Athena Parquet publication is required before the batch succeeds. - `IngestionBatch.BusinessDate` remains null for the whole-folder batch. - Curated Redshift schema remains unchanged. - The feature remains outside `IEtlSourceAdapter`; promotion to the standard ETL architecture is a separate future decision. ## Technical Refinements The detailed implementation must make two guarantees that an S3 key-only manifest cannot provide by itself: 1. **Immutable content:** a manifest freezes URLs, not mutable object bytes. Each listed source object must therefore be conditionally copied, using its listed ETag/version precondition, to a batch-owned platform prefix before loading. 2. **Path-to-payload date proof:** Redshift JSON `COPY` into one `SUPER` column does not expose the source S3 key. Snapshot entries must therefore be grouped into one manifest per path business date. Each date manifest is loaded and transformed with that path date supplied as a trusted literal, allowing comparison with the JSON partition and `timestamp_local`. This means the implementation performs one bulk `COPY` per populated business date, not one `COPY` per file and not one untraceable year-wide `COPY`. The validated `sourceFolder` is case-sensitive and remains unchanged. S3 keys are case-sensitive, so lowercasing a valid folder would select a different source. The exact canonical value is used for the source prefix, terminal/store identity, operational scope, and curated discriminator. The active Athena key is fixed by this specification: ```text curated/company_id={customerId}/tenant_id={tenantId}/entity={entity}/ dt={yyyy-MM-dd}/source_type=fleet_s3/source_id={canonicalSourceFolder}/data.parquet ``` The deployed Glue table location for an entity/date must scan descendants of the `dt=.../` root. Catalog compatibility is a Task 0 prerequisite and a release gate, not a reason to change key ordering during Task 4. ## Scope - shared constants, options, models, and validation rules; - race-safe operational reservation keyed by tenant and Fleet source folder, plus a renewable per-batch worker lease; - paginated S3 inventory, conditional server-side snapshot copy, manifest index, and date manifests; - dedicated Redshift JSON COPY, temporary candidate transformation, complete validation, and atomic replacement; - Fleet-specific explicit export projections and stable source-scoped Athena publication; - resumable post-commit publication and retained-workspace cleanup; - admin endpoint, TickerQ functions, DI/configuration wiring, diagnostics, and tests; - live Redshift/S3/Athena validation gated by existing live-test conventions. ## Out of Scope - permanent Fleet raw/normalized/staging schemas; - `IEtlSourceAdapter` stages, checkpoints, schedules, or event-driven imports; - incremental file ledger or ETag-based business reconciliation; - arbitrary caller-supplied buckets, prefixes, SQL, database names, or target tables; - curated table migrations; - KDS lifecycle reconstruction; - Glue/Athena catalog redesign; - genericizing Fleet-only behavior before a second concrete use case exists. --- ## Task Checklist ### Phase 0: Contracts And Configuration - [x] [TASK 0](./TASK0.md) - Freeze implementation contracts, constants, options, and phase details ### Phase 1: Operational Reservation - [ ] [TASK 1](./TASK1.md) - Add durable per-tenant/source batch reservation and publication-resume support ### Phase 2: Immutable Source Snapshot - [ ] [TASK 2](./TASK2.md) - Add paginated Fleet source inventory, conditional snapshot copies, and date manifests ### Phase 3: Redshift Import - [ ] [TASK 3](./TASK3.md) - Add one-session JSON load, candidate validation, and atomic curated replacement ### Phase 4: Athena Publication - [ ] [TASK 4](./TASK4.md) - Add Fleet-specific staged Parquet export and stable source-scoped activation ### Phase 5: Orchestration And Recovery - [ ] [TASK 5](./TASK5.md) - Add import orchestration, durable phase recovery, diagnostics, and cleanup ### Phase 6: Admin And Scheduler Wiring - [ ] [TASK 6](./TASK6.md) - Add admin endpoints, TickerQ functions, and module registration ### Phase 7: End-To-End Validation - [ ] [TASK 7](./TASK7.md) - Add regression, live Redshift/S3/Athena, performance, and removal validation --- ## Dependency Graph ```text TASK 0 contracts/options +-- TASK 1 operational reservation +-- TASK 2 immutable S3 snapshot +-- TASK 3 Redshift import +-- TASK 4 Athena publication TASK 5 orchestration +-- depends on TASKS 1, 2, 3, 4 TASK 6 API/TickerQ wiring +-- depends on TASKS 1 and 5 TASK 7 end-to-end validation +-- depends on TASKS 2, 3, 4, 5, 6 ``` Tasks 1 through 4 can be implemented in parallel only after Task 0 freezes all shared request/result contracts and verifies the deployed Athena table location against the fixed active key. ## Recommended Session Order | Session | Tasks | Description | |---|---|---| | 1 | 0 | Freeze names, options, inputs/results, details schema, and invariants. | | 2 | 1 | Add operational source scope, unique running reservation, and resume transitions. | | 3 | 2 | Build immutable batch workspace and date manifests. | | 4 | 3 | Implement temporary `SUPER` load, validation, and curated transaction. | | 5 | 4 | Implement staged Parquet export and stable Fleet activation. | | 6 | 5 | Compose phases and prove crash-safe resume behavior. | | 7 | 6 | Expose admin enqueue/retry routes and TickerQ execution. | | 8 | 7 | Run full regression and opt-in live validation. | ## Success Boundary The import is successful only when all of the following are true: 1. the immutable source snapshot and manifest index exist; 2. all candidate rows passed validation; 3. the exact Redshift source slice committed and is stamped with the current batch ID; 4. all affected date/entity Parquet objects were activated or removed as required; 5. batch details contain final counts and publication diagnostics; 6. the operational batch is `Succeeded`. A Redshift commit followed by an Athena publication failure is a failed, resumable batch, not success. ## Delivery Guardrails - Keep Fleet-specific SQL, path rules, and models together under a dedicated Fleet namespace. - Do not modify Brink/KDS catalogs to understand Fleet source identity. - Do not broaden `IRedshiftCopyCommandFactory`; its current Parquet contract must remain unchanged. - Do not put all 36,500+ object keys in `IngestionBatch.DetailsJson`; persist inventory/manifests in S3 and store bounded summaries plus URIs. - Never log source payloads or unbounded validation rows. - Do not claim cross-system atomicity between Redshift and S3. Make activation idempotent and recoverable instead. - Every permanent delete must include canonical tenant ID and exact Fleet source discriminator.