diff --git a/backend/src/api/routes/backfill.routes.ts b/backend/src/api/routes/backfill.routes.ts new file mode 100644 index 00000000..88af883b --- /dev/null +++ b/backend/src/api/routes/backfill.routes.ts @@ -0,0 +1,95 @@ +import type { FastifyInstance } from "fastify"; +import { backfillService } from "../../services/backfill.service.js"; +import { logger } from "../../utils/logger.js"; + +export async function backfillRoutes(server: FastifyInstance) { + server.post<{ Params: { sourceId: string } }>( + "/:sourceId/start", + { + schema: { + tags: ["Backfill"], + summary: "Start a backfill for a source", + params: { + type: "object", + required: ["sourceId"], + properties: { sourceId: { type: "string" } }, + }, + body: { + type: "object", + required: ["rangeStart", "rangeEnd", "chunkSize"], + properties: { + rangeStart: { type: "integer" }, + rangeEnd: { type: "integer" }, + chunkSize: { type: "integer", minimum: 1 }, + } + }, + response: { 200: { type: "object", additionalProperties: true } }, + }, + }, + async (request, reply) => { + const { sourceId } = request.params; + const config = request.body as any; + + try { + const jobId = await backfillService.startBackfillForSource(sourceId, config, { + processChunk: async (chunk) => { + // Placeholder: actual processing logic would be injected or handled here + // e.g. await fetchHistoricalDataForChunk(sourceId, chunk); + await new Promise(r => setTimeout(r, 100)); // mock work + } + }); + return { success: true, jobId }; + } catch (err: any) { + return reply.code(400).send({ error: err.message }); + } + } + ); + + server.post<{ Params: { sourceId: string } }>( + "/:sourceId/stop", + { + schema: { + tags: ["Backfill"], + summary: "Stop a backfill for a source", + params: { + type: "object", + required: ["sourceId"], + properties: { sourceId: { type: "string" } }, + }, + response: { 200: { type: "object", additionalProperties: true } }, + }, + }, + async (request, reply) => { + const { sourceId } = request.params; + try { + await backfillService.stopBackfillForSource(sourceId); + return { success: true }; + } catch (err: any) { + return reply.code(400).send({ error: err.message }); + } + } + ); + + server.get<{ Params: { sourceId: string } }>( + "/:sourceId/status", + { + schema: { + tags: ["Backfill"], + summary: "Get backfill status for a source", + params: { + type: "object", + required: ["sourceId"], + properties: { sourceId: { type: "string" } }, + }, + response: { 200: { type: "object", additionalProperties: true } }, + }, + }, + async (request, reply) => { + const status = await backfillService.getBackfillStatus(request.params.sourceId); + if (!status) { + return reply.code(404).send({ error: "No backfill job found for source" }); + } + return status; + } + ); +} diff --git a/backend/src/api/routes/index.ts b/backend/src/api/routes/index.ts index 384aff8a..8a52bcf4 100644 --- a/backend/src/api/routes/index.ts +++ b/backend/src/api/routes/index.ts @@ -19,6 +19,8 @@ import { registerCompatibilityRoutes } from "./route-groups/compatibility-routes import { registerOperationalRoutes } from "./route-groups/operational-routes.js"; import { registerOperationalMonitoringRoutes } from "./route-groups/operational-monitoring-routes.js"; import { registerLiquidityRoutes } from "./route-groups/liquidity-routes.js"; +import { sorobanEventsRoutes } from "./sorobanEvents.routes.js"; +import { backfillRoutes } from "./backfill.routes.js"; export async function registerRoutes(server: FastifyInstance): Promise { // Core routes: health, websocket, config, preferences, caching @@ -80,4 +82,7 @@ export async function registerRoutes(server: FastifyInstance): Promise { // Operational routes: query baseline, rollback readiness, canary metrics, promotion gates, risk clustering await registerOperationalRoutes(server); + + server.register(sorobanEventsRoutes, { prefix: "/api/v1/soroban-events" }); + server.register(backfillRoutes, { prefix: "/api/v1/backfill" }); } diff --git a/backend/src/api/routes/sorobanEvents.routes.ts b/backend/src/api/routes/sorobanEvents.routes.ts new file mode 100644 index 00000000..61a89ad3 --- /dev/null +++ b/backend/src/api/routes/sorobanEvents.routes.ts @@ -0,0 +1,62 @@ +import type { FastifyInstance } from "fastify"; +import { sorobanEventIndexService } from "../../services/sorobanEventIndex.service.js"; + +export async function sorobanEventsRoutes(server: FastifyInstance) { + server.get( + "/", + { + schema: { + tags: ["Soroban Events"], + summary: "List paginated Soroban events", + querystring: { + type: "object", + additionalProperties: false, + properties: { + contractId: { type: "string" }, + limit: { type: "integer", minimum: 1, maximum: 500, default: 100 }, + cursor: { type: "string" }, + }, + }, + response: { 200: { type: "object", additionalProperties: true } }, + }, + }, + async (request) => { + const q = request.query as { + contractId?: string; + limit?: number; + cursor?: string; + }; + + return sorobanEventIndexService.getPaginatedEvents( + q.contractId, + q.limit ?? 100, + q.cursor + ); + } + ); + + server.post( + "/sync", + { + schema: { + tags: ["Soroban Events"], + summary: "Trigger a manual sync of Soroban events", + body: { + type: "object", + required: ["contractId"], + additionalProperties: false, + properties: { + contractId: { type: "string" }, + limit: { type: "integer", minimum: 1, maximum: 10000, default: 1000 }, + }, + }, + response: { 200: { type: "object", additionalProperties: true } }, + }, + }, + async (request) => { + const body = request.body as { contractId: string; limit?: number }; + const syncedCount = await sorobanEventIndexService.syncEvents(body.contractId, body.limit); + return { syncedCount }; + } + ); +} diff --git a/backend/src/database/migrations/042_health_score_confidence.ts b/backend/src/database/migrations/042_health_score_confidence.ts new file mode 100644 index 00000000..1c5b5462 --- /dev/null +++ b/backend/src/database/migrations/042_health_score_confidence.ts @@ -0,0 +1,25 @@ +import type { Knex } from "knex"; + +export async function up(knex: Knex): Promise { + await knex.schema.alterTable("source_health_scores", (table) => { + table.integer("confidence_score"); + table.string("confidence_band"); + }); + + await knex.schema.alterTable("source_health_score_history", (table) => { + table.integer("confidence_score"); + table.string("confidence_band"); + }); +} + +export async function down(knex: Knex): Promise { + await knex.schema.alterTable("source_health_score_history", (table) => { + table.dropColumn("confidence_band"); + table.dropColumn("confidence_score"); + }); + + await knex.schema.alterTable("source_health_scores", (table) => { + table.dropColumn("confidence_band"); + table.dropColumn("confidence_score"); + }); +} diff --git a/backend/src/database/migrations/043_soroban_event_index.ts b/backend/src/database/migrations/043_soroban_event_index.ts new file mode 100644 index 00000000..c9261ada --- /dev/null +++ b/backend/src/database/migrations/043_soroban_event_index.ts @@ -0,0 +1,22 @@ +import type { Knex } from "knex"; + +export async function up(knex: Knex): Promise { + await knex.schema.createTable("soroban_events", (table) => { + table.uuid("id").primary().defaultTo(knex.raw("gen_random_uuid()")); + table.string("cursor").notNullable().unique(); + table.integer("ledger").notNullable(); + table.timestamp("ledger_closed_at").notNullable(); + table.string("contract_id").notNullable(); + table.string("topic").notNullable(); + table.jsonb("value").notNullable(); + table.timestamp("created_at").defaultTo(knex.fn.now()); + }); + + await knex.raw("CREATE INDEX soroban_events_contract_idx ON soroban_events(contract_id, ledger_closed_at DESC);"); + await knex.raw("SELECT create_hypertable('soroban_events', 'ledger_closed_at', if_not_exists => TRUE);"); + await knex.raw("SELECT add_retention_policy('soroban_events', INTERVAL '90 days', if_not_exists => TRUE);"); +} + +export async function down(knex: Knex): Promise { + await knex.schema.dropTableIfExists("soroban_events"); +} diff --git a/backend/src/database/migrations/044_per_source_backfill.ts b/backend/src/database/migrations/044_per_source_backfill.ts new file mode 100644 index 00000000..50129f0c --- /dev/null +++ b/backend/src/database/migrations/044_per_source_backfill.ts @@ -0,0 +1,24 @@ +import type { Knex } from "knex"; + +export async function up(knex: Knex): Promise { + await knex.schema.createTable("backfill_jobs", (table) => { + table.uuid("id").primary().defaultTo(knex.raw("gen_random_uuid()")); + table.string("source_id").notNullable(); + table.string("status").notNullable().defaultTo("PENDING"); // PENDING, RUNNING, COMPLETED, FAILED + table.integer("range_start").notNullable(); + table.integer("range_end").notNullable(); + table.integer("chunk_size").notNullable(); + table.jsonb("completed_chunks").defaultTo("[]"); + table.jsonb("failed_chunks").defaultTo("[]"); + table.timestamp("started_at"); + table.timestamp("completed_at"); + table.timestamp("created_at").defaultTo(knex.fn.now()); + table.timestamp("updated_at").defaultTo(knex.fn.now()); + }); + + await knex.raw("CREATE INDEX backfill_jobs_source_idx ON backfill_jobs(source_id);"); +} + +export async function down(knex: Knex): Promise { + await knex.schema.dropTableIfExists("backfill_jobs"); +} diff --git a/backend/src/database/models/backfillJob.model.ts b/backend/src/database/models/backfillJob.model.ts new file mode 100644 index 00000000..6e5fae3c --- /dev/null +++ b/backend/src/database/models/backfillJob.model.ts @@ -0,0 +1,44 @@ +import { getDatabase } from "../connection.js"; + +export interface BackfillJobRecord { + id?: string; + source_id: string; + status: "PENDING" | "RUNNING" | "COMPLETED" | "FAILED" | "STOPPED"; + range_start: number; + range_end: number; + chunk_size: number; + completed_chunks: string; // JSON + failed_chunks: string; // JSON + started_at?: Date; + completed_at?: Date; + created_at?: Date; + updated_at?: Date; +} + +export class BackfillJobModel { + private db = getDatabase(); + private table = "backfill_jobs"; + + async create(job: Partial): Promise { + const [inserted] = await this.db(this.table).insert(job).returning("*"); + return inserted as BackfillJobRecord; + } + + async updateStatus(id: string, status: BackfillJobRecord["status"], details?: Partial): Promise { + const updateData: any = { status, updated_at: new Date(), ...details }; + if (status === "RUNNING") updateData.started_at = new Date(); + if (status === "COMPLETED" || status === "FAILED" || status === "STOPPED") updateData.completed_at = new Date(); + + await this.db(this.table).where("id", id).update(updateData); + } + + async getLatestForSource(sourceId: string): Promise { + const record = await this.db(this.table) + .where("source_id", sourceId) + .orderBy("created_at", "desc") + .first(); + return record as BackfillJobRecord | undefined; + } +} + +export const backfillJobModel = new BackfillJobModel(); diff --git a/backend/src/database/models/healthScore.model.ts b/backend/src/database/models/healthScore.model.ts index 3e1f33f0..af6ddb38 100644 --- a/backend/src/database/models/healthScore.model.ts +++ b/backend/src/database/models/healthScore.model.ts @@ -9,6 +9,8 @@ export interface HealthScoreRecord { bridge_uptime_score: number; reserve_backing_score: number; volume_trend_score: number; + confidence_score?: number; + confidence_band?: string; } export class HealthScoreModel { diff --git a/backend/src/database/models/sorobanEvent.model.ts b/backend/src/database/models/sorobanEvent.model.ts new file mode 100644 index 00000000..71d3d93d --- /dev/null +++ b/backend/src/database/models/sorobanEvent.model.ts @@ -0,0 +1,63 @@ +import { getDatabase } from "../connection.js"; + +export interface SorobanEventRecord { + id?: string; + cursor: string; + ledger: number; + ledger_closed_at: Date; + contract_id: string; + topic: string; + value: any; + created_at?: Date; +} + +export class SorobanEventModel { + private db = getDatabase(); + private table = "soroban_events"; + + async insert(events: SorobanEventRecord[]): Promise { + if (events.length === 0) return; + await this.db(this.table) + .insert(events) + .onConflict("cursor") + .ignore(); // cursor is unique + } + + async getLatestCursor(contractId: string): Promise { + const record = await this.db(this.table) + .where("contract_id", contractId) + .orderBy("ledger_closed_at", "desc") + .first("cursor"); + return record?.cursor; + } + + async getPaginatedEvents( + contractId?: string, + limit = 100, + cursor?: string + ): Promise<{ data: SorobanEventRecord[]; nextCursor?: string }> { + let query = this.db(this.table).orderBy("ledger_closed_at", "desc").limit(limit); + + if (contractId) { + query = query.where("contract_id", contractId); + } + + if (cursor) { + // Find the event with this cursor to get its ledger_closed_at + const cursorEvent = await this.db(this.table).where("cursor", cursor).first(); + if (cursorEvent) { + query = query.where("ledger_closed_at", "<=", cursorEvent.ledger_closed_at).andWhereNot("cursor", cursor); + } + } + + const rows = await query; + const nextCursor = rows.length === limit ? rows[rows.length - 1].cursor : undefined; + + return { + data: rows as SorobanEventRecord[], + nextCursor, + }; + } +} + +export const sorobanEventModel = new SorobanEventModel(); diff --git a/backend/src/database/schema.sql b/backend/src/database/schema.sql index 617c1c0e..37964100 100644 --- a/backend/src/database/schema.sql +++ b/backend/src/database/schema.sql @@ -226,7 +226,9 @@ CREATE TABLE health_scores ( price_stability_score SMALLINT NOT NULL, bridge_uptime_score SMALLINT NOT NULL, reserve_backing_score SMALLINT NOT NULL, - volume_trend_score SMALLINT NOT NULL + volume_trend_score SMALLINT NOT NULL, + confidence_score SMALLINT, + confidence_band TEXT ); CREATE INDEX health_scores_symbol_time_idx ON health_scores (symbol, time DESC); SELECT create_hypertable('health_scores', 'time', if_not_exists => TRUE); diff --git a/backend/src/services/backfill.service.ts b/backend/src/services/backfill.service.ts new file mode 100644 index 00000000..0067c927 --- /dev/null +++ b/backend/src/services/backfill.service.ts @@ -0,0 +1,105 @@ +import { backfillJobModel } from "../database/models/backfillJob.model.js"; +import { runBackfill, type BackfillJobConfig, type BackfillDeps } from "./backfillOrchestrator.js"; +import { logger } from "../utils/logger.js"; + +export class BackfillService { + private activeJobs = new Map void }>(); + + async startBackfillForSource(sourceId: string, config: Omit, deps: Omit): Promise { + if (this.activeJobs.has(sourceId)) { + throw new Error(`A backfill is already running for source ${sourceId}`); + } + + const previousJob = await backfillJobModel.getLatestForSource(sourceId); + let completedChunks: number[] = []; + + if (previousJob && (previousJob.status === "STOPPED" || previousJob.status === "FAILED")) { + try { + completedChunks = JSON.parse(previousJob.completed_chunks); + } catch { + completedChunks = []; + } + } + + const job = await backfillJobModel.create({ + source_id: sourceId, + status: "PENDING", + range_start: config.rangeStart, + range_end: config.rangeEnd, + chunk_size: config.chunkSize, + completed_chunks: JSON.stringify(completedChunks), + failed_chunks: "[]", + }); + + const jobId = job.id!; + + let abortFlag = false; + this.activeJobs.set(sourceId, { + abort: () => { abortFlag = true; } + }); + + // Run asynchronously + setImmediate(async () => { + try { + await backfillJobModel.updateStatus(jobId, "RUNNING"); + + const fullDeps: BackfillDeps = { + ...deps, + processChunk: async (chunk) => { + if (abortFlag) throw new Error("AbortRequested"); + await deps.processChunk(chunk); + }, + onEvent: (event) => { + // Ideally we could persist progress here if needed + } + }; + + const result = await runBackfill({ ...config, completedChunks }, fullDeps); + + if (abortFlag) { + await backfillJobModel.updateStatus(jobId, "STOPPED", { + completed_chunks: JSON.stringify(result.completedChunks), + failed_chunks: JSON.stringify(result.failedChunks), + }); + } else { + await backfillJobModel.updateStatus(jobId, result.failedChunks.length > 0 ? "FAILED" : "COMPLETED", { + completed_chunks: JSON.stringify(result.completedChunks), + failed_chunks: JSON.stringify(result.failedChunks), + }); + } + } catch (err) { + logger.error({ sourceId, jobId, err }, "Backfill crashed completely"); + await backfillJobModel.updateStatus(jobId, "FAILED"); + } finally { + this.activeJobs.delete(sourceId); + } + }); + + return jobId; + } + + async stopBackfillForSource(sourceId: string): Promise { + const active = this.activeJobs.get(sourceId); + if (!active) { + throw new Error(`No active backfill for source ${sourceId}`); + } + active.abort(); + } + + async getBackfillStatus(sourceId: string) { + const job = await backfillJobModel.getLatestForSource(sourceId); + if (!job) return null; + return { + id: job.id, + sourceId: job.source_id, + status: job.status, + startedAt: job.started_at, + completedAt: job.completed_at, + rangeStart: job.range_start, + rangeEnd: job.range_end, + chunkSize: job.chunk_size, + }; + } +} + +export const backfillService = new BackfillService(); diff --git a/backend/src/services/incidentIngestion.service.ts b/backend/src/services/incidentIngestion.service.ts index d0dca599..ac68ee58 100644 --- a/backend/src/services/incidentIngestion.service.ts +++ b/backend/src/services/incidentIngestion.service.ts @@ -1,127 +1,488 @@ -import { Knex } from "knex"; -import { - IncidentIngestionModel, - ThirdPartyIncident, - IncidentIngestionSource, -} from "../database/models/IncidentIngestion.js"; -import logger from "../utils/logger.js"; +import crypto from "node:crypto"; +import { getDatabase } from "../database/connection.js"; +import { logger } from "../utils/logger.js"; +import { enrichmentPipelineService, type EnrichmentResult } from "./enrichment/index.js"; +import { IncidentService, type IncidentSeverity, type BridgeIncident } from "./incident.service.js"; + +export type IncidentSourceType = "github" | "webhook" | "partner" | "manual"; + +export interface RawIncidentPayload { + sourceType?: IncidentSourceType | string; + externalId?: string; + bridgeId?: string; + assetCode?: string; + severity?: string; + title?: string; + description?: string; + sourceUrl?: string; + occurredAt?: string; + repository?: string; + repoAvatarUrl?: string; + actor?: string; + followUpActions?: string[]; + source?: { + type?: string; + externalId?: string; + repository?: string; + repoAvatarUrl?: string; + actor?: string; + url?: string; + }; + metadata?: Record; +} + +export interface IngestIncidentResult { + incident: BridgeIncident | null; + duplicate: boolean; + queuedForReview: boolean; + reviewReason?: string; +} + +interface NormalizedIncident { + sourceType: IncidentSourceType; + sourceExternalId: string | null; + bridgeId: string; + assetCode: string | null; + severity: IncidentSeverity; + title: string; + description: string; + sourceUrl: string | null; + followUpActions: string[]; + occurredAt: string; + sourceRepository: string | null; + sourceRepoAvatarUrl: string | null; + sourceActor: string | null; + sourceAttribution: Record; + enrichmentMetadata: Record; + enrichmentTags: string[]; + derivedFields: Record; + enrichmentValidation: Record; + normalizedFingerprint: string; + requiresManualReview: boolean; + reviewReason: string | null; +} + +const MANUAL_REVIEW_REASONS = { + missingBridgeId: "missing_bridge_id", + missingTitle: "missing_title", + missingDescription: "missing_description", +} as const; + +const SEVERITY_MAP: Record = { + critical: "critical", + crit: "critical", + sev0: "critical", + severe: "critical", + high: "high", + sev1: "high", + major: "high", + medium: "medium", + med: "medium", + moderate: "medium", + sev2: "medium", + low: "low", + minor: "low", + info: "low", + informational: "low", + sev3: "low", +}; + +const RETRYABLE_ERROR_CODES = new Set(["ECONNRESET", "ETIMEDOUT", "ENOTFOUND", "EAI_AGAIN"]); export class IncidentIngestionService { - private model: IncidentIngestionModel; + private db = getDatabase(); + private incidentService = new IncidentService(); - constructor(private db: Knex) { - this.model = new IncidentIngestionModel(db); - } + normalize(raw: RawIncidentPayload): NormalizedIncident { + const sourceType = this.normalizeSourceType(raw.sourceType ?? raw.source?.type); + const sourceExternalId = this.normalizeString(raw.externalId ?? raw.source?.externalId); + const bridgeId = this.normalizeString(raw.bridgeId) ?? "unknown"; + const title = this.normalizeString(raw.title) ?? "Untitled incident"; + const description = this.normalizeString(raw.description) ?? "No description provided."; - async ingestIncident( - source: string, - externalIncident: any, - ): Promise { - const existing = await this.model.getIncidentByExternalId( - source, - externalIncident.id, - ); + const sourceRepository = this.normalizeString(raw.repository ?? raw.source?.repository); + const sourceRepoAvatarUrl = this.normalizeString(raw.repoAvatarUrl ?? raw.source?.repoAvatarUrl); + const sourceActor = this.normalizeString(raw.actor ?? raw.source?.actor); + const sourceUrl = this.normalizeString(raw.sourceUrl ?? raw.source?.url); - const incidentData = { - source, - external_id: externalIncident.id, - title: externalIncident.title || externalIncident.name, - description: externalIncident.description, - status: this.normalizeStatus(externalIncident.status), - severity: this.normalizeSeverity( - externalIncident.impact || externalIncident.severity, - ), - affected_component: externalIncident.components?.[0] || null, - incident_started_at: new Date( - externalIncident.started_at || externalIncident.created_at, - ), - incident_resolved_at: externalIncident.resolved_at - ? new Date(externalIncident.resolved_at) - : null, - metadata: externalIncident, + const severity = this.mapSeverity(raw.severity); + const occurredAt = this.toIsoDate(raw.occurredAt) ?? new Date().toISOString(); + const followUpActions = Array.isArray(raw.followUpActions) ? raw.followUpActions.filter(Boolean) : []; + + const missing: string[] = []; + if (!this.normalizeString(raw.bridgeId)) missing.push(MANUAL_REVIEW_REASONS.missingBridgeId); + if (!this.normalizeString(raw.title)) missing.push(MANUAL_REVIEW_REASONS.missingTitle); + if (!this.normalizeString(raw.description)) missing.push(MANUAL_REVIEW_REASONS.missingDescription); + + const normalizedFingerprint = this.buildFingerprint({ + sourceType, + sourceExternalId, + bridgeId, + title, + occurredAt, + sourceUrl, + }); + + return { + sourceType, + sourceExternalId, + bridgeId, + assetCode: this.normalizeString(raw.assetCode), + severity, + title, + description, + sourceUrl, + followUpActions, + occurredAt, + sourceRepository, + sourceRepoAvatarUrl, + sourceActor, + sourceAttribution: { + sourceType, + sourceExternalId, + repository: sourceRepository, + repoAvatarUrl: sourceRepoAvatarUrl, + actor: sourceActor, + sourceUrl, + metadata: raw.metadata ?? {}, + }, + enrichmentMetadata: {}, + enrichmentTags: [], + derivedFields: {}, + enrichmentValidation: {}, + normalizedFingerprint, + requiresManualReview: missing.length > 0, + reviewReason: missing.length > 0 ? missing.join(",") : null, }; + } + async ingest(raw: RawIncidentPayload): Promise { + const normalized = await this.enrichNormalized(this.normalize(raw), raw); + + if (normalized.requiresManualReview) { + await this.enqueueReview(normalized, raw); + await this.recordHistory({ + incidentId: null, + normalized, + eventType: "queued_for_review", + status: "queued", + errorMessage: normalized.reviewReason, + attemptNumber: 1, + }); + + return { + incident: null, + duplicate: false, + queuedForReview: true, + reviewReason: normalized.reviewReason ?? undefined, + }; + } + + const existing = await this.findDuplicate(normalized); if (existing) { - const updated = await this.model.updateIncident( - existing.id, - incidentData, - ); - logger.info({ incidentId: updated.id }, "Updated existing incident"); - return updated; - } else { - const created = await this.model.createIncident(incidentData); - logger.info({ incidentId: created.id }, "Created new incident"); - return created; + const existingIncidentId = typeof existing.id === "string" ? existing.id : null; + + await this.recordHistory({ + incidentId: existingIncidentId, + normalized, + eventType: "duplicate_detected", + status: "duplicate", + attemptNumber: Number((existing as any).ingestion_attempt_count ?? 0) + 1, + }); + + return { + incident: this.incidentService.mapDatabaseRow(existing as unknown as Record), + duplicate: true, + queuedForReview: false, + }; } + + const inserted = await this.createIncidentFromNormalized(normalized); + + await this.recordHistory({ + incidentId: inserted.id, + normalized, + eventType: "ingested", + status: "processed", + attemptNumber: 1, + }); + + logger.info( + { + incidentId: inserted.id, + sourceType: normalized.sourceType, + sourceExternalId: normalized.sourceExternalId, + }, + "Bridge incident ingested" + ); + + return { incident: inserted, duplicate: false, queuedForReview: false }; } - async pollSource( - sourceId: string, - ): Promise<{ success: boolean; incidentsProcessed: number; error?: string }> { - const sources = await this.model.getActiveSources(); - const source = sources.find((s) => s.id === sourceId); + async ingestWithRetry(raw: RawIncidentPayload, maxAttempts = 3): Promise { + let attempt = 0; + let lastError: unknown; - if (!source) { - throw new Error(`Source ${sourceId} not found or inactive`); + while (attempt < maxAttempts) { + attempt += 1; + try { + return await this.ingest(raw); + } catch (error) { + lastError = error; + if (!this.isRetryable(error) || attempt >= maxAttempts) { + const normalized = this.normalize(raw); + await this.recordHistory({ + incidentId: null, + normalized, + eventType: "ingestion_failed", + status: "failed", + errorMessage: error instanceof Error ? error.message : "Unknown ingestion error", + attemptNumber: attempt, + }); + throw error; + } + } } - try { - const incidents = await this.fetchIncidentsFromSource(source); - let processed = 0; + throw lastError instanceof Error ? lastError : new Error("Unknown ingestion failure"); + } - for (const incident of incidents) { - await this.ingestIncident(source.source_name, incident); - processed++; - } + async listManualReviewQueue(limit = 50): Promise { + return this.db("bridge_incident_review_queue") + .where("status", "pending") + .orderBy("created_at", "asc") + .limit(limit) + .select("*"); + } - await this.model.updateSource(sourceId, { - last_poll_at: new Date(), - last_success_at: new Date(), - last_error: null, - }); + async searchIngestionHistory(filters: { + sourceType?: string; + status?: string; + incidentId?: string; + startDate?: Date; + endDate?: Date; + limit?: number; + offset?: number; + }): Promise<{ data: any[]; total: number }> { + let query = this.db("bridge_incident_ingestion_history"); - return { success: true, incidentsProcessed: processed }; - } catch (error: any) { - await this.model.updateSource(sourceId, { - last_poll_at: new Date(), - last_error: error.message, - }); + if (filters.sourceType) query = query.where("source_type", filters.sourceType); + if (filters.status) query = query.where("status", filters.status); + if (filters.incidentId) query = query.where("incident_id", filters.incidentId); + if (filters.startDate) query = query.where("created_at", ">=", filters.startDate); + if (filters.endDate) query = query.where("created_at", "<=", filters.endDate); + + const countRow = await query.clone().count("id as count").first(); + const total = Number((countRow as any)?.count ?? 0); + + const data = await query + .orderBy("created_at", "desc") + .limit(filters.limit ?? 100) + .offset(filters.offset ?? 0); + + return { data, total }; + } + + private async findDuplicate(normalized: NormalizedIncident): Promise | null> { + const byFingerprint = await this.db("bridge_incidents") + .where("normalized_fingerprint", normalized.normalizedFingerprint) + .first(); + + if (byFingerprint) return byFingerprint as Record; + + if (normalized.sourceExternalId) { + const byExternalId = await this.db("bridge_incidents") + .where({ + source_type: normalized.sourceType, + source_external_id: normalized.sourceExternalId, + }) + .first(); + + if (byExternalId) return byExternalId as Record; + } + + return null; + } + + private async createIncidentFromNormalized(normalized: NormalizedIncident): Promise { + const [row] = await this.db("bridge_incidents") + .insert({ + bridge_id: normalized.bridgeId, + asset_code: normalized.assetCode, + severity: normalized.severity, + title: normalized.title, + description: normalized.description, + source_url: normalized.sourceUrl, + follow_up_actions: JSON.stringify(normalized.followUpActions), + occurred_at: new Date(normalized.occurredAt), + source_type: normalized.sourceType, + source_external_id: normalized.sourceExternalId, + source_repository: normalized.sourceRepository, + source_repo_avatar_url: normalized.sourceRepoAvatarUrl, + source_actor: normalized.sourceActor, + source_attribution: JSON.stringify(normalized.sourceAttribution), + enrichment_metadata: JSON.stringify(normalized.enrichmentMetadata), + enrichment_tags: normalized.enrichmentTags, + derived_fields: JSON.stringify(normalized.derivedFields), + enrichment_validation: JSON.stringify(normalized.enrichmentValidation), + normalized_fingerprint: normalized.normalizedFingerprint, + requires_manual_review: false, + ingestion_attempt_count: 1, + last_ingestion_error: null, + }) + .returning("*"); + + return this.incidentService.mapDatabaseRow(row as unknown as Record); + } + + private async enqueueReview(normalized: NormalizedIncident, raw: RawIncidentPayload): Promise { + await this.db("bridge_incident_review_queue").insert({ + source_type: normalized.sourceType, + source_external_id: normalized.sourceExternalId, + raw_payload: JSON.stringify(raw), + enriched_payload: JSON.stringify({ + metadata: normalized.enrichmentMetadata, + tags: normalized.enrichmentTags, + derivedFields: normalized.derivedFields, + validation: normalized.enrichmentValidation, + }), + reason: normalized.reviewReason, + status: "pending", + incident_id: null, + }); + } + + private async recordHistory(input: { + incidentId: string | null; + normalized: NormalizedIncident; + eventType: string; + status: string; + errorMessage?: string | null; + attemptNumber: number; + }): Promise { + await this.db("bridge_incident_ingestion_history").insert({ + incident_id: input.incidentId, + source_type: input.normalized.sourceType, + source_external_id: input.normalized.sourceExternalId, + event_type: input.eventType, + payload: JSON.stringify(input.normalized.sourceAttribution), + enrichment_metadata: JSON.stringify(input.normalized.enrichmentMetadata), + enrichment_tags: input.normalized.enrichmentTags, + derived_fields: JSON.stringify(input.normalized.derivedFields), + status: input.status, + error_message: input.errorMessage ?? null, + attempt_number: input.attemptNumber, + }); + } + + private async enrichNormalized(normalized: NormalizedIncident, raw: RawIncidentPayload): Promise { + const enrichment = await enrichmentPipelineService.enrich({ + recordType: "incident", + provider: normalized.sourceType, + data: { + sourceType: normalized.sourceType, + sourceExternalId: normalized.sourceExternalId, + bridgeId: normalized.bridgeId, + assetCode: normalized.assetCode, + severity: normalized.severity, + title: normalized.title, + description: normalized.description, + sourceUrl: normalized.sourceUrl, + occurredAt: normalized.occurredAt, + followUpActions: normalized.followUpActions, + requiresManualReview: normalized.requiresManualReview, + }, + context: { + rawMetadata: raw.metadata ?? {}, + repository: normalized.sourceRepository, + actor: normalized.sourceActor, + }, + }); + + return this.applyEnrichment(normalized, enrichment); + } - return { success: false, incidentsProcessed: 0, error: error.message }; + private applyEnrichment( + normalized: NormalizedIncident, + enrichment: EnrichmentResult, + ): NormalizedIncident { + const enrichmentMetadata = { + ...enrichment.metadata, + rawMetadata: normalized.sourceAttribution.metadata ?? {}, + }; + + return { + ...normalized, + sourceAttribution: { + ...normalized.sourceAttribution, + enrichment: { + metadata: enrichmentMetadata, + tags: enrichment.tags, + derivedFields: enrichment.derivedFields, + validation: enrichment.validation, + attempts: enrichment.attempts, + }, + }, + enrichmentMetadata, + enrichmentTags: enrichment.tags, + derivedFields: enrichment.derivedFields, + enrichmentValidation: { + ...enrichment.validation, + attempts: enrichment.attempts, + }, + }; + } + + private mapSeverity(sourceSeverity: string | undefined): IncidentSeverity { + const key = this.normalizeString(sourceSeverity)?.toLowerCase(); + if (!key) return "medium"; + return SEVERITY_MAP[key] ?? "medium"; + } + + private normalizeSourceType(value: string | undefined): IncidentSourceType { + const normalized = this.normalizeString(value)?.toLowerCase(); + if (normalized === "github" || normalized === "partner" || normalized === "manual") { + return normalized; } + return "webhook"; } - async getActiveIncidents(): Promise { - return await this.model.getActiveIncidents(); + private normalizeString(value: string | undefined | null): string | null { + if (typeof value !== "string") return null; + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : null; } - private async fetchIncidentsFromSource( - source: IncidentIngestionSource, - ): Promise { - // Placeholder implementation - would integrate with actual APIs - // StatusPage, PagerDuty, etc. - return []; + private toIsoDate(value: string | undefined): string | null { + if (!value) return null; + const d = new Date(value); + if (Number.isNaN(d.getTime())) return null; + return d.toISOString(); } - private normalizeStatus( - status: string, - ): "investigating" | "identified" | "monitoring" | "resolved" { - const normalized = status.toLowerCase(); - if (normalized.includes("investigating")) return "investigating"; - if (normalized.includes("identified")) return "identified"; - if (normalized.includes("monitoring") || normalized.includes("watching")) - return "monitoring"; - if (normalized.includes("resolved") || normalized.includes("fixed")) - return "resolved"; - return "investigating"; + private buildFingerprint(input: { + sourceType: IncidentSourceType; + sourceExternalId: string | null; + bridgeId: string; + title: string; + occurredAt: string; + sourceUrl: string | null; + }): string { + const material = [ + input.sourceType, + input.sourceExternalId ?? "", + input.bridgeId, + input.title.toLowerCase(), + input.occurredAt, + input.sourceUrl ?? "", + ].join("|"); + + return crypto.createHash("sha256").update(material).digest("hex"); } - private normalizeSeverity(impact: string): "minor" | "major" | "critical" { - const normalized = impact.toLowerCase(); - if (normalized.includes("critical") || normalized.includes("high")) - return "critical"; - if (normalized.includes("major") || normalized.includes("medium")) - return "major"; - return "minor"; + private isRetryable(error: unknown): boolean { + if (!error || typeof error !== "object") return false; + const err = error as { code?: string }; + return typeof err.code === "string" && RETRYABLE_ERROR_CODES.has(err.code); } } diff --git a/backend/src/services/sorobanEventIndex.service.ts b/backend/src/services/sorobanEventIndex.service.ts new file mode 100644 index 00000000..91a5dff5 --- /dev/null +++ b/backend/src/services/sorobanEventIndex.service.ts @@ -0,0 +1,55 @@ +import { sorobanEventModel, type SorobanEventRecord } from "../database/models/sorobanEvent.model.js"; +import { SorobanRpcClient } from "./stellar/soroban.client.js"; +import { config } from "../config/index.js"; +import { logger } from "../utils/logger.js"; + +export class SorobanEventIndexService { + private rpcClient: SorobanRpcClient; + + constructor() { + this.rpcClient = new SorobanRpcClient({ + rpcUrls: [config.SOROBAN_RPC_URL || "https://soroban-testnet.stellar.org"], + }); + } + + async getPaginatedEvents( + contractId?: string, + limit = 100, + cursor?: string + ): Promise<{ data: SorobanEventRecord[]; nextCursor?: string }> { + return sorobanEventModel.getPaginatedEvents(contractId, limit, cursor); + } + + // Called periodically via a worker to index events + async syncEvents(contractId: string, limit = 1000): Promise { + const cursor = await sorobanEventModel.getLatestCursor(contractId) ?? "0"; + try { + const response = await this.rpcClient.getEvents({ + cursor, + limit, + filters: [{ type: "contract", contractIds: [contractId] }], + }) as { events?: any[] }; + + const events = response.events || []; + if (events.length === 0) return 0; + + const records: SorobanEventRecord[] = events.map((e: any) => ({ + cursor: e.pagingToken, + ledger: e.ledger, + ledger_closed_at: new Date(e.ledgerClosedAt), + contract_id: e.contractId, + topic: JSON.stringify(e.topic || []), + value: e.value, + })); + + await sorobanEventModel.insert(records); + logger.info({ contractId, count: records.length }, "Indexed new Soroban events"); + return records.length; + } catch (error) { + logger.error({ contractId, error }, "Failed to sync Soroban events"); + return 0; + } + } +} + +export const sorobanEventIndexService = new SorobanEventIndexService(); diff --git a/backend/src/services/sourceHealthScoring.service.ts b/backend/src/services/sourceHealthScoring.service.ts index 11a8fc0e..707dbc32 100644 --- a/backend/src/services/sourceHealthScoring.service.ts +++ b/backend/src/services/sourceHealthScoring.service.ts @@ -54,6 +54,8 @@ export interface SourceHealthScore { contributingFactors: SourceContributingFactors; thresholdViolations: SourceThresholdViolation[]; sampleCount: number; + confidenceScore: number; + confidenceBand: string; computedAt: string; updatedAt: string; } @@ -69,6 +71,8 @@ export interface SourceHealthHistoryEntry { grade: string; alertState: string; sampleCount: number; + confidenceScore: number; + confidenceBand: string; computedAt: string; } @@ -315,6 +319,9 @@ export class SourceHealthScoringService { const now = new Date(); + const confidenceScore = Math.min(100, Math.round((totalChecks / 144) * 100)); // Assuming 10m intervals over 24h + const confidenceBand = confidenceScore >= 80 ? "HIGH" : confidenceScore >= 50 ? "MEDIUM" : "LOW"; + const [row] = await this.db("source_health_scores") .insert({ source_key: sourceKey, @@ -330,6 +337,8 @@ export class SourceHealthScoringService { contributing_factors: JSON.stringify(factors), threshold_violations: JSON.stringify(violations), sample_count: totalChecks, + confidence_score: confidenceScore, + confidence_band: confidenceBand, computed_at: now, updated_at: now, }) @@ -347,6 +356,8 @@ export class SourceHealthScoringService { "contributing_factors", "threshold_violations", "sample_count", + "confidence_score", + "confidence_band", "computed_at", "updated_at", ]) @@ -362,6 +373,8 @@ export class SourceHealthScoringService { grade, alert_state: alertState, sample_count: totalChecks, + confidence_score: confidenceScore, + confidence_band: confidenceBand, computed_at: now, }); @@ -475,6 +488,8 @@ export class SourceHealthScoringService { }), thresholdViolations: parseJson(row.threshold_violations, []), sampleCount: Number(row.sample_count ?? 0), + confidenceScore: Number(row.confidence_score ?? 0), + confidenceBand: String(row.confidence_band ?? "LOW"), computedAt: isoTimestamp(row.computed_at), updatedAt: isoTimestamp(row.updated_at), }; @@ -492,6 +507,8 @@ export class SourceHealthScoringService { grade: String(row.grade), alertState: String(row.alert_state), sampleCount: Number(row.sample_count ?? 0), + confidenceScore: Number(row.confidence_score ?? 0), + confidenceBand: String(row.confidence_band ?? "LOW"), computedAt: isoTimestamp(row.computed_at), }; }