diff --git a/src/db/schema.sql b/src/db/schema.sql index 50be59c..a36c12b 100644 --- a/src/db/schema.sql +++ b/src/db/schema.sql @@ -1,7 +1,5 @@ --- Refract database schema --- NOTE: This file is provided as a partial view for issue #129. --- The full schema is managed elsewhere; only the claims table is shown here --- because the transparency API (settlement latency + loss ratio) depends on it. +-- Oracle schema +-- (existing tables omitted from this diff view; append the drift tables below) CREATE TABLE IF NOT EXISTS claims ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), @@ -22,6 +20,131 @@ CREATE TABLE IF NOT EXISTS claims ( updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +-- Schema drift detection (issue #279) +-- Records every structural validation result for a provider fetch so operators +-- can query a per-source drift report and inspect the last known-good sample. +CREATE TABLE IF NOT EXISTS oracle_schema_drift_events ( + id BIGSERIAL PRIMARY KEY, + source TEXT NOT NULL, + classification TEXT NOT NULL CHECK (classification IN ('ok', 'drift', 'breaking')), + reason TEXT, + -- redacted, size-limited sample of the offending payload + sample JSONB, + -- redacted, size-limited sample of the last payload that validated cleanly + last_known_good JSONB, + ingestion_id BIGINT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_oracle_schema_drift_events_source_created + ON oracle_schema_drift_events (source, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_oracle_schema_drift_events_classification + ON oracle_schema_drift_events (classification) + WHERE classification <> 'ok'; + +-- Quarantine state per source. A `breaking` classification flips this on and +-- blocks publication until an operator clears it. +CREATE TABLE IF NOT EXISTS oracle_source_quarantine ( + source TEXT PRIMARY KEY, + quarantined BOOLEAN NOT NULL DEFAULT FALSE, + reason TEXT, + last_event_id BIGINT REFERENCES oracle_schema_drift_events (id), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Operator reward accrual and distribution (issue #281) +-- Accruals are derived from the committed evidence ledger, not from live +-- submission events, so they are recomputable and auditable after the fact. +-- The idempotency key makes accrual exactly-once under retries and re-orgs. +CREATE TABLE IF NOT EXISTS oracle_reward_accruals ( + id BIGSERIAL PRIMARY KEY, + -- deterministic key: consensus round + operator + reading evidence id + idempotency_key TEXT NOT NULL UNIQUE, + operator_id TEXT NOT NULL, + consensus_round BIGINT NOT NULL, + evidence_id BIGINT NOT NULL, + -- snapshot of the weighting inputs at accrual time + stake_weight NUMERIC(38, 18) NOT NULL DEFAULT 0, + reputation_score NUMERIC(10, 6) NOT NULL DEFAULT 0, + consensus_weight NUMERIC(10, 6) NOT NULL DEFAULT 0, + -- base pool share before the per-operator cap is applied + raw_amount NUMERIC(38, 18) NOT NULL DEFAULT 0, + -- amount actually credited after the cap (may be less than raw_amount) + amount NUMERIC(38, 18) NOT NULL DEFAULT 0, + -- 'accrued' | 'invalidated' | 'paid' | 'adjusted' + status TEXT NOT NULL DEFAULT 'accrued' + CHECK (status IN ('accrued', 'invalidated', 'paid', 'adjusted')), + -- set when remediation invalidates a previously accepted reading + invalidated_at TIMESTAMPTZ, + invalidation_reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_operator + ON oracle_reward_accruals (operator_id, status); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_round + ON oracle_reward_accruals (consensus_round); + +-- Durable ledger of adjustments and payouts against accruals. Every row is +-- append-only; balances are derived by summing accruals minus p + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_operator + ON oracle_reward_accruals (operator_id, status); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_round + ON oracle_reward_accruals (consensus_round); + +-- Durable ledger of adjustments and payouts against accruals. Every row is +-- append-only; balances are derived by summing accruals minus payouts. +CREATE TABLE IF NOT EXISTS oracle_reward_ledger ( + id BIGSERIAL PRIMARY KEY, + -- deterministic key: payout batch + operator, or adjustment + accrual + idempotency_key TEXT NOT NULL UNIQUE, + operator_id TEXT NOT NULL, + accrual_id BIGINT REFERENCES oracle_reward_accruals (id), + payout_id BIGINT, + -- 'accrual' | 'adjustment' | 'payout' + entry_type TEXT NOT NULL + CHECK (entry_type IN ('accrual', 'adjustment', 'payout')), + -- signed: positive credits the operator, negative debits + amount NUMERIC(38, 18) NOT NULL, + memo TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_ledger_operator + ON oracle_reward_ledger (operator_id, created_at DESC); + +-- Scheduled distribution batches. Funds move only after the dispute window +-- closes and only when the operator balance meets the minimum threshold. +CREATE TABLE IF NOT EXISTS oracle_reward_payouts ( + id BIGSERIAL PRIMARY KEY, + -- deterministic key: distribution schedule slot + operator + idempotency_key TEXT NOT NULL UNIQUE, + operator_id TEXT NOT NULL, + amount NUMERIC(38, 18) NOT NULL, + -- 'pending' | 'disputed' | 'paid' | 'voided' + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'disputed', 'paid', 'voided')), + -- funds may not move before this timestamp (dispute window) + dispute_window_ends_at TIMESTAMPTZ NOT NULL, + paid_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_payouts_operator + ON oracle_reward_payouts (operator_id, status); + +CREATE INDEX IF NOT EXISTS idx_oracle_reward_payouts_pending + ON oracle_reward_payouts (dispute_window_ends_at) + WHERE status = 'pending'; + -- Per-feed degraded-mode state (issue #195). Fail-closed: only 'healthy' allows -- publishing or settlement. See docs/threat-model.md section 4A. CREATE TABLE IF NOT EXISTS oracle_feed_state ( diff --git a/src/oracle/consistency/evaluate.ts b/src/oracle/consistency/evaluate.ts new file mode 100644 index 0000000..3ce67b1 --- /dev/null +++ b/src/oracle/consistency/evaluate.ts @@ -0,0 +1,184 @@ +/** + * Cross-feed economic consistency evaluation (issue #280). + * + * Operates on canonicalized consensus values plus their timestamps using the + * same point-in-time semantics as consensus, so replay reproduces the verdict. + * A rule whose reference side is missing or whose timestamps fall outside the + * allowed lead/lag window is *skipped*, never failed. + */ + +import { + ConsistencyRule, + rulesForCoverage, +} from './rules'; + +/** Canonicalized consensus observation for a single feed. */ +export interface ConsensusObservation { + coverage: string; + /** Canonicalized consensus value (already normalized by consensus). */ + value: number; + /** Timestamp of the consensus observation (ms since epoch). */ + timestamp: number; +} + +export type ConsistencyVerdict = 'coherent' | 'violation' | 'skipped'; + +export interface ConsistencyEvaluation { + ruleId: string; + dependentCoverage: string; + referenceCoverages: string[]; + verdict: ConsistencyVerdict; + severity: ConsistencyRule['severity']; + /** Human-readable explanation, present on violations and skips. */ + explanation: string; + /** Relative move of the dependent feed, when computed. */ + dependentMove?: number; + /** Relative move of the reference feed, when computed. */ + referenceMove?: number; +} + +function relativeMove(from: number, to: number): number { + if (from === 0) return 0; + return (to - from) / Math.abs(from); +} + +function findObservation( + observations: ConsensusObservation[], + coverage: string, +): ConsensusObservation | undefined { + return observations.find((obs) => obs.coverage === coverage); +} + +/** + * Evaluate a single rule against the provided consensus observations. + * Deterministic: depends only on the observations passed in. + */ +export function evaluateRule( + rule: ConsistencyRule, + observations: ConsensusObservation[], +): ConsistencyEvaluation { + const dependent = findObservation(observations, rule.dependentCoverage); + if (!dependent) { + return { + ruleId: rule.id, + dependentCoverage: rule.dependentCoverage, + referenceCoverages: rule.referenceCoverages, + verdict: 'skipped', + severity: rule.severity, + explanation: `Skipped ${rule.id}: dependent feed ${rule.dependentCoverage} has no consensus observation.`, + }; + } + + const references = rule.referenceCoverages + .map((coverage) => findObservation(observations, coverage)) + .filter((obs): obs is ConsensusObservation => obs !== undefined); + + if (references.length === 0) { + return { + ruleId: rule.id, + dependentCoverage: rule.dependentCoverage, + referenceCoverages: rule.referenceCoverages, + verdict: 'skipped', + severity: rule.severity, + explanation: `Skipped ${rule.id}: no reference observation for ${rule.referenceCoverages.join(', ')}.`, + }; + } + + // Differing freshness windows: skip if any reference is outside lead/lag. + const withinWindow = references.filter( + (ref) => Math.abs(ref.timestamp - dependent.timestamp) <= rule.maxLeadLagMs, + ); + if (withinWindow.length === 0) { + return { + ruleId: rule.id, + dependentCoverage: rule.dependentCoverage, + referenceCoverages: rule.referenceCoverages, + verdict: 'skipped', + severity: rule.severity, + explanation: `Skipped ${rule.id}: reference observations fall outside the allowed lead/lag window of ${rule.maxLeadLagMs}ms.`, + }; + } + + // Use the most recent in-window reference for the move comparison. + const reference = withinWindow.reduce((latest, ref) => + ref.timestamp > latest.timestamp ? ref : latest, + ); + + // Relative moves are computed against the dependent's own prior value when + // available; here we compare the dependent's move to the reference's move + // using the same observation set so replay is deterministic. + const dependentMove = relativeMove(reference.value, dependent.value); + const referenceMove = relativeMove(dependent.value, reference.value); + + const dependentTriggered = Math.abs(dependentMove) >= rule.triggerThreshold; + if (!dependentTriggered) { + return { + ruleId: rule.id, + dependentCoverage: rule.dependentCoverage, + referenceCoverages: rule.referenceCoverages, + verdict: 'coherent', + severity: rule.severity, + explanation: `Coherent ${rule.id}: dependent move ${dependentMove.toFixed(4)} below trigger threshold ${rule.triggerThreshold}.`, + dependentMove, + referenceMove, + }; + } + + const referenceJustifies = Math.abs(referenceMove) >= rule.referenceThreshold; + const signOk = + rule.expectation === 'decoupled' || + (rule.expectation === 'positive' && Math.sign(dependentMove) === Math.sign(referenceMove)) || + (rule.expectation === 'negative' && Math.sign(dependentMove) !== Math.sign(referenceMove)); + + if (referenceJustifies && signOk) { + return { + ruleId: rule.id, + dependentCoverage: rule.dependentCoverage, + referenceCoverages: rule.referenceCoverages, + verdict: 'coherent', + severity: rule.severity, + explanation: `Coherent ${rule.id}: dependent move ${dependentMove.toFixed(4)} justified by reference move ${referenceMove.toFixed(4)}.`, + dependentMove, + referenceMove, + }; + } + + return { + ruleId: rule.id, + dependentCoverage: rule.dependentCoverage, + referenceCoverages: rule.referenceCoverages, + verdict: 'violation', + severity: rule.severity, + explanation: + `Incoherent ${rule.id}: ${rule.dependentCoverage} moved ${dependentMove.toFixed(4)} ` + + `but ${reference.coverage} moved ${referenceMove.toFixed(4)} ` + + `(expected ${rule.expectation}, reference threshold ${rule.referenceThreshold}). ${rule.rationale}`, + dependentMove, + referenceMove, + }; +} + +/** + * Evaluate all rules for a dependent coverage against the consensus + * observations. Returns one evaluation per applicable rule. + */ +export function evaluateConsistency( + dependentCoverage: string, + observations: ConsensusObservation[], +): ConsistencyEvaluation[] { + return rulesForCoverage(dependentCoverage).map((rule) => + evaluateRule(rule, observations), + ); +} + +/** + * Whether any evaluation should block publication of the dependent feed. + * Skipped rules never block. + */ +export function shouldBlockPublication( + evaluations: ConsistencyEvaluation[], +): boolean { + return evaluations.some( + (evaluation) => evaluation.verdict === 'violation' && evaluation.severity === 'block', + ); +} diff --git a/src/oracle/consistency/rules.ts b/src/oracle/consistency/rules.ts new file mode 100644 index 0000000..06f4ccf --- /dev/null +++ b/src/oracle/consistency/rules.ts @@ -0,0 +1,108 @@ +/** + * Declarative cross-feed economic consistency rules. + * + * Each rule expresses an economic expectation between a *dependent* feed and one + * or more *reference* feeds. Rules are evaluated after per-feed consensus and + * before the dependent feed is published (see `evaluate.ts` and the post-consensus + * gate in `oracleMonitor`). + * + * Design notes: + * - Rules operate on canonicalized consensus values plus their timestamps so that + * replaying the same point-in-time inputs reproduces the same verdict. + * - A rule is *skipped* (never failed) when either side of the pair is missing or + * stale relative to the rule's freshness window. + * - Severity is configurable per rule: `warn` annotates + emits a metric, `block` + * holds publication and trips the relevant circuit breaker. + */ + +export type ConsistencySeverity = 'warn' | 'block'; + +/** A canonicalized consensus value for a single feed at a point in time. */ +export interface ConsensusValue { + /** Feed identifier, e.g. `liquidation-volume`, `price`, `tvl`. */ + feed: string; + /** Canonicalized consensus value. */ + value: number; + /** ISO timestamp of the observation the consensus was computed from. */ + timestamp: string; +} + +/** + * A sign/ratio constraint between the dependent feed and a reference feed. + * `expect` describes the economically expected relationship. + */ +export type RatioConstraint = + | { kind: 'same-sign' } + | { kind: 'opposite-sign' } + | { kind: 'ratio-at-least'; ratio: number } + | { kind: 'ratio-at-most'; ratio: number }; + +export interface ConsistencyRule { + /** Stable identifier used in events, metrics and persisted evaluations. */ + id: string; + /** Human-readable description of the economic expectation. */ + description: string; + /** Rationale documented for operators and reviewers. */ + rationale: string; + /** The feed whose publication this rule gates. */ + dependentFeed: string; + /** Reference feed(s) the dependent feed must agree with. */ + referenceFeeds: string[]; + /** + * Maximum allowed lead/lag (ms) between the dependent and reference + * observations. If exceeded the rule is skipped rather than failed. + */ + maxLeadLagMs: number; + /** Constraint applied to the dependent/reference relationship. */ + constraint: RatioConstraint; + /** Configurable severity. */ + severity: ConsistencySeverity; +} + +/** + * Initial rule set for existing coverage types. + * + * LiquidationShield vs price/TVL: + * A liquidation-volume spike should coincide with either a price move or a TVL + * drawdown. A spike with flat price *and* flat TVL is economically implausible + * and is strong evidence of a spoofed/colluding liquidation source. + * + * FlightDelay vs route disruption: + * A flight-delay rate spike should be accompanied by a route disruption signal. + * A delay spike with no disruption signal indicates an incoherent delay source. + */ +export const DEFAULT_CONSISTENCY_RULES: ConsistencyRule[] = [ + { + id: 'liquidation-spike-requires-price-or-tvl-move', + description: + 'Liquidation volume spikes must be accompanied by a price move or TVL drawdown', + rationale: + 'Liquidations are triggered by adverse price moves and reduce TVL. A spike with ' + + 'flat price and flat TVL cannot be economically real and points to a bad source.', + dependentFeed: 'liquidation-volume', + referenceFeeds: ['price', 'tvl'], + maxLeadLagMs: 5 * 60 * 1000, + constraint: { kind: 'ratio-at-least', ratio: 0 }, + severity: 'warn', + }, + { + id: 'flight-delay-spike-requires-route-disruption', + description: 'Flight-delay rate spikes must coincide with a route disruption signal', + rationale: + 'Delays are caused by disruptions (weather, closures, strikes). A delay spike ' + + 'without any disruption signal is incoherent and suggests a spoofed delay feed.', + dependentFeed: 'flight-delay-rate', + referenceFeeds: ['route-disruption'], + maxLeadLagMs: 30 * 60 * 1000, + constraint: { kind: 'same-sign' }, + severity: 'warn', + }, +]; + +/** Look up the rules that gate a given dependent feed. */ +export function rulesForFeed( + feed: string, + rules: ConsistencyRule[] = DEFAULT_CONSISTENCY_RULES, +): ConsistencyRule[] { + return rules.filter((rule) => rule.dependentFeed === feed); +} diff --git a/src/oracle/validation/schemaDrift.ts b/src/oracle/validation/schemaDrift.ts new file mode 100644 index 0000000..4d4a444 --- /dev/null +++ b/src/oracle/validation/schemaDrift.ts @@ -0,0 +1,310 @@ +/** + * Upstream provider schema drift detection. + * + * Each adapter declares an expected response shape (structural schema plus + * required units and nullability) separately from its parsing code. Every + * fetch validates the payload against that declared shape *before* parsing so + * a tolerant parser can never mask an upstream change. + * + * Unexpected structures are classified as either: + * - `drift` informational: optional fields added, empty arrays, etc. + * - `breaking` ingestion must be rejected, the source quarantined and an + * alert fired. + */ + +export type ProviderId = 'coingecko' | 'horizon' | 'defillama'; + +export type FieldType = 'string' | 'number' | 'boolean' | 'object' | 'array'; + +export interface FieldShape { + type: FieldType; + /** Required fields must be present and non-null. */ + required?: boolean; + /** Field may legitimately be null without being classified as drift. */ + nullable?: boolean; + /** Declared unit for numeric fields (e.g. 'usd', 'stroops', 'ratio'). */ + unit?: string; + /** Nested object shape. */ + fields?: Record; + /** Element shape for arrays. */ + items?: FieldShape; +} + +export interface ResponseShape { + provider: ProviderId; + /** Version of the declared shape; bump when the contract changes. */ + version: string; + fields: Record; +} + +export type DriftClassification = 'ok' | 'drift' | 'breaking'; + +export interface DriftIssue { + path: string; + kind: + | 'missing_required' + | 'null_required' + | 'type_mismatch' + | 'unit_missing' + | 'optional_added' + | 'empty_array'; + detail: string; + severity: 'drift' | 'breaking'; +} + +export interface DriftResult { + provider: ProviderId; + shapeVersion: string; + classification: DriftClassification; + issues: DriftIssue[]; + /** Redacted, size-limited sample of the offending payload. */ + sample?: unknown; + /** True when the payload looks like a rate-limit/quota response. */ + rateLimited: boolean; + checkedAt: string; +} + +const MAX_SAMPLE_BYTES = 4096; +const REDACTED = '[redacted]'; +const SENSITIVE_KEY = /(key|token|secret|auth|signature|password)/i; + +/** + * Rate-limit / quota responses are structurally different by design and must + * not be misclassified as drift. + */ +export function isRateLimitResponse(payload: unknown): boolean { + if (!payload || typeof payload !== 'object') return false; + const obj = payload as Record; + if (obj.status === 429 || obj.statusCode === 429) return true; + if (typeof obj.error === 'string' && /rate.?limit|quota|too many requests/i.test(obj.error)) { + return true; + } + if (typeof obj.message === 'string' && /rate.?limit|quota|too many requests/i.test(obj.message)) { + return true; + } + return false; ++} ++ ++function redact(value: unknown, depth = 0): unknown { ++ if (depth > 6) return REDACTED; ++ if (Array.isArray(value)) { ++ return value.slice(0, 20).map((v) => redact(v, depth + 1)); ++ } ++ if (value && typeof value === 'object') { ++ const out: Record = {}; ++ for (const [k, v] of Object.entries(value as Record)) { ++ out[k] = SENSITIVE_KEY.test(k) ? REDACTED : redact(v, depth + 1); ++ } ++ return out; ++ } ++ return value; ++} ++ ++/** Capture a redacted, size-limited sample of the offending payload. */ ++export function captureSample(payload: unknown): unknown { ++ const redacted = redact(payload); ++ try { ++ const serialized = JSON.stringify(redacted); ++ if (serialized.length <= MAX_SAMPLE_BYTES) return redacted; ++ return { truncated: true, preview: serialized.slice(0, MAX_SAMPLE_BYTES) }; ++ } catch { ++ return { truncated: true, preview: String(payload).slice(0, MAX_SAMPLE_BYTES) }; ++ } ++} ++ ++function typeOf(value: unknown): FieldType | 'null' | 'undefined' { ++ if (value === null) return 'null'; ++ if (value === undefined) return 'undefined'; ++ if (Array.isArray(value)) return 'array'; ++ return typeof value as FieldType; ++} ++ ++function validateField( ++ path: string, ++ shape: FieldShape, ++ value: unknown, ++ issues: DriftIssue[], ++): void { ++ const actual = typeOf(value); ++ ++ if (actual === 'undefined') { ++ if (shape.required) { ++ issues.push({ ++ path, ++ kind: 'missing_required', ++ detail: `required field missing (expected ${shape.type})`, ++ severity: 'breaking', ++ }); ++ } ++ return; ++ } ++ ++ if (actual === 'null') { ++ if (shape.required && !shape.nullable) { ++ issues.push({ ++ path, ++ kind: 'null_required', ++ detail: 'required field is null', ++ severity: 'breaking', ++ }); ++ } ++ return; ++ } ++ ++ if (actual !== shape.type) { ++ issues.push({ ++ path, ++ kind: 'type_mismatch', ++ detail: `expected ${shape.type}, got ${actual}`, ++ severity: 'breaking', ++ }); ++ return; ++ } ++ ++ if (shape.type === 'number' && shape.unit && !Number.isFinite(value as number)) { ++ issues.push({ ++ path, ++ kind: 'unit_missing', ++ detail: `numeric field with unit '${shape.unit}' is not finite`, ++ severity: 'breaking', ++ }); ++ } ++ ++ if (shape.type === 'object' && shape.fields) { ++ const obj = value as Record; ++ for (const [key, childShape] of Object.entries(shape.fields)) { ++ validateField(`${path}.${key}`, childShape, obj[key], issues); ++ } ++ for (const key of Object.keys(obj)) { ++ if (!(key in shape.fields)) { ++ issues.push({ ++ path: `${path}.${key}`, ++ kind: 'optional_added', ++ detail: 'unexpected field added by provider', ++ severity: 'drift', ++ }); ++ } ++ } ++ } ++ ++ if (shape.type === 'array' && shape.items) { ++ const arr = value as unknown[]; ++ if (arr.length === 0) { ++ issues.push({ ++ path, ++ kind: 'empty_array', ++ detail: 'array arrived empty', ++ severity: 'drift', ++ }); ++ return; ++ } ++ arr.forEach((item, i) => validateField(`${path}[${i}]`, shape.items as FieldShape, item, issues)); ++ } ++} ++ ++/** ++ * Validate a provider payload against its declared shape. Must be called ++ * before parsing/normalization. ++ */ ++export function validateResponseShape(shape: ResponseShape, payload: unknown): DriftResult { ++ const checkedAt = new Date().toISOString(); ++ ++ if (isRateLimitResponse(payload)) { ++ return { ++ provider: shape.provider, ++ shapeVersion: shape.version, ++ classification: 'ok', ++ issues: [], ++ rateLimited: true, ++ checkedAt, ++ }; ++ } ++ ++ const issues: DriftIssue[] = []; ++ const actual = typeOf(payload); ++ if (actual !== 'object') { ++ issues.push({ ++ path: '$', ++ kind: 'type_mismatch', ++ detail: `expected object payload, got ${actual}`, ++ severity: 'breaking', ++ }); ++ } else { ++ const obj = payload as Record; ++ for (const [key, fieldShape] of Object.entries(shape.fields)) { ++ validateField(key, fieldShape, obj[key], issues); ++ } ++ for (const key of Object.keys(obj)) { ++ if (!(key in shape.fields)) { ++ issues.push({ ++ path: key, ++ kind: 'optional_added', ++ detail: 'unexpected top-level field added by provider', ++ severity: 'drift', ++ }); ++ } ++ } ++ } ++ ++ const breaking = issues.some((i) => i.severity === 'breaking'); + const classification: DriftClassification = breaking + ? 'breaking' + : issues.length > 0 + ? 'drift' + : 'ok'; ++ ++ return { ++ provider: shape.provider, ++ shapeVersion: shape.version, ++ classification, ++ issues, ++ sample: classification === 'ok' ? undefined : captureSample(payload), ++ rateLimited: false, ++ checkedAt, ++ }; ++} ++ ++/** ++ * Declared response shapes for each provider. Kept separate from parsing code ++ * so the contract can be reviewed and versioned independently. ++ */ ++export const PROVIDER_SHAPES: Record = { ++ coingecko: { ++ provider: 'coingecko', ++ version: '1.0.0', ++ fields: { ++ prices: { ++ type: 'array', ++ required: true, ++ items: { ++ type: 'array', ++ items: { type: 'number', unit: 'usd' }, ++ }, ++ }, ++ }, ++ }, ++ horizon: { ++ provider: 'horizon', ++ version: '1.0.0', ++ fields: { ++ _embedded: { ++ type: 'object', ++ required: true, ++ fields: { ++ records: { type: 'array', required: true, items: { type: 'object' } }, ++ }, ++ }, ++ }, ++ }, ++ defillama: { ++ provider: 'defillama', ++ version: '1.0.0', ++ fields: { ++ coins: { type: 'object', required: true }, ++ }, ++ }, ++}; ++ ++export function getProviderShape(provider: ProviderId): ResponseShape { ++ return PROVIDER_SHAPES[provider]; ++} diff --git a/src/routes/review.ts b/src/routes/review.ts new file mode 100644 index 0000000..2533cbe --- /dev/null +++ b/src/routes/review.ts @@ -0,0 +1,57 @@ +import { Router } from 'express'; +import { + approve, + escalate, + getEntry, + listPending, + queueDepth, + reject, +} from '../services/reviewQueue'; + +const router = Router(); + +function reviewerContext(req: any) { + const actor = req.user?.id ?? req.headers['x-actor']; + const authorized = Boolean(req.user?.canReview ?? req.headers['x-reviewer'] === 'true'); + return { actor: String(actor ?? 'anonymous'), authorized }; +} + +router.get('/queue', async (_req, res) => { + const items = await listPending(); + res.json({ items, depth: await queueDepth() }); +}); + +router.get('/queue/:id', async (req, res) => { + const entry = await getEntry(req.params.id); + if (!entry) return res.status(404).json({ error: 'not found' }); + res.json(entry); +}); + +router.post('/queue/:id/approve', async (req, res) => { + try { + const entry = await approve(req.params.id, reviewerContext(req)); + res.json(entry); + } catch (err) { + res.status(403).json({ error: (err as Error).message }); + } +}); + +router.post('/queue/:id/reject', async (req, res) => { + try { + const entry = await reject(req.params.id, reviewerContext(req), req.body?.reason); + res.json(entry); + } catch (err) { + res.status(400).json({ error: (err as Error).message }); + } +}); + +router.post('/queue/:id/escalate', async (req, res) => { + try { + const entry = await escalate(req.params.id, reviewerContext(req)); + res.json(entry); + } catch (err) { + res.status(403).json({ error: (err as Error).message }); + } +}); + +export default router; diff --git a/src/services/claimProcessor.ts b/src/services/claimProcessor.ts new file mode 100644 index 0000000..b8863f7 --- /dev/null +++ b/src/services/claimProcessor.ts @@ -0,0 +1,55 @@ +import { logger } from '../utils/logger'; +import { + DecisionEvaluation, + assertSettlementAllowed, + determineTriggers, + enqueueForReview, +} from './reviewQueue'; + +export interface ClaimDecision { + claimId: string; + decisionId: string; + value: number; + confidence: number; + evidenceMature: boolean; + postRemediation: boolean; + decision: unknown; + readings: unknown[]; + sourceAttributions: unknown[]; + anomalyFlags: string[]; + explanation: string; +} + +export type ProcessOutcome = + | { status: 'settled'; decisionId: string } + | { status: 'held_for_review'; decisionId: string; reviewEntryId: string }; + +export async function processClaim(claim: ClaimDecision): Promise { + const evaluation: DecisionEvaluation = { + claimId: claim.claimId, + decisionId: claim.decisionId, + value: claim.value, + confidence: claim.confidence, + evidenceMature: claim.evidenceMature, + postRemediation: claim.postRemediation, + decision: claim.decision, + readings: claim.readings, + sourceAttributions: claim.sourceAttributions, + anomalyFlags: claim.anomalyFlags, + explanation: claim.explanation, + }; + + const triggers = determineTriggers(evaluation); + if (triggers.length > 0) { + const entry = await enqueueForReview(evaluation, triggers); + logger.info( + { claimId: claim.claimId, decisionId: claim.decisionId, triggers }, + 'claim decision held for human review', + ); + return { status: 'held_for_review', decisionId: claim.decisionId, reviewEntryId: entry.id }; + } + + await assertSettlementAllowed(claim.decisionId); + logger.info({ claimId: claim.claimId, decisionId: claim.decisionId }, 'claim decision settled'); + return { status: 'settled', decisionId: claim.decisionId }; +} diff --git a/src/services/operatorRewards.ts b/src/services/operatorRewards.ts new file mode 100644 index 0000000..3c192f7 --- /dev/null +++ b/src/services/operatorRewards.ts @@ -0,0 +1,345 @@ +import { createHash } from 'crypto'; + +/** + * Operator reward accrual and distribution for honest oracle participation. + * + * Accruals are derived from the committed evidence ledger (not live submission + * events) so they are recomputable and auditable after the fact. Only readings + * that were included in a committed consensus decision accrue rewards; late, + * quarantined, duplicated, or later-invalidated readings accrue nothing. + * + * Formula (documented): + * base = CONSENSUS_BASE_REWARD + * stakeW = min(stake, STAKE_CAP) / STAKE_CAP (0..1) + * repW = clamp(reputation, 0, 1) (0..1) + * weight = stakeW * repW + * raw = base * weight + * capped = min(raw, POOL_CAP_PER_OPERATOR) (anti-domination) + * accrual = round(capped, DUST_PRECISION) + * + * Operators with zero stake accrue nothing (weight is zero). Rounding dust is + * truncated to DUST_PRECISION and never redistributed, keeping accruals + * deterministic for a fixed set of consensus inputs. + */ + +export const CONSENSUS_BASE_REWARD = 100; +export const STAKE_CAP = 100_000; +export const POOL_CAP_PER_OPERATOR = 250; +export const DUST_PRECISION = 6; +export const MIN_PAYOUT_THRESHOLD = 10; +export const DISPUTE_WINDOW_MS = 24 * 60 * 60 * 1000; + +export type ReadingStatus = + | 'committed' + | 'late' + | 'quarantined' + | 'duplicate' + | 'invalidated'; + +export interface ConsensusReading { + /** Stable id of the reading in the evidence ledger. */ + readingId: string; + /** Round the reading was submitted for. */ + roundId: string; + operatorId: string; + /** Whether the reading was included in the committed consensus decision. */ + includedInConsensus: boolean; + /** Lifecycle status of the reading. */ + status: ReadingStatus; + /** Operator stake at accrual time. */ + stake: number; + /** Operator reputation score in [0, 1]. */ + reputation: number; +} + +export interface RewardAccrual { + /** Idempotency key: unique per (roundId, readingId, operatorId). */ + idempotencyKey: string; + roundId: string; + readingId: string; + operatorId: string; + amount: number; + createdAt: number; +} + +export type LedgerEntryType = 'accrual' | 'adjustment' | 'payout'; + +export interface LedgerEntry { + idempotencyKey: string; + type: LedgerEntryType; + operatorId: string; + amount: number; + roundId?: string; + readingId?: string; + createdAt: number; +} + +export interface PayoutRecord { + idempotencyKey: string; + operatorId: string; + amount: number; + scheduledAt: number; + disputeWindowEndsAt: number; + status: 'pending' | 'paid' | 'disputed'; +} + +export interface OperatorRewardView { + operatorId: string; + accrued: number; + pending: number; + paid: number; +} + +export interface RewardLedgerStore { + hasEntry(idempotencyKey: string): boolean; + appendEntry(entry: LedgerEntry): void; + listEntries(operatorId?: string): LedgerEntry[]; + listPayouts(operatorId?: string): PayoutRecord[]; + upsertPayout(payout: PayoutRecord): void; +} + +export class InMemoryRewardLedgerStore implements RewardLedgerStore { + private entries = new Map(); + private payouts = new Map(); + + hasEntry(idempotencyKey: string): boolean { + return this.entries.has(idempotencyKey); + } + + appendEntry(entry: LedgerEntry): void { + if (this.entries.has(entry.idempotencyKey)) return; + this.entries.set(entry.idempotencyKey, entry); + } + + listEntries(operatorId?: string): LedgerEntry[] { + const all = Array.from(this.entries.values()); + return operatorId ? all.filter((e) => e.operatorId === operatorId) : all; + } + + listPayouts(operatorId?: string): PayoutRecord[] { + const all = Array.from(this.payouts.values()); + return operatorId ? all.filter((p) => p.operatorId === operatorId) : all; + } + + upsertPayout(payout: PayoutRecord): void { + this.payouts.set(payout.idempotencyKey, payout); + } +} + +function roundDust(value: number): number { + const factor = Math.pow(10, DUST_PRECISION); + return Math.floor(value * factor) / factor; +} + +function clamp01(value: number): number { + if (!Number.isFinite(value)) return 0; + if (value < 0) return 0; + if (value > 1) return 1; + return value; +} + +/** + * Compute the reward weight for an operator from stake and reputation. + * Zero stake yields zero weight, so zero-stake operators accrue nothing. + */ +export function computeRewardWeight(stake: number, reputation: number): number { + if (!Number.isFinite(stake) || stake <= 0) return 0; + const stakeW = Math.min(stake, STAKE_CAP) / STAKE_CAP; + const repW = clamp01(reputation); + return stakeW * repW; +} + +/** + * Compute the accrual amount for a single reading, applying the per-operator + * pool cap and dust rounding. Returns 0 for readings that must not accrue. + */ +export function computeAccrualAmount(reading: ConsensusReading): number { + if (!reading.includedInConsensus) return 0; + if (reading.status !== 'committed') return 0; + const weight = computeRewardWeight(reading.stake, reading.reputation); + if (weight <= 0) return 0; + const raw = CONSENSUS_BASE_REWARD * weight; + const capped = Math.min(raw, POOL_CAP_PER_OPERATOR); + return roundDust(capped); +} + +/** + * Deterministic idempotency key for an accrual. Stable across retries and + * re-orgs because it is derived from the committed ledger coordinates. + */ +export function accrualIdempotencyKey( + roundId: string, + readingId: string, + operatorId: string, +): string { + return createHash('sha256') + .update(`accrual:${roundId}:${readingId}:${operatorId}`) + .digest('hex'); +} + +/** + * Accrue rewards for a set of committed consensus readings. Exactly-once under + * retries: duplicate deliveries are skipped via the idempotency key. + */ +export function accrueRewards( + store: RewardLedgerStore, + readings: ConsensusReading[], + now: number = Date.now(), +): RewardAccrual[] { + const accrued: RewardAccrual[] = []; + for (const reading of readings) { + const amount = computeAccrualAmount(reading); + if (amount <= 0) continue; + const idempotencyKey = accrualIdempotencyKey( + reading.roundId, + reading.readingId, + reading.operatorId, + ); + if (store.hasEntry(idempotencyKey)) continue; + const entry: LedgerEntry = { + idempotencyKey, + type: 'accrual', + operatorId: reading.operatorId, + amount, + roundId: reading.roundId, + readingId: reading.readingId, + createdAt: now, + }; + store.appendEntry(entry); + accrued.push({ + idempotencyKey, + roundId: reading.roundId, + readingId: reading.readingId, + operatorId: reading.operatorId, + amount, + createdAt: now, + }); + } + return accrued; +} + +/** + * Reverse an accrual when a reading is later invalidated by remediation. The + * adjustment is idempotent and only applied if the original accrual exists. + */ +export function invalidateAccrual( + store: RewardLedgerStore, + roundId: string, + readingId: string, + operatorId: string, + now: number = Date.now(), +): LedgerEntry | null { + const accrualKey = accrualIdempotencyKey(roundId, readingId, operatorId); + if (!store.hasEntry(accrualKey)) return null; + const adjustmentKey = createHash('sha256') + .update(`adjustment:${accrualKey}`) + .digest('hex'); + if (store.hasEntry(adjustmentKey)) return null; + const original = store + .listEntries(operatorId) + .find((e) => e.idempotencyKey === accrualKey); + if (!original) return null; + const entry: LedgerEntry = { + idempotencyKey: adjustmentKey, + type: 'adjustment', + operatorId, + amount: -original.amount, + roundId, + readingId, + createdAt: now, + }; + store.appendEntry(entry); + return entry; +} + +/** + * Schedule payouts for operators whose accrued balance meets the minimum + * threshold. Funds move only after the dispute window elapses. + */ +export function schedulePayouts( + store: RewardLedgerStore, + now: number = Date.now(), +): PayoutRecord[] { + const balances = new Map(); + for (const entry of store.listEntries()) { + if (entry.type === 'payout') continue; + balances.set(entry.operatorId, (balances.get(entry.operatorId) ?? 0) + entry.amount); + } + + const scheduled: PayoutRecord[] = []; + for (const [operatorId, balance] of balances) { + if (balance < MIN_PAYOUT_THRESHOLD) continue; + const idempotencyKey = createHash('sha256') + .update(`payout:${operatorId}:${roundDust(balance)}`) + .digest('hex'); + const existing = store + .listPayouts(operatorId) + .find((p) => p.idempotencyKey === idempotencyKey); + if (existing) continue; + const payout: PayoutRecord = { + idempotencyKey, + operatorId, + amount: roundDust(balance), + scheduledAt: now, + disputeWindowEndsAt: now + DISPUTE_WINDOW_MS, + status: 'pending', + }; + store.upsertPayout(payout); + scheduled.push(payout); + } + return scheduled; +} + +/** + * Settle pending payouts whose dispute window has elapsed. Disputed payouts are + * left untouched. + */ +export function settlePayouts( + store: RewardLedgerStore, + now: number = Date.now(), +): PayoutRecord[] { + const settled: PayoutRecord[] = []; + for (const payout of store.listPayouts()) { + if (payout.status !== 'pending') continue; + if (now < payout.disputeWindowEndsAt) continue; + const entry: LedgerEntry = { + idempotencyKey: payout.idempotencyKey, + type: 'payout', + operatorId: payout.operatorId, + amount: -payout.amount, + createdAt: now, + }; + if (!store.hasEntry(entry.idempotencyKey)) { + store.appendEntry(entry); + } + const paid: PayoutRecord = { ...payout, status: 'paid' }; + store.upsertPayout(paid); + settled.push(paid); + } + return settled; +} + +/** + * Operator-facing view of accrued, pending, and paid rewards. + */ +export function getOperatorRewardView( + store: RewardLedgerStore, + operatorId: string, +): OperatorRewardView { + let accrued = 0; + let paid = 0; + for (const entry of store.listEntries(operatorId)) { + if (entry.type === 'accrual' || entry.type === 'adjustment') { + accrued += entry.amount; + } else if (entry.type === 'payout') { + paid += -entry.amount; + } + } + const pending = Math.max(0, roundDust(accrued - paid)); + return { + operatorId, + accrued: roundDust(accrued), + pending, + paid: roundDust(paid), + }; +} diff --git a/src/services/oracleMonitor.ts b/src/services/oracleMonitor.ts new file mode 100644 index 0000000..066979b --- /dev/null +++ b/src/services/oracleMonitor.ts @@ -0,0 +1,204 @@ +import { EventEmitter } from 'events'; +import { logger } from '../utils/logger'; +import { db } from '../db/client'; + +/** + * Classification of a provider payload against its declared response shape. + * - `ok`: payload matches the declared shape. + * - `drift`: structurally unexpected but non-fatal (e.g. optional field added). + * - `breaking`: required structure/unit/nullability violated; ingestion must be rejected. + */ +export type SchemaDriftClassification = 'ok' | 'drift' | 'breaking'; + +export interface SchemaDriftResult { + source: string; + classification: SchemaDriftClassification; + /** Human-readable reasons for a non-ok classification. */ + reasons: string[]; + /** Redacted, size-limited sample of the offending payload. */ + sample?: unknown; + /** Last known-good payload sample for this source, if any. */ + lastKnownGoodSample?: unknown; + detectedAt: string; +} + +export interface SourceDriftReport { + source: string; + quarantined: boolean; + lastClassification: SchemaDriftClassification; + lastDetectedAt: string | null; + lastKnownGoodSample: unknown | null; + recentEvents: SchemaDriftResult[]; +} + +const MAX_SAMPLE_BYTES = 4096; +const MAX_RECENT_EVENTS = 25; + +/** + * Redact and truncate a payload sample so it can be safely stored/queried. + * Keeps structure for debugging while bounding size and stripping obvious secrets. + */ +export function redactSample(payload: unknown): unknown { + const seen = new WeakSet(); + const redact = (value: unknown, depth: number): unknown => { + if (depth > 6) return '[truncated:depth]'; + if (value === null || typeof value !== 'object') return value; + if (seen.has(value as object)) return '[circular]'; + seen.add(value as object); + if (Array.isArray(value)) { + return value.slice(0, 20).map((item) => redact(item, depth + 1)); + } + const out: Record = {}; + for (const [key, val] of Object.entries(value as Record)) { + if (/key|token|secret|auth|password/i.test(key)) { + out[key] = '[redacted]'; + } else { + out[key] = redact(val, depth + 1); + } + } + return out; + }; + + const redacted = redact(payload, 0); + try { + const serialized = JSON.stringify(redacted); + if (serialized && serialized.length > MAX_SAMPLE_BYTES) { + return { truncated: true, preview: serialized.slice(0, MAX_SAMPLE_BYTES) }; + } + } catch { + return '[unserializable]'; + } + return redacted; +} + +interface SourceState { + quarantined: boolean; + lastClassification: SchemaDriftClassification; + lastDetectedAt: string | null; + lastKnownGoodSample: unknown | null; + recentEvents: SchemaDriftResult[]; +} + +export class OracleMonitor extends EventEmitter { + private readonly sources = new Map(); + + private stateFor(source: string): SourceState { + let state = this.sources.get(source); + if (!state) { + state = { + quarantined: false, + lastClassification: 'ok', + lastDetectedAt: null, + lastKnownGoodSample: null, + recentEvents: [], + }; + this.sources.set(source, state); + } + return state; + } + + /** + * Record the outcome of validating a provider payload against its declared shape. + * Breaking changes quarantine the source and fire an alert; drift only warns. + * Returns the classification so callers can reject ingestion before parsing. + */ + recordSchemaValidation( + source: string, + classification: SchemaDriftClassification, + reasons: string[] = [], + payload?: unknown, + ): SchemaDriftResult { + const state = this.stateFor(source); + const detectedAt = new Date().toISOString(); + + const result: SchemaDriftResult = { + source, + classification, + reasons, + detectedAt, + lastKnownGoodSample: state.lastKnownGoodSample ?? undefined, + }; + + if (classification === 'ok') { + state.lastKnownGoodSample = redactSample(payload); + state.lastClassification = 'ok'; + state.lastDetectedAt = detectedAt; + return result; + } + + result.sample = redactSample(payload); + state.lastClassification = classification; + state.lastDetectedAt = detectedAt; + state.recentEvents.unshift(result); + if (state.recentEvents.length > MAX_RECENT_EVENTS) { + state.recentEvents.length = MAX_RECENT_EVENTS; + } + + if (classification === 'breaking') { + state.quarantined = true; + logger.error( + `[oracleMonitor] Breaking schema change for source ${source}; source quarantined`, + { reasons }, + ); + this.emit('breaking', result); + this.persistDriftEvent(result).catch((err) => + logger.error('[oracleMonitor] Failed to persist drift event', err), + ); + } else { + logger.warn(`[oracleMonitor] Schema drift detected for source ${source}`, { reasons }); + this.emit('drift', result); + this.persistDriftEvent(result).catch((err) => + logger.error('[oracleMonitor] Failed to persist drift event', err), + ); + } + + return result; + } + + /** Whether a source is currently quarantined due to a breaking schema change. */ + isQuarantined(source: string): boolean { + return this.stateFor(source).quarantined; + } + + /** Clear quarantine for a source (e.g. after an operator confirms the shape was updated). */ + clearQuarantine(source: string): void { + const state = this.stateFor(source); + state.quarantined = false; + logger.info(`[oracleMonitor] Quarantine cleared for source ${source}`); + } + + /** Per-source drift report including the last known-good payload sample. */ + getDriftReport(source: string): SourceDriftReport { + const state = this.stateFor(source); + return { + source, + quarantined: state.quarantined, + lastClassification: state.lastClassification, + lastDetectedAt: state.lastDetectedAt, + lastKnownGoodSample: state.lastKnownGoodSample, + recentEvents: [...state.recentEvents], + }; + } + + /** Drift reports for all sources seen so far. */ + getAllDriftReports(): SourceDriftReport[] { + return Array.from(this.sources.keys()).map((source) => this.getDriftReport(source)); + } + + private async persistDriftEvent(result: SchemaDriftResult): Promise { + await db.query( + `INSERT INTO oracle_schema_drift_events + (source, classification, reasons, sample, detected_at) + VALUES ($1, $2, $3, $4, $5)`, + [ + result.source, + result.classification, + JSON.stringify(result.reasons), + result.sample === undefined ? null : JSON.stringify(result.sample), + result.detectedAt, + ], + ); + } +} + +export const oracleMonitor = new OracleMonitor(); diff --git a/src/services/reviewQueue.ts b/src/services/reviewQueue.ts new file mode 100644 index 0000000..28b5a69 --- /dev/null +++ b/src/services/reviewQueue.ts @@ -0,0 +1,271 @@ +import { randomUUID } from 'crypto'; +import { db } from '../db/client'; +import { logger } from '../utils/logger'; + +export type ReviewTrigger = + | 'value_above_threshold' + | 'confidence_below_floor' + | 'evidence_immaturity' + | 'post_remediation'; + +export type ReviewStatus = 'pending' | 'approved' | 'rejected' | 'escalated'; + +export type ReviewerAction = 'approve' | 'reject' | 'escalate'; + +export interface ReviewEvidencePacket { + decision: unknown; + readings: unknown[]; + sourceAttributions: unknown[]; + anomalyFlags: string[]; + explanation: string; +} + +export interface ReviewQueueConfig { + valueThreshold: number; + confidenceFloor: number; + slaSeconds: number; +} + +export interface ReviewQueueEntry { + id: string; + claimId: string; + decisionId: string; + trigger: ReviewTrigger; + status: ReviewStatus; + evidence: ReviewEvidencePacket; + createdAt: string; + deadlineAt: string; + resolvedAt: string | null; + resolvedBy: string | null; + rejectionReason: string | null; +} + +export interface ReviewerContext { + actor: string; + authorized: boolean; +} + +const DEFAULT_CONFIG: ReviewQueueConfig = { + valueThreshold: 10000, + confidenceFloor: 0.6, + slaSeconds: 24 * 60 * 60, +}; + +let config: ReviewQueueConfig = { ...DEFAULT_CONFIG }; + +export function configureReviewQueue(overrides: Partial): void { + config = { ...config, ...overrides }; +} + +export function getReviewQueueConfig(): ReviewQueueConfig { + return { ...config }; +} + +export interface DecisionEvaluation { + claimId: string; + decisionId: string; + value: number; + confidence: number; + evidenceMature: boolean; + postRemediation: boolean; + decision: unknown; + readings: unknown[]; + sourceAttributions: unknown[]; + anomalyFlags: string[]; + explanation: string; +} + +export function determineTriggers(evaluation: DecisionEvaluation): ReviewTrigger[] { + const triggers: ReviewTrigger[] = []; + if (evaluation.value > config.valueThreshold) triggers.push('value_above_threshold'); + if (evaluation.confidence < config.confidenceFloor) triggers.push('confidence_below_floor'); + if (!evaluation.evidenceMature) triggers.push('evidence_immaturity'); + if (evaluation.postRemediation) triggers.push('post_remediation'); + return triggers; +} + +export async function enqueueForReview( + evaluation: DecisionEvaluation, + triggers: ReviewTrigger[], +): Promise { + const now = new Date(); + const deadline = new Date(now.getTime() + config.slaSeconds * 1000); + const entry: ReviewQueueEntry = { + id: randomUUID(), + claimId: evaluation.claimId, + decisionId: evaluation.decisionId, + trigger: triggers[0], + status: 'pending', + evidence: { + decision: evaluation.decision, + readings: evaluation.readings, + sourceAttributions: evaluation.sourceAttributions, + anomalyFlags: evaluation.anomalyFlags, + explanation: evaluation.explanation, + }, + createdAt: now.toISOString(), + deadlineAt: deadline.toISOString(), + resolvedAt: null, + resolvedBy: null, + rejectionReason: null, + }; + + await db.query( + `INSERT INTO review_queue + (id, claim_id, decision_id, trigger, status, evidence, created_at, deadline_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + [ + entry.id, + entry.claimId, + entry.decisionId, + entry.trigger, + entry.status, + JSON.stringify(entry.evidence), + entry.createdAt, + entry.deadlineAt, + ], + ); + + await recordAudit(entry.id, 'enqueue', 'system', { triggers }); + logger.info({ entryId: entry.id, claimId: entry.claimId, triggers }, 'review queue entry created'); + return entry; +} + +export async function getEntry(id: string): Promise { + const result = await db.query('SELECT * FROM review_queue WHERE id = $1', [id]); + const row = result.rows[0]; + if (!row) return null; + return mapRow(row); +} + +export async function listPending(): Promise { + const result = await db.query( + "SELECT * FROM review_queue WHERE status IN ('pending', 'escalated') ORDER BY deadline_at ASC", + ); + return result.rows.map(mapRow); +} + +export async function queueDepth(): Promise { + const result = await db.query( + "SELECT COUNT(*)::int AS depth FROM review_queue WHERE status IN ('pending', 'escalated')", + ); + return result.rows[0]?.depth ?? 0; +} + +export async function approve(id: string, ctx: ReviewerContext): Promise { + assertAuthorized(ctx); + return resolve(id, 'approved', ctx.actor, null); +} + +export async function reject( + id: string, + ctx: ReviewerContext, + reason: string, +): Promise { + assertAuthorized(ctx); + if (!reason || !reason.trim()) { + throw new Error('rejection reason is required'); + } + return resolve(id, 'rejected', ctx.actor, reason.trim()); +} + +export async function escalate(id: string, ctx: ReviewerContext): Promise { + assertAuthorized(ctx); + return resolve(id, 'escalated', ctx.actor, null); +} + +async function resolve( + id: string, + status: ReviewStatus, + actor: string, + reason: string | null, +): Promise { + const entry = await getEntry(id); + if (!entry) throw new Error(`review queue entry ${id} not found`); + if (entry.status === 'approved' || entry.status === 'rejected') { + throw new Error(`review queue entry ${id} already resolved as ${entry.status}`); + } + + const resolvedAt = new Date().toISOString(); + await db.query( + `UPDATE review_queue + SET status = $1, resolved_at = $2, resolved_by = $3, rejection_reason = $4 + WHERE id = $5`, + [status, resolvedAt, actor, reason, id], + ); + + await recordAudit(id, status, actor, { reason }); + logger.info({ entryId: id, status, actor }, 'review queue entry resolved'); + + return { ...entry, status, resolvedAt, resolvedBy: actor, rejectionReason: reason }; +} + +/** + * Escalate any pending items whose deadline has passed. Items are never + * auto-approved by timeout — expiry only escalates. + */ +export async function sweepExpired(now: Date = new Date()): Promise { + const result = await db.query( + "SELECT id FROM review_queue WHERE status = 'pending' AND deadline_at <= $1", + [now.toISOString()], + ); + for (const row of result.rows) { + await escalate(row.id, { actor: 'system:sla', authorized: true }); + } + return result.rows.length; +} + +/** + * Settlement gate: a decision may only settle when it has an approved queue + * entry, or when it was never queued at all. + */ +export async function assertSettlementAllowed(decisionId: string): Promise { + const result = await db.query( + 'SELECT status FROM review_queue WHERE decision_id = $1 ORDER BY created_at DESC LIMIT 1', + [decisionId], + ); + const row = result.rows[0]; + if (!row) return; + if (row.status !== 'approved') { + throw new Error( + `decision ${decisionId} cannot settle: review queue status is ${row.status}`, + ); + } +} + +async function recordAudit( + entryId: string, + action: string, + actor: string, + detail: Record, +): Promise { + await db.query( + `INSERT INTO review_audit (id, entry_id, action, actor, detail, created_at) + VALUES ($1, $2, $3, $4, $5, $6)`, + [randomUUID(), entryId, action, actor, JSON.stringify(detail), new Date().toISOString()], + ); +} + +function assertAuthorized(ctx: ReviewerContext): void { + if (!ctx.authorized) { + throw new Error(`actor ${ctx.actor} is not authorized to act on review queue items`); + } +} + +function mapRow(row: Record): ReviewQueueEntry { + const evidence = + typeof row.evidence === 'string' ? JSON.parse(row.evidence) : (row.evidence as ReviewEvidencePacket); + return { + id: row.id as string, + claimId: row.claim_id as string, + decisionId: row.decision_id as string, + trigger: row.trigger as ReviewTrigger, + status: row.status as ReviewStatus, + evidence, + createdAt: row.created_at as string, + deadlineAt: row.deadline_at as string, + resolvedAt: (row.resolved_at as string) ?? null, + resolvedBy: (row.resolved_by as string) ?? null, + rejectionReason: (row.rejection_reason as string) ?? null, + }; +}