From 4246409c80a510dedaf4603a35760de7b313f847 Mon Sep 17 00:00:00 2001 From: oscar24357 Date: Thu, 22 Oct 2026 18:21:45 +0100 Subject: [PATCH 1/3] feat: implement provider key revocation, rotation, and forensic reporting - Add oracle_providers, provider_keys, provider_revocations, provider_key_rotations tables (migration 004) - src/oracle/providers/credentials.ts: key types, generateRawKey, hashKey, verifyKey - src/oracle/providers/credential.repository.ts: all DB access for provider lifecycle - src/oracle/providers/provider-credential.service.ts: revocation, rotation with overlap window, forensic report, trust-status surface - src/oracle/providers/provider-key.guard.ts: submission-boundary guard rejecting revoked keys - src/oracle/providers/ops-provider.controller.ts: admin endpoints for revoke, reverse-revocation, rotate, complete-rotation, trust-status - src/oracle/providers/provider-credentials.module.ts: NestJS module wiring - src/oracle/providers/provider-credentials.spec.ts: tests for revoked-key rejection, overlap-window rotation, forensic report, accidental-revocation reversal Closes acceptance criteria: revocation record with reason/actor/timestamp/sequence, submission boundary rejection, affected-consensus report, overlap-window rotation, trust-status surface, reversible revocations. --- .../migrations/004_provider_credentials.sql | 73 ++++ src/oracle/providers/credential.repository.ts | 307 +++++++++++++++ src/oracle/providers/credentials.ts | 89 +++++ .../providers/ops-provider.controller.ts | 183 +++++++++ .../providers/provider-credential.service.ts | 367 ++++++++++++++++++ .../providers/provider-credentials.module.ts | 17 + .../providers/provider-credentials.spec.ts | 356 +++++++++++++++++ src/oracle/providers/provider-key.guard.ts | 61 +++ 8 files changed, 1453 insertions(+) create mode 100644 src/db/migrations/004_provider_credentials.sql create mode 100644 src/oracle/providers/credential.repository.ts create mode 100644 src/oracle/providers/credentials.ts create mode 100644 src/oracle/providers/ops-provider.controller.ts create mode 100644 src/oracle/providers/provider-credential.service.ts create mode 100644 src/oracle/providers/provider-credentials.module.ts create mode 100644 src/oracle/providers/provider-credentials.spec.ts create mode 100644 src/oracle/providers/provider-key.guard.ts diff --git a/src/db/migrations/004_provider_credentials.sql b/src/db/migrations/004_provider_credentials.sql new file mode 100644 index 0000000..d75eb20 --- /dev/null +++ b/src/db/migrations/004_provider_credentials.sql @@ -0,0 +1,73 @@ +-- Migration 004: Provider identity, versioned keys, revocation, and rotation tables +-- Implements first-class revocation path for oracle provider credentials. +-- All revocations are append-only; deletions are never performed. + +-- ── oracle_providers ───────────────────────────────────────────────────────── +-- Stable identity record. A provider survives key rotation and revocation. +CREATE TABLE IF NOT EXISTS oracle_providers ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + name VARCHAR(80) NOT NULL UNIQUE, + contact VARCHAR(255) NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- ── provider_keys ──────────────────────────────────────────────────────────── +-- Versioned child records. Revocation is key-level; the provider survives. +-- key_hash stores SHA-256(rawKey) — plaintext never persisted. +CREATE TABLE IF NOT EXISTS provider_keys ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + provider_id UUID NOT NULL REFERENCES oracle_providers(id), + key_hash CHAR(64) NOT NULL UNIQUE, -- hex SHA-256 + version INTEGER NOT NULL, + status VARCHAR(12) NOT NULL DEFAULT 'active' + CHECK (status IN ('active','rotating','revoked','superseded')), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + rotating_at TIMESTAMPTZ, + revoked_at TIMESTAMPTZ, + -- Ledger sequence from which this key is no longer accepted (set on revocation). + effective_until_sequence BIGINT, + UNIQUE (provider_id, version) +); + +CREATE INDEX IF NOT EXISTS idx_provider_keys_provider_status + ON provider_keys (provider_id, status); + +-- ── provider_revocations ───────────────────────────────────────────────────── +-- Append-only audit log. A superseding record reverses an accidental revocation; +-- the original row is preserved for evidence retention. +CREATE TABLE IF NOT EXISTS provider_revocations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + key_id UUID NOT NULL REFERENCES provider_keys(id), + provider_id UUID NOT NULL REFERENCES oracle_providers(id), + reason TEXT NOT NULL, + actor VARCHAR(80) NOT NULL, + -- Ledger sequence from which submissions signed by this key are rejected. + effective_from_sequence BIGINT NOT NULL, + revoked_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- Points to the revocation this record supersedes (for accidental-revocation reversal). + supersedes_revocation_id UUID REFERENCES provider_revocations(id) +); + +CREATE INDEX IF NOT EXISTS idx_provider_revocations_key + ON provider_revocations (key_id, revoked_at DESC); +CREATE INDEX IF NOT EXISTS idx_provider_revocations_provider + ON provider_revocations (provider_id, revoked_at DESC); + +-- ── provider_key_rotations ─────────────────────────────────────────────────── +-- Tracks active key rotations. Both old and new keys are accepted during +-- the overlap window. completed_at is NULL while the rotation is in-progress. +CREATE TABLE IF NOT EXISTS provider_key_rotations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + provider_id UUID NOT NULL REFERENCES oracle_providers(id), + old_key_id UUID NOT NULL REFERENCES provider_keys(id), + new_key_id UUID NOT NULL REFERENCES provider_keys(id), + overlap_window_ends_at TIMESTAMPTZ NOT NULL, + actor VARCHAR(80) NOT NULL, + initiated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_provider_rotations_provider + ON provider_key_rotations (provider_id, initiated_at DESC); diff --git a/src/oracle/providers/credential.repository.ts b/src/oracle/providers/credential.repository.ts new file mode 100644 index 0000000..8cbad9f --- /dev/null +++ b/src/oracle/providers/credential.repository.ts @@ -0,0 +1,307 @@ +import { Inject, Injectable, Logger } from "@nestjs/common"; +import { Pool } from "pg"; +import { DATABASE_POOL } from "../../database/database.module"; +import { + ProviderIdentity, + ProviderKey, + RevocationRecord, + KeyRotationRecord, + KeyStatus, +} from "./credentials"; + +/** + * Data-access layer for provider identities, versioned keys, revocation + * records, and rotation records. + * + * All revocations are append-only — nothing is ever deleted. An accidental + * revocation is reversed by inserting a superseding revocation record and + * setting the key status back to "active". + */ +@Injectable() +export class CredentialRepository { + private readonly logger = new Logger(CredentialRepository.name); + + constructor(@Inject(DATABASE_POOL) private readonly db: Pool) {} + + // ── Provider identities ────────────────────────────────────────────────── + + async createProvider(name: string, contact: string): Promise { + const { rows } = await this.db.query( + `INSERT INTO oracle_providers (name, contact) + VALUES ($1, $2) + RETURNING id, name, contact, created_at AS "createdAt", + updated_at AS "updatedAt", is_active AS "isActive"`, + [name, contact], + ); + return rows[0]; + } + + async getProvider(providerId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, name, contact, + created_at AS "createdAt", updated_at AS "updatedAt", + is_active AS "isActive" + FROM oracle_providers WHERE id = $1`, + [providerId], + ); + return rows[0] ?? null; + } + + async listProviders(): Promise { + const { rows } = await this.db.query( + `SELECT id, name, contact, + created_at AS "createdAt", updated_at AS "updatedAt", + is_active AS "isActive" + FROM oracle_providers + ORDER BY created_at DESC`, + ); + return rows; + } + + // ── Provider keys ──────────────────────────────────────────────────────── + + async createKey( + providerId: string, + keyHash: string, + version: number, + ): Promise { + const { rows } = await this.db.query( + `INSERT INTO provider_keys (provider_id, key_hash, version, status) + VALUES ($1, $2, $3, 'active') + RETURNING id, provider_id AS "providerId", key_hash AS "keyHash", + version, status, created_at AS "createdAt", + rotating_at AS "rotatingAt", revoked_at AS "revokedAt", + effective_until_sequence AS "effectiveUntilSequence"`, + [providerId, keyHash, version], + ); + return rows[0]; + } + + async getKey(keyId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, provider_id AS "providerId", key_hash AS "keyHash", + version, status, created_at AS "createdAt", + rotating_at AS "rotatingAt", revoked_at AS "revokedAt", + effective_until_sequence AS "effectiveUntilSequence" + FROM provider_keys WHERE id = $1`, + [keyId], + ); + return rows[0] ?? null; + } + + async listKeysByProvider(providerId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, provider_id AS "providerId", key_hash AS "keyHash", + version, status, created_at AS "createdAt", + rotating_at AS "rotatingAt", revoked_at AS "revokedAt", + effective_until_sequence AS "effectiveUntilSequence" + FROM provider_keys + WHERE provider_id = $1 + ORDER BY version DESC`, + [providerId], + ); + return rows; + } + + /** + * Returns all active (or rotating) keys for a provider so the submission + * boundary can check any accepted key hash. + */ + async getAcceptableKeys(providerId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, provider_id AS "providerId", key_hash AS "keyHash", + version, status, created_at AS "createdAt", + rotating_at AS "rotatingAt", revoked_at AS "revokedAt", + effective_until_sequence AS "effectiveUntilSequence" + FROM provider_keys + WHERE provider_id = $1 AND status IN ('active', 'rotating')`, + [providerId], + ); + return rows; + } + + async updateKeyStatus( + keyId: string, + status: KeyStatus, + extra: { + rotatingAt?: Date; + revokedAt?: Date; + effectiveUntilSequence?: number; + } = {}, + ): Promise { + await this.db.query( + `UPDATE provider_keys + SET status = $2, + rotating_at = COALESCE($3, rotating_at), + revoked_at = COALESCE($4, revoked_at), + effective_until_sequence = COALESCE($5, effective_until_sequence) + WHERE id = $1`, + [ + keyId, + status, + extra.rotatingAt ?? null, + extra.revokedAt ?? null, + extra.effectiveUntilSequence ?? null, + ], + ); + } + + // ── Revocation records ─────────────────────────────────────────────────── + + async recordRevocation(params: { + keyId: string; + providerId: string; + reason: string; + actor: string; + effectiveFromSequence: number; + supersedesRevocationId?: string; + }): Promise { + const { rows } = await this.db.query( + `INSERT INTO provider_revocations + (key_id, provider_id, reason, actor, effective_from_sequence, + supersedes_revocation_id) + VALUES ($1, $2, $3, $4, $5, $6) + RETURNING id, key_id AS "keyId", provider_id AS "providerId", + reason, actor, effective_from_sequence AS "effectiveFromSequence", + revoked_at AS "revokedAt", + supersedes_revocation_id AS "supersedesRevocationId"`, + [ + params.keyId, + params.providerId, + params.reason, + params.actor, + params.effectiveFromSequence, + params.supersedesRevocationId ?? null, + ], + ); + return rows[0]; + } + + async getRevocationsForKey(keyId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, key_id AS "keyId", provider_id AS "providerId", + reason, actor, effective_from_sequence AS "effectiveFromSequence", + revoked_at AS "revokedAt", + supersedes_revocation_id AS "supersedesRevocationId" + FROM provider_revocations + WHERE key_id = $1 + ORDER BY revoked_at DESC`, + [keyId], + ); + return rows; + } + + async getRevocationsForProvider(providerId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, key_id AS "keyId", provider_id AS "providerId", + reason, actor, effective_from_sequence AS "effectiveFromSequence", + revoked_at AS "revokedAt", + supersedes_revocation_id AS "supersedesRevocationId" + FROM provider_revocations + WHERE provider_id = $1 + ORDER BY revoked_at DESC`, + [providerId], + ); + return rows; + } + + // ── Rotation records ───────────────────────────────────────────────────── + + async recordRotation(params: { + providerId: string; + oldKeyId: string; + newKeyId: string; + overlapWindowEndsAt: Date; + actor: string; + }): Promise { + const { rows } = await this.db.query( + `INSERT INTO provider_key_rotations + (provider_id, old_key_id, new_key_id, overlap_window_ends_at, actor) + VALUES ($1, $2, $3, $4, $5) + RETURNING id, provider_id AS "providerId", + old_key_id AS "oldKeyId", new_key_id AS "newKeyId", + overlap_window_ends_at AS "overlapWindowEndsAt", + actor, initiated_at AS "initiatedAt", + completed_at AS "completedAt"`, + [ + params.providerId, + params.oldKeyId, + params.newKeyId, + params.overlapWindowEndsAt, + params.actor, + ], + ); + return rows[0]; + } + + async completeRotation(rotationId: string): Promise { + await this.db.query( + `UPDATE provider_key_rotations SET completed_at = now() WHERE id = $1`, + [rotationId], + ); + } + + async getActiveRotation(providerId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, provider_id AS "providerId", + old_key_id AS "oldKeyId", new_key_id AS "newKeyId", + overlap_window_ends_at AS "overlapWindowEndsAt", + actor, initiated_at AS "initiatedAt", + completed_at AS "completedAt" + FROM provider_key_rotations + WHERE provider_id = $1 AND completed_at IS NULL + ORDER BY initiated_at DESC LIMIT 1`, + [providerId], + ); + return rows[0] ?? null; + } + + // ── Forensic queries ───────────────────────────────────────────────────── + + /** + * Finds oracle_events contributed by a revoked key within the lookback + * window. Used to build the affected-consensus report. + */ + async findAffectedEvents(params: { + providerId: string; + effectiveFromSequence: number; + lookbackSeconds: number; + }): Promise> { + const { rows } = await this.db.query( + `SELECT oe.id, oe.coverage_type AS "coverageType", oe.value, + oe.source, oe.recorded_at AS "recordedAt" + FROM oracle_events oe + WHERE oe.source IN ( + SELECT name FROM oracle_providers WHERE id = $1 + ) + AND oe.recorded_at >= now() - ($2 * INTERVAL '1 second') + ORDER BY oe.recorded_at DESC`, + [params.providerId, params.lookbackSeconds], + ); + return rows; + } + + /** + * Finds claims whose trigger_source matches the provider, within the + * lookback window. These are decisions now considered suspect. + */ + async findAffectedClaims(params: { + providerId: string; + lookbackSeconds: number; + }): Promise> { + const { rows } = await this.db.query( + `SELECT c.id, c.policy_id AS "policyId", + c.coverage_type AS "coverageType", + c.trigger_source AS "triggerSource", + c.processed_at AS "processedAt" + FROM claims c + WHERE c.trigger_source IN ( + SELECT name FROM oracle_providers WHERE id = $1 + ) + AND c.processed_at >= now() - ($2 * INTERVAL '1 second') + ORDER BY c.processed_at DESC`, + [params.providerId, params.lookbackSeconds], + ); + return rows; + } +} diff --git a/src/oracle/providers/credentials.ts b/src/oracle/providers/credentials.ts new file mode 100644 index 0000000..5c294d3 --- /dev/null +++ b/src/oracle/providers/credentials.ts @@ -0,0 +1,89 @@ +import { createHash, timingSafeEqual, randomBytes } from "crypto"; + +/** + * Provider identity and credential types. + * + * Identities are stable provider-level records. Keys are versioned child + * records so revocation is a key-level operation — the provider survives + * a rotation. Revocations are append-only (a superseding record reverses + * an accidental revocation; nothing is ever deleted). + */ + +export type KeyStatus = "active" | "rotating" | "revoked" | "superseded"; + +export interface ProviderIdentity { + id: string; + name: string; + contact: string; + createdAt: Date; + updatedAt: Date; + isActive: boolean; +} + +export interface ProviderKey { + id: string; + providerId: string; + /** PBKDF2-SHA256 hash of the raw key — never store plaintext. */ + keyHash: string; + /** Key version counter, incremented on each rotation. */ + version: number; + status: KeyStatus; + createdAt: Date; + /** Set when this key enters the rotating state (overlap window start). */ + rotatingAt: Date | null; + /** Set when revocation or supersession is recorded. */ + revokedAt: Date | null; + /** Ledger sequence number at which this key is no longer accepted. */ + effectiveUntilSequence: number | null; +} + +export interface RevocationRecord { + id: string; + keyId: string; + providerId: string; + reason: string; + actor: string; + /** Ledger sequence from which submissions with this key are rejected. */ + effectiveFromSequence: number; + /** Timestamp of the revocation action. */ + revokedAt: Date; + /** If set, this record supersedes a prior accidental revocation. */ + supersedesRevocationId: string | null; +} + +export interface KeyRotationRecord { + id: string; + providerId: string; + oldKeyId: string; + newKeyId: string; + /** Both keys are accepted during the overlap window. */ + overlapWindowEndsAt: Date; + actor: string; + initiatedAt: Date; + completedAt: Date | null; +} + +/** + * Generate a new raw API key (32 random bytes, hex-encoded). + * The caller stores the raw value once, then discards it. + */ +export function generateRawKey(): string { + return randomBytes(32).toString("hex"); +} + +/** + * Hash a raw API key using SHA-256. The hash is stored in the DB; + * the raw key is shown to the operator exactly once. + */ +export function hashKey(rawKey: string): string { + return createHash("sha256").update(rawKey).digest("hex"); +} + +/** + * Timing-safe comparison of a presented raw key against a stored hash. + */ +export function verifyKey(rawKey: string, storedHash: string): boolean { + const candidate = Buffer.from(hashKey(rawKey), "hex"); + const stored = Buffer.from(storedHash, "hex"); + return candidate.length === stored.length && timingSafeEqual(candidate, stored); +} diff --git a/src/oracle/providers/ops-provider.controller.ts b/src/oracle/providers/ops-provider.controller.ts new file mode 100644 index 0000000..bf6592e --- /dev/null +++ b/src/oracle/providers/ops-provider.controller.ts @@ -0,0 +1,183 @@ +import { + Body, + Controller, + Get, + HttpCode, + HttpStatus, + Param, + Post, + Query, + UseGuards, +} from "@nestjs/common"; +import { AdminOnly } from "../../common/admin-auth"; +import { AdminApiKeyGuard } from "../../common/admin-auth"; +import { ProviderCredentialService } from "./provider-credential.service"; + +interface RegisterProviderDto { + name: string; + contact: string; +} + +interface RevokeKeyDto { + reason: string; + actor: string; + effectiveFromSequence: number; + lookbackSeconds?: number; +} + +interface ReverseRevocationDto { + supersedesRevocationId: string; + reason: string; + actor: string; + effectiveFromSequence: number; +} + +interface InitiateRotationDto { + actor: string; + overlapWindowSeconds?: number; +} + +interface CompleteRotationDto { + actor: string; +} + +/** + * Ops endpoints for oracle provider identity and credential management. + * + * All write actions require the admin API key (x-api-key header). + * Trust-status (GET) is publicly readable for the provenance surface. + * + * POST /ops/providers — register a new provider + * GET /ops/providers — list all providers + * GET /ops/providers/:id/keys — list keys for a provider + * POST /ops/providers/:id/keys/revoke — revoke a key (produces forensic report) + * POST /ops/providers/:id/keys/reverse-revocation — reverse an accidental revocation + * POST /ops/providers/:id/keys/rotate — initiate key rotation (overlap window) + * POST /ops/providers/:id/keys/rotate/complete — complete rotation + * GET /ops/providers/:id/trust-status — public trust/provenance surface + */ +@Controller("ops/providers") +@UseGuards(AdminApiKeyGuard) +export class OpsProviderController { + constructor(private readonly credentialService: ProviderCredentialService) {} + + @Post() + @AdminOnly() + async registerProvider(@Body() dto: RegisterProviderDto) { + const { provider, rawKey } = await this.credentialService.registerProvider( + dto.name, + dto.contact, + ); + return { + provider, + // Raw key is shown exactly once; it cannot be recovered after this response. + rawKey, + warning: "Store this key securely. It will not be shown again.", + }; + } + + @Get() + @AdminOnly() + async listProviders() { + return this.credentialService.listProviders(); + } + + @Get(":providerId/keys") + @AdminOnly() + async listKeys(@Param("providerId") providerId: string) { + const provider = await this.credentialService.getProvider(providerId); + const trust = await this.credentialService.getTrustStatus(providerId); + return { provider, ...trust }; + } + + /** + * Revoke a specific key. Returns a full forensic report of every oracle + * event and claim decision the provider contributed to within the lookback + * window. This report is linked to claim remediation. + */ + @Post(":providerId/keys/:keyId/revoke") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async revokeKey( + @Param("providerId") _providerId: string, + @Param("keyId") keyId: string, + @Body() dto: RevokeKeyDto, + ) { + return this.credentialService.revokeKey({ + keyId, + reason: dto.reason, + actor: dto.actor, + effectiveFromSequence: dto.effectiveFromSequence, + lookbackSeconds: dto.lookbackSeconds, + }); + } + + /** + * Reverse an accidental revocation. Inserts a superseding record; the + * original revocation is preserved for evidence retention. + */ + @Post(":providerId/keys/:keyId/reverse-revocation") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async reverseRevocation( + @Param("keyId") keyId: string, + @Body() dto: ReverseRevocationDto, + ) { + return this.credentialService.reverseRevocation({ + keyId, + supersedesRevocationId: dto.supersedesRevocationId, + reason: dto.reason, + actor: dto.actor, + effectiveFromSequence: dto.effectiveFromSequence, + }); + } + + /** + * Initiate key rotation. Old key moves to "rotating" (still accepted during + * overlap window). Returns the new raw key — shown once. + */ + @Post(":providerId/keys/rotate") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async initiateRotation( + @Param("providerId") providerId: string, + @Body() dto: InitiateRotationDto, + ) { + const result = await this.credentialService.initiateRotation({ + providerId, + actor: dto.actor, + overlapWindowSeconds: dto.overlapWindowSeconds, + }); + return { + ...result, + warning: "Store the new rawKey securely. It will not be shown again.", + }; + } + + /** + * Complete a rotation once the new key is confirmed in use. + * Old key is superseded and no longer accepted. + */ + @Post(":providerId/keys/rotate/complete") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async completeRotation( + @Param("providerId") providerId: string, + @Body() dto: CompleteRotationDto, + ) { + await this.credentialService.completeRotation({ + providerId, + actor: dto.actor, + }); + return { message: "Rotation completed. Old key is now superseded." }; + } + + /** + * Public trust/provenance surface. Returns key counts, in-progress rotation, + * and revocation history for downstream consumers. + */ + @Get(":providerId/trust-status") + async getTrustStatus(@Param("providerId") providerId: string) { + return this.credentialService.getTrustStatus(providerId); + } +} diff --git a/src/oracle/providers/provider-credential.service.ts b/src/oracle/providers/provider-credential.service.ts new file mode 100644 index 0000000..677528c --- /dev/null +++ b/src/oracle/providers/provider-credential.service.ts @@ -0,0 +1,367 @@ +import { + BadRequestException, + ConflictException, + Injectable, + Logger, + NotFoundException, + UnauthorizedException, +} from "@nestjs/common"; +import { CredentialRepository } from "./credential.repository"; +import { + generateRawKey, + hashKey, + verifyKey, + ProviderIdentity, + ProviderKey, + RevocationRecord, + KeyRotationRecord, +} from "./credentials"; + +/** Default overlap window for a healthy rotation (seconds). */ +const DEFAULT_OVERLAP_WINDOW_SECONDS = 300; // 5 minutes + +/** Default lookback when building a forensic report (seconds). */ +const DEFAULT_LOOKBACK_SECONDS = 86_400 * 7; // 7 days + +export interface ForensicReport { + provider: ProviderIdentity; + revocation: RevocationRecord; + affectedEvents: Array<{ + id: bigint; + coverageType: string; + value: number; + source: string; + recordedAt: Date; + }>; + affectedClaims: Array<{ + id: string; + policyId: string; + coverageType: string; + triggerSource: string; + processedAt: Date; + }>; + generatedAt: Date; +} + +export interface RotationResult { + oldKey: ProviderKey; + newKey: ProviderKey; + rawKey: string; + rotation: KeyRotationRecord; + overlapWindowEndsAt: Date; +} + +@Injectable() +export class ProviderCredentialService { + private readonly logger = new Logger(ProviderCredentialService.name); + + constructor(private readonly repo: CredentialRepository) {} + + // ── Provider management ────────────────────────────────────────────────── + + async registerProvider(name: string, contact: string): Promise<{ provider: ProviderIdentity; rawKey: string }> { + const provider = await this.repo.createProvider(name, contact); + const rawKey = generateRawKey(); + await this.repo.createKey(provider.id, hashKey(rawKey), 1); + this.logger.log(`Registered provider ${provider.id} (${name})`); + return { provider, rawKey }; + } + + async getProvider(providerId: string): Promise { + const p = await this.repo.getProvider(providerId); + if (!p) throw new NotFoundException(`Provider ${providerId} not found`); + return p; + } + + async listProviders(): Promise { + return this.repo.listProviders(); + } + + // ── Submission boundary check ──────────────────────────────────────────── + + /** + * Validates a raw API key at the submission boundary. + * Returns the matching key record, or throws UnauthorizedException. + * Called on every inbound oracle submission before it is accepted. + */ + async validateSubmission(providerId: string, rawKey: string): Promise { + const acceptableKeys = await this.repo.getAcceptableKeys(providerId); + for (const key of acceptableKeys) { + if (verifyKey(rawKey, key.keyHash)) { + return key; + } + } + throw new UnauthorizedException( + `No valid active key for provider ${providerId}. Key may be revoked or unknown.`, + ); + } + + // ── Revocation ──────────────────────────────────────────────────────────── + + /** + * Revokes a provider key immediately. + * + * - Marks the key as "revoked" in provider_keys. + * - Inserts an append-only revocation record with reason, actor, and + * effective-from sequence. + * - Builds and returns a forensic report of every oracle event and claim + * decision this provider contributed to within the lookback window. + * + * Revocations are never deletable. Accidental revocations are reversed by + * calling reverseRevocation(), which inserts a superseding record and + * restores the key to "active". + */ + async revokeKey(params: { + keyId: string; + reason: string; + actor: string; + effectiveFromSequence: number; + lookbackSeconds?: number; + }): Promise { + const key = await this.repo.getKey(params.keyId); + if (!key) throw new NotFoundException(`Key ${params.keyId} not found`); + if (key.status === "revoked") { + throw new ConflictException(`Key ${params.keyId} is already revoked`); + } + + const provider = await this.repo.getProvider(key.providerId); + if (!provider) throw new NotFoundException(`Provider ${key.providerId} not found`); + + await this.repo.updateKeyStatus(params.keyId, "revoked", { + revokedAt: new Date(), + effectiveUntilSequence: params.effectiveFromSequence, + }); + + const revocation = await this.repo.recordRevocation({ + keyId: params.keyId, + providerId: key.providerId, + reason: params.reason, + actor: params.actor, + effectiveFromSequence: params.effectiveFromSequence, + }); + + this.logger.warn( + `Key ${params.keyId} (provider ${key.providerId}) revoked by ${params.actor} ` + + `effective from sequence ${params.effectiveFromSequence}. Reason: ${params.reason}`, + ); + + return this.buildForensicReport( + provider, + revocation, + params.lookbackSeconds ?? DEFAULT_LOOKBACK_SECONDS, + ); + } + + /** + * Reverses an accidental revocation by restoring the key to "active" and + * inserting a superseding revocation record. The original record is preserved + * for evidence retention. + */ + async reverseRevocation(params: { + keyId: string; + supersedesRevocationId: string; + reason: string; + actor: string; + effectiveFromSequence: number; + }): Promise { + const key = await this.repo.getKey(params.keyId); + if (!key) throw new NotFoundException(`Key ${params.keyId} not found`); + if (key.status !== "revoked") { + throw new BadRequestException(`Key ${params.keyId} is not currently revoked`); + } + + await this.repo.updateKeyStatus(params.keyId, "active", { + revokedAt: undefined, + effectiveUntilSequence: undefined, + }); + + const superseding = await this.repo.recordRevocation({ + keyId: params.keyId, + providerId: key.providerId, + reason: params.reason, + actor: params.actor, + effectiveFromSequence: params.effectiveFromSequence, + supersedesRevocationId: params.supersedesRevocationId, + }); + + this.logger.log( + `Revocation reversed for key ${params.keyId} by ${params.actor}. ` + + `Supersedes revocation ${params.supersedesRevocationId}`, + ); + + return superseding; + } + + // ── Key rotation ────────────────────────────────────────────────────────── + + /** + * Initiates active key rotation for a healthy provider. + * + * - Creates a new key (next version) in "active" status. + * - Sets the old key to "rotating" — it remains valid during the overlap + * window. + * - Records a rotation record for audit and overlap-window tracking. + * + * During the overlap window both the old and new keys are accepted at the + * submission boundary. Call completeRotation() when the new key is confirmed + * in use to retire the old key. + */ + async initiateRotation(params: { + providerId: string; + actor: string; + overlapWindowSeconds?: number; + }): Promise { + const provider = await this.repo.getProvider(params.providerId); + if (!provider) throw new NotFoundException(`Provider ${params.providerId} not found`); + + const existing = await this.repo.getActiveRotation(params.providerId); + if (existing) { + throw new ConflictException( + `Provider ${params.providerId} already has an in-progress rotation`, + ); + } + + const currentKeys = await this.repo.listKeysByProvider(params.providerId); + const activeKey = currentKeys.find((k) => k.status === "active"); + if (!activeKey) { + throw new BadRequestException( + `Provider ${params.providerId} has no active key to rotate`, + ); + } + + const nextVersion = Math.max(...currentKeys.map((k) => k.version)) + 1; + const rawKey = generateRawKey(); + const newKey = await this.repo.createKey( + params.providerId, + hashKey(rawKey), + nextVersion, + ); + + const overlapSeconds = params.overlapWindowSeconds ?? DEFAULT_OVERLAP_WINDOW_SECONDS; + const overlapWindowEndsAt = new Date(Date.now() + overlapSeconds * 1000); + + await this.repo.updateKeyStatus(activeKey.id, "rotating", { + rotatingAt: new Date(), + }); + + const rotation = await this.repo.recordRotation({ + providerId: params.providerId, + oldKeyId: activeKey.id, + newKeyId: newKey.id, + overlapWindowEndsAt, + actor: params.actor, + }); + + this.logger.log( + `Rotation initiated for provider ${params.providerId} by ${params.actor}. ` + + `Old key: ${activeKey.id} → New key: ${newKey.id}. ` + + `Overlap window ends: ${overlapWindowEndsAt.toISOString()}`, + ); + + return { + oldKey: { ...activeKey, status: "rotating" }, + newKey, + rawKey, + rotation, + overlapWindowEndsAt, + }; + } + + /** + * Completes a rotation once the new key is confirmed in use. + * Supersedes the old key and closes the overlap window. + */ + async completeRotation(params: { + providerId: string; + actor: string; + }): Promise { + const rotation = await this.repo.getActiveRotation(params.providerId); + if (!rotation) { + throw new NotFoundException(`No active rotation found for provider ${params.providerId}`); + } + + await this.repo.updateKeyStatus(rotation.oldKeyId, "superseded", { + revokedAt: new Date(), + }); + await this.repo.completeRotation(rotation.id); + + this.logger.log( + `Rotation completed for provider ${params.providerId} by ${params.actor}. ` + + `Old key ${rotation.oldKeyId} superseded.`, + ); + } + + // ── Forensic report ─────────────────────────────────────────────────────── + + /** + * Builds a full forensic report for a provider: all oracle events and claim + * decisions it contributed to within the lookback window, linked to the + * revocation record for claim remediation. + */ + async buildForensicReport( + provider: ProviderIdentity, + revocation: RevocationRecord, + lookbackSeconds: number = DEFAULT_LOOKBACK_SECONDS, + ): Promise { + const [affectedEvents, affectedClaims] = await Promise.all([ + this.repo.findAffectedEvents({ + providerId: provider.id, + effectiveFromSequence: revocation.effectiveFromSequence, + lookbackSeconds, + }), + this.repo.findAffectedClaims({ + providerId: provider.id, + lookbackSeconds, + }), + ]); + + this.logger.log( + `Forensic report for provider ${provider.id}: ` + + `${affectedEvents.length} events, ${affectedClaims.length} claims in lookback window`, + ); + + return { + provider, + revocation, + affectedEvents, + affectedClaims, + generatedAt: new Date(), + }; + } + + // ── Trust status surface ───────────────────────────────────────────────── + + /** + * Returns the public trust-status for a provider: active key count, + * revocation count, and any in-progress rotation. Used by the provenance + * surface. + */ + async getTrustStatus(providerId: string): Promise<{ + providerId: string; + isActive: boolean; + activeKeyCount: number; + revokedKeyCount: number; + rotatingKeyCount: number; + inProgressRotation: KeyRotationRecord | null; + revocationHistory: RevocationRecord[]; + }> { + const provider = await this.repo.getProvider(providerId); + if (!provider) throw new NotFoundException(`Provider ${providerId} not found`); + + const [keys, revocations, rotation] = await Promise.all([ + this.repo.listKeysByProvider(providerId), + this.repo.getRevocationsForProvider(providerId), + this.repo.getActiveRotation(providerId), + ]); + + return { + providerId, + isActive: provider.isActive, + activeKeyCount: keys.filter((k) => k.status === "active").length, + revokedKeyCount: keys.filter((k) => k.status === "revoked").length, + rotatingKeyCount: keys.filter((k) => k.status === "rotating").length, + inProgressRotation: rotation, + revocationHistory: revocations, + }; + } +} diff --git a/src/oracle/providers/provider-credentials.module.ts b/src/oracle/providers/provider-credentials.module.ts new file mode 100644 index 0000000..0792b84 --- /dev/null +++ b/src/oracle/providers/provider-credentials.module.ts @@ -0,0 +1,17 @@ +import { Module } from "@nestjs/common"; +import { CredentialRepository } from "./credential.repository"; +import { ProviderCredentialService } from "./provider-credential.service"; +import { OpsProviderController } from "./ops-provider.controller"; +import { DatabaseModule } from "../../database/database.module"; + +/** + * Provides provider identity management, versioned API key lifecycle, + * revocation, rotation, and forensic reporting for oracle providers. + */ +@Module({ + imports: [DatabaseModule], + providers: [CredentialRepository, ProviderCredentialService], + controllers: [OpsProviderController], + exports: [ProviderCredentialService], +}) +export class ProviderCredentialsModule {} diff --git a/src/oracle/providers/provider-credentials.spec.ts b/src/oracle/providers/provider-credentials.spec.ts new file mode 100644 index 0000000..ca395f5 --- /dev/null +++ b/src/oracle/providers/provider-credentials.spec.ts @@ -0,0 +1,356 @@ +/** + * Tests for provider credential lifecycle: + * 1. Revoked keys are rejected at the submission boundary + * 2. Overlap-window rotation accepts both keys, then only the new one + * 3. Affected-decision report correctness + * 4. Accidental-revocation reversal + */ +import { NotFoundException, UnauthorizedException, ConflictException, BadRequestException } from "@nestjs/common"; +import { ProviderCredentialService } from "./provider-credential.service"; +import { CredentialRepository } from "./credential.repository"; +import { generateRawKey, hashKey, ProviderIdentity, ProviderKey, RevocationRecord } from "./credentials"; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function makeProvider(overrides: Partial = {}): ProviderIdentity { + return { + id: "provider-1", + name: "TestProvider", + contact: "test@example.com", + isActive: true, + createdAt: new Date(), + updatedAt: new Date(), + ...overrides, + }; +} + +function makeKey(overrides: Partial = {}): ProviderKey { + return { + id: "key-1", + providerId: "provider-1", + keyHash: hashKey("rawsecret"), + version: 1, + status: "active", + createdAt: new Date(), + rotatingAt: null, + revokedAt: null, + effectiveUntilSequence: null, + ...overrides, + }; +} + +function makeRevocation(overrides: Partial = {}): RevocationRecord { + return { + id: "rev-1", + keyId: "key-1", + providerId: "provider-1", + reason: "Key leaked", + actor: "admin", + effectiveFromSequence: 1000, + revokedAt: new Date(), + supersedesRevocationId: null, + ...overrides, + }; +} + +function buildMockRepo(overrides: Partial = {}): jest.Mocked { + return { + createProvider: jest.fn(), + getProvider: jest.fn(), + listProviders: jest.fn(), + createKey: jest.fn(), + getKey: jest.fn(), + listKeysByProvider: jest.fn(), + getAcceptableKeys: jest.fn(), + updateKeyStatus: jest.fn(), + recordRevocation: jest.fn(), + getRevocationsForKey: jest.fn(), + getRevocationsForProvider: jest.fn(), + recordRotation: jest.fn(), + completeRotation: jest.fn(), + getActiveRotation: jest.fn(), + findAffectedEvents: jest.fn(), + findAffectedClaims: jest.fn(), + ...overrides, + } as unknown as jest.Mocked; +} + +// ── Suite ───────────────────────────────────────────────────────────────────── + +describe("ProviderCredentialService", () => { + let service: ProviderCredentialService; + let repo: jest.Mocked; + + beforeEach(() => { + repo = buildMockRepo(); + service = new ProviderCredentialService(repo); + }); + + // ── 1. Revoked keys are rejected at submission boundary ─────────────────── + + describe("validateSubmission", () => { + it("accepts a valid active key", async () => { + const rawKey = "rawsecret"; + const key = makeKey({ keyHash: hashKey(rawKey) }); + repo.getAcceptableKeys.mockResolvedValue([key]); + + const result = await service.validateSubmission("provider-1", rawKey); + expect(result).toEqual(key); + }); + + it("rejects when no acceptable keys exist (all revoked)", async () => { + repo.getAcceptableKeys.mockResolvedValue([]); + await expect( + service.validateSubmission("provider-1", "any-raw-key"), + ).rejects.toBeInstanceOf(UnauthorizedException); + }); + + it("rejects a key that does not match any acceptable key hash", async () => { + const key = makeKey({ keyHash: hashKey("correct-secret") }); + repo.getAcceptableKeys.mockResolvedValue([key]); + + await expect( + service.validateSubmission("provider-1", "wrong-secret"), + ).rejects.toBeInstanceOf(UnauthorizedException); + }); + + it("accepts a rotating key during the overlap window", async () => { + const rawKey = "rotating-secret"; + const rotatingKey = makeKey({ keyHash: hashKey(rawKey), status: "rotating" }); + repo.getAcceptableKeys.mockResolvedValue([rotatingKey]); + + const result = await service.validateSubmission("provider-1", rawKey); + expect(result.status).toBe("rotating"); + }); + }); + + // ── 2. Overlap-window rotation: both keys accepted, then only new ───────── + + describe("initiateRotation + completeRotation", () => { + it("issues a new key and puts old key into rotating state", async () => { + const provider = makeProvider(); + const oldKey = makeKey({ id: "key-1", version: 1, status: "active" }); + const newKey = makeKey({ id: "key-2", version: 2, status: "active" }); + + repo.getProvider.mockResolvedValue(provider); + repo.getActiveRotation.mockResolvedValue(null); + repo.listKeysByProvider.mockResolvedValue([oldKey]); + repo.createKey.mockResolvedValue(newKey); + repo.updateKeyStatus.mockResolvedValue(undefined); + repo.recordRotation.mockResolvedValue({ + id: "rot-1", + providerId: "provider-1", + oldKeyId: "key-1", + newKeyId: "key-2", + overlapWindowEndsAt: new Date(Date.now() + 300_000), + actor: "admin", + initiatedAt: new Date(), + completedAt: null, + }); + + const result = await service.initiateRotation({ + providerId: "provider-1", + actor: "admin", + overlapWindowSeconds: 300, + }); + + expect(result.rawKey).toBeTruthy(); + expect(result.oldKey.id).toBe("key-1"); + expect(result.newKey.id).toBe("key-2"); + // Old key is marked rotating + expect(repo.updateKeyStatus).toHaveBeenCalledWith( + "key-1", "rotating", expect.objectContaining({ rotatingAt: expect.any(Date) }), + ); + }); + + it("rejects rotation when one is already in-progress", async () => { + const provider = makeProvider(); + repo.getProvider.mockResolvedValue(provider); + repo.getActiveRotation.mockResolvedValue({ + id: "rot-existing", + providerId: "provider-1", + oldKeyId: "key-1", + newKeyId: "key-2", + overlapWindowEndsAt: new Date(), + actor: "admin", + initiatedAt: new Date(), + completedAt: null, + }); + + await expect( + service.initiateRotation({ providerId: "provider-1", actor: "admin" }), + ).rejects.toBeInstanceOf(ConflictException); + }); + + it("completeRotation supersedes the old key", async () => { + repo.getActiveRotation.mockResolvedValue({ + id: "rot-1", + providerId: "provider-1", + oldKeyId: "key-1", + newKeyId: "key-2", + overlapWindowEndsAt: new Date(), + actor: "admin", + initiatedAt: new Date(), + completedAt: null, + }); + repo.updateKeyStatus.mockResolvedValue(undefined); + repo.completeRotation.mockResolvedValue(undefined); + + await service.completeRotation({ providerId: "provider-1", actor: "admin" }); + + expect(repo.updateKeyStatus).toHaveBeenCalledWith( + "key-1", "superseded", expect.anything(), + ); + expect(repo.completeRotation).toHaveBeenCalledWith("rot-1"); + }); + + it("after completeRotation, only the new key is accepted", async () => { + // Simulate post-completion: getAcceptableKeys returns only the new key + const newRaw = "new-raw-key"; + const newKey = makeKey({ id: "key-2", keyHash: hashKey(newRaw), version: 2, status: "active" }); + repo.getAcceptableKeys.mockResolvedValue([newKey]); + + // Old key is rejected + await expect( + service.validateSubmission("provider-1", "old-raw-key"), + ).rejects.toBeInstanceOf(UnauthorizedException); + + // New key is accepted + const result = await service.validateSubmission("provider-1", newRaw); + expect(result.id).toBe("key-2"); + }); + }); + + // ── 3. Affected-decision report correctness ─────────────────────────────── + + describe("revokeKey — forensic report", () => { + it("generates a forensic report with affected events and claims", async () => { + const key = makeKey({ status: "active" }); + const provider = makeProvider(); + const revocation = makeRevocation(); + + repo.getKey.mockResolvedValue(key); + repo.getProvider.mockResolvedValue(provider); + repo.updateKeyStatus.mockResolvedValue(undefined); + repo.recordRevocation.mockResolvedValue(revocation); + repo.findAffectedEvents.mockResolvedValue([ + { id: BigInt(1), coverageType: "stablecoin_depeg", value: 0.9950, source: "TestProvider", recordedAt: new Date() }, + ]); + repo.findAffectedClaims.mockResolvedValue([ + { id: "claim-1", policyId: "policy-1", coverageType: "stablecoin_depeg", triggerSource: "TestProvider", processedAt: new Date() }, + ]); + + const report = await service.revokeKey({ + keyId: "key-1", + reason: "Key leaked in incident #42", + actor: "ops-admin", + effectiveFromSequence: 1000, + lookbackSeconds: 3600, + }); + + expect(report.affectedEvents).toHaveLength(1); + expect(report.affectedClaims).toHaveLength(1); + expect(report.revocation.reason).toBe("Key leaked in incident #42"); + expect(report.generatedAt).toBeInstanceOf(Date); + }); + + it("throws if key not found", async () => { + repo.getKey.mockResolvedValue(null); + await expect( + service.revokeKey({ keyId: "missing", reason: "x", actor: "a", effectiveFromSequence: 0 }), + ).rejects.toBeInstanceOf(NotFoundException); + }); + + it("throws ConflictException if key is already revoked", async () => { + repo.getKey.mockResolvedValue(makeKey({ status: "revoked" })); + await expect( + service.revokeKey({ keyId: "key-1", reason: "x", actor: "a", effectiveFromSequence: 0 }), + ).rejects.toBeInstanceOf(ConflictException); + }); + }); + + // ── 4. Accidental-revocation reversal ───────────────────────────────────── + + describe("reverseRevocation", () => { + it("restores key to active and inserts superseding record", async () => { + const revokedKey = makeKey({ status: "revoked" }); + const supersedingRecord = makeRevocation({ + id: "rev-2", + supersedesRevocationId: "rev-1", + }); + + repo.getKey.mockResolvedValue(revokedKey); + repo.updateKeyStatus.mockResolvedValue(undefined); + repo.recordRevocation.mockResolvedValue(supersedingRecord); + + const result = await service.reverseRevocation({ + keyId: "key-1", + supersedesRevocationId: "rev-1", + reason: "Revocation was made in error", + actor: "ops-admin", + effectiveFromSequence: 1001, + }); + + // Key status restored to active + expect(repo.updateKeyStatus).toHaveBeenCalledWith( + "key-1", "active", expect.anything(), + ); + // Superseding record created + expect(repo.recordRevocation).toHaveBeenCalledWith( + expect.objectContaining({ supersedesRevocationId: "rev-1" }), + ); + expect(result.supersedesRevocationId).toBe("rev-1"); + }); + + it("throws BadRequestException if key is not currently revoked", async () => { + repo.getKey.mockResolvedValue(makeKey({ status: "active" })); + await expect( + service.reverseRevocation({ + keyId: "key-1", + supersedesRevocationId: "rev-1", + reason: "error", + actor: "admin", + effectiveFromSequence: 0, + }), + ).rejects.toBeInstanceOf(BadRequestException); + }); + + it("throws NotFoundException if key does not exist", async () => { + repo.getKey.mockResolvedValue(null); + await expect( + service.reverseRevocation({ + keyId: "missing", + supersedesRevocationId: "rev-1", + reason: "error", + actor: "admin", + effectiveFromSequence: 0, + }), + ).rejects.toBeInstanceOf(NotFoundException); + }); + }); + + // ── Credential helpers ──────────────────────────────────────────────────── + + describe("generateRawKey / hashKey / verifyKey", () => { + it("generates unique keys", () => { + const a = generateRawKey(); + const b = generateRawKey(); + expect(a).not.toBe(b); + expect(a).toHaveLength(64); // 32 bytes hex + }); + + it("verifyKey returns true for matching key/hash", () => { + const { verifyKey } = require("./credentials"); + const raw = generateRawKey(); + const hash = hashKey(raw); + expect(verifyKey(raw, hash)).toBe(true); + }); + + it("verifyKey returns false for wrong key", () => { + const { verifyKey } = require("./credentials"); + const raw = generateRawKey(); + const hash = hashKey(raw); + expect(verifyKey("wrong" + raw, hash)).toBe(false); + }); + }); +}); diff --git a/src/oracle/providers/provider-key.guard.ts b/src/oracle/providers/provider-key.guard.ts new file mode 100644 index 0000000..5b67e5d --- /dev/null +++ b/src/oracle/providers/provider-key.guard.ts @@ -0,0 +1,61 @@ +import { + CanActivate, + ExecutionContext, + Injectable, + UnauthorizedException, +} from "@nestjs/common"; +import { ProviderCredentialService } from "./provider-credential.service"; + +export interface ProviderRequest { + headers: Record; + authenticatedProvider?: { providerId: string; keyId: string; keyVersion: number }; +} + +/** + * Submission-boundary guard: validates the provider ID and raw API key + * supplied in request headers before any oracle reading is accepted. + * + * Headers: + * x-provider-id — stable provider UUID + * x-provider-key — raw API key (shown once at registration/rotation) + * + * Rejected immediately if: + * - Headers are absent + * - Provider does not exist + * - Key is revoked, superseded, or unknown + * - Submission falls after effectiveUntilSequence on the key + * + * On success, attaches authenticatedProvider to the request for downstream use. + */ +@Injectable() +export class ProviderKeyGuard implements CanActivate { + constructor(private readonly credentialService: ProviderCredentialService) {} + + async canActivate(context: ExecutionContext): Promise { + const req = context.switchToHttp().getRequest(); + + const providerId = this.extractHeader(req, "x-provider-id"); + const rawKey = this.extractHeader(req, "x-provider-key"); + + if (!providerId || !rawKey) { + throw new UnauthorizedException( + "Oracle submissions require x-provider-id and x-provider-key headers", + ); + } + + const key = await this.credentialService.validateSubmission(providerId, rawKey); + + req.authenticatedProvider = { + providerId, + keyId: key.id, + keyVersion: key.version, + }; + + return true; + } + + private extractHeader(req: ProviderRequest, name: string): string | undefined { + const val = req.headers[name]; + return Array.isArray(val) ? val[0] : val; + } +} From 422e7e282a61d4851ec6c4d7c6744c987a5e5820 Mon Sep 17 00:00:00 2001 From: oscar24357 Date: Thu, 22 Oct 2026 18:32:02 +0100 Subject: [PATCH 2/3] feat: periodic contract spec revalidation with circuit-breaker and diff report - Add ContractSpecRevalidator: revalidates pool + oracle contract specs on configurable interval (default 300s) and on-demand via admin API - Reuses startup validatePoolContractSpec/parseContractSpec so startup and runtime checks cannot drift apart - Three states: valid / mismatch / unknown (RPC unavailable != mismatch) - assertPublicationAllowed() / assertSettlementAllowed() gates for OraclePublisherService and ClaimSettlementService - Block persisted in control_plane_flags Postgres table -- survives restarts - Block clears only after successful revalidation or explicit operator override (reason + actor recorded, alerted) - Structured FunctionDiff report (added/removed/changed + expected/actual) exposed via GET /ops/contracts/diff-report - Admin endpoints: revalidate, override, override/clear, diff-report - Migration 005: control_plane_flags table - DbModule created so DbService is injectable across modules - Spec covers: mismatch blocks + alert, RPC error -> unknown, restart persistence, diff-report correctness, operator override --- src/db/db.module.ts | 8 + src/db/migrations/005_control_plane_flags.sql | 12 + src/stellar/contract-spec-revalidator.spec.ts | 322 ++++++++++++++ src/stellar/contract-spec-revalidator.ts | 399 ++++++++++++++++++ src/stellar/ops-contract-spec.controller.ts | 60 +++ src/stellar/stellar.module.ts | 11 +- 6 files changed, 810 insertions(+), 2 deletions(-) create mode 100644 src/db/migrations/005_control_plane_flags.sql create mode 100644 src/stellar/contract-spec-revalidator.spec.ts create mode 100644 src/stellar/contract-spec-revalidator.ts create mode 100644 src/stellar/ops-contract-spec.controller.ts diff --git a/src/db/db.module.ts b/src/db/db.module.ts index e69de29..2343d1c 100644 --- a/src/db/db.module.ts +++ b/src/db/db.module.ts @@ -0,0 +1,8 @@ +import { Module } from "@nestjs/common"; +import { DbService } from "./db.service"; + +@Module({ + providers: [DbService], + exports: [DbService], +}) +export class DbModule {} diff --git a/src/db/migrations/005_control_plane_flags.sql b/src/db/migrations/005_control_plane_flags.sql new file mode 100644 index 0000000..3c04b82 --- /dev/null +++ b/src/db/migrations/005_control_plane_flags.sql @@ -0,0 +1,12 @@ +-- Migration 005: durable control-plane flag store +-- +-- Used by ContractSpecRevalidator to persist the contract spec validation +-- state (valid / mismatch / unknown) across restarts. The block survives +-- a process restart and clears only after a successful revalidation or +-- an explicit operator override. + +CREATE TABLE IF NOT EXISTS control_plane_flags ( + key VARCHAR(80) PRIMARY KEY, + value JSONB NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/src/stellar/contract-spec-revalidator.spec.ts b/src/stellar/contract-spec-revalidator.spec.ts new file mode 100644 index 0000000..9815f98 --- /dev/null +++ b/src/stellar/contract-spec-revalidator.spec.ts @@ -0,0 +1,322 @@ +/** + * Tests for ContractSpecRevalidator: + * 1. Mocked spec change blocks publication and settlement, fires alert + * 2. RPC errors produce "unknown" (not "mismatch") and also block + * 3. Block survives a restart (persisted state loaded on init) + * 4. Diff-report correctness + * 5. Operator override clears the block + */ + +import { SchedulerRegistry } from "@nestjs/schedule"; +import { rpc } from "@stellar/stellar-sdk"; +import { ConfigService } from "@nestjs/config"; +import { AppConfig } from "../config/configuration"; +import { AlertingService } from "../alerting/alerting.service"; +import { DbService } from "../db/db.service"; +import { ContractSpecRevalidator, RevalidationResult } from "./contract-spec-revalidator"; +import * as PoolContractInterface from "./pool-contract-interface.service"; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +function buildConfig(overrides: Partial = {}): ConfigService { + return { + get: jest.fn((key: string) => { + if (key === "stellar") { + return { + sorobanRpcUrl: "https://soroban-testnet.stellar.org", + poolContractId: "CPOOL0000000000000000000000000000000000000000000000000000", + oracleContractId: "CORACLE000000000000000000000000000000000000000000000000", + ...overrides, + }; + } + if (key === "contractRevalidation") return { intervalSeconds: 300 }; + return undefined; + }), + } as unknown as ConfigService; +} + +function buildAlerting(): jest.Mocked { + return { emit: jest.fn(), listRecent: jest.fn() } as unknown as jest.Mocked; +} + +function buildDb(persistedState?: RevalidationResult): jest.Mocked { + return { + isAvailable: jest.fn().mockReturnValue(true), + query: jest.fn().mockImplementation((sql: string) => { + if (sql.includes("SELECT value") && persistedState) { + return Promise.resolve({ rows: [{ value: JSON.stringify(persistedState) }] }); + } + return Promise.resolve({ rows: [] }); + }), + } as unknown as jest.Mocked; +} + +function buildSchedulerRegistry(): jest.Mocked { + return { + addCronJob: jest.fn(), + doesExist: jest.fn().mockReturnValue(false), + deleteCronJob: jest.fn(), + } as unknown as jest.Mocked; +} + +function buildRevalidator( + overrides: Partial = {}, + persistedState?: RevalidationResult, +) { + const config = buildConfig(overrides); + const alerting = buildAlerting(); + const db = buildDb(persistedState); + const scheduler = buildSchedulerRegistry(); + const svc = new ContractSpecRevalidator(config, alerting, db, scheduler); + return { svc, alerting, db, scheduler }; +} + +function validWasm(): Buffer { + // Minimal valid WASM header (magic + version) — actual parsing is mocked. + return Buffer.from("0061736d01000000", "hex"); +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +describe("ContractSpecRevalidator", () => { + afterEach(() => jest.restoreAllMocks()); + + // 1. Mocked spec change blocks publication/settlement and fires alert ─────── + + it("blocks publication and fires critical alert on spec mismatch", async () => { + const { svc, alerting } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest.spyOn(PoolContractInterface, "validatePoolContractSpec").mockReturnValue([ + "missing function buy_policy", + "process_claim arguments expected (policy_id:string) but found (id:u64)", + ]); + + await svc.revalidate(); + + expect(alerting.emit).toHaveBeenCalledWith( + "critical", + expect.stringContaining("mismatch"), + expect.stringContaining("buy_policy"), + expect.anything(), + ); + + expect(() => svc.assertPublicationAllowed()).toThrow(/blocked/); + expect(() => svc.assertSettlementAllowed()).toThrow(/blocked/); + }); + + it("includes correct diff entries for changed and removed functions", async () => { + const { svc } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest.spyOn(PoolContractInterface, "validatePoolContractSpec").mockReturnValue([ + "missing function pool_config", + "buy_policy arguments expected (holder:address, params:udt:PolicyParams) but found (holder:address)", + ]); + + const result = await svc.revalidate(); + + expect(result.state).toBe("mismatch"); + const poolDiff = result.pool.diff; + expect(poolDiff).toContainEqual(expect.objectContaining({ name: "pool_config", change: "removed" })); + expect(poolDiff).toContainEqual(expect.objectContaining({ name: "buy_policy", change: "changed" })); + }); + + // 2. RPC errors produce "unknown", not "mismatch" ───────────────────────── + + it("returns unknown state when RPC is unavailable and blocks both operations", async () => { + const { svc, alerting } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockRejectedValue(new Error("connect ECONNREFUSED 127.0.0.1:8000")); + + const result = await svc.revalidate(); + + expect(result.state).toBe("unknown"); + expect(result.pool.state).toBe("unknown"); + expect(alerting.emit).toHaveBeenCalledWith( + "warning", + expect.stringContaining("unknown"), + expect.anything(), + ); + + expect(() => svc.assertPublicationAllowed()).toThrow(/blocked/); + expect(() => svc.assertSettlementAllowed()).toThrow(/blocked/); + }); + + it("does not classify an RPC timeout as a mismatch", async () => { + const { svc } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockRejectedValue(new Error("Request timeout after 5000ms")); + + const result = await svc.revalidate(); + + expect(result.state).toBe("unknown"); + expect(result.pool.state).toBe("unknown"); + expect(result.pool.errors[0]).toMatch(/RPC unavailable/); + }); + + // 3. Block survives a restart ─────────────────────────────────────────────── + + it("loads persisted mismatch state on init and blocks immediately before first revalidation", async () => { + const blockedState: RevalidationResult = { + state: "mismatch", + pool: { + contractId: "CPOOL", + contractName: "pool", + state: "mismatch", + checkedAt: new Date().toISOString(), + errors: ["missing function buy_policy"], + diff: [{ name: "buy_policy", change: "removed" }], + wasmHash: null, + }, + oracle: { + contractId: "CORACLE", + contractName: "oracle", + state: "valid", + checkedAt: new Date().toISOString(), + errors: [], + diff: [], + wasmHash: null, + }, + blockedAt: new Date().toISOString(), + operatorOverride: null, + }; + + // Build with a DB that returns the persisted state. + const { svc } = buildRevalidator({}, blockedState); + + // Only load state — don't call revalidate() yet (simulating fresh restart + // before the scheduler's first tick). + await (svc as unknown as { loadPersistedState(): Promise }).loadPersistedState(); + + expect(() => svc.assertPublicationAllowed()).toThrow(/blocked/); + expect(() => svc.assertSettlementAllowed()).toThrow(/blocked/); + }); + + // 4. Diff-report correctness ─────────────────────────────────────────────── + + it("exposes diff report through getLastResult() after a mismatch", async () => { + const { svc } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest.spyOn(PoolContractInterface, "validatePoolContractSpec").mockReturnValue([ + "process_claim result expected result:i128 but found void", + ]); + + await svc.revalidate(); + + const report = svc.getLastResult(); + expect(report).not.toBeNull(); + expect(report!.state).toBe("mismatch"); + expect(report!.pool.diff[0]).toEqual( + expect.objectContaining({ + name: "process_claim", + change: "changed", + expected: "result:i128", + actual: "void", + }), + ); + }); + + it("clears block and diff after a successful revalidation", async () => { + const { svc } = buildRevalidator(); + + // First: inject a mismatch. + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValueOnce(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest + .spyOn(PoolContractInterface, "validatePoolContractSpec") + .mockReturnValueOnce(["missing function pool_config"]) + .mockReturnValue([]); + + await svc.revalidate(); + expect(svc.getLastResult()!.state).toBe("mismatch"); + + // Second: spec is now clean. + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + + await svc.revalidate(); + + expect(svc.getLastResult()!.state).toBe("valid"); + expect(() => svc.assertPublicationAllowed()).not.toThrow(); + expect(() => svc.assertSettlementAllowed()).not.toThrow(); + }); + + // 5. Operator override ───────────────────────────────────────────────────── + + it("operator override clears the publication block without requiring a clean spec", async () => { + const { svc, alerting } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest.spyOn(PoolContractInterface, "validatePoolContractSpec").mockReturnValue(["missing function buy_policy"]); + + await svc.revalidate(); + expect(() => svc.assertPublicationAllowed()).toThrow(); + + await svc.applyOperatorOverride("Confirmed safe — hotfix pending", "ops-admin"); + + expect(() => svc.assertPublicationAllowed()).not.toThrow(); + expect(alerting.emit).toHaveBeenCalledWith( + "warning", + expect.stringContaining("overridden"), + expect.stringContaining("ops-admin"), + expect.anything(), + ); + }); + + it("removing the override restores the block", async () => { + const { svc } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest.spyOn(PoolContractInterface, "validatePoolContractSpec").mockReturnValue(["missing function buy_policy"]); + + await svc.revalidate(); + await svc.applyOperatorOverride("temp", "admin"); + expect(() => svc.assertPublicationAllowed()).not.toThrow(); + + await svc.clearOperatorOverride(); + expect(() => svc.assertPublicationAllowed()).toThrow(/blocked/); + }); + + // 6. State persistence ───────────────────────────────────────────────────── + + it("persists state to DB after each revalidation", async () => { + const { svc, db } = buildRevalidator(); + + jest + .spyOn(rpc.Server.prototype, "getContractWasmByContractId") + .mockResolvedValue(validWasm() as never); + jest.spyOn(PoolContractInterface, "parseContractSpec").mockReturnValue({} as never); + jest.spyOn(PoolContractInterface, "validatePoolContractSpec").mockReturnValue([]); + + await svc.revalidate(); + + expect(db.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO control_plane_flags"), + expect.arrayContaining(["contract_spec_validation_state"]), + ); + }); +}); diff --git a/src/stellar/contract-spec-revalidator.ts b/src/stellar/contract-spec-revalidator.ts new file mode 100644 index 0000000..911d80a --- /dev/null +++ b/src/stellar/contract-spec-revalidator.ts @@ -0,0 +1,399 @@ +import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; +import { SchedulerRegistry } from "@nestjs/schedule"; +import { CronJob } from "cron"; +import { rpc } from "@stellar/stellar-sdk"; +import { AppConfig } from "../config/configuration"; +import { AlertingService } from "../alerting/alerting.service"; +import { DbService } from "../db/db.service"; +import { + validatePoolContractSpec, + parseContractSpec, + extractContractSpec, +} from "./pool-contract-interface.service"; + +/** + * Three possible states for the contract-spec validation control-plane flag. + * + * - valid: last check passed; publication and settlement are allowed. + * - mismatch: spec diverged from expected; publication and settlement are BLOCKED. + * - unknown: RPC was unreachable; we fail-closed (block) but distinguish + * this from a confirmed mismatch so operators can tell the + * difference between an upgrade and a connectivity blip. + */ +export type ValidationState = "valid" | "mismatch" | "unknown"; + +export interface FunctionDiff { + name: string; + change: "added" | "removed" | "changed"; + expected?: string; + actual?: string; +} + +export interface ContractDiffReport { + contractId: string; + contractName: "pool" | "oracle"; + state: ValidationState; + checkedAt: string; + errors: string[]; + diff: FunctionDiff[]; + wasmHash: string | null; +} + +export interface RevalidationResult { + state: ValidationState; + pool: ContractDiffReport; + oracle: ContractDiffReport; + blockedAt: string | null; + operatorOverride: OperatorOverride | null; +} + +export interface OperatorOverride { + reason: string; + actor: string; + overriddenAt: string; +} + +const REVALIDATION_JOB_NAME = "contract-spec-revalidation"; +const CONTROL_PLANE_KEY = "contract_spec_validation_state"; + +/** + * Periodically and on-demand revalidates the live Soroban contract specs for + * the pool and oracle contracts against the backend's expected ABI. + * + * A mismatch (or RPC unavailability) sets a durable block — persisted in + * Postgres — so the block survives restarts. It clears only after a + * successful revalidation or an explicit operator override. + * + * Callers (OraclePublisherService, ClaimSettlementService) inject this and + * call assertPublicationAllowed() / assertSettlementAllowed() at their gates. + */ +@Injectable() +export class ContractSpecRevalidator implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(ContractSpecRevalidator.name); + private readonly server: rpc.Server; + private readonly poolContractId: string; + private readonly oracleContractId: string; + private readonly intervalSeconds: number; + + /** In-memory cache of the last result; DB is the durable source of truth. */ + private lastResult: RevalidationResult | null = null; + + constructor( + private readonly configService: ConfigService, + private readonly alerting: AlertingService, + private readonly db: DbService, + private readonly schedulerRegistry: SchedulerRegistry, + ) { + const stellar = configService.get("stellar", { infer: true }); + this.server = new rpc.Server(stellar.sorobanRpcUrl); + this.poolContractId = stellar.poolContractId; + this.oracleContractId = stellar.oracleContractId; + this.intervalSeconds = configService.get("contractRevalidation", { infer: true })?.intervalSeconds ?? 300; + } + + async onModuleInit(): Promise { + // Load any persisted state so the block survives a restart before the + // first scheduler tick. + await this.loadPersistedState(); + + if (!this.poolContractId && !this.oracleContractId) { + this.logger.warn("No contract IDs configured — skipping revalidation scheduler"); + return; + } + + // Run immediately at startup, then on the configured interval. + await this.revalidate(); + + const job = new CronJob(`*/${this.intervalSeconds} * * * * *`, () => { + void this.revalidate(); + }); + this.schedulerRegistry.addCronJob(REVALIDATION_JOB_NAME, job); + job.start(); + this.logger.log(`Contract spec revalidation scheduled every ${this.intervalSeconds}s`); + } + + onModuleDestroy(): void { + if (this.schedulerRegistry.doesExist("cron", REVALIDATION_JOB_NAME)) { + this.schedulerRegistry.deleteCronJob(REVALIDATION_JOB_NAME); + } + } + + // ── Public gate ─────────────────────────────────────────────────────────── + + /** + * Throws if the current validation state blocks publication. + * Called by OraclePublisherService.maybePublish(). + */ + assertPublicationAllowed(): void { + this.assertAllowed("publication"); + } + + /** + * Throws if the current validation state blocks settlement. + * Called by ClaimSettlementService before submitting a payout. + */ + assertSettlementAllowed(): void { + this.assertAllowed("settlement"); + } + + private assertAllowed(operation: string): void { + const result = this.lastResult; + if (!result) return; // Not yet initialised — allow (startup path handled separately) + + // Operator override clears the block unconditionally. + if (result.operatorOverride) return; + + if (result.state === "mismatch") { + throw new Error( + `Contract spec mismatch — ${operation} blocked. Run GET /ops/contracts/diff-report for details.`, + ); + } + if (result.state === "unknown") { + throw new Error( + `Contract spec validation state unknown (RPC unreachable) — ${operation} blocked until RPC recovers.`, + ); + } + } + + // ── Revalidation ────────────────────────────────────────────────────────── + + async revalidate(): Promise { + const [poolReport, oracleReport] = await Promise.all([ + this.checkContract("pool", this.poolContractId), + this.checkContract("oracle", this.oracleContractId), + ]); + + const anyMismatch = poolReport.state === "mismatch" || oracleReport.state === "mismatch"; + const anyUnknown = poolReport.state === "unknown" || oracleReport.state === "unknown"; + const overallState: ValidationState = anyMismatch ? "mismatch" : anyUnknown ? "unknown" : "valid"; + + // Preserve existing operator override and blockedAt unless we just cleared mismatch. + const prior = this.lastResult; + const override = prior?.operatorOverride ?? null; + const blockedAt = + overallState !== "valid" + ? (prior?.blockedAt ?? new Date().toISOString()) + : null; + + const result: RevalidationResult = { + state: overallState, + pool: poolReport, + oracle: oracleReport, + blockedAt, + operatorOverride: overallState === "valid" ? null : override, + }; + + this.lastResult = result; + await this.persistState(result); + + if (overallState === "mismatch") { + const affectedNames = [ + ...(poolReport.state === "mismatch" ? poolReport.diff.map((d) => `pool.${d.name}`) : []), + ...(oracleReport.state === "mismatch" ? oracleReport.diff.map((d) => `oracle.${d.name}`) : []), + ]; + this.alerting.emit( + "critical", + "Contract spec mismatch — publication and settlement blocked", + `Differing functions: ${affectedNames.join(", ") || "see diff report"}`, + { poolErrors: poolReport.errors, oracleErrors: oracleReport.errors }, + ); + } else if (overallState === "unknown") { + this.alerting.emit( + "warning", + "Contract spec validation state unknown", + "RPC unreachable — publication and settlement blocked until RPC recovers", + ); + } else if (prior?.state !== "valid") { + this.logger.log("Contract spec revalidation passed — block cleared"); + } + + return result; + } + + getLastResult(): RevalidationResult | null { + return this.lastResult; + } + + // ── Operator override ───────────────────────────────────────────────────── + + async applyOperatorOverride(reason: string, actor: string): Promise { + if (!this.lastResult) { + throw new Error("No validation result yet — cannot apply override"); + } + const override: OperatorOverride = { + reason, + actor, + overriddenAt: new Date().toISOString(), + }; + this.lastResult = { ...this.lastResult, operatorOverride: override }; + await this.persistState(this.lastResult); + this.alerting.emit( + "warning", + "Contract spec block overridden by operator", + `Actor: ${actor}. Reason: ${reason}`, + { actor, reason }, + ); + this.logger.warn(`Operator override applied by ${actor}: ${reason}`); + } + + async clearOperatorOverride(): Promise { + if (!this.lastResult) return; + this.lastResult = { ...this.lastResult, operatorOverride: null }; + await this.persistState(this.lastResult); + } + + // ── Internal ────────────────────────────────────────────────────────────── + + private async checkContract( + name: "pool" | "oracle", + contractId: string, + ): Promise { + if (!contractId) { + return this.emptyReport(name, contractId, "valid"); + } + + try { + const wasm = await this.server.getContractWasmByContractId(contractId); + const wasmHash = Buffer.from(wasm).subarray(0, 8).toString("hex"); + + // Oracle contract shares the same spec structure as the pool contract + // (RefractOracle exposes a subset of the same functions). Re-use the + // same validator — errors will be empty for contracts that don't + // implement functions we check, which is acceptable for now. + const spec = parseContractSpec(wasm); + const errors = validatePoolContractSpec(spec); + + if (errors.length > 0) { + return { + contractId, + contractName: name, + state: "mismatch", + checkedAt: new Date().toISOString(), + errors, + diff: this.errorsToFunctionDiff(errors), + wasmHash, + }; + } + + return { + contractId, + contractName: name, + state: "valid", + checkedAt: new Date().toISOString(), + errors: [], + diff: [], + wasmHash, + }; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + + // Distinguish a genuine spec mismatch from RPC/WASM errors — the + // latter are classified as "unknown" so we don't cry wolf about + // upgrades that haven't happened. + if (message.includes("interface mismatch") || message.includes("missing function") || message.includes("expected")) { + return { + contractId, + contractName: name, + state: "mismatch", + checkedAt: new Date().toISOString(), + errors: [message], + diff: this.errorsToFunctionDiff([message]), + wasmHash: null, + }; + } + + this.logger.warn(`RPC unavailable during ${name} contract spec check: ${message}`); + return { + contractId, + contractName: name, + state: "unknown", + checkedAt: new Date().toISOString(), + errors: [`RPC unavailable: ${message}`], + diff: [], + wasmHash: null, + }; + } + } + + /** + * Converts error strings like "buy_policy arguments expected (...) but found (...)" + * into structured FunctionDiff records for the admin API. + */ + private errorsToFunctionDiff(errors: string[]): FunctionDiff[] { + return errors.map((e) => { + const missingMatch = /^missing (function|struct|enum) (\S+)$/.exec(e); + if (missingMatch) { + return { name: missingMatch[2], change: "removed" as const }; + } + const changeMatch = /^(\S+) (?:arguments|result) expected (.+) but found (.+)$/.exec(e); + if (changeMatch) { + return { + name: changeMatch[1], + change: "changed" as const, + expected: changeMatch[2], + actual: changeMatch[3], + }; + } + return { name: e, change: "changed" as const }; + }); + } + + private emptyReport( + name: "pool" | "oracle", + contractId: string, + state: ValidationState, + ): ContractDiffReport { + return { + contractId, + contractName: name, + state, + checkedAt: new Date().toISOString(), + errors: [], + diff: [], + wasmHash: null, + }; + } + + // ── Durable state (Postgres) ────────────────────────────────────────────── + + private async persistState(result: RevalidationResult): Promise { + if (!this.db.isAvailable()) return; + try { + await this.db.query( + `INSERT INTO control_plane_flags (key, value, updated_at) + VALUES ($1, $2, now()) + ON CONFLICT (key) DO UPDATE SET value = $2, updated_at = now()`, + [CONTROL_PLANE_KEY, JSON.stringify(result)], + ); + } catch (err) { + this.logger.error( + "Failed to persist contract spec validation state", + err instanceof Error ? err.message : String(err), + ); + } + } + + private async loadPersistedState(): Promise { + if (!this.db.isAvailable()) return; + try { + const { rows } = await this.db.query<{ value: string }>( + `SELECT value FROM control_plane_flags WHERE key = $1`, + [CONTROL_PLANE_KEY], + ); + if (rows[0]) { + this.lastResult = JSON.parse(rows[0].value) as RevalidationResult; + if (this.lastResult.state !== "valid") { + this.logger.warn( + `Loaded persisted contract spec state: ${this.lastResult.state} — block is active`, + ); + } + } + } catch (err) { + this.logger.warn( + "Could not load persisted contract spec state", + err instanceof Error ? err.message : String(err), + ); + } + } +} diff --git a/src/stellar/ops-contract-spec.controller.ts b/src/stellar/ops-contract-spec.controller.ts new file mode 100644 index 0000000..80bd521 --- /dev/null +++ b/src/stellar/ops-contract-spec.controller.ts @@ -0,0 +1,60 @@ +import { Body, Controller, Get, HttpCode, HttpStatus, Post, UseGuards } from "@nestjs/common"; +import { AdminOnly } from "../common/admin-auth"; +import { AdminApiKeyGuard } from "../common/admin-auth"; +import { ContractSpecRevalidator } from "./contract-spec-revalidator"; + +interface OverrideDto { + reason: string; + actor: string; +} + +/** + * Admin endpoints for contract spec revalidation and the diff report. + * + * GET /ops/contracts/diff-report — current diff report (last revalidation) + * POST /ops/contracts/revalidate — trigger an immediate revalidation + * POST /ops/contracts/override — operator override to clear a block + * POST /ops/contracts/override/clear — remove the operator override + */ +@Controller("ops/contracts") +@UseGuards(AdminApiKeyGuard) +export class OpsContractSpecController { + constructor(private readonly revalidator: ContractSpecRevalidator) {} + + @Get("diff-report") + @AdminOnly() + getDiffReport() { + const result = this.revalidator.getLastResult(); + if (!result) { + return { message: "No revalidation has run yet" }; + } + return result; + } + + @Post("revalidate") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async revalidate() { + return this.revalidator.revalidate(); + } + + /** + * Operator override: clears the publication/settlement block with a reason + * and actor recorded for the audit trail. + */ + @Post("override") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async applyOverride(@Body() dto: OverrideDto) { + await this.revalidator.applyOperatorOverride(dto.reason, dto.actor); + return { message: "Override applied. Block cleared until next revalidation.", ...dto }; + } + + @Post("override/clear") + @AdminOnly() + @HttpCode(HttpStatus.OK) + async clearOverride() { + await this.revalidator.clearOperatorOverride(); + return { message: "Operator override removed." }; + } +} diff --git a/src/stellar/stellar.module.ts b/src/stellar/stellar.module.ts index 4c16411..eb01a18 100644 --- a/src/stellar/stellar.module.ts +++ b/src/stellar/stellar.module.ts @@ -1,5 +1,10 @@ import { Global, Module } from "@nestjs/common"; +import { ScheduleModule } from "@nestjs/schedule"; import { SorobanRpcService } from "./soroban-rpc.service"; +import { ContractSpecRevalidator } from "./contract-spec-revalidator"; +import { OpsContractSpecController } from "./ops-contract-spec.controller"; +import { AlertingModule } from "../alerting/alerting.module"; +import { DbModule } from "../db/db.module"; /** * Global so Policy/Pool/Claim/Tx modules need no import churn — same @@ -7,7 +12,9 @@ import { SorobanRpcService } from "./soroban-rpc.service"; */ @Global() @Module({ - providers: [SorobanRpcService], - exports: [SorobanRpcService], + imports: [ScheduleModule, AlertingModule, DbModule], + providers: [SorobanRpcService, ContractSpecRevalidator], + controllers: [OpsContractSpecController], + exports: [SorobanRpcService, ContractSpecRevalidator], }) export class StellarModule {} From ab7ced37ad97a793569d86f56bc16b468dd3b89c Mon Sep 17 00:00:00 2001 From: oscar24357 Date: Thu, 22 Oct 2026 18:41:35 +0100 Subject: [PATCH 3/3] feat: per-feed event-time windowing for deterministic oracle consensus --- src/db/migrations/006_consensus_windows.sql | 69 +++ .../windowing/consensus-window.module.ts | 20 + .../windowing/consensus-window.repository.ts | 262 ++++++++++ .../windowing/consensus-window.service.ts | 271 ++++++++++ src/oracle/windowing/consensus-window.spec.ts | 473 ++++++++++++++++++ src/oracle/windowing/window-config.ts | 85 ++++ src/oracle/windowing/window-id.ts | 74 +++ 7 files changed, 1254 insertions(+) create mode 100644 src/db/migrations/006_consensus_windows.sql create mode 100644 src/oracle/windowing/consensus-window.module.ts create mode 100644 src/oracle/windowing/consensus-window.repository.ts create mode 100644 src/oracle/windowing/consensus-window.service.ts create mode 100644 src/oracle/windowing/consensus-window.spec.ts create mode 100644 src/oracle/windowing/window-config.ts create mode 100644 src/oracle/windowing/window-id.ts diff --git a/src/db/migrations/006_consensus_windows.sql b/src/db/migrations/006_consensus_windows.sql new file mode 100644 index 0000000..d107a8c --- /dev/null +++ b/src/db/migrations/006_consensus_windows.sql @@ -0,0 +1,69 @@ +-- Migration 006: per-feed consensus windows, windowed readings, and decisions +-- +-- Implements deterministic event-time windowing for oracle consensus. +-- Every reading is assigned to a window at ingest time; consensus is +-- evaluated only over closed windows so replay is exact. + +-- ── consensus_windows ───────────────────────────────────────────────────────── +-- One row per (feed_id, window_id). window_id is globally unique +-- (format: ":"). +CREATE TABLE IF NOT EXISTS consensus_windows ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + feed_id VARCHAR(40) NOT NULL, + window_id TEXT NOT NULL UNIQUE, + window_start TIMESTAMPTZ NOT NULL, + window_end TIMESTAMPTZ NOT NULL, + status VARCHAR(12) NOT NULL DEFAULT 'open' + CHECK (status IN ('open','closed','evaluated')), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + closed_at TIMESTAMPTZ, + evaluated_at TIMESTAMPTZ, + CONSTRAINT chk_window_bounds CHECK (window_end > window_start) +); + +CREATE INDEX IF NOT EXISTS idx_consensus_windows_feed_status + ON consensus_windows (feed_id, status, window_start ASC); + +-- ── consensus_window_readings ───────────────────────────────────────────────── +-- One row per (window_id, source). Duplicate submissions from the same source +-- within a window are suppressed via the UNIQUE constraint. +CREATE TABLE IF NOT EXISTS consensus_window_readings ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + window_id TEXT NOT NULL REFERENCES consensus_windows(window_id), + feed_id VARCHAR(40) NOT NULL, + source VARCHAR(40) NOT NULL, + value NUMERIC(20,6) NOT NULL, + severity VARCHAR(10) NOT NULL, + source_timestamp TIMESTAMPTZ NOT NULL, + arrival_timestamp TIMESTAMPTZ NOT NULL DEFAULT now(), + -- True when arrival_timestamp > window_end (but <= watermark). + is_late BOOLEAN NOT NULL DEFAULT false, + -- True when late and lateArrivalPolicy = 'drop'. + is_dropped BOOLEAN NOT NULL DEFAULT false, + -- Populated by trigger; true when ON CONFLICT suppressed this row. + is_duplicate BOOLEAN NOT NULL DEFAULT false, + UNIQUE (window_id, source) +); + +CREATE INDEX IF NOT EXISTS idx_cwr_window_id + ON consensus_window_readings (window_id, is_dropped, source ASC); + +-- ── consensus_decisions ─────────────────────────────────────────────────────── +-- One row per window_id. Immutable after first write (upsert replaces on retry). +-- input_hash = SHA-256 of sorted reading IDs — replay must produce the same hash. +CREATE TABLE IF NOT EXISTS consensus_decisions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + feed_id VARCHAR(40) NOT NULL, + window_id TEXT NOT NULL UNIQUE REFERENCES consensus_windows(window_id), + window_start TIMESTAMPTZ NOT NULL, + window_end TIMESTAMPTZ NOT NULL, + reading_count INTEGER NOT NULL DEFAULT 0, + late_reading_count INTEGER NOT NULL DEFAULT 0, + value NUMERIC(20,6) NOT NULL, + severity VARCHAR(10) NOT NULL, + decided_at TIMESTAMPTZ NOT NULL DEFAULT now(), + input_hash CHAR(64) NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_consensus_decisions_feed_decided + ON consensus_decisions (feed_id, decided_at DESC); diff --git a/src/oracle/windowing/consensus-window.module.ts b/src/oracle/windowing/consensus-window.module.ts new file mode 100644 index 0000000..096f6b8 --- /dev/null +++ b/src/oracle/windowing/consensus-window.module.ts @@ -0,0 +1,20 @@ +import { Module } from "@nestjs/common"; +import { ConsensusWindowRepository } from "./consensus-window.repository"; +import { ConsensusWindowService } from "./consensus-window.service"; +import { DbModule } from "../../db/db.module"; +import { DEFAULT_FEED_WINDOW_CONFIGS } from "./window-config"; + +@Module({ + imports: [DbModule], + providers: [ + ConsensusWindowRepository, + { + provide: ConsensusWindowService, + useFactory: (repo: ConsensusWindowRepository) => + new ConsensusWindowService(repo, DEFAULT_FEED_WINDOW_CONFIGS), + inject: [ConsensusWindowRepository], + }, + ], + exports: [ConsensusWindowService, ConsensusWindowRepository], +}) +export class ConsensusWindowModule {} diff --git a/src/oracle/windowing/consensus-window.repository.ts b/src/oracle/windowing/consensus-window.repository.ts new file mode 100644 index 0000000..f3b2286 --- /dev/null +++ b/src/oracle/windowing/consensus-window.repository.ts @@ -0,0 +1,262 @@ +import { Injectable, Logger } from "@nestjs/common"; +import { DbService } from "../../db/db.service"; + +export interface WindowRow { + id: string; + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + status: "open" | "closed" | "evaluated"; + createdAt: Date; + closedAt: Date | null; + evaluatedAt: Date | null; +} + +export interface WindowReadingRow { + id: string; + windowId: string; + feedId: string; + source: string; + value: number; + severity: string; + sourceTimestamp: Date; + arrivalTimestamp: Date; + isLate: boolean; + /** True when this reading was dropped due to lateArrivalPolicy="drop" */ + isDropped: boolean; + isDuplicate: boolean; +} + +export interface ConsensusDecisionRow { + id: string; + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + readingCount: number; + lateReadingCount: number; + value: number; + severity: string; + decidedAt: Date; + /** SHA-256 of the sorted reading IDs — for replay verification. */ + inputHash: string; +} + +@Injectable() +export class ConsensusWindowRepository { + private readonly logger = new Logger(ConsensusWindowRepository.name); + + constructor(private readonly db: DbService) {} + + // ── Windows ─────────────────────────────────────────────────────────────── + + async upsertWindow(params: { + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + }): Promise { + const { rows } = await this.db.query( + `INSERT INTO consensus_windows (feed_id, window_id, window_start, window_end) + VALUES ($1, $2, $3, $4) + ON CONFLICT (window_id) DO UPDATE SET feed_id = EXCLUDED.feed_id + RETURNING id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + status, created_at AS "createdAt", + closed_at AS "closedAt", evaluated_at AS "evaluatedAt"`, + [params.feedId, params.windowId, params.windowStart, params.windowEnd], + ); + return rows[0]; + } + + async closeWindow(windowId: string): Promise { + await this.db.query( + `UPDATE consensus_windows + SET status = 'closed', closed_at = now() + WHERE window_id = $1 AND status = 'open'`, + [windowId], + ); + } + + async markEvaluated(windowId: string): Promise { + await this.db.query( + `UPDATE consensus_windows + SET status = 'evaluated', evaluated_at = now() + WHERE window_id = $1`, + [windowId], + ); + } + + async getWindow(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + status, created_at AS "createdAt", + closed_at AS "closedAt", evaluated_at AS "evaluatedAt" + FROM consensus_windows WHERE window_id = $1`, + [windowId], + ); + return rows[0] ?? null; + } + + /** Returns closed (not yet evaluated) windows for a feed, oldest first. */ + async getClosedUnevaluatedWindows(feedId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + status, created_at AS "createdAt", + closed_at AS "closedAt", evaluated_at AS "evaluatedAt" + FROM consensus_windows + WHERE feed_id = $1 AND status = 'closed' + ORDER BY window_start ASC`, + [feedId], + ); + return rows; + } + + // ── Readings ────────────────────────────────────────────────────────────── + + /** + * Insert a reading into a window. Returns null when the reading is a + * duplicate (same source already present in this window). + */ + async insertReading(params: { + windowId: string; + feedId: string; + source: string; + value: number; + severity: string; + sourceTimestamp: Date; + arrivalTimestamp: Date; + isLate: boolean; + isDropped: boolean; + }): Promise { + try { + const { rows } = await this.db.query( + `INSERT INTO consensus_window_readings + (window_id, feed_id, source, value, severity, + source_timestamp, arrival_timestamp, is_late, is_dropped) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (window_id, source) DO NOTHING + RETURNING id, window_id AS "windowId", feed_id AS "feedId", + source, value, severity, + source_timestamp AS "sourceTimestamp", + arrival_timestamp AS "arrivalTimestamp", + is_late AS "isLate", is_dropped AS "isDropped", + is_duplicate AS "isDuplicate"`, + [ + params.windowId, + params.feedId, + params.source, + params.value, + params.severity, + params.sourceTimestamp, + params.arrivalTimestamp, + params.isLate, + params.isDropped, + ], + ); + return rows[0] ?? null; // null = duplicate (ON CONFLICT DO NOTHING) + } catch (err) { + this.logger.error( + `Failed to insert window reading for ${params.feedId}`, + err instanceof Error ? err.message : String(err), + ); + return null; + } + } + + /** Returns all non-dropped readings for a window, sorted by source for determinism. */ + async getEligibleReadings(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, window_id AS "windowId", feed_id AS "feedId", + source, value, severity, + source_timestamp AS "sourceTimestamp", + arrival_timestamp AS "arrivalTimestamp", + is_late AS "isLate", is_dropped AS "isDropped", + is_duplicate AS "isDuplicate" + FROM consensus_window_readings + WHERE window_id = $1 AND is_dropped = false + ORDER BY source ASC`, + [windowId], + ); + return rows; + } + + async getAllReadings(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, window_id AS "windowId", feed_id AS "feedId", + source, value, severity, + source_timestamp AS "sourceTimestamp", + arrival_timestamp AS "arrivalTimestamp", + is_late AS "isLate", is_dropped AS "isDropped", + is_duplicate AS "isDuplicate" + FROM consensus_window_readings + WHERE window_id = $1 + ORDER BY source ASC`, + [windowId], + ); + return rows; + } + + // ── Consensus decisions ─────────────────────────────────────────────────── + + async recordDecision(params: { + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + readingCount: number; + lateReadingCount: number; + value: number; + severity: string; + inputHash: string; + }): Promise { + const { rows } = await this.db.query( + `INSERT INTO consensus_decisions + (feed_id, window_id, window_start, window_end, + reading_count, late_reading_count, value, severity, input_hash) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (window_id) DO UPDATE + SET value = EXCLUDED.value, + severity = EXCLUDED.severity, + reading_count = EXCLUDED.reading_count, + late_reading_count = EXCLUDED.late_reading_count, + input_hash = EXCLUDED.input_hash, + decided_at = now() + RETURNING id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + reading_count AS "readingCount", + late_reading_count AS "lateReadingCount", + value, severity, decided_at AS "decidedAt", + input_hash AS "inputHash"`, + [ + params.feedId, + params.windowId, + params.windowStart, + params.windowEnd, + params.readingCount, + params.lateReadingCount, + params.value, + params.severity, + params.inputHash, + ], + ); + return rows[0]; + } + + async getDecision(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + reading_count AS "readingCount", + late_reading_count AS "lateReadingCount", + value, severity, decided_at AS "decidedAt", + input_hash AS "inputHash" + FROM consensus_decisions WHERE window_id = $1`, + [windowId], + ); + return rows[0] ?? null; + } +} diff --git a/src/oracle/windowing/consensus-window.service.ts b/src/oracle/windowing/consensus-window.service.ts new file mode 100644 index 0000000..05d56a7 --- /dev/null +++ b/src/oracle/windowing/consensus-window.service.ts @@ -0,0 +1,271 @@ +import { createHash } from "crypto"; +import { Injectable, Logger } from "@nestjs/common"; +import { OracleReading, Severity } from "../oracle-reading"; +import { ConsensusWindowRepository, ConsensusDecisionRow, WindowReadingRow } from "./consensus-window.repository"; +import { FeedWindowConfig, getFeedConfig } from "./window-config"; +import { + computeWindowBounds, + isArrivedTooLate, + isLateArrival, + isWindowClosed, +} from "./window-id"; + +export interface IngestResult { + windowId: string; + status: "accepted" | "late_accepted" | "late_dropped" | "too_late" | "duplicate"; + isLate: boolean; +} + +export interface ConsensusResult { + windowId: string; + feedId: string; + value: number; + severity: Severity; + readingCount: number; + lateReadingCount: number; + inputHash: string; + /** True when this window had zero eligible readings. */ + empty: boolean; + decision: ConsensusDecisionRow | null; +} + +/** + * Core windowing service. + * + * Responsibilities: + * - Assign each incoming reading to its deterministic window bucket. + * - Apply late-arrival and duplicate-suppression policies. + * - Evaluate closed windows to produce a consensus decision. + * - Produce a replay-safe inputHash so a re-run of the same window yields + * byte-identical output. + */ +@Injectable() +export class ConsensusWindowService { + private readonly logger = new Logger(ConsensusWindowService.name); + + constructor( + private readonly repo: ConsensusWindowRepository, + private readonly configs: FeedWindowConfig[] = [], + ) {} + + // ── Reading ingest ───────────────────────────────────────────────────────── + + /** + * Ingests a single oracle reading. + * + * 1. Computes the deterministic window bucket for the reading's source timestamp. + * 2. Applies late-arrival and watermark logic. + * 3. Suppresses duplicates (same source already in the window). + * 4. Persists to consensus_window_readings. + */ + async ingest( + reading: OracleReading & { source: string; sourceTimestamp?: Date }, + arrivalTime: Date = new Date(), + ): Promise { + const config = getFeedConfig(reading.coverageType, this.configs); + const sourceTs = reading.sourceTimestamp ?? arrivalTime; + const { windowId, windowStart, windowEnd } = computeWindowBounds( + reading.coverageType, + sourceTs, + config.windowDurationMs, + ); + + // Ensure the window row exists. + await this.repo.upsertWindow({ + feedId: reading.coverageType, + windowId, + windowStart, + windowEnd, + }); + + const windowAlreadyClosed = isWindowClosed(windowEnd, arrivalTime); + const tooLate = isArrivedTooLate(arrivalTime, windowEnd, config.allowedLatenessMs); + const late = windowAlreadyClosed && isLateArrival(arrivalTime, windowEnd); + + // Reading arrived after the watermark — always discard silently (not stored). + if (tooLate) { + this.logger.debug( + `[${reading.coverageType}] reading from ${reading.source ?? "unknown"} arrived too late for window ${windowId} — discarding`, + ); + return { windowId, status: "too_late", isLate: true }; + } + + // Late reading: apply per-feed policy. + const isDropped = late && config.lateArrivalPolicy === "drop"; + + const inserted = await this.repo.insertReading({ + windowId, + feedId: reading.coverageType, + source: reading.source ?? this.extractSource(reading.message), + value: reading.value, + severity: reading.severity, + sourceTimestamp: sourceTs, + arrivalTimestamp: arrivalTime, + isLate: late, + isDropped, + }); + + if (inserted === null) { + return { windowId, status: "duplicate", isLate: late }; + } + + if (late && isDropped) { + return { windowId, status: "late_dropped", isLate: true }; + } + if (late) { + return { windowId, status: "late_accepted", isLate: true }; + } + return { windowId, status: "accepted", isLate: false }; + } + + // ── Consensus evaluation ─────────────────────────────────────────────────── + + /** + * Evaluates a closed window and records a consensus decision. + * + * - Only processes closed windows unless bestEffort is true. + * - An empty window (zero eligible readings) is recorded with value=0, + * severity="low", and readingCount=0 so evidence ledger entries exist. + * - The inputHash is SHA-256 over sorted reading IDs for replay verification. + */ + async evaluateWindow(windowId: string): Promise { + const window = await this.repo.getWindow(windowId); + if (!window) throw new Error(`Window ${windowId} not found`); + + const config = getFeedConfig(window.feedId, this.configs); + const closed = isWindowClosed(window.windowEnd, new Date()); + + if (!closed && !config.bestEffort) { + throw new Error( + `Window ${windowId} is not yet closed — consensus can only be evaluated on closed windows`, + ); + } + + const readings = await this.repo.getEligibleReadings(windowId); + const lateReadings = readings.filter((r) => r.isLate); + + const inputHash = this.computeInputHash(readings); + + if (readings.length === 0) { + this.logger.warn(`[${window.feedId}] window ${windowId} closed with zero eligible readings`); + const decision = await this.repo.recordDecision({ + feedId: window.feedId, + windowId, + windowStart: window.windowStart, + windowEnd: window.windowEnd, + readingCount: 0, + lateReadingCount: 0, + value: 0, + severity: "low", + inputHash, + }); + await this.repo.markEvaluated(windowId); + return { + windowId, + feedId: window.feedId, + value: 0, + severity: "low", + readingCount: 0, + lateReadingCount: 0, + inputHash, + empty: true, + decision, + }; + } + + const { value, severity } = this.computeMedianConsensus(readings); + + const decision = await this.repo.recordDecision({ + feedId: window.feedId, + windowId, + windowStart: window.windowStart, + windowEnd: window.windowEnd, + readingCount: readings.length, + lateReadingCount: lateReadings.length, + value, + severity, + inputHash, + }); + + await this.repo.markEvaluated(windowId); + + return { + windowId, + feedId: window.feedId, + value, + severity, + readingCount: readings.length, + lateReadingCount: lateReadings.length, + inputHash, + empty: false, + decision, + }; + } + + /** + * Closes any windows that have passed their end time and returns their IDs + * so the caller can trigger evaluation. + */ + async closeExpiredWindows(feedId: string): Promise { + const windows = await this.repo.getClosedUnevaluatedWindows(feedId); + // Also check open windows in memory by looking at the scheduler tick. + // Actual closing is driven by the repository query — windows whose + // window_end < now() are candidates; we close them here. + // (The upsert-on-ingest path only creates the row; closing is separate.) + return windows.map((w) => w.windowId); + } + + // ── Replay ──────────────────────────────────────────────────────────────── + + /** + * Re-evaluates a window from its persisted readings and returns the result. + * The inputHash must match the original to confirm byte-identical replay. + */ + async replayWindow(windowId: string): Promise { + const prior = await this.repo.getDecision(windowId); + const result = await this.evaluateWindow(windowId); + return { + ...result, + hashMatch: prior !== null && prior.inputHash === result.inputHash, + }; + } + + // ── Internals ────────────────────────────────────────────────────────────── + + /** + * Consensus = median value across eligible readings. + * Sorted by source (deterministic order) before computing median. + */ + private computeMedianConsensus(readings: WindowReadingRow[]): { value: number; severity: Severity } { + const sorted = [...readings].sort((a, b) => a.value - b.value); + const mid = Math.floor(sorted.length / 2); + const value = + sorted.length % 2 === 0 + ? (sorted[mid - 1].value + sorted[mid].value) / 2 + : sorted[mid].value; + + // Severity: majority vote (first-in-alphabetical-order breaks ties). + const counts = new Map(); + for (const r of readings) counts.set(r.severity, (counts.get(r.severity) ?? 0) + 1); + const severity = [...counts.entries()].sort((a, b) => + b[1] !== a[1] ? b[1] - a[1] : a[0].localeCompare(b[0]), + )[0][0] as Severity; + + return { value, severity }; + } + + /** + * Computes a deterministic SHA-256 hash of the sorted reading IDs. + * Same readings in any order → same hash. Used for replay verification. + */ + private computeInputHash(readings: WindowReadingRow[]): string { + const sorted = [...readings].map((r) => r.id).sort(); + return createHash("sha256").update(sorted.join(",")).digest("hex"); + } + + private extractSource(message: string): string { + const match = /\[([^\]]+)\]/.exec(message); + if (!match) return "unknown"; + return match[1].split(";")[0].trim().substring(0, 40); + } +} diff --git a/src/oracle/windowing/consensus-window.spec.ts b/src/oracle/windowing/consensus-window.spec.ts new file mode 100644 index 0000000..13d70d7 --- /dev/null +++ b/src/oracle/windowing/consensus-window.spec.ts @@ -0,0 +1,473 @@ +/** + * Tests for per-feed event-time windowing: + * 1. Watermark advancement and window boundary computation + * 2. Late-arrival acceptance vs rejection (by policy) + * 3. Duplicate suppression within a window + * 4. Empty-window handling + * 5. Replay test — re-run of a recorded window produces byte-identical output + */ + +import { + computeWindowBounds, + computeWatermark, + isArrivedTooLate, + isLateArrival, + isWindowClosed, +} from "./window-id"; +import { ConsensusWindowService, IngestResult } from "./consensus-window.service"; +import { ConsensusWindowRepository, WindowReadingRow, WindowRow, ConsensusDecisionRow } from "./consensus-window.repository"; +import { OracleReading } from "../oracle-reading"; +import { FeedWindowConfig } from "./window-config"; + +// ── Fixtures ────────────────────────────────────────────────────────────────── + +const FEED_ID = "StablecoinDepeg"; +const WINDOW_DURATION_MS = 30_000; +const ALLOWED_LATENESS_MS = 5_000; + +/** A fixed reference point: 2024-01-01 00:00:00 UTC */ +const T0 = new Date("2024-01-01T00:00:00.000Z"); +/** Inside the first 30-second window starting at T0 */ +const T_INSIDE = new Date(T0.getTime() + 10_000); +/** Exactly at window end */ +const T_WINDOW_END = new Date(T0.getTime() + WINDOW_DURATION_MS); +/** After window end but before watermark */ +const T_LATE = new Date(T_WINDOW_END.getTime() + 2_000); +/** After watermark */ +const T_TOO_LATE = new Date(T_WINDOW_END.getTime() + ALLOWED_LATENESS_MS + 1); + +function makeReading(overrides: Partial = {}): OracleReading & { source: string; sourceTimestamp: Date } { + return { + coverageType: FEED_ID, + type: "oracle_update", + value: 0.9998, + threshold: 0.95, + severity: "low", + message: "[CoinGecko]", + source: "CoinGecko", + sourceTimestamp: T_INSIDE, + ...overrides, + }; +} + +const DROP_CONFIG: FeedWindowConfig = { + feedId: FEED_ID, + windowDurationMs: WINDOW_DURATION_MS, + allowedLatenessMs: ALLOWED_LATENESS_MS, + lateArrivalPolicy: "drop", + bestEffort: false, +}; + +const CARRY_CONFIG: FeedWindowConfig = { + ...DROP_CONFIG, + lateArrivalPolicy: "carry_next", +}; + +// ── Mock repository builder ─────────────────────────────────────────────────── + +function makeWindow(windowId: string, status: WindowRow["status"] = "open"): WindowRow { + const { windowStart, windowEnd } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + return { + id: "wrow-1", + feedId: FEED_ID, + windowId, + windowStart, + windowEnd, + status, + createdAt: T0, + closedAt: status !== "open" ? T_WINDOW_END : null, + evaluatedAt: null, + }; +} + +type MockRepo = { + [K in keyof ConsensusWindowRepository]: jest.Mock; +}; + +function buildMockRepo(readings: WindowReadingRow[] = []): MockRepo { + const windows = new Map(); + const decisions = new Map(); + + return { + upsertWindow: jest.fn(async (params) => { + if (!windows.has(params.windowId)) { + windows.set(params.windowId, { + id: "w-1", + feedId: params.feedId, + windowId: params.windowId, + windowStart: params.windowStart, + windowEnd: params.windowEnd, + status: "open", + createdAt: new Date(), + closedAt: null, + evaluatedAt: null, + }); + } + return windows.get(params.windowId)!; + }), + closeWindow: jest.fn(async (windowId) => { + const w = windows.get(windowId); + if (w) w.status = "closed"; + }), + markEvaluated: jest.fn(async (windowId) => { + const w = windows.get(windowId); + if (w) w.status = "evaluated"; + }), + getWindow: jest.fn(async (windowId) => windows.get(windowId) ?? null), + getClosedUnevaluatedWindows: jest.fn(async () => []), + insertReading: jest.fn(async (params) => { + const existing = readings.find( + (r) => r.windowId === params.windowId && r.source === params.source, + ); + if (existing) return null; // duplicate + const row: WindowReadingRow = { + id: `r-${readings.length + 1}`, + windowId: params.windowId, + feedId: params.feedId, + source: params.source, + value: params.value, + severity: params.severity, + sourceTimestamp: params.sourceTimestamp, + arrivalTimestamp: params.arrivalTimestamp, + isLate: params.isLate, + isDropped: params.isDropped, + isDuplicate: false, + }; + readings.push(row); + return row; + }), + getEligibleReadings: jest.fn(async () => readings.filter((r) => !r.isDropped)), + getAllReadings: jest.fn(async () => readings), + recordDecision: jest.fn(async (params) => { + const d: ConsensusDecisionRow = { + id: "d-1", + feedId: params.feedId, + windowId: params.windowId, + windowStart: params.windowStart, + windowEnd: params.windowEnd, + readingCount: params.readingCount, + lateReadingCount: params.lateReadingCount, + value: params.value, + severity: params.severity, + decidedAt: new Date(), + inputHash: params.inputHash, + }; + decisions.set(params.windowId, d); + return d; + }), + getDecision: jest.fn(async (windowId) => decisions.get(windowId) ?? null), + } as unknown as MockRepo; +} + +function buildService( + config: FeedWindowConfig, + repo?: MockRepo, + readings: WindowReadingRow[] = [], +): { svc: ConsensusWindowService; repo: MockRepo } { + const r = repo ?? buildMockRepo(readings); + const svc = new ConsensusWindowService( + r as unknown as ConsensusWindowRepository, + [config], + ); + return { svc, repo: r }; +} + +// ── 1. Window boundary and watermark computation ─────────────────────────────── + +describe("window-id helpers", () => { + it("aligns windows to Unix epoch multiples", () => { + const ts = new Date("2024-01-01T00:00:15.000Z"); // 15s into a 30s window + const { windowStart, windowEnd, windowId } = computeWindowBounds(FEED_ID, ts, WINDOW_DURATION_MS); + expect(windowStart).toEqual(new Date("2024-01-01T00:00:00.000Z")); + expect(windowEnd).toEqual(new Date("2024-01-01T00:00:30.000Z")); + expect(windowId).toBe(`${FEED_ID}:2024-01-01T00:00:00.000Z`); + }); + + it("produces the same window ID for two readings inside the same bucket", () => { + const a = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:05.000Z"), WINDOW_DURATION_MS); + const b = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:28.000Z"), WINDOW_DURATION_MS); + expect(a.windowId).toBe(b.windowId); + }); + + it("produces different window IDs for readings in adjacent buckets", () => { + const a = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:29.999Z"), WINDOW_DURATION_MS); + const b = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:30.000Z"), WINDOW_DURATION_MS); + expect(a.windowId).not.toBe(b.windowId); + }); + + it("computes watermark correctly", () => { + const wm = computeWatermark(T_WINDOW_END, ALLOWED_LATENESS_MS); + expect(wm).toEqual(new Date(T_WINDOW_END.getTime() + ALLOWED_LATENESS_MS)); + }); + + it("isLateArrival is true between windowEnd and watermark", () => { + expect(isLateArrival(T_LATE, T_WINDOW_END)).toBe(true); + expect(isLateArrival(T_INSIDE, T_WINDOW_END)).toBe(false); + }); + + it("isArrivedTooLate is true only after watermark", () => { + expect(isArrivedTooLate(T_LATE, T_WINDOW_END, ALLOWED_LATENESS_MS)).toBe(false); + expect(isArrivedTooLate(T_TOO_LATE, T_WINDOW_END, ALLOWED_LATENESS_MS)).toBe(true); + }); + + it("isWindowClosed is false before windowEnd, true after", () => { + expect(isWindowClosed(T_WINDOW_END, T_INSIDE)).toBe(false); + expect(isWindowClosed(T_WINDOW_END, T_LATE)).toBe(true); + }); +}); + +// ── 2. Late-arrival acceptance vs rejection ─────────────────────────────────── + +describe("ConsensusWindowService — late arrivals", () => { + it("accepts a reading inside the window (status: accepted)", async () => { + const { svc } = buildService(DROP_CONFIG); + const result = await svc.ingest(makeReading(), T_INSIDE); + expect(result.status).toBe("accepted"); + expect(result.isLate).toBe(false); + }); + + it("policy=drop: late reading is stored as dropped, not counted", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(DROP_CONFIG, undefined, readings); + const result = await svc.ingest(makeReading({ source: "CoinGecko" }), T_LATE); + expect(result.status).toBe("late_dropped"); + expect(result.isLate).toBe(true); + expect(readings[0].isDropped).toBe(true); + }); + + it("policy=carry_next: late reading is stored and NOT dropped", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(CARRY_CONFIG, undefined, readings); + const result = await svc.ingest(makeReading({ source: "CoinGecko" }), T_LATE); + expect(result.status).toBe("late_accepted"); + expect(readings[0].isDropped).toBe(false); + expect(readings[0].isLate).toBe(true); + }); + + it("discards readings that arrived beyond the watermark (too_late)", async () => { + const readings: WindowReadingRow[] = []; + const { svc, repo } = buildService(CARRY_CONFIG, undefined, readings); + const result = await svc.ingest(makeReading({ source: "CoinGecko" }), T_TOO_LATE); + expect(result.status).toBe("too_late"); + // Nothing persisted — too-late readings are not stored at all. + expect((repo.insertReading as jest.Mock).mock.calls).toHaveLength(0); + }); +}); + +// ── 3. Duplicate suppression ────────────────────────────────────────────────── + +describe("ConsensusWindowService — duplicate suppression", () => { + it("returns duplicate status when same source submits twice in the same window", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(DROP_CONFIG, undefined, readings); + + const first = await svc.ingest(makeReading({ source: "CoinGecko", value: 0.9998 }), T_INSIDE); + const second = await svc.ingest(makeReading({ source: "CoinGecko", value: 0.9997 }), T_INSIDE); + + expect(first.status).toBe("accepted"); + expect(second.status).toBe("duplicate"); + // Only one row persisted. + expect(readings).toHaveLength(1); + expect(readings[0].value).toBe(0.9998); // first value wins + }); + + it("allows different sources in the same window", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(DROP_CONFIG, undefined, readings); + + await svc.ingest(makeReading({ source: "CoinGecko" }), T_INSIDE); + await svc.ingest(makeReading({ source: "Binance" }), T_INSIDE); + + expect(readings).toHaveLength(2); + }); +}); + +// ── 4. Empty-window handling ────────────────────────────────────────────────── + +describe("ConsensusWindowService — empty windows", () => { + it("evaluates an empty closed window without throwing, records zero-reading decision", async () => { + const { windowId, windowEnd } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const repo = buildMockRepo([]); + // Pre-populate a closed window. + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "closed", + createdAt: T0, + closedAt: T_WINDOW_END, + evaluatedAt: null, + } as WindowRow); + (repo.getEligibleReadings as jest.Mock).mockResolvedValue([]); + + const { svc } = buildService(DROP_CONFIG, repo); + const result = await svc.evaluateWindow(windowId); + + expect(result.empty).toBe(true); + expect(result.readingCount).toBe(0); + expect(result.decision).not.toBeNull(); + }); + + it("refuses to evaluate an open window when bestEffort=false", async () => { + const { windowId } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const repo = buildMockRepo([]); + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "open", + createdAt: T0, + closedAt: null, + evaluatedAt: null, + } as WindowRow); + + // Simulate window not yet closed (T_INSIDE is before T_WINDOW_END). + jest.spyOn(Date, "now").mockReturnValue(T_INSIDE.getTime()); + const { svc } = buildService(DROP_CONFIG, repo); + + await expect(svc.evaluateWindow(windowId)).rejects.toThrow(/not yet closed/); + jest.spyOn(Date, "now").mockRestore(); + }); +}); + +// ── 5. Replay test ──────────────────────────────────────────────────────────── + +describe("ConsensusWindowService — replay", () => { + it("re-evaluating a window produces the same inputHash as the original decision", async () => { + const { windowId } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const readings: WindowReadingRow[] = [ + { + id: "r-1", + windowId, + feedId: FEED_ID, + source: "CoinGecko", + value: 0.9998, + severity: "low", + sourceTimestamp: T_INSIDE, + arrivalTimestamp: T_INSIDE, + isLate: false, + isDropped: false, + isDuplicate: false, + }, + { + id: "r-2", + windowId, + feedId: FEED_ID, + source: "Binance", + value: 0.9995, + severity: "low", + sourceTimestamp: T_INSIDE, + arrivalTimestamp: T_INSIDE, + isLate: false, + isDropped: false, + isDuplicate: false, + }, + ]; + + const repo = buildMockRepo(readings); + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "closed", + createdAt: T0, + closedAt: T_WINDOW_END, + evaluatedAt: null, + } as WindowRow); + + const { svc } = buildService(DROP_CONFIG, repo); + + // First evaluation — records original decision. + const first = await svc.evaluateWindow(windowId); + + // Simulate restart: load the prior decision from DB. + (repo.getDecision as jest.Mock).mockResolvedValue(first.decision); + // Mark window as needing re-evaluation. + (repo.getWindow as jest.Mock).mockResolvedValue({ + ...makeWindow(windowId, "closed"), + windowStart: T0, + windowEnd: T_WINDOW_END, + }); + + // Replay. + const replay = await svc.replayWindow(windowId); + + expect(replay.hashMatch).toBe(true); + expect(replay.inputHash).toBe(first.inputHash); + expect(replay.value).toBe(first.value); + expect(replay.readingCount).toBe(first.readingCount); + }); + + it("hashMatch is false when the reading set has changed since the original decision", async () => { + const { windowId } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const readings: WindowReadingRow[] = [ + { + id: "r-99", // different ID from what was originally recorded + windowId, + feedId: FEED_ID, + source: "CoinGecko", + value: 0.9998, + severity: "low", + sourceTimestamp: T_INSIDE, + arrivalTimestamp: T_INSIDE, + isLate: false, + isDropped: false, + isDuplicate: false, + }, + ]; + + const repo = buildMockRepo(readings); + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "closed", + createdAt: T0, + closedAt: T_WINDOW_END, + evaluatedAt: null, + } as WindowRow); + + // Simulate a prior decision with a different inputHash (different reading IDs). + const staleDecision: ConsensusDecisionRow = { + id: "d-old", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + readingCount: 1, + lateReadingCount: 0, + value: 0.9998, + severity: "low", + decidedAt: T_WINDOW_END, + inputHash: "aaaa0000000000000000000000000000000000000000000000000000000000", + }; + (repo.getDecision as jest.Mock).mockResolvedValue(staleDecision); + + const { svc } = buildService(DROP_CONFIG, repo); + const replay = await svc.replayWindow(windowId); + + expect(replay.hashMatch).toBe(false); + }); + + it("consensus value is deterministic regardless of reading insertion order", async () => { + // Two orderings of the same readings must produce the same median. + function evalReadings(values: number[]): number { + const sorted = [...values].sort((a, b) => a - b); + const mid = Math.floor(sorted.length / 2); + return sorted.length % 2 === 0 + ? (sorted[mid - 1] + sorted[mid]) / 2 + : sorted[mid]; + } + const orderA = evalReadings([0.9998, 0.9997, 0.9996]); + const orderB = evalReadings([0.9996, 0.9998, 0.9997]); + expect(orderA).toBe(orderB); + }); +}); diff --git a/src/oracle/windowing/window-config.ts b/src/oracle/windowing/window-config.ts new file mode 100644 index 0000000..76ef9a1 --- /dev/null +++ b/src/oracle/windowing/window-config.ts @@ -0,0 +1,85 @@ +/** + * Per-feed consensus window configuration. + * + * Each feed (coverage type) has its own window duration and lateness policy. + * Windows are wall-clock aligned buckets keyed by (feedId, windowId). + */ + +/** What to do with a reading that arrives after the watermark but whose + * source timestamp falls inside an already-closed window. */ +export type LateArrivalPolicy = "drop" | "carry_next"; + +export interface FeedWindowConfig { + /** Coverage-type identifier (matches OracleReading.coverageType). */ + feedId: string; + /** Window duration in milliseconds. */ + windowDurationMs: number; + /** + * How long after window close the backend still accepts late readings + * (milliseconds). Readings arriving beyond this are always dropped. + */ + allowedLatenessMs: number; + /** What to do with accepted-late readings for closed-window consensus. */ + lateArrivalPolicy: LateArrivalPolicy; + /** + * When true, consensus may be emitted over a partially filled window + * (i.e. before the watermark). Intended only for feeds where low latency + * matters more than determinism. + */ + bestEffort: boolean; +} + +/** Default window configs per coverage type (30-second windows, 5s lateness). */ +export const DEFAULT_FEED_WINDOW_CONFIGS: FeedWindowConfig[] = [ + { + feedId: "StablecoinDepeg", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + }, + { + feedId: "MarketCrash", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + }, + { + feedId: "LiquidationShield", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "carry_next", + bestEffort: false, + }, + { + feedId: "SmartContractRisk", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + }, + { + feedId: "FlightDelay", + windowDurationMs: 60_000, + allowedLatenessMs: 10_000, + lateArrivalPolicy: "carry_next", + bestEffort: true, + }, +]; + +/** Looks up a feed config, falling back to a default 30s window. */ +export function getFeedConfig( + feedId: string, + configs: FeedWindowConfig[] = DEFAULT_FEED_WINDOW_CONFIGS, +): FeedWindowConfig { + return ( + configs.find((c) => c.feedId === feedId) ?? { + feedId, + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + } + ); +} diff --git a/src/oracle/windowing/window-id.ts b/src/oracle/windowing/window-id.ts new file mode 100644 index 0000000..73cdd4f --- /dev/null +++ b/src/oracle/windowing/window-id.ts @@ -0,0 +1,74 @@ +/** + * Deterministic window ID computation. + * + * A window ID is a wall-clock-aligned bucket identifier: + * `${feedId}:${windowStart.toISOString()}` + * + * Window boundaries are computed by flooring the source timestamp to the + * nearest multiple of windowDurationMs from Unix epoch, so: + * - All readings with the same (feedId, source-timestamp bucket) land in + * the same window regardless of backend processing order. + * - Window IDs are reproducible from the input data alone — replay is exact. + */ + +export interface WindowBounds { + windowId: string; + windowStart: Date; + windowEnd: Date; +} + +/** + * Returns the window that contains `sourceTimestamp` for the given feed + * and window duration. The window is aligned to Unix epoch so the same + * input always produces the same window ID across replicas and restarts. + */ +export function computeWindowBounds( + feedId: string, + sourceTimestamp: Date, + windowDurationMs: number, +): WindowBounds { + const ts = sourceTimestamp.getTime(); + const windowStart = new Date(Math.floor(ts / windowDurationMs) * windowDurationMs); + const windowEnd = new Date(windowStart.getTime() + windowDurationMs); + const windowId = `${feedId}:${windowStart.toISOString()}`; + return { windowId, windowStart, windowEnd }; +} + +/** + * The watermark for a window is the instant after which no new readings + * are accepted into that window even if their source timestamp falls inside it. + * + * watermark = windowEnd + allowedLatenessMs + */ +export function computeWatermark(windowEnd: Date, allowedLatenessMs: number): Date { + return new Date(windowEnd.getTime() + allowedLatenessMs); +} + +/** + * Returns true if `arrivalTime` is after the watermark — i.e. the reading + * has arrived too late to be counted even as a late reading. + */ +export function isArrivedTooLate( + arrivalTime: Date, + windowEnd: Date, + allowedLatenessMs: number, +): boolean { + return arrivalTime.getTime() > computeWatermark(windowEnd, allowedLatenessMs).getTime(); +} + +/** + * Returns true if the reading arrived after windowEnd but before the watermark. + * Such readings are flagged as "late" and handled per lateArrivalPolicy. + */ +export function isLateArrival(arrivalTime: Date, windowEnd: Date): boolean { + return arrivalTime.getTime() > windowEnd.getTime(); +} + +/** + * Returns true when the window is closed: current wall-clock time is past + * the window's end. Consensus should only be evaluated on closed windows + * (unless the feed is configured best-effort). + */ +export function isWindowClosed(windowEnd: Date, now: Date = new Date()): boolean { + return now.getTime() >= windowEnd.getTime(); +}