diff --git a/.gitea/workflows/vehiq-oltp-ingest.yml b/.gitea/workflows/vehiq-oltp-ingest.yml new file mode 100644 index 0000000..3ddfe45 --- /dev/null +++ b/.gitea/workflows/vehiq-oltp-ingest.yml @@ -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 diff --git a/docs/P070_VEHIQ_OLTP_EXTRACTOR.md b/docs/P070_VEHIQ_OLTP_EXTRACTOR.md new file mode 100644 index 0000000..acc5976 --- /dev/null +++ b/docs/P070_VEHIQ_OLTP_EXTRACTOR.md @@ -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. diff --git a/package.json b/package.json index bc17b56..5270928 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7eeb8f7..36c9826 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -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 diff --git a/scripts/vehiq-oltp-ingest/run.ts b/scripts/vehiq-oltp-ingest/run.ts new file mode 100644 index 0000000..fcae5a5 --- /dev/null +++ b/scripts/vehiq-oltp-ingest/run.ts @@ -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 { + 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; +}); diff --git a/src/connectors/postgres-keyed-lookup-connector.ts b/src/connectors/postgres-keyed-lookup-connector.ts new file mode 100644 index 0000000..7d164e6 --- /dev/null +++ b/src/connectors/postgres-keyed-lookup-connector.ts @@ -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; + /** 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(); + private connected = false; + + /** Query calls actually issued, for tests to assert the narrow-scan property against. */ + readonly queryLog: Array<{ sql: string; keys: Array }> = []; + + 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 { + 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 { + const config = this.requireSession("discover"); + return { + partitions: [{ partition_key: config.partition_key, name: config.table, schema: config.schema }], + modes: ["batch"], + }; + } + + async health(): Promise { + 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 { + 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 { + 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 { + 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 { + 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; + } +} diff --git a/src/connectors/postgres-query-impl.ts b/src/connectors/postgres-query-impl.ts new file mode 100644 index 0000000..bfe4407 --- /dev/null +++ b/src/connectors/postgres-query-impl.ts @@ -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; close: () => Promise } { + 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 => { + assertReadOnlyStatement(text); + const rows = await sql.unsafe(text, params as never[]); + return rows.map((r) => ({ ...r })); + }; + return { query, sql, close: () => sql.end({ timeout: 5 }) }; +} diff --git a/src/connectors/postgres-watermark-connector.ts b/src/connectors/postgres-watermark-connector.ts new file mode 100644 index 0000000..f27614a --- /dev/null +++ b/src/connectors/postgres-watermark-connector.ts @@ -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; + +/** + * 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; + +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(); + 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 { + 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 { + 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 { + 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 { + 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 { + this.requireSession("resume"); + return this.read({ partition_key: checkpoint.partition_key, cursor: checkpoint.cursor, limit }); + } + + async backfill(req: BackfillRequest): Promise { + 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 { + 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; + } +} diff --git a/src/ingestion/projection-sink.ts b/src/ingestion/projection-sink.ts new file mode 100644 index 0000000..145e93b --- /dev/null +++ b/src/ingestion/projection-sink.ts @@ -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 = {}; + const measures: Record = {}; + 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 { + return async (envelopes: IngestionEnvelope[]): Promise => { + if (envelopes.length === 0) return; + const byPartition = new Map(); + 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`, + ); + } + } + }; +} diff --git a/src/ingestion/vehiq-oltp-ingest-runner.ts b/src/ingestion/vehiq-oltp-ingest-runner.ts new file mode 100644 index 0000000..2bac136 --- /dev/null +++ b/src/ingestion/vehiq-oltp-ingest-runner.ts @@ -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, + opts: { sourceId: string; tenantId: string; projectId: string; limit?: number } = { + sourceId: "vehiq_oltp", + tenantId: "t_vehiq", + projectId: "vehiq", + }, +): Promise { + 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 { + const runner = new IngestionRunner(deps.repo, deps.schemas, deps.lineage); + const sourceIdPrefix = deps.sourceIdPrefix ?? "vehiq_oltp"; + const oltpReports: IngestionRunReport[] = []; + const referencedVehicleIds = new Set(); + + 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 }; diff --git a/tests/connectors/postgres-keyed-lookup-connector.test.ts b/tests/connectors/postgres-keyed-lookup-connector.test.ts new file mode 100644 index 0000000..be3f683 --- /dev/null +++ b/tests/connectors/postgres-keyed-lookup-connector.test.ts @@ -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]; + return VEHICLE_TABLE.filter((row) => ids.includes(row.id as string)); + }; + return { query, calls }; +} + +function config(keys: Array, 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); + }); +}); diff --git a/tests/connectors/postgres-watermark-connector.test.ts b/tests/connectors/postgres-watermark-connector.test.ts new file mode 100644 index 0000000..2cf703c --- /dev/null +++ b/tests/connectors/postgres-watermark-connector.test.ts @@ -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"); + } + }); +}); diff --git a/tests/ingestion/vehiq-oltp-ingest-runner.test.ts b/tests/ingestion/vehiq-oltp-ingest-runner.test.ts new file mode 100644 index 0000000..5468016 --- /dev/null +++ b/tests/ingestion/vehiq-oltp-ingest-runner.test.ts @@ -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 , `) 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): { 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]; + 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): 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 => { + 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 + }); +});