# Spensara (KDSAI) — Brink POS ingestion path: current state, gaps, recommended build
Prepared 2026-10-23 (ET). Read-only analysis. No commits, pushes, or AWS changes.

## 0. What I could and could not read (read this first)
| Source | Result |
|---|---|
| GitHub (primary) | **Not reachable.** No Origin namespaces visible; `gh` on the box is not logged in. |
| Xstore Gitea mirror (192.168.1.247:3000) | Reachable from TomDevAI, but an anonymous read shows **0 repos**. Repo `Bematechus/KDSAI`, branch `main`, needs a token. I did not use the vault `GITEA_XSTORE_TOKEN`, because pulling a credential to give myself access isn't sanctioned without Tom's go-ahead. |
| Box: `/workspace/research/kg/trees/tree_KDSAI.json` | A file listing of KDSAI `main` from an earlier read-only research pass (Oct 8). It has 1,000 of 2,091 paths, so it is truncated. **It has file names only, no file contents.** |
| `Y:\xampp\htdocs\Spensara\SESSION_LOG.md` (TomHO11) | Sage's notes from cloning and running KDSAI on `spensara-sandbox` (commit `f2962c7`). These were the main source for runtime behavior. |
| Memory / deliverables | Tom plans to take over the Redshift load from the Poland team. Maciej Kryca has an open "Redshift-to-S3 migration PR, then Athena for multi-tenant reads." |
| AWS | No CLI or credentials on the box. **I verified no live resources:** no Lambda, Glue, EventBridge, Redshift Serverless workgroups, or S3 buckets. |

Everything below about how the code behaves comes from file and class names plus the session log. Each point is labelled **[seen]** (a path exists or the log states it) or **[inferred]**. To confirm the inferred points, someone with Gitea access needs to read the roughly 12 files listed in §5.

## 1. End-to-end map (current)
Everything runs in **one .NET 10 / Aspire app** (`backend/Kds.Ai.slnx`, orchestrated by `backend/Kds.Ai.AppHost/AppHost.cs`). There are **no Lambda, Glue or Step Functions** in the tree for Brink [seen: no such paths in KDSAI]. A separate repo, `KDS.AWS.Devops.Infrastructure`, holds the infrastructure code.

```
Brink POS API (per-location token + service URL)
  │  BrinkPosEndpointCatalog / BrinkPosOperationCatalog / BrinkPosEndpointExecutor
  ▼
[1 Extract]  BrinkPosExtractOrchestrator → BrinkPosExtractStage / BrinkPosExtractJobs
  │  business-date windows: BrinkPosBusinessDateResolver, BrinkPosScheduleDateWindowBuilder,
  │  BrinkPosScheduleTimeZoneResolver, BrinkPosScheduleDefaults
  │  → raw payloads to S3 (EtlS3Options:BucketName; Services/Storage/AmazonS3ObjectStoreClient.cs)   [inferred raw zone]
  ▼
[2 Load→Staging]  BrinkPosLoadToStagingStage/Jobs, BrinkPosStagingTableCatalog
  │  → Redshift tenant DB, schema `staging` (37 tables; e.g. 0619_create_staging_brink_settings_get_modifier_groups.sql uses SUPER) [seen]
  ▼
[3 Normalize]  BrinkPosNormalizeLoadStage/Jobs, BrinkPosNormalizationCatalog, BrinkPosNormalizationContextFactory
  ▼
[4 Staging→Curated]  BrinkPosStagingToCuratedStage/Jobs, BrinkPosCuratedTargetCatalog, Etl/CuratedTargetExecutor.cs
  │  → schema `curated` (11 tables) [seen]
  ▼
[5 Curated→S3 export]  BrinkPosCuratedToS3Stage/Jobs, BrinkPosCuratedExportCatalog, Etl/CuratedExportExecutor.cs
  │  (Parquet curated zone; this is the Athena/multi-tenant-read direction in Maciej's PR) [inferred]
  ▼
Spensara API / analysis layer
   ApiService/Services/RedshiftWarehouseOrderReadService.cs, Endpoints/WarehouseOrderEndpoints.cs
   AnomalyDetector/Tools/WarehouseQueryToolProvider.cs, Agents/DataAccessAgentFactory.cs (read-only, tenant-scoped, 500 rows / 30 s)
```
Shared framework (`backend/Kds.Ai.Integrations/Etl/`) includes `EtlPipelineRunner`, `IEtlStage`, `IEtlSourceAdapter`, `StageExecutionSupport`, `StageCheckpointAdvanceRequest` and `StageBatchFailureRequest`. State is kept in the operational Postgres through `Data/Entities/IngestionBatch.cs`, `IngestionCheckpoint.cs`, `Enums/IngestionMode.cs` and `IngestionBatchStatus.cs`, added by migration `20260508093108_AddEtlIngestionTables`. Credentials come from `Data/Credentials/BrinkPosCredentialPayload.cs` and `CustomerCredentialBackedBrinkCredentialReadService.cs`, falling back to `BrinkConnectionOptionsFallbackResolver.cs` (the AppHost parameters `brink-pos-access-token`, `-location-token` and `-service-url`). Per-tenant provisioning is handled by `Services/CustomerEtlProvisioningService.cs`, `Jobs/CustomerWarehouseProvisioningJobs.cs` and `Data/Entities/TenantWarehouseMapping.cs`.

The KDS side has a more mature sibling pipeline: Debezium → SQS FIFO → `KdsCdc*`, with a duplicate detector, quarantine records, a replay index and a promotion retry service. **The Brink pipeline has none of these** (no quarantine or replay files exist under BrinkPos).

Warehouse status: Postgres RDS (`kds-ai-postgres…us-west-2`) is still in use, and Redshift Serverless is a migration in progress (per `docs/REDSHIFT.md`, quoted in the session log).

## 2. Manual steps and their triggers today
| # | Step | Trigger today |
|---|---|---|
| M1 | Tenant warehouse schema creation (`SetupTenantWarehouse`, Aspire `.WithExplicitStart()`) | A person clicks Start in the Aspire dashboard or runs the CLI (`Kds.Ai.Cli` `Commands/Warehouse/WarehouseCommand.cs`) [seen: "deliberately manual, by design"] |
| M2 | Tenant mapping (`ensure-tenant-warehouse`) | CLI, by a person [seen] |
| M3 | Brink credentials and location binding | Entered by an operator into the customer credential store, or AppHost user-secrets [seen] |
| M4 | Running Extract → Staging → Normalize → Curated → S3 | **An HTTP call to `Endpoints/BrinkPosEtlEndpoints.cs`** (or CLI) by a developer. `Models/BrinkPosEtlTriggers.cs` / `BrinkPosExtractTrigger.cs` suggest a "manual" vs "scheduled" trigger enum [inferred]. |
| M5 | Scheduling | `Jobs/BrinkCronSyncService.cs` exists (a hosted cron service) [seen]. Tom reports there is no schedule in practice, so it is most likely **disabled by config, not deployed, or covering extract only** [inferred; must confirm]. |
| M6 | Redshift-only SQL fixes (SUPER vs JSONB) and the CLI launch-profile typo (`--tenantId`) | Fixed by hand when they come up [seen] |
| M7 | Re-running a failed day | Manual re-invocation; there is no replay or quarantine tooling for Brink [inferred] |

## 3. Gaps blocking automatic near-real-time landing in curated
1. **The trigger is in-process.** Any cron runs inside the ApiService pod. If the API scales out, it fires twice; if the API is down, nothing runs. There's no external scheduler and no lease or lock [inferred].
2. **Incremental / watermark handling.** `IngestionCheckpoint` exists, but the Brink window builder works in *business-date* windows. Brink's documented model is SOAP/REST pulls by business date (for example GetOrders per location per day), not a change feed. Near-real-time therefore means **re-pulling "today" every N minutes**, which needs an intraday watermark (last order modified time / order id) [inferred; Brink has no webhooks/stream to my knowledge — **assumption**].
3. **Idempotency / dedupe.** Re-pulling today returns the same orders again. Staging → curated must be a key-based **MERGE** (Redshift supports `MERGE`, not `ON CONFLICT`). The Brink path has no duplicate detector like `KdsCdcDuplicateDetector` [seen absent].
4. **Schema drift.** Raw lands in `SUPER` (good), but `BrinkPosNormalizationCatalog` is hard-coded contracts. There's no drift detection, quarantine (`KdsCdcQuarantineRecord` has no Brink twin), or alert when a new field or type appears.
5. **Retries / backoff.** No visible Polly or retry policy for Brink rate limits and timeouts, and no dead-letter. `StageBatchFailureRequest` records failure but doesn't retry [inferred].
6. **Secrets.** Three plaintext credentials are committed (`appsettings.Development.json`, `Cli/Properties/launchSettings.json` to real RDS as the `postgres` superuser). The Brink tokens fall back to app parameters. They should move to Secrets Manager per tenant and the exposed ones should be rotated [seen].
7. **Monitoring.** No CloudWatch metrics or alarms on freshness, rows per run or failures. Batch status sits only in Postgres tables.
8. **Warehouse target is ambiguous.** RDS vs Redshift Serverless vs (Maciej's PR) S3+Athena. The automation has to target one source of truth for "curated", or it will be rebuilt twice.
9. **Provisioning is manual (M1/M2).** A new customer can't go live without a person.

## 4. Recommended build
**Principle:** keep the existing .NET stages (they're already modular via `IEtlStage`). Replace the *trigger and state machine*, and harden each stage. Don't rewrite in Glue.

| # | Component | Detail | Effort |
|---|---|---|---|
| B1 | **Stage runner as an ECS Fargate task (or a worker container)** | Package `Kds.Ai.Cli`, or a new `Kds.Ai.EtlWorker`, with a `run-brink --tenant --location --stage --window` command. Same code, out of the API process. | 3–4 d |
| B2 | **EventBridge Scheduler → Step Functions** (one state machine per tenant, using a Map over locations) | Extract → LoadToStaging → Normalize → StagingToCurated → (optional) CuratedToS3. Each step is an ECS RunTask (`.sync`) with retry (3×, exponential backoff, jitter) and Catch → failure state → SNS. One execution name per tenant/window gives idempotent start, so no double-fire. | 4–5 d |
| B3 | **Cadence** | Intraday: every **5 min** for the current business date (target: **curated ≤ 10 min behind the POS**; the Brink rate limit may force 15 min, **to be confirmed with PAR**). Nightly 04:00 local: a full re-pull of D-1 and D-2 to catch voids, refunds and late closes. Weekly: a D-7 reconcile. | incl. |
| B4 | **Watermarks** | Extend `IngestionCheckpoint` to (tenant, location, endpoint, business_date, last_modified_utc / max_order_id). Advance only after the curated MERGE commits. | 2 d |
| B5 | **Idempotent curated load** | Staging gets a `batch_id`, a `_ingested_at` column and a natural key. Curated uses `MERGE INTO curated.x USING (dedup with ROW_NUMBER) …` inside one transaction per batch. Delete-then-insert per (location, business_date) for small dimension tables. | 3–4 d |
| B6 | **Drift and quarantine** | Raw JSON stays in S3 + `SUPER`. The normalizer validates against a versioned contract, and unknown or invalid records go to `brink_quarantine` (mirror `KdsCdcQuarantineRecord`). New fields alert and stay in raw; breaking changes stop that endpoint only. | 3 d |
| B7 | **Secrets** | Per-tenant Brink tokens in Secrets Manager (`spensara/{tenant}/brink`), read by the task role. Rotate the three exposed DB credentials and remove them from git history. | 1–2 d |
| B8 | **Monitoring / alerting** | Emit CloudWatch metrics (rows, latency, freshness lag = now − max curated order time) for each location. Alarms: lag > 30 min intraday, any Step Functions failure, quarantine > 0. SNS → email/Teams. A freshness view in Bridge Admin is optional. | 2–3 d |
| B9 | **Provisioning automation** | When a customer is created, run SetupTenantWarehouse + mapping + schedule creation as a Step Functions branch (removes M1–M3). | 2–3 d |
| B10 | **Decide the warehouse** | Pick Redshift Serverless (with MERGE and SUPER) as curated, with S3 Parquet export kept for Athena and replay. Fix the SUPER/JSONB dev split with a dialect switch. Merge Maciej's PR to match. | 1 d (decision + review) |
| B11 | Tests and runbook | Use the simulator plus a recorded-Brink-payload replay test, and a backfill command (`--from --to`). | 3 d |

**Total ≈ 27–33 engineer-days (about 6 weeks for one engineer, or 3–4 weeks for two).** The smallest useful slice is B1+B2+B5+B7 with a 15-minute cadence, about 2.5 weeks.

**Streaming (Kinesis / Redshift streaming ingestion)** doesn't fit unless PAR offers a push feed. Brink is pull-based, so scheduled micro-batches are the right model [assumption].

## 5. To confirm with Gitea or GitHub read access (about 1 hour)
`Jobs/BrinkCronSyncService.cs` (is it enabled, and what's the cron?), `Endpoints/BrinkPosEtlEndpoints.cs`, `Models/BrinkPosEtlTriggers.cs`, `BrinkPosScheduleDefaults.cs`, `BrinkPosStagingToCuratedStage.cs` (insert vs merge), `Etl/CuratedTargetExecutor.cs`, `BrinkPosExtractStage.cs` (does raw go to S3 first?), `AppHost.cs`, `.github/workflows/backend-pipeline.yml`, `docs/REDSHIFT.md`, and `KDS.AWS.Devops.Infrastructure` (is there an existing ECS/EventBridge/Redshift definition?). Also needed: read-only AWS access in us-west-2 to list Redshift Serverless workgroups, S3 buckets, ECS services and EventBridge schedules.