0pt/tests/analytics/analytics-ingestion-providers.test.ts

/**
 * 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> = {}): 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<Response> {
  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> = {}): 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> = {}): 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<Record<string, unknown> | 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"/);
  });
});