Categories
Brink & Borg Architecture (2)
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
- 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].
- Incremental / watermark handling.
IngestionCheckpointexists, 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]. - Idempotency / dedupe. Re-pulling today returns the same orders again. Staging → curated must be a key-based MERGE (Redshift supports
MERGE, notON CONFLICT). The Brink path has no duplicate detector likeKdsCdcDuplicateDetector[seen absent]. - Schema drift. Raw lands in
SUPER(good), butBrinkPosNormalizationCatalogis hard-coded contracts. There's no drift detection, quarantine (KdsCdcQuarantineRecordhas no Brink twin), or alert when a new field or type appears. - Retries / backoff. No visible Polly or retry policy for Brink rate limits and timeouts, and no dead-letter.
StageBatchFailureRequestrecords failure but doesn't retry [inferred]. - Secrets. Three plaintext credentials are committed (
appsettings.Development.json,Cli/Properties/launchSettings.jsonto real RDS as thepostgressuperuser). The Brink tokens fall back to app parameters. They should move to Secrets Manager per tenant and the exposed ones should be rotated [seen]. - Monitoring. No CloudWatch metrics or alarms on freshness, rows per run or failures. Batch status sits only in Postgres tables.
- 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.
- 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.