Fix: Build a Versioned Reference Data Registry for Feed Entity Validation

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 <path> --kind <kind> --version <ver> --effective-from <iso> [--effective-to <iso>] [--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<string, string>[] {
+  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<string, string> = {};
+    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<void> {
+  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;
+  });
+}

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<Record<string, string | null>>((acc, feed) => {
+    acc[feed] = pinnedReferenceDataVersion(feed);
+    return acc;
+  }, {}),
+}));

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_<FEED>=<dataset>@<version>`
+     * (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_<FEED>` keys; each must be a
+    // `<dataset>@<version>` pin. Unknown keys are rejected below.
+    z.unknown(),
+  )
   .superRefine((env, ctx) => {
+    const raw = env as Record<string, unknown>;
+    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 <dataset>@<version> (e.g. aviation@3)`,
+        });
+      }
+    }
+
     if (!env.REQUIRE_CHAIN_CONFIG) {
       return;
     }
@@ -192,6 +245,34 @@ export const envSchema = z
 
 export type Env = z.infer<typeof envSchema>;
 
+/**
+ * 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<string, { dataset: string; version: number }> {
+  const pins: Record<string, { dataset: string; version: number }> = {};
+  for (const [key, value] of Object.entries(env as Record<string, unknown>)) {
+    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.
  *