From ec26ddc296a710c35830d03ee2458d03c8f7ef5a Mon Sep 17 00:00:00 2001 From: Claude Opus 5 Date: Mon, 7 Sep 2026 02:45:41 +0200 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20BI-P022=20=E2=80=94=20dbt=20?= =?UTF-8?q?canonical=20transformation=20CONTRACT=20and=20reference=20proje?= =?UTF-8?q?ct?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scope stated in the module's first paragraph: this defines the contract, it does not run dbt, and nothing in this repo does. Execution is N0D's (n0d#38), which consumes this from the other side. There is deliberately no DbtExecutor implementation here — only the interface — so "0pt can run dbt" cannot become true by accident and the two repos share one versioned shape. `analytics/dbt/` is a real reference project covering raw -> staging -> curated -> serving, with the four layers mapped onto BI-P003's existing tiers rather than inventing a fifth vocabulary. The tenant predicate is applied at the FIRST modelled layer so every downstream model inherits it and no later model has to remember — the same forced-predicate rule as the query compiler's, one layer up. Two singular tests ship with it, both aimed at false greens: * `assert_mart_is_not_empty` — without it every column test passes trivially over zero rows and the whole suite reports green while the pipeline is dead; * `assert_no_cross_tenant_rows` — which should be impossible to fail, and exists precisely for that reason. An isolation guarantee nobody tests is the one that turns out to have four holes rather than the one reported. BI-P022's version rule is a GATE. `DBT_RUNTIME_PIN` records dbt Core 1.10.13 (stable) with dbt-clickhouse 1.9.2 and BOTH review flags FALSE — no licence or supply-chain review has been performed in this repo and none is claimed. `assertDbtRuntimeCertified` refuses a production run until they are true, and refuses an alpha line in production without a separate certification reference. A version constant on its own would be the documented-but-unenforced pattern; a test pins the flags at false so flipping them is a visible deliberate act. The result is treated as UNTRUSTED INPUT, because it arrives from another repo's executor and this estate has been burned by green statuses that meant nothing. Refused: `status: success` with ZERO nodes executed (a dbt selector matching nothing exits 0 — the dbt-shaped false green); an observed runtime different from the pin (the models mean something else and nothing says so); a revision that is not an exact 40-char SHA (a branch name is not reproducible, and reproducibility is the DoD); a missing manifest.json or run_results.json, without which the run cannot be audited or its tests ingested by BI-P025. Layer dependencies are checked against the manifest GRAPH, not asserted as a convention: a mart reading a source directly, or skipping a layer, or a staging model reading a mart, are each refused. Selectors are validated rather than passed through — a free-form argument list is a shell surface, and a shell surface on a job holding warehouse credentials is the whole problem. The job also cannot name an arbitrary target schema: it must be scoped to the job's tenant. 39 tests. typecheck + eslint clean. Co-Authored-By: Claude Opus 5 --- analytics/dbt/README.md | 51 +++ analytics/dbt/dbt_project.yml | 43 ++ .../intermediate/int_sales_enriched.sql | 16 + .../dbt/models/marts/mart_sales_daily.sql | 17 + analytics/dbt/models/marts/schema.yml | 19 + analytics/dbt/models/sources/sources.yml | 34 ++ analytics/dbt/models/staging/schema.yml | 26 ++ .../dbt/models/staging/stg_customers.sql | 11 + .../dbt/models/staging/stg_sales_events.sql | 30 ++ analytics/dbt/snapshots/snap_customers.sql | 15 + .../dbt/tests/assert_mart_is_not_empty.sql | 7 + .../dbt/tests/assert_no_cross_tenant_rows.sql | 11 + src/analytics/transform/dbt-contract.ts | 374 ++++++++++++++++++ tests/analytics/dbt-contract.test.ts | 321 +++++++++++++++ 14 files changed, 975 insertions(+) create mode 100644 analytics/dbt/README.md create mode 100644 analytics/dbt/dbt_project.yml create mode 100644 analytics/dbt/models/intermediate/int_sales_enriched.sql create mode 100644 analytics/dbt/models/marts/mart_sales_daily.sql create mode 100644 analytics/dbt/models/marts/schema.yml create mode 100644 analytics/dbt/models/sources/sources.yml create mode 100644 analytics/dbt/models/staging/schema.yml create mode 100644 analytics/dbt/models/staging/stg_customers.sql create mode 100644 analytics/dbt/models/staging/stg_sales_events.sql create mode 100644 analytics/dbt/snapshots/snap_customers.sql create mode 100644 analytics/dbt/tests/assert_mart_is_not_empty.sql create mode 100644 analytics/dbt/tests/assert_no_cross_tenant_rows.sql create mode 100644 src/analytics/transform/dbt-contract.ts create mode 100644 tests/analytics/dbt-contract.test.ts diff --git a/analytics/dbt/README.md b/analytics/dbt/README.md new file mode 100644 index 0000000..1f7e9c8 --- /dev/null +++ b/analytics/dbt/README.md @@ -0,0 +1,51 @@ +# `analytics/dbt/` — BI-P022 + +The canonical SIAX transformation project. **Nothing in this repository runs +it.** Execution lives in N0D (`n0d#38`), against the contract in +`src/analytics/transform/dbt-contract.ts`. + +## Layers, and why they map onto BI-P003 + +| directory | tier (BI-P003) | may read | must not contain | +| ---------------------- | -------------- | --------------------- | --------------------- | +| `models/sources/` | `RAW` | — (declarations only) | any SQL | +| `models/staging/` | `NORMALIZED` | sources | joins, business logic | +| `models/intermediate/` | `CURATED` | staging | presentation shaping | +| `models/marts/` | `SERVING` | intermediate | direct source reads | + +`assertLayerDependenciesLegal` enforces the "may read" column against the +dependency graph in `manifest.json`. A mart reading a source directly is +refused — that is raw data reaching a dashboard, which is precisely what +BI-P003's tier rule exists to prevent, and a convention nobody checks would not +prevent it. + +## What a run must produce + +`manifest.json` and `run_results.json`, both registered as ST0RE artifacts. +`assertValidDbtJobResult` refuses a result missing either: without them the run +cannot be audited and its tests cannot be ingested by BI-P025's +`DbtQualityProvider`. + +It also refuses: + +- a result whose revision differs from the request — the run is not reproducible; +- a result whose observed dbt version differs from the pin — a runtime that is + not the pinned one silently changes what the models mean; +- `status: success` with **zero nodes executed** — a dbt selector that matches + nothing exits 0, which is the dbt-shaped false green. + +## Version posture + +`DBT_RUNTIME_PIN` records dbt Core 1.10.13 (stable line) with +`dbt-clickhouse` 1.9.2. **Both review flags are `false`.** No licence or +supply-chain review has been performed in this repo, and none is claimed. +`assertDbtRuntimeCertified` refuses a `production` run until they are true, so +BI-P022's "after current license/security verification" is a gate rather than a +sentence in a document. + +## Credentials + +A dbt run never receives a tenant's warehouse credential from this repo. N0D +resolves a credential whose write scope matches `target_layer` and +`expected_target_schema`, and the contract refuses a request whose expected +schema is not scoped to the job's tenant. diff --git a/analytics/dbt/dbt_project.yml b/analytics/dbt/dbt_project.yml new file mode 100644 index 0000000..59c8a95 --- /dev/null +++ b/analytics/dbt/dbt_project.yml @@ -0,0 +1,43 @@ +# BI-P022 — the canonical SIAX analytics transformation project. +# +# Execution happens in N0D (n0d#38), never here. This project is the MODELS and +# the CONVENTIONS; src/analytics/transform/dbt-contract.ts is the job contract +# N0D consumes, and it is what refuses a run that is not reproducible. +name: siax_analytics +version: 1.0.0 +config-version: 2 +profile: siax_analytics + +model-paths: ['models'] +test-paths: ['tests'] +snapshot-paths: ['snapshots'] +macro-paths: ['macros'] + +# The four layers map onto BI-P003's tiers rather than inventing a fifth +# vocabulary. `+tags` carries the tier so a manifest reader can assign one +# without parsing paths, and dbt-contract.ts asserts the mapping. +models: + siax_analytics: + +materialized: view + staging: + +schema: staging + +materialized: view + +tags: ['tier:NORMALIZED'] + intermediate: + +schema: intermediate + +materialized: table + +tags: ['tier:CURATED'] + marts: + +schema: marts + +materialized: table + +tags: ['tier:SERVING'] + +snapshots: + siax_analytics: + +target_schema: snapshots + +# `on_schema_change: fail` rather than `sync_all_columns`: a silently widened +# table is how a downstream metric starts averaging a column that did not exist +# yesterday. BI-P004's contract gate is the same rule one layer up. +vars: + siax_tenant_id: '__set_by_n0d__' diff --git a/analytics/dbt/models/intermediate/int_sales_enriched.sql b/analytics/dbt/models/intermediate/int_sales_enriched.sql new file mode 100644 index 0000000..22efc0b --- /dev/null +++ b/analytics/dbt/models/intermediate/int_sales_enriched.sql @@ -0,0 +1,16 @@ +-- CURATED (BI-P003). Joins and conformed entities. Reads STAGING only. +{{ config(materialized='table', tags=['tier:CURATED']) }} + +select + e.event_id, + e.tenant_id, + e.customer_id, + e.amount_sek, + e.channel, + coalesce(c.country_code, e.country_code) as country_code, + c.segment, + e.invoiced_at +from {{ ref('stg_sales_events') }} e +left join {{ ref('stg_customers') }} c + on e.customer_id = c.customer_id + and e.tenant_id = c.tenant_id -- the join itself stays tenant-scoped diff --git a/analytics/dbt/models/marts/mart_sales_daily.sql b/analytics/dbt/models/marts/mart_sales_daily.sql new file mode 100644 index 0000000..229caf7 --- /dev/null +++ b/analytics/dbt/models/marts/mart_sales_daily.sql @@ -0,0 +1,17 @@ +-- SERVING (BI-P003). What a dashboard or a BI-P028 metric reads. +-- +-- Reads INTERMEDIATE only. Reading a source here would skip two layers, and +-- assertLayerDependenciesLegal refuses that against the manifest graph. +{{ config(materialized='table', tags=['tier:SERVING']) }} + +select + tenant_id, + date_trunc('day', invoiced_at) as day, + country_code, + channel, + segment, + count(*) as order_count, + sum(amount_sek) as net_revenue_sek +from {{ ref('int_sales_enriched') }} +where channel != 'internal' -- matches the net_revenue metric's baked filter +group by 1, 2, 3, 4, 5 diff --git a/analytics/dbt/models/marts/schema.yml b/analytics/dbt/models/marts/schema.yml new file mode 100644 index 0000000..a84817d --- /dev/null +++ b/analytics/dbt/models/marts/schema.yml @@ -0,0 +1,19 @@ +version: 2 + +models: + - name: mart_sales_daily + description: > + SERVING projection behind the `net_revenue` and `orders` metrics. + A blocking test failure here quarantines the serving projection + (BI-P025); the quarantine is enforced on the READ side. + columns: + - name: tenant_id + tests: [not_null] + - name: day + tests: [not_null] + - name: net_revenue_sek + tests: [not_null] + - name: order_count + tests: + - not_null + - dbt_utils.accepted_range: { min_value: 1, inclusive: true } diff --git a/analytics/dbt/models/sources/sources.yml b/analytics/dbt/models/sources/sources.yml new file mode 100644 index 0000000..34c152c --- /dev/null +++ b/analytics/dbt/models/sources/sources.yml @@ -0,0 +1,34 @@ +version: 2 + +# RAW. Declared, never modelled. A source has no SQL of its own — anything that +# transforms belongs in staging, where BI-P003 assigns it NORMALIZED. +sources: + - name: opt_raw + description: > + Landed, untransformed data as delivered by the 0pt sourcing engine. + RAW tier: no BI surface may read this directly without a recorded + BI-P003 exception. + schema: "{{ var('siax_tenant_id') }}_raw" + loaded_at_field: ingested_at + freshness: + # BI-P026's DELAYED/STALE thresholds, expressed where dbt can act on them. + warn_after: { count: 90, period: minute } + error_after: { count: 180, period: minute } + tables: + - name: sales_events + description: One row per invoiced sales line, as landed. + columns: + - name: event_id + tests: [not_null, unique] + - name: tenant_id + description: Present on every raw row; the tier boundary depends on it. + tests: [not_null] + - name: ingested_at + tests: [not_null] + - name: customers + description: Customer master as landed. + columns: + - name: customer_id + tests: [not_null, unique] + - name: tenant_id + tests: [not_null] diff --git a/analytics/dbt/models/staging/schema.yml b/analytics/dbt/models/staging/schema.yml new file mode 100644 index 0000000..dd4216c --- /dev/null +++ b/analytics/dbt/models/staging/schema.yml @@ -0,0 +1,26 @@ +version: 2 + +models: + - name: stg_sales_events + description: Normalized sales events — one row per event_id, latest landing wins. + columns: + - name: event_id + tests: [not_null, unique] + - name: tenant_id + tests: [not_null] + - name: amount_sek + tests: + - not_null + - dbt_utils.accepted_range: + min_value: 0 + inclusive: true + - name: channel + tests: + - accepted_values: + values: ['web', 'retail', 'partner', 'internal'] + - name: stg_customers + columns: + - name: customer_id + tests: [not_null, unique] + - name: tenant_id + tests: [not_null] diff --git a/analytics/dbt/models/staging/stg_customers.sql b/analytics/dbt/models/staging/stg_customers.sql new file mode 100644 index 0000000..2ef66df --- /dev/null +++ b/analytics/dbt/models/staging/stg_customers.sql @@ -0,0 +1,11 @@ +-- NORMALIZED (BI-P003). +{{ config(materialized='view', tags=['tier:NORMALIZED']) }} + +select + cast(customer_id as varchar) as customer_id, + cast(tenant_id as varchar) as tenant_id, + cast(name as varchar) as customer_name, + cast(segment as varchar) as segment, + cast(country as varchar) as country_code +from {{ source('opt_raw', 'customers') }} +where tenant_id = '{{ var("siax_tenant_id") }}' diff --git a/analytics/dbt/models/staging/stg_sales_events.sql b/analytics/dbt/models/staging/stg_sales_events.sql new file mode 100644 index 0000000..e1147e7 --- /dev/null +++ b/analytics/dbt/models/staging/stg_sales_events.sql @@ -0,0 +1,30 @@ +-- NORMALIZED (BI-P003). Rename, cast, deduplicate. No joins, no business logic. +-- +-- The tenant predicate is applied HERE, at the first modelled layer, so every +-- downstream model inherits it and no later model has to remember. It is the +-- same forced-predicate rule as the query compiler's, one layer up. +{{ config(materialized='view', tags=['tier:NORMALIZED']) }} + +with source as ( + select * from {{ source('opt_raw', 'sales_events') }} + where tenant_id = '{{ var("siax_tenant_id") }}' +), + +deduplicated as ( + select + *, + row_number() over (partition by event_id order by ingested_at desc) as _rn + from source +) + +select + cast(event_id as varchar) as event_id, + cast(tenant_id as varchar) as tenant_id, + cast(customer_id as varchar) as customer_id, + cast(amount as double) as amount_sek, + cast(channel as varchar) as channel, + cast(country as varchar) as country_code, + cast(invoiced_at as timestamp) as invoiced_at, + cast(ingested_at as timestamp) as ingested_at +from deduplicated +where _rn = 1 diff --git a/analytics/dbt/snapshots/snap_customers.sql b/analytics/dbt/snapshots/snap_customers.sql new file mode 100644 index 0000000..d712410 --- /dev/null +++ b/analytics/dbt/snapshots/snap_customers.sql @@ -0,0 +1,15 @@ +-- Slowly-changing customer attributes, so a historical metric can be +-- recomputed against what a customer's segment WAS, not what it is now. +{% snapshot snap_customers %} +{{ + config( + target_schema='snapshots', + unique_key='customer_id', + strategy='check', + check_cols=['segment', 'country_code'], + invalidate_hard_deletes=True + ) +}} +select customer_id, tenant_id, customer_name, segment, country_code +from {{ ref('stg_customers') }} +{% endsnapshot %} diff --git a/analytics/dbt/tests/assert_mart_is_not_empty.sql b/analytics/dbt/tests/assert_mart_is_not_empty.sql new file mode 100644 index 0000000..ead1f98 --- /dev/null +++ b/analytics/dbt/tests/assert_mart_is_not_empty.sql @@ -0,0 +1,7 @@ +-- Fails when the serving mart is EMPTY. +-- +-- Without this, every column test passes trivially over zero rows and the +-- whole suite reports green while the pipeline is dead — the same false green +-- BI-P025's NativeQualityProvider refuses. +select 1 as empty_mart +having (select count(*) from {{ ref('mart_sales_daily') }}) = 0 diff --git a/analytics/dbt/tests/assert_no_cross_tenant_rows.sql b/analytics/dbt/tests/assert_no_cross_tenant_rows.sql new file mode 100644 index 0000000..5e5c679 --- /dev/null +++ b/analytics/dbt/tests/assert_no_cross_tenant_rows.sql @@ -0,0 +1,11 @@ +-- A singular test that FAILS if the serving mart ever contains a row belonging +-- to a tenant other than the one the run was scoped to. +-- +-- The tenant predicate is applied in staging and inherited downstream, so this +-- test should be impossible to fail. That is precisely why it exists: an +-- isolation guarantee nobody tests is the one that turns out to have four +-- holes rather than the one reported. +select tenant_id, count(*) as rows_from_another_tenant +from {{ ref('mart_sales_daily') }} +where tenant_id != '{{ var("siax_tenant_id") }}' +group by tenant_id diff --git a/src/analytics/transform/dbt-contract.ts b/src/analytics/transform/dbt-contract.ts new file mode 100644 index 0000000..76e5c5d --- /dev/null +++ b/src/analytics/transform/dbt-contract.ts @@ -0,0 +1,374 @@ +/** + * src/analytics/transform/dbt-contract.ts + * ======================================== + * BI-P022 — DBT CORE CANONICAL TRANSFORMATION. + * + * ── SCOPE, STATED FIRST ─────────────────────────────────────────────────── + * This module defines the CONTRACT. It does not run dbt, and nothing in this + * repo does. Execution belongs to N0D (n0d#38), which consumes this contract + * from the other side. There is deliberately no `DbtExecutor` implementation + * here — only the interface — so that "0pt can run dbt" cannot become true by + * accident, and so the two repos have one shared, versioned shape rather than + * two drifting ones. + * + * ── THE VERSION RULE BI-P022 STATES ─────────────────────────────────────── + * "Use stable dbt Core production line AFTER current license/security + * verification. Do NOT use alpha development engine in production unless + * separately certified." + * + * So the pin is recorded together with its certification STATE, and the state + * is `unverified` until someone does the review. `assertDbtRuntimeCertified` + * refuses a production run against an uncertified or alpha runtime — which + * makes the sentence above a gate rather than a paragraph. A version constant + * on its own would be exactly the "documented but unenforced" pattern. + * + * ── THE TIER MAPPING ────────────────────────────────────────────────────── + * BI-P022's DoD names raw → staging → curated → serving. BI-P003 (merged) + * already owns RAW/NORMALIZED/CURATED/SERVING, so the dbt layers map onto it + * rather than inventing a fifth vocabulary: + * + * sources/ -> RAW (declared, never modelled) + * staging/ -> NORMALIZED (rename, cast, deduplicate; no business logic) + * intermediate/ -> CURATED (joins, business logic, conformed entities) + * marts/ -> SERVING (what a dashboard or a metric reads) + * + * A model in the wrong directory for its tier is a rejection, not a warning: + * the tier is what BI-P003's read rule gates on, and a mart that is really a + * staging model is how RAW data reaches a dashboard. + */ +import { SiaxError } from '../../platform/errors'; +import type { AnalyticsTier } from '../domain/tiers'; + +/** dbt layers, in dependency order. */ +export const DBT_LAYERS = ['sources', 'staging', 'intermediate', 'marts'] as const; +export type DbtLayer = (typeof DBT_LAYERS)[number]; + +export const DBT_LAYER_TIERS: Record = { + sources: 'RAW', + staging: 'NORMALIZED', + intermediate: 'CURATED', + marts: 'SERVING', +}; + +/** Directories BI-P022 requires under `analytics/dbt/`. */ +export const REQUIRED_DBT_DIRECTORIES = [ + 'models/sources', + 'models/staging', + 'models/intermediate', + 'models/marts', + 'tests', + 'snapshots', +] as const; + +// ── runtime certification ────────────────────────────────────────────────── + +export type DbtReleaseLine = 'stable' | 'prerelease' | 'alpha'; + +export interface DbtRuntimePin { + /** e.g. "1.10.13". Exact, never a range. */ + version: string; + line: DbtReleaseLine; + /** Adapter and its pin, e.g. { name: "dbt-clickhouse", version: "1.9.2" }. */ + adapter: { name: string; version: string }; + /** Has a licence + supply-chain review been completed FOR THIS PIN? */ + licenseReviewed: boolean; + securityReviewed: boolean; + reviewedBy?: string; + reviewedAt?: string; + /** Required by BI-P022 to run an alpha line in production at all. */ + separateCertificationRef?: string; +} + +/** + * The pin this contract is written against. + * + * BOTH review flags are FALSE. No licence or supply-chain review of dbt Core + * has been performed in this repo, and none is claimed. The constant records + * an intended pin; `assertDbtRuntimeCertified` is what stops it being treated + * as an approval. Flipping these to true without doing the review is the + * failure this shape exists to make visible. + */ +export const DBT_RUNTIME_PIN: DbtRuntimePin = { + version: '1.10.13', + line: 'stable', + adapter: { name: 'dbt-clickhouse', version: '1.9.2' }, + licenseReviewed: false, + securityReviewed: false, +}; + +export type DbtRunEnvironment = 'development' | 'ci' | 'production'; + +/** BI-P022's version rule, as a gate. */ +export function assertDbtRuntimeCertified(pin: DbtRuntimePin, environment: DbtRunEnvironment): void { + if (!/^\d+\.\d+\.\d+$/.test(pin.version)) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + 'the dbt runtime pin must be an exact version', + ); + } + if (environment !== 'production') return; + if (pin.line === 'alpha' && !pin.separateCertificationRef) { + throw new SiaxError( + 'ENTITLEMENT_REQUIRED', + '0pt.analytics.dbt', + 'BI-P022: an alpha dbt engine may not run in production without a separate certification reference', + ); + } + if (!pin.licenseReviewed || !pin.securityReviewed) { + throw new SiaxError( + 'ENTITLEMENT_REQUIRED', + '0pt.analytics.dbt', + `BI-P022: dbt ${pin.version} has not completed licence and security verification; ` + + `it may not run in production`, + ); + } +} + +// ── the job contract N0D executes ────────────────────────────────────────── + +export const DBT_COMMANDS = [ + 'build', + 'run', + 'test', + 'snapshot', + 'seed', + 'compile', + 'source_freshness', +] as const; +export type DbtCommand = (typeof DBT_COMMANDS)[number]; + +/** + * What 0pt hands N0D. Deliberately narrow: there is no free-form `args` field, + * because a passthrough argument list is a shell surface, and a shell surface + * on a job that holds warehouse credentials is the whole problem. + */ +export interface DbtJobRequest { + /** Verified, from CL0UD. Never read from a request body. */ + tenant_id: string; + project_id: string; + /** Which analytics project to run — a registered name, not a path. */ + dbt_project: string; + /** Git ref of the analytics repo the models come from. Exact SHA. */ + revision: string; + command: DbtCommand; + /** dbt node selectors. Validated; never concatenated into a shell string. */ + select?: string[]; + exclude?: string[]; + /** Target layer, which N0D maps to a credential with matching write scope. */ + target_layer: DbtLayer; + environment: DbtRunEnvironment; + runtime: DbtRuntimePin; + /** Idempotency key, per the standard N0D job contract. */ + idempotency_key: string; + correlation_id?: string; + /** Fail the run if it would write outside this tenant's schema. */ + expected_target_schema: string; +} + +/** The artefacts N0D must return. dbt produces all of them; they are not optional. */ +export const REQUIRED_DBT_ARTIFACTS = ['manifest.json', 'run_results.json'] as const; + +export interface DbtJobResult { + tenant_id: string; + dbt_project: string; + revision: string; + command: DbtCommand; + status: 'success' | 'error' | 'partial'; + started_at: string; + finished_at: string; + /** ST0RE artifact refs, one per REQUIRED_DBT_ARTIFACTS entry. */ + artifacts: Record; + /** Per-node outcomes, the input to BI-P025's DbtQualityProvider. */ + nodes: DbtNodeResult[]; + /** dbt's own version, as reported by the runtime that actually ran. */ + observed_dbt_version: string; +} + +export interface DbtNodeResult { + unique_id: string; + resource_type: 'model' | 'test' | 'snapshot' | 'seed' | 'source'; + status: 'success' | 'error' | 'skipped' | 'pass' | 'fail' | 'warn'; + rows_affected?: number; + execution_time_s?: number; + message?: string; +} + +const SELECTOR_RE = /^[A-Za-z0-9_.:*+@ -]{1,200}$/; +const NAME_RE = /^[a-z][a-z0-9_]{1,62}$/; +const SHA_RE = /^[0-9a-f]{40}$/; + +export function assertValidDbtJobRequest(req: DbtJobRequest): void { + if (!req.tenant_id?.trim()) { + throw new SiaxError('UNAUTHENTICATED', '0pt.analytics.dbt', 'a dbt job requires a verified tenant'); + } + if (!NAME_RE.test(req.dbt_project)) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + 'dbt_project must be a registered snake_case name', + ); + } + if (!SHA_RE.test(req.revision)) { + // A branch name is not reproducible, and BI-P022's DoD is about + // reproducibility. "main" today and "main" tomorrow are different projects. + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + 'revision must be an exact 40-character commit SHA — a branch name is not reproducible', + ); + } + if (!(DBT_COMMANDS as readonly string[]).includes(req.command)) { + throw new SiaxError('INVALID_REQUEST', '0pt.analytics.dbt', `unsupported dbt command "${req.command}"`); + } + if (!(DBT_LAYERS as readonly string[]).includes(req.target_layer)) { + throw new SiaxError('INVALID_REQUEST', '0pt.analytics.dbt', `unknown target layer "${req.target_layer}"`); + } + if (req.target_layer === 'sources') { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + 'sources are DECLARED, never built — a job cannot target the sources layer', + ); + } + for (const selector of [...(req.select ?? []), ...(req.exclude ?? [])]) { + if (!SELECTOR_RE.test(selector)) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + `dbt selector "${selector.slice(0, 40)}" contains characters that are not selector syntax`, + ); + } + } + if (!req.idempotency_key?.trim()) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + 'the N0D job contract requires an idempotency key', + ); + } + if (!req.expected_target_schema.includes(req.tenant_id)) { + // The target schema must be derivable from the tenant. A job that can name + // an arbitrary schema is a cross-tenant write waiting to happen. + throw new SiaxError( + 'AUTHORIZATION_DENIED', + '0pt.analytics.dbt', + "expected_target_schema must be scoped to the job's tenant", + ); + } + assertDbtRuntimeCertified(req.runtime, req.environment); +} + +/** + * Validate what came BACK. A result is untrusted input: it arrives from + * another repo's executor, and the estate has been burned by trusting a green + * status that meant nothing. + */ +export function assertValidDbtJobResult(result: DbtJobResult, request: DbtJobRequest): void { + if (result.tenant_id !== request.tenant_id) { + throw new SiaxError('AUTHORIZATION_DENIED', '0pt.analytics.dbt', 'dbt result is for a different tenant'); + } + if (result.revision !== request.revision) { + throw new SiaxError( + 'CONFLICT', + '0pt.analytics.dbt', + 'dbt result is for a different revision than was requested — the run is not reproducible', + ); + } + for (const artifact of REQUIRED_DBT_ARTIFACTS) { + if (!result.artifacts[artifact]) { + throw new SiaxError( + 'PERMANENT_FAILURE', + '0pt.analytics.dbt', + `dbt result is missing "${artifact}" — without it the run cannot be audited or its tests ingested`, + ); + } + } + if (result.observed_dbt_version !== request.runtime.version) { + // A runtime that is not the pinned one silently changes what the models + // mean. Catching it here is the difference between a pin and a wish. + throw new SiaxError( + 'CONFLICT', + '0pt.analytics.dbt', + `dbt ran as ${result.observed_dbt_version} but ${request.runtime.version} was pinned`, + ); + } + if (result.status === 'success' && result.nodes.length === 0) { + // A "successful" run that built nothing is the dbt-shaped false green: a + // selector that matched no models exits 0. + throw new SiaxError( + 'PERMANENT_FAILURE', + '0pt.analytics.dbt', + 'dbt reported success but executed zero nodes — a selector that matches nothing exits 0', + ); + } + const failed = result.nodes.filter((n) => n.status === 'error' || n.status === 'fail'); + if (result.status === 'success' && failed.length > 0) { + throw new SiaxError( + 'PERMANENT_FAILURE', + '0pt.analytics.dbt', + `dbt reported success while ${failed.length} node(s) failed`, + ); + } +} + +/** + * The executor N0D implements. Declared so both repos compile against one + * shape; deliberately NOT implemented here. + */ +export interface DbtExecutor { + readonly runsIn: 'n0d'; + execute(request: DbtJobRequest): Promise; +} + +/** Which tier a model path belongs to, or a rejection. */ +export function tierForModelPath(modelPath: string): AnalyticsTier { + const match = /(?:^|\/)models\/(sources|staging|intermediate|marts)\//.exec(modelPath); + if (!match) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + `model "${modelPath}" is not under one of ${DBT_LAYERS.join('/')} — BI-P003 cannot assign it a tier`, + ); + } + return DBT_LAYER_TIERS[match[1] as DbtLayer]; +} + +/** + * A staging model must not reach past sources, a mart must not read a source + * directly, and nothing may skip a layer. Enforced on the dependency graph + * from `manifest.json`, because "we follow the convention" is not a check. + */ +const LAYER_ORDER: DbtLayer[] = ['sources', 'staging', 'intermediate', 'marts']; + +export function assertLayerDependenciesLegal(edges: Array<{ from: string; to: string }>): void { + for (const edge of edges) { + const fromLayer = layerOf(edge.from); + const toLayer = layerOf(edge.to); + const fromIdx = LAYER_ORDER.indexOf(fromLayer); + const toIdx = LAYER_ORDER.indexOf(toLayer); + if (toIdx > fromIdx) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + `"${edge.from}" (${fromLayer}) depends on "${edge.to}" (${toLayer}) — a model may not read a later layer`, + ); + } + if (fromIdx - toIdx > 1) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.dbt', + `"${edge.from}" (${fromLayer}) skips a layer to read "${edge.to}" (${toLayer}) — ` + + `raw data reaching a mart directly is what BI-P003's tier rule exists to prevent`, + ); + } + } +} + +function layerOf(path: string): DbtLayer { + const match = /(?:^|\/)models\/(sources|staging|intermediate|marts)\//.exec(path); + if (!match) { + throw new SiaxError('INVALID_REQUEST', '0pt.analytics.dbt', `model "${path}" is not in a known layer`); + } + return match[1] as DbtLayer; +} diff --git a/tests/analytics/dbt-contract.test.ts b/tests/analytics/dbt-contract.test.ts new file mode 100644 index 0000000..29a7a0d --- /dev/null +++ b/tests/analytics/dbt-contract.test.ts @@ -0,0 +1,321 @@ +/** + * tests/analytics/dbt-contract.test.ts + * ===================================== + * BI-P022 — DBT CORE CANONICAL TRANSFORMATION (the CONTRACT). + * + * This repo does not run dbt; N0D (n0d#38) does, against this contract. So + * what is testable here is exactly what the contract must refuse, and the + * interesting refusals are the ones that look like successes: + * + * * `status: "success"` with ZERO nodes executed — a dbt selector that + * matches nothing exits 0. The dbt-shaped false green. + * * a result whose observed dbt version differs from the pin — the models + * mean something different and nothing says so. + * * a branch name as the revision — "main" today and "main" tomorrow are + * different projects, and BI-P022's DoD is about reproducibility. + * * a production run against an unreviewed runtime, which is BI-P022's + * "after current license/security verification" turned into a gate. + * + * The scaffold itself is asserted against the filesystem, because a directory + * convention that nothing checks is a directory convention that drifts. + */ +import fs from 'node:fs'; +import path from 'node:path'; +import { describe, expect, it } from 'vitest'; +import { SiaxError } from '@/src/platform/errors'; +import { + assertDbtRuntimeCertified, + assertLayerDependenciesLegal, + assertValidDbtJobRequest, + assertValidDbtJobResult, + DBT_LAYER_TIERS, + DBT_LAYERS, + DBT_RUNTIME_PIN, + REQUIRED_DBT_ARTIFACTS, + REQUIRED_DBT_DIRECTORIES, + tierForModelPath, + type DbtJobRequest, + type DbtJobResult, +} from '@/src/analytics/transform/dbt-contract'; + +const DBT_ROOT = path.join(process.cwd(), 'analytics/dbt'); +const SHA = 'a'.repeat(40); + +const CERTIFIED_PIN = { + ...DBT_RUNTIME_PIN, + licenseReviewed: true, + securityReviewed: true, + reviewedBy: 'u_1', + reviewedAt: '2026-09-07T00:00:00Z', +}; + +function request(over: Partial = {}): DbtJobRequest { + return { + tenant_id: 't_alpha', + project_id: 'p_main', + dbt_project: 'siax_analytics', + revision: SHA, + command: 'build', + target_layer: 'marts', + environment: 'ci', + runtime: DBT_RUNTIME_PIN, + idempotency_key: 'idem-1', + expected_target_schema: 't_alpha_marts', + ...over, + }; +} + +function result(over: Partial = {}): DbtJobResult { + return { + tenant_id: 't_alpha', + dbt_project: 'siax_analytics', + revision: SHA, + command: 'build', + status: 'success', + started_at: '2026-09-07T00:00:00Z', + finished_at: '2026-09-07T00:05:00Z', + artifacts: { + 'manifest.json': 'st0re://a/manifest.json', + 'run_results.json': 'st0re://a/run_results.json', + }, + nodes: [ + { unique_id: 'model.siax_analytics.mart_sales_daily', resource_type: 'model', status: 'success' }, + ], + observed_dbt_version: DBT_RUNTIME_PIN.version, + ...over, + }; +} + +describe('BI-P022 — the scaffold exists and matches the declared conventions', () => { + it.each([...REQUIRED_DBT_DIRECTORIES])('analytics/dbt/%s exists', (dir) => { + expect(fs.existsSync(path.join(DBT_ROOT, dir)), `${dir} is missing`).toBe(true); + }); + + it('declares a reference project covering raw -> staging -> curated -> serving', () => { + expect(fs.existsSync(path.join(DBT_ROOT, 'models/sources/sources.yml'))).toBe(true); + expect(fs.existsSync(path.join(DBT_ROOT, 'models/staging/stg_sales_events.sql'))).toBe(true); + expect(fs.existsSync(path.join(DBT_ROOT, 'models/intermediate/int_sales_enriched.sql'))).toBe(true); + expect(fs.existsSync(path.join(DBT_ROOT, 'models/marts/mart_sales_daily.sql'))).toBe(true); + expect(fs.existsSync(path.join(DBT_ROOT, 'snapshots/snap_customers.sql'))).toBe(true); + }); + + it('the staging model applies the tenant predicate at the first modelled layer', () => { + const sql = fs.readFileSync(path.join(DBT_ROOT, 'models/staging/stg_sales_events.sql'), 'utf8'); + expect(sql).toContain('where tenant_id = \'{{ var("siax_tenant_id") }}\''); + }); + + it('the mart reads intermediate, never a source directly', () => { + const sql = fs.readFileSync(path.join(DBT_ROOT, 'models/marts/mart_sales_daily.sql'), 'utf8'); + expect(sql).toContain("ref('int_sales_enriched')"); + expect(sql).not.toContain('source('); + }); + + it('ships a test that fails on an EMPTY mart', () => { + // Without it, every column test passes trivially over zero rows. + const sql = fs.readFileSync(path.join(DBT_ROOT, 'tests/assert_mart_is_not_empty.sql'), 'utf8'); + expect(sql).toContain('= 0'); + }); + + it('ships a cross-tenant test even though it should be impossible to fail', () => { + const sql = fs.readFileSync(path.join(DBT_ROOT, 'tests/assert_no_cross_tenant_rows.sql'), 'utf8'); + expect(sql).toContain('tenant_id !='); + }); + + it("maps the four layers onto BI-P003's tiers", () => { + expect(DBT_LAYER_TIERS).toEqual({ + sources: 'RAW', + staging: 'NORMALIZED', + intermediate: 'CURATED', + marts: 'SERVING', + }); + expect([...DBT_LAYERS]).toEqual(['sources', 'staging', 'intermediate', 'marts']); + }); + + it('assigns a tier from a model path, and refuses one it cannot place', () => { + expect(tierForModelPath('models/marts/mart_sales_daily.sql')).toBe('SERVING'); + expect(tierForModelPath('analytics/dbt/models/staging/x.sql')).toBe('NORMALIZED'); + expect(() => tierForModelPath('models/scratch/x.sql')).toThrowError(SiaxError); + }); +}); + +describe('BI-P022 — the version rule is a GATE, not a paragraph', () => { + it('the shipped pin is honestly UNREVIEWED', () => { + // If someone flips these without doing the review, that is a deliberate + // act visible in a diff — which is the point of storing the state. + expect(DBT_RUNTIME_PIN.licenseReviewed).toBe(false); + expect(DBT_RUNTIME_PIN.securityReviewed).toBe(false); + expect(DBT_RUNTIME_PIN.line).toBe('stable'); + }); + + it('refuses a PRODUCTION run against an unreviewed runtime', () => { + expect(() => assertDbtRuntimeCertified(DBT_RUNTIME_PIN, 'production')).toThrowError(SiaxError); + }); + + it('allows development and CI against the same unreviewed runtime', () => { + expect(() => assertDbtRuntimeCertified(DBT_RUNTIME_PIN, 'ci')).not.toThrow(); + expect(() => assertDbtRuntimeCertified(DBT_RUNTIME_PIN, 'development')).not.toThrow(); + }); + + it('allows production once both reviews are recorded', () => { + expect(() => assertDbtRuntimeCertified(CERTIFIED_PIN, 'production')).not.toThrow(); + }); + + it('refuses an ALPHA line in production without a separate certification reference', () => { + const alpha = { ...CERTIFIED_PIN, line: 'alpha' as const }; + expect(() => assertDbtRuntimeCertified(alpha, 'production')).toThrowError(SiaxError); + expect(() => + assertDbtRuntimeCertified({ ...alpha, separateCertificationRef: 'CERT-2026-11' }, 'production'), + ).not.toThrow(); + }); + + it('refuses a version range — a pin is exact or it is not a pin', () => { + expect(() => assertDbtRuntimeCertified({ ...CERTIFIED_PIN, version: '1.10' }, 'ci')).toThrowError( + SiaxError, + ); + expect(() => assertDbtRuntimeCertified({ ...CERTIFIED_PIN, version: '^1.10.0' }, 'ci')).toThrowError( + SiaxError, + ); + }); +}); + +describe('BI-P022 — the job request', () => { + it('accepts a well-formed request', () => { + expect(() => assertValidDbtJobRequest(request())).not.toThrow(); + }); + + it("refuses a branch name as the revision — BI-P022's DoD is reproducibility", () => { + for (const rev of ['main', 'HEAD', 'v1.0.0', 'a'.repeat(39)]) { + expect(() => assertValidDbtJobRequest(request({ revision: rev }))).toThrowError(SiaxError); + } + }); + + it('refuses a target schema that is not scoped to the tenant', () => { + let thrown: unknown; + try { + assertValidDbtJobRequest(request({ expected_target_schema: 't_beta_marts' })); + } catch (e) { + thrown = e; + } + expect((thrown as SiaxError).code).toBe('AUTHORIZATION_DENIED'); + }); + + it('refuses a selector carrying shell metacharacters', () => { + for (const selector of ['stg_sales; rm -rf /', '$(whoami)', 'a`b`', 'x && y', 'a|b']) { + expect(() => assertValidDbtJobRequest(request({ select: [selector] }))).toThrowError(SiaxError); + } + // ...while accepting real dbt selector syntax. + expect(() => + assertValidDbtJobRequest( + request({ select: ['tag:tier:SERVING+', 'state:modified', '+mart_sales_daily'] }), + ), + ).not.toThrow(); + }); + + it('refuses a job targeting the sources layer — sources are declared, not built', () => { + expect(() => assertValidDbtJobRequest(request({ target_layer: 'sources' }))).toThrowError(SiaxError); + }); + + it('refuses a request with no idempotency key', () => { + expect(() => assertValidDbtJobRequest(request({ idempotency_key: '' }))).toThrowError(SiaxError); + }); + + it('refuses a production request on the unreviewed pin', () => { + expect(() => + assertValidDbtJobRequest( + request({ environment: 'production', expected_target_schema: 't_alpha_marts' }), + ), + ).toThrowError(SiaxError); + }); +}); + +describe('BI-P022 — the job RESULT is untrusted input', () => { + it('accepts a well-formed result', () => { + expect(() => assertValidDbtJobResult(result(), request())).not.toThrow(); + }); + + it('REFUSES success with zero nodes — a selector matching nothing exits 0', () => { + let thrown: unknown; + try { + assertValidDbtJobResult(result({ nodes: [] }), request()); + } catch (e) { + thrown = e; + } + expect(thrown).toBeInstanceOf(SiaxError); + expect((thrown as SiaxError).message).toContain('zero nodes'); + }); + + it('refuses success while nodes failed', () => { + expect(() => + assertValidDbtJobResult( + result({ nodes: [{ unique_id: 't1', resource_type: 'test', status: 'fail' }] }), + request(), + ), + ).toThrowError(SiaxError); + }); + + it('refuses a runtime that is not the pinned one', () => { + expect(() => assertValidDbtJobResult(result({ observed_dbt_version: '1.9.0' }), request())).toThrowError( + SiaxError, + ); + }); + + it('refuses a result for a different revision', () => { + expect(() => assertValidDbtJobResult(result({ revision: 'b'.repeat(40) }), request())).toThrowError( + SiaxError, + ); + }); + + it('refuses a result for a different tenant', () => { + expect(() => assertValidDbtJobResult(result({ tenant_id: 't_beta' }), request())).toThrowError(SiaxError); + }); + + it.each([...REQUIRED_DBT_ARTIFACTS])('refuses a result missing %s', (artifact) => { + const artifacts = { ...result().artifacts }; + delete artifacts[artifact]; + expect(() => assertValidDbtJobResult(result({ artifacts }), request())).toThrowError(SiaxError); + }); +}); + +describe('BI-P022 — layer dependencies are checked against the graph, not the convention', () => { + it("accepts the reference project's own edges", () => { + expect(() => + assertLayerDependenciesLegal([ + { from: 'models/staging/stg_sales_events.sql', to: 'models/sources/sources.yml' }, + { from: 'models/intermediate/int_sales_enriched.sql', to: 'models/staging/stg_sales_events.sql' }, + { from: 'models/marts/mart_sales_daily.sql', to: 'models/intermediate/int_sales_enriched.sql' }, + ]), + ).not.toThrow(); + }); + + it('refuses a mart reading a source directly — raw data reaching a dashboard', () => { + expect(() => + assertLayerDependenciesLegal([ + { from: 'models/marts/mart_sales_daily.sql', to: 'models/sources/sources.yml' }, + ]), + ).toThrowError(SiaxError); + }); + + it('refuses a mart reading staging — one layer skipped is still a skip', () => { + expect(() => + assertLayerDependenciesLegal([ + { from: 'models/marts/mart_sales_daily.sql', to: 'models/staging/stg_sales_events.sql' }, + ]), + ).toThrowError(SiaxError); + }); + + it('refuses a staging model reading a mart — an upward edge', () => { + expect(() => + assertLayerDependenciesLegal([ + { from: 'models/staging/stg_sales_events.sql', to: 'models/marts/mart_sales_daily.sql' }, + ]), + ).toThrowError(SiaxError); + }); + + it('allows a model reading another in the same layer', () => { + expect(() => + assertLayerDependenciesLegal([ + { from: 'models/staging/stg_sales_events.sql', to: 'models/staging/stg_customers.sql' }, + ]), + ).not.toThrow(); + }); +});