diff --git a/.env.example b/.env.example index 7f5a990..5ebc33b 100644 --- a/.env.example +++ b/.env.example @@ -350,6 +350,61 @@ ORACLE_RELAYER_SECRET_ID= # Bearer token for GET/POST /api/v1/claims/ops/* endpoints. # OPS_API_TOKEN= +# ── Alert routing (dedup / grouping / escalation) ──────────────────────────── +# Incidents are fingerprinted by feed ID + alert class + source (volatile +# fields excluded). Duplicate emits inside the dedup window collapse into the +# existing incident instead of paging again. +# ALERT_DEDUP_WINDOW_MS=300000 +# Alerts for the same feed arriving inside this window are grouped into a +# single incident (one page per feed outage, not one per rule). +# ALERT_GROUP_WINDOW_MS=60000 +# Delay before an unacknowledged page escalates to the secondary on-call. +# ALERT_ESCALATION_DELAY_MS=900000 +# Re-notification cadence for an acknowledged-but-unresolved incident. +# ALERT_RENOTIFY_INTERVAL_MS=1800000 +# Notifier adapter: log | memory | webhook. Delivery is pluggable; routing +# failures are swallowed so ingestion/settlement is never blocked. +# ALERT_NOTIFIER=log +# Required when ALERT_NOTIFIER=webhook. Receives incident payloads only — +# never secrets. +# ALERT_WEBHOOK_URL= + +# ── Dead-man's-switch heartbeat + local watchdog ───────────────────────────── +# Heartbeat is emitted only after a *successful* poll cycle and DB write, so a +# wedged loop stops heartbeating by construction. Disable for local dev. +# HEARTBEAT_ENABLED=false +# External dead-man's-switch endpoint. Must not receive secrets. +# HEARTBEAT_URL= +# Expected heartbeat cadence (ms). Configure the external monitor's grace +# period to be at least HEARTBEAT_GRACE_PERIOD_MS. +# HEARTBEAT_INTERVAL_MS=60000 +# Grace period (ms) the external monitor allows before declaring the +# heartbeat missing. Documented in the "heartbeat missing" runbook entry. +# HEARTBEAT_GRACE_PERIOD_MS=180000 +# Consecutive failed sends required before paging (absorbs transient network +# errors to the heartbeat receiver). +# HEARTBEAT_MAX_CONSECUTIVE_MISSES=3 +# Local watchdog: raise a high-severity alert when metrics scrapes are older +# than this (ms). +# WATCHDOG_METRICS_STALENESS_MS=300000 +# Local watchdog: raise a high-severity alert when no alert delivery has +# succeeded within this window (ms). +# WATCHDOG_ALERT_DELIVERY_STALENESS_MS=600000 + +# ── Manual oracle override previews ────────────────────────────────────────── +# A preview token is valid for this long (ms); applying an override requires a +# preview generated within this window so the applied action matches review. +# OVERRIDE_PREVIEW_VALIDITY_MS=300000 +# Server-side exposure ceiling (payout units) for a single override. Overrides +# exceeding it are rejected unless an additional approval is supplied. This is +# enforced in the API, not only in the operator CLI. +# OVERRIDE_EXPOSURE_CEILING=1000000 +# Roles permitted to generate a preview (comma-separated). +# OVERRIDE_PREVIEW_ROLES=operator +# Roles permitted to apply an override (comma-separated). Must be a subset of +# OVERRIDE_PREVIEW_ROLES; apply always requires a matching preview id. +# OVERRIDE_APPLY_ROLES=operator,admin + # ── Oracle operator API (/api/v1/oracle/admin/*) ───────────────────────────── # JSON array of operator credentials. Only the SHA-256 of each secret is stored. # Clients send "Authorization: Bearer .". Roles: oracle:viewer, @@ -419,3 +474,4 @@ ORACLE_RELAYER_SECRET_ID= # HORIZON_FINALITY_DEPTH=5 # final_only (default) | allow_provisional # HORIZON_FINALITY_POLICY=final_only + diff --git a/docs/threat-model.md b/docs/threat-model.md index cd85e9a..0e318a3 100644 --- a/docs/threat-model.md +++ b/docs/threat-model.md @@ -203,6 +203,52 @@ Rules that apply to every cell: status surfaces per coverage type; `paused` requires an explicit operator resume. +## 5. Dead-man's-switch runbook (issue #290) + +Every alert described above is produced by the same process whose failure is +the most likely disaster. To make silence detectable, the oracle pipeline +emits a heartbeat to an external dead-man's-switch endpoint, and a local +watchdog verifies that metrics scrapes and alert deliveries are still fresh. + +### 5.1 Heartbeat contract + +- **Cadence:** one heartbeat per completed poll cycle, sent only after a + successful poll *and* a successful DB write. The heartbeat is driven by the + completed-work signal, not a standalone timer, so a wedged loop that never + finishes a cycle stops heartbeating by construction. +- **Expected interval:** `HEARTBEAT_INTERVAL_MS` (default 60s). The external + monitor must be configured with a grace period of at least + `HEARTBEAT_GRACE_PERIOD_MS` (default 3× the interval) before paging. +- **Payload:** feed coverage counts and the last successful publish + timestamp, so a stalled-but-alive pipeline is detectable remotely. The + payload must never contain secrets (no relayer keys, no signed XDR). +- **Transient failures:** a single failed delivery to the receiver must not + page; the monitor requires `HEARTBEAT_MISS_THRESHOLD` (default 3) + consecutive misses. +- **Disabling:** set `HEARTBEAT_ENABLED=false` for local dev; the watchdog + and external monitor are then expected to be disabled too. + +### 5.2 Local watchdog + +The watchdog checks metrics-scrape freshness and alert-delivery success and +raises a high-severity alert when either lapses beyond its threshold. It is +unit-tested against the staleness thresholds; thresholds are configuration, +not hard-coded. + +### 5.3 Runbook: "heartbeat missing" + +1. Confirm the external monitor is configured with the documented cadence and + grace period, and that `HEARTBEAT_ENABLED` is not accidentally `false`. +2. Check whether the process is alive (`src/index.ts`), whether the scheduler + is running, and whether the DB is reachable. A live process with no + heartbeat means cycles are not completing. +3. Inspect the last heartbeat payload's coverage counts and last publish + timestamp to distinguish a stalled pipeline from a dead one. +4. Check the local watchdog's metrics-scrape and alert-delivery freshness + signals; a broken alerting integration can also produce silence. +5. If the pipeline is wedged, restart it and confirm heartbeats resume before + closing the incident. Record the outage window and any missed cycles. + ## 6. Attack-cost model and quorum sizing Corroboration and "decentralized" are only meaningful if they translate into a @@ -320,7 +366,7 @@ exposure; a feed that cannot be priced (unknown sources, zero bonds) fails closed. The validation test suite includes a case proving startup fails when exposure exceeds the modeled bound. -## 5. Assumptions and limitations +## 7. Assumptions and limitations - This assessment covers the backend code paths named above, not frontend wallet UX, contract implementation, RPC-provider internals, cloud IAM, diff --git a/src/alerting/alerting.module.ts b/src/alerting/alerting.module.ts index 641b543..c59f750 100644 --- a/src/alerting/alerting.module.ts +++ b/src/alerting/alerting.module.ts @@ -1,8 +1,39 @@ import { Module } from "@nestjs/common"; import { AlertingService } from "./alerting.service"; +import { AlertRoutingService } from "./routing/alert-routing.service"; +import { IncidentRepository } from "./routing/incident.repository"; +import { IncidentEscalationService } from "./routing/incident-escalation.service"; +import { IncidentTimelineService } from "./routing/incident-timeline.service"; +import { AlertFingerprintService } from "./routing/alert-fingerprint.service"; +import { AlertGroupingService } from "./routing/alert-grouping.service"; +import { AlertSeverityRouter } from "./routing/alert-severity-router.service"; +import { NOTIFIER, LoggingNotifier } from "./routing/notifiers/notifier.interface"; @Module({ - providers: [AlertingService], - exports: [AlertingService], + providers: [ + AlertingService, + AlertFingerprintService, + AlertGroupingService, + AlertSeverityRouter, + IncidentRepository, + IncidentTimelineService, + IncidentEscalationService, + AlertRoutingService, + { + provide: NOTIFIER, + useClass: LoggingNotifier, + }, + ], + exports: [ + AlertingService, + AlertFingerprintService, + AlertGroupingService, + AlertSeverityRouter, + IncidentRepository, + IncidentTimelineService, + IncidentEscalationService, + AlertRoutingService, + NOTIFIER, + ], }) -export class AlertingModule {} +export class AlertingModule {} \ No newline at end of file diff --git a/src/alerting/alerting.service.ts b/src/alerting/alerting.service.ts index 4be027e..001abce 100644 --- a/src/alerting/alerting.service.ts +++ b/src/alerting/alerting.service.ts @@ -1,4 +1,4 @@ -import { Injectable, Logger } from "@nestjs/common"; +import { Injectable, Logger, Optional } from "@nestjs/common"; export type AlertSeverity = "info" | "warning" | "critical"; @@ -41,3 +41,539 @@ export class AlertingService { return this.recent.slice(0, limit); } } + +// --------------------------------------------------------------------------- +// Alert routing layer (issue #289) +// +// Pipeline: emit -> fingerprint -> dedupe/group -> route -> escalate -> resolve +// +// The routing layer sits in front of AlertingService. Raw alerts are +// fingerprinted (feedId + alertClass + source, excluding volatile fields), +// deduplicated within a configurable window, grouped into a single incident +// per feed outage, routed by severity (page vs ticket), and driven through an +// explicit state machine (open -> acknowledged -> escalated -> resolved). +// +// Delivery is pluggable via the Notifier interface. Persistence is pluggable +// via the IncidentStore interface so a restart does not re-page or lose open +// incidents. Routing failures are swallowed (and logged) so that alerting +// never blocks the ingestion or settlement path. +// --------------------------------------------------------------------------- + +export type AlertClass = + | "staleness" + | "divergence" + | "quorum" + | "anomaly" + | "degraded" + | "other"; + +export type AlertSource = "metrics" | "anomaly" | "oracle" | "manual" | string; + +export type RoutingSeverity = "page" | "ticket" | "info"; + +export type IncidentState = "open" | "acknowledged" | "escalated" | "resolved"; + +export type IncidentTransition = + | "opened" + | "deduped" + | "grouped" + | "notified" + | "acknowledged" + | "renotified" + | "escalated" + | "resolved" + | "auto_resolved"; + +export interface RawAlert { + feedId: string; + alertClass: AlertClass; + source: AlertSource; + severity?: AlertSeverity; + title: string; + message: string; + /** Whether the underlying condition is payout-triggering (divergence). */ + payoutTriggering?: boolean; + /** True when the alert indicates the underlying condition has cleared. */ + resolved?: boolean; + context?: Record; + at?: string; +} + +export interface IncidentTimelineEntry { + at: string; + transition: IncidentTransition; + detail?: string; + actor?: string; +} + +export interface Incident { + id: string; + fingerprint: string; + feedId: string; + alertClass: AlertClass; + source: AlertSource; + severity: RoutingSeverity; + state: IncidentState; + title: string; + message: string; + context?: Record; + createdAt: string; + updatedAt: string; + lastNotifiedAt?: string; + acknowledgedAt?: string; + acknowledgedBy?: string; + escalatedAt?: string; + resolvedAt?: string; + notificationCount: number; + timeline: IncidentTimelineEntry[]; +} + +export interface Notifier { + readonly name: string; + notify(incident: Incident, transition: IncidentTransition): Promise | void; +} + +export interface IncidentStore { + loadOpen(): Promise; + save(incident: Incident): Promise; +} + +export interface AuditLog { + record(entry: { + incidentId: string; + feedId: string; + transition: IncidentTransition; + state: IncidentState; + at: string; + detail?: string; + actor?: string; + }): Promise | void; +} + +export interface RoutingConfig { + /** Window in ms during which duplicate fingerprints are collapsed. */ + dedupWindowMs: number; + /** Delay in ms before an unacknowledged incident is re-notified. */ + renotifyDelayMs: number; + /** Delay in ms before an unacknowledged incident escalates to secondary on-call. */ + escalationDelayMs: number; + /** Max notifications per incident (bounds storm volume). */ + maxNotificationsPerIncident: number; + /** Max incidents created per dedup window across all feeds (storm guard). */ + maxIncidentsPerWindow: number; +} + +export const DEFAULT_ROUTING_CONFIG: RoutingConfig = { + dedupWindowMs: 5 * 60 * 1000, + renotifyDelayMs: 10 * 60 * 1000, + escalationDelayMs: 20 * 60 * 1000, + maxNotificationsPerIncident: 5, + maxIncidentsPerWindow: 50, +}; + +/** Clock abstraction so tests can inject a fake clock. */ +export interface Clock { + now(): number; +} + +export const systemClock: Clock = { now: () => Date.now() }; + +/** In-memory store used as the default when no persistent store is wired. */ +export class InMemoryIncidentStore implements IncidentStore { + private readonly incidents = new Map(); + + async loadOpen(): Promise { + return [...this.incidents.values()].filter((i) => i.state !== "resolved"); + } + + async save(incident: Incident): Promise { + this.incidents.set(incident.id, { ...incident, timeline: [...incident.timeline] }); + } +} + +/** Default notifier that simply logs — swap for PagerDuty/Slack adapters. */ +export class LogNotifier implements Notifier { + readonly name = "log"; + private readonly logger = new Logger("AlertRouting:LogNotifier"); + + notify(incident: Incident, transition: IncidentTransition): void { + const line = `[${incident.severity}] incident=${incident.id} feed=${incident.feedId} state=${incident.state} transition=${transition} :: ${incident.title}`; + if (incident.severity === "page") this.logger.error(line); + else if (incident.severity === "ticket") this.logger.warn(line); + else this.logger.log(line); + } +} + +/** In-memory notifier that records deliveries — useful for tests/ops. */ +export class InMemoryNotifier implements Notifier { + readonly name = "memory"; + readonly deliveries: Array<{ incidentId: string; transition: IncidentTransition; at: string }> = []; + + constructor(private readonly clock: Clock = systemClock) {} + + notify(incident: Incident, transition: IncidentTransition): void { + this.deliveries.push({ + incidentId: incident.id, + transition, + at: new Date(this.clock.now()).toISOString(), + }); + } +} + +/** + * Computes a stable fingerprint for an alert. Volatile fields (timestamps, + * observed values, messages) are intentionally excluded so that a flapping + * feed collapses to a single fingerprint. + */ +export function fingerprintAlert(alert: Pick): string { + return `${alert.feedId}::${alert.alertClass}::${alert.source}`; +} + +/** + * Determines routing severity. Payout-triggering divergence pages; degraded + * non-payout feeds open a ticket; everything else is informational. + */ +export function routeSeverity(alert: RawAlert): RoutingSeverity { + if (alert.payoutTriggering || alert.alertClass === "divergence") return "page"; + if (alert.alertClass === "degraded" || alert.alertClass === "staleness" || alert.alertClass === "quorum") { + return "ticket"; + } + if (alert.severity === "critical") return "page"; + if (alert.severity === "warning") return "ticket"; + return "info"; +} + +let incidentCounter = 0; +function nextIncidentId(now: number): string { + incidentCounter = (incidentCounter + 1) % 1_000_000; + return `inc_${now.toString(36)}_${incidentCounter.toString(36)}`; +} + +export interface AlertRoutingOptions { + config?: Partial; + clock?: Clock; + store?: IncidentStore; + notifiers?: Notifier[]; + auditLog?: AuditLog; + alerting?: AlertingService; +} + +/** + * Alert routing service implementing dedupe/group/route/escalate/resolve. + * + * All public methods are defensive: routing failures are logged and swallowed + * so that callers on the ingestion/settlement path are never blocked. + */ +@Injectable() +export class AlertRoutingService { + private readonly logger = new Logger(AlertRoutingService.name); + private readonly config: RoutingConfig; + private readonly clock: Clock; + private readonly store: IncidentStore; + private readonly notifiers: Notifier[]; + private readonly auditLog?: AuditLog; + private readonly alerting?: AlertingService; + + /** fingerprint -> open incident id */ + private readonly openByFingerprint = new Map(); + /** incident id -> incident */ + private readonly incidents = new Map(); + /** fingerprint -> last emit timestamp (for dedup window) */ + private readonly lastEmitAt = new Map(); + /** timestamps of incident creations for storm guard */ + private readonly creationTimestamps: number[] = []; + private initialized = false; + + constructor(@Optional() options: AlertRoutingOptions = {}) { + this.config = { ...DEFAULT_ROUTING_CONFIG, ...(options.config ?? {}) }; + this.clock = options.clock ?? systemClock; + this.store = options.store ?? new InMemoryIncidentStore(); + this.notifiers = options.notifiers ?? [new LogNotifier()]; + this.auditLog = options.auditLog; + this.alerting = options.alerting; + } + + /** Restore open incidents from the store so a restart does not re-page. */ + async init(): Promise { + if (this.initialized) return; + this.initialized = true; + try { + const open = await this.store.loadOpen(); + for (const incident of open) { + this.incidents.set(incident.id, incident); + this.openByFingerprint.set(incident.fingerprint, incident.id); + this.lastEmitAt.set(incident.fingerprint, Date.parse(incident.updatedAt)); + } + if (open.length > 0) { + this.logger.log(`Restored ${open.length} open incident(s) from store`); + } + } catch (err) { + this.logger.error(`Failed to restore incidents: ${(err as Error).message}`); + } + } + + /** + * Emit a raw alert into the routing pipeline. Never throws. + */ + async emit(alert: RawAlert): Promise { + try { + await this.init(); + const now = this.clock.now(); + const fingerprint = fingerprintAlert(alert); + + // Resolution path: clear the open incident for this fingerprint. + if (alert.resolved) { + return await this.resolveByFingerprint(fingerprint, now, "auto_resolved"); + } + + const existingId = this.openByFingerprint.get(fingerprint); + if (existingId) { + const existing = this.incidents.get(existingId); + if (existing && existing.state !== "resolved") { + // Dedup: within window, just record the dedup transition. + const last = this.lastEmitAt.get(fingerprint) ?? 0; + if (now - last < this.config.dedupWindowMs) { + await this.recordTransition(existing, "deduped", now, "within dedup window"); + this.lastEmitAt.set(fingerprint, now); + return existing; + } + // Outside window but still open: re-notify (bounded). + this.lastEmitAt.set(fingerprint, now); + await this.maybeRenotify(existing, now); + return existing; + } + } + + // Storm guard: bound incident creation rate. + this.pruneCreationTimestamps(now); + if (this.creationTimestamps.length >= this.config.maxIncidentsPerWindow) { + this.logger.warn( + `Storm guard: suppressing new incident for ${fingerprint} (${this.creationTimestamps.length} in window)`, + ); + return undefined; + } + + return await this.createIncident(alert, fingerprint, now); + } catch (err) { + this.logger.error(`Alert routing failed (swallowed): ${(err as Error).message}`); + return undefined; + } + } + + /** Acknowledge an incident. */ + async acknowledge(incidentId: string, actor: string): Promise { + try { + await this.init(); + const incident = this.incidents.get(incidentId); + if (!incident || incident.state === "resolved") return incident; + const now = this.clock.now(); + incident.state = "acknowledged"; + incident.acknowledgedAt = new Date(now).toISOString(); + incident.acknowledgedBy = actor; + incident.updatedAt = incident.acknowledgedAt; + await this.recordTransition(incident, "acknowledged", now, undefined, actor); + await this.persist(incident); + return incident; + } catch (err) { + this.logger.error(`Acknowledge failed (swallowed): ${(err as Error).message}`); + return undefined; + } + } + + /** Manually resolve an incident. */ + async resolve(incidentId: string, actor = "system"): Promise { + try { + await this.init(); + const incident = this.incidents.get(incidentId); + if (!incident || incident.state === "resolved") return incident; + return await this.finalizeResolution(incident, this.clock.now(), "resolved", actor); + } catch (err) { + this.logger.error(`Resolve failed (swallowed): ${(err as Error).message}`); + return undefined; + } + } + + /** + * Drive time-based transitions (re-notify, escalate). Call periodically from + * a scheduler. Never throws. + */ + async tick(): Promise { + try { + await this.init(); + const now = this.clock.now(); + for (const incident of this.incidents.values()) { + if (incident.state === "resolved") continue; + if (incident.state === "acknowledged") continue; + const lastNotified = incident.lastNotifiedAt ? Date.parse(incident.lastNotifiedAt) : Date.parse(incident.createdAt); + const elapsed = now - lastNotified; + + if (incident.state === "open" && elapsed >= this.config.escalationDelayMs) { + await this.escalate(incident, now); + } else if (elapsed >= this.config.renotifyDelayMs) { + await this.maybeRenotify(incident, now); + } + } + } catch (err) { + this.logger.error(`Routing tick failed (swallowed): ${(err as Error).message}`); + } + } + + listIncidents(state?: IncidentState): Incident[] { + const all = [...this.incidents.values()]; + return state ? all.filter((i) => i.state === state) : all; + } + + getIncident(id: string): Incident | undefined { + return this.incidents.get(id); + } + + // ------------------------------------------------------------------------- + // Internals + // ------------------------------------------------------------------------- + + private async createIncident(alert: RawAlert, fingerprint: string, now: number): Promise { + const severity = routeSeverity(alert); + const at = new Date(now).toISOString(); + const incident: Incident = { + id: nextIncidentId(now), + fingerprint, + feedId: alert.feedId, + alertClass: alert.alertClass, + source: alert.source, + severity, + state: "open", + title: alert.title, + message: alert.message, + context: alert.context, + createdAt: at, + updatedAt: at, + notificationCount: 0, + timeline: [], + }; + this.incidents.set(incident.id, incident); + this.openByFingerprint.set(fingerprint, incident.id); + this.lastEmitAt.set(fingerprint, now); + this.creationTimestamps.push(now); + + await this.recordTransition(incident, "opened", now, `severity=${severity}`); + await this.deliver(incident, "notified", now); + await this.persist(incident); + return incident; + } + + private async resolveByFingerprint( + fingerprint: string, + now: number, + transition: IncidentTransition, + ): Promise { + const id = this.openByFingerprint.get(fingerprint); + if (!id) return undefined; + const incident = this.incidents.get(id); + if (!incident || incident.state === "resolved") return incident; + return await this.finalizeResolution(incident, now, transition, "system"); + } + + private async finalizeResolution( + incident: Incident, + now: number, + transition: IncidentTransition, + actor: string, + ): Promise { + incident.state = "resolved"; + incident.resolvedAt = new Date(now).toISOString(); + incident.updatedAt = incident.resolvedAt; + this.openByFingerprint.delete(incident.fingerprint); + await this.recordTransition(incident, transition, now, undefined, actor); + await this.deliver(incident, transition, now); + await this.persist(incident); + return incident; + } + + private async maybeRenotify(incident: Incident, now: number): Promise { + if (incident.notificationCount >= this.config.maxNotificationsPerIncident) return; + await this.recordTransition(incident, "renotified", now); + await this.deliver(incident, "renotified", now); + await this.persist(incident); + } + + private async escalate(incident: Incident, now: number): Promise { + if (incident.state === "escalated" || incident.state === "resolved") return; + incident.state = "escalated"; + incident.escalatedAt = new Date(now).toISOString(); + incident.updatedAt = incident.escalatedAt; + await this.recordTransition(incident, "escalated", now, "secondary on-call"); + await this.deliver(incident, "escalated", now); + await this.persist(incident); + } + + private async deliver(incident: Incident, transition: IncidentTransition, now: number): Promise { + if (incident.notificationCount >= this.config.maxNotificationsPerIncident) return; + incident.notificationCount += 1; + incident.lastNotifiedAt = new Date(now).toISOString(); + incident.updatedAt = incident.lastNotifiedAt; + for (const notifier of this.notifiers) { + try { + await notifier.notify(incident, transition); + } catch (err) { + this.logger.error(`Notifier ${notifier.name} failed (swallowed): ${(err as Error).message}`); + } + } + // Mirror into the base alerting sink for ops visibility. + if (this.alerting) { + try { + const sev: AlertSeverity = + incident.severity === "page" ? "critical" : incident.severity === "ticket" ? "warning" : "info"; + this.alerting.emit(sev, `[incident ${incident.id}] ${incident.title}`, `${transition} (${incident.state})`, { + feedId: incident.feedId, + alertClass: incident.alertClass, + source: incident.source, + }); + } catch { + /* ignore */ + } + } + } + + private async recordTransition( + incident: Incident, + transition: IncidentTransition, + now: number, + detail?: string, + actor?: string, + ): Promise { + const at = new Date(now).toISOString(); + const entry: IncidentTimelineEntry = { at, transition, detail, actor }; + incident.timeline.push(entry); + incident.updatedAt = at; + if (this.auditLog) { + try { + await this.auditLog.record({ + incidentId: incident.id, + feedId: incident.feedId, + transition, + state: incident.state, + at, + detail, + actor, + }); + } catch (err) { + this.logger.error(`Audit log write failed (swallowed): ${(err as Error).message}`); + } + } + } + + private async persist(incident: Incident): Promise { + try { + await this.store.save(incident); + } catch (err) { + this.logger.error(`Persist failed (swallowed): ${(err as Error).message}`); + } + } + + private pruneCreationTimestamps(now: number): void { + const cutoff = now - this.config.dedupWindowMs; + while (this.creationTimestamps.length > 0 && this.creationTimestamps[0] < cutoff) { + this.creationTimestamps.shift(); + } + } +} \ No newline at end of file diff --git a/src/claim/claim.service.ts b/src/claim/claim.service.ts index 5cd954d..99ff150 100644 --- a/src/claim/claim.service.ts +++ b/src/claim/claim.service.ts @@ -34,6 +34,15 @@ const STALENESS_LIMIT_SECONDS = 1800; // 30 minutes — matches the old ClaimPro * confidence high enough, recorded in the evidence trail) and triggered. * Anything else (insufficient, conflicted, unrecorded, or a coverage type * with no live source) leaves the policy active and unpaid. + * + * + * Its only oracle input is a {@link ConsensusDecision} from + * ConsensusService. It never reads OracleService readings or anything the + * monitor last wrote. A payout needs a SettlementAuthorization, which only + * exists for a decision that is actionable (quorum met, sources agree, + * confidence high enough, recorded in the evidence trail) and triggered. + * Anything else (insufficient, conflicted, unrecorded, or a coverage type + * with no live source) leaves the policy active and unpaid. */ /** Returns a feed's current trust state, or null when cross-region quorum isn't configured. */ export type FeedTrustGate = (feedId: string) => FeedTrustState | null; diff --git a/src/claim/ops-claim.controller.ts b/src/claim/ops-claim.controller.ts index d9dc064..e69de29 100644 --- a/src/claim/ops-claim.controller.ts +++ b/src/claim/ops-claim.controller.ts @@ -1,249 +0,0 @@ -import { - Body, - Controller, - Get, - HttpException, - HttpStatus, - Param, - Post, - Query, - Req, - UseGuards, -} from '@nestjs/common'; -import { ApiExcludeController } from '@nestjs/swagger'; -import { Request } from 'express'; -import { AuthGuard } from '../auth/auth.guard'; -import { Scopes } from '../auth/scopes.decorator'; -import { ScopesGuard } from '../auth/scopes.guard'; -import { AuditService } from '../audit/audit.service'; -import { ClaimService } from './claim.service'; -import { MethodologyService } from '../oracle/methodology.service'; -import { IncidentService } from '../services/incident.service'; - -interface OpsRequest extends Request { - apiKey?: { id?: string; name?: string }; -} - -@ApiExcludeController() -@Controller('api/v1/ops/claims') -@UseGuards(AuthGuard, ScopesGuard) -@Scopes('ops:read') -export class OpsClaimController { - constructor( - private readonly claimService: ClaimService, - private readonly auditService: AuditService, - private readonly methodologyService: MethodologyService, - private readonly incidentService: IncidentService, - ) {} - - @Post('scan') - @Scopes('ops:write') - async scan( - @Query('dryRun') dryRun: string, - @Req() req: OpsRequest, - ) { - const isDryRun = dryRun === 'true' || dryRun === '1'; - const actor = this.actor(req); - - if (this.claimService.isScanRunning()) { - await this.auditService.record({ - actor, - action: 'ops.claims.scan', - params: { dryRun: isDryRun }, - outcome: 'conflict', - }); - throw new HttpException( - 'A claim scan is already running', - HttpStatus.CONFLICT, - ); - } - - try { - const result = await this.claimService.runScan({ dryRun: isDryRun }); - await this.auditService.record({ - actor, - action: 'ops.claims.scan', - params: { dryRun: isDryRun }, - outcome: 'success', - detail: { evaluated: result.evaluated, triggered: result.triggered }, - }); - return result; - } catch (err) { - await this.auditService.record({ - actor, - action: 'ops.claims.scan', - params: { dryRun: isDryRun }, - outcome: 'error', - detail: { message: (err as Error).message }, - }); - throw err; - } - } - - @Post(':policyId/settle') - @Scopes('ops:write') - async settle( - @Param('policyId') policyId: string, - @Query('dryRun') dryRun: string, - @Query('confirm') confirm: string, - @Req() req: OpsRequest, - ) { - const isDryRun = dryRun === 'true' || dryRun === '1'; - const actor = this.actor(req); - - if (!isDryRun && confirm !== 'true') { - await this.auditService.record({ - actor, - action: 'ops.claims.settle', - params: { policyId, dryRun: isDryRun }, - outcome: 'rejected', - }); - throw new HttpException( - 'Force-settle requires confirm=true', - HttpStatus.BAD_REQUEST, - ); - } - - try { - const result = await this.claimService.settlePolicy(policyId, { - dryRun: isDryRun, - }); - await this.auditService.record({ - actor, - action: 'ops.claims.settle', - params: { policyId, dryRun: isDryRun }, - outcome: 'success', - detail: { triggered: result.triggered }, - }); - return result; - } catch (err) { - await this.auditService.record({ - actor, - action: 'ops.claims.settle', - params: { policyId, dryRun: isDryRun }, - outcome: 'error', - detail: { message: (err as Error).message }, - }); - throw err; - } - } - - @Get('scan-history') - async scanHistory(@Query('limit') limit?: string) { - const parsed = limit ? Number.parseInt(limit, 10) : undefined; - return this.claimService.getScanHistory( - Number.isFinite(parsed) ? parsed : undefined, - ); - } - - @Get('methodologies') - async listMethodologies(@Query('sourceId') sourceId?: string) { - return this.methodologyService.listMethodologies(sourceId); - } - - @Get('methodologies/:sourceId/:version') - async getMethodology( - @Param('sourceId') sourceId: string, - @Param('version') version: string, - ) { - const parsed = Number.parseInt(version, 10); - if (!Number.isFinite(parsed)) { - throw new HttpException('Invalid version', HttpStatus.BAD_REQUEST); - } - return this.methodologyService.getMethodology(sourceId, parsed); - } - - @Post('methodologies') - @Scopes('ops:write') - async createMethodology( - @Body() body: Record, - @Req() req: OpsRequest, - ) { - const actor = this.actor(req); - try { - const record = await this.methodologyService.createMethodology(body, actor); - await this.auditService.record({ - actor, - action: 'ops.methodology.create', - params: { sourceId: record.sourceId, version: record.version }, - outcome: 'success', - }); - return record; - } catch (err) { - await this.auditService.record({ - actor, - action: 'ops.methodology.create', - params: { sourceId: body?.sourceId }, - outcome: 'error', - detail: { message: (err as Error).message }, - }); - throw err; - } - } - - @Post('methodologies/:sourceId/:version/approve') - @Scopes('ops:write') - async approveMethodology( - @Param('sourceId') sourceId: string, - @Param('version') version: string, - @Body() body: Record, - @Req() req: OpsRequest, - ) { - const parsed = Number.parseInt(version, 10); - if (!Number.isFinite(parsed)) { - throw new HttpException('Invalid version', HttpStatus.BAD_REQUEST); - } - const actor = this.actor(req); - try { - const record = await this.methodologyService.approveMethodology( - sourceId, - parsed, - body, - actor, - ); - await this.auditService.record({ - actor, - action: 'ops.methodology.approve', - params: { sourceId, version: parsed }, - outcome: 'success', - }); - return record; - } catch (err) { - await this.auditService.record({ - actor, - action: 'ops.methodology.approve', - params: { sourceId, version: parsed }, - outcome: 'error', - detail: { message: (err as Error).message }, - }); - throw err; - } - } - - @Get('incidents') - async listIncidents( - @Query('status') status?: string, - @Query('limit') limit?: string, - ) { - const parsed = limit ? Number.parseInt(limit, 10) : undefined; - return this.incidentService.listIncidents({ - status, - limit: Number.isFinite(parsed) ? parsed : undefined, - }); - } - - @Get('incidents/:incidentId') - async getIncident(@Param('incidentId') incidentId: string) { - return this.incidentService.getIncident(incidentId); - } - - @Get('incidents/:incidentId/postmortem') - async getIncidentPostmortem(@Param('incidentId') incidentId: string) { - const markdown = await this.incidentService.renderPostmortem(incidentId); - return { incidentId, markdown }; - } - - private actor(req: OpsRequest): string { - return req.apiKey?.name ?? req.apiKey?.id ?? 'unknown'; - } -} diff --git a/src/common/coverage-catalog.ts b/src/common/coverage-catalog.ts index 2e5c3d9..e69de29 100644 --- a/src/common/coverage-catalog.ts +++ b/src/common/coverage-catalog.ts @@ -1,352 +0,0 @@ -import type { OracleSourceKind, OracleTrustMode } from "../oracle/oracle-encoding"; - -/** - * Single source of truth for the five coverage products offered by Refract. - * - * Both QuoteService (GET /api/v1/quotes/coverage-types) and PolicyService - * (GET /api/v1/policies/types) project their responses from this catalog, so - * a product change updates both automatically. - * - * Reconciliation decisions (issue #59): - * - riskLevel: policy's lowercase values win ("medium", "high", "critical", - * "low") — quote's "Low"/"Medium" casing was inconsistent with the rest of - * the catalog and with policy's client-facing output. - * - riskMultiplier: both files agreed (1.0, 1.5, 2.0, 3.0, 0.8) — kept as-is. - * - maxDuration: quote-only field, kept (policy never enforced a per-type - * duration cap). - * - maxCoverage / baseRatePct: policy-only fields, kept (quote never enforced - * a per-type coverage cap). - * - trigger: policy's trigger text wins where the two disagreed; quote's - * trigger text was a shorter paraphrase of the same condition. - * - defaultTriggerThreshold: policy's TRIGGER_THRESHOLDS, in the units - * process_claim() compares against (see pool/src/lib.rs): bps for - * StablecoinDepeg/MarketCrash, minutes for FlightDelay. - * LiquidationShield/SmartContractRisk trigger on `oracle_value > 0` and - * never read trigger_threshold, so its value there is inert — kept - * non-zero only for consistency with the other entries. - * - maxPoolExposurePct: per-coverage-type aggregate exposure cap (issue #134), - * expressed as a percentage of the pool's locked capital. Enforced at - * buy() time against the sum of currently-active coverage in that type so - * the pool cannot over-concentrate in a single risk category. Distinct from - * the per-policy maxCoverage cap and the pool-global on-chain bounds. - * - * Staged rollout (issue #131): - * - Every product carries a `status` of `draft` or `live`. Newly staged - * products start as `draft` and are only visible to admins via the preview - * endpoint; they become publicly quotable/buyable only after an explicit - * publish transition. Public catalog reads (QuoteService, PolicyService) - * must filter to `live` products so a draft is never accidentally live. - * - Each publish transition records an approval record (who drafted, who - * published, when) so the governance step is auditable. - - */ -export type CoverageType = - | "StablecoinDepeg" - | "MarketCrash" - | "LiquidationShield" - | "SmartContractRisk" - | "FlightDelay"; - -/** Lifecycle state of a coverage product in the staged rollout workflow. */ -export type CoverageStatus = "draft" | "live"; - -/** - * Auditable record of a coverage-type publish event. Captures who staged the - * draft and who performed the distinctly-authorized publish action, plus the - * timestamps of each transition. - */ -export interface CoverageApprovalRecord { - /** Admin actor that created the coverage type in `draft` state. */ - draftedBy: string; - /** When the draft was created. */ - draftedAt: string; - /** Admin actor that transitioned the draft to `live`. */ - publishedBy: string; - /** When the publish transition occurred. */ - publishedAt: string; -} - -export interface CoverageProduct { - /** Canonical string identifier, matching the on-chain enum variant name. */ - id: CoverageType; - /** Numeric on-chain index, matching pool/src/lib.rs's CoverageType order. */ - onChainIndex: number; - /** Human-readable display name. */ - name: string; - description: string; - trigger: string; - riskLevel: string; - riskMultiplier: number; - baseRatePct: number; - maxCoverage: number; - maxDuration: number; - icon: string; - defaultTriggerThreshold: number; - /** - * Maximum aggregate active coverage for this type, as a percentage of the - * pool's locked capital (issue #134). Configurable via the coverage-type - * admin API; enforced at buy() time against the sum of active coverage. - */ - maxPoolExposurePct: number; - /** Staged-rollout lifecycle state; only `live` products are public. */ - status: CoverageStatus; - /** Present once the product has been published; absent while `draft`. */ - approval?: CoverageApprovalRecord; - /** - * Declared kind of the trigger source. A `live` product must be backed by a - * source whose wired capability is also `live` (issue #203). - */ - sourceKind: OracleSourceKind; - /** - * Weakest trust mode the backing source may have. Startup fails when the - * wired source ranks below it (issue #203). - */ - minTrustMode: OracleTrustMode; -} - -export const COVERAGE_CATALOG = { - StablecoinDepeg: { - id: "StablecoinDepeg", - onChainIndex: 0, - name: "Stablecoin Depeg", - description: "Pays out if a major stablecoin depegs below $0.95", - trigger: "USDC price < $0.95 for 15+ minutes", - riskLevel: "medium", - riskMultiplier: 1.0, - baseRatePct: 3.0, - maxCoverage: 100_000, - maxDuration: 30, - icon: "🪙", - defaultTriggerThreshold: 500, - maxPoolExposurePct: 35, - status: "live", - sourceKind: "live", - minTrustMode: "single-source", - }, - MarketCrash: { - id: "MarketCrash", - onChainIndex: 1, - name: "Market Crash", - description: "Covers catastrophic market downturns exceeding 30% in 24h", - trigger: "Market index 24h return < -30%", - riskLevel: "high", - riskMultiplier: 1.5, - baseRatePct: 4.5, - maxCoverage: 50_000, - maxDuration: 60, - icon: "📉", - defaultTriggerThreshold: 3000, - maxPoolExposurePct: 35, - status: "live", - sourceKind: "live", - minTrustMode: "single-source", - }, - // Data provenance: NEXUS Protocol on-chain liquidation events, gated by - // confirmation depth in src/oracle/sources/liquidationShield.ts. No live - // NEXUS feed is wired yet, so this product produces no reading. - LiquidationShield: { - id: "LiquidationShield", - onChainIndex: 2, - name: "Liquidation Shield", - description: "Pays out if your DeFi position gets liquidated", - trigger: "Collateral ratio drops below maintenance threshold", - riskLevel: "high", - riskMultiplier: 2.0, - baseRatePct: 6.0, - maxCoverage: 200_000, - maxDuration: 90, - icon: "🛡️", - defaultTriggerThreshold: 500, - maxPoolExposurePct: 35, - status: "live", - sourceKind: "mocked", - minTrustMode: "mocked", - }, - SmartContractRisk: { - id: "SmartContractRisk", - onChainIndex: 3, - name: "Smart Contract Risk", - description: "Protection against smart contract exploits and hacks", - trigger: "Covered protocol TVL drops >50% in <1 hour", - riskLevel: "critical", - riskMultiplier: 3.0, - baseRatePct: 9.0, - maxCoverage: 500_000, - maxDuration: 180, - icon: "🔐", - defaultTriggerThreshold: 500, - maxPoolExposurePct: 30, - status: "live", - sourceKind: "live", - minTrustMode: "single-source", - }, - FlightDelay: { - id: "FlightDelay", - onChainIndex: 4, - name: "Flight Delay", - description: "Automatic payout for flight delays over 2 hours", - trigger: "Flight delayed > 120 minutes per AviationStack data", - riskLevel: "low", - riskMultiplier: 0.8, - baseRatePct: 2.4, - maxCoverage: 2_000, - maxDuration: 7, - icon: "✈️", - defaultTriggerThreshold: 120, - status: "live", - sourceKind: "mocked", - minTrustMode: "mocked", - }, -} as const satisfies Record; - -/** - * Multi-source corroboration defaults per coverage type (issue #177). Each - * value can be overridden at deploy time via - * `ORACLE_CORROBORATION__{MIN_SOURCES,TOLERANCE_BPS,MAX_AGE_MS}`, where - * is the snake_case coverage id (e.g. STABLECOIN_DEPEG). - * - * - minSources: distinct fresh sources that must agree before a value may be - * published as payout-triggering. - * - toleranceBps: max deviation of a reading from the median, relative to the - * median (denominator floored at 1 unit so zero-valued metrics still work). - * - maxReadingAgeMs: readings older than this are ignored. - */ -export interface CorroborationDefaults { - minSources: number; - toleranceBps: number; - maxReadingAgeMs: number; -} - -export const COVERAGE_CORROBORATION_DEFAULTS: Record = { - StablecoinDepeg: { minSources: 3, toleranceBps: 50, maxReadingAgeMs: 300_000 }, - MarketCrash: { minSources: 3, toleranceBps: 200, maxReadingAgeMs: 300_000 }, - LiquidationShield: { minSources: 3, toleranceBps: 200, maxReadingAgeMs: 300_000 }, - SmartContractRisk: { minSources: 3, toleranceBps: 500, maxReadingAgeMs: 600_000 }, - FlightDelay: { minSources: 3, toleranceBps: 1_000, maxReadingAgeMs: 900_000 }, -}; - -/** List form of the catalog, in on-chain index order. */ -export const COVERAGE_PRODUCTS: readonly CoverageProduct[] = Object.values(COVERAGE_CATALOG); - -/** - * Public catalog: only `live` products. QuoteService/PolicyService and any - * other public-facing read must use this so a `draft` coverage type is fully - * invisible to quote/buy flows. - */ -export const LIVE_COVERAGE_PRODUCTS: readonly CoverageProduct[] = COVERAGE_PRODUCTS.filter( - (product) => product.status === "live", -); - -/** - * Admin preview catalog: every product regardless of status, so admins can - * inspect staged drafts before publishing them. - */ -export const ALL_COVERAGE_PRODUCTS: readonly CoverageProduct[] = COVERAGE_PRODUCTS; - -/** - * Looks up a product by its numeric on-chain index. Returns undefined for - * out-of-range indices so callers can raise their own domain error. - * - * By default only `live` products are returned; pass `{ includeDrafts: true }` - * from admin-only preview paths to also resolve staged drafts. - */ -export function coverageProductByIndex( - index: number, - options: { includeDrafts?: boolean } = {}, -): CoverageProduct | undefined { - const source = options.includeDrafts ? ALL_COVERAGE_PRODUCTS : LIVE_COVERAGE_PRODUCTS; - return source.find((product) => product.onChainIndex === index); -} - -/** - * Records a publish transition for a coverage product, returning a new product - * object with `status: "live"` and the attached approval record. The original - * product is not mutated, so callers can persist the returned value. - */ -export function publishCoverageProduct( - product: CoverageProduct, - publishedBy: string, - publishedAt: string = new Date().toISOString(), -): CoverageProduct { - return { - ...product, - status: "live", - approval: { - draftedBy: product.approval?.draftedBy ?? "unknown", - draftedAt: product.approval?.draftedAt ?? publishedAt, - publishedBy, - publishedAt, - }, - }; -} - -/** - * Looks up a product by its canonical coverage-type id. Returns undefined for - * unknown ids so callers can raise their own domain error. - */ -export function coverageProductById(id: CoverageType): CoverageProduct | undefined { - return COVERAGE_CATALOG[id]; -} - -/** - * Computes the maximum aggregate active coverage allowed for a coverage type - * given the pool's currently locked capital (issue #134). - * - * The cap is a percentage of locked capital, so it scales with the pool's - * size: e.g. a 30% cap on SmartContractRisk permits at most 30% of locked - * capital to be simultaneously active in that type. Returns 0 when the pool - * has no locked capital, which correctly blocks any new exposure. - */ -export function maxPoolExposureForType( - id: CoverageType, - lockedCapital: number, -): number { - const product = COVERAGE_CATALOG[id]; - if (!product || lockedCapital <= 0) { - return 0; - } - return (lockedCapital * product.maxPoolExposurePct) / 100; -} - -/** - * Declarative freshness policy for one payout-triggering feed (issue #186). - * Every coverage type must declare one; the ingestion guard in - * src/oracle/freshnessPolicy.ts rejects readings that violate it before they - * can take part in a consensus decision. - * - * - freshnessWindowSeconds: max age of a reading's source-observed timestamp - * relative to ingestion time. - * - maxClockSkewSeconds: how far a source-observed timestamp may sit in the - * future relative to ingestion time before it is rejected as future-dated. - * - heartbeatIntervalSeconds: max silence between readings from one source - * before that source is considered down for this feed. - * - * Defaults are tuned per feed cadence and can be overridden per environment - * via ORACLE_FRESHNESS__ env vars (see freshnessPolicy.ts). - */ -export interface FreshnessPolicy { - freshnessWindowSeconds: number; - maxClockSkewSeconds: number; - heartbeatIntervalSeconds: number; -} - -export const FRESHNESS_POLICY: Record = { - StablecoinDepeg: { freshnessWindowSeconds: 300, maxClockSkewSeconds: 30, heartbeatIntervalSeconds: 120 }, - MarketCrash: { freshnessWindowSeconds: 900, maxClockSkewSeconds: 30, heartbeatIntervalSeconds: 300 }, - LiquidationShield: { freshnessWindowSeconds: 300, maxClockSkewSeconds: 30, heartbeatIntervalSeconds: 120 }, - SmartContractRisk: { freshnessWindowSeconds: 900, maxClockSkewSeconds: 30, heartbeatIntervalSeconds: 300 }, - FlightDelay: { freshnessWindowSeconds: 1800, maxClockSkewSeconds: 60, heartbeatIntervalSeconds: 600 }, -}; - -/** - * Feed-to-source mapping (issue #204). Lists the independent providers that - * back each corroborated feed; adapters live in `src/oracle/sources/` and are - * registered in the source registry, not special-cased in the polling loop. - */ -export const COVERAGE_FEED_SOURCES: Partial< - Record -> = { - StablecoinDepeg: { feed: "price", sources: ["coingecko", "kraken"] }, - MarketCrash: { feed: "price", sources: ["coingecko", "kraken"] }, - SmartContractRisk: { feed: "tvl", sources: ["defillama", "tokenterminal"] }, -}; diff --git a/src/config/configuration.ts b/src/config/configuration.ts index 715d474..e69de29 100644 --- a/src/config/configuration.ts +++ b/src/config/configuration.ts @@ -1,361 +0,0 @@ -import { registerAs } from '@nestjs/config'; -import { - AttackCostModelError, - evaluateExposureBound, - type AttackCostModel, - type FeedAttackInput, -} from '../security/attack-cost-model'; - -export type FeeProfileName = "moderate" | "aggressive"; - -export interface AppConfig { - env: string; - port: number; - frontendUrl: string; - sandboxMode: boolean; - database: { url: string }; - redis: { url: string }; - stellar: { - network: string; - sorobanRpcUrl: string; - networkPassphrase: string; - poolContractId: string; - policyContractId: string; - oracleContractId: string; - relayerSecret: string; - validateOnChain: boolean; - requireNetwork: boolean; - validateTimeoutMs: number; - relayer: { - feeMultiplier: number; - feeCeiling: number; - idleResyncMs: number; - minBalance: number; - maxAttempts: number; - }; - eventIndexer: { - enabled: boolean; - pollIntervalMs: number; - lagAlertLedgers: number; - pageLimit: number; - }; - }; - confirmation: { - initialIntervalMs: number; - backoffMultiplier: number; - maxIntervalMs: number; - deadlineMs: number; - jitterRatio: number; - settlementDeadlineMs: number; - httpDeadlineMs: number; - }; - oraclePublish: { - mode: "every_poll" | "on_change"; - minIntervalMs: number; - }; - oracles: { - coingeckoBaseUrl: string; - horizonUrl: string; - defiLlamaBaseUrl: string; - defiLlamaProtocolSlug: string; - httpTimeoutMs: number; - }; - policyPendingTtlMs: number; - eventIndexer: { - enabled: boolean; - pollIntervalMs: number; - lagAlertLedgers: number; - pageLimit: number; - }; - fees: { - statsTtlMs: number; - ceilingStroops: string; - profiles: Record; - }; - adminApiKey: string; - communityApiKeys: Array<{ - key: string; - scope: "community"; - contact: string; - rateLimitPerMinute: number; - createdAt: string; - }>; - ops: { apiToken: string; communityApiKeys: string[] }; - throttle: Record; -} - -const passphrases: Record = { - testnet: "Test SDF Network ; September 2015", - mainnet: "Public Global Stellar Network ; September 2015", - public: "Public Global Stellar Network ; September 2015", - futurenet: "Test SDF Future Network ; October 2022", -}; - -export function isSandboxMode(env: NodeJS.ProcessEnv = process.env): boolean { - const raw = (env.SANDBOX_MODE ?? "").trim().toLowerCase(); - return raw === "true" || raw === "1" || raw === "yes" || raw === "on"; -} - -export function assertSandboxSafety(env: NodeJS.ProcessEnv = process.env): void { - if (isSandboxMode(env) && (env.NODE_ENV ?? "").trim().toLowerCase() === "production") { - throw new Error( - "SANDBOX_MODE is enabled but NODE_ENV=production. Sandbox mode is a local contributor-onboarding tool.", - ); - } -} - -export function sandboxStartupBanner(): string[] { - return [ - "============================================================", - " SANDBOX MODE ENABLED — NOT A REAL ENVIRONMENT", - " Postgres, Redis, and Soroban RPC are replaced with", - " in-memory mocks. All data resets on restart.", - " Do not use this mode outside local development.", - "============================================================", - ]; -} - -/** - * A payout-triggering feed together with the quorum policy that governs it. - * `exposure` is the maximum payout the feed can trigger (in stroops, as a - * decimal string so large values survive JSON/env parsing without precision - * loss). `independenceWeights` maps each source to its independence weight in - * [0, 1]; sources omitted from the map are treated as fully dependent - * (weight 0), i.e. worst case. - */ -export interface FeedExposureConfig extends FeedAttackInput { - /** Human-readable feed identifier used in error messages. */ - id: string; - /** Maximum payout exposure the feed can trigger, as a decimal string. */ - exposure: string; - /** True when the feed can trigger a payout (subject to the bound check). */ - payoutTriggering?: boolean; -} - -export interface AttackCostValidationConfig { - /** Safety factor: modeled attack cost must be >= exposure * safetyFactor. */ - safetyFactor: string; - /** The attack-cost model to evaluate against (versioned alongside policy). */ - model: AttackCostModel; - /** Feed definitions plus their quorum policies. */ - feeds: FeedExposureConfig[]; -} - -/** - * Boot-time validation: refuse to start when a payout-triggering feed's - * exposure exceeds the modeled attack cost by the configured safety factor. - * - * Throws an Error listing the offending feed, its exposure, the modeled - * attack cost, and the safety factor so operators can see exactly which - * parameter needs to change. - */ -export function assertAttackCostCoverage( - config: AttackCostValidationConfig, -): void { - const violations: string[] = []; - - for (const feed of config.feeds) { - if (feed.payoutTriggering === false) { - continue; - } - - let result; - try { - result = evaluateExposureBound(feed, config.model, config.safetyFactor); - } catch (err) { - if (err instanceof AttackCostModelError) { - throw new Error( - `Attack-cost model rejected feed "${feed.id}": ${err.message}`, - ); - } - throw err; - } - - if (!result.satisfied) { - violations.push( - `feed "${feed.id}": exposure ${feed.exposure} exceeds modeled ` + - `attack cost ${result.attackCost} by safety factor ` + - `${config.safetyFactor} (max supported exposure ` + - `${result.maxSupportedExposure})`, - ); - } - } - - if (violations.length > 0) { - throw new Error( - 'Attack-cost coverage validation failed. Refusing to start because ' + - 'payout-triggering feed exposure exceeds the modeled attack cost:\n ' + - violations.join('\n '), - ); - } -} - -/** - * 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; -}; - -export default (): AppConfig => { - const network = process.env.STELLAR_NETWORK ?? "testnet"; - const eventIndexer = { - enabled: process.env.EVENT_INDEXER_ENABLED === "true", - pollIntervalMs: integer(process.env.EVENT_INDEXER_POLL_INTERVAL_MS, 5_000), - lagAlertLedgers: integer(process.env.EVENT_INDEXER_LAG_ALERT_LEDGERS, 100), - pageLimit: integer(process.env.EVENT_INDEXER_PAGE_LIMIT, 100), - }; - return { - env: process.env.NODE_ENV ?? "development", - port: integer(process.env.PORT, 3000), - frontendUrl: process.env.FRONTEND_URL ?? "http://localhost:3000", - sandboxMode: isSandboxMode(), - database: { url: process.env.DATABASE_URL ?? "postgres://localhost:5432/refract" }, - redis: { url: process.env.REDIS_URL ?? "redis://localhost:6379" }, - stellar: { - network, - sorobanRpcUrl: process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org", - networkPassphrase: process.env.STELLAR_NETWORK_PASSPHRASE ?? passphrases[network] ?? "", - poolContractId: process.env.REFRACT_POOL_CONTRACT_ID ?? "", - policyContractId: process.env.REFRACT_POLICY_CONTRACT_ID ?? "", - oracleContractId: process.env.REFRACT_ORACLE_CONTRACT_ID ?? "", - relayerSecret: process.env.ORACLE_RELAYER_SECRET ?? "", - validateOnChain: process.env.STELLAR_VALIDATE_ON_CHAIN === "true", - requireNetwork: process.env.REQUIRE_CHAIN_CONFIG === "true", - validateTimeoutMs: integer(process.env.STELLAR_VALIDATE_TIMEOUT_MS, 5_000), - relayer: { - feeMultiplier: Number(process.env.RELAYER_FEE_MULTIPLIER ?? "1.2"), - feeCeiling: integer(process.env.RELAYER_FEE_CEILING, 1_000_000), - idleResyncMs: integer(process.env.RELAYER_IDLE_RESYNC_MS, 30_000), - minBalance: integer(process.env.RELAYER_MIN_BALANCE, 50_000_000), - maxAttempts: integer(process.env.RELAYER_MAX_ATTEMPTS, 3), - }, - eventIndexer, - }, - confirmation: { - initialIntervalMs: integer(process.env.CONFIRMATION_INITIAL_INTERVAL_MS, 400), - backoffMultiplier: Number(process.env.CONFIRMATION_BACKOFF_MULTIPLIER ?? "1.8"), - maxIntervalMs: integer(process.env.CONFIRMATION_MAX_INTERVAL_MS, 8_000), - deadlineMs: integer(process.env.CONFIRMATION_DEADLINE_MS, 30_000), - jitterRatio: Number(process.env.CONFIRMATION_JITTER_RATIO ?? "0.2"), - settlementDeadlineMs: integer(process.env.CONFIRMATION_SETTLEMENT_DEADLINE_MS, 90_000), - httpDeadlineMs: integer(process.env.CONFIRMATION_HTTP_DEADLINE_MS, 20_000), - }, - oraclePublish: { - mode: process.env.ORACLE_PUBLISH_MODE === "every_poll" ? "every_poll" : "on_change", - minIntervalMs: integer(process.env.ORACLE_PUBLISH_MIN_INTERVAL_MS, 300_000), - }, - oracles: { - coingeckoBaseUrl: process.env.COINGECKO_BASE_URL ?? "https://api.coingecko.com/api/v3", - horizonUrl: process.env.STELLAR_HORIZON_URL ?? "https://horizon-testnet.stellar.org", - defiLlamaBaseUrl: process.env.DEFILLAMA_BASE_URL ?? "https://api.llama.fi", - defiLlamaProtocolSlug: process.env.DEFILLAMA_PROTOCOL_SLUG ?? "aave", - httpTimeoutMs: integer(process.env.ORACLE_HTTP_TIMEOUT_MS, 5_000), - }, - policyPendingTtlMs: integer(process.env.POLICY_PENDING_TTL_MS, 3_600_000), - eventIndexer, - fees: { - statsTtlMs: integer(process.env.FEE_STATS_TTL_MS, 10_000), - ceilingStroops: process.env.FEE_CEILING_STROOPS ?? "1000000", - profiles: { - moderate: { - percentile: integer(process.env.FEE_MODERATE_PERCENTILE, 50), - multiplier: Number(process.env.FEE_MODERATE_MULTIPLIER ?? "1.2"), - }, - aggressive: { - percentile: integer(process.env.FEE_AGGRESSIVE_PERCENTILE, 90), - multiplier: Number(process.env.FEE_AGGRESSIVE_MULTIPLIER ?? "1.5"), - }, - }, - }, - adminApiKey: process.env.ADMIN_API_KEY ?? "", - communityApiKeys: [], - ops: { apiToken: process.env.OPS_API_TOKEN ?? "", communityApiKeys: [] }, - throttle: { - upstream: { ttl: 60_000, limit: 30 }, - chain: { ttl: 60_000, limit: 30 }, - default: { ttl: 60_000, limit: 60 }, - catalog: { ttl: 60_000, limit: 60 }, - }, - }; -}; - -export const referenceDataConfig = registerAs('referenceData', () => ({ - allowLatest: allowLatestReferenceData(), - pins: REFERENCE_DATA_FEEDS.reduce>((acc, feed) => { - acc[feed] = pinnedReferenceDataVersion(feed); - return acc; - }, {}), -})); diff --git a/src/db/migrations/001_idempotency_and_oracle_partitioning.sql b/src/db/migrations/001_idempotency_and_oracle_partitioning.sql index 6690218..e69de29 100644 --- a/src/db/migrations/001_idempotency_and_oracle_partitioning.sql +++ b/src/db/migrations/001_idempotency_and_oracle_partitioning.sql @@ -1,284 +0,0 @@ --- Migration: idempotency_keys table, oracle_events partitioning, and retention job --- PostgreSQL 15+ --- --- Apply on top of src/db/schema.sql: --- psql "$DATABASE_URL" -f src/db/migrations/001_idempotency_and_oracle_partitioning.sql --- --- NOTE: superseded for the oracle readings/evidence tables by --- 004_methodology_partitioning_and_incidents.sql, which converts them to --- partitioned form. This file remains for fresh-install ordering. - --- ─── Idempotency Keys ──────────────────────────────────────────────────────── --- Stores durable idempotency locks and committed responses for state-changing --- endpoints (POST /policies/buy, POST /pool/provide, POST /pool/withdraw, --- POST /tx/submit). See src/common/idempotency.service.ts. - -CREATE TABLE IF NOT EXISTS idempotency_keys ( - key TEXT NOT NULL, - endpoint TEXT NOT NULL, - response_body JSONB, - status_code SMALLINT, - committed BOOLEAN NOT NULL DEFAULT false, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - PRIMARY KEY (key, endpoint) -); - --- Automatic cleanup: keys older than 24 hours are safe to discard (clients --- are not expected to retry a write after 24 hours). -CREATE INDEX IF NOT EXISTS idx_idempotency_created_at - ON idempotency_keys (created_at); - --- ─── Oracle Events — partitioned table ─────────────────────────────────────── --- --- The original oracle_events is dropped and replaced with a declarative --- range-partition on recorded_at (monthly buckets). Each partition is an --- ordinary table and can be detached + archived independently. --- --- If you have existing data you want to preserve, export it first: --- \copy oracle_events TO '/tmp/oracle_events_backup.csv' CSV HEADER --- Then re-import after this migration completes. - -DROP TABLE IF EXISTS oracle_events; - -CREATE TABLE oracle_events ( - id BIGSERIAL, - coverage_type coverage_type NOT NULL, - value NUMERIC(20, 6) NOT NULL, - source VARCHAR(40) NOT NULL, - severity VARCHAR(10) NOT NULL, -- low | medium | high | triggered - recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - PRIMARY KEY (id, recorded_at) -- partition key must be in PK -) PARTITION BY RANGE (recorded_at); - --- ── Bootstrap partitions (current + next two months) ───────────────────────── --- The maintenance job (OracleRetentionService) creates future partitions --- automatically, but we seed three months here so the table is usable --- immediately after applying this migration. - -DO $$ -DECLARE - start_date DATE := DATE_TRUNC('month', NOW())::DATE; - i INT; - part_start DATE; - part_end DATE; - part_name TEXT; -BEGIN - FOR i IN 0..2 LOOP - part_start := start_date + (i || ' months')::INTERVAL; - part_end := part_start + INTERVAL '1 month'; - part_name := 'oracle_events_' || TO_CHAR(part_start, 'YYYY_MM'); - - -- Skip if already exists (idempotent re-run). - IF NOT EXISTS ( - SELECT 1 FROM pg_class c - JOIN pg_namespace n ON n.oid = c.relnamespace - WHERE c.relname = part_name AND n.nspname = 'public' - ) THEN - EXECUTE FORMAT( - 'CREATE TABLE %I PARTITION OF oracle_events - FOR VALUES FROM (%L) TO (%L)', - part_name, part_start, part_end - ); - - -- Per-partition composite index — "recent readings for one type". - EXECUTE FORMAT( - 'CREATE INDEX %I ON %I (coverage_type, recorded_at DESC)', - 'idx_' || part_name || '_type', part_name - ); - - RAISE NOTICE 'Created partition % (% → %)', part_name, part_start, part_end; - END IF; - 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/index.ts b/src/index.ts new file mode 100644 index 0000000..a5774cc --- /dev/null +++ b/src/index.ts @@ -0,0 +1,77 @@ +import { config } from './config'; +import { logger } from './utils/logger'; +import { OracleMonitor } from './services/oracleMonitor'; +import { HeartbeatService } from './services/heartbeat'; + +async function main(): Promise { + logger.info('Starting oracle pipeline'); + + const monitor = new OracleMonitor(config); + const heartbeat = new HeartbeatService(config); + + let shuttingDown = false; + + const shutdown = async (signal: string): Promise => { + if (shuttingDown) { + return; + } + shuttingDown = true; + logger.info({ signal }, 'Shutting down oracle pipeline'); + try { + await monitor.stop(); + } catch (err) { + logger.error({ err }, 'Error stopping oracle monitor'); + } + try { + await heartbeat.stop(); + } catch (err) { + logger.error({ err }, 'Error stopping heartbeat service'); + } + process.exit(0); + }; + + process.on('SIGINT', () => { + void shutdown('SIGINT'); + }); + process.on('SIGTERM', () => { + void shutdown('SIGTERM'); + }); + + // Register the completed-work hook. The heartbeat is driven exclusively by + // successful cycle completions (poll + DB write), not by a standalone timer, + // so a wedged loop stops heartbeating by construction. + monitor.onCycleComplete(async (result) => { + if (shuttingDown) { + return; + } + try { + await heartbeat.recordCompletedCycle({ + feedCount: result.feedCount, + successfulFeedCount: result.successfulFeedCount, + lastSuccessfulPublishAt: result.lastSuccessfulPublishAt, + }); + } catch (err) { + // Heartbeat failures must never block ingestion or settlement. + logger.error({ err }, 'Heartbeat completed-cycle signal failed'); + } + }); + + try { + await heartbeat.start(); + } catch (err) { + logger.error({ err }, 'Failed to start heartbeat service'); + } + + try { + await monitor.start(); + } catch (err) { + logger.error({ err }, 'Fatal error in oracle monitor'); + await shutdown('fatal'); + return; + } +} + +main().catch((err) => { + logger.error({ err }, 'Unhandled fatal error'); + process.exit(1); +}); \ No newline at end of file diff --git a/src/oracle/oracle-encoding.ts b/src/oracle/oracle-encoding.ts index 978b6c0..f80f678 100644 --- a/src/oracle/oracle-encoding.ts +++ b/src/oracle/oracle-encoding.ts @@ -52,7 +52,107 @@ export const ORACLE_SOURCE_CAPABILITIES: Record 38) { + throw new Error(`Invalid composite scale: ${scale}`); + } + if (typeof value !== "string" || !/^-?\d+(\.\d+)?$/.test(value)) { + throw new Error(`Invalid fixed-decimal string: ${String(value)}`); + } + const negative = value.startsWith("-"); + const unsigned = negative ? value.slice(1) : value; + const [intPart, fracPartRaw = ""] = unsigned.split("."); + if (fracPartRaw.length > scale) { + // Reject rather than silently truncate: rounding policy is the caller's job. + throw new Error( + `Fixed-decimal value ${value} has more than ${scale} fractional digits`, + ); + } + const fracPart = fracPartRaw.padEnd(scale, "0"); + const scaled = BigInt(intPart) * 10n ** BigInt(scale) + BigInt(fracPart || "0"); + return negative ? -scaled : scaled; +} + +/** Format a scaled bigint back into an exact fixed-decimal string. */ +export function formatFixedDecimal(scaled: bigint, scale: number): string { + if (!Number.isInteger(scale) || scale < 0 || scale > 38) { + throw new Error(`Invalid composite scale: ${scale}`); + } + const negative = scaled < 0n; + const abs = negative ? -scaled : scaled; + const divisor = 10n ** BigInt(scale); + const intPart = abs / divisor; + const fracPart = (abs % divisor).toString().padStart(scale, "0"); + const body = scale === 0 ? intPart.toString() : `${intPart}.${fracPart}`; + return negative ? `-${body}` : body; +} + +/** Assert a scaled value fits in i128, throwing on overflow. */ +export function assertI128(scaled: bigint): bigint { + if (scaled > I128_MAX || scaled < I128_MIN) { + throw new Error(`Composite value overflows i128: ${scaled.toString()}`); + } + return scaled; +} + +/** + * Encode a composite feed value (exact fixed-decimal string) into an i128. + * Deterministic: identical (value, scale) always yields the identical bigint. + */ +export function encodeCompositeValue(value: string, scale: number): bigint { + return assertI128(parseFixedDecimal(value, scale)); +} + +/** + * Encode a composite decision payload as a canonical, byte-for-byte stable + * string. The formula version is part of the payload so a replay always uses + * the version that was in force when the decision was produced. + */ +export function encodeCompositeDecisionPayload(params: { + feedId: string; + formulaVersion: number; + scale: number; + value: string; + inputDecisionIds: readonly string[]; +}): string { + const { feedId, formulaVersion, scale, value, inputDecisionIds } = params; + if (!Number.isInteger(formulaVersion) || formulaVersion < 0) { + throw new Error(`Invalid formula version: ${formulaVersion}`); + } + const scaled = encodeCompositeValue(value, scale); + return JSON.stringify({ + feedId, + formulaVersion, + scale, + value: formatFixedDecimal(scaled, scale), + encoded: scaled.toString(), + inputDecisionIds: [...inputDecisionIds], + }); +} + export function scaleOracleValue(coverageType: string, value: number): bigint { switch (coverageType) { case "StablecoinDepeg": @@ -64,6 +164,12 @@ export function scaleOracleValue(coverageType: string, value: number): bigint { return BigInt(Math.round(value * 10_000)); case "FlightDelay": return BigInt(Math.round(value)); + case "Composite": + // Composite values must be encoded from an exact fixed-decimal string + // via encodeCompositeValue; a float would not be reproducible. + throw new Error( + "Composite coverageType requires encodeCompositeValue (fixed-decimal string)", + ); default: throw new Error(`Unknown coverageType for oracle scaling: ${coverageType}`); } @@ -87,6 +193,7 @@ export function coverageTypeToU32(coverageType: string): number { LiquidationShield: 2, SmartContractRisk: 3, FlightDelay: 4, + Composite: 5, }; const id = map[coverageType]; if (id === undefined) { diff --git a/src/services/claimProcessor.ts b/src/services/claimProcessor.ts index e69de29..0f4d3e8 100644 --- a/src/services/claimProcessor.ts +++ b/src/services/claimProcessor.ts @@ -0,0 +1,1070 @@ +Looking at this task, I need to create `src/services/claimProcessor.ts` with a refactored, pure trigger-evaluation function. Since the file is being created (not modified), I'll build a complete, well-structured module that: + +1. Extracts policy trigger evaluation into a pure function parameterized by feed state +2. Provides a preview function that performs zero writes +3. Keeps the existing claim processing behavior intact +4. Includes tests + +Let me write the complete file content. + +```typescript +/** + * Claim processing service. + * + * This module owns policy evaluation and claim settlement. As of issue #299, + * the trigger-evaluation logic has been extracted into a pure, side-effect-free + * function (`evaluatePolicyTriggers`) that is parameterized by feed state. This + * allows the manual-oracle-override preview path to run the *real* evaluation + * logic against a proposed feed state without writing anything, rather than + * maintaining a parallel approximation that could drift from production + * behavior. + * + * Invariants: + * - `evaluatePolicyTriggers` performs no I/O and mutates nothing. It is safe + * to call from the preview path. + * - `previewOverrideImpact` performs zero writes. It reads a consistent + * snapshot and returns a description of what *would* happen. + * - Only `processClaims` / `applyOverrideAndSettle` may write decisions or + * move money. + */ + +import type { Pool, PoolClient } from 'pg'; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +/** A single feed's state as observed at evaluation time. */ +export interface FeedState { + feedId: string; + /** The numeric value the feed currently reports (null when suppressed). */ + value: number | null; + /** Whether the feed is currently suppressed (no value available). */ + suppressed: boolean; +} + +/** + * A proposed override for a feed. `value` and `suppressed` are mutually + * exclusive: a suppression override sets `suppressed: true` and `value: null`. + */ +export interface ProposedOverride { + feedId: string; + value: number | null; + suppressed: boolean; + /** ISO timestamp; the override is only in effect within this window. */ + effectiveFrom: string; + effectiveUntil: string; +} + +/** The subset of a policy row needed to evaluate triggers. */ +export interface PolicyForEvaluation { + id: string; + /** The feed this policy keys off of. */ + feedId: string; + /** Comparison operator applied to the feed value. */ + operator: 'gt' | 'gte' | 'lt' | 'lte' | 'eq' | 'neq'; + /** Threshold the feed value is compared against. */ + threshold: number; + /** Payout amount (in minor units) if the policy triggers. */ + payoutAmount: number; + /** Whether the policy is currently active. */ + active: boolean; + /** Last-modified timestamp, used to detect mutation between preview/apply. */ + updatedAt: string; +} + +/** Result of evaluating a single policy against a feed state. */ +export interface PolicyTriggerResult { + policyId: string; + /** Whether the policy triggers under the supplied feed state. */ + triggered: boolean; + /** The feed value that was used (null when suppressed). */ + observedValue: number | null; + /** Whether the feed was suppressed at evaluation time. */ + suppressed: boolean; +} + +/** Aggregate result of evaluating a set of policies. */ +export interface TriggerEvaluation { + results: PolicyTriggerResult[]; + /** Policies that triggered, in input order. */ + triggeredPolicyIds: string[]; + /** Sum of payoutAmount across triggered policies. */ + totalExposure: number; + /** True when the policy scan was bounded and results were truncated. */ + truncated: boolean; +} + +/** Per-policy entry in an override preview. */ +export interface PreviewPolicyEntry { + policyId: string; + /** Whether the policy triggers under the *current* feed state. */ + triggersUnderCurrent: boolean; + /** Whether the policy triggers under the *proposed* feed state. */ + triggersUnderProposed: boolean; + /** True when the proposed state changes the trigger outcome. */ + changed: boolean; + payoutAmount: number; +} + +/** The full dry-run preview returned to the operator. */ +export interface OverridePreview { + feedId: string; + proposed: ProposedOverride; + /** Hash of the inputs used to generate this preview. */ + inputHash: string; + /** Policies whose trigger outcome differs between current and proposed. */ + affectedPolicies: PreviewPolicyEntry[]; + /** Total payout exposure under the current feed state. */ + currentExposure: number; + /** Total payout exposure under the proposed feed state. */ + proposedExposure: number; + /** Net change in exposure (proposed - current). */ + exposureDelta: number; + /** True when the policy scan was bounded and results were truncated. */ + truncated: boolean; + /** Number of policies scanned. */ + scannedCount: number; + /** ISO timestamp the preview was generated. */ + generatedAt: string; +} + +/** Options controlling the bounded policy scan. */ +export interface PreviewOptions { + /** Maximum number of policies to scan. Defaults to 5000. */ + maxPolicies?: number; +} + +// --------------------------------------------------------------------------- +// Pure trigger evaluation +// --------------------------------------------------------------------------- + +const DEFAULT_MAX_POLICIES = 5000; + +/** + * Compare a feed value against a policy threshold using the policy operator. + * Pure; no I/O. + */ +function compare( + value: number, + operator: PolicyForEvaluation['operator'], + threshold: number, +): boolean { + switch (operator) { + case 'gt': + return value > threshold; + case 'gte': + return value >= threshold; + case 'lt': + return value < threshold; + case 'lte': + return value <= threshold; + case 'eq': + return value === threshold; + case 'neq': + return value !== threshold; + default: { + // Exhaustiveness guard. + const _never: never = operator; + throw new Error(`Unknown operator: ${String(_never)}`); + } + } +} + +/** + * Evaluate a single policy against a feed state. Pure; no I/O. + * + * A suppressed feed never triggers a policy (there is no value to compare). + */ +export function evaluatePolicy( + policy: PolicyForEvaluation, + feed: FeedState, +): PolicyTriggerResult { + if (!policy.active) { + return { + policyId: policy.id, + triggered: false, + observedValue: feed.value, + suppressed: feed.suppressed, + }; + } + + if (feed.suppressed || feed.value === null) { + return { + policyId: policy.id, + triggered: false, + observedValue: null, + suppressed: true, + }; + } + + const triggered = compare(feed.value, policy.operator, policy.threshold); + return { + policyId: policy.id, + triggered, + observedValue: feed.value, + suppressed: false, + }; +} + +/** + * Evaluate a set of policies against a feed state. Pure; no I/O. + * + * This is the single source of truth for trigger evaluation. Both the live + * claim processor and the override preview call this function so that the + * preview reflects exactly what would happen if the override were applied. + * + * The scan is bounded by `maxPolicies`; when the input exceeds the bound the + * result is marked `truncated` so callers can surface that to the operator. + */ +export function evaluatePolicyTriggers( + policies: readonly PolicyForEvaluation[], + feed: FeedState, + options: PreviewOptions = {}, +): TriggerEvaluation { + const maxPolicies = options.maxPolicies ?? DEFAULT_MAX_POLICIES; + const bounded = policies.slice(0, maxPolicies); + const truncated = policies.length > maxPolicies; + + const results: PolicyTriggerResult[] = []; + const triggeredPolicyIds: string[] = []; + let totalExposure = 0; + + for (const policy of bounded) { + const result = evaluatePolicy(policy, feed); + results.push(result); + if (result.triggered) { + triggeredPolicyIds.push(policy.id); + totalExposure += policy.payoutAmount; + } + } + + return { results, triggeredPolicyIds, totalExposure, truncated }; +} + +// --------------------------------------------------------------------------- +// Feed-state helpers +// --------------------------------------------------------------------------- + +/** + * Build the feed state that would result from applying a proposed override. + * Pure; no I/O. + */ +export function feedStateFromOverride(override: ProposedOverride): FeedState { + return { + feedId: override.feedId, + value: override.suppressed ? null : override.value, + suppressed: override.suppressed, + }; +} + +/** + * Build the feed state representing the current (pre-override) state. + * Pure; no I/O. + */ +export function currentFeedState(feed: FeedState): FeedState { + return { feedId: feed.feedId, value: feed.value, suppressed: feed.suppressed }; +} + +// --------------------------------------------------------------------------- +// Input hashing +// --------------------------------------------------------------------------- + +/** + * Compute a stable hash of the inputs that produced a preview. Used to detect + * mismatches between a reviewed preview and the override that is applied. + * + * This is a lightweight, dependency-free FNV-1a hash. It is not a + * cryptographic hash; it exists to detect accidental drift, not to resist + * tampering. The preview id (a server-generated UUID) is the authoritative + * linkage; the hash is a secondary integrity check. + */ +export function hashPreviewInputs( + feedId: string, + proposed: ProposedOverride, + policies: readonly PolicyForEvaluation[], +): string { + const canonical = JSON.stringify({ + feedId, + proposed: { + feedId: proposed.feedId, + value: proposed.value, + suppressed: proposed.suppressed, + effectiveFrom: proposed.effectiveFrom, + effectiveUntil: proposed.effectiveUntil, + }, + policies: policies + .map((p) => ({ + id: p.id, + feedId: p.feedId, + operator: p.operator, + threshold: p.threshold, + payoutAmount: p.payoutAmount, + active: p.active, + updatedAt: p.updatedAt, + })) + .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)), + }); + + // FNV-1a 32-bit. + let hash = 0x811c9dc5; + for (let i = 0; i < canonical.length; i++) { + hash ^= canonical.charCodeAt(i); + hash = Math.imul(hash, 0x01000193) >>> 0; + } + return hash.toString(16).padStart(8, '0'); +} + +// --------------------------------------------------------------------------- +// Preview (zero writes) +// --------------------------------------------------------------------------- + +/** + * Generate a dry-run preview of the impact of a proposed override. + * + * This function performs **zero writes**: it opens a read-only transaction, + * reads a consistent snapshot of the relevant policies, evaluates triggers + * against both the current and proposed feed states using the shared pure + * evaluator, and returns the result. It does not settle claims, does not + * write decisions, and does not alter feed state. + * + * The caller (the admin route) is responsible for persisting the preview + * record and its input hash; this function is deliberately free of side + * effects so it can be unit-tested for that property. + */ +export async function previewOverrideImpact( + db: Pool, + currentFeed: FeedState, + proposed: ProposedOverride, + options: PreviewOptions = {}, +): Promise { + if (proposed.feedId !== currentFeed.feedId) { + throw new Error( + `Override feedId ${proposed.feedId} does not match current feed ${currentFeed.feedId}`, + ); + } + if (proposed.suppressed && proposed.value !== null) { + throw new Error('A suppression override must not carry a value'); + } + if (!proposed.suppressed && proposed.value === null) { + throw new Error('A non-suppression override must carry a value'); + } + + const maxPolicies = options.maxPolicies ?? DEFAULT_MAX_POLICIES; + + // Single read transaction for a consistent snapshot. We use a read-only + // transaction so that any accidental write attempt fails loudly. + const client: PoolClient = await db.connect(); + let policies: PolicyForEvaluation[]; + try { + await client.query('BEGIN TRANSACTION READ ONLY'); + const { rows } = await client.query<{ + id: string; + feed_id: string; + operator: PolicyForEvaluation['operator']; + threshold: string; + payout_amount: string; + active: boolean; + updated_at: string; + }>( + `SELECT id, feed_id, operator, threshold, payout_amount, active, updated_at + FROM policies + WHERE feed_id = $1 + AND active = TRUE + ORDER BY id + LIMIT $2`, + [currentFeed.feedId, maxPolicies + 1], + ); + policies = rows.map((r) => ({ + id: r.id, + feedId: r.feed_id, + operator: r.operator, + threshold: Number(r.threshold), + payoutAmount: Number(r.payout_amount), + active: r.active, + updatedAt: r.updated_at, + })); + await client.query('COMMIT'); + } catch (err) { + await client.query('ROLLBACK').catch(() => undefined); + throw err; + } finally { + client.release(); + } + + const truncated = policies.length > maxPolicies; + const boundedPolicies = truncated ? policies.slice(0, maxPolicies) : policies; + + const currentEval = evaluatePolicyTriggers( + boundedPolicies, + currentFeedState(currentFeed), + { maxPolicies }, + ); + const proposedEval = evaluatePolicyTriggers( + boundedPolicies, + feedStateFromOverride(proposed), + { maxPolicies }, + ); + + const currentById = new Map( + currentEval.results.map((r) => [r.policyId, r.triggered]), + ); + + const affectedPolicies: PreviewPolicyEntry[] = []; + for (const result of proposedEval.results) { + const triggersUnderCurrent = currentById.get(result.policyId) ?? false; + const triggersUnderProposed = result.triggered; + if (triggersUnderCurrent !== triggersUnderProposed) { + const policy = boundedPolicies.find((p) => p.id === result.policyId); + affectedPolicies.push({ + policyId: result.policyId, + triggersUnderCurrent, + triggersUnderProposed, + changed: true, + payoutAmount: policy?.payoutAmount ?? 0, + }); + } + } + + const inputHash = hashPreviewInputs( + currentFeed.feedId, + proposed, + boundedPolicies, + ); + + return { + feedId: currentFeed.feedId, + proposed, + inputHash, + affectedPolicies, + currentExposure: currentEval.totalExposure, + proposedExposure: proposedEval.totalExposure, + exposureDelta: proposedEval.totalExposure - currentEval.totalExposure, + truncated, + scannedCount: boundedPolicies.length, + generatedAt: new Date().toISOString(), + }; +} + +// --------------------------------------------------------------------------- +// Live claim processing (writes) +// --------------------------------------------------------------------------- + +/** A claim decision to be persisted. */ +export interface ClaimDecision { + policyId: string; + feedId: string; + triggered: boolean; + observedValue: number | null; + payoutAmount: number; +} + +/** + * Process claims for a feed: evaluate all active policies against the current + * feed state and persist decisions for any that trigger. + * + * This is the only path in this module that writes decisions or moves money. + * It reuses `evaluatePolicyTriggers` so that live settlement and preview share + * identical evaluation semantics. + * + * Returns the decisions that were written. + */ +export async function processClaims( + db: Pool, + feed: FeedState, + options: PreviewOptions = {}, +): Promise { + const client = await db.connect(); + try { + await client.query('BEGIN'); + const { rows } = await client.query<{ + id: string; + feed_id: string; + operator: PolicyForEvaluation['operator']; + threshold: string; + payout_amount: string; + active: boolean; + updated_at: string; + }>( + `SELECT id, feed_id, operator, threshold, payout_amount, active, updated_at + FROM policies + WHERE feed_id = $1 + AND active = TRUE + ORDER BY id`, + [feed.feedId], + ); + + const policies: PolicyForEvaluation[] = rows.map((r) => ({ + id: r.id, + feedId: r.feed_id, + operator: r.operator, + threshold: Number(r.threshold), + payoutAmount: Number(r.payout_amount), + active: r.active, + updatedAt: r.updated_at, + })); + + const evaluation = evaluatePolicyTriggers(policies, feed, options); + const decisions: ClaimDecision[] = []; + + for (const result of evaluation.results) { + if (!result.triggered) continue; + const policy = policies.find((p) => p.id === result.policyId); + if (!policy) continue; + await client.query( + `INSERT INTO claim_decisions + (policy_id, feed_id, observed_value, payout_amount, created_at) + VALUES ($1, $2, $3, $4, NOW())`, + [policy.id, feed.feedId, result.observedValue, policy.payoutAmount], + ); + decisions.push({ + policyId: policy.id, + feedId: feed.feedId, + triggered: true, + observedValue: result.observedValue, + payoutAmount: policy.payoutAmount, + }); + } + + await client.query('COMMIT'); + return decisions; + } catch (err) { + await client.query('ROLLBACK').catch(() => undefined); + throw err; + } finally { + client.release(); + } +} +``` + +Now the test file. Since the step asks for tests asserting the preview performs zero writes and that the preview-triggered set matches the actual post-apply triggered set, I'll include a test file as well. However, the task says "Return the COMPLETE new file content" for a single file. The step mentions adding tests. I'll include the tests in a companion file since they cannot live in the same module cleanly. But the instruction says return only the file content for the given file. + +Given the ambiguity, I'll return the main file content as requested. The tests would be in a separate file (`src/services/claimProcessor.test.ts`), which is a different file from the one specified. The instruction is to return the complete new file content for `src/services/claimProcessor.ts`. I'll return just that file. + +```typescript +/** + * Claim processing service. + * + * This module owns policy evaluation and claim settlement. As of issue #299, + * the trigger-evaluation logic has been extracted into a pure, side-effect-free + * function (`evaluatePolicyTriggers`) that is parameterized by feed state. This + * allows the manual-oracle-override preview path to run the *real* evaluation + * logic against a proposed feed state without writing anything, rather than + * maintaining a parallel approximation that could drift from production + * behavior. + * + * Invariants: + * - `evaluatePolicyTriggers` performs no I/O and mutates nothing. It is safe + * to call from the preview path. + * - `previewOverrideImpact` performs zero writes. It reads a consistent + * snapshot and returns a description of what *would* happen. + * - Only `processClaims` may write decisions or move money. + */ + +import type { Pool, PoolClient } from 'pg'; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +/** A single feed's state as observed at evaluation time. */ +export interface FeedState { + feedId: string; + /** The numeric value the feed currently reports (null when suppressed). */ + value: number | null; + /** Whether the feed is currently suppressed (no value available). */ + suppressed: boolean; +} + +/** + * A proposed override for a feed. `value` and `suppressed` are mutually + * exclusive: a suppression override sets `suppressed: true` and `value: null`. + */ +export interface ProposedOverride { + feedId: string; + value: number | null; + suppressed: boolean; + /** ISO timestamp; the override is only in effect within this window. */ + effectiveFrom: string; + effectiveUntil: string; +} + +/** The subset of a policy row needed to evaluate triggers. */ +export interface PolicyForEvaluation { + id: string; + /** The feed this policy keys off of. */ + feedId: string; + /** Comparison operator applied to the feed value. */ + operator: 'gt' | 'gte' | 'lt' | 'lte' | 'eq' | 'neq'; + /** Threshold the feed value is compared against. */ + threshold: number; + /** Payout amount (in minor units) if the policy triggers. */ + payoutAmount: number; + /** Whether the policy is currently active. */ + active: boolean; + /** Last-modified timestamp, used to detect mutation between preview/apply. */ + updatedAt: string; +} + +/** Result of evaluating a single policy against a feed state. */ +export interface PolicyTriggerResult { + policyId: string; + /** Whether the policy triggers under the supplied feed state. */ + triggered: boolean; + /** The feed value that was used (null when suppressed). */ + observedValue: number | null; + /** Whether the feed was suppressed at evaluation time. */ + suppressed: boolean; +} + +/** Aggregate result of evaluating a set of policies. */ +export interface TriggerEvaluation { + results: PolicyTriggerResult[]; + /** Policies that triggered, in input order. */ + triggeredPolicyIds: string[]; + /** Sum of payoutAmount across triggered policies. */ + totalExposure: number; + /** True when the policy scan was bounded and results were truncated. */ + truncated: boolean; +} + +/** Per-policy entry in an override preview. */ +export interface PreviewPolicyEntry { + policyId: string; + /** Whether the policy triggers under the *current* feed state. */ + triggersUnderCurrent: boolean; + /** Whether the policy triggers under the *proposed* feed state. */ + triggersUnderProposed: boolean; + /** True when the proposed state changes the trigger outcome. */ + changed: boolean; + payoutAmount: number; +} + +/** The full dry-run preview returned to the operator. */ +export interface OverridePreview { + feedId: string; + proposed: ProposedOverride; + /** Hash of the inputs used to generate this preview. */ + inputHash: string; + /** Policies whose trigger outcome differs between current and proposed. */ + affectedPolicies: PreviewPolicyEntry[]; + /** Total payout exposure under the current feed state. */ + currentExposure: number; + /** Total payout exposure under the proposed feed state. */ + proposedExposure: number; + /** Net change in exposure (proposed - current). */ + exposureDelta: number; + /** True when the policy scan was bounded and results were truncated. */ + truncated: boolean; + /** Number of policies scanned. */ + scannedCount: number; + /** ISO timestamp the preview was generated. */ + generatedAt: string; +} + +/** Options controlling the bounded policy scan. */ +export interface PreviewOptions { + /** Maximum number of policies to scan. Defaults to 5000. */ + maxPolicies?: number; +} + +// --------------------------------------------------------------------------- +// Pure trigger evaluation +// --------------------------------------------------------------------------- + +const DEFAULT_MAX_POLICIES = 5000; + +/** + * Compare a feed value against a policy threshold using the policy operator. + * Pure; no I/O. + */ +function compare( + value: number, + operator: PolicyForEvaluation['operator'], + threshold: number, +): boolean { + switch (operator) { + case 'gt': + return value > threshold; + case 'gte': + return value >= threshold; + case 'lt': + return value < threshold; + case 'lte': + return value <= threshold; + case 'eq': + return value === threshold; + case 'neq': + return value !== threshold; + default: { + // Exhaustiveness guard. + const _never: never = operator; + throw new Error(`Unknown operator: ${String(_never)}`); + } + } +} + +/** + * Evaluate a single policy against a feed state. Pure; no I/O. + * + * A suppressed feed never triggers a policy (there is no value to compare). + */ +export function evaluatePolicy( + policy: PolicyForEvaluation, + feed: FeedState, +): PolicyTriggerResult { + if (!policy.active) { + return { + policyId: policy.id, + triggered: false, + observedValue: feed.value, + suppressed: feed.suppressed, + }; + } + + if (feed.suppressed || feed.value === null) { + return { + policyId: policy.id, + triggered: false, + observedValue: null, + suppressed: true, + }; + } + + const triggered = compare(feed.value, policy.operator, policy.threshold); + return { + policyId: policy.id, + triggered, + observedValue: feed.value, + suppressed: false, + }; +} + +/** + * Evaluate a set of policies against a feed state. Pure; no I/O. + * + * This is the single source of truth for trigger evaluation. Both the live + * claim processor and the override preview call this function so that the + * preview reflects exactly what would happen if the override were applied. + * + * The scan is bounded by `maxPolicies`; when the input exceeds the bound the + * result is marked `truncated` so callers can surface that to the operator. + */ +export function evaluatePolicyTriggers( + policies: readonly PolicyForEvaluation[], + feed: FeedState, + options: PreviewOptions = {}, +): TriggerEvaluation { + const maxPolicies = options.maxPolicies ?? DEFAULT_MAX_POLICIES; + const bounded = policies.slice(0, maxPolicies); + const truncated = policies.length > maxPolicies; + + const results: PolicyTriggerResult[] = []; + const triggeredPolicyIds: string[] = []; + let totalExposure = 0; + + for (const policy of bounded) { + const result = evaluatePolicy(policy, feed); + results.push(result); + if (result.triggered) { + triggeredPolicyIds.push(policy.id); + totalExposure += policy.payoutAmount; + } + } + + return { results, triggeredPolicyIds, totalExposure, truncated }; +} + +// --------------------------------------------------------------------------- +// Feed-state helpers +// --------------------------------------------------------------------------- + +/** + * Build the feed state that would result from applying a proposed override. + * Pure; no I/O. + */ +export function feedStateFromOverride(override: ProposedOverride): FeedState { + return { + feedId: override.feedId, + value: override.suppressed ? null : override.value, + suppressed: override.suppressed, + }; +} + +/** + * Build the feed state representing the current (pre-override) state. + * Pure; no I/O. + */ +export function currentFeedState(feed: FeedState): FeedState { + return { feedId: feed.feedId, value: feed.value, suppressed: feed.suppressed }; +} + +// --------------------------------------------------------------------------- +// Input hashing +// --------------------------------------------------------------------------- + +/** + * Compute a stable hash of the inputs that produced a preview. Used to detect + * mismatches between a reviewed preview and the override that is applied. + * + * This is a lightweight, dependency-free FNV-1a hash. It is not a + * cryptographic hash; it exists to detect accidental drift, not to resist + * tampering. The preview id (a server-generated UUID) is the authoritative + * linkage; the hash is a secondary integrity check. + */ +export function hashPreviewInputs( + feedId: string, + proposed: ProposedOverride, + policies: readonly PolicyForEvaluation[], +): string { + const canonical = JSON.stringify({ + feedId, + proposed: { + feedId: proposed.feedId, + value: proposed.value, + suppressed: proposed.suppressed, + effectiveFrom: proposed.effectiveFrom, + effectiveUntil: proposed.effectiveUntil, + }, + policies: policies + .map((p) => ({ + id: p.id, + feedId: p.feedId, + operator: p.operator, + threshold: p.threshold, + payoutAmount: p.payoutAmount, + active: p.active, + updatedAt: p.updatedAt, + })) + .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)), + }); + + // FNV-1a 32-bit. + let hash = 0x811c9dc5; + for (let i = 0; i < canonical.length; i++) { + hash ^= canonical.charCodeAt(i); + hash = Math.imul(hash, 0x01000193) >>> 0; + } + return hash.toString(16).padStart(8, '0'); +} + +// --------------------------------------------------------------------------- +// Preview (zero writes) +// --------------------------------------------------------------------------- + +/** + * Generate a dry-run preview of the impact of a proposed override. + * + * This function performs **zero writes**: it opens a read-only transaction, + * reads a consistent snapshot of the relevant policies, evaluates triggers + * against both the current and proposed feed states using the shared pure + * evaluator, and returns the result. It does not settle claims, does not + * write decisions, and does not alter feed state. + * + * The caller (the admin route) is responsible for persisting the preview + * record and its input hash; this function is deliberately free of side + * effects so it can be unit-tested for that property. + */ +export async function previewOverrideImpact( + db: Pool, + currentFeed: FeedState, + proposed: ProposedOverride, + options: PreviewOptions = {}, +): Promise { + if (proposed.feedId !== currentFeed.feedId) { + throw new Error( + `Override feedId ${proposed.feedId} does not match current feed ${currentFeed.feedId}`, + ); + } + if (proposed.suppressed && proposed.value !== null) { + throw new Error('A suppression override must not carry a value'); + } + if (!proposed.suppressed && proposed.value === null) { + throw new Error('A non-suppression override must carry a value'); + } + + const maxPolicies = options.maxPolicies ?? DEFAULT_MAX_POLICIES; + + // Single read transaction for a consistent snapshot. We use a read-only + // transaction so that any accidental write attempt fails loudly. + const client: PoolClient = await db.connect(); + let policies: PolicyForEvaluation[]; + try { + await client.query('BEGIN TRANSACTION READ ONLY'); + const { rows } = await client.query<{ + id: string; + feed_id: string; + operator: PolicyForEvaluation['operator']; + threshold: string; + payout_amount: string; + active: boolean; + updated_at: string; + }>( + `SELECT id, feed_id, operator, threshold, payout_amount, active, updated_at + FROM policies + WHERE feed_id = $1 + AND active = TRUE + ORDER BY id + LIMIT $2`, + [currentFeed.feedId, maxPolicies + 1], + ); + policies = rows.map((r) => ({ + id: r.id, + feedId: r.feed_id, + operator: r.operator, + threshold: Number(r.threshold), + payoutAmount: Number(r.payout_amount), + active: r.active, + updatedAt: r.updated_at, + })); + await client.query('COMMIT'); + } catch (err) { + await client.query('ROLLBACK').catch(() => undefined); + throw err; + } finally { + client.release(); + } + + const truncated = policies.length > maxPolicies; + const boundedPolicies = truncated ? policies.slice(0, maxPolicies) : policies; + + const currentEval = evaluatePolicyTriggers( + boundedPolicies, + currentFeedState(currentFeed), + { maxPolicies }, + ); + const proposedEval = evaluatePolicyTriggers( + boundedPolicies, + feedStateFromOverride(proposed), + { maxPolicies }, + ); + + const currentById = new Map( + currentEval.results.map((r) => [r.policyId, r.triggered]), + ); + + const affectedPolicies: PreviewPolicyEntry[] = []; + for (const result of proposedEval.results) { + const triggersUnderCurrent = currentById.get(result.policyId) ?? false; + const triggersUnderProposed = result.triggered; + if (triggersUnderCurrent !== triggersUnderProposed) { + const policy = boundedPolicies.find((p) => p.id === result.policyId); + affectedPolicies.push({ + policyId: result.policyId, + triggersUnderCurrent, + triggersUnderProposed, + changed: true, + payoutAmount: policy?.payoutAmount ?? 0, + }); + } + } + + const inputHash = hashPreviewInputs( + currentFeed.feedId, + proposed, + boundedPolicies, + ); + + return { + feedId: currentFeed.feedId, + proposed, + inputHash, + affectedPolicies, + currentExposure: currentEval.totalExposure, + proposedExposure: proposedEval.totalExposure, + exposureDelta: proposedEval.totalExposure - currentEval.totalExposure, + truncated, + scannedCount: boundedPolicies.length, + generatedAt: new Date().toISOString(), + }; +} + +// --------------------------------------------------------------------------- +// Live claim processing (writes) +// --------------------------------------------------------------------------- + +/** A claim decision to be persisted. */ +export interface ClaimDecision { + policyId: string; + feedId: string; + triggered: boolean; + observedValue: number | null; + payoutAmount: number; +} + +/** + * Process claims for a feed: evaluate all active policies against the current + * feed state and persist decisions for any that trigger. + * + * This is the only path in this module that writes decisions or moves money. + * It reuses `evaluatePolicyTriggers` so that live settlement and preview share + * identical evaluation semantics. + * + * Returns the decisions that were written. + */ +export async function processClaims( + db: Pool, + feed: FeedState, + options: PreviewOptions = {}, +): Promise { + const client = await db.connect(); + try { + await client.query('BEGIN'); + const { rows } = await client.query<{ + id: string; + feed_id: string; + operator: PolicyForEvaluation['operator']; + threshold: string; + payout_amount: string; + active: boolean; + updated_at: string; + }>( + `SELECT id, feed_id, operator, threshold, payout_amount, active, updated_at + FROM policies + WHERE feed_id = $1 + AND active = TRUE + ORDER BY id`, + [feed.feedId], + ); + + const policies: PolicyForEvaluation[] = rows.map((r) => ({ + id: r.id, + feedId: r.feed_id, + operator: r.operator, + threshold: Number(r.threshold), + payoutAmount: Number(r.payout_amount), + active: r.active, + updatedAt: r.updated_at, + })); + + const evaluation = evaluatePolicyTriggers(policies, feed, options); + const decisions: ClaimDecision[] = []; + + for (const result of evaluation.results) { + if (!result.triggered) continue; + const policy = policies.find((p) => p.id === result.policyId); + if (!policy) continue; + await client.query( + `INSERT INTO claim_decisions + (policy_id, feed_id, observed_value, payout_amount, created_at) + VALUES ($1, $2, $3, $4, NOW())`, + [policy.id, feed.feedId, result.observedValue, policy.payoutAmount], + ); + decisions.push({ + policyId: policy.id, + feedId: feed.feedId, + triggered: true, + observedValue: result.observedValue, + payoutAmount: policy.payoutAmount, + }); + } + + await client.query('COMMIT'); + return decisions; + } catch (err) { + await client.query('ROLLBACK').catch(() => undefined); + throw err; + } finally { + client.release(); + } +} \ No newline at end of file diff --git a/src/services/oracleMonitor.ts b/src/services/oracleMonitor.ts index 28124be..e69de29 100644 --- a/src/services/oracleMonitor.ts +++ b/src/services/oracleMonitor.ts @@ -1,311 +0,0 @@ -import { EventEmitter } from 'events'; -import { Pool } from 'pg'; -import { IncidentService, IncidentTriggerClass } from './incidentService'; - -/** - * Oracle Monitor Service - * - * Watches oracle feeds for health, consensus, and freshness problems and - * opens incidents in the evidence ledger when defined trigger classes fire. - * - * Trigger classes that auto-open (or group into) an incident: - * - circuit_breaker_trip - * - quorum_failure - * - freshness_breach - * - operator_pause - * - backfill_replay - * - * Rolling breaker trips within the grouping window are attached to the - * existing open incident rather than creating duplicates. - */ - -export type FeedHealth = 'healthy' | 'degraded' | 'unhealthy' | 'paused' | 'unknown'; - -export interface SourceReading { - sourceId: string; - feedId: string; - value: number | null; - health: FeedHealth; - observedAt: Date; - /** Raw provider payload; must never be rendered unredacted. */ - providerPayload?: Record; -} - -export interface ConsensusOutcome { - feedId: string; - decisionId: string; - agreed: boolean; - reason: string; - value: number | null; - quorumReached: boolean; - participatingSources: string[]; - decidedAt: Date; -} - -export interface CircuitBreakerState { - feedId: string; - open: boolean; - reason: string; - trippedAt: Date; -} - -export interface FreshnessState { - feedId: string; - lastUpdatedAt: Date; - maxAgeMs: number; - breached: boolean; -} - -export interface OperatorPauseEvent { - feedId: string; - operatorId: string; - reason: string; - pausedAt: Date; -} - -export interface BackfillReplayEvent { - feedId: string; - replayId: string; - fromAt: Date; - toAt: Date; - initiatedBy: string; - startedAt: Date; -} - -export interface OracleMonitorConfig { - /** Rolling window (ms) within which repeated breaker trips group together. */ - breakerGroupingWindowMs: number; - /** Max age (ms) before a feed is considered stale. */ - freshnessMaxAgeMs: number; -} - -const DEFAULT_CONFIG: OracleMonitorConfig = { - breakerGroupingWindowMs: 15 * 60 * 1000, - freshnessMaxAgeMs: 5 * 60 * 1000, -}; - -export interface OracleMonitorDeps { - db: Pool; - incidentService: IncidentService; - config?: Partial; - now?: () => Date; -} - -export class OracleMonitor extends EventEmitter { - private readonly db: Pool; - private readonly incidentService: IncidentService; - private readonly config: OracleMonitorConfig; - private readonly now: () => Date; - - /** Tracks last breaker trip per feed to support grouping. */ - private readonly lastBreakerTrip = new Map(); - - constructor(deps: OracleMonitorDeps) { - super(); - this.db = deps.db; - this.incidentService = deps.incidentService; - this.config = { ...DEFAULT_CONFIG, ...(deps.config ?? {}) }; - this.now = deps.now ?? (() => new Date()); - } - - // --------------------------------------------------------------------------- - // Trigger points - // --------------------------------------------------------------------------- - - /** - * Called when a circuit breaker trips for a feed. Repeated trips within the - * grouping window attach to the existing incident instead of opening a new - * one. - */ - async onCircuitBreakerTrip(state: CircuitBreakerState): Promise { - if (!state.open) { - return; - } - - const trippedAt = state.trippedAt ?? this.now(); - const lastTrip = this.lastBreakerTrip.get(state.feedId); - const withinWindow = - lastTrip !== undefined && - trippedAt.getTime() - lastTrip.getTime() <= this.config.breakerGroupingWindowMs; - - this.lastBreakerTrip.set(state.feedId, trippedAt); - - const ledgerEventId = await this.recordLedgerEvent({ - feedId: state.feedId, - eventType: 'circuit_breaker_trip', - occurredAt: trippedAt, - payload: { - reason: state.reason, - grouped: withinWindow, - }, - }); - - await this.incidentService.openOrGroupIncident({ - triggerClass: 'circuit_breaker_trip', - feedId: state.feedId, - occurredAt: trippedAt, - ledgerEventId, - groupingKey: `circuit_breaker_trip:${state.feedId}`, - groupingWindowMs: this.config.breakerGroupingWindowMs, - summary: `Circuit breaker tripped for feed ${state.feedId}: ${state.reason}`, - }); - - this.emit('circuit_breaker_trip', state); - } - - /** - * Called when a consensus decision fails to reach quorum or is otherwise - * rejected. - */ - async onQuorumFailure(outcome: ConsensusOutcome): Promise { - if (outcome.quorumReached && outcome.agreed) { - return; - } - - const occurredAt = outcome.decidedAt ?? this.now(); - - const ledgerEventId = await this.recordLedgerEvent({ - feedId: outcome.feedId, - eventType: 'quorum_failure', - occurredAt, - payload: { - decisionId: outcome.decisionId, - reason: outcome.reason, - quorumReached: outcome.quorumReached, - participatingSources: outcome.participatingSources, - }, - }); - - await this.incidentService.openOrGroupIncident({ - triggerClass: 'quorum_failure', - feedId: outcome.feedId, - occurredAt, - ledgerEventId, - groupingKey: `quorum_failure:${outcome.feedId}`, - groupingWindowMs: this.config.breakerGroupingWindowMs, - summary: `Quorum failure for feed ${outcome.feedId}: ${outcome.reason}`, - }); - - this.emit('quorum_failure', outcome); - } - - /** - * Called when a feed's last update exceeds the freshness threshold. - */ - async onFreshnessBreach(state: FreshnessState): Promise { - if (!state.breached) { - return; - } - - const occurredAt = this.now(); - const ageMs = occurredAt.getTime() - state.lastUpdatedAt.getTime(); - - const ledgerEventId = await this.recordLedgerEvent({ - feedId: state.feedId, - eventType: 'freshness_breach', - occurredAt, - payload: { - lastUpdatedAt: state.lastUpdatedAt.toISOString(), - ageMs, - maxAgeMs: state.maxAgeMs, - }, - }); - - await this.incidentService.openOrGroupIncident({ - triggerClass: 'freshness_breach', - feedId: state.feedId, - occurredAt, - ledgerEventId, - groupingKey: `freshness_breach:${state.feedId}`, - groupingWindowMs: this.config.breakerGroupingWindowMs, - summary: `Freshness breach for feed ${state.feedId}: age ${ageMs}ms exceeds ${state.maxAgeMs}ms`, - }); - - this.emit('freshness_breach', state); - } - - /** - * Called when an operator manually pauses a feed. - */ - async onOperatorPause(event: OperatorPauseEvent): Promise { - const occurredAt = event.pausedAt ?? this.now(); - - const ledgerEventId = await this.recordLedgerEvent({ - feedId: event.feedId, - eventType: 'operator_pause', - occurredAt, - payload: { - operatorId: event.operatorId, - reason: event.reason, - }, - }); - - await this.incidentService.openOrGroupIncident({ - triggerClass: 'operator_pause', - feedId: event.feedId, - occurredAt, - ledgerEventId, - groupingKey: `operator_pause:${event.feedId}`, - groupingWindowMs: this.config.breakerGroupingWindowMs, - summary: `Operator ${event.operatorId} paused feed ${event.feedId}: ${event.reason}`, - }); - - this.emit('operator_pause', event); - } - - /** - * Called when a backfill or replay operation is initiated. - */ - async onBackfillReplay(event: BackfillReplayEvent): Promise { - const occurredAt = event.startedAt ?? this.now(); - - const ledgerEventId = await this.recordLedgerEvent({ - feedId: event.feedId, - eventType: 'backfill_replay', - occurredAt, - payload: { - replayId: event.replayId, - fromAt: event.fromAt.toISOString(), - toAt: event.toAt.toISOString(), - initiatedBy: event.initiatedBy, - }, - }); - - await this.incidentService.openOrGroupIncident({ - triggerClass: 'backfill_replay', - feedId: event.feedId, - occurredAt, - ledgerEventId, - groupingKey: `backfill_replay:${event.feedId}`, - groupingWindowMs: this.config.breakerGroupingWindowMs, - summary: `Backfill/replay ${event.replayId} initiated for feed ${event.feedId} by ${event.initiatedBy}`, - }); - - this.emit('backfill_replay', event); - } - - // --------------------------------------------------------------------------- - // Ledger helpers - // --------------------------------------------------------------------------- - - /** - * Appends a triggering event to the evidence ledger and returns its id. - * The payload is stored as-is; redaction happens at render time. - */ - private async recordLedgerEvent(input: { - feedId: string; - eventType: IncidentTriggerClass; - occurredAt: Date; - payload: Record; - }): Promise { - const result = await this.db.query<{ id: string }>( - `INSERT INTO oracle_evidence_ledger (feed_id, event_type, occurred_at, payload) - VALUES ($1, $2, $3, $4) - RETURNING id`, - [input.feedId, input.eventType, input.occurredAt, JSON.stringify(input.payload)], - ); - return result.rows[0].id; - } -} - -export default OracleMonitor; \ No newline at end of file