diff --git a/src/analytics/ingestion/analytics-ingestion-service.ts b/src/analytics/ingestion/analytics-ingestion-service.ts new file mode 100644 index 0000000..3899bc4 --- /dev/null +++ b/src/analytics/ingestion/analytics-ingestion-service.ts @@ -0,0 +1,143 @@ +/** + * src/analytics/ingestion/analytics-ingestion-service.ts + * ======================================================= + * BI-P129: one entry point that takes an `AnalyticsIngestionSpec` and lands + * rows in a canonical analytics tier, whichever provider supplies the + * connector. + * + * The sequence, and why it is this sequence: + * + * 1. plan — every gate checkable without touching the source + * (tier, physical home, licence, provider readiness). + * Cheap checks first: a spec that can never be legal + * should not open a session to find that out. + * 2. connect+discover— ask the source what it actually looks like. + * 3. landability — the discovered shape must satisfy the BI-P004 + * contract. This is the check that cannot be done at + * step 1, and doing it here is the whole reason the + * service exists rather than a bare `planX()` call. + * 4. run — hand the connector to the EXISTING P062 runner with + * a landing sink. The runner owns paging, envelopes, + * schema conformance, quality gating, the idempotency + * ledger and the durable checkpoint. None of that is + * re-implemented here. + * 5. disconnect — always, including on failure. + * + * Step 4 is worth spelling out because it is where a "new ingestion system" + * would normally get built by accident. There is no second runner. The + * analytics plane contributes a TARGET and four GATES; the highway is the one + * that already exists and already has crash/resume tests. + */ +import { SiaxError } from '../../platform/errors'; +import { IngestionRunner, type IngestionRunReport } from '../../ingestion/ingestion-runner'; +import type { IngestionRepository } from '../../repositories/ingestion-repository'; +import type { SchemaRegistry } from '../../schema/schema-registry'; +import type { LineageStore } from '../../lineage/lineage-store'; +import type { QualityEnforcementPolicy } from '../../quality/quality-policy'; +import type { RecordSchemaFields } from '../../schema/record-schema'; +import { + planAnalyticsIngestion, + type AnalyticsConnectorProvider, + type AnalyticsIngestionSpec, + type PlannedAnalyticsIngestion, +} from './connector-provider'; +import { assertDiscoveryLandable, landingSink, type LandingWriter } from './landing'; + +export interface AnalyticsIngestionDeps { + repo: IngestionRepository; + schemas: SchemaRegistry; + writer: LandingWriter; + lineage?: LineageStore; + /** Runtime environment name; "production" turns gate 4 on. */ + runtimeEnv: string; +} + +export interface AnalyticsIngestionResult { + plan: PlannedAnalyticsIngestion; + /** What discovery reported, per partition — recorded so a later drift is provable. */ + discovered: Record; + run: IngestionRunReport; +} + +export interface AnalyticsIngestionOptions { + run_id?: string; + page_size?: number; + max_pages?: number; + partition_key?: string; + quality_policy?: QualityEnforcementPolicy; + correlation_id?: string; +} + +export class AnalyticsIngestionService { + constructor(private readonly deps: AnalyticsIngestionDeps) {} + + /** Gates 1, 3 and 4 — no IO, no session. Safe to call from a UI pre-flight. */ + plan(provider: AnalyticsConnectorProvider, spec: AnalyticsIngestionSpec): PlannedAnalyticsIngestion { + return planAnalyticsIngestion(provider, spec, this.deps.runtimeEnv); + } + + async ingest( + provider: AnalyticsConnectorProvider, + spec: AnalyticsIngestionSpec, + options: AnalyticsIngestionOptions = {}, + ): Promise { + const plan = this.plan(provider, spec); + const runId = options.run_id ?? `run_${spec.source_id}_${Date.now()}`; + const connector = plan.connector; + + const discovered: Record = {}; + await connector.connect(spec.config, { + tenant_id: spec.tenant_id, + project_id: spec.project_id, + source_id: spec.source_id, + run_id: runId, + }); + try { + const discovery = await connector.discover(); + for (const partition of discovery.partitions) { + if (options.partition_key && partition.partition_key !== options.partition_key) continue; + discovered[partition.partition_key] = partition.schema; + } + // Gate 2 — the one that needs the source to have answered. + assertDiscoveryLandable(spec.target, discovered); + } finally { + // The runner opens its own session; this discovery session is ours to close. + await connector.disconnect(); + } + + const runner = new IngestionRunner(this.deps.repo, this.deps.schemas, this.deps.lineage); + const run = await runner.run({ + source_id: spec.source_id, + tenant_id: spec.tenant_id, + project_id: spec.project_id, + mode: spec.mode, + connector, + config: spec.config, + sink: landingSink({ + target: spec.target, + tenant_id: spec.tenant_id, + project_id: spec.project_id, + run_id: runId, + writer: this.deps.writer, + }), + run_id: runId, + page_size: options.page_size, + max_pages: options.max_pages, + partition_key: options.partition_key, + quality_policy: options.quality_policy, + output_ref: `0pt:dataset:${spec.target.dataset_id}`, + correlation_id: options.correlation_id, + }); + + if (run.status === 'failed') { + throw new SiaxError( + 'RETRYABLE_FAILURE', + '0pt.analytics.ingestion', + `ingestion run ${run.run_id} for dataset ${spec.target.dataset_id} failed: ` + + `${run.failure?.code ?? 'unknown'} ${run.failure?.detail ?? ''}`.trim(), + ); + } + + return { plan, discovered, run }; + } +} diff --git a/src/analytics/ingestion/connector-provider.ts b/src/analytics/ingestion/connector-provider.ts new file mode 100644 index 0000000..212501f --- /dev/null +++ b/src/analytics/ingestion/connector-provider.ts @@ -0,0 +1,442 @@ +/** + * src/analytics/ingestion/connector-provider.ts + * ============================================== + * BI-P129 — ANALYTICS DATA INGESTION. + * + * BI-P129 names four provider kinds for 0pt's `ConnectorProvider`: + * + * native — connectors 0pt implements and owns (P061 contract) + * dlt — the dlt Python framework (BI-P130) + * Meltano/Singer — the Singer tap ecosystem (BI-P131) + * Airbyte — "only optional where legal/license architecture accepts it" + * + * WHAT THIS FILE IS. It is the seam between 0pt's EXISTING ingestion highway + * (`src/connectors/connector.ts` P061 + `src/ingestion/ingestion-runner.ts` + * P062) and the ANALYTICS tier/contract model (BI-P003 tiers, BI-P004 + * `DatasetContract`). It deliberately does NOT define a second connector + * interface: a provider's job is to hand back a P061 `Connector`, and the + * runner that already exists then does the reading, checkpointing and + * quality gating. A parallel ingestion stack would be exactly the "do not + * define parallel models" failure this programme keeps paying for. + * + * WHAT IT ADDS, which the P061 stack has no opinion about: an ingestion run + * must LAND somewhere, and where it lands is an analytics contract question. + * Four gates, all fail-closed, all attacked in + * tests/analytics/analytics-ingestion-providers.test.ts: + * + * 1. assertLandingTierIngestible — source data may land in RAW, or in + * NORMALIZED with a declared normalization. It may NEVER land directly + * in CURATED or SERVING. BI-P003's rule ("no business dashboard reads + * arbitrary raw ingestion tables by default") is trivially defeated by + * relabelling the raw landing table SERVING; this gate is what stops + * that, because the tier is checked at the point of WRITE, not only at + * the point of read. + * + * 2. assertContractCoversDiscovery — every field the connector actually + * discovered must be declared in the target `DatasetContract`, with a + * compatible type. An UNDECLARED field is refused rather than landed: + * an undeclared column is an unclassified column, and an unclassified + * column carrying personal data is a GDPR incident that every freshness + * and quality check would still report green. + * + * 3. assertPhysicalHomeMatchesTier — a RAW/NORMALIZED landing goes to + * ST0RE Parquet/Iceberg (the shapes st0re#45 standardised), never + * straight into ClickHouse. Writing ingestion output into the serving + * engine skips the lake, and with it every replay, backfill and + * erasure path that depends on the files existing. + * + * 4. assertProviderRunnable — a provider whose own `providerReadiness()` + * says `verified: false` cannot execute in a production runtime without + * a recorded exception naming a human. This is the structural version + * of the honesty rule `0pt#35` set with MetricFlow: an unverified + * provider may exist, may be planned, may be tested — but it cannot + * quietly become production capability because someone forgot it was + * unverified. + * + * ARCHITECTURE BOUNDARY (BI programme §0): 0pt owns ingestion into analytics + * tiers. A product repo declares what the rows MEAN through + * `BIApplicationBinding`; it never runs its own ingestion into these tiers + * and never receives ClickHouse credentials. + */ +import { SiaxError } from '../../platform/errors'; +import type { Connector, ConnectorConfig, ValidationResult } from '../../connectors/connector'; +import type { RecordSchemaFields, FieldType } from '../../schema/record-schema'; +import type { IngestionMode } from '../../types/ingestion'; +import type { AnalyticsFieldType } from '../domain/entities'; +import type { AnalyticsTier } from '../domain/tiers'; +import type { DatasetContract } from '../domain/contract'; +import type { ProviderReadiness } from '../metrics/semantic-provider'; +import type { N0dRequiredRole } from '../../clients/n0d-client'; + +/** + * Deliberately the SAME `ProviderReadiness` shape BI-P029 introduced, not a + * second one. An estate with two readiness vocabularies has none. + */ +export type { ProviderReadiness }; + +// ── provider kinds ───────────────────────────────────────────────────────── + +export const CONNECTOR_PROVIDER_KINDS = ['native', 'dlt', 'singer', 'airbyte'] as const; +export type ConnectorProviderKind = (typeof CONNECTOR_PROVIDER_KINDS)[number]; + +/** + * BI-P129: "Airbyte only optional where legal/license architecture accepts + * it." That acceptance has NOT been given in this estate — Airbyte's server + * is Elastic License 2.0 (ELv2), which restricts offering the software as a + * managed service to third parties, and 0pt's whole purpose is offering + * managed data capability to customers. So `airbyte` is a declared kind with + * no implementation, and registering one is refused here rather than + * somewhere downstream. When a licence review says otherwise, this constant + * is the single place that changes. + */ +export const PROVIDER_KINDS_REQUIRING_LICENCE_ACCEPTANCE: readonly ConnectorProviderKind[] = ['airbyte']; + +// ── landing target ───────────────────────────────────────────────────────── + +/** Physical formats an analytics landing may use. st0re#45's shapes. */ +export const LANDING_FORMATS = ['parquet', 'iceberg'] as const; +export type LandingFormat = (typeof LANDING_FORMATS)[number]; + +/** Where an ingestion run's output physically goes. */ +export interface LandingPhysical { + format: LandingFormat; + /** ST0RE lakehouse namespace. */ + namespace: string; + table_name: string; +} + +export interface AnalyticsLandingTarget { + dataset_id: string; + tier: AnalyticsTier; + /** The BI-P004 contract the landed rows must satisfy. */ + contract: DatasetContract; + physical: LandingPhysical; + /** + * Required when `tier` is NORMALIZED: what normalization was applied + * between the source bytes and the landed rows. Free text, recorded — an + * empty or missing value means the dataset is really RAW wearing a + * NORMALIZED label, and is refused. + */ + normalization?: string; +} + +/** One ingestion the analytics plane is asked to perform. */ +export interface AnalyticsIngestionSpec { + source_id: string; + tenant_id: string; + project_id: string; + mode: IngestionMode; + provider: ConnectorProviderKind; + /** Provider-specific connector selector (a native connector id, a tap name, a dlt source). */ + connector_ref: string; + config: ConnectorConfig; + target: AnalyticsLandingTarget; + /** + * Records the human who accepted running an UNVERIFIED provider in a + * production runtime. Absent = an unverified provider is refused. + */ + unverified_provider_exception?: string; +} + +// ── execution plan ───────────────────────────────────────────────────────── + +export type ProviderRuntime = 'in_process' | 'n0d'; + +/** + * Where a provider's work actually runs. BI-P130 says dlt executes on N0D; + * this is the type that carries that decision instead of a comment. + */ +export interface ProviderExecutionPlan { + runtime: ProviderRuntime; + /** N0D job_type; set iff runtime is "n0d". */ + job_type?: string; + required_role?: N0dRequiredRole; + /** Named profile from N0D_RESOURCE_PROFILES; set iff runtime is "n0d". */ + resource_profile?: string; + /** Operator-facing reason, in plain words. */ + reason: string; +} + +// ── the port ─────────────────────────────────────────────────────────────── + +export interface AnalyticsConnectorProvider { + readonly provider: ConnectorProviderKind; + /** What this provider has and has NOT demonstrated. Never optimistic. */ + providerReadiness(): ProviderReadiness; + /** Pure, no IO, never throws — the P061 `validate` posture, at spec level. */ + validateSpec(spec: AnalyticsIngestionSpec): ValidationResult; + /** The P061 connector this spec ingests through. */ + connectorFor(spec: AnalyticsIngestionSpec): Connector; + executionPlan(spec: AnalyticsIngestionSpec): ProviderExecutionPlan; +} + +// ── gate 1: landing tier ─────────────────────────────────────────────────── + +/** Tiers an ingestion run may write into at all. */ +export const INGESTIBLE_TIERS: readonly AnalyticsTier[] = ['RAW', 'NORMALIZED']; + +export function assertLandingTierIngestible(target: AnalyticsLandingTarget): void { + if (!INGESTIBLE_TIERS.includes(target.tier)) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `ingestion may not land directly in ${target.tier} (dataset ${target.dataset_id}): ` + + `CURATED and SERVING are DERIVED tiers, built from a landed RAW/NORMALIZED dataset. ` + + `Landing source data there relabels raw ingestion as dashboard-readable and defeats BI-P003.`, + ); + } + if (target.tier === 'NORMALIZED' && !(target.normalization ?? '').trim()) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `dataset ${target.dataset_id} lands in NORMALIZED but declares no normalization — ` + + `that is a RAW dataset wearing a NORMALIZED label (BI-P003)`, + ); + } +} + +// ── gate 2: contract covers what was discovered ──────────────────────────── + +/** + * How a connector's wire-level field type maps onto the analytics type + * vocabulary. Conservative on purpose: a connector's `number` may be an + * integer or a float, so it satisfies EITHER analytics numeric type, but + * nothing else. Anything not listed here is a mismatch, not a guess. + */ +const CONNECTOR_TYPE_SATISFIES: Record = { + string: ['string', 'decimal', 'date', 'timestamp'], + number: ['int64', 'float64', 'decimal'], + boolean: ['bool'], + timestamp: ['timestamp', 'date'], + object: ['json'], + array: ['json'], +}; + +export interface ContractCoverageProblem { + code: 'undeclared_field' | 'type_mismatch' | 'missing_required_field' | 'nullability_mismatch'; + field: string; + detail: string; +} + +/** + * Compare what the connector says the source LOOKS LIKE against what the + * contract says the dataset IS. Returns every problem; never throws. + */ +export function contractCoverageProblems( + contract: DatasetContract, + discovered: RecordSchemaFields, +): ContractCoverageProblem[] { + const problems: ContractCoverageProblem[] = []; + const byName = new Map(contract.fields.map((f) => [f.name, f])); + + for (const [name, spec] of Object.entries(discovered)) { + const field = byName.get(name); + if (!field) { + problems.push({ + code: 'undeclared_field', + field: name, + detail: + `source field "${name}" is not declared in contract ${contract.dataset_id}@${contract.version}; ` + + `an undeclared field is an unclassified field and is refused, not landed`, + }); + continue; + } + const satisfies = CONNECTOR_TYPE_SATISFIES[spec.type] ?? []; + if (!satisfies.includes(field.type)) { + problems.push({ + code: 'type_mismatch', + field: name, + detail: `source field "${name}" is ${spec.type}; contract declares ${field.type}`, + }); + } + // `FieldSpec.required` is the connector-side vocabulary; the contract + // speaks `nullable`. required:false === nullable:true. + if (spec.required === false && field.nullable === false) { + problems.push({ + code: 'nullability_mismatch', + field: name, + detail: `source field "${name}" is optional at the source; contract declares it NOT NULL`, + }); + } + } + + for (const field of contract.fields) { + if (field.nullable === false && !(field.name in discovered)) { + problems.push({ + code: 'missing_required_field', + field: field.name, + detail: `contract requires non-null field "${field.name}" but the source does not produce it`, + }); + } + } + return problems; +} + +export function assertContractCoversDiscovery( + contract: DatasetContract, + discovered: RecordSchemaFields, +): void { + const problems = contractCoverageProblems(contract, discovered); + if (problems.length > 0) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `source shape does not satisfy dataset contract ${contract.dataset_id}@${contract.version}: ` + + problems.map((p) => `[${p.code}] ${p.detail}`).join('; '), + ); + } +} + +// ── gate 3: physical home ────────────────────────────────────────────────── + +export function assertPhysicalHomeMatchesTier(target: AnalyticsLandingTarget): void { + if (!(LANDING_FORMATS as readonly string[]).includes(target.physical.format)) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `landing format "${target.physical.format}" is not a lakehouse format; ` + + `RAW/NORMALIZED land as Parquet or Iceberg in ST0RE (BI-P003, st0re#45)`, + ); + } + if (!target.physical.namespace.trim() || !target.physical.table_name.trim()) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `landing target for ${target.dataset_id} is missing a ST0RE namespace/table name`, + ); + } +} + +// ── gate 4: unverified providers may not run in production ───────────────── + +export function assertProviderRunnable( + provider: AnalyticsConnectorProvider, + spec: AnalyticsIngestionSpec, + runtimeEnv: string, +): void { + const readiness = provider.providerReadiness(); + if (readiness.verified) return; + const isProduction = runtimeEnv === 'production'; + const exception = (spec.unverified_provider_exception ?? '').trim(); + if (!isProduction) return; + if (exception.length === 0) { + throw new SiaxError( + 'CAPABILITY_UNAVAILABLE', + '0pt.analytics.ingestion', + `provider "${provider.provider}" reports verified: false ` + + `(open: ${readiness.open.join('; ')}) and may not run in production ` + + `without a recorded unverified_provider_exception`, + ); + } +} + +// ── planning ─────────────────────────────────────────────────────────────── + +export interface PlannedAnalyticsIngestion { + spec: AnalyticsIngestionSpec; + connector: Connector; + execution: ProviderExecutionPlan; + readiness: ProviderReadiness; + /** The tier the rows land in — echoed so a caller cannot claim a different one later. */ + landing_tier: AnalyticsTier; +} + +/** + * Run every gate that can be checked WITHOUT touching the source, then hand + * back the connector. `assertContractCoversDiscovery` is deliberately NOT + * here: discovery requires a live session, so it runs in `landingSink`'s + * caller / `AnalyticsIngestionService` once `discover()` has answered. + */ +export function planAnalyticsIngestion( + provider: AnalyticsConnectorProvider, + spec: AnalyticsIngestionSpec, + runtimeEnv: string, +): PlannedAnalyticsIngestion { + if (provider.provider !== spec.provider) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `spec asks for provider "${spec.provider}" but was given "${provider.provider}"`, + ); + } + if (PROVIDER_KINDS_REQUIRING_LICENCE_ACCEPTANCE.includes(spec.provider)) { + throw new SiaxError( + 'CAPABILITY_UNAVAILABLE', + '0pt.analytics.ingestion', + `provider "${spec.provider}" requires a licence/architecture acceptance this estate has not given ` + + `(BI-P129: "Airbyte only optional where legal/license architecture accepts it")`, + ); + } + const verdict = provider.validateSpec(spec); + if (!verdict.valid) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `provider "${provider.provider}" rejected the spec: ${verdict.errors.join('; ')}`, + ); + } + assertLandingTierIngestible(spec.target); + assertPhysicalHomeMatchesTier(spec.target); + assertProviderRunnable(provider, spec, runtimeEnv); + + return { + spec, + connector: provider.connectorFor(spec), + execution: provider.executionPlan(spec), + readiness: provider.providerReadiness(), + landing_tier: spec.target.tier, + }; +} + +// ── registry ─────────────────────────────────────────────────────────────── + +/** + * The set of providers a runtime actually has. Registration is explicit: + * there is no discovery-by-filename, because a provider appearing in the + * registry because a file exists is how an unreviewed integration becomes + * live. + */ +export class ConnectorProviderRegistry { + private readonly providers = new Map(); + + register(provider: AnalyticsConnectorProvider): this { + if (this.providers.has(provider.provider)) { + throw new SiaxError( + 'CONFLICT', + '0pt.analytics.ingestion', + `provider "${provider.provider}" is already registered`, + ); + } + if (PROVIDER_KINDS_REQUIRING_LICENCE_ACCEPTANCE.includes(provider.provider)) { + throw new SiaxError( + 'CAPABILITY_UNAVAILABLE', + '0pt.analytics.ingestion', + `provider "${provider.provider}" may not be registered: no licence/architecture acceptance exists ` + + `for it in this estate (BI-P129)`, + ); + } + this.providers.set(provider.provider, provider); + return this; + } + + get(kind: ConnectorProviderKind): AnalyticsConnectorProvider { + const p = this.providers.get(kind); + if (!p) { + throw new SiaxError( + 'NOT_FOUND', + '0pt.analytics.ingestion', + `no connector provider registered for kind "${kind}"`, + ); + } + return p; + } + + has(kind: ConnectorProviderKind): boolean { + return this.providers.has(kind); + } + + /** Readiness of every registered provider — the operator's one-screen answer. */ + readinessReport(): ProviderReadiness[] { + return [...this.providers.values()].map((p) => p.providerReadiness()); + } +} diff --git a/src/analytics/ingestion/landing.ts b/src/analytics/ingestion/landing.ts new file mode 100644 index 0000000..9b56ff3 --- /dev/null +++ b/src/analytics/ingestion/landing.ts @@ -0,0 +1,160 @@ +/** + * src/analytics/ingestion/landing.ts + * =================================== + * BI-P129: the write side. An ingestion run's envelopes have to become FILES + * in a canonical tier, and this is the only place in the analytics plane that + * decides what those files look like. + * + * WHERE THE BOUNDARY IS. 0pt does not own object storage. ST0RE owns the + * lakehouse (namespaces, Iceberg tables, snapshots — `st0re#45`), and 0pt's + * existing `src/services/iceberg-write-service.ts` is already the client for + * it. So `LandingWriter` is a PORT, not an implementation: it names exactly + * what the analytics plane needs from a lake writer (append a batch to one + * tier-owned table, tell me what landed) and nothing more. The default + * implementation here is in-memory and is used by tests and by dry runs; the + * ST0RE-backed one is the existing service, injected. + * + * WHAT THIS FILE ENFORCES, and why each rule is not decoration: + * + * - Every landed batch carries `tier` and `dataset_id`. A batch of rows + * with no tier is a batch nobody can decide the read rules for later. + * + * - `assertContractCoversDiscovery` (BI-P129 gate 2) runs HERE, at the + * point discovery has actually answered, not at plan time when the + * source has not been asked yet. A gate that runs before the fact it + * checks is available is a gate that checks nothing. + * + * - Deletes are LANDED, not dropped. `IngestionEnvelope` with + * `operation: "delete"` carries `payload: null`; a landing that silently + * skipped it would produce exactly the failure this programme has already + * paid for once — an aggregate that never sees a DELETE and overstates a + * number while every freshness check reports green. Staleness is not + * correctness, and a delete that never lands is not staleness at all, + * it is permanent wrongness. + * + * - The batch records `envelope_id`s, so a replay is detectable downstream. + * The runner's guarantee is at-least-once (documented in + * src/ingestion/ingestion-runner.ts); a landing that pretended otherwise + * would be lying about the pipeline it sits in. + */ +import { SiaxError } from '../../platform/errors'; +import type { IngestionEnvelope } from '../../types/ingestion'; +import type { RecordSchemaFields } from '../../schema/record-schema'; +import type { AnalyticsTier } from '../domain/tiers'; +import { + assertContractCoversDiscovery, + type AnalyticsLandingTarget, + type LandingFormat, +} from './connector-provider'; + +/** One physical batch appended to a tier-owned table. */ +export interface LandedBatch { + dataset_id: string; + tier: AnalyticsTier; + format: LandingFormat; + namespace: string; + table_name: string; + tenant_id: string; + project_id: string; + run_id: string; + partition_key: string; + /** Rows written, including tombstones. */ + row_count: number; + /** Rows whose payload was null (CDC deletes). Counted separately, never dropped. */ + tombstone_count: number; + envelope_ids: string[]; +} + +/** + * The port. One method: append a batch. Deliberately narrow — every extra + * method here is a piece of ST0RE's contract leaking into 0pt's. + */ +export interface LandingWriter { + append(batch: LandedBatch): Promise; +} + +/** Test/dry-run writer. Keeps everything it was given so a test can assert on it. */ +export class InMemoryLandingWriter implements LandingWriter { + readonly batches: LandedBatch[] = []; + async append(batch: LandedBatch): Promise { + this.batches.push(batch); + } + get rowsLanded(): number { + return this.batches.reduce((n, b) => n + b.row_count, 0); + } + get tombstonesLanded(): number { + return this.batches.reduce((n, b) => n + b.tombstone_count, 0); + } +} + +export interface LandingSinkInput { + target: AnalyticsLandingTarget; + tenant_id: string; + project_id: string; + run_id: string; + writer: LandingWriter; +} + +/** + * Build an `IngestionSink` (the runner's own type) that lands envelopes into + * the declared analytics tier. + * + * Note the sink is per-run: the runner commits its checkpoint only AFTER + * this resolves, so a writer that throws leaves the page unconsumed and the + * next run re-reads it. That ordering belongs to the runner and is not + * re-implemented here — it is relied upon. + */ +export function landingSink(input: LandingSinkInput): (envelopes: IngestionEnvelope[]) => Promise { + return async (envelopes: IngestionEnvelope[]): Promise => { + if (envelopes.length === 0) return; + const partitions = new Set(envelopes.map((e) => e.partition_key)); + for (const partition of partitions) { + const forPartition = envelopes.filter((e) => e.partition_key === partition); + await input.writer.append({ + dataset_id: input.target.dataset_id, + tier: input.target.tier, + format: input.target.physical.format, + namespace: input.target.physical.namespace, + table_name: input.target.physical.table_name, + tenant_id: input.tenant_id, + project_id: input.project_id, + run_id: input.run_id, + partition_key: partition, + row_count: forPartition.length, + tombstone_count: forPartition.filter((e) => e.payload === null).length, + envelope_ids: forPartition.map((e) => e.envelope_id), + }); + } + }; +} + +/** + * BI-P129 gate 2, applied once the connector's `discover()` has answered. + * Called by `AnalyticsIngestionService`; exported so an operator pre-flight + * can run it without starting a run. + */ +export function assertDiscoveryLandable( + target: AnalyticsLandingTarget, + discoveredByPartition: Record, +): void { + const partitions = Object.keys(discoveredByPartition); + if (partitions.length === 0) { + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `source for dataset ${target.dataset_id} discovered no partitions; there is nothing to land`, + ); + } + for (const partition of partitions) { + try { + assertContractCoversDiscovery(target.contract, discoveredByPartition[partition]); + } catch (err) { + const detail = err instanceof Error ? err.message : String(err); + throw new SiaxError( + 'INVALID_REQUEST', + '0pt.analytics.ingestion', + `partition "${partition}": ${detail}`, + ); + } + } +} diff --git a/src/analytics/ingestion/native-provider.ts b/src/analytics/ingestion/native-provider.ts new file mode 100644 index 0000000..c8abd8c --- /dev/null +++ b/src/analytics/ingestion/native-provider.ts @@ -0,0 +1,97 @@ +/** + * src/analytics/ingestion/native-provider.ts + * =========================================== + * BI-P129, provider kind `native`: connectors 0pt implements and owns. + * + * There is nothing to integrate here — the connectors already exist and are + * already conformance-tested (`src/connectors/conformance.ts`, P061). This + * provider's whole job is to select one and report, honestly, that it is + * verified: these connectors run in this repo's CI, in-process, with no + * third-party runtime and no unreviewed dependency. + * + * The two shipped natives are deliberately heterogeneous (that was P061's + * point and it is BI-P129's DoD too — "two real source types"): + * + * http_json — remote HTTP API, opaque server-issued page token, health + * is a network round trip, record count unknown until the + * server says it is done. + * jsonl_file — local file drop, cursor is a line offset the connector + * computes itself, health is a stat(), record count exact, + * and it emits real CDC deletes. + * + * They share no code but the contract, so "both land in a canonical tier" + * is a statement about the SEAM, not about one connector. + */ +import { HttpJsonConnector, type FetchImpl } from '../../connectors/http-json-connector'; +import { JsonlFileConnector } from '../../connectors/jsonl-file-connector'; +import type { Connector, ValidationResult } from '../../connectors/connector'; +import type { + AnalyticsConnectorProvider, + AnalyticsIngestionSpec, + ProviderExecutionPlan, + ProviderReadiness, +} from './connector-provider'; + +/** Native connector ids this provider can build. */ +export const NATIVE_CONNECTOR_IDS = ['http_json', 'jsonl_file'] as const; +export type NativeConnectorId = (typeof NATIVE_CONNECTOR_IDS)[number]; + +export class NativeConnectorProvider implements AnalyticsConnectorProvider { + readonly provider = 'native' as const; + + /** + * Optional injection point so a caller can supply an `HttpJsonConnector` + * with a specific `fetchImpl` (the connector's own documented test seam). + * With nothing supplied, a real connector talking to the configured URL is + * built — this is not a mock fallback. + */ + constructor(private readonly fetchImpl?: FetchImpl) {} + + providerReadiness(): ProviderReadiness { + return { + provider: this.provider, + pinned_version: 'in-repo', + verified: true, + proven: [ + 'both native connectors pass the P061 conformance suite (tests/p0/connector-conformance.test.ts)', + 'both land into a canonical analytics tier through the P062 ingestion runner ' + + '(tests/analytics/analytics-ingestion-providers.test.ts)', + 'no third-party runtime and no unreviewed dependency is involved', + ], + open: [ + 'only http_json and jsonl_file exist natively; every other source type needs dlt/Singer or a new native connector', + ], + }; + } + + validateSpec(spec: AnalyticsIngestionSpec): ValidationResult { + const errors: string[] = []; + if (!(NATIVE_CONNECTOR_IDS as readonly string[]).includes(spec.connector_ref)) { + errors.push( + `unknown native connector "${spec.connector_ref}" (have: ${NATIVE_CONNECTOR_IDS.join(', ')})`, + ); + return { valid: false, errors }; + } + // Delegate the config check to the connector itself — it is the authority + // on its own config, and its `validate` is pure and never throws. + const connector = this.connectorFor(spec); + const verdict = connector.validate(spec.config); + errors.push(...verdict.errors); + return { valid: errors.length === 0, errors }; + } + + connectorFor(spec: AnalyticsIngestionSpec): Connector { + if (spec.connector_ref === 'http_json') { + return this.fetchImpl ? new HttpJsonConnector(this.fetchImpl) : new HttpJsonConnector(); + } + if (spec.connector_ref === 'jsonl_file') return new JsonlFileConnector(); + throw new Error(`unknown native connector "${spec.connector_ref}"`); + } + + executionPlan(_spec: AnalyticsIngestionSpec): ProviderExecutionPlan { + return { + runtime: 'in_process', + reason: 'native connectors are Node code in this repo; they need no separate runtime and no N0D job', + }; + } +} diff --git a/src/connectors/http-json-connector.ts b/src/connectors/http-json-connector.ts index cb11627..cce7384 100644 --- a/src/connectors/http-json-connector.ts +++ b/src/connectors/http-json-connector.ts @@ -66,7 +66,7 @@ export interface HttpJsonConnectorConfig extends ConnectorConfig { timeout_ms?: number; } -type FetchImpl = (url: string, init: RequestInit) => Promise; +export type FetchImpl = (url: string, init: RequestInit) => Promise; /** Wire shape this connector expects from the upstream endpoint. */ interface UpstreamPage { diff --git a/tests/analytics/analytics-ingestion-providers.test.ts b/tests/analytics/analytics-ingestion-providers.test.ts new file mode 100644 index 0000000..a42db7d --- /dev/null +++ b/tests/analytics/analytics-ingestion-providers.test.ts @@ -0,0 +1,521 @@ +/** + * BI-P129 — ANALYTICS DATA INGESTION. + * + * DoD: "Two real source types ingest into canonical analytics tiers." + * + * The two source types here are the REAL P061 connectors, sharing no code but + * the contract: a JSONL file on actual disk, and an HTTP JSON API served by a + * real `fetch` implementation over a real request/response cycle. Both go + * through the REAL P062 ingestion runner and the REAL landing sink into a + * declared tier. Nothing about the tier or the contract is stubbed. + * + * Every gate is attacked TWICE: once with the guard in place (it must + * refuse), and once in a counterfactual that removes only the guard (it must + * succeed). A guard whose counterfactual also fails is not a guard — the + * behaviour was coming from somewhere else, and the estate has shipped three + * of those this month. + */ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { promises as fs } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { InMemoryIngestionRepository } from '@/src/repositories/in-memory-ingestion-repository'; +import { SchemaRegistry } from '@/src/schema/schema-registry'; +import { LineageStore } from '@/src/lineage/lineage-store'; +import { SiaxError } from '@/src/platform/errors'; +import type { RecordSchemaFields } from '@/src/schema/record-schema'; +import { + ConnectorProviderRegistry, + assertContractCoversDiscovery, + assertLandingTierIngestible, + assertPhysicalHomeMatchesTier, + assertProviderRunnable, + contractCoverageProblems, + planAnalyticsIngestion, + type AnalyticsConnectorProvider, + type AnalyticsIngestionSpec, + type AnalyticsLandingTarget, +} from '@/src/analytics/ingestion/connector-provider'; +import { NativeConnectorProvider } from '@/src/analytics/ingestion/native-provider'; +import { InMemoryLandingWriter, assertDiscoveryLandable } from '@/src/analytics/ingestion/landing'; +import { AnalyticsIngestionService } from '@/src/analytics/ingestion/analytics-ingestion-service'; +import type { DatasetContract } from '@/src/analytics/domain/contract'; + +// ── fixtures ─────────────────────────────────────────────────────────────── + +const SOURCE_SCHEMA: RecordSchemaFields = { + org_number: { type: 'string', required: true }, + name: { type: 'string', required: true }, + employees: { type: 'number', required: false }, +}; + +const CONTRACT: DatasetContract = { + dataset_id: 'ds_companies_raw', + version: '1.0.0', + tier: 'RAW', + fields: [ + { name: 'org_number', type: 'string', nullable: false, role: 'identifier', primary_identifier: true }, + { name: 'name', type: 'string', nullable: false, role: 'dimension' }, + { name: 'employees', type: 'int64', nullable: true, role: 'measure' }, + ], + primary_identifiers: ['org_number'], + dimensions: ['name'], + measures: ['employees'], + freshness_slo_seconds: 86_400, + retention_days: 3650, + pii_classification: 'none', + source_lineage: ['bolagsverket'], + quality_rules: [], + allowed_purposes: ['analytics'], +}; + +function target(overrides: Partial = {}): AnalyticsLandingTarget { + return { + dataset_id: 'ds_companies_raw', + tier: 'RAW', + contract: CONTRACT, + physical: { format: 'parquet', namespace: 'analytics_raw', table_name: 'companies' }, + ...overrides, + }; +} + +const ROWS = Array.from({ length: 9 }, (_, i) => ({ + org_number: `55600${i}-000${i}`, + name: `Bolag ${i} AB`, + employees: (i + 1) * 10, +})); + +const tmpDir = path.join(os.tmpdir(), `opt-bi129-${process.pid}`); +const filePath = path.join(tmpDir, 'companies.jsonl'); + +beforeAll(async () => { + await fs.mkdir(tmpDir, { recursive: true }); + await fs.writeFile(filePath, ROWS.map((r) => JSON.stringify(r)).join('\n') + '\n', 'utf8'); +}); +afterAll(async () => { + await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => undefined); +}); + +/** + * A real HTTP JSON source: an actual fetch implementation returning real + * `Response` objects, paging by an opaque server token. The connector under + * test does a genuine request/response cycle against it — the seam is the + * transport, not the connector. + */ +function httpSourceFetch(): (url: string, init: RequestInit) => Promise { + return async (url: string) => { + const parsed = new URL(url); + if (parsed.pathname.endsWith('/health')) { + return new Response(JSON.stringify({ ok: true }), { status: 200 }); + } + const cursor = parsed.searchParams.get('cursor'); + const limit = Number(parsed.searchParams.get('limit') ?? '100'); + const offset = cursor ? Number(cursor) : 0; + const page = ROWS.slice(offset, offset + limit); + const nextOffset = offset + page.length; + return new Response( + JSON.stringify({ + records: page, + next_cursor: nextOffset < ROWS.length ? String(nextOffset) : undefined, + }), + { status: 200, headers: { 'content-type': 'application/json' } }, + ); + }; +} + +function jsonlSpec(overrides: Partial = {}): AnalyticsIngestionSpec { + return { + source_id: 'src_jsonl', + tenant_id: 't_test', + project_id: 'p_test', + mode: 'batch', + provider: 'native', + connector_ref: 'jsonl_file', + config: { + file_path: filePath, + partition_key: 'companies', + schema: SOURCE_SCHEMA, + record_key_field: 'org_number', + }, + target: target(), + ...overrides, + }; +} + +function httpSpec(overrides: Partial = {}): AnalyticsIngestionSpec { + return { + source_id: 'src_http', + tenant_id: 't_test', + project_id: 'p_test', + mode: 'batch', + provider: 'native', + connector_ref: 'http_json', + config: { + base_url: 'http://source.invalid', + records_path: '/v1/companies', + partition_key: 'companies', + schema: SOURCE_SCHEMA, + record_key_field: 'org_number', + record_cursor_field: 'org_number', + health_path: '/health', + }, + target: target({ dataset_id: 'ds_companies_raw' }), + ...overrides, + }; +} + +let repo: InMemoryIngestionRepository; +let writer: InMemoryLandingWriter; +let service: AnalyticsIngestionService; + +function buildService(runtimeEnv = 'test'): AnalyticsIngestionService { + repo = new InMemoryIngestionRepository(); + writer = new InMemoryLandingWriter(); + return new AnalyticsIngestionService({ + repo, + schemas: new SchemaRegistry(repo), + lineage: new LineageStore(repo), + writer, + runtimeEnv, + }); +} + +beforeEach(() => { + service = buildService(); +}); + +// ── DoD: two real source types land in a canonical tier ──────────────────── + +describe('BI-P129 DoD — two real source types ingest into canonical analytics tiers', () => { + it('source type 1 (JSONL file on disk) lands every row in the declared RAW tier', async () => { + const result = await service.ingest(new NativeConnectorProvider(), jsonlSpec(), { page_size: 4 }); + + expect(result.run.status).toBe('succeeded'); + expect(writer.rowsLanded).toBe(ROWS.length); + expect(writer.batches.every((b) => b.tier === 'RAW')).toBe(true); + expect(writer.batches.every((b) => b.dataset_id === 'ds_companies_raw')).toBe(true); + expect(writer.batches.every((b) => b.format === 'parquet')).toBe(true); + expect(writer.batches.every((b) => b.namespace === 'analytics_raw')).toBe(true); + // Paged: 4 + 4 + 1, so the tier landing is exercised across batches. + expect(writer.batches.length).toBeGreaterThan(1); + }); + + it('source type 2 (HTTP JSON API, opaque page token) lands every row in the same tier', async () => { + const provider = new NativeConnectorProvider(httpSourceFetch()); + const result = await service.ingest(provider, httpSpec(), { page_size: 4 }); + + expect(result.run.status).toBe('succeeded'); + expect(writer.rowsLanded).toBe(ROWS.length); + expect(writer.batches.every((b) => b.tier === 'RAW')).toBe(true); + expect(new Set(writer.batches.flatMap((b) => b.envelope_ids)).size).toBe(ROWS.length); + }); + + it('the two source types share no connector code — different ids, versions and cursor vocabulary', async () => { + const nativeProvider = new NativeConnectorProvider(httpSourceFetch()); + const jsonl = nativeProvider.connectorFor(jsonlSpec()); + const http = nativeProvider.connectorFor(httpSpec()); + expect(jsonl.connector_id).toBe('jsonl_file'); + expect(http.connector_id).toBe('http_json'); + expect(jsonl.connector_version).not.toBe(http.connector_version); + expect(Object.getPrototypeOf(jsonl)).not.toBe(Object.getPrototypeOf(http)); + }); +}); + +// ── gate 1: landing tier ─────────────────────────────────────────────────── + +describe('BI-P129 gate 1 — ingestion may not land directly in a derived tier', () => { + it('WITH the guard: landing in SERVING is refused', () => { + expect(() => assertLandingTierIngestible(target({ tier: 'SERVING' }))).toThrow( + /may not land directly in SERVING/, + ); + expect(() => assertLandingTierIngestible(target({ tier: 'CURATED' }))).toThrow( + /may not land directly in CURATED/, + ); + }); + + it('WITHOUT the guard: the same spec lands source rows into a dashboard-readable tier', async () => { + // Counterfactual — the service is bypassed and the sink is called directly, + // which is exactly what a provider that skipped the gate would do. + const { landingSink } = await import('@/src/analytics/ingestion/landing'); + const bypassWriter = new InMemoryLandingWriter(); + const sink = landingSink({ + target: target({ tier: 'SERVING' }), + tenant_id: 't_test', + project_id: 'p_test', + run_id: 'run_bypass', + writer: bypassWriter, + }); + await sink([ + { + envelope_id: 'e1', + source_id: 'src_jsonl', + tenant_id: 't_test', + project_id: 'p_test', + mode: 'batch', + operation: 'upsert', + partition_key: 'companies', + record_key: '1', + schema_version: 1, + sequence: 1, + ingested_at: new Date().toISOString(), + payload: { org_number: '1', name: 'x' }, + provenance: { connector_id: 'jsonl_file', connector_version: 'v1', run_id: 'run_bypass' }, + }, + ]); + // Raw source rows now sit in SERVING, which BI-P003 makes readable by + // every dashboard with no exception recorded anywhere. That is the harm. + expect(bypassWriter.batches[0].tier).toBe('SERVING'); + expect(bypassWriter.rowsLanded).toBe(1); + }); + + it('NORMALIZED without a declared normalization is a RAW dataset wearing a label', () => { + expect(() => assertLandingTierIngestible(target({ tier: 'NORMALIZED' }))).toThrow( + /declares no normalization/, + ); + expect(() => + assertLandingTierIngestible( + target({ tier: 'NORMALIZED', normalization: 'typed + deduped on org_number' }), + ), + ).not.toThrow(); + // An all-whitespace declaration is not a declaration. + expect(() => assertLandingTierIngestible(target({ tier: 'NORMALIZED', normalization: ' ' }))).toThrow( + /declares no normalization/, + ); + }); + + it('the service refuses a SERVING spec before it opens a session to the source', async () => { + await expect( + service.ingest(new NativeConnectorProvider(), jsonlSpec({ target: target({ tier: 'SERVING' }) })), + ).rejects.toThrow(/may not land directly in SERVING/); + expect(writer.batches).toHaveLength(0); + }); +}); + +// ── gate 2: contract covers what was discovered ──────────────────────────── + +describe('BI-P129 gate 2 — an undeclared source field is refused, not landed', () => { + it('WITH the guard: a new undeclared column stops the run', () => { + const drifted: RecordSchemaFields = { + ...SOURCE_SCHEMA, + contact_email: { type: 'string', required: false }, + }; + expect(() => assertContractCoversDiscovery(CONTRACT, drifted)).toThrow(/undeclared_field/); + expect(() => assertContractCoversDiscovery(CONTRACT, drifted)).toThrow(/contact_email/); + }); + + it('the harm being prevented is named: an undeclared column is an unclassified column', () => { + const problems = contractCoverageProblems(CONTRACT, { + ...SOURCE_SCHEMA, + personnummer: { type: 'string', required: false }, + }); + expect(problems).toHaveLength(1); + expect(problems[0].code).toBe('undeclared_field'); + expect(problems[0].detail).toMatch(/unclassified/); + }); + + it('WITHOUT the guard: the drifted source lands and the PII column is invisible to the contract', async () => { + const driftedFile = path.join(tmpDir, 'drifted.jsonl'); + await fs.writeFile( + driftedFile, + ROWS.map((r) => JSON.stringify({ ...r, personnummer: '19800101-1234' })).join('\n') + '\n', + 'utf8', + ); + const driftedSchema: RecordSchemaFields = { + ...SOURCE_SCHEMA, + personnummer: { type: 'string', required: false }, + }; + // Counterfactual: the runner alone (no analytics gate) happily lands it. + const bypass = buildService(); + const bypassWriter = writer; + const { IngestionRunner } = await import('@/src/ingestion/ingestion-runner'); + const { JsonlFileConnector } = await import('@/src/connectors/jsonl-file-connector'); + const { landingSink } = await import('@/src/analytics/ingestion/landing'); + const runner = new IngestionRunner(repo, new SchemaRegistry(repo)); + const landed: Array | null> = []; + const report = await runner.run({ + source_id: 'src_drift', + tenant_id: 't_test', + project_id: 'p_test', + mode: 'batch', + connector: new JsonlFileConnector(), + config: { + file_path: driftedFile, + partition_key: 'companies', + schema: driftedSchema, + record_key_field: 'org_number', + }, + sink: async (envelopes) => { + for (const e of envelopes) landed.push(e.payload); + await landingSink({ + target: target(), + tenant_id: 't_test', + project_id: 'p_test', + run_id: 'run_drift', + writer: bypassWriter, + })(envelopes); + }, + }); + expect(report.status).toBe('succeeded'); + expect(landed).toHaveLength(ROWS.length); + // The personal number is now sitting in a RAW table whose contract says + // pii_classification: "none". Nothing else in the pipeline notices. + expect(landed[0]).toHaveProperty('personnummer'); + expect(CONTRACT.pii_classification).toBe('none'); + void bypass; + }); + + it('a type mismatch and a missing required field are both refused', () => { + expect(() => + assertContractCoversDiscovery(CONTRACT, { + ...SOURCE_SCHEMA, + employees: { type: 'string', required: false }, + }), + ).toThrow(/type_mismatch/); + const { org_number: _dropped, ...withoutKey } = SOURCE_SCHEMA; + expect(() => assertContractCoversDiscovery(CONTRACT, withoutKey)).toThrow(/missing_required_field/); + }); + + it('a source field that is optional where the contract says NOT NULL is refused', () => { + const problems = contractCoverageProblems(CONTRACT, { + ...SOURCE_SCHEMA, + name: { type: 'string', required: false }, + }); + expect(problems.map((p) => p.code)).toContain('nullability_mismatch'); + }); + + it('discovery with no partitions is refused rather than reported as a clean empty run', () => { + expect(() => assertDiscoveryLandable(target(), {})).toThrow(/discovered no partitions/); + }); + + it('the live path runs gate 2 against what the connector actually discovered', async () => { + // The contract is fine for the file; narrow the CONTRACT instead so the + // real discovery result is what fails the gate. + const narrow: DatasetContract = { ...CONTRACT, fields: CONTRACT.fields.slice(0, 2) }; + await expect( + service.ingest(new NativeConnectorProvider(), jsonlSpec({ target: target({ contract: narrow }) })), + ).rejects.toThrow(/undeclared_field/); + expect(writer.batches).toHaveLength(0); + }); +}); + +// ── gate 3: physical home ────────────────────────────────────────────────── + +describe('BI-P129 gate 3 — RAW/NORMALIZED land as Parquet/Iceberg in ST0RE', () => { + it('a ClickHouse landing is refused', () => { + expect(() => + assertPhysicalHomeMatchesTier( + target({ physical: { format: 'clickhouse' as never, namespace: 'n', table_name: 't' } }), + ), + ).toThrow(/not a lakehouse format/); + }); + + it('iceberg and parquet are both accepted; a nameless table is not', () => { + expect(() => + assertPhysicalHomeMatchesTier( + target({ physical: { format: 'iceberg', namespace: 'n', table_name: 't' } }), + ), + ).not.toThrow(); + expect(() => + assertPhysicalHomeMatchesTier( + target({ physical: { format: 'parquet', namespace: ' ', table_name: 't' } }), + ), + ).toThrow(/missing a ST0RE namespace/); + }); +}); + +// ── gate 4: unverified providers ─────────────────────────────────────────── + +class UnverifiedProvider implements AnalyticsConnectorProvider { + readonly provider = 'dlt' as const; + providerReadiness() { + return { + provider: 'dlt', + pinned_version: '0.0.0', + verified: false, + proven: [], + open: ['nothing has been executed'], + }; + } + validateSpec() { + return { valid: true, errors: [] }; + } + connectorFor(): never { + throw new Error('not used in this test'); + } + executionPlan() { + return { runtime: 'n0d' as const, reason: 'test' }; + } +} + +describe('BI-P129 gate 4 — an unverified provider cannot quietly become production capability', () => { + const provider = new UnverifiedProvider(); + const spec = jsonlSpec({ provider: 'dlt', connector_ref: 'whatever' }); + + it('WITH the guard: production refuses an unverified provider', () => { + expect(() => assertProviderRunnable(provider, spec, 'production')).toThrow(SiaxError); + expect(() => assertProviderRunnable(provider, spec, 'production')).toThrow(/verified: false/); + }); + + it('WITHOUT the guard (non-production runtime): the same provider is allowed to plan', () => { + expect(() => assertProviderRunnable(provider, spec, 'staging')).not.toThrow(); + }); + + it('a recorded, named exception unblocks production — and is required to be non-empty', () => { + expect(() => + assertProviderRunnable(provider, { ...spec, unverified_provider_exception: ' ' }, 'production'), + ).toThrow(/unverified_provider_exception/); + expect(() => + assertProviderRunnable( + provider, + { + ...spec, + unverified_provider_exception: 'accepted by simon@siax.io 2026-09-07 for a bounded backfill', + }, + 'production', + ), + ).not.toThrow(); + }); + + it('the native provider is verified, so it needs no exception in production', () => { + expect(() => + assertProviderRunnable(new NativeConnectorProvider(), jsonlSpec(), 'production'), + ).not.toThrow(); + expect(new NativeConnectorProvider().providerReadiness().verified).toBe(true); + }); +}); + +// ── licence boundary + registry ──────────────────────────────────────────── + +describe('BI-P129 — Airbyte is a declared kind with no acceptance in this estate', () => { + it('registering an airbyte provider is refused at the registry', () => { + const registry = new ConnectorProviderRegistry(); + const fake = { ...new UnverifiedProvider(), provider: 'airbyte' as const }; + expect(() => registry.register(fake as unknown as AnalyticsConnectorProvider)).toThrow( + /no licence\/architecture acceptance/, + ); + }); + + it('planning an airbyte ingestion is refused even if a provider object is supplied directly', () => { + const fake = { + ...new UnverifiedProvider(), + provider: 'airbyte' as const, + } as unknown as AnalyticsConnectorProvider; + expect(() => planAnalyticsIngestion(fake, jsonlSpec({ provider: 'airbyte' }), 'test')).toThrow( + /licence\/architecture acceptance/, + ); + }); + + it('the registry refuses duplicates and unknown kinds, and reports readiness for what it has', () => { + const registry = new ConnectorProviderRegistry(); + registry.register(new NativeConnectorProvider()); + expect(() => registry.register(new NativeConnectorProvider())).toThrow(/already registered/); + expect(() => registry.get('singer')).toThrow(/no connector provider registered/); + expect(registry.readinessReport().map((r) => r.provider)).toEqual(['native']); + }); + + it('a provider handed a spec for a different kind is refused', () => { + expect(() => + planAnalyticsIngestion(new NativeConnectorProvider(), jsonlSpec({ provider: 'dlt' }), 'test'), + ).toThrow(/asks for provider "dlt"/); + }); +});