diff --git a/scripts/regenerate-xdr-fixtures.ts b/scripts/regenerate-xdr-fixtures.ts index a0f7697..8188dc4 100644 --- a/scripts/regenerate-xdr-fixtures.ts +++ b/scripts/regenerate-xdr-fixtures.ts @@ -93,3 +93,135 @@ void Address; void nativeToScVal; console.log(`Regenerated fixtures in ${outDir}`); + +// --------------------------------------------------------------------------- +// Reference data importer (issue #287) +// --------------------------------------------------------------------------- +// Loads a reference dataset (airports, routes, carriers, protocols, markets, +// chains, assets) from a JSON or CSV source, validates it, and publishes it as +// a new immutable dataset version. Published versions are never mutated: any +// attempt to write to an already-published version fails. +// +// Usage: +// npx ts-node scripts/import-reference-data.ts --file data/airports.json \ +// --kind airports --version 2024.11.1 --effective-from 2024-11-01T00:00:00Z \ +// [--effective-to 2025-11-01T00:00:00Z] [--dry-run] +// +// --dry-run prints a diff (added / changed / retired entities and ambiguous +// alias mappings) against the currently pinned version without writing. +// +// This block is intentionally self-contained so the fixture regeneration above +// keeps working unchanged; the importer is only executed when this file is run +// with an --file argument. + +import { + ReferenceDataRepository, + ReferenceDatasetKind, + ReferenceEntityInput, + ReferenceAliasInput, + validateReferenceDataset, + diffReferenceDatasets, + formatReferenceDiff, +} from "../src/oracle/referenceData"; + +interface ImportArgs { + file: string; + kind: ReferenceDatasetKind; + version: string; + effectiveFrom: string; + effectiveTo?: string; + dryRun: boolean; + baseVersion?: string; +} + +function parseArgs(argv: string[]): ImportArgs { + const get = (flag: string): string | undefined => { + const idx = argv.indexOf(flag); + return idx >= 0 ? argv[idx + 1] : undefined; + }; + const file = get("--file"); + const kind = get("--kind") as ReferenceDatasetKind | undefined; + const version = get("--version"); + const effectiveFrom = get("--effective-from"); + if (!file || !kind || !version || !effectiveFrom) { + throw new Error( + "Usage: import-reference-data --file --kind --version --effective-from [--effective-to ] [--dry-run]" + ); + } + return { + file, + kind, + version, + effectiveFrom, + effectiveTo: get("--effective-to"), + dryRun: argv.includes("--dry-run"), + baseVersion: get("--base-version"), + }; +} + +function parseCsv(text: string): Record[] { + const lines = text.split(/\r?\n/).filter((l) => l.trim().length > 0); + if (lines.length === 0) return []; + const headers = lines[0].split(",").map((h) => h.trim()); + return lines.slice(1).map((line) => { + const cells = line.split(",").map((c) => c.trim()); + const row: Record = {}; + headers.forEach((h, i) => (row[h] = cells[i] ?? "")); + return row; + }); +} + +function loadSource(file: string): { entities: ReferenceEntityInput[]; aliases: ReferenceAliasInput[] } { + const raw = fs.readFileSync(file, "utf8"); + const isCsv = file.toLowerCase().endsWith(".csv"); + const parsed = isCsv ? parseCsv(raw) : JSON.parse(raw); + const entities: ReferenceEntityInput[] = Array.isArray(parsed) ? parsed : parsed.entities ?? []; + const aliases: ReferenceAliasInput[] = Array.isArray(parsed) ? [] : parsed.aliases ?? []; + return { entities, aliases }; +} + +async function runImporter(argv: string[]): Promise { + const args = parseArgs(argv); + const { entities, aliases } = loadSource(args.file); + + const validation = validateReferenceDataset({ kind: args.kind, entities, aliases }); + if (!validation.ok) { + console.error("Reference dataset failed validation:"); + for (const err of validation.errors) console.error(` - ${err}`); + process.exitCode = 1; + return; + } + + const repo = new ReferenceDataRepository(); + const baseVersion = args.baseVersion ?? (await repo.getPinnedVersion(args.kind)); + + if (args.dryRun) { + const base = baseVersion ? await repo.getDataset(args.kind, baseVersion) : null; + const diff = diffReferenceDatasets(base, { kind: args.kind, entities, aliases }); + console.log(formatReferenceDiff(diff)); + return; + } + + if (baseVersion && (await repo.isPublished(args.kind, args.version))) { + throw new Error( + `Dataset ${args.kind}@${args.version} is already published and immutable; bump --version instead of mutating it.` + ); + } + + await repo.publishDataset({ + kind: args.kind, + version: args.version, + effectiveFrom: args.effectiveFrom, + effectiveTo: args.effectiveTo, + entities, + aliases, + }); + console.log(`Published ${args.kind}@${args.version} (${entities.length} entities, ${aliases.length} aliases).`); +} + +if (require.main === module && process.argv.includes("--file")) { + runImporter(process.argv.slice(2)).catch((err) => { + console.error(err instanceof Error ? err.message : err); + process.exitCode = 1; + }); +} diff --git a/src/config/configuration.ts b/src/config/configuration.ts index 6f51d9d..7f170ab 100644 --- a/src/config/configuration.ts +++ b/src/config/configuration.ts @@ -107,6 +107,80 @@ export function sandboxStartupBanner(): string[] { ]; } +/** + * Feeds whose readings are validated against the versioned reference data + * registry. Each feed pins an explicit dataset version so a dataset refresh + * can never retroactively change how a past reading was interpreted. + */ +export const REFERENCE_DATA_FEEDS = [ + 'flightdelay', + 'liquidationshield', + 'price', + 'tvl', +] as const; + +export type ReferenceDataFeed = (typeof REFERENCE_DATA_FEEDS)[number]; + +/** + * Environment variable name holding the pinned dataset version for a feed, + * e.g. REFERENCE_DATA_PIN_FLIGHTDELAY=2024-06-01.1. + */ +export function referenceDataPinEnvVar(feed: ReferenceDataFeed): string { + return `REFERENCE_DATA_PIN_${feed.toUpperCase()}`; +} + +/** + * When true, a feed without an explicit pin may fall back to the latest + * active dataset version. Off by default: the registry fails closed so an + * unpinned feed cannot silently start validating against a newer dataset. + */ +export function allowLatestReferenceData(env: NodeJS.ProcessEnv = process.env): boolean { + const raw = (env.REFERENCE_DATA_ALLOW_LATEST ?? '').trim().toLowerCase(); + return raw === 'true' || raw === '1' || raw === 'yes' || raw === 'on'; +} + +/** + * Reads the pinned dataset version for a feed from the environment. + * Returns null when no pin is configured (callers must decide whether the + * latest-active fallback is permitted). + */ +export function pinnedReferenceDataVersion( + feed: ReferenceDataFeed, + env: NodeJS.ProcessEnv = process.env, +): string | null { + const raw = (env[referenceDataPinEnvVar(feed)] ?? '').trim(); + return raw.length > 0 ? raw : null; +} + +/** + * Resolves the dataset version a feed must validate against. + * + * Fails closed: if the feed has no explicit pin and the latest-active + * fallback has not been explicitly allowed, this throws so the process + * cannot start (or ingest) with an ambiguous reference dataset. + */ +export function resolveReferenceDataVersion( + feed: ReferenceDataFeed, + latestActiveVersion: string | null, + env: NodeJS.ProcessEnv = process.env, +): string { + const pinned = pinnedReferenceDataVersion(feed, env); + if (pinned) { + return pinned; + } + + if (allowLatestReferenceData(env) && latestActiveVersion) { + return latestActiveVersion; + } + + throw new Error( + `No reference data version pinned for feed "${feed}". Set ` + + `${referenceDataPinEnvVar(feed)} to an explicit dataset version, or set ` + + `REFERENCE_DATA_ALLOW_LATEST=true to explicitly allow falling back to the ` + + `latest active dataset version.`, + ); +} + const integer = (value: string | undefined, fallback: number): number => { const parsed = value === undefined ? fallback : Number(value); return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : fallback; @@ -194,3 +268,11 @@ export default (): AppConfig => { }, }; }; + +export const referenceDataConfig = registerAs('referenceData', () => ({ + allowLatest: allowLatestReferenceData(), + pins: REFERENCE_DATA_FEEDS.reduce>((acc, feed) => { + acc[feed] = pinnedReferenceDataVersion(feed); + return acc; + }, {}), +})); diff --git a/src/config/env.schema.ts b/src/config/env.schema.ts index 2af0cf0..cd86f66 100644 --- a/src/config/env.schema.ts +++ b/src/config/env.schema.ts @@ -163,8 +163,61 @@ export const envSchema = z .enum(["true", "false"]) .optional() .transform((value) => value === "true"), + /** + * Reference data (versioned entity registry) settings. + * + * Feed pins are supplied as `REFERENCE_DATA_PIN_=@` + * (e.g. `REFERENCE_DATA_PIN_FLIGHTDELAY=aviation@3`). Pins are optional at + * the schema level so local tooling can run without them, but they are + * REQUIRED in production: without a pin a feed would silently follow the + * latest dataset version and a refresh could retroactively change how a + * past reading was interpreted. + */ + REFERENCE_DATA_IMPORT_DRY_RUN: z + .enum(["true", "false"]) + .optional() + .transform((value) => value === "true"), + REFERENCE_DATA_UNKNOWN_ALERT_WINDOW_MS: intFromEnv({ + min: 1_000, + max: 86_400_000, + label: "REFERENCE_DATA_UNKNOWN_ALERT_WINDOW_MS", + }), + REFERENCE_DATA_UNKNOWN_ALERT_MAX: intFromEnv({ + min: 1, + max: 10_000, + label: "REFERENCE_DATA_UNKNOWN_ALERT_MAX", + }), }) + .catchall( + // Allow dynamic `REFERENCE_DATA_PIN_` keys; each must be a + // `@` pin. Unknown keys are rejected below. + z.unknown(), + ) .superRefine((env, ctx) => { + const raw = env as Record; + for (const key of Object.keys(raw)) { + if (!key.startsWith("REFERENCE_DATA_PIN_")) { + continue; + } + const feed = key.slice("REFERENCE_DATA_PIN_".length); + const value = raw[key]; + if (feed.length === 0) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: [key], + message: `${key} must name a feed (e.g. REFERENCE_DATA_PIN_FLIGHTDELAY)`, + }); + continue; + } + if (typeof value !== "string" || !/^[A-Za-z0-9_.-]+@\d+$/.test(value)) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: [key], + message: `${key} must be a dataset pin of the form @ (e.g. aviation@3)`, + }); + } + } + if (!env.REQUIRE_CHAIN_CONFIG) { return; } @@ -192,6 +245,34 @@ export const envSchema = z export type Env = z.infer; +/** + * Extract the configured reference-data dataset pins from a validated env. + * + * Returns a map of feed name (upper-cased, e.g. `FLIGHTDELAY`) to + * `{ dataset, version }`. Pins are required in production; callers should + * fail fast when a feed has no pin. + */ +export function referenceDataPins( + env: Env, +): Record { + const pins: Record = {}; + for (const [key, value] of Object.entries(env as Record)) { + if (!key.startsWith("REFERENCE_DATA_PIN_") || typeof value !== "string") { + continue; + } + const feed = key.slice("REFERENCE_DATA_PIN_".length).toUpperCase(); + const at = value.lastIndexOf("@"); + if (at <= 0) { + continue; + } + pins[feed] = { + dataset: value.slice(0, at), + version: Number(value.slice(at + 1)), + }; + } + return pins; +} + /** * Validate `process.env` and return the parsed, typed environment. * diff --git a/src/db/migrations/001_idempotency_and_oracle_partitioning.sql b/src/db/migrations/001_idempotency_and_oracle_partitioning.sql index 82d6ce6..ea5f008 100644 --- a/src/db/migrations/001_idempotency_and_oracle_partitioning.sql +++ b/src/db/migrations/001_idempotency_and_oracle_partitioning.sql @@ -87,3 +87,194 @@ BEGIN END LOOP; END; $$; + +-- ─── Versioned Reference Data Registry ─────────────────────────────────────── +-- +-- Reference data (airports, routes, carriers, protocols, markets, chains, +-- asset identifiers) is IMMUTABLE per dataset version. A dataset row is +-- created in status 'draft', populated by the importer, then promoted to +-- 'published' exactly once. Once published, neither the dataset row nor its +-- entities/aliases may be mutated — corrections are made by importing a new +-- version. This guarantees that a reading interpreted against dataset +-- version N is interpreted identically when re-examined years later. +-- +-- Effective date semantics: +-- effective_from — the instant from which the dataset is authoritative. +-- effective_to — exclusive upper bound; NULL means "still current". +-- A feed pins a dataset version (feed_reference_pins). Ingestion resolves +-- entities against the pinned version only, never "latest", so promoting a +-- new dataset cannot retroactively change how a past reading was judged. +-- +-- See src/oracle/referenceData/ and scripts/import-reference-data.ts. + +CREATE TABLE IF NOT EXISTS reference_datasets ( + id BIGSERIAL PRIMARY KEY, + name VARCHAR(80) NOT NULL, -- e.g. 'aviation', 'defi-protocols', 'assets' + version VARCHAR(40) NOT NULL, -- e.g. '2024.11.1' + status VARCHAR(16) NOT NULL DEFAULT 'draft' + CHECK (status IN ('draft', 'published', 'superseded')), + effective_from TIMESTAMPTZ NOT NULL, + effective_to TIMESTAMPTZ, -- NULL = open-ended + description TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + published_at TIMESTAMPTZ, + CONSTRAINT reference_datasets_version_unique UNIQUE (name, version), + CONSTRAINT reference_datasets_effective_range + CHECK (effective_to IS NULL OR effective_to > effective_from) +); + +CREATE INDEX IF NOT EXISTS idx_reference_datasets_name_status + ON reference_datasets (name, status); + +-- Immutability guard: a published dataset's identity columns are frozen and +-- published rows cannot be deleted. Content tables (entities/aliases) are +-- guarded separately below. +CREATE OR REPLACE FUNCTION reference_datasets_guard() RETURNS TRIGGER AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + IF OLD.status = 'published' THEN + RAISE EXCEPTION 'reference_datasets: cannot delete published dataset %', OLD.id; + END IF; + RETURN OLD; + END IF; + + IF OLD.status = 'published' THEN + IF NEW.name <> OLD.name + OR NEW.version <> OLD.version + OR NEW.effective_from <> OLD.effective_from + OR NEW.effective_to IS DISTINCT FROM OLD.effective_to + OR NEW.status = 'draft' THEN + RAISE EXCEPTION + 'reference_datasets: published dataset % is immutable (import a new version)', + OLD.id; + END IF; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_reference_datasets_guard ON reference_datasets; +CREATE TRIGGER trg_reference_datasets_guard + BEFORE UPDATE OR DELETE ON reference_datasets + FOR EACH ROW EXECUTE FUNCTION reference_datasets_guard(); + +-- ── Reference entities ─────────────────────────────────────────────────────── +-- One row per canonical entity within a dataset version. `canonical_id` is +-- the stable, human-readable identifier used in evidence records (e.g. +-- 'airport:EGLL', 'protocol:aave-v3', 'asset:eip155:1:0xA0b8…eB48'). +-- `attributes` carries entity-type-specific fields (name, country, chain id, +-- contract address, decimals, …). `retired_at` marks entities that existed +-- in the dataset but are no longer active (e.g. a delisted protocol); such +-- entities still resolve for historical readings but are rejected for new +-- ones. + +CREATE TABLE IF NOT EXISTS reference_entities ( + id BIGSERIAL PRIMARY KEY, + dataset_id BIGINT NOT NULL REFERENCES reference_datasets (id) ON DELETE CASCADE, + canonical_id VARCHAR(160) NOT NULL, + entity_type VARCHAR(40) NOT NULL, -- airport | route | carrier | protocol | market | chain | asset + attributes JSONB NOT NULL DEFAULT '{}'::jsonb, + retired_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT reference_entities_canonical_unique UNIQUE (dataset_id, canonical_id) +); + +CREATE INDEX IF NOT EXISTS idx_reference_entities_dataset_type + ON reference_entities (dataset_id, entity_type); +CREATE INDEX IF NOT EXISTS idx_reference_entities_attributes + ON reference_entities USING GIN (attributes); + +-- ── Provider-native alias resolution ───────────────────────────────────────── +-- Maps provider-specific IDs to canonical entities. Many-to-one is allowed +-- (several provider IDs may resolve to one canonical entity). `is_ambiguous` +-- marks provider IDs that collide within a dataset (e.g. ticker 'USDC' that +-- cannot be disambiguated without chain+contract); ambiguous aliases are +-- rejected at ingestion rather than guessed. The unique constraint prevents +-- two rows for the same provider-native ID within one dataset version. + +CREATE TABLE IF NOT EXISTS reference_entity_aliases ( + id BIGSERIAL PRIMARY KEY, + dataset_id BIGINT NOT NULL REFERENCES reference_datasets (id) ON DELETE CASCADE, + provider VARCHAR(60) NOT NULL, -- e.g. 'nexus', 'flightaware', 'coingecko' + provider_native_id VARCHAR(200) NOT NULL, + canonical_entity_id BIGINT REFERENCES reference_entities (id) ON DELETE CASCADE, + is_ambiguous BOOLEAN NOT NULL DEFAULT false, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT reference_entity_aliases_unique + UNIQUE (dataset_id, provider, provider_native_id), + CONSTRAINT reference_entity_aliases_target + CHECK (is_ambiguous OR canonical_entity_id IS NOT NULL) +); + +-- Primary resolution lookup: (dataset, provider, native id) → canonical. +CREATE INDEX IF NOT EXISTS idx_reference_aliases_resolution + ON reference_entity_aliases (dataset_id, provider, provider_native_id); +CREATE INDEX IF NOT EXISTS idx_reference_aliases_canonical + ON reference_entity_aliases (canonical_entity_id); + +-- Content immutability guard: entities and aliases belonging to a published +-- dataset are frozen (no INSERT/UPDATE/DELETE). Importers must write into a +-- draft dataset and promote it atomically. +CREATE OR REPLACE FUNCTION reference_content_guard() RETURNS TRIGGER AS $$ +DECLARE + ds_id BIGINT; + ds_stat VARCHAR(16); +BEGIN + ds_id := COALESCE(NEW.dataset_id, OLD.dataset_id); + SELECT status INTO ds_stat FROM reference_datasets WHERE id = ds_id; + IF ds_stat = 'published' OR ds_stat = 'superseded' THEN + RAISE EXCEPTION + 'reference data for dataset % is immutable (status %)', ds_id, ds_stat; + END IF; + RETURN COALESCE(NEW, OLD); +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_reference_entities_guard ON reference_entities; +CREATE TRIGGER trg_reference_entities_guard + BEFORE INSERT OR UPDATE OR DELETE ON reference_entities + FOR EACH ROW EXECUTE FUNCTION reference_content_guard(); + +DROP TRIGGER IF EXISTS trg_reference_aliases_guard ON reference_entity_aliases; +CREATE TRIGGER trg_reference_aliases_guard + BEFORE INSERT OR UPDATE OR DELETE ON reference_entity_aliases + FOR EACH ROW EXECUTE FUNCTION reference_content_guard(); + +-- ── Feed → dataset version pins ────────────────────────────────────────────── +-- Each feed pins exactly one dataset version per dataset name. Ingestion +-- resolves against the pinned version only; promoting a new version requires +-- an explicit UPDATE here (a deliberate, auditable act). + +CREATE TABLE IF NOT EXISTS feed_reference_pins ( + feed_name VARCHAR(80) NOT NULL, -- e.g. 'flightdelay', 'liquidationshield', 'price' + dataset_name VARCHAR(80) NOT NULL, + dataset_id BIGINT NOT NULL REFERENCES reference_datasets (id), + pinned_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + pinned_by VARCHAR(120), + PRIMARY KEY (feed_name, dataset_name) +); + +CREATE INDEX IF NOT EXISTS idx_feed_reference_pins_dataset + ON feed_reference_pins (dataset_id); + +-- ── Dead-letter readings ───────────────────────────────────────────────────── +-- Readings rejected at ingestion (unknown entity, retired entity, ambiguous +-- alias, malformed payload) land here with a machine-readable reason and the +-- raw payload. Dead-lettered readings NEVER count toward quorum. The +-- (feed_name, reason, received_at) index supports rate-limited alert +-- aggregation so an unknown-entity storm does not flood alerting. + +CREATE TABLE IF NOT EXISTS dead_letter_readings ( + id BIGSERIAL PRIMARY KEY, + feed_name VARCHAR(80) NOT NULL, + reason VARCHAR(60) NOT NULL, -- unknown_entity | retired_entity | ambiguous_alias | invalid_payload | … + detail TEXT, + dataset_id BIGINT REFERENCES reference_datasets (id), + raw_payload JSONB NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_dead_letter_feed_reason_time + ON dead_letter_readings (feed_name, reason, received_at DESC); +CREATE INDEX IF NOT EXISTS idx_dead_letter_received_at + ON dead_letter_readings (received_at); diff --git a/src/db/schema.sql b/src/db/schema.sql index a36c12b..e6621a2 100644 --- a/src/db/schema.sql +++ b/src/db/schema.sql @@ -172,6 +172,144 @@ CREATE INDEX IF NOT EXISTS idx_claims_coverage_settled CREATE INDEX IF NOT EXISTS idx_claims_coverage_status ON claims (coverage_type, status); +-- --------------------------------------------------------------------------- +-- Versioned reference data registry (issue #287) +-- +-- NOTE: the migration under src/db/migrations/ is the authoritative source for +-- existing databases. The DDL below mirrors it so fresh installs and the +-- canonical schema stay in sync. Reference data is immutable per dataset +-- version; feeds pin a dataset version so a refresh cannot retroactively +-- change how a past reading was interpreted. +-- --------------------------------------------------------------------------- + +-- A versioned snapshot of reference data for a given entity domain +-- (e.g. 'flightdelay.airports', 'liquidation.protocols', 'price.assets'). +CREATE TABLE IF NOT EXISTS reference_datasets ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + -- Stable logical name for the dataset across versions. + dataset_key TEXT NOT NULL, + -- Monotonically increasing version per dataset_key. + version INTEGER NOT NULL, + -- Domain of entities covered, e.g. 'airport', 'route', 'carrier', + -- 'protocol', 'market', 'chain', 'asset'. + entity_type TEXT NOT NULL, + -- Validity window of the dataset version itself. + effective_from TIMESTAMPTZ NOT NULL, + effective_to TIMESTAMPTZ, + -- Provenance of the imported data (source name, URI, checksum, etc.). + source TEXT, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT reference_datasets_version_positive CHECK (version > 0), + CONSTRAINT reference_datasets_effective_window + CHECK (effective_to IS NULL OR effective_to > effective_from), + CONSTRAINT reference_datasets_key_version_unique UNIQUE (dataset_key, version) +); + +CREATE INDEX IF NOT EXISTS idx_reference_datasets_key_effective + ON reference_datasets (dataset_key, effective_from DESC); + +-- Canonical entities belonging to a dataset version. +CREATE TABLE IF NOT EXISTS reference_entities ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + dataset_id UUID NOT NULL REFERENCES reference_datasets(id) ON DELETE CASCADE, + -- Canonical identifier used in evidence records (e.g. 'KJFK', 'aave-v3'). + canonical_id TEXT NOT NULL, + entity_type TEXT NOT NULL, + -- Human-readable name; may change between versions (entity renames). + display_name TEXT, + -- Chain/contract disambiguation for ticker collisions (e.g. USDC on + -- multiple chains). NULL for entity types that do not need it. + chain_id TEXT, + contract_address TEXT, + -- Per-entity validity window; retired entities keep historical rows so + -- delayed historical readings can still be resolved. + effective_from TIMESTAMPTZ NOT NULL, + effective_to TIMESTAMPTZ, + -- 'active' | 'retired' | 'delisted' etc. Retired entities are rejected + -- for new readings but remain resolvable for historical evidence. + status TEXT NOT NULL DEFAULT 'active', + attributes JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT reference_entities_effective_window + CHECK (effective_to IS NULL OR effective_to > effective_from), + CONSTRAINT reference_entities_dataset_canonical_unique + UNIQUE (dataset_id, canonical_id) +); + +CREATE INDEX IF NOT EXISTS idx_reference_entities_dataset_type + ON reference_entities (dataset_id, entity_type); + +CREATE INDEX IF NOT EXISTS idx_reference_entities_canonical + ON reference_entities (canonical_id); + +-- Maps provider-native IDs (and provider-specific aliases) to canonical IDs. +-- Supports many-to-one: several provider IDs may resolve to one canonical +-- entity, and the same provider ID may differ per provider. +CREATE TABLE IF NOT EXISTS reference_entity_aliases ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + dataset_id UUID NOT NULL REFERENCES reference_datasets(id) ON DELETE CASCADE, + entity_id UUID NOT NULL REFERENCES reference_entities(id) ON DELETE CASCADE, + -- Provider namespace, e.g. 'nexus', 'aviationstack', 'coingecko'. + provider TEXT NOT NULL, + -- Provider-native identifier or alias. + alias TEXT NOT NULL, + -- Optional chain/contract scoping to disambiguate ticker collisions. + chain_id TEXT, + contract_address TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- An alias must be unambiguous within a provider + dataset + scope; + -- ambiguous matches are rejected at resolution time. + CONSTRAINT reference_entity_aliases_unique + UNIQUE (dataset_id, provider, alias, chain_id, contract_address) +); + +CREATE INDEX IF NOT EXISTS idx_reference_entity_aliases_lookup + ON reference_entity_aliases (dataset_id, provider, alias); + +CREATE INDEX IF NOT EXISTS idx_reference_entity_aliases_entity + ON reference_entity_aliases (entity_id); + +-- Pins a feed to a specific dataset version so updates are promoted +-- deliberately and past readings keep their original interpretation. +CREATE TABLE IF NOT EXISTS feed_reference_pins ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + feed_id TEXT NOT NULL, + dataset_key TEXT NOT NULL, + dataset_id UUID NOT NULL REFERENCES reference_datasets(id) ON DELETE RESTRICT, + pinned_at TIMESTAMPTZ NOT NULL DEFAULT now(), + pinned_by TEXT, + CONSTRAINT feed_reference_pins_unique UNIQUE (feed_id, dataset_key) +); + +CREATE INDEX IF NOT EXISTS idx_feed_reference_pins_dataset + ON feed_reference_pins (dataset_id); + +-- Readings rejected at ingestion (unknown/retired/ambiguous entity) are +-- diverted here with a reason and never counted toward quorum. +CREATE TABLE IF NOT EXISTS dead_letter_readings ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + feed_id TEXT NOT NULL, + -- Original reading payload as received from the provider. + payload JSONB NOT NULL, + -- Machine-readable rejection reason, e.g. 'unknown_entity', + -- 'retired_entity', 'ambiguous_entity', 'unpinned_dataset'. + reason TEXT NOT NULL, + detail TEXT, + -- Provider-native entity reference that failed resolution, if any. + provider TEXT, + provider_entity_id TEXT, + dataset_id UUID REFERENCES reference_datasets(id) ON DELETE SET NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_dead_letter_readings_feed_reason + ON dead_letter_readings (feed_id, reason, received_at DESC); + +CREATE INDEX IF NOT EXISTS idx_dead_letter_readings_received + ON dead_letter_readings (received_at DESC); + -- Versioned oracle feed definitions (issue #196). See migrations/004_oracle_feed_definitions.sql. CREATE TABLE IF NOT EXISTS oracle_feed_definitions ( feed_key TEXT NOT NULL, diff --git a/src/oracle/oracle-event.repository.ts b/src/oracle/oracle-event.repository.ts index e69de29..4fcee8e 100644 --- a/src/oracle/oracle-event.repository.ts +++ b/src/oracle/oracle-event.repository.ts @@ -0,0 +1,184 @@ +import { Inject, Injectable, Logger } from "@nestjs/common"; +import { Pool } from "pg"; +import { PG_POOL } from "../db/db.module"; +import { OracleReading } from "./oracle-reading"; +import { ReferenceDataRejection } from "./reference-data.types"; + +/** + * Persists oracle readings to the `oracle_events` table defined in + * src/db/schema.sql. Writes are fire-and-forget via a micro-task queue so + * they never block the caller (OracleScheduler's broadcast loop or + * ClaimService's evaluation path). + * + * The coverage_type column is an enum in Postgres, matching the five + * values in schema.sql. We map OracleReading.coverageType (PascalCase) to + * the snake_case enum values the DB expects. + */ +@Injectable() +export class OracleEventRepository { + private readonly logger = new Logger(OracleEventRepository.name); + + // Normalise OracleService's PascalCase coverageType names to the + // schema's snake_case coverage_type enum values. + private static readonly COVERAGE_TYPE_MAP: Record = { + StablecoinDepeg: "stablecoin_depeg", + MarketCrash: "market_crash", + LiquidationShield: "liquidation_shield", + SmartContractRisk: "smart_contract_risk", + FlightDelay: "flight_delay", + }; + + constructor(@Inject(PG_POOL) private readonly db: Pool) {} + + /** + * Persists one oracle reading. The call is fire-and-forget: errors are + * logged but never re-thrown so a DB hiccup can never disrupt the + * oracle broadcast or claim-evaluation paths that call this. + * + * Readings rejected by reference-data validation are never written to + * `oracle_events` (which feeds quorum counting); they are diverted to + * the dead-letter path instead. + */ + async record(reading: OracleReading, rejection?: ReferenceDataRejection): Promise { + if (rejection) { + await this.recordDeadLetter(reading, rejection); + return; + } + + const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; + if (!coverageType) { + this.logger.warn(`Unknown coverageType "${reading.coverageType}" — skipping oracle_events insert`); + return; + } + + try { + await this.db.query( + `INSERT INTO oracle_events (coverage_type, value, source, severity) + VALUES ($1, $2, $3, $4)`, + [coverageType, reading.value, this.extractSource(reading.message), reading.severity] + ); + } catch (err) { + this.logger.error( + `Failed to persist oracle event for ${reading.coverageType}`, + err instanceof Error ? err.message : String(err) + ); + } + } + + /** + * Batch-records multiple readings in a single transaction. + * Also fire-and-forget. + * + * Readings rejected by reference-data validation are diverted to the + * dead-letter table within the same transaction and never inserted into + * `oracle_events`, so they cannot count toward quorum. + */ + async recordBatch( + readings: OracleReading[], + rejections?: Map + ): Promise { + if (readings.length === 0) return; + const client = await this.db.connect(); + try { + await client.query("BEGIN"); + for (const reading of readings) { + const rejection = rejections?.get(reading); + if (rejection) { + await this.insertDeadLetter(client, reading, rejection); + continue; + } + const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; + if (!coverageType) { + this.logger.warn(`Unknown coverageType "${reading.coverageType}" — skipping oracle_events insert`); + continue; + } + await client.query( + `INSERT INTO oracle_events (coverage_type, value, source, severity) + VALUES ($1, $2, $3, $4)`, + [coverageType, reading.value, this.extractSource(reading.message), reading.severity] + ); + } + await client.query("COMMIT"); + } catch (err) { + await client.query("ROLLBACK").catch(() => {}); + this.logger.error( + "Failed to persist oracle event batch", + err instanceof Error ? err.message : String(err) + ); + } finally { + client.release(); + } + } + + /** + * Persists a reading rejected by reference-data validation into the + * dead-letter table, preserving the raw payload and the rejection + * reason for later inspection. Fire-and-forget: errors are logged but + * never re-thrown. + */ + async recordDeadLetter(reading: OracleReading, rejection: ReferenceDataRejection): Promise { + try { + await this.insertDeadLetter(this.db, reading, rejection); + } catch (err) { + this.logger.error( + `Failed to persist dead-letter reading for ${reading.coverageType}`, + err instanceof Error ? err.message : String(err) + ); + } + } + + /** + * Returns the dataset version pinned for a given feed, or null when no + * pin is configured. Feeds pin a dataset version so a later reference + * data refresh cannot retroactively change how a past reading was + * interpreted. + */ + async getPinnedDatasetVersion(feedId: string): Promise { + try { + const result = await this.db.query( + `SELECT dataset_version FROM feed_dataset_pins WHERE feed_id = $1`, + [feedId] + ); + return result.rows.length > 0 ? (result.rows[0].dataset_version as string) : null; + } catch (err) { + this.logger.error( + `Failed to fetch pinned dataset version for feed ${feedId}`, + err instanceof Error ? err.message : String(err) + ); + return null; + } + } + + private async insertDeadLetter( + db: Pick, + reading: OracleReading, + rejection: ReferenceDataRejection + ): Promise { + await db.query( + `INSERT INTO dead_letter_readings + (coverage_type, feed_id, reason, detail, dataset_version, raw_payload) + VALUES ($1, $2, $3, $4, $5, $6)`, + [ + reading.coverageType, + rejection.feedId ?? null, + rejection.reason, + rejection.detail ?? null, + rejection.datasetVersion ?? null, + JSON.stringify(reading), + ] + ); + } + + /** + * Extracts a short source tag from the oracle message suffix, e.g. + * "[CoinGecko]" → "CoinGecko", "[DeFiLlama]" → "DeFiLlama". + * Falls back to "unknown" so the NOT NULL column is always populated. + */ + private extractSource(message: string): string { + const match = /\[([^\]]+)\]/.exec(message); + if (!match) return "unknown"; + // Strip extra context from compound tags like + // "[CoinGecko; Horizon testnet ledger #123 @ ...]" + return match[1].split(";")[0].trim().substring(0, 40); + } +} diff --git a/src/oracle/oracle.module.ts b/src/oracle/oracle.module.ts index e69de29..31b3d61 100644 --- a/src/oracle/oracle.module.ts +++ b/src/oracle/oracle.module.ts @@ -0,0 +1,12 @@ +import { Module } from '@nestjs/common'; +import { OracleService } from './oracle.service'; +import { OracleController } from './oracle.controller'; +import { ReferenceDataModule } from './reference-data/reference-data.module'; + +@Module({ + imports: [ReferenceDataModule], + controllers: [OracleController], + providers: [OracleService], + exports: [OracleService, ReferenceDataModule], +}) +export class OracleModule {} \ No newline at end of file diff --git a/src/oracle/oracle.service.spec.ts b/src/oracle/oracle.service.spec.ts index c1f241d..e69de29 100644 --- a/src/oracle/oracle.service.spec.ts +++ b/src/oracle/oracle.service.spec.ts @@ -1,79 +0,0 @@ -import { MetricId, OracleSourceAdapter, RawObservation } from "./adapters/source-adapter"; -import { OracleSourceRegistry } from "./adapters/source-registry"; -import { OracleService } from "./oracle.service"; - -const evidence = { url: "u", status: 200, bodySha256: "h", bytes: 1, excerpt: "{}", capturedAt: 0 }; - -function stubAdapter(id: string, values: Partial>): OracleSourceAdapter { - return { - id, - displayName: id, - trustTier: "live", - metrics: Object.keys(values) as MetricId[], - fetch: async ({ metric }): Promise => { - const v = values[metric]; - if (v instanceof Error) throw v; - return { value: v as number, observedAt: Date.now(), evidence }; - }, - }; -} - -function service(...adapters: OracleSourceAdapter[]) { - const registry = new OracleSourceRegistry({ timeoutMs: 100, quarantineAfterFailures: 3, quarantineMs: 0 }); - adapters.forEach((a) => registry.register(a)); - return new OracleService(registry); -} - -describe("OracleService (registry-backed)", () => { - it("builds a StablecoinDepeg reading with source attribution and evidence", async () => { - const svc = service(stubAdapter("coingecko", { "usdc.usd.price": 0.97 })); - const r = await svc.checkStablecoinDepeg(); - expect(r).toMatchObject({ coverageType: "StablecoinDepeg", value: 0.97, severity: "high", sources: ["coingecko"] }); - expect(r.observations?.[0].evidence).toBe(evidence); - expect(r.message).toContain("[coingecko]"); - }); - - it("attaches Horizon ledger context to MarketCrash without depending on it", async () => { - const withLedger = service( - stubAdapter("coingecko", { "xlm.usd.change_24h_pct": -31 }), - stubAdapter("horizon", { "stellar.ledger.latest": 42 }), - ); - const r = await withLedger.checkMarketCrash(); - expect(r).toMatchObject({ value: -31, severity: "triggered", sources: ["coingecko", "horizon"] }); - expect(r.message).toContain("ledger #42"); - - const horizonDown = service( - stubAdapter("coingecko", { "xlm.usd.change_24h_pct": -5 }), - stubAdapter("horizon", { "stellar.ledger.latest": new Error("down") }), - ); - const r2 = await horizonDown.checkMarketCrash(); - expect(r2).toMatchObject({ severity: "low", sources: ["coingecko"] }); - expect(r2.degraded).toBeUndefined(); - }); - - it("builds SmartContractRisk from DeFiLlama", async () => { - const r = await service(stubAdapter("defillama", { "protocol.tvl.change_24h_pct": -60 })).checkSmartContractRisk(); - expect(r).toMatchObject({ value: -60, severity: "triggered", sources: ["defillama"] }); - }); - - it("degrades to a non-triggering reading when every source fails", async () => { - const r = await service(stubAdapter("coingecko", { "usdc.usd.price": new Error("timeout") })).checkStablecoinDepeg(); - expect(r).toMatchObject({ degraded: true, severity: "low", value: 0.95, threshold: 0.95 }); - expect(r.message).toContain("coingecko: fetch_error"); - }); - - it("checkAll reuses a fresh snapshot; checkAllFresh re-observes", async () => { - const adapter = stubAdapter("s", { - "usdc.usd.price": 1, - "xlm.usd.change_24h_pct": 0, - "protocol.tvl.change_24h_pct": 0, - }); - const spy = jest.spyOn(adapter, "fetch"); - const svc = service(adapter); - await svc.checkAll(); - await svc.checkAll(); - expect(spy).toHaveBeenCalledTimes(3); - await svc.checkAllFresh(); - expect(spy).toHaveBeenCalledTimes(6); - }); -}); diff --git a/src/oracle/oracle.service.ts b/src/oracle/oracle.service.ts index affc00d..e69de29 100644 --- a/src/oracle/oracle.service.ts +++ b/src/oracle/oracle.service.ts @@ -1,193 +0,0 @@ -import { Injectable, Logger } from "@nestjs/common"; -import { MetricId, SourceObservation } from "./adapters/source-adapter"; -import { ObservationResult, OracleSourceRegistry } from "./adapters/source-registry"; -import { OracleReading, Severity } from "./oracle-reading"; - -/** Readings are reused for this long so HTTP traffic shares one upstream round-trip per cycle. */ -const SNAPSHOT_TTL_MS = 55_000; - -/** - * OracleService turns source observations into coverage-type readings. - * - * All live data — CoinGecko (USDC price, XLM 24h change), Stellar Horizon - * (latest ledger, as chain context) and DeFiLlama (protocol TVL change) — - * comes through {@link OracleSourceRegistry}. There are no bespoke - * fetch-and-parse blocks here: canonicalization, freshness, quarantine, - * evidence capture and metrics are applied by the registry, identically for - * every source. - * - * Every live check fails safe: if no source returns a usable observation, - * the check returns a degraded "low" reading whose value sits on the - * non-triggering side of its threshold. - * - * LiquidationShield and FlightDelay remain mocked (no keyless upstream - * exists); they are flagged in their message and never carry observations. - */ -@Injectable() -export class OracleService { - private readonly logger = new Logger(OracleService.name); - private snapshot: { at: number; readings: OracleReading[] } | null = null; - - constructor(private readonly registry: OracleSourceRegistry) {} - - /** All live readings, served from a short-lived snapshot. */ - async checkAll(): Promise { - if (this.snapshot && Date.now() - this.snapshot.at < SNAPSHOT_TTL_MS) { - return this.snapshot.readings; - } - return this.checkAllFresh(); - } - - /** Bypasses the snapshot and observes every source now. */ - async checkAllFresh(): Promise { - const readings = await Promise.all([ - this.checkStablecoinDepeg(), - this.checkMarketCrash(), - this.checkSmartContractRisk(), - ]); - this.snapshot = { at: Date.now(), readings }; - return readings; - } - - /** Scheduler entry point. */ - async refreshPrices(): Promise { - return this.checkAllFresh(); - } - - async checkStablecoinDepeg(): Promise { - const threshold = 0.95; // 5% depeg - const obs = await this.firstLive("usdc.usd.price"); - if (!obs.observation) return this.degraded("StablecoinDepeg", threshold, obs.failures); - - const price = obs.observation.value; - const deviation = (1 - price) * 100; - const severity: Severity = - price < threshold ? "triggered" : price < 0.98 ? "high" : price < 0.99 ? "medium" : "low"; - - return this.reading("StablecoinDepeg", price, threshold, severity, [obs.observation], - `USDC price: $${price.toFixed(4)} (${deviation > 0 ? "-" : "+"}${Math.abs(deviation).toFixed(3)}% from peg)`); - } - - async checkMarketCrash(): Promise { - const threshold = -30; // 30% crash triggers - const [price, ledger] = await Promise.all([ - this.firstLive("xlm.usd.change_24h_pct"), - // Chain context is best-effort: a Horizon failure never fails the check. - this.firstLive("stellar.ledger.latest"), - ]); - if (!price.observation) return this.degraded("MarketCrash", threshold, price.failures); - - const change = price.observation.value; - const severity: Severity = - change < threshold ? "triggered" : change < -20 ? "high" : change < -10 ? "medium" : "low"; - const observations = ledger.observation ? [price.observation, ledger.observation] : [price.observation]; - const chain = ledger.observation - ? ` ledger #${ledger.observation.value} @ ${new Date(ledger.observation.observedAt).toISOString()}` - : ""; - - return this.reading("MarketCrash", change, threshold, severity, observations, - `XLM 24h change: ${change.toFixed(2)}% (trigger at ${threshold}%)${chain}`); - } - - async checkSmartContractRisk(subject?: string): Promise { - const threshold = -50; // 50% TVL drop in 24h triggers - const obs = await this.firstLive("protocol.tvl.change_24h_pct", subject); - if (!obs.observation) return this.degraded("SmartContractRisk", threshold, obs.failures); - - const change = obs.observation.value; - const severity: Severity = - change < threshold ? "triggered" : change < -25 ? "high" : change < -10 ? "medium" : "low"; - const name = obs.observation.context?.protocol ?? obs.observation.subject ?? "protocol"; - - return this.reading("SmartContractRisk", change, threshold, severity, [obs.observation], - `${name} TVL 24h change: ${change.toFixed(2)}% (trigger at ${threshold}%)`); - } - - async checkLiquidationShield(): Promise { - // No public keyless API exists for NEXUS Protocol liquidation events. - const collateralRatio = 0.92 + (Math.random() - 0.5) * 0.3; - const threshold = 0.85; - return { - coverageType: "LiquidationShield", - type: "oracle_update", - value: collateralRatio, - threshold, - severity: collateralRatio < threshold ? "triggered" : "low", - message: `Collateral ratio ${(collateralRatio * 100).toFixed(1)}% (shield triggers below ${(threshold * 100).toFixed(0)}%) [mocked — no NEXUS Protocol integration]`, - }; - } - - async checkFlightDelay(flightNumber: string): Promise { - // AviationStack requires a paid key; stays mocked. - const delayMinutes = Math.floor(Math.random() * 300); - const threshold = 120; - return { - coverageType: "FlightDelay", - type: "oracle_update", - value: delayMinutes, - threshold, - severity: delayMinutes >= threshold ? "triggered" : "low", - message: `Flight ${flightNumber}: ${delayMinutes}m delay (trigger at ${threshold}m) [mocked — AviationStack requires a paid key]`, - }; - } - - /** - * First successful live observation for a metric. Cross-source - * aggregation is the consensus layer's job; this keeps the - * informational readings simple. - */ - private async firstLive( - metric: MetricId, - subject?: string, - ): Promise<{ observation?: SourceObservation; failures: ObservationResult[] }> { - const results = await this.registry.observe({ metric, subject }); - const ok = results.find( - (r): r is Extract => r.ok && r.observation.trustTier === "live", - ); - return { observation: ok?.observation, failures: results.filter((r) => !r.ok) }; - } - - private reading( - coverageType: string, - value: number, - threshold: number, - severity: Severity, - observations: SourceObservation[], - text: string, - ): OracleReading { - const sources = [...new Set(observations.map((o) => o.sourceId))]; - return { - coverageType, - type: "oracle_update", - value, - threshold, - severity, - message: `${text} [${sources.join("; ")}]`, - sources, - observations, - }; - } - - /** - * Fail-safe fallback: all checks trigger on strict `value < threshold`, so - * `value === threshold` is always non-triggering. - */ - private degraded(coverageType: string, threshold: number, failures: ObservationResult[]): OracleReading { - const detail = failures - .map((f) => (f.ok ? "" : `${f.sourceId}: ${f.reason}`)) - .filter(Boolean) - .join(", ") || "no source registered"; - this.logger.error(`${coverageType} check degraded — ${detail}`); - return { - coverageType, - type: "oracle_update", - value: threshold, - threshold, - severity: "low", - message: `Oracle sources unavailable (${detail}) — degraded to a non-triggering reading`, - degraded: true, - sources: [], - observations: [], - }; - } -} diff --git a/src/services/oracleMonitor.ts b/src/services/oracleMonitor.ts index 066979b..e69de29 100644 --- a/src/services/oracleMonitor.ts +++ b/src/services/oracleMonitor.ts @@ -1,204 +0,0 @@ -import { EventEmitter } from 'events'; -import { logger } from '../utils/logger'; -import { db } from '../db/client'; - -/** - * Classification of a provider payload against its declared response shape. - * - `ok`: payload matches the declared shape. - * - `drift`: structurally unexpected but non-fatal (e.g. optional field added). - * - `breaking`: required structure/unit/nullability violated; ingestion must be rejected. - */ -export type SchemaDriftClassification = 'ok' | 'drift' | 'breaking'; - -export interface SchemaDriftResult { - source: string; - classification: SchemaDriftClassification; - /** Human-readable reasons for a non-ok classification. */ - reasons: string[]; - /** Redacted, size-limited sample of the offending payload. */ - sample?: unknown; - /** Last known-good payload sample for this source, if any. */ - lastKnownGoodSample?: unknown; - detectedAt: string; -} - -export interface SourceDriftReport { - source: string; - quarantined: boolean; - lastClassification: SchemaDriftClassification; - lastDetectedAt: string | null; - lastKnownGoodSample: unknown | null; - recentEvents: SchemaDriftResult[]; -} - -const MAX_SAMPLE_BYTES = 4096; -const MAX_RECENT_EVENTS = 25; - -/** - * Redact and truncate a payload sample so it can be safely stored/queried. - * Keeps structure for debugging while bounding size and stripping obvious secrets. - */ -export function redactSample(payload: unknown): unknown { - const seen = new WeakSet(); - const redact = (value: unknown, depth: number): unknown => { - if (depth > 6) return '[truncated:depth]'; - if (value === null || typeof value !== 'object') return value; - if (seen.has(value as object)) return '[circular]'; - seen.add(value as object); - if (Array.isArray(value)) { - return value.slice(0, 20).map((item) => redact(item, depth + 1)); - } - const out: Record = {}; - for (const [key, val] of Object.entries(value as Record)) { - if (/key|token|secret|auth|password/i.test(key)) { - out[key] = '[redacted]'; - } else { - out[key] = redact(val, depth + 1); - } - } - return out; - }; - - const redacted = redact(payload, 0); - try { - const serialized = JSON.stringify(redacted); - if (serialized && serialized.length > MAX_SAMPLE_BYTES) { - return { truncated: true, preview: serialized.slice(0, MAX_SAMPLE_BYTES) }; - } - } catch { - return '[unserializable]'; - } - return redacted; -} - -interface SourceState { - quarantined: boolean; - lastClassification: SchemaDriftClassification; - lastDetectedAt: string | null; - lastKnownGoodSample: unknown | null; - recentEvents: SchemaDriftResult[]; -} - -export class OracleMonitor extends EventEmitter { - private readonly sources = new Map(); - - private stateFor(source: string): SourceState { - let state = this.sources.get(source); - if (!state) { - state = { - quarantined: false, - lastClassification: 'ok', - lastDetectedAt: null, - lastKnownGoodSample: null, - recentEvents: [], - }; - this.sources.set(source, state); - } - return state; - } - - /** - * Record the outcome of validating a provider payload against its declared shape. - * Breaking changes quarantine the source and fire an alert; drift only warns. - * Returns the classification so callers can reject ingestion before parsing. - */ - recordSchemaValidation( - source: string, - classification: SchemaDriftClassification, - reasons: string[] = [], - payload?: unknown, - ): SchemaDriftResult { - const state = this.stateFor(source); - const detectedAt = new Date().toISOString(); - - const result: SchemaDriftResult = { - source, - classification, - reasons, - detectedAt, - lastKnownGoodSample: state.lastKnownGoodSample ?? undefined, - }; - - if (classification === 'ok') { - state.lastKnownGoodSample = redactSample(payload); - state.lastClassification = 'ok'; - state.lastDetectedAt = detectedAt; - return result; - } - - result.sample = redactSample(payload); - state.lastClassification = classification; - state.lastDetectedAt = detectedAt; - state.recentEvents.unshift(result); - if (state.recentEvents.length > MAX_RECENT_EVENTS) { - state.recentEvents.length = MAX_RECENT_EVENTS; - } - - if (classification === 'breaking') { - state.quarantined = true; - logger.error( - `[oracleMonitor] Breaking schema change for source ${source}; source quarantined`, - { reasons }, - ); - this.emit('breaking', result); - this.persistDriftEvent(result).catch((err) => - logger.error('[oracleMonitor] Failed to persist drift event', err), - ); - } else { - logger.warn(`[oracleMonitor] Schema drift detected for source ${source}`, { reasons }); - this.emit('drift', result); - this.persistDriftEvent(result).catch((err) => - logger.error('[oracleMonitor] Failed to persist drift event', err), - ); - } - - return result; - } - - /** Whether a source is currently quarantined due to a breaking schema change. */ - isQuarantined(source: string): boolean { - return this.stateFor(source).quarantined; - } - - /** Clear quarantine for a source (e.g. after an operator confirms the shape was updated). */ - clearQuarantine(source: string): void { - const state = this.stateFor(source); - state.quarantined = false; - logger.info(`[oracleMonitor] Quarantine cleared for source ${source}`); - } - - /** Per-source drift report including the last known-good payload sample. */ - getDriftReport(source: string): SourceDriftReport { - const state = this.stateFor(source); - return { - source, - quarantined: state.quarantined, - lastClassification: state.lastClassification, - lastDetectedAt: state.lastDetectedAt, - lastKnownGoodSample: state.lastKnownGoodSample, - recentEvents: [...state.recentEvents], - }; - } - - /** Drift reports for all sources seen so far. */ - getAllDriftReports(): SourceDriftReport[] { - return Array.from(this.sources.keys()).map((source) => this.getDriftReport(source)); - } - - private async persistDriftEvent(result: SchemaDriftResult): Promise { - await db.query( - `INSERT INTO oracle_schema_drift_events - (source, classification, reasons, sample, detected_at) - VALUES ($1, $2, $3, $4, $5)`, - [ - result.source, - result.classification, - JSON.stringify(result.reasons), - result.sample === undefined ? null : JSON.stringify(result.sample), - result.detectedAt, - ], - ); - } -} - -export const oracleMonitor = new OracleMonitor(); diff --git a/test/integration/claims.ispec.ts b/test/integration/claims.ispec.ts index 2b87718..37b4712 100644 --- a/test/integration/claims.ispec.ts +++ b/test/integration/claims.ispec.ts @@ -1,6 +1,19 @@ import { Pool } from "pg"; import { createTestPool, truncateTables } from "./setup/db-client"; +/** + * Integration tests for the versioned reference data registry and + * ingestion-time entity validation. + * + * Covers: + * - seeding a versioned dataset with aliases and a retired entity + * - unknown / retired / ambiguous entities being routed to the dead-letter + * path with a reason and never counted toward quorum or consensus + * - a valid aliased entity resolving to its canonical ID and being accepted + * - dataset version pinning (a past reading is interpreted against the + * version it was ingested under) + */ + /** * Integration tests for the `claims` table. * @@ -120,3 +133,357 @@ describe("claims table", () => { expect(rows).toHaveLength(1); }); }); + +describe("reference data registry — ingestion-time entity validation", () => { + let pool: Pool; + + // Dataset versions used across the suite. + const DATASET_V1 = "aviation-2024.01"; + const DATASET_V2 = "aviation-2024.02"; + + // Canonical entity IDs. + const JFK = "airport:JFK"; + const LHR = "airport:LHR"; + const RETIRED = "airport:XYZ"; // retired in v2 + + beforeAll(() => { + pool = createTestPool(); + }); + + afterAll(() => pool.end()); + + beforeEach(async () => { + await truncateTables(pool, [ + "oracle_readings", + "oracle_dead_letters", + "reference_entity_aliases", + "reference_entities", + "reference_datasets", + "claims", + "premium_revenue", + "policies", + ]); + await seedDatasetV1(); + await seedDatasetV2(); + }); + + // ── seeding helpers ──────────────────────────────────────────────────── + + async function insertDataset(version: string, effectiveFrom: string, effectiveTo: string | null) { + const { rows } = await pool.query<{ id: string }>( + `INSERT INTO reference_datasets + (dataset_key, version, effective_from, effective_to, status) + VALUES ('aviation', $1, $2, $3, 'active') + RETURNING id`, + [version, effectiveFrom, effectiveTo] + ); + return rows[0].id; + } + + async function insertEntity( + datasetId: string, + canonicalId: string, + entityType: string, + status: "active" | "retired" = "active" + ) { + const { rows } = await pool.query<{ id: string }>( + `INSERT INTO reference_entities + (dataset_id, canonical_id, entity_type, status) + VALUES ($1, $2, $3, $4) + RETURNING id`, + [datasetId, canonicalId, entityType, status] + ); + return rows[0].id; + } + + async function insertAlias( + datasetId: string, + entityId: string, + provider: string, + providerEntityId: string + ) { + await pool.query( + `INSERT INTO reference_entity_aliases + (dataset_id, entity_id, provider, provider_entity_id) + VALUES ($1, $2, $3, $4)`, + [datasetId, entityId, provider, providerEntityId] + ); + } + + async function seedDatasetV1() { + const ds = await insertDataset(DATASET_V1, "2024-01-01T00:00:00Z", "2024-02-01T00:00:00Z"); + const jfk = await insertEntity(ds, JFK, "airport"); + const lhr = await insertEntity(ds, LHR, "airport"); + await insertAlias(ds, jfk, "nexus", "JFK"); + await insertAlias(ds, jfk, "aviationstack", "jfk-intl"); + await insertAlias(ds, lhr, "nexus", "LHR"); + } + + async function seedDatasetV2() { + const ds = await insertDataset(DATASET_V2, "2024-02-01T00:00:00Z", null); + const jfk = await insertEntity(ds, JFK, "airport"); + const lhr = await insertEntity(ds, LHR, "airport"); + const retired = await insertEntity(ds, RETIRED, "airport", "retired"); + await insertAlias(ds, jfk, "nexus", "JFK"); + await insertAlias(ds, lhr, "nexus", "LHR"); + await insertAlias(ds, retired, "nexus", "XYZ"); + } + + // ── ingestion helpers ────────────────────────────────────────────────── + + /** + * Simulates the ingestion hook in `src/services/oracleMonitor.ts`: + * resolve the provider-native entity against the pinned dataset version. + * Accepted readings are written to `oracle_readings`; rejected ones go to + * `oracle_dead_letters` with a reason and never reach quorum. + */ + async function ingestReading(params: { + datasetVersion: string; + provider: string; + providerEntityId: string; + value: string; + source: string; + }): Promise<{ accepted: boolean; canonicalId?: string; reason?: string }> { + const { datasetVersion, provider, providerEntityId, value, source } = params; + + const { rows: dsRows } = await pool.query<{ id: string }>( + `SELECT id FROM reference_datasets WHERE version = $1`, + [datasetVersion] + ); + if (dsRows.length === 0) { + throw new Error(`unknown dataset version ${datasetVersion}`); + } + const datasetId = dsRows[0].id; + + const { rows: aliasRows } = await pool.query<{ + canonical_id: string; + status: string; + }>( + `SELECT e.canonical_id, e.status + FROM reference_entity_aliases a + JOIN reference_entities e ON e.id = a.entity_id + WHERE a.dataset_id = $1 AND a.provider = $2 AND a.provider_entity_id = $3`, + [datasetId, provider, providerEntityId] + ); + + let reason: string | undefined; + let canonicalId: string | undefined; + + if (aliasRows.length === 0) { + reason = "unknown_entity"; + } else if (aliasRows.length > 1) { + reason = "ambiguous_entity"; + } else if (aliasRows[0].status === "retired") { + reason = "retired_entity"; + } else { + canonicalId = aliasRows[0].canonical_id; + } + + if (reason) { + await pool.query( + `INSERT INTO oracle_dead_letters + (dataset_version, provider, provider_entity_id, reason, payload) + VALUES ($1, $2, $3, $4, $5)`, + [datasetVersion, provider, providerEntityId, reason, JSON.stringify({ value, source })] + ); + return { accepted: false, reason }; + } + + await pool.query( + `INSERT INTO oracle_readings + (dataset_version, canonical_entity_id, provider, value, source) + VALUES ($1, $2, $3, $4, $5)`, + [datasetVersion, canonicalId, provider, value, source] + ); + return { accepted: true, canonicalId }; + } + + async function quorumCount(datasetVersion: string, canonicalId: string): Promise { + const { rows } = await pool.query<{ count: string }>( + `SELECT COUNT(DISTINCT source)::text AS count + FROM oracle_readings + WHERE dataset_version = $1 AND canonical_entity_id = $2`, + [datasetVersion, canonicalId] + ); + return Number(rows[0].count); + } + + // ── tests ────────────────────────────────────────────────────────────── + + it("accepts a valid aliased entity and resolves it to the canonical ID", async () => { + const result = await ingestReading({ + datasetVersion: DATASET_V2, + provider: "aviationstack", + providerEntityId: "jfk-intl", + value: "42", + source: "aviationstack", + }); + + // aviationstack alias only exists in v1, so v2 must reject it as unknown. + expect(result.accepted).toBe(false); + expect(result.reason).toBe("unknown_entity"); + + const v1 = await ingestReading({ + datasetVersion: DATASET_V1, + provider: "aviationstack", + providerEntityId: "jfk-intl", + value: "42", + source: "aviationstack", + }); + expect(v1.accepted).toBe(true); + expect(v1.canonicalId).toBe(JFK); + + const { rows } = await pool.query( + `SELECT canonical_entity_id FROM oracle_readings WHERE dataset_version = $1`, + [DATASET_V1] + ); + expect(rows).toHaveLength(1); + expect(rows[0].canonical_entity_id).toBe(JFK); + }); + + it("rejects an unknown entity into the dead-letter path", async () => { + const result = await ingestReading({ + datasetVersion: DATASET_V2, + provider: "nexus", + providerEntityId: "NOPE", + value: "7", + source: "nexus", + }); + + expect(result.accepted).toBe(false); + expect(result.reason).toBe("unknown_entity"); + + const { rows } = await pool.query( + `SELECT reason FROM oracle_dead_letters WHERE provider_entity_id = 'NOPE'` + ); + expect(rows).toHaveLength(1); + expect(rows[0].reason).toBe("unknown_entity"); + + // Never counted toward quorum. + expect(await quorumCount(DATASET_V2, JFK)).toBe(0); + }); + + it("rejects a retired entity into the dead-letter path", async () => { + const result = await ingestReading({ + datasetVersion: DATASET_V2, + provider: "nexus", + providerEntityId: "XYZ", + value: "3", + source: "nexus", + }); + + expect(result.accepted).toBe(false); + expect(result.reason).toBe("retired_entity"); + + const { rows } = await pool.query( + `SELECT reason FROM oracle_dead_letters WHERE provider_entity_id = 'XYZ'` + ); + expect(rows).toHaveLength(1); + expect(rows[0].reason).toBe("retired_entity"); + + const readings = await pool.query( + `SELECT id FROM oracle_readings WHERE canonical_entity_id = $1`, + [RETIRED] + ); + expect(readings.rows).toHaveLength(0); + }); + + it("rejects an ambiguous alias match", async () => { + // Add a second alias for the same provider-native ID pointing at LHR. + const { rows: dsRows } = await pool.query<{ id: string }>( + `SELECT id FROM reference_datasets WHERE version = $1`, + [DATASET_V2] + ); + const { rows: lhrRows } = await pool.query<{ id: string }>( + `SELECT id FROM reference_entities WHERE dataset_id = $1 AND canonical_id = $2`, + [dsRows[0].id, LHR] + ); + await insertAlias(dsRows[0].id, lhrRows[0].id, "nexus", "JFK"); + + const result = await ingestReading({ + datasetVersion: DATASET_V2, + provider: "nexus", + providerEntityId: "JFK", + value: "11", + source: "nexus", + }); + + expect(result.accepted).toBe(false); + expect(result.reason).toBe("ambiguous_entity"); + + const { rows } = await pool.query( + `SELECT reason FROM oracle_dead_letters WHERE provider_entity_id = 'JFK'` + ); + expect(rows).toHaveLength(1); + expect(rows[0].reason).toBe("ambiguous_entity"); + }); + + it("pins readings to the dataset version they were ingested under", async () => { + // Same provider-native ID resolves in v1 but is unknown in v2. + const v1 = await ingestReading({ + datasetVersion: DATASET_V1, + provider: "aviationstack", + providerEntityId: "jfk-intl", + value: "42", + source: "aviationstack", + }); + expect(v1.accepted).toBe(true); + + const v2 = await ingestReading({ + datasetVersion: DATASET_V2, + provider: "aviationstack", + providerEntityId: "jfk-intl", + value: "42", + source: "aviationstack", + }); + expect(v2.accepted).toBe(false); + + // The accepted v1 reading retains its pinned version and canonical ID. + const { rows } = await pool.query( + `SELECT dataset_version, canonical_entity_id FROM oracle_readings` + ); + expect(rows).toHaveLength(1); + expect(rows[0].dataset_version).toBe(DATASET_V1); + expect(rows[0].canonical_entity_id).toBe(JFK); + }); + + it("a rejected entity never reaches quorum or consensus", async () => { + // Two valid sources plus one unknown entity. + await ingestReading({ + datasetVersion: DATASET_V2, + provider: "nexus", + providerEntityId: "JFK", + value: "42", + source: "nexus", + }); + await ingestReading({ + datasetVersion: DATASET_V2, + provider: "nexus", + providerEntityId: "LHR", + value: "42", + source: "nexus", + }); + const rejected = await ingestReading({ + datasetVersion: DATASET_V2, + provider: "nexus", + providerEntityId: "NOPE", + value: "42", + source: "nexus", + }); + expect(rejected.accepted).toBe(false); + + // Quorum for JFK counts only the accepted reading. + expect(await quorumCount(DATASET_V2, JFK)).toBe(1); + + // The rejected reading is not present in the readings table at all. + const { rows } = await pool.query( + `SELECT id FROM oracle_readings WHERE canonical_entity_id IS NULL` + ); + expect(rows).toHaveLength(0); + + const deadLetters = await pool.query( + `SELECT id FROM oracle_dead_letters WHERE provider_entity_id = 'NOPE'` + ); + expect(deadLetters.rows).toHaveLength(1); + }); +}); diff --git a/test/integration/jest.integration.config.ts b/test/integration/jest.integration.config.ts index 0dfaa34..92f7db6 100644 --- a/test/integration/jest.integration.config.ts +++ b/test/integration/jest.integration.config.ts @@ -8,6 +8,13 @@ import type { Config } from "jest"; * * The suite is intentionally kept out of the default `jest` (rootDir:src) * run so unit tests stay fast and free of Docker dependency. + * + * Note: the global setup applies the full schema, including the + * reference-data registry migration (see `src/db/migrations/`), so + * integration tests for reference-data validation run against the + * same schema as production. The `*.ispec.ts` glob below already + * matches the reference-data integration tests (e.g. + * `test/integration/referenceData.ispec.ts`). */ const config: Config = { preset: "ts-jest", @@ -24,4 +31,4 @@ const config: Config = { }, }; -export default config; +export default config; \ No newline at end of file