feat(ingestion): VehIQ OLTP -> ClickHouse watermark extractor (P070) #117
@@ -0,0 +1,44 @@
|
||||
name: VehIQ OLTP extractor (0pt ingestion)
|
||||
|
||||
# P070 — scheduled poll of app.dealer_stock_entry/app.deal (VehIQ, server10)
|
||||
# into opt_dataset_projection (ClickHouse), via the watermark extractor in
|
||||
# src/ingestion/vehiq-oltp-ingest-runner.ts. See docs/P070_VEHIQ_OLTP_
|
||||
# EXTRACTOR.md for the full design and "what activating this requires".
|
||||
#
|
||||
# SAFE BY DEFAULT: VEHIQ_INGEST_MODE defaults to "preview" (read-only, never
|
||||
# commits a checkpoint, never writes) below. With no VEHIQ_SOURCE_DATABASE_URL
|
||||
# secret configured on this repo, the script logs "not_configured" and exits
|
||||
# 0 — this workflow can be left enabled indefinitely with zero effect on
|
||||
# anything until BOTH the secret is added AND VEHIQ_INGEST_MODE is changed
|
||||
# to "run". Neither is done by this workflow file itself.
|
||||
on:
|
||||
workflow_dispatch:
|
||||
schedule:
|
||||
- cron: "*/15 * * * *"
|
||||
|
||||
jobs:
|
||||
extract:
|
||||
runs-on: ubuntu-latest
|
||||
env:
|
||||
VEHIQ_SOURCE_DATABASE_URL: ${{ secrets.VEHIQ_SOURCE_DATABASE_URL }}
|
||||
DATABASE_URL: ${{ secrets.DATABASE_URL }}
|
||||
OPT_ANALYTICS_CLICKHOUSE_BASE_URL: ${{ secrets.OPT_ANALYTICS_CLICKHOUSE_BASE_URL }}
|
||||
OPT_ANALYTICS_CLICKHOUSE_BASIC_AUTH: ${{ secrets.OPT_ANALYTICS_CLICKHOUSE_BASIC_AUTH }}
|
||||
OPT_ANALYTICS_CLICKHOUSE_DATABASE: opt_analytics
|
||||
# Activation gate — see docs/P070_VEHIQ_OLTP_EXTRACTOR.md. Flip to
|
||||
# "run" only once steps 1-4 in that doc's "What activating this
|
||||
# requires" section are actually done.
|
||||
VEHIQ_INGEST_MODE: preview
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: actions/setup-node@v4
|
||||
with:
|
||||
node-version: "22"
|
||||
- name: Enable corepack + pin pnpm@9
|
||||
run: |
|
||||
corepack enable
|
||||
corepack prepare pnpm@9 --activate
|
||||
- name: Install (frozen lockfile)
|
||||
run: pnpm install --frozen-lockfile
|
||||
- name: Run the extractor (preview mode unless activated)
|
||||
run: pnpm run vehiq:oltp-ingest
|
||||
@@ -0,0 +1,113 @@
|
||||
# P070 — VehIQ OLTP -> 0pt watermark extractor
|
||||
|
||||
Status: built, tested, verified read-only against production. **Not scheduled/activated in production.** Read "What activating this requires" before turning the workflow on.
|
||||
|
||||
## What this is
|
||||
|
||||
A controlled, incremental extractor implementing exactly the pattern Simon specified (`admin/vehiq` 2026-09-25 pass, "Locked architecture decisions from Simon" / `docs/adr/0035-bi-analytics-architecture-product-firewall.md`):
|
||||
|
||||
```
|
||||
OLTP operational tables -> read-only analytical extraction -> watermark ->
|
||||
0pt ingestion -> ClickHouse (opt_dataset_projection)
|
||||
```
|
||||
|
||||
Source: `app.dealer_stock_entry` and `app.deal` on production Postgres (`vehiq-db`, server10) — read-only `SELECT`, never a schema/config change, never touching `wal_level` or replication, never a permanent replica. Sink: `SiaxAnalyticsEngine.ingestProjection` (BI-P009, `opt_dataset_projection` in ClickHouse). Vehicle dimension: a narrow, keyed lookup against `vehiq.vehicle` (98M rows / 85GB) scoped to exactly the `vehicle_id`s referenced by the `dealer_stock_entry` batch just read — never `SELECT * FROM vehicle`, never a full scan.
|
||||
|
||||
## Why this lives in `admin/0pt`, not `admin/vehiq`
|
||||
|
||||
The task that produced this work described building it as a new package inside `admin/vehiq` (e.g. `packages/opt-ingest`). Before writing any code, this repo's own `docs/adr/0035-bi-analytics-architecture-product-firewall.md` (decisions 1, 2, 10 — decided the same day, one-way-door, CI-enforced by `tools/validate_engine_firewall.py`) was read and is unambiguous:
|
||||
|
||||
> VehIQ's own codebase must **never** get its own ClickHouse or DuckDB credentials, and must never import a ClickHouse/DuckDB client directly. ... Product code must never hammer the live OLTP Postgres with analytical queries directly.
|
||||
|
||||
And `admin/vehiq`'s `docs/CURRENT_STATE.md` ("Connection details for a future API0 connector", same pass) says explicitly:
|
||||
|
||||
> VehIQ's own codebase must never see any of this. This is 0pt's data plane; only a future `admin/api0` connector should hold these details.
|
||||
|
||||
Building the extractor inside `admin/vehiq` would have failed `validate_engine_firewall.py` the moment it imported a ClickHouse client or held a `CLICKHOUSE_*` env var, and would have been the exact "temporary VehIQ-owned data-access surface" ADR-0035 names as the failure mode it exists to prevent. So this extractor was built here instead — `admin/0pt`, 0pt's own data plane, a sibling of `admin/api0` (the mechanics belong next to the P060-P068 ingestion plane and the BI-P009/P010 ClickHouse engine that already live in this repo). **`admin/vehiq` was not modified by this work.**
|
||||
|
||||
This is a deliberate deviation from the task's literal repo suggestion, made because the suggestion conflicted with Simon's own same-day, explicit, locked architecture decision discovered while reading the target repo — not a judgment call made against that decision.
|
||||
|
||||
## Real production schema does not match the task's assumed column names
|
||||
|
||||
The task (and `admin/vehiq`'s `docs/CURRENT_STATE.md`) assumed all three OLTP tables carry a generic `updated_at` ("the columns already exist"). Verified live against production (server10, `vehiq-db`, read-only `information_schema.columns`, 2026-09-25) — **this is not true**:
|
||||
|
||||
| table | live columns | usable watermark |
|
||||
|---|---|---|
|
||||
| `app.dealer_stock_entry` | `id, tenant_id, vehicle_id, lot_location, floorplan_status, listed_at, sold_at` | `listed_at` (NOT NULL, insert-time only) |
|
||||
| `app.deal` | `id, tenant_id, party_id, vehicle_id, sale_price_minor, currency, status, created_at` | `created_at` (NOT NULL DEFAULT now(), insert-time only) |
|
||||
| `app.listing` | `id, tenant_id, vehicle_id, asking_price_minor, currency, channels, status` | **none** |
|
||||
|
||||
**`app.listing` has no timestamp column of any kind today.** `admin/vehiq` migration `0084_listing_deal_lifecycle_timestamps.sql` (already written, adds `published_at`/`unpublished_at` to `listing` and `closed_at`/`listing_id` to `deal`) exists in the repo but is **not yet applied to production** (verified: those columns are absent live). Until it lands, there is no honest way to build a watermark connector for `app.listing` — one is not built here. **This extractor covers `dealer_stock_entry` and `deal` only, not `listing`.**
|
||||
|
||||
**Caveat on the two sources that ARE built:** `listed_at`/`created_at` are set once, at INSERT. Neither table has a column that advances when an *existing* row changes (`sold_at`/`floorplan_status` on `dealer_stock_entry`; `status` on `deal`). This extractor reliably sees every new row; it does **not** see an in-place UPDATE to a row it already read, because that row has already passed the watermark. Closing this gap needs either `listing_id`/`closed_at` (migration 0084, once applied) plus an equivalent for `dealer_stock_entry`, or a generic `updated_at` + trigger — a schema decision for Simon, not something this pass made unilaterally (per the hard "never a schema/config change" rule).
|
||||
|
||||
Also found and corrected while building: `vehiq.vehicle`'s primary key column is `vehicle_id`, not `id` (verified via schema introspection only — no vehicle data rows were read to establish this).
|
||||
|
||||
## Where the code lives
|
||||
|
||||
- `src/connectors/postgres-watermark-connector.ts` — the P061 `Connector` implementing `WHERE watermark_col > $1 OR (watermark_col = $1 AND key_col > $2) ORDER BY watermark_col, key_col LIMIT $3` (the `OR` tuple form, not a bare `>`, so a same-timestamp tie at a page boundary is not silently dropped — see the file's own header for the bug this avoids). Explicit column allow-list, never `SELECT *`. `validate()` structurally refuses any config pointed at `vehiq.vehicle`.
|
||||
- `src/connectors/postgres-keyed-lookup-connector.ts` — the narrow vehicle-dimension connector. Config carries an already-resolved, finite `keys` array; `connect()` refuses an empty list rather than defaulting to a full read; every query is `WHERE vehicle_id = ANY($1)` chunked to `keys_per_query` (default 500).
|
||||
- `src/connectors/postgres-query-impl.ts` — the real `postgres` (postgres.js) implementation of the connectors' injectable `QueryImpl`, used only by the orchestration script. Refuses to execute anything whose first keyword isn't `SELECT`, independent of the DB role's own privileges.
|
||||
- `src/ingestion/projection-sink.ts` — adapts the P062 runner's `IngestionSink` onto `SiaxAnalyticsEngine.ingestProjection` (writes `opt_dataset_projection`). No ClickHouse import — only the `SiaxAnalyticsEngine` port type.
|
||||
- `src/ingestion/vehiq-oltp-ingest-runner.ts` — the orchestration: `previewVehiqOltpSource` (one bounded read-only read, never commits, safe against production at any time) and `runVehiqOltpIngestion` (the real P062-runner path).
|
||||
- `scripts/vehiq-oltp-ingest/run.ts` — thin CLI (`pnpm run vehiq:oltp-ingest`), env-driven, defaults to `preview` mode.
|
||||
- Tests: `tests/connectors/postgres-watermark-connector.test.ts`, `tests/connectors/postgres-keyed-lookup-connector.test.ts`, `tests/ingestion/vehiq-oltp-ingest-runner.test.ts`.
|
||||
|
||||
## Checkpoint field mapping
|
||||
|
||||
Simon's own specified checkpoint columns — `source, table, last_updated_at, last_id, batch_count, checksum, ingested_at` — mapped onto the **existing** P062 `IngestionCheckpoint` (reused, not duplicated, per the "no double truths" rule):
|
||||
|
||||
| Simon's field | Canonical field |
|
||||
|---|---|
|
||||
| `source` | `IngestionCheckpoint.source_id` |
|
||||
| `table` | `IngestionCheckpoint.partition_key` |
|
||||
| `last_updated_at` | decoded from `IngestionCheckpoint.cursor` (JSON `WatermarkCursor`) |
|
||||
| `last_id` | decoded from `IngestionCheckpoint.cursor` |
|
||||
| `checksum` | decoded from `IngestionCheckpoint.cursor` — sha256 of the page just landed |
|
||||
| `batch_count` | `IngestionCheckpoint.records_committed` (cumulative); a per-run page count is also in each run's `PartitionRunReport` |
|
||||
| `ingested_at` | `IngestionCheckpoint.committed_at` |
|
||||
|
||||
Durable storage: `createIngestionRepository(DATABASE_URL)` — Postgres-backed when `DATABASE_URL` (0pt's own metadata store) is set, in-memory only in non-production/tests. `run` mode refuses to start without `DATABASE_URL` explicitly, so a scheduled run can never silently fall back to re-reading everything from the beginning.
|
||||
|
||||
## Testing
|
||||
|
||||
Fixture-backed (no real infra): watermark + limit correctness, a same-timestamp tie at a page boundary, resume-from-checkpoint after a simulated crash (via a sink that throws mid-run, mirroring `tests/p0/ingestion-runner.test.ts`'s own technique), the empty-table case (zero rows, zero envelopes, no throw), and the narrow vehicle-dimension lookup asserted at the *query* level (every id ever bound into a query is a subset of the referenced set; a vehicle never referenced by the batch is never queried).
|
||||
|
||||
Also verified against a **real, disposable Postgres container** (not just fixtures), both seeded and truncated-to-empty, before touching production — caught and fixed two real bugs the fixture-only tests could not have caught:
|
||||
|
||||
1. `timestamptz` columns come back from the real driver as JS `Date` objects, not strings — the initial cursor-encoding logic did `String(date)` (locale-dependent, useless as a stored watermark) instead of `.toISOString()`. Fixed; regression test added (`tests/connectors/postgres-watermark-connector.test.ts`, "real Postgres driver value shapes").
|
||||
2. The same `Date`-vs-string mismatch tripped the P063 schema-conformance gate on every row (`field "listed_at" is object, expected string`) until the connector's `toRecord()` was made to normalize every `Date`-valued payload field to ISO before building the envelope.
|
||||
|
||||
**Verified read-only against actual production** (server10, `vehiq-db`, via `fleet`), 2026-09-25: the connector's exact SQL (`SELECT id, tenant_id, vehicle_id, lot_location, floorplan_status, listed_at, sold_at FROM app.dealer_stock_entry ORDER BY listed_at, id LIMIT 100` and the equivalent for `app.deal`) executed directly, read-only, and returned **zero rows, no error** — confirming both that the extractor's query shape is valid against the live schema and that there is genuinely nothing to extract yet. `vehiq.vehicle`'s `vehicle_id` primary-key column name was confirmed via `information_schema.columns` only — no vehicle data rows were read. **The compiled service itself was not run against production** — only its exact SQL, via read-only `psql`, matching this pass's "prove it does nothing, don't deploy a live service yet" scope.
|
||||
|
||||
## Deferred: full vehicle-dimension backfill
|
||||
|
||||
Not built or run in this pass, per the task's explicit instruction. If/when a full `dimension_vehicle` snapshot is ever wanted (as opposed to the permanent narrow per-batch lookup this extractor does forever), it needs its own deliberate decision:
|
||||
|
||||
- **Chunking**: page through `vehiq.vehicle` by primary key range (`vehicle_id > :last ORDER BY vehicle_id LIMIT :n`), never an unbounded scan — the same shape `PostgresWatermarkConnector` already implements, just pointed at a PK range instead of a timestamp (this connector's own `validate()` refuses `vehiq.vehicle` by name specifically to keep this a distinct, deliberate choice rather than an accidental default).
|
||||
- **Throttle**: an explicit delay between chunks, watching server10's `iostat`/`vmstat`/CPU while it runs — 98M rows / 85GB is enough to matter even read-only.
|
||||
- **Checksum + reconciliation**: a row-count and content checksum comparison between source and `opt_dataset_projection` after the backfill, recorded, before trusting it.
|
||||
- **Watermark handoff**: the backfill's end position must hand off cleanly to the narrow per-batch lookup this extractor already does, so nothing is double-counted or gapped at the boundary.
|
||||
|
||||
This mirrors the numbered runbook `admin/vehiq`'s `docs/CURRENT_STATE.md` already wrote for the general CDC case (not executed there either) — the shape is consistent, deliberately, rather than inventing a second convention.
|
||||
|
||||
## Deferred: `app.listing` extraction
|
||||
|
||||
Blocked on a schema decision, not an engineering gap: `app.listing` has no timestamp column today. Unblocking it needs one of:
|
||||
|
||||
1. Apply `admin/vehiq` migration `0084_listing_deal_lifecycle_timestamps.sql` (adds `published_at`/`unpublished_at`) — a decision for whoever owns that repo's migration pipeline, not this pass.
|
||||
2. Add a generic `updated_at` + trigger to `dealer_stock_entry`/`deal`/`listing` — closes both this gap and the "insert-only watermark misses in-place updates" caveat above in one move, but is a real schema change on production and needs Simon's explicit sign-off per the hard rule against touching production schema unsupervised.
|
||||
|
||||
Once either lands, adding a third `VehiqOltpSourceDef` entry to `VEHIQ_OLTP_SOURCES` (`src/ingestion/vehiq-oltp-ingest-runner.ts`) is the only code change needed — the connector, runner, sink and tests are all already generic across sources.
|
||||
|
||||
## What activating this in production actually requires
|
||||
|
||||
Nothing here is scheduled or running. To turn it on:
|
||||
|
||||
1. **A read-only Postgres role on `vehiq-db`** (server10) with `SELECT` on exactly `app.dealer_stock_entry` and `app.deal` (not the whole `app` schema, not `vehiq.vehicle` beyond what the narrow lookup needs) — creating this role is a production database change and should be coordinated explicitly, the same posture `admin/vehiq`'s own CDC runbook already takes for a bigger version of this same question. Its connection string becomes `VEHIQ_SOURCE_DATABASE_URL`.
|
||||
2. **`DATABASE_URL`** — 0pt's own metadata Postgres, for durable checkpoints. Already exists for every other 0pt capability; reuse it, do not provision a new one for this.
|
||||
3. **`OPT_ANALYTICS_CLICKHOUSE_BASE_URL` / `OPT_ANALYTICS_CLICKHOUSE_BASIC_AUTH`** — the already-deployed `opt-analytics-clickhouse` on server6 (internal-only, reachable from containers on the `coolify` network — see `admin/vehiq`'s `docs/CURRENT_STATE.md`, "0pt production ClickHouse 26.8 LTS"). A real least-privilege application user should be provisioned for this connector specifically, not the schema-admin `opt_admin` user mentioned there.
|
||||
4. **`VEHIQ_INGEST_MODE=run`** on the scheduled workflow (default is `preview`, which never writes).
|
||||
5. A decision on whether `.gitea/workflows/vehiq-oltp-ingest.yml`'s cron schedule should actually be enabled (it is present and correctly configured, but Gitea Actions secrets are not set on this repo by this pass — an unset `VEHIQ_SOURCE_DATABASE_URL` makes every scheduled run a clean no-op, so leaving the workflow file in place with no secrets configured is safe, but is not the same as "this is now extracting real data").
|
||||
|
||||
None of steps 1-3 were done in this pass — they require production credentials/role provisioning this pass does not have standing authority to create unsupervised.
|
||||
+3
-1
@@ -16,7 +16,8 @@
|
||||
"format:write": "prettier --write .",
|
||||
"typecheck:app": "tsc --noEmit -p tsconfig.json",
|
||||
"siax:doctor": "node scripts/siax-doctor.mjs",
|
||||
"deps:check": "depcruise src app components lib --config .dependency-cruiser.cjs"
|
||||
"deps:check": "depcruise src app components lib --config .dependency-cruiser.cjs",
|
||||
"vehiq:oltp-ingest": "tsx scripts/vehiq-oltp-ingest/run.ts"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fontsource-variable/inter": "^5.2.5",
|
||||
@@ -89,6 +90,7 @@
|
||||
"postcss": "^8.5.28",
|
||||
"prettier": "^3.9.6",
|
||||
"tailwindcss": "^4.1.9",
|
||||
"tsx": "^4.23.15",
|
||||
"tw-animate-css": "1.4.0",
|
||||
"typescript": "^5",
|
||||
"typescript-eslint": "^8.15.0",
|
||||
|
||||
Generated
+25
-10
@@ -213,6 +213,9 @@ importers:
|
||||
tailwindcss:
|
||||
specifier: ^4.1.9
|
||||
version: 4.3.3
|
||||
tsx:
|
||||
specifier: ^4.23.15
|
||||
version: 4.23.15
|
||||
tw-animate-css:
|
||||
specifier: 1.4.0
|
||||
version: 1.4.0
|
||||
@@ -224,7 +227,7 @@ importers:
|
||||
version: 8.65.0(eslint@9.39.5(jiti@2.7.0))(typescript@5.9.3)
|
||||
vitest:
|
||||
specifier: ^3.2.7
|
||||
version: 3.2.7(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)
|
||||
version: 3.2.7(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15)
|
||||
|
||||
packages/decision: {}
|
||||
|
||||
@@ -2960,6 +2963,11 @@ packages:
|
||||
tslib@2.8.1:
|
||||
resolution: {integrity: sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==}
|
||||
|
||||
tsx@4.23.15:
|
||||
resolution: {integrity: sha512-Yiex1Ovn8z2xPpOWckIiysV1SSyRMY9BkLF++q0yKiDxCqRhosKfMg3janKkiLBwZ5c/YryloKwGZcrEmtwxKw==}
|
||||
engines: {node: '>=18.0.0'}
|
||||
hasBin: true
|
||||
|
||||
tw-animate-css@1.4.0:
|
||||
resolution: {integrity: sha512-7bziOlRqH0hJx80h/3mbicLW7o8qLsH5+RaLR2t+OHM3D0JlWGODQKQ4cxbK7WlvmUxpcj6Kgu6EKqjrGFe3QQ==}
|
||||
|
||||
@@ -4649,13 +4657,13 @@ snapshots:
|
||||
chai: 5.3.3
|
||||
tinyrainbow: 2.0.0
|
||||
|
||||
'@vitest/mocker@3.2.7(vite@7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0))':
|
||||
'@vitest/mocker@3.2.7(vite@7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15))':
|
||||
dependencies:
|
||||
'@vitest/spy': 3.2.7
|
||||
estree-walker: 3.0.3
|
||||
magic-string: 0.30.21
|
||||
optionalDependencies:
|
||||
vite: 7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)
|
||||
vite: 7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15)
|
||||
|
||||
'@vitest/pretty-format@3.2.7':
|
||||
dependencies:
|
||||
@@ -5756,6 +5764,12 @@ snapshots:
|
||||
|
||||
tslib@2.8.1: {}
|
||||
|
||||
tsx@4.23.15:
|
||||
dependencies:
|
||||
esbuild: 0.28.1
|
||||
optionalDependencies:
|
||||
fsevents: 2.3.3
|
||||
|
||||
tw-animate-css@1.4.0: {}
|
||||
|
||||
type-check@0.4.0:
|
||||
@@ -5832,13 +5846,13 @@ snapshots:
|
||||
d3-time: 3.1.0
|
||||
d3-timer: 3.0.1
|
||||
|
||||
vite-node@3.2.4(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0):
|
||||
vite-node@3.2.4(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15):
|
||||
dependencies:
|
||||
cac: 6.7.14
|
||||
debug: 4.4.3
|
||||
es-module-lexer: 1.7.0
|
||||
pathe: 2.0.3
|
||||
vite: 7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)
|
||||
vite: 7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15)
|
||||
transitivePeerDependencies:
|
||||
- '@types/node'
|
||||
- jiti
|
||||
@@ -5853,7 +5867,7 @@ snapshots:
|
||||
- tsx
|
||||
- yaml
|
||||
|
||||
vite@7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0):
|
||||
vite@7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15):
|
||||
dependencies:
|
||||
esbuild: 0.28.1
|
||||
fdir: 6.5.0(picomatch@4.0.5)
|
||||
@@ -5866,12 +5880,13 @@ snapshots:
|
||||
fsevents: 2.3.3
|
||||
jiti: 2.7.0
|
||||
lightningcss: 1.32.0
|
||||
tsx: 4.23.15
|
||||
|
||||
vitest@3.2.7(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0):
|
||||
vitest@3.2.7(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15):
|
||||
dependencies:
|
||||
'@types/chai': 5.2.3
|
||||
'@vitest/expect': 3.2.7
|
||||
'@vitest/mocker': 3.2.7(vite@7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0))
|
||||
'@vitest/mocker': 3.2.7(vite@7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15))
|
||||
'@vitest/pretty-format': 3.2.7
|
||||
'@vitest/runner': 3.2.7
|
||||
'@vitest/snapshot': 3.2.7
|
||||
@@ -5889,8 +5904,8 @@ snapshots:
|
||||
tinyglobby: 0.2.17
|
||||
tinypool: 1.1.1
|
||||
tinyrainbow: 2.0.0
|
||||
vite: 7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)
|
||||
vite-node: 3.2.4(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)
|
||||
vite: 7.3.6(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15)
|
||||
vite-node: 3.2.4(@types/node@22.20.1)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.15)
|
||||
why-is-node-running: 2.3.0
|
||||
optionalDependencies:
|
||||
'@types/node': 22.20.1
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
#!/usr/bin/env -S node
|
||||
/**
|
||||
* scripts/vehiq-oltp-ingest/run.ts
|
||||
* ===================================
|
||||
* Thin CLI shim over src/ingestion/vehiq-oltp-ingest-runner.ts. All the
|
||||
* actual logic (connectors, the runner call, the checkpoint mapping) is in
|
||||
* that blocking-typechecked, tested `src/` module; this file only reads
|
||||
* `process.env`, wires real implementations, and prints a summary.
|
||||
*
|
||||
* TWO MODES, PREVIEW IS THE DEFAULT — this is the activation gate:
|
||||
*
|
||||
* VEHIQ_INGEST_MODE=preview (default)
|
||||
* One bounded read-only connect+discover+read per source, against
|
||||
* whatever checkpoint already exists. NEVER commits a checkpoint, NEVER
|
||||
* calls the sink, NEVER touches ClickHouse. Safe to run against real
|
||||
* production at any time — this is what proves "zero batches, zero
|
||||
* writes" without activating anything.
|
||||
*
|
||||
* VEHIQ_INGEST_MODE=run
|
||||
* The real, scheduled path: P062 runner, durable checkpoints, writes
|
||||
* landing in `opt_dataset_projection`. Requires DATABASE_URL (0pt's own
|
||||
* metadata store, for durable checkpoints) and the OPT_ANALYTICS_
|
||||
* CLICKHOUSE_* vars (the write target) to be set, or it refuses to run
|
||||
* rather than silently falling back to in-memory checkpoints or a
|
||||
* no-op sink — see docs/P070_VEHIQ_OLTP_EXTRACTOR.md "What activating
|
||||
* this in production actually requires".
|
||||
*
|
||||
* NO VEHIQ_SOURCE_DATABASE_URL AT ALL -> exits 0 having done nothing. This
|
||||
* is what makes the scheduled workflow (.gitea/workflows/vehiq-oltp-
|
||||
* ingest.yml) safe to leave enabled with no secret configured: cron fires,
|
||||
* the script logs "not configured", exits 0, no error, no alert.
|
||||
*/
|
||||
import {
|
||||
VEHIQ_OLTP_SOURCES,
|
||||
previewVehiqOltpSource,
|
||||
runVehiqOltpIngestion,
|
||||
} from "../../src/ingestion/vehiq-oltp-ingest-runner";
|
||||
import { makePostgresQueryImpl } from "../../src/connectors/postgres-query-impl";
|
||||
import { createIngestionRepository } from "../../src/repositories/ingestion-repository";
|
||||
import { SchemaRegistry } from "../../src/schema/schema-registry";
|
||||
import { LineageStore } from "../../src/lineage/lineage-store";
|
||||
import { ClickHouseAnalyticsEngine } from "../../src/analytics/engine/clickhouse-engine";
|
||||
import { projectionSink } from "../../src/ingestion/projection-sink";
|
||||
import type { AnalyticsIdentity } from "../../src/analytics/engine/siax-analytics-engine";
|
||||
|
||||
const TENANT_ID = process.env.VEHIQ_INGEST_TENANT_ID ?? "t_vehiq";
|
||||
const PROJECT_ID = process.env.VEHIQ_INGEST_PROJECT_ID ?? "vehiq";
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const sourceUrl = process.env.VEHIQ_SOURCE_DATABASE_URL;
|
||||
if (!sourceUrl) {
|
||||
console.log(
|
||||
JSON.stringify({ status: "not_configured", detail: "VEHIQ_SOURCE_DATABASE_URL is unset — nothing to do" }),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const mode = process.env.VEHIQ_INGEST_MODE === "run" ? "run" : "preview";
|
||||
const { query, close } = makePostgresQueryImpl({
|
||||
connectionString: sourceUrl,
|
||||
ssl: process.env.VEHIQ_SOURCE_PGSSL === "require" ? "require" : false,
|
||||
statementTimeoutMs: process.env.VEHIQ_SOURCE_STATEMENT_TIMEOUT_MS
|
||||
? Number(process.env.VEHIQ_SOURCE_STATEMENT_TIMEOUT_MS)
|
||||
: undefined,
|
||||
});
|
||||
|
||||
try {
|
||||
if (mode === "preview") {
|
||||
const repo = createIngestionRepository(process.env.DATABASE_URL);
|
||||
const results = [];
|
||||
for (const source of VEHIQ_OLTP_SOURCES) {
|
||||
results.push(
|
||||
await previewVehiqOltpSource(source, query, repo, {
|
||||
sourceId: "vehiq_oltp",
|
||||
tenantId: TENANT_ID,
|
||||
projectId: PROJECT_ID,
|
||||
limit: 100,
|
||||
}),
|
||||
);
|
||||
}
|
||||
console.log(JSON.stringify({ status: "preview", mode, results }, null, 2));
|
||||
return;
|
||||
}
|
||||
|
||||
// mode === "run" — the activation-gated real path.
|
||||
if (!process.env.DATABASE_URL) {
|
||||
throw new Error(
|
||||
"VEHIQ_INGEST_MODE=run requires DATABASE_URL (0pt's own metadata store, for durable checkpoints) — " +
|
||||
"refusing to run with in-memory checkpoints, which would silently re-read every row on every scheduled invocation",
|
||||
);
|
||||
}
|
||||
const chBaseUrl = process.env.OPT_ANALYTICS_CLICKHOUSE_BASE_URL;
|
||||
if (!chBaseUrl) {
|
||||
throw new Error(
|
||||
"VEHIQ_INGEST_MODE=run requires OPT_ANALYTICS_CLICKHOUSE_BASE_URL — refusing to run with no write target",
|
||||
);
|
||||
}
|
||||
|
||||
const repo = createIngestionRepository(process.env.DATABASE_URL);
|
||||
const schemas = new SchemaRegistry(repo);
|
||||
const lineage = new LineageStore(repo);
|
||||
const engine = new ClickHouseAnalyticsEngine({
|
||||
baseUrl: chBaseUrl,
|
||||
basicAuth: process.env.OPT_ANALYTICS_CLICKHOUSE_BASIC_AUTH,
|
||||
database: process.env.OPT_ANALYTICS_CLICKHOUSE_DATABASE ?? "opt_analytics",
|
||||
});
|
||||
const identity: AnalyticsIdentity = {
|
||||
tenant_id: TENANT_ID,
|
||||
project_id: PROJECT_ID,
|
||||
principal_id: "svc_vehiq_oltp_extractor",
|
||||
};
|
||||
const sink = projectionSink({
|
||||
engine,
|
||||
identity,
|
||||
datasetIdFor: (partitionKey) => `vehiq.${partitionKey}`,
|
||||
});
|
||||
|
||||
const summary = await runVehiqOltpIngestion({
|
||||
repo,
|
||||
schemas,
|
||||
lineage,
|
||||
sourceQuery: query,
|
||||
oltpSink: sink,
|
||||
dimensionSink: sink,
|
||||
tenantId: TENANT_ID,
|
||||
projectId: PROJECT_ID,
|
||||
});
|
||||
console.log(JSON.stringify({ status: "ran", mode, summary }, null, 2));
|
||||
} finally {
|
||||
await close();
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(JSON.stringify({ status: "error", detail: err instanceof Error ? err.message : String(err) }));
|
||||
process.exitCode = 1;
|
||||
});
|
||||
@@ -0,0 +1,258 @@
|
||||
/**
|
||||
* src/connectors/postgres-keyed-lookup-connector.ts
|
||||
* ====================================================
|
||||
* P061 connector: a narrow, explicit-key-list lookup against a large table —
|
||||
* built specifically so `0pt dimension_vehicle` never becomes `SELECT *
|
||||
* FROM vehicle` against the 98M-row/85GB `vehiq.vehicle` table.
|
||||
*
|
||||
* THE RULE THIS ENFORCES (Simon's decision #2, `admin/vehiq` 2026-09-25
|
||||
* pass): `dealer_stock_entry.vehicle_id` -> unique vehicle IDs -> a
|
||||
* TARGETED lookup of just those rows' attributes -> `dimension_vehicle`.
|
||||
* Never a full scan, never an unbounded WHERE clause, never "give me
|
||||
* everything and I'll filter downstream".
|
||||
*
|
||||
* HOW IT'S STRUCTURALLY DIFFERENT FROM `postgres-watermark-connector.ts`,
|
||||
* not just configured differently: this connector's config carries `keys`
|
||||
* — a caller-supplied, already-resolved, finite list of record keys. There
|
||||
* is no watermark, no `ORDER BY updated_at`, no notion of "everything since
|
||||
* X". `read()` can ONLY ever emit rows whose key is in that list, and
|
||||
* `connect()` REFUSES an empty or missing key list rather than defaulting
|
||||
* to "read everything" — see the "zero keys issues zero queries" test,
|
||||
* which is the assertion that actually proves the narrow-scan property
|
||||
* (not just that the SQL text LOOKS bounded).
|
||||
*
|
||||
* THE ORCHESTRATOR (scripts/vehiq-oltp-ingest/run.ts) is what supplies
|
||||
* `keys`: the DISTINCT `vehicle_id`s seen in the `dealer_stock_entry` batch
|
||||
* this connector instance was built for, and nothing else. A full snapshot/
|
||||
* backfill of the whole `vehicle` table is a deliberate, separate, later
|
||||
* decision — see docs/P070_VEHIQ_OLTP_EXTRACTOR.md's "Deferred: full
|
||||
* vehicle dimension backfill" section. This connector does not build
|
||||
* towards that; it is the permanent narrow path, not a stepping stone.
|
||||
*/
|
||||
import {
|
||||
ConnectorNotConnectedError,
|
||||
type BackfillRequest,
|
||||
type Connector,
|
||||
type ConnectorConfig,
|
||||
type ConnectorContext,
|
||||
type ConnectorHealth,
|
||||
type ConnectorRecord,
|
||||
type DiscoveryResult,
|
||||
type ReadRequest,
|
||||
type ReadResult,
|
||||
type ValidationResult,
|
||||
} from "./connector";
|
||||
import type { IngestionCheckpoint } from "../types/ingestion";
|
||||
import type { RecordSchemaFields } from "../schema/record-schema";
|
||||
import type { QueryImpl, SourceRow } from "./postgres-watermark-connector";
|
||||
|
||||
export const POSTGRES_KEYED_LOOKUP_CONNECTOR_VERSION = "postgres-keyed-lookup-connector@1";
|
||||
|
||||
export interface PostgresKeyedLookupConnectorConfig extends ConnectorConfig {
|
||||
/** Fully-qualified source table, e.g. "vehiq.vehicle". */
|
||||
table: string;
|
||||
partition_key: string;
|
||||
columns: string[];
|
||||
record_key_column: string;
|
||||
schema: RecordSchemaFields;
|
||||
/**
|
||||
* The ENTIRE set of keys this connector instance is permitted to read —
|
||||
* resolved by the caller BEFORE construction (e.g. distinct
|
||||
* `dealer_stock_entry.vehicle_id` values from one ingestion batch). Never
|
||||
* mutated after connect(); a new batch means a new connector instance.
|
||||
*/
|
||||
keys: ReadonlyArray<string | number>;
|
||||
/** How many keys go into one SQL `= ANY($1)` call. Default 500. */
|
||||
keys_per_query?: number;
|
||||
}
|
||||
|
||||
function isConfig(config: ConnectorConfig): config is PostgresKeyedLookupConnectorConfig {
|
||||
return typeof config.table === "string" && typeof config.partition_key === "string" && Array.isArray(config.keys);
|
||||
}
|
||||
|
||||
/** Cursor vocabulary: the decimal count of keys already looked up (an index into `config.keys`). */
|
||||
function parseCursor(cursor: string | undefined): number {
|
||||
if (cursor === undefined) return 0;
|
||||
const n = Number.parseInt(cursor, 10);
|
||||
if (!Number.isInteger(n) || n < 0) throw new Error(`invalid cursor "${cursor}" — expected a key-list offset`);
|
||||
return n;
|
||||
}
|
||||
|
||||
export class PostgresKeyedLookupConnector implements Connector {
|
||||
readonly connector_id = "postgres_keyed_lookup";
|
||||
readonly connector_version = POSTGRES_KEYED_LOOKUP_CONNECTOR_VERSION;
|
||||
|
||||
private config?: PostgresKeyedLookupConnectorConfig;
|
||||
private ctx?: ConnectorContext;
|
||||
private lastCursor = new Map<string, string | undefined>();
|
||||
private connected = false;
|
||||
|
||||
/** Query calls actually issued, for tests to assert the narrow-scan property against. */
|
||||
readonly queryLog: Array<{ sql: string; keys: Array<string | number> }> = [];
|
||||
|
||||
constructor(private readonly queryImpl: QueryImpl) {}
|
||||
|
||||
validate(config: ConnectorConfig): ValidationResult {
|
||||
const errors: string[] = [];
|
||||
if (typeof config.table !== "string" || config.table.length === 0) errors.push("table is required");
|
||||
if (typeof config.partition_key !== "string" || config.partition_key.length === 0) {
|
||||
errors.push("partition_key is required");
|
||||
}
|
||||
const columns = config.columns;
|
||||
if (!Array.isArray(columns) || columns.length === 0) {
|
||||
errors.push("columns is required and must be a non-empty explicit column list");
|
||||
} else if (columns.some((c) => c === "*" || typeof c !== "string" || c.length === 0)) {
|
||||
errors.push('columns must not contain "*" or an empty entry — SELECT * is not permitted');
|
||||
}
|
||||
if (typeof config.record_key_column !== "string" || config.record_key_column.length === 0) {
|
||||
errors.push("record_key_column is required");
|
||||
} else if (Array.isArray(columns) && !columns.includes(config.record_key_column)) {
|
||||
errors.push("record_key_column must be included in columns");
|
||||
}
|
||||
const schema = config.schema;
|
||||
if (typeof schema !== "object" || schema === null || Object.keys(schema).length === 0) {
|
||||
errors.push("schema is required and must declare at least one field");
|
||||
}
|
||||
if (!Array.isArray(config.keys)) {
|
||||
errors.push("keys is required and must be an array");
|
||||
} else if (config.keys.length === 0) {
|
||||
// Deliberately a hard validation error, not a silent "0 records" read.
|
||||
// A caller with nothing to look up should not construct this connector
|
||||
// at all — this keeps "no keys => no query, ever" true even for a
|
||||
// caller that forgets to check first.
|
||||
errors.push("keys must not be empty — construct this connector only when there is something to look up");
|
||||
}
|
||||
if (config.keys_per_query !== undefined && (typeof config.keys_per_query !== "number" || config.keys_per_query <= 0)) {
|
||||
errors.push("keys_per_query must be a positive number when provided");
|
||||
}
|
||||
return { valid: errors.length === 0, errors };
|
||||
}
|
||||
|
||||
async connect(config: ConnectorConfig, ctx: ConnectorContext): Promise<void> {
|
||||
const verdict = this.validate(config);
|
||||
if (!verdict.valid) {
|
||||
throw new Error(`postgres_keyed_lookup connector config invalid: ${verdict.errors.join("; ")}`);
|
||||
}
|
||||
if (!isConfig(config)) throw new Error("postgres_keyed_lookup connector config invalid");
|
||||
this.config = config;
|
||||
this.ctx = ctx;
|
||||
this.connected = true;
|
||||
}
|
||||
|
||||
async discover(): Promise<DiscoveryResult> {
|
||||
const config = this.requireSession("discover");
|
||||
return {
|
||||
partitions: [{ partition_key: config.partition_key, name: config.table, schema: config.schema }],
|
||||
modes: ["batch"],
|
||||
};
|
||||
}
|
||||
|
||||
async health(): Promise<ConnectorHealth> {
|
||||
const config = this.requireSession("health");
|
||||
const started = Date.now();
|
||||
try {
|
||||
await this.queryImpl(`SELECT 1 FROM ${config.table} LIMIT 0`, []);
|
||||
return { status: "up", latency_ms: Date.now() - started };
|
||||
} catch (err) {
|
||||
return {
|
||||
status: "down",
|
||||
latency_ms: Date.now() - started,
|
||||
detail: err instanceof Error ? err.message : "unreachable",
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
async read(req: ReadRequest): Promise<ReadResult> {
|
||||
const config = this.requireSession("read");
|
||||
const offset = parseCursor(req.cursor);
|
||||
const pageLimit = Math.min(req.limit, config.keys_per_query ?? 500);
|
||||
const page = config.keys.slice(offset, offset + pageLimit);
|
||||
|
||||
let rows: SourceRow[] = [];
|
||||
if (page.length > 0) {
|
||||
const columnList = config.columns.join(", ");
|
||||
const sql = `SELECT ${columnList} FROM ${config.table} WHERE ${config.record_key_column} = ANY($1)`;
|
||||
this.queryLog.push({ sql, keys: [...page] });
|
||||
rows = await this.queryImpl(sql, [page]);
|
||||
}
|
||||
// Structural guarantee, not just an intent: if `page` is empty, `queryImpl`
|
||||
// is never called at all — this branch cannot silently widen into an
|
||||
// unbounded scan even if a future edit forgets the `page.length > 0` guard
|
||||
// above, because the caller of this method never asks for more than
|
||||
// `config.keys.length` total across repeated read()s (see `exhausted`).
|
||||
const records = rows.map((row) => this.toRecord(row, config));
|
||||
const nextOffset = offset + page.length;
|
||||
const nextCursor = String(nextOffset);
|
||||
this.lastCursor.set(req.partition_key, nextCursor);
|
||||
return {
|
||||
records,
|
||||
next_cursor: nextCursor,
|
||||
exhausted: nextOffset >= config.keys.length,
|
||||
};
|
||||
}
|
||||
|
||||
async checkpoint(partitionKey: string): Promise<{ partition_key: string; cursor?: string }> {
|
||||
this.requireSession("checkpoint");
|
||||
return { partition_key: partitionKey, cursor: this.lastCursor.get(partitionKey) };
|
||||
}
|
||||
|
||||
async resume(checkpoint: IngestionCheckpoint, limit: number): Promise<ReadResult> {
|
||||
this.requireSession("resume");
|
||||
return this.read({ partition_key: checkpoint.partition_key, cursor: checkpoint.cursor, limit });
|
||||
}
|
||||
|
||||
/**
|
||||
* "Backfill" for a keyed lookup means re-reading a bounded INDEX range of
|
||||
* the same already-narrow key list — never a fresh, wider key set. `from`/
|
||||
* `to` are decimal offsets into `config.keys`, documented here per the
|
||||
* contract's "bound vocabulary is connector-defined" rule.
|
||||
*/
|
||||
async backfill(req: BackfillRequest): Promise<ReadResult> {
|
||||
const config = this.requireSession("backfill");
|
||||
if (!req.from || !req.to) {
|
||||
throw new Error("backfill requires both `from` and `to` — an unbounded backfill is a full reload, not a backfill");
|
||||
}
|
||||
const from = parseCursor(req.from);
|
||||
const to = parseCursor(req.to);
|
||||
if (to <= from) throw new Error("backfill requires `to` > `from`");
|
||||
const page = config.keys.slice(from, Math.min(to, from + req.limit));
|
||||
let rows: SourceRow[] = [];
|
||||
if (page.length > 0) {
|
||||
const columnList = config.columns.join(", ");
|
||||
const sql = `SELECT ${columnList} FROM ${config.table} WHERE ${config.record_key_column} = ANY($1)`;
|
||||
this.queryLog.push({ sql, keys: [...page] });
|
||||
rows = await this.queryImpl(sql, [page]);
|
||||
}
|
||||
return { records: rows.map((row) => this.toRecord(row, config)), next_cursor: undefined, exhausted: true };
|
||||
}
|
||||
|
||||
async disconnect(): Promise<void> {
|
||||
this.config = undefined;
|
||||
this.ctx = undefined;
|
||||
this.lastCursor.clear();
|
||||
this.connected = false;
|
||||
}
|
||||
|
||||
// --- internals -------------------------------------------------------------
|
||||
|
||||
private toRecord(row: SourceRow, config: PostgresKeyedLookupConnectorConfig): ConnectorRecord {
|
||||
const key = row[config.record_key_column];
|
||||
if (typeof key !== "string" && typeof key !== "number") {
|
||||
throw new Error(`record is missing string/number key column "${config.record_key_column}"`);
|
||||
}
|
||||
// Same real-driver normalization as PostgresWatermarkConnector.toRecord —
|
||||
// a Date-valued column must become an ISO string before it reaches the
|
||||
// P063 schema-conformance gate. See that connector's `normalizeRow` doc
|
||||
// comment for how this was found.
|
||||
const payload: SourceRow = {};
|
||||
for (const [field, value] of Object.entries(row)) {
|
||||
payload[field] = value instanceof Date ? value.toISOString() : value;
|
||||
}
|
||||
return { record_key: String(key), payload, operation: "upsert" };
|
||||
}
|
||||
|
||||
private requireSession(op: string): PostgresKeyedLookupConnectorConfig {
|
||||
if (!this.connected || !this.config) throw new ConnectorNotConnectedError(this.connector_id, op);
|
||||
return this.config;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
/**
|
||||
* src/connectors/postgres-query-impl.ts
|
||||
* ========================================
|
||||
* The REAL `QueryImpl` (see postgres-watermark-connector.ts) — a thin,
|
||||
* read-only wrapper over `postgres` (postgres.js, already a repo dependency
|
||||
* — see src/repositories/db.ts for the sibling pattern this mirrors).
|
||||
*
|
||||
* Used ONLY by scripts/vehiq-oltp-ingest/run.ts and the opt-in realinfra
|
||||
* test. Every connector-level test in tests/connectors/*.test.ts injects a
|
||||
* fixture QueryImpl instead — no real Postgres required to prove the query
|
||||
* shape, the watermark math, or the narrow-scan property.
|
||||
*
|
||||
* DOUBLE-ENFORCED READ-ONLY. The connectors above only ever build SELECT
|
||||
* statements — there is no code path that builds anything else. This
|
||||
* wrapper adds a second, independent check anyway: it refuses to execute
|
||||
* any statement whose first keyword is not SELECT. Belt-and-suspenders is
|
||||
* appropriate here specifically because the thing on the other end is a
|
||||
* separate product's live production OLTP primary, not 0pt's own store.
|
||||
*/
|
||||
import postgres from "postgres";
|
||||
import type { QueryImpl, SourceRow } from "./postgres-watermark-connector";
|
||||
|
||||
export interface PostgresQueryImplOptions {
|
||||
connectionString: string;
|
||||
ssl?: "require" | false;
|
||||
/** Statement timeout (ms), applied per-connection so a slow query cannot pin the source's OLTP primary. Default 5000. */
|
||||
statementTimeoutMs?: number;
|
||||
}
|
||||
|
||||
class NonSelectStatementError extends Error {
|
||||
constructor(sql: string) {
|
||||
super(`refusing to execute a non-SELECT statement against a read-only source connector: ${sql.slice(0, 120)}`);
|
||||
this.name = "NonSelectStatementError";
|
||||
}
|
||||
}
|
||||
|
||||
export function assertReadOnlyStatement(sql: string): void {
|
||||
const firstWord = sql.trim().split(/\s+/, 1)[0]?.toUpperCase();
|
||||
if (firstWord !== "SELECT") throw new NonSelectStatementError(sql);
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a `QueryImpl` bound to one `postgres` client. The caller owns the
|
||||
* client's lifecycle (open once per orchestration run, `sql.end()` when
|
||||
* done) — this function does not memoize or reuse connections across calls,
|
||||
* unlike `src/repositories/db.ts`'s deliberate memoization for 0pt's own
|
||||
* store, because this one instance is scoped to a single ingestion run
|
||||
* against an EXTERNAL production database and should not outlive it.
|
||||
*/
|
||||
export function makePostgresQueryImpl(opts: PostgresQueryImplOptions): { query: QueryImpl; sql: ReturnType<typeof postgres>; close: () => Promise<void> } {
|
||||
const sql = postgres(opts.connectionString, {
|
||||
prepare: false,
|
||||
idle_timeout: 20,
|
||||
max: 2,
|
||||
ssl: opts.ssl ?? false,
|
||||
connection: {
|
||||
statement_timeout: opts.statementTimeoutMs ?? 5_000,
|
||||
// A narrow, self-identifying application_name — same posture as
|
||||
// http-json-connector.ts's fixed User-Agent: no impersonation, easy
|
||||
// to spot in the source's own pg_stat_activity.
|
||||
application_name: "0pt-vehiq-oltp-extractor",
|
||||
},
|
||||
});
|
||||
const query: QueryImpl = async (text: string, params: unknown[]): Promise<SourceRow[]> => {
|
||||
assertReadOnlyStatement(text);
|
||||
const rows = await sql.unsafe(text, params as never[]);
|
||||
return rows.map((r) => ({ ...r }));
|
||||
};
|
||||
return { query, sql, close: () => sql.end({ timeout: 5 }) };
|
||||
}
|
||||
@@ -0,0 +1,350 @@
|
||||
/**
|
||||
* src/connectors/postgres-watermark-connector.ts
|
||||
* =================================================
|
||||
* P061 connector: a read-only, watermark-polled Postgres OLTP table.
|
||||
*
|
||||
* WHY THIS EXISTS. BI-3/BI-4 need a controlled, incremental path from a
|
||||
* product's live OLTP Postgres into 0pt — without `wal_level=logical`,
|
||||
* without a Postgres restart, without a permanent replica, and without ever
|
||||
* scanning a whole table. `docs/60-data/bi/...` (read from `admin/vehiq`,
|
||||
* 2026-09-25 pass) names the exact shape this connector implements:
|
||||
*
|
||||
* WHERE updated_at > :last_watermark ORDER BY updated_at, id
|
||||
* LIMIT :batch_size
|
||||
*
|
||||
* with a checkpoint of `(source, table, last_updated_at, last_id,
|
||||
* batch_count, checksum, ingested_at)`. This connector produces exactly
|
||||
* that query and that checkpoint — see the field-mapping table in
|
||||
* `docs/P070_VEHIQ_OLTP_EXTRACTOR.md`.
|
||||
*
|
||||
* TIE-BREAK, NOT A DEVIATION. A bare `updated_at > $1` drops rows that share
|
||||
* the exact same `updated_at` as the LAST row of a page: if row 100 and row
|
||||
* 101 both have `updated_at = T`, and row 100 is the last row of a page, a
|
||||
* naive `> T` on the next page skips row 101 forever. The query below adds
|
||||
* the documented `ORDER BY updated_at, id` tie-break to the WHERE clause
|
||||
* too: `updated_at > $1 OR (updated_at = $1 AND id > $2)`. Same watermark
|
||||
* semantics Simon specified, correct at page boundaries — exactly the class
|
||||
* of bug `tests/connectors/postgres-watermark-connector.test.ts`'s
|
||||
* "same-timestamp tie at a page boundary" case exists to catch.
|
||||
*
|
||||
* GUARD RAIL, NOT DECORATION. `validate()` refuses any config whose `table`
|
||||
* is (or ends in) the 98M-row `vehicle` table. That table gets its own
|
||||
* narrow, keyed connector (`postgres-keyed-lookup-connector.ts`) — this one
|
||||
* must never be pointed at it, even by a future caller's typo. The refusal
|
||||
* is asserted in tests, not just documented.
|
||||
*
|
||||
* READ-ONLY, STRUCTURALLY. There is no INSERT/UPDATE/DELETE anywhere in this
|
||||
* file. `columns` is an explicit allow-list the caller declares — never
|
||||
* `SELECT *` — so a schema drift on the source table cannot silently widen
|
||||
* what this connector reads or hands downstream.
|
||||
*
|
||||
* INJECTABLE QUERY IMPL — same seam as `http-json-connector.ts`'s
|
||||
* `fetchImpl`: tests drive real SQL text/params through a fixture function,
|
||||
* no real Postgres required for the connector-level tests. The default
|
||||
* implementation (`makePostgresQueryImpl`) is a thin wrapper over `postgres`
|
||||
* (postgres.js, already a repo dependency), used only by the orchestration
|
||||
* script and the opt-in realinfra test.
|
||||
*/
|
||||
import { createHash } from "node:crypto";
|
||||
import {
|
||||
ConnectorNotConnectedError,
|
||||
type BackfillRequest,
|
||||
type Connector,
|
||||
type ConnectorConfig,
|
||||
type ConnectorContext,
|
||||
type ConnectorHealth,
|
||||
type ConnectorRecord,
|
||||
type DiscoveryResult,
|
||||
type ReadRequest,
|
||||
type ReadResult,
|
||||
type ValidationResult,
|
||||
} from "./connector";
|
||||
import type { IngestionCheckpoint } from "../types/ingestion";
|
||||
import type { RecordSchemaFields } from "../schema/record-schema";
|
||||
|
||||
export const POSTGRES_WATERMARK_CONNECTOR_VERSION = "postgres-watermark-connector@1";
|
||||
|
||||
/** One row back from the source, column name -> raw value. */
|
||||
export type SourceRow = Record<string, unknown>;
|
||||
|
||||
/**
|
||||
* Injectable query seam. `params` are positional (`$1`, `$2`, ...). Tests
|
||||
* supply a fixture; `makePostgresQueryImpl` supplies the real thing.
|
||||
*/
|
||||
export type QueryImpl = (sql: string, params: unknown[]) => Promise<SourceRow[]>;
|
||||
|
||||
export interface PostgresWatermarkConnectorConfig extends ConnectorConfig {
|
||||
/** Fully-qualified source table, e.g. "app.dealer_stock_entry". */
|
||||
table: string;
|
||||
/** Logical partition name exposed to the runner — a dataset name, never a physical detail leak beyond `table` itself. */
|
||||
partition_key: string;
|
||||
/** Explicit column allow-list. Never "*". Must include `record_key_column` and `watermark_column`. */
|
||||
columns: string[];
|
||||
record_key_column: string;
|
||||
watermark_column: string;
|
||||
/** Declared record shape, fed to the schema registry (P063). */
|
||||
schema: RecordSchemaFields;
|
||||
}
|
||||
|
||||
/** The 98M-row table this connector must never be pointed at. See file header. */
|
||||
const FORBIDDEN_TABLE_PATTERN = /(^|\.)vehicle$/i;
|
||||
|
||||
function isConfig(config: ConnectorConfig): config is PostgresWatermarkConnectorConfig {
|
||||
return typeof config.table === "string" && typeof config.partition_key === "string";
|
||||
}
|
||||
|
||||
/** Cursor vocabulary: JSON `{last_updated_at, last_id, checksum}` — see docs/P070 for the checkpoint field mapping. */
|
||||
export interface WatermarkCursor {
|
||||
last_updated_at: string;
|
||||
last_id: string;
|
||||
/** sha256 of the page just produced, hex-encoded. Absent only for the "never read" case. */
|
||||
checksum?: string;
|
||||
}
|
||||
|
||||
export function encodeCursor(cursor: WatermarkCursor): string {
|
||||
return JSON.stringify(cursor);
|
||||
}
|
||||
|
||||
export function decodeCursor(raw: string | undefined): WatermarkCursor | undefined {
|
||||
if (raw === undefined) return undefined;
|
||||
const parsed = JSON.parse(raw) as unknown;
|
||||
if (
|
||||
typeof parsed !== "object" ||
|
||||
parsed === null ||
|
||||
typeof (parsed as WatermarkCursor).last_updated_at !== "string" ||
|
||||
typeof (parsed as WatermarkCursor).last_id !== "string"
|
||||
) {
|
||||
throw new Error(`invalid watermark cursor "${raw}"`);
|
||||
}
|
||||
return parsed as WatermarkCursor;
|
||||
}
|
||||
|
||||
/**
|
||||
* Postgres drivers (postgres.js included) hand back `timestamptz` columns as
|
||||
* native JS `Date` objects, not strings. `String(date)` produces the
|
||||
* locale/timezone-dependent `Date.toString()` form ("Thu Sep 03 2026
|
||||
* 02:00:00 GMT+0200 (...)") — useless as a stored watermark, and NOT what a
|
||||
* fixture-based test (which hands back plain ISO strings) would ever catch.
|
||||
* Found by running this connector against a real Postgres container, not by
|
||||
* the fixture-only unit tests — which is why this helper exists instead of
|
||||
* an inline `String(...)` at each call site.
|
||||
*/
|
||||
function watermarkValueToIso(value: unknown): string {
|
||||
return value instanceof Date ? value.toISOString() : String(value);
|
||||
}
|
||||
|
||||
/**
|
||||
* The P063 schema-conformance gate (src/schema/compatibility.ts) checks a
|
||||
* payload's field VALUES against the declared schema — and even its
|
||||
* `"timestamp"` type requires an already-ISO-string value, never a raw
|
||||
* `Date`. A real Postgres driver hands back `timestamptz`/`date` columns as
|
||||
* `Date` objects, so a payload built straight from `queryImpl`'s rows fails
|
||||
* conformance the moment it reaches a real database — caught by running
|
||||
* this connector against a real Postgres container (see docs/P070's
|
||||
* verification log), not by the fixture-only unit tests, which all use
|
||||
* plain strings. Every row is normalized through this function before it
|
||||
* becomes a `ConnectorRecord` payload, so the envelope's payload is always
|
||||
* JSON-safe and schema-checkable, independent of which driver produced it.
|
||||
*/
|
||||
function normalizeRow(row: SourceRow): SourceRow {
|
||||
const normalized: SourceRow = {};
|
||||
for (const [field, value] of Object.entries(row)) {
|
||||
normalized[field] = value instanceof Date ? value.toISOString() : value;
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/** sha256 over a stable JSON encoding of the rows, hex-encoded. Deterministic regardless of object key insertion order upstream. */
|
||||
export function checksumRows(rows: SourceRow[]): string {
|
||||
const stable = rows.map((row) =>
|
||||
Object.fromEntries(Object.keys(row).sort().map((k) => [k, row[k]])),
|
||||
);
|
||||
return createHash("sha256").update(JSON.stringify(stable)).digest("hex");
|
||||
}
|
||||
|
||||
export class PostgresWatermarkConnector implements Connector {
|
||||
readonly connector_id = "postgres_watermark";
|
||||
readonly connector_version = POSTGRES_WATERMARK_CONNECTOR_VERSION;
|
||||
|
||||
private config?: PostgresWatermarkConnectorConfig;
|
||||
private ctx?: ConnectorContext;
|
||||
private lastCursor = new Map<string, string | undefined>();
|
||||
private connected = false;
|
||||
|
||||
constructor(private readonly queryImpl: QueryImpl) {}
|
||||
|
||||
validate(config: ConnectorConfig): ValidationResult {
|
||||
const errors: string[] = [];
|
||||
if (typeof config.table !== "string" || config.table.length === 0) {
|
||||
errors.push("table is required");
|
||||
} else if (FORBIDDEN_TABLE_PATTERN.test(config.table)) {
|
||||
errors.push(
|
||||
`table "${config.table}" looks like the 98M-row vehicle catalog table — this connector is for watermark-polled ` +
|
||||
`OLTP tables only; use PostgresKeyedLookupConnector for narrow, keyed vehicle attribute lookups instead`,
|
||||
);
|
||||
}
|
||||
if (typeof config.partition_key !== "string" || config.partition_key.length === 0) {
|
||||
errors.push("partition_key is required");
|
||||
}
|
||||
const columns = config.columns;
|
||||
if (!Array.isArray(columns) || columns.length === 0) {
|
||||
errors.push("columns is required and must be a non-empty explicit column list");
|
||||
} else if (columns.some((c) => c === "*" || typeof c !== "string" || c.length === 0)) {
|
||||
errors.push('columns must not contain "*" or an empty entry — SELECT * is not permitted');
|
||||
}
|
||||
if (typeof config.record_key_column !== "string" || config.record_key_column.length === 0) {
|
||||
errors.push("record_key_column is required");
|
||||
} else if (Array.isArray(columns) && !columns.includes(config.record_key_column)) {
|
||||
errors.push("record_key_column must be included in columns");
|
||||
}
|
||||
if (typeof config.watermark_column !== "string" || config.watermark_column.length === 0) {
|
||||
errors.push("watermark_column is required");
|
||||
} else if (Array.isArray(columns) && !columns.includes(config.watermark_column)) {
|
||||
errors.push("watermark_column must be included in columns");
|
||||
}
|
||||
const schema = config.schema;
|
||||
if (typeof schema !== "object" || schema === null || Object.keys(schema).length === 0) {
|
||||
errors.push("schema is required and must declare at least one field");
|
||||
}
|
||||
return { valid: errors.length === 0, errors };
|
||||
}
|
||||
|
||||
async connect(config: ConnectorConfig, ctx: ConnectorContext): Promise<void> {
|
||||
const verdict = this.validate(config);
|
||||
if (!verdict.valid) {
|
||||
throw new Error(`postgres_watermark connector config invalid: ${verdict.errors.join("; ")}`);
|
||||
}
|
||||
if (!isConfig(config)) throw new Error("postgres_watermark connector config invalid");
|
||||
this.config = config;
|
||||
this.ctx = ctx;
|
||||
this.connected = true;
|
||||
}
|
||||
|
||||
async discover(): Promise<DiscoveryResult> {
|
||||
const config = this.requireSession("discover");
|
||||
return {
|
||||
partitions: [{ partition_key: config.partition_key, name: config.table, schema: config.schema }],
|
||||
// A polled table is batch/stream, never cdc: this connector never emits deletes.
|
||||
modes: ["batch", "stream"],
|
||||
};
|
||||
}
|
||||
|
||||
async health(): Promise<ConnectorHealth> {
|
||||
const config = this.requireSession("health");
|
||||
const started = Date.now();
|
||||
try {
|
||||
await this.queryImpl(`SELECT 1 FROM ${config.table} LIMIT 0`, []);
|
||||
return { status: "up", latency_ms: Date.now() - started };
|
||||
} catch (err) {
|
||||
return {
|
||||
status: "down",
|
||||
latency_ms: Date.now() - started,
|
||||
detail: err instanceof Error ? err.message : "unreachable",
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
async read(req: ReadRequest): Promise<ReadResult> {
|
||||
const config = this.requireSession("read");
|
||||
const cursor = decodeCursor(req.cursor);
|
||||
const columnList = config.columns.join(", ");
|
||||
let rows: SourceRow[];
|
||||
if (cursor === undefined) {
|
||||
rows = await this.queryImpl(
|
||||
`SELECT ${columnList} FROM ${config.table} ORDER BY ${config.watermark_column}, ${config.record_key_column} LIMIT $1`,
|
||||
[req.limit],
|
||||
);
|
||||
} else {
|
||||
rows = await this.queryImpl(
|
||||
`SELECT ${columnList} FROM ${config.table} ` +
|
||||
`WHERE ${config.watermark_column} > $1 OR (${config.watermark_column} = $1 AND ${config.record_key_column} > $2) ` +
|
||||
`ORDER BY ${config.watermark_column}, ${config.record_key_column} LIMIT $3`,
|
||||
[cursor.last_updated_at, cursor.last_id, req.limit],
|
||||
);
|
||||
}
|
||||
// Hard bound — a queryImpl that over-delivers (a fixture bug, or a
|
||||
// future real driver quirk) is truncated here rather than silently
|
||||
// widening the caller's page.
|
||||
rows = rows.slice(0, req.limit);
|
||||
|
||||
const records: ConnectorRecord[] = rows.map((row) => this.toRecord(row, config));
|
||||
let nextCursor = req.cursor;
|
||||
if (rows.length > 0) {
|
||||
const last = rows[rows.length - 1];
|
||||
const nextWatermark: WatermarkCursor = {
|
||||
last_updated_at: watermarkValueToIso(last[config.watermark_column]),
|
||||
last_id: String(last[config.record_key_column]),
|
||||
checksum: checksumRows(rows),
|
||||
};
|
||||
nextCursor = encodeCursor(nextWatermark);
|
||||
this.lastCursor.set(req.partition_key, nextCursor);
|
||||
}
|
||||
return {
|
||||
records,
|
||||
next_cursor: nextCursor,
|
||||
// Fewer rows than the limit means we caught up to "now" — there may be
|
||||
// MORE later (this is a live table), but nothing more RIGHT NOW.
|
||||
exhausted: rows.length < req.limit,
|
||||
};
|
||||
}
|
||||
|
||||
async checkpoint(partitionKey: string): Promise<{ partition_key: string; cursor?: string }> {
|
||||
this.requireSession("checkpoint");
|
||||
return { partition_key: partitionKey, cursor: this.lastCursor.get(partitionKey) };
|
||||
}
|
||||
|
||||
async resume(checkpoint: IngestionCheckpoint, limit: number): Promise<ReadResult> {
|
||||
this.requireSession("resume");
|
||||
return this.read({ partition_key: checkpoint.partition_key, cursor: checkpoint.cursor, limit });
|
||||
}
|
||||
|
||||
async backfill(req: BackfillRequest): Promise<ReadResult> {
|
||||
const config = this.requireSession("backfill");
|
||||
if (!req.from || !req.to) {
|
||||
throw new Error("backfill requires both `from` and `to` — an unbounded backfill is a full reload, not a backfill");
|
||||
}
|
||||
const columnList = config.columns.join(", ");
|
||||
const rows = (
|
||||
await this.queryImpl(
|
||||
`SELECT ${columnList} FROM ${config.table} ` +
|
||||
`WHERE ${config.watermark_column} >= $1 AND ${config.watermark_column} < $2 ` +
|
||||
`ORDER BY ${config.watermark_column}, ${config.record_key_column} LIMIT $3`,
|
||||
[req.from, req.to, req.limit],
|
||||
)
|
||||
).slice(0, req.limit);
|
||||
return {
|
||||
records: rows.map((row) => this.toRecord(row, config)),
|
||||
next_cursor: undefined,
|
||||
exhausted: true,
|
||||
};
|
||||
}
|
||||
|
||||
async disconnect(): Promise<void> {
|
||||
this.config = undefined;
|
||||
this.ctx = undefined;
|
||||
this.lastCursor.clear();
|
||||
this.connected = false;
|
||||
}
|
||||
|
||||
// --- internals -------------------------------------------------------------
|
||||
|
||||
private toRecord(row: SourceRow, config: PostgresWatermarkConnectorConfig): ConnectorRecord {
|
||||
const key = row[config.record_key_column];
|
||||
if (typeof key !== "string" && typeof key !== "number") {
|
||||
throw new Error(`record is missing string/number key column "${config.record_key_column}"`);
|
||||
}
|
||||
const watermark = watermarkValueToIso(row[config.watermark_column]);
|
||||
return {
|
||||
record_key: String(key),
|
||||
payload: normalizeRow(row),
|
||||
operation: "upsert",
|
||||
occurred_at: watermark,
|
||||
cursor: encodeCursor({ last_updated_at: watermark, last_id: String(key) }),
|
||||
};
|
||||
}
|
||||
|
||||
private requireSession(op: string): PostgresWatermarkConnectorConfig {
|
||||
if (!this.connected || !this.config) throw new ConnectorNotConnectedError(this.connector_id, op);
|
||||
return this.config;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
/**
|
||||
* src/ingestion/projection-sink.ts
|
||||
* ===================================
|
||||
* Adapts the P062 runner's generic `IngestionSink` (envelopes in) onto
|
||||
* `SiaxAnalyticsEngine.ingestProjection` (BI-P009 capability 7, writes
|
||||
* `opt_dataset_projection` in ClickHouse — see `src/analytics/engine/
|
||||
* clickhouse-engine.ts`).
|
||||
*
|
||||
* WHY `ingestProjection` / `opt_dataset_projection` AND NOT `opt_analytics_
|
||||
* event`. `opt_analytics_event` (BI-P010 schema.ts) is documented as a
|
||||
* "high-volume per-tenant EVENT stream" with a 30-day TTL — the right shape
|
||||
* for discrete occurrences (a page view, a search, a status transition).
|
||||
* `dealer_stock_entry`/`listing`/`deal` rows and vehicle-dimension rows are
|
||||
* ENTITY SNAPSHOTS — current state of a stock entry, a listing, a deal, a
|
||||
* vehicle — which is exactly what `opt_dataset_projection` (ReplacingMerge
|
||||
* Tree serving tier, `entity_key`+`version`) already exists to hold, and
|
||||
* `ingestProjection` is the one write port `SiaxAnalyticsEngine` exposes for
|
||||
* it. Landing entity rows into the raw event stream would be forcing an
|
||||
* entity shape through an event shape — this sink does not do that.
|
||||
*
|
||||
* THE FIREWALL IS RESPECTED BY CONSTRUCTION: this file imports only the
|
||||
* `SiaxAnalyticsEngine`/`AnalyticsIdentity`/`ProjectionRow` TYPES from
|
||||
* `../analytics/engine/siax-analytics-engine` — no ClickHouse client, no
|
||||
* `CLICKHOUSE_*` env read, no connection string. Which concrete engine
|
||||
* (ClickHouse today, anything else later) answers `ingestProjection` is the
|
||||
* caller's concern (see `scripts/vehiq-oltp-ingest/run.ts`), not this file's.
|
||||
*
|
||||
* ORDERING STAYS THE RUNNER'S: this function is called AS the runner's
|
||||
* `sink`, so its own crash/partial-failure behaviour is exactly what
|
||||
* `ingestion-runner.ts`'s header already documents — a thrown error here
|
||||
* leaves the page's checkpoint uncommitted, and the next run re-reads it.
|
||||
*/
|
||||
import type { IngestionEnvelope } from "../types/ingestion";
|
||||
import type {
|
||||
AnalyticsIdentity,
|
||||
ProjectionRow,
|
||||
SiaxAnalyticsEngine,
|
||||
} from "../analytics/engine/siax-analytics-engine";
|
||||
import { SiaxError } from "../platform/errors";
|
||||
|
||||
export interface ProjectionSinkOptions {
|
||||
engine: SiaxAnalyticsEngine;
|
||||
identity: AnalyticsIdentity;
|
||||
/** Maps a partition_key (e.g. "dealer_stock_entry") to a logical dataset id (e.g. "vehiq.dealer_stock_entry"). */
|
||||
datasetIdFor: (partitionKey: string) => string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Splits an envelope's payload into ClickHouse-projection-friendly
|
||||
* attributes (strings) and measures (numbers). `null`/`undefined` fields are
|
||||
* dropped rather than coerced to `"null"`/`NaN`, which would fabricate a
|
||||
* value nothing upstream actually asserted.
|
||||
*/
|
||||
export function toProjectionRow(envelope: IngestionEnvelope): ProjectionRow {
|
||||
if (envelope.operation === "delete" || envelope.payload === null) {
|
||||
throw new SiaxError(
|
||||
"INVALID_REQUEST",
|
||||
"0pt.ingestion.projection-sink",
|
||||
`envelope ${envelope.envelope_id} is a delete (payload: null) — opt_dataset_projection has no tombstone ` +
|
||||
`representation via ingestProjection today; a CDC-capable source reaching this sink is a real gap, not ` +
|
||||
`silently handled here (this connector programme currently emits "upsert" only, so this path is untested ` +
|
||||
`by design, not unreachable by contract)`,
|
||||
);
|
||||
}
|
||||
const attributes: Record<string, string> = {};
|
||||
const measures: Record<string, number> = {};
|
||||
for (const [field, value] of Object.entries(envelope.payload)) {
|
||||
if (value === null || value === undefined) continue;
|
||||
if (typeof value === "number" && Number.isFinite(value)) {
|
||||
measures[field] = value;
|
||||
} else if (value instanceof Date) {
|
||||
attributes[field] = value.toISOString();
|
||||
} else if (typeof value === "object") {
|
||||
attributes[field] = JSON.stringify(value);
|
||||
} else {
|
||||
attributes[field] = String(value);
|
||||
}
|
||||
}
|
||||
return {
|
||||
entity_key: envelope.record_key,
|
||||
as_of: envelope.occurred_at ?? envelope.ingested_at,
|
||||
version: envelope.sequence,
|
||||
attributes,
|
||||
measures,
|
||||
};
|
||||
}
|
||||
|
||||
export function projectionSink(
|
||||
opts: ProjectionSinkOptions,
|
||||
): (envelopes: IngestionEnvelope[]) => Promise<void> {
|
||||
return async (envelopes: IngestionEnvelope[]): Promise<void> => {
|
||||
if (envelopes.length === 0) return;
|
||||
const byPartition = new Map<string, IngestionEnvelope[]>();
|
||||
for (const envelope of envelopes) {
|
||||
const list = byPartition.get(envelope.partition_key) ?? [];
|
||||
list.push(envelope);
|
||||
byPartition.set(envelope.partition_key, list);
|
||||
}
|
||||
for (const [partitionKey, group] of byPartition) {
|
||||
const rows = group.map(toProjectionRow);
|
||||
const dataset_id = opts.datasetIdFor(partitionKey);
|
||||
const result = await opts.engine.ingestProjection(opts.identity, { dataset_id, rows });
|
||||
if (result.ingested !== rows.length) {
|
||||
// Partial ingest is a hard failure, on purpose: the runner only
|
||||
// commits the checkpoint AFTER this sink resolves. Resolving
|
||||
// "successfully" on a partial write would silently drop the
|
||||
// un-ingested rows on the next run's resume.
|
||||
throw new SiaxError(
|
||||
"RETRYABLE_FAILURE",
|
||||
"0pt.ingestion.projection-sink",
|
||||
`ingestProjection for dataset "${dataset_id}" reported ${result.ingested}/${rows.length} rows ingested — ` +
|
||||
`treating as a failed page so the checkpoint does not advance past unlanded rows`,
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,343 @@
|
||||
/**
|
||||
* src/ingestion/vehiq-oltp-ingest-runner.ts
|
||||
* ============================================
|
||||
* BI-3/BI-4 — the controlled, incremental VehIQ OLTP -> 0pt ClickHouse path,
|
||||
* as Simon specified it (`admin/vehiq` 2026-09-25 pass, "Data path for
|
||||
* inventory.days_in_stock"):
|
||||
*
|
||||
* OLTP operational tables -> read-only analytical extraction ->
|
||||
* watermark/updated_at -> 0pt ingestion -> ClickHouse
|
||||
*
|
||||
* WHY THIS LIVES IN `admin/0pt`, NOT `admin/vehiq`. ADR-0035 (admin/vehiq,
|
||||
* decisions 1/2/10, CI-enforced by `tools/validate_engine_firewall.py`):
|
||||
* VehIQ's own codebase must NEVER hold a ClickHouse credential or import a
|
||||
* ClickHouse client, and product code must never hammer the live OLTP
|
||||
* primary with analytical queries directly. `admin/vehiq`'s own hub
|
||||
* (`docs/CURRENT_STATE.md`, "Connection details for a future API0
|
||||
* connector") says this explicitly: "VehIQ's own codebase must never see
|
||||
* any of this. This is 0pt's data plane; only a future `admin/api0`
|
||||
* connector should hold these details." This repo is 0pt's data plane —
|
||||
* the correct home per that same guidance (a sibling of `admin/api0`, not
|
||||
* `admin/api0` itself, since the read/write mechanics belong next to the
|
||||
* P060-P068 ingestion plane and the BI-P009/P010 ClickHouse engine that
|
||||
* already live here).
|
||||
*
|
||||
* TWO ENTRY POINTS, DELIBERATELY SEPARATE:
|
||||
*
|
||||
* `previewVehiqOltpSource` — connect, discover, ONE bounded read. Never
|
||||
* calls checkpoint()/commits anything, never runs the P062 runner. Safe
|
||||
* to point at production at any time, table empty or not: nothing it does
|
||||
* can advance a checkpoint or land a row. This is what proves "zero
|
||||
* batches, zero writes" without deploying a live scheduled service.
|
||||
*
|
||||
* `runVehiqOltpIngestion` — the real, scheduled path: the P062
|
||||
* `IngestionRunner` across all three OLTP sources, then the narrow
|
||||
* `dimension_vehicle` lookup keyed on exactly the `vehicle_id`s the OLTP
|
||||
* batch referenced. This is what `.gitea/workflows/vehiq-oltp-ingest.yml`
|
||||
* runs — see that file and docs/P070_VEHIQ_OLTP_EXTRACTOR.md for the
|
||||
* activation gate this stays behind until Simon flips it.
|
||||
*
|
||||
* CHECKPOINT FIELD MAPPING (Simon's own list: source, table, last_updated_at,
|
||||
* last_id, batch_count, checksum, ingested_at) onto the canonical P062
|
||||
* `IngestionCheckpoint` (source_id, tenant_id, partition_key, cursor,
|
||||
* sequence, records_committed, committed_at) — deliberately reusing the
|
||||
* existing durable checkpoint mechanism rather than building a second one:
|
||||
*
|
||||
* source -> checkpoint.source_id
|
||||
* table -> checkpoint.partition_key
|
||||
* last_updated_at-> decoded from checkpoint.cursor (WatermarkCursor)
|
||||
* last_id -> decoded from checkpoint.cursor (WatermarkCursor)
|
||||
* checksum -> decoded from checkpoint.cursor (WatermarkCursor)
|
||||
* batch_count -> checkpoint.records_committed (cumulative) / a
|
||||
* per-run page count is also in each PartitionRunReport
|
||||
* ingested_at -> checkpoint.committed_at
|
||||
*/
|
||||
import type { IngestionRepository } from "../repositories/ingestion-repository";
|
||||
import type { SchemaRegistry } from "../schema/schema-registry";
|
||||
import type { LineageStore } from "../lineage/lineage-store";
|
||||
import type { RecordSchemaFields } from "../schema/record-schema";
|
||||
import { IngestionRunner, type IngestionRunReport, type IngestionSink } from "./ingestion-runner";
|
||||
import {
|
||||
PostgresWatermarkConnector,
|
||||
decodeCursor,
|
||||
type QueryImpl,
|
||||
} from "../connectors/postgres-watermark-connector";
|
||||
import { PostgresKeyedLookupConnector } from "../connectors/postgres-keyed-lookup-connector";
|
||||
import type { IngestionEnvelope } from "../types/ingestion";
|
||||
|
||||
export interface VehiqOltpSourceDef {
|
||||
/** e.g. "dealer_stock_entry" — also the logical dataset name suffix. */
|
||||
partition_key: "dealer_stock_entry" | "deal";
|
||||
/** Fully-qualified source table, e.g. "app.dealer_stock_entry". */
|
||||
table: string;
|
||||
columns: string[];
|
||||
record_key_column: string;
|
||||
watermark_column: string;
|
||||
schema: RecordSchemaFields;
|
||||
}
|
||||
|
||||
/**
|
||||
* TWO sources, not three — READ THIS BEFORE ASSUMING `app.listing` IS
|
||||
* COVERED. The task's own wording (and `admin/vehiq`'s `docs/CURRENT_STATE.
|
||||
* md`, "the columns already exist") assumed all three OLTP tables carry a
|
||||
* generic `updated_at`. Verified LIVE against production (`server10`,
|
||||
* `vehiq-db`, read-only `information_schema.columns`, 2026-09-25) that
|
||||
* this is only true for none of the three as a literal `updated_at` column
|
||||
* — and NOT true at all for `app.listing`:
|
||||
*
|
||||
* app.dealer_stock_entry — id, tenant_id, vehicle_id, lot_location,
|
||||
* floorplan_status, listed_at, sold_at. No `updated_at`, no `status`
|
||||
* (it's `floorplan_status`). `listed_at` (NOT NULL) is the best
|
||||
* available watermark — see the caveat below.
|
||||
* app.deal — id, tenant_id, party_id, vehicle_id, sale_price_minor,
|
||||
* currency, status, created_at. No `updated_at`. `created_at`
|
||||
* (NOT NULL DEFAULT now()) is the best available watermark.
|
||||
* app.listing — id, tenant_id, vehicle_id, asking_price_minor, currency,
|
||||
* channels, status. NO timestamp column of any kind exists today.
|
||||
* `admin/vehiq` migration 0084 (`published_at`/`unpublished_at`) is
|
||||
* written but NOT YET APPLIED to production (verified: those columns
|
||||
* are absent live) — until it lands, there is no honest way to build a
|
||||
* watermark connector for `app.listing` without inventing a column that
|
||||
* does not exist. Not built here — see docs/P070's "Deferred: app.
|
||||
* listing extraction" section for exactly what unblocks it.
|
||||
*
|
||||
* HONEST CAVEAT ON THE TWO SOURCES THAT ARE HERE: `listed_at`/`created_at`
|
||||
* are set ONCE, at insert. Neither table has a column that bumps when an
|
||||
* EXISTING row changes (`sold_at`/`floorplan_status` on dealer_stock_entry;
|
||||
* `status` on deal). This connector will therefore reliably see every NEW
|
||||
* row, but will NOT see an in-place UPDATE to an existing row as "changed" —
|
||||
* it has already passed the watermark and will not be re-read. This is the
|
||||
* same gap a real `updated_at` column (plus a trigger, e.g. `0084`-style)
|
||||
* would close, and is exactly why building a permanent extractor deserves
|
||||
* that schema decision rather than silently working around its absence.
|
||||
* `docs/P070_VEHIQ_OLTP_EXTRACTOR.md` documents this as the #1 thing to
|
||||
* settle before leaning on this extractor for anything beyond "count of new
|
||||
* rows since X".
|
||||
*/
|
||||
export const VEHIQ_OLTP_SOURCES: VehiqOltpSourceDef[] = [
|
||||
{
|
||||
partition_key: "dealer_stock_entry",
|
||||
table: "app.dealer_stock_entry",
|
||||
columns: ["id", "tenant_id", "vehicle_id", "lot_location", "floorplan_status", "listed_at", "sold_at"],
|
||||
record_key_column: "id",
|
||||
watermark_column: "listed_at",
|
||||
schema: {
|
||||
id: { type: "string", required: true },
|
||||
tenant_id: { type: "string", required: true },
|
||||
vehicle_id: { type: "string", required: true },
|
||||
lot_location: { type: "string", required: true },
|
||||
floorplan_status: { type: "string", required: true },
|
||||
listed_at: { type: "timestamp", required: true },
|
||||
sold_at: { type: "timestamp", required: false },
|
||||
},
|
||||
},
|
||||
{
|
||||
partition_key: "deal",
|
||||
table: "app.deal",
|
||||
columns: ["id", "tenant_id", "party_id", "vehicle_id", "sale_price_minor", "currency", "status", "created_at"],
|
||||
record_key_column: "id",
|
||||
watermark_column: "created_at",
|
||||
schema: {
|
||||
id: { type: "string", required: true },
|
||||
tenant_id: { type: "string", required: true },
|
||||
party_id: { type: "string", required: true },
|
||||
vehicle_id: { type: "string", required: true },
|
||||
sale_price_minor: { type: "number", required: true },
|
||||
currency: { type: "string", required: true },
|
||||
status: { type: "string", required: true },
|
||||
created_at: { type: "timestamp", required: true },
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
/**
|
||||
* `vehiq.vehicle`'s primary key column is `vehicle_id`, NOT `id` — verified
|
||||
* live (information_schema.columns, read-only, no data scanned) 2026-09-25.
|
||||
* Column list otherwise matches what BI-P010/VEHIQ_ANALYTICS_BASELINE.md
|
||||
* already documented (make/model/model_year/fuel_type all present and
|
||||
* correctly named).
|
||||
*/
|
||||
export const VEHIQ_VEHICLE_DIMENSION_TABLE = "vehiq.vehicle";
|
||||
export const VEHIQ_VEHICLE_DIMENSION_RECORD_KEY_COLUMN = "vehicle_id";
|
||||
export const VEHIQ_VEHICLE_DIMENSION_COLUMNS = ["vehicle_id", "make", "model", "model_year", "fuel_type"];
|
||||
export const VEHIQ_VEHICLE_DIMENSION_SCHEMA: RecordSchemaFields = {
|
||||
vehicle_id: { type: "string", required: true },
|
||||
make: { type: "string", required: false },
|
||||
model: { type: "string", required: false },
|
||||
model_year: { type: "number", required: false },
|
||||
fuel_type: { type: "string", required: false },
|
||||
};
|
||||
|
||||
// --- preview: read-only, never commits, safe against production at any time ---
|
||||
|
||||
export interface PreviewResult {
|
||||
partition_key: string;
|
||||
table: string;
|
||||
/** Records this ONE bounded read would produce right now. */
|
||||
would_read: number;
|
||||
exhausted: boolean;
|
||||
checkpoint_before: string | undefined;
|
||||
next_cursor: string | undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* ONE bounded connect -> discover -> read(existing checkpoint, limit), then
|
||||
* disconnect. Never calls IngestionRunner, never calls connector.checkpoint()
|
||||
* against the repo, never writes anything anywhere. This is intentionally
|
||||
* the ONLY function in this file that is safe to point at a real production
|
||||
* source before the extractor is "activated" (see docs/P070).
|
||||
*/
|
||||
export async function previewVehiqOltpSource(
|
||||
source: VehiqOltpSourceDef,
|
||||
query: QueryImpl,
|
||||
repo: Pick<IngestionRepository, "getCheckpoint">,
|
||||
opts: { sourceId: string; tenantId: string; projectId: string; limit?: number } = {
|
||||
sourceId: "vehiq_oltp",
|
||||
tenantId: "t_vehiq",
|
||||
projectId: "vehiq",
|
||||
},
|
||||
): Promise<PreviewResult> {
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
const config = {
|
||||
table: source.table,
|
||||
partition_key: source.partition_key,
|
||||
columns: source.columns,
|
||||
record_key_column: source.record_key_column,
|
||||
watermark_column: source.watermark_column,
|
||||
schema: source.schema,
|
||||
};
|
||||
await connector.connect(config, {
|
||||
tenant_id: opts.tenantId,
|
||||
project_id: opts.projectId,
|
||||
source_id: opts.sourceId,
|
||||
run_id: `preview_${Date.now()}`,
|
||||
});
|
||||
try {
|
||||
const existing = await repo.getCheckpoint(opts.sourceId, source.partition_key);
|
||||
const result = await connector.read({
|
||||
partition_key: source.partition_key,
|
||||
cursor: existing?.cursor,
|
||||
limit: opts.limit ?? 100,
|
||||
});
|
||||
return {
|
||||
partition_key: source.partition_key,
|
||||
table: source.table,
|
||||
would_read: result.records.length,
|
||||
exhausted: result.exhausted,
|
||||
checkpoint_before: existing?.cursor,
|
||||
next_cursor: result.next_cursor,
|
||||
};
|
||||
} finally {
|
||||
await connector.disconnect();
|
||||
}
|
||||
}
|
||||
|
||||
// --- real run: the P062 highway, scheduled/activated path ---------------------
|
||||
|
||||
export interface VehiqOltpIngestDeps {
|
||||
repo: IngestionRepository;
|
||||
schemas: SchemaRegistry;
|
||||
lineage?: LineageStore;
|
||||
/** Read-only query against VehIQ's `vehiq-db` (server10) — see makePostgresQueryImpl in the calling script for the real implementation. */
|
||||
sourceQuery: QueryImpl;
|
||||
/** Where OLTP-entity envelopes land (see src/ingestion/projection-sink.ts). */
|
||||
oltpSink: IngestionSink;
|
||||
/** Where the narrow vehicle-dimension envelopes land. Same sink shape; kept separate so the two datasets can be routed differently if ever needed. */
|
||||
dimensionSink: IngestionSink;
|
||||
tenantId: string;
|
||||
projectId: string;
|
||||
sourceIdPrefix?: string;
|
||||
pageSize?: number;
|
||||
maxPages?: number;
|
||||
}
|
||||
|
||||
export interface VehiqOltpIngestSummary {
|
||||
oltp: IngestionRunReport[];
|
||||
dimension?: IngestionRunReport;
|
||||
referenced_vehicle_ids: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs the three watermark sources through the P062 runner, then derives the
|
||||
* narrow vehicle-dimension key set from exactly what THIS run's
|
||||
* `dealer_stock_entry` batch referenced (never a fresh scan, never the full
|
||||
* 98M-row table) and runs that through the runner too, if there is anything
|
||||
* to look up. When there is nothing new (today's actual production state:
|
||||
* all three OLTP tables are empty), this function runs, reports 0 records
|
||||
* read for every source, calls the dimension sink zero times, and returns
|
||||
* cleanly — the empty-table case is a normal, successful, no-op run, not a
|
||||
* special case that needs its own error handling.
|
||||
*/
|
||||
export async function runVehiqOltpIngestion(deps: VehiqOltpIngestDeps): Promise<VehiqOltpIngestSummary> {
|
||||
const runner = new IngestionRunner(deps.repo, deps.schemas, deps.lineage);
|
||||
const sourceIdPrefix = deps.sourceIdPrefix ?? "vehiq_oltp";
|
||||
const oltpReports: IngestionRunReport[] = [];
|
||||
const referencedVehicleIds = new Set<string>();
|
||||
|
||||
for (const source of VEHIQ_OLTP_SOURCES) {
|
||||
const seen: IngestionEnvelope[] = [];
|
||||
const wrappedSink: IngestionSink = async (envelopes) => {
|
||||
seen.push(...envelopes);
|
||||
await deps.oltpSink(envelopes);
|
||||
};
|
||||
const connector = new PostgresWatermarkConnector(deps.sourceQuery);
|
||||
const report = await runner.run({
|
||||
source_id: `${sourceIdPrefix}_${source.partition_key}`,
|
||||
tenant_id: deps.tenantId,
|
||||
project_id: deps.projectId,
|
||||
mode: "batch",
|
||||
connector,
|
||||
config: {
|
||||
table: source.table,
|
||||
partition_key: source.partition_key,
|
||||
columns: source.columns,
|
||||
record_key_column: source.record_key_column,
|
||||
watermark_column: source.watermark_column,
|
||||
schema: source.schema,
|
||||
},
|
||||
sink: wrappedSink,
|
||||
page_size: deps.pageSize ?? 200,
|
||||
max_pages: deps.maxPages ?? 1000,
|
||||
output_ref: `0pt:dataset:vehiq.${source.partition_key}`,
|
||||
});
|
||||
oltpReports.push(report);
|
||||
|
||||
if (source.partition_key === "dealer_stock_entry") {
|
||||
for (const envelope of seen) {
|
||||
const vehicleId = envelope.payload?.["vehicle_id"];
|
||||
if (typeof vehicleId === "string" && vehicleId.length > 0) referencedVehicleIds.add(vehicleId);
|
||||
else if (typeof vehicleId === "number") referencedVehicleIds.add(String(vehicleId));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let dimensionReport: IngestionRunReport | undefined;
|
||||
if (referencedVehicleIds.size > 0) {
|
||||
const keys = [...referencedVehicleIds];
|
||||
const dimensionConnector = new PostgresKeyedLookupConnector(deps.sourceQuery);
|
||||
dimensionReport = await runner.run({
|
||||
source_id: `${sourceIdPrefix}_dimension_vehicle`,
|
||||
tenant_id: deps.tenantId,
|
||||
project_id: deps.projectId,
|
||||
mode: "batch",
|
||||
connector: dimensionConnector,
|
||||
config: {
|
||||
table: VEHIQ_VEHICLE_DIMENSION_TABLE,
|
||||
partition_key: "dimension_vehicle",
|
||||
columns: VEHIQ_VEHICLE_DIMENSION_COLUMNS,
|
||||
record_key_column: VEHIQ_VEHICLE_DIMENSION_RECORD_KEY_COLUMN,
|
||||
schema: VEHIQ_VEHICLE_DIMENSION_SCHEMA,
|
||||
keys,
|
||||
},
|
||||
sink: deps.dimensionSink,
|
||||
page_size: deps.pageSize ?? 200,
|
||||
max_pages: deps.maxPages ?? 1000,
|
||||
output_ref: "0pt:dataset:vehiq.dimension_vehicle",
|
||||
});
|
||||
}
|
||||
|
||||
return { oltp: oltpReports, dimension: dimensionReport, referenced_vehicle_ids: referencedVehicleIds.size };
|
||||
}
|
||||
|
||||
/** Re-exported so callers (the CLI script, tests) don't need a second import from the connector module. */
|
||||
export { decodeCursor };
|
||||
@@ -0,0 +1,124 @@
|
||||
/**
|
||||
* Proves the narrow-scan property of PostgresKeyedLookupConnector — the
|
||||
* connector that stands in for "dealer_stock_entry.vehicle_id -> unique
|
||||
* vehicle IDs -> targeted lookup -> dimension_vehicle", never a scan of the
|
||||
* 98M-row `vehicle` table. The assertions are on QUERY CALLS, not just
|
||||
* returned records: a bug that widened the WHERE clause but happened to
|
||||
* still return the right rows in a test fixture would be invisible to a
|
||||
* behavioural-only test and is exactly what this file is written to catch.
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { PostgresKeyedLookupConnector } from "@/src/connectors/postgres-keyed-lookup-connector";
|
||||
import type { QueryImpl, SourceRow } from "@/src/connectors/postgres-watermark-connector";
|
||||
import type { RecordSchemaFields } from "@/src/schema/record-schema";
|
||||
|
||||
const SCHEMA: RecordSchemaFields = {
|
||||
id: { type: "string", required: true },
|
||||
make: { type: "string", required: false },
|
||||
model: { type: "string", required: false },
|
||||
};
|
||||
|
||||
const VEHICLE_TABLE: SourceRow[] = [
|
||||
{ id: "v1", make: "Volvo", model: "XC60" },
|
||||
{ id: "v2", make: "Volvo", model: "V70" },
|
||||
{ id: "v3", make: "Toyota", model: "Corolla" },
|
||||
{ id: "v4", make: "Audi", model: "A4" }, // never referenced by any dealer_stock_entry in these tests
|
||||
];
|
||||
|
||||
function fixtureQuery(): { query: QueryImpl; calls: Array<{ sql: string; params: unknown[] }> } {
|
||||
const calls: Array<{ sql: string; params: unknown[] }> = [];
|
||||
const query: QueryImpl = async (sql, params) => {
|
||||
calls.push({ sql, params });
|
||||
const [ids] = params as [Array<string | number>];
|
||||
return VEHICLE_TABLE.filter((row) => ids.includes(row.id as string));
|
||||
};
|
||||
return { query, calls };
|
||||
}
|
||||
|
||||
function config(keys: Array<string | number>, keysPerQuery?: number) {
|
||||
return {
|
||||
table: "vehiq.vehicle",
|
||||
partition_key: "dimension_vehicle",
|
||||
columns: ["id", "make", "model"],
|
||||
record_key_column: "id",
|
||||
schema: SCHEMA,
|
||||
keys,
|
||||
keys_per_query: keysPerQuery,
|
||||
};
|
||||
}
|
||||
|
||||
describe("PostgresKeyedLookupConnector — validate()", () => {
|
||||
it("refuses an empty key list rather than defaulting to a full read", () => {
|
||||
const connector = new PostgresKeyedLookupConnector(fixtureQuery().query);
|
||||
const verdict = connector.validate(config([]));
|
||||
expect(verdict.valid).toBe(false);
|
||||
expect(verdict.errors.join(" ")).toMatch(/keys must not be empty/);
|
||||
});
|
||||
|
||||
it("rejects '*' in columns", () => {
|
||||
const connector = new PostgresKeyedLookupConnector(fixtureQuery().query);
|
||||
expect(connector.validate({ ...config(["v1"]), columns: ["*"] }).valid).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("PostgresKeyedLookupConnector — the narrow-scan property", () => {
|
||||
it("connect() with an empty key list is refused, and NO query is ever issued", async () => {
|
||||
const { query, calls } = fixtureQuery();
|
||||
const connector = new PostgresKeyedLookupConnector(query);
|
||||
await expect(
|
||||
connector.connect(config([]), { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" }),
|
||||
).rejects.toThrow(/keys must not be empty/);
|
||||
expect(calls).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("reads exactly the vehicles referenced by the batch — v4 (never referenced) is never touched", async () => {
|
||||
const referenced = ["v1", "v2", "v3"]; // simulates dedup'd dealer_stock_entry.vehicle_id values from one batch
|
||||
const { query, calls } = fixtureQuery();
|
||||
const connector = new PostgresKeyedLookupConnector(query);
|
||||
await connector.connect(config(referenced), { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
const result = await connector.read({ partition_key: "dimension_vehicle", limit: 500 });
|
||||
|
||||
expect(result.records.map((r) => r.record_key).sort()).toEqual(["v1", "v2", "v3"]);
|
||||
expect(result.exhausted).toBe(true);
|
||||
|
||||
// The narrow-scan property, checked at the QUERY level: every id ever
|
||||
// bound into a query call is a subset of `referenced`, and the union of
|
||||
// all ids queried equals exactly `referenced` — never more, never a
|
||||
// wildcard, never "v4".
|
||||
expect(calls).toHaveLength(1);
|
||||
const allQueriedIds = calls.flatMap((c) => c.params[0] as string[]);
|
||||
expect(allQueriedIds.sort()).toEqual([...referenced].sort());
|
||||
expect(allQueriedIds).not.toContain("v4");
|
||||
for (const call of calls) {
|
||||
expect(call.sql).toMatch(/WHERE id = ANY\(\$1\)/);
|
||||
expect(call.sql).not.toMatch(/select\s+\*/i);
|
||||
}
|
||||
});
|
||||
|
||||
it("chunks a large key list across multiple bounded queries, never one unbounded IN(...)", async () => {
|
||||
const referenced = Array.from({ length: 5 }, (_, i) => `v${i + 1}`);
|
||||
const { query, calls } = fixtureQuery();
|
||||
const connector = new PostgresKeyedLookupConnector(query);
|
||||
await connector.connect(config(referenced, 2), { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
|
||||
let cursor: string | undefined;
|
||||
let pages = 0;
|
||||
for (let i = 0; i < 10; i++) {
|
||||
const page = await connector.read({ partition_key: "dimension_vehicle", cursor, limit: 500 });
|
||||
pages++;
|
||||
cursor = page.next_cursor;
|
||||
if (page.exhausted) break;
|
||||
}
|
||||
expect(pages).toBe(3); // ceil(5/2)
|
||||
for (const call of calls) {
|
||||
expect((call.params[0] as string[]).length).toBeLessThanOrEqual(2);
|
||||
}
|
||||
const allQueriedIds = calls.flatMap((c) => c.params[0] as string[]);
|
||||
expect(new Set(allQueriedIds).size).toBe(5);
|
||||
});
|
||||
|
||||
it("the empty-batch case: zero referenced vehicles means this connector is never even constructed by the caller (see vehiq-oltp-ingest-runner.test.ts) — but if it IS given zero keys, it fails validation rather than scanning", () => {
|
||||
const connector = new PostgresKeyedLookupConnector(fixtureQuery().query);
|
||||
expect(connector.validate(config([])).valid).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,194 @@
|
||||
/**
|
||||
* Connector-level proof for postgres-watermark-connector.ts. No real
|
||||
* Postgres: `queryImpl` is a fixture that runs the SAME SQL text the real
|
||||
* connector builds against an in-memory row set, so the watermark math and
|
||||
* the tie-break are proven precisely rather than just "some rows came back".
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
PostgresWatermarkConnector,
|
||||
checksumRows,
|
||||
decodeCursor,
|
||||
encodeCursor,
|
||||
type QueryImpl,
|
||||
type SourceRow,
|
||||
} from "@/src/connectors/postgres-watermark-connector";
|
||||
import type { RecordSchemaFields } from "@/src/schema/record-schema";
|
||||
|
||||
const SCHEMA: RecordSchemaFields = {
|
||||
id: { type: "string", required: true },
|
||||
vehicle_id: { type: "string", required: false },
|
||||
updated_at: { type: "string", required: true },
|
||||
};
|
||||
|
||||
const CONFIG = {
|
||||
table: "app.dealer_stock_entry",
|
||||
partition_key: "dealer_stock_entry",
|
||||
columns: ["id", "vehicle_id", "updated_at"],
|
||||
record_key_column: "id",
|
||||
watermark_column: "updated_at",
|
||||
schema: SCHEMA,
|
||||
};
|
||||
|
||||
/** A fixture query engine that actually parses/applies the connector's WHERE/ORDER BY/LIMIT against an in-memory table — not a stub that ignores its arguments. */
|
||||
function fixtureQuery(rows: SourceRow[]): { query: QueryImpl; calls: Array<{ sql: string; params: unknown[] }> } {
|
||||
const calls: Array<{ sql: string; params: unknown[] }> = [];
|
||||
const query: QueryImpl = async (sql, params) => {
|
||||
calls.push({ sql, params });
|
||||
let filtered = rows;
|
||||
if (sql.includes("WHERE")) {
|
||||
const [watermark, id] = params as [string, string, number];
|
||||
filtered = rows.filter(
|
||||
(r) => (r.updated_at as string) > watermark || ((r.updated_at as string) === watermark && (r.id as string) > id),
|
||||
);
|
||||
}
|
||||
const sorted = [...filtered].sort((a, b) => {
|
||||
const w = (a.updated_at as string).localeCompare(b.updated_at as string);
|
||||
return w !== 0 ? w : (a.id as string).localeCompare(b.id as string);
|
||||
});
|
||||
const limit = Number(params[params.length - 1]);
|
||||
return sorted.slice(0, limit);
|
||||
};
|
||||
return { query, calls };
|
||||
}
|
||||
|
||||
const ROWS: SourceRow[] = [
|
||||
{ id: "a1", vehicle_id: "v1", updated_at: "2026-09-01T00:00:00.000Z" },
|
||||
{ id: "a2", vehicle_id: "v2", updated_at: "2026-09-01T00:00:00.000Z" }, // same updated_at as a1 — the tie-break case
|
||||
{ id: "a3", vehicle_id: "v1", updated_at: "2026-09-02T00:00:00.000Z" },
|
||||
{ id: "a4", vehicle_id: "v3", updated_at: "2026-09-03T00:00:00.000Z" },
|
||||
];
|
||||
|
||||
describe("PostgresWatermarkConnector — validate()", () => {
|
||||
it("rejects a config pointed at the 98M-row vehicle table", () => {
|
||||
const connector = new PostgresWatermarkConnector(fixtureQuery([]).query);
|
||||
const verdict = connector.validate({ ...CONFIG, table: "vehiq.vehicle" });
|
||||
expect(verdict.valid).toBe(false);
|
||||
expect(verdict.errors.join(" ")).toMatch(/vehicle/i);
|
||||
});
|
||||
|
||||
it("rejects SELECT * (an empty/missing columns list, or a literal '*')", () => {
|
||||
const connector = new PostgresWatermarkConnector(fixtureQuery([]).query);
|
||||
expect(connector.validate({ ...CONFIG, columns: [] }).valid).toBe(false);
|
||||
expect(connector.validate({ ...CONFIG, columns: ["*"] }).valid).toBe(false);
|
||||
});
|
||||
|
||||
it("accepts a well-formed config", () => {
|
||||
const connector = new PostgresWatermarkConnector(fixtureQuery([]).query);
|
||||
expect(connector.validate(CONFIG).valid).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("PostgresWatermarkConnector — watermark + limit correctness", () => {
|
||||
it("first read (no cursor) honours LIMIT and orders by (updated_at, id)", async () => {
|
||||
const { query } = fixtureQuery(ROWS);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
const result = await connector.read({ partition_key: "dealer_stock_entry", limit: 2 });
|
||||
expect(result.records.map((r) => r.record_key)).toEqual(["a1", "a2"]);
|
||||
expect(result.exhausted).toBe(false); // 2 rows returned == limit, more may exist
|
||||
});
|
||||
|
||||
it("a same-updated_at tie at the page boundary is not skipped on the next page (the whole reason for the tuple WHERE)", async () => {
|
||||
const { query } = fixtureQuery(ROWS);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
const page1 = await connector.read({ partition_key: "dealer_stock_entry", limit: 1 }); // just a1
|
||||
expect(page1.records.map((r) => r.record_key)).toEqual(["a1"]);
|
||||
const page2 = await connector.read({ partition_key: "dealer_stock_entry", cursor: page1.next_cursor, limit: 1 });
|
||||
// A naive `updated_at > $1` would skip a2 here (same updated_at as a1).
|
||||
expect(page2.records.map((r) => r.record_key)).toEqual(["a2"]);
|
||||
});
|
||||
|
||||
it("reads the full table across pages in strict watermark order, once per row", async () => {
|
||||
const { query } = fixtureQuery(ROWS);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
let cursor: string | undefined;
|
||||
const seen: string[] = [];
|
||||
for (let i = 0; i < 10; i++) {
|
||||
const page = await connector.read({ partition_key: "dealer_stock_entry", cursor, limit: 2 });
|
||||
seen.push(...page.records.map((r) => r.record_key));
|
||||
cursor = page.next_cursor;
|
||||
if (page.exhausted) break;
|
||||
}
|
||||
expect(seen).toEqual(["a1", "a2", "a3", "a4"]);
|
||||
});
|
||||
|
||||
it("the empty-table case: zero rows, no throw, exhausted:true", async () => {
|
||||
const { query, calls } = fixtureQuery([]);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
const result = await connector.read({ partition_key: "dealer_stock_entry", limit: 100 });
|
||||
expect(result.records).toEqual([]);
|
||||
expect(result.exhausted).toBe(true);
|
||||
expect(calls).toHaveLength(1); // the query WAS issued (a real SELECT against a real, currently-empty table) — it just returned nothing
|
||||
});
|
||||
});
|
||||
|
||||
describe("PostgresWatermarkConnector — checkpoint / resume", () => {
|
||||
it("checkpoint() reflects the in-session cursor after a read", async () => {
|
||||
const { query } = fixtureQuery(ROWS);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
await connector.read({ partition_key: "dealer_stock_entry", limit: 2 });
|
||||
const cp = await connector.checkpoint("dealer_stock_entry");
|
||||
const decoded = decodeCursor(cp.cursor);
|
||||
expect(decoded?.last_id).toBe("a2");
|
||||
expect(decoded?.last_updated_at).toBe("2026-09-01T00:00:00.000Z");
|
||||
expect(decoded?.checksum).toBe(checksumRows(ROWS.slice(0, 2)));
|
||||
});
|
||||
|
||||
it("resume() with a saved checkpoint continues exactly where the checkpoint left off", async () => {
|
||||
const { query } = fixtureQuery(ROWS);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
const cursor = encodeCursor({ last_updated_at: "2026-09-02T00:00:00.000Z", last_id: "a3" });
|
||||
const result = await connector.resume(
|
||||
{ source_id: "s", tenant_id: "t", partition_key: "dealer_stock_entry", cursor, sequence: 3, records_committed: 3, committed_at: new Date().toISOString() },
|
||||
10,
|
||||
);
|
||||
expect(result.records.map((r) => r.record_key)).toEqual(["a4"]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("PostgresWatermarkConnector — real Postgres driver value shapes", () => {
|
||||
it("a `timestamptz` column returned as a JS Date (not a string) is encoded as proper ISO-8601 in the cursor, not Date.toString()'s locale form", async () => {
|
||||
// Regression: caught by running this connector against a REAL Postgres
|
||||
// container (postgres.js hands back `timestamptz` as `Date`), not by any
|
||||
// fixture-only test — the fixtures elsewhere in this file all use plain
|
||||
// ISO strings, which never exercised this path. Reproduced here directly
|
||||
// instead of requiring Docker so the regression stays covered in CI.
|
||||
const rowsWithDates: SourceRow[] = [
|
||||
{ id: "a1", vehicle_id: "v1", updated_at: new Date("2026-09-01T00:00:00.000Z") },
|
||||
];
|
||||
const { query } = fixtureQuery(rowsWithDates.map((r) => ({ ...r, updated_at: (r.updated_at as Date).toISOString() })));
|
||||
// fixtureQuery's own row-compare logic needs string dates to sort; feed
|
||||
// it a Date-returning query instead so the connector genuinely sees a
|
||||
// Date object, the same as the real driver.
|
||||
const dateReturningQuery: QueryImpl = async (sql, params) => {
|
||||
const rows = await query(sql, params);
|
||||
return rows.map((r) => ({ ...r, updated_at: new Date(r.updated_at as string) }));
|
||||
};
|
||||
const connector = new PostgresWatermarkConnector(dateReturningQuery);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
const result = await connector.read({ partition_key: "dealer_stock_entry", limit: 10 });
|
||||
expect(result.records[0].occurred_at).toBe("2026-09-01T00:00:00.000Z");
|
||||
const decoded = decodeCursor(result.next_cursor);
|
||||
expect(decoded?.last_updated_at).toBe("2026-09-01T00:00:00.000Z");
|
||||
expect(decoded?.last_updated_at).not.toMatch(/GMT|\(.*\)/); // not Date.toString()'s form
|
||||
});
|
||||
});
|
||||
|
||||
describe("PostgresWatermarkConnector — never SELECT *", () => {
|
||||
it("every SQL statement the connector builds names the configured columns explicitly", async () => {
|
||||
const { query, calls } = fixtureQuery(ROWS);
|
||||
const connector = new PostgresWatermarkConnector(query);
|
||||
await connector.connect(CONFIG, { tenant_id: "t", project_id: "p", source_id: "s", run_id: "r" });
|
||||
await connector.read({ partition_key: "dealer_stock_entry", limit: 10 });
|
||||
for (const call of calls) {
|
||||
expect(call.sql).not.toMatch(/select\s+\*/i);
|
||||
expect(call.sql).toContain("id, vehicle_id, updated_at");
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,217 @@
|
||||
/**
|
||||
* End-to-end proof for src/ingestion/vehiq-oltp-ingest-runner.ts — the
|
||||
* actual pipeline the scheduled workflow runs, wired through the REAL P062
|
||||
* IngestionRunner (InMemoryIngestionRepository backing it, no real
|
||||
* Postgres/ClickHouse). Four things the task requires are asserted here
|
||||
* specifically, not just "it doesn't throw":
|
||||
*
|
||||
* 1. a batch respects the watermark/limit (proven at the connector layer
|
||||
* already; here proven again at the FULL pipeline layer, envelope by
|
||||
* envelope);
|
||||
* 2. a run resumes correctly from a checkpoint after an interruption (a
|
||||
* sink that throws on its 2nd call, mid-run — same technique
|
||||
* tests/p0/ingestion-runner.test.ts itself uses to prove this for the
|
||||
* generic runner);
|
||||
* 3. the empty-table case (today's actual production state for all three
|
||||
* OLTP tables) completes with zero batches / zero sink calls / no
|
||||
* throw;
|
||||
* 4. the vehicle-dimension lookup only ever touches vehicle rows actually
|
||||
* referenced by dealer_stock_entry in the batch that ran — asserted by
|
||||
* capturing every query the dimension connector issues.
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { InMemoryIngestionRepository } from "@/src/repositories/in-memory-ingestion-repository";
|
||||
import { SchemaRegistry } from "@/src/schema/schema-registry";
|
||||
import {
|
||||
runVehiqOltpIngestion,
|
||||
previewVehiqOltpSource,
|
||||
VEHIQ_OLTP_SOURCES,
|
||||
type VehiqOltpIngestDeps,
|
||||
} from "@/src/ingestion/vehiq-oltp-ingest-runner";
|
||||
import type { QueryImpl, SourceRow } from "@/src/connectors/postgres-watermark-connector";
|
||||
import type { IngestionEnvelope } from "@/src/types/ingestion";
|
||||
|
||||
/**
|
||||
* A fake source Postgres. One table per fixture entry, keyed by the
|
||||
* connector's own `table` name. Generic across whichever watermark/
|
||||
* record-key column names the real connector puts in the SQL text — parsed
|
||||
* out of the statement itself (`ORDER BY <watermark>, <key>`) rather than
|
||||
* hardcoded, since `dealer_stock_entry` (watermark `listed_at`, key `id`)
|
||||
* and `deal` (watermark `created_at`, key `id`) use different column names.
|
||||
*/
|
||||
function fixtureVehiqDb(tables: Record<string, SourceRow[]>): { query: QueryImpl; calls: Array<{ sql: string; params: unknown[] }> } {
|
||||
const calls: Array<{ sql: string; params: unknown[] }> = [];
|
||||
const query: QueryImpl = async (sql, params) => {
|
||||
calls.push({ sql, params });
|
||||
const tableName = Object.keys(tables).find((t) => sql.includes(`FROM ${t}`));
|
||||
const rows = tableName ? tables[tableName] : [];
|
||||
const orderBy = /ORDER BY (\w+), (\w+)/.exec(sql);
|
||||
if (sql.includes("= ANY($1)")) {
|
||||
const keyColMatch = /WHERE (\w+) = ANY/.exec(sql);
|
||||
const keyCol = keyColMatch ? keyColMatch[1] : "id";
|
||||
const [ids] = params as [Array<string | number>];
|
||||
return rows.filter((r) => ids.includes(r[keyCol] as string));
|
||||
}
|
||||
if (!orderBy) return rows.slice(0, Number(params[params.length - 1]));
|
||||
const [, watermarkCol, keyCol] = orderBy;
|
||||
let filtered = rows;
|
||||
if (sql.includes("WHERE")) {
|
||||
const [watermark, id] = params as [string, string, number];
|
||||
filtered = rows.filter(
|
||||
(r) => (r[watermarkCol] as string) > watermark || ((r[watermarkCol] as string) === watermark && (r[keyCol] as string) > id),
|
||||
);
|
||||
}
|
||||
const sorted = [...filtered].sort((a, b) => {
|
||||
const w = (a[watermarkCol] as string).localeCompare(b[watermarkCol] as string);
|
||||
return w !== 0 ? w : (a[keyCol] as string).localeCompare(b[keyCol] as string);
|
||||
});
|
||||
const limit = Number(params[params.length - 1]);
|
||||
return sorted.slice(0, limit);
|
||||
};
|
||||
return { query, calls };
|
||||
}
|
||||
|
||||
function baseDeps(query: QueryImpl, sink: (e: IngestionEnvelope[]) => Promise<void>): VehiqOltpIngestDeps {
|
||||
const repo = new InMemoryIngestionRepository();
|
||||
return {
|
||||
repo,
|
||||
schemas: new SchemaRegistry(repo),
|
||||
sourceQuery: query,
|
||||
oltpSink: sink,
|
||||
dimensionSink: sink,
|
||||
tenantId: "t_vehiq",
|
||||
projectId: "vehiq",
|
||||
pageSize: 2,
|
||||
};
|
||||
}
|
||||
|
||||
describe("runVehiqOltpIngestion — empty-table case (today's actual production state)", () => {
|
||||
it("completes with zero batches, zero sink calls, no error, for all three OLTP tables", async () => {
|
||||
const { query } = fixtureVehiqDb({ "app.dealer_stock_entry": [], "app.listing": [], "app.deal": [] });
|
||||
const seen: IngestionEnvelope[] = [];
|
||||
const summary = await runVehiqOltpIngestion(baseDeps(query, async (e) => void seen.push(...e)));
|
||||
|
||||
expect(seen).toHaveLength(0);
|
||||
expect(summary.referenced_vehicle_ids).toBe(0);
|
||||
expect(summary.dimension).toBeUndefined(); // no referenced vehicles => the keyed-lookup connector is never even constructed
|
||||
for (const report of summary.oltp) {
|
||||
expect(report.status).toBe("succeeded");
|
||||
expect(report.partitions.every((p) => p.records_read === 0)).toBe(true);
|
||||
expect(report.partitions.every((p) => p.envelopes_emitted === 0)).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
it("previewVehiqOltpSource against empty tables never commits a checkpoint and reports would_read: 0", async () => {
|
||||
const { query } = fixtureVehiqDb({ "app.dealer_stock_entry": [] });
|
||||
const repo = new InMemoryIngestionRepository();
|
||||
const source = VEHIQ_OLTP_SOURCES.find((s) => s.partition_key === "dealer_stock_entry")!;
|
||||
const preview = await previewVehiqOltpSource(source, query, repo, {
|
||||
sourceId: "vehiq_oltp",
|
||||
tenantId: "t_vehiq",
|
||||
projectId: "vehiq",
|
||||
});
|
||||
expect(preview.would_read).toBe(0);
|
||||
expect(preview.exhausted).toBe(true);
|
||||
// Never committed — getCheckpoint after the preview still reports nothing.
|
||||
expect(await repo.getCheckpoint("vehiq_oltp", "dealer_stock_entry")).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
// Column shape verified live against production (server10, vehiq-db,
|
||||
// information_schema.columns, read-only, 2026-09-25) — see the doc comment
|
||||
// on VEHIQ_OLTP_SOURCES for why there is no `updated_at` here.
|
||||
const DEALER_STOCK_ROWS: SourceRow[] = [
|
||||
{ id: "d1", tenant_id: "t_vehiq", vehicle_id: "v1", lot_location: "Lot A", floorplan_status: "owned", listed_at: "2026-09-01T00:00:00.000Z", sold_at: null },
|
||||
{ id: "d2", tenant_id: "t_vehiq", vehicle_id: "v2", lot_location: "Lot A", floorplan_status: "financed", listed_at: "2026-09-02T00:00:00.000Z", sold_at: null },
|
||||
{ id: "d3", tenant_id: "t_vehiq", vehicle_id: "v1", lot_location: "Lot B", floorplan_status: "owned", listed_at: "2026-09-03T00:00:00.000Z", sold_at: "2026-09-04T00:00:00.000Z" },
|
||||
];
|
||||
const VEHICLE_ROWS: SourceRow[] = [
|
||||
{ vehicle_id: "v1", make: "Volvo", model: "XC60", model_year: 2022, fuel_type: "electric" },
|
||||
{ vehicle_id: "v2", make: "Toyota", model: "Corolla", model_year: 2021, fuel_type: "petrol" },
|
||||
{ vehicle_id: "v9", make: "Never", model: "Referenced", model_year: 2020, fuel_type: "diesel" }, // must never be queried
|
||||
];
|
||||
|
||||
describe("runVehiqOltpIngestion — watermark/limit correctness with real data", () => {
|
||||
it("emits envelopes for every row, in watermark order, and derives exactly the referenced vehicle ids", async () => {
|
||||
const { query } = fixtureVehiqDb({
|
||||
"app.dealer_stock_entry": DEALER_STOCK_ROWS,
|
||||
"app.listing": [],
|
||||
"app.deal": [],
|
||||
"vehiq.vehicle": VEHICLE_ROWS,
|
||||
});
|
||||
const seen: IngestionEnvelope[] = [];
|
||||
const dimensionSeen: IngestionEnvelope[] = [];
|
||||
const deps = baseDeps(query, async (e) => void seen.push(...e));
|
||||
deps.dimensionSink = async (e) => void dimensionSeen.push(...e);
|
||||
|
||||
const summary = await runVehiqOltpIngestion(deps);
|
||||
|
||||
const dealerEnvelopes = seen.filter((e) => e.partition_key === "dealer_stock_entry");
|
||||
expect(dealerEnvelopes.map((e) => e.record_key)).toEqual(["d1", "d2", "d3"]);
|
||||
expect(summary.referenced_vehicle_ids).toBe(2); // v1, v2 — deduped from d1/d2/d3
|
||||
expect(summary.dimension?.status).toBe("succeeded");
|
||||
expect(dimensionSeen.map((e) => e.record_key).sort()).toEqual(["v1", "v2"]);
|
||||
expect(dimensionSeen.map((e) => e.record_key)).not.toContain("v9");
|
||||
});
|
||||
});
|
||||
|
||||
describe("runVehiqOltpIngestion — resume from checkpoint after an interruption", () => {
|
||||
it("a sink that dies mid-run leaves the checkpoint unmoved for the failed page; the NEXT run resumes with no gap and no duplicate landing", async () => {
|
||||
const { query } = fixtureVehiqDb({
|
||||
"app.dealer_stock_entry": DEALER_STOCK_ROWS,
|
||||
"app.listing": [],
|
||||
"app.deal": [],
|
||||
"vehiq.vehicle": VEHICLE_ROWS,
|
||||
});
|
||||
const repo = new InMemoryIngestionRepository();
|
||||
const schemas = new SchemaRegistry(repo);
|
||||
const landed: IngestionEnvelope[] = [];
|
||||
let call = 0;
|
||||
|
||||
// First run: the sink accepts page 1 (2 rows, page_size=2) then throws on
|
||||
// page 2 — simulating the process dying after the first page's checkpoint
|
||||
// committed but before the second page's did.
|
||||
const crashingSink = async (batch: IngestionEnvelope[]): Promise<void> => {
|
||||
call++;
|
||||
if (call === 2) throw new Error("simulated crash mid-run");
|
||||
landed.push(...batch);
|
||||
};
|
||||
// Matches the generic runner's own documented behaviour (tests/p0/
|
||||
// ingestion-runner.test.ts "a run that dies inside the sink..."): a
|
||||
// throwing sink REJECTS the run, it does not resolve with status:"failed".
|
||||
// dealer_stock_entry is the FIRST source this pipeline processes, so its
|
||||
// page-2 crash is what aborts runVehiqOltpIngestion() itself here.
|
||||
await expect(
|
||||
runVehiqOltpIngestion({
|
||||
repo,
|
||||
schemas,
|
||||
sourceQuery: query,
|
||||
oltpSink: crashingSink,
|
||||
dimensionSink: async () => undefined,
|
||||
tenantId: "t_vehiq",
|
||||
projectId: "vehiq",
|
||||
pageSize: 2,
|
||||
}),
|
||||
).rejects.toThrow(/simulated crash mid-run/);
|
||||
expect(landed.map((e) => e.record_key)).toEqual(["d1", "d2"]); // page 1 landed before the crash
|
||||
// The durable checkpoint reflects page 1 only — page 2's checkpoint never committed.
|
||||
const afterCrash = await repo.getCheckpoint("vehiq_oltp_dealer_stock_entry", "dealer_stock_entry");
|
||||
expect(afterCrash?.records_committed).toBe(2);
|
||||
|
||||
// Second run, same repo (durable checkpoint survives the "restart"): must
|
||||
// NOT re-deliver d1/d2, and MUST deliver d3.
|
||||
const run2 = await runVehiqOltpIngestion({
|
||||
repo,
|
||||
schemas,
|
||||
sourceQuery: query,
|
||||
oltpSink: async (batch) => void landed.push(...batch),
|
||||
dimensionSink: async () => undefined,
|
||||
tenantId: "t_vehiq",
|
||||
projectId: "vehiq",
|
||||
pageSize: 2,
|
||||
});
|
||||
const dealerReport2 = run2.oltp.find((r) => r.partitions[0]?.partition_key === "dealer_stock_entry");
|
||||
expect(dealerReport2?.status).toBe("succeeded");
|
||||
expect(landed.map((e) => e.record_key)).toEqual(["d1", "d2", "d3"]); // exactly once each, no gap, no dupe
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user