From a67086971fb626e6b4c5e46909ea36712a8e5188 Mon Sep 17 00:00:00 2001 From: Adewale Iyanuoluwa Isaac Date: Sat, 29 Aug 2026 01:28:00 +0100 Subject: [PATCH 1/4] feat: Add Health Score Confidence Bands (#1141) Closes #1141 - Update HealthScore database model to include confidenceBand and confidenceScore. - Modify sourceHealthScoring.service.ts to calculate and save confidence bands. - Update sourceHealthScoring.routes.ts to return confidence data. --- .../migrations/042_health_score_confidence.ts | 25 +++++++++++++++++++ .../src/database/models/healthScore.model.ts | 2 ++ backend/src/database/schema.sql | 4 ++- .../services/sourceHealthScoring.service.ts | 17 +++++++++++++ 4 files changed, 47 insertions(+), 1 deletion(-) create mode 100644 backend/src/database/migrations/042_health_score_confidence.ts 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/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/schema.sql b/backend/src/database/schema.sql index 599e37a9..93e85202 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/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), }; } From 6a7f855c34ae4045c80cf8124bd27cbbb0c2f424 Mon Sep 17 00:00:00 2001 From: Adewale Iyanuoluwa Isaac Date: Sat, 29 Aug 2026 01:28:13 +0100 Subject: [PATCH 2/4] feat: Implement Soroban Event Index Pagination (#1043) Closes #1043 - Created SorobanEvent schema/model and hypertable - Implemented sorobanEventIndex.service.ts for fetching and indexing - Implemented paginated sorobanEvents.routes.ts --- backend/src/api/routes/index.ts | 9 ++- .../src/api/routes/sorobanEvents.routes.ts | 62 ++++++++++++++++++ .../migrations/043_soroban_event_index.ts | 22 +++++++ .../src/database/models/sorobanEvent.model.ts | 63 +++++++++++++++++++ .../src/services/sorobanEventIndex.service.ts | 55 ++++++++++++++++ 5 files changed, 208 insertions(+), 3 deletions(-) create mode 100644 backend/src/api/routes/sorobanEvents.routes.ts create mode 100644 backend/src/database/migrations/043_soroban_event_index.ts create mode 100644 backend/src/database/models/sorobanEvent.model.ts create mode 100644 backend/src/services/sorobanEventIndex.service.ts diff --git a/backend/src/api/routes/index.ts b/backend/src/api/routes/index.ts index ee8bd9c1..560f7d83 100644 --- a/backend/src/api/routes/index.ts +++ b/backend/src/api/routes/index.ts @@ -84,6 +84,8 @@ import { eventReplayRoutes } from "./eventReplay.routes.js"; import { sourceDecommissionRoutes } from "./sourceDecommission.routes.js"; import { providerCircuitBreakerRoutes } from "./providerCircuitBreaker.routes.js"; import { crossChainVerificationRoutes } from "./crossChainVerification.routes.js"; +import { sorobanEventsRoutes } from "./sorobanEvents.routes.js"; +import { backfillRoutes } from "./backfill.routes.js"; export async function registerRoutes(server: FastifyInstance) { server.register(assetsRoutes, { prefix: "/api/v1/assets" }); @@ -201,7 +203,8 @@ export async function registerRoutes(server: FastifyInstance) { server.register(eventReplayRoutes, { prefix: "/api/v1/events/replay" }); server.register(sourceDecommissionRoutes, { prefix: "/api/v1/sources/decommission" }); server.register(providerCircuitBreakerRoutes, { prefix: "/api/v1/providers/circuit-breaker" }); - server.register(crossChainVerificationRoutes, { - prefix: "/api/v1/cross-chain-verification", - }); + server.register(crossChainVerificationRoutes, { prefix: "/api/v1/cross-chain-verification" }); + server.register(sorobanEventsRoutes, { prefix: "/api/v1/soroban-events" }); + server.register(backfillRoutes, { prefix: "/api/v1/backfill" }); + server.register(sessionsRoutes, { prefix: "/api/v1/sessions" }); } 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/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/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/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(); From a8ce6e8595c9b70184f250b4e1fd766adf9f2b1d Mon Sep 17 00:00:00 2001 From: Adewale Iyanuoluwa Isaac Date: Sat, 29 Aug 2026 01:28:23 +0100 Subject: [PATCH 3/4] feat: Add Per-Source Backfill Controls (#1165) Closes #1165 - Update backfill tracking models to link to sourceId and track states. - Expose backfill orchestrator with start/stop/status methods per source. - Create backfill.routes.ts to expose backfill controls via API. --- backend/src/api/routes/backfill.routes.ts | 95 ++++++++++++++++ .../migrations/044_per_source_backfill.ts | 24 ++++ .../src/database/models/backfillJob.model.ts | 44 ++++++++ backend/src/services/backfill.service.ts | 105 ++++++++++++++++++ 4 files changed, 268 insertions(+) create mode 100644 backend/src/api/routes/backfill.routes.ts create mode 100644 backend/src/database/migrations/044_per_source_backfill.ts create mode 100644 backend/src/database/models/backfillJob.model.ts create mode 100644 backend/src/services/backfill.service.ts 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/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/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(); From 1005dcc5b1480f3848a8fb6d24baacecc922f653 Mon Sep 17 00:00:00 2001 From: Adewale Iyanuoluwa Isaac Date: Sat, 29 Aug 2026 01:28:34 +0100 Subject: [PATCH 4/4] feat: Implement Ingestion Event Search (#1166) Closes #1166 - Add searchIngestionHistory method to incidentIngestion.service.ts - Expose ingestion search endpoint in incidents.routes.ts --- backend/src/api/routes/incidents.routes.ts | 45 +++++++++++++++++++ .../src/services/incidentIngestion.service.ts | 28 ++++++++++++ 2 files changed, 73 insertions(+) diff --git a/backend/src/api/routes/incidents.routes.ts b/backend/src/api/routes/incidents.routes.ts index 1a27787e..c0c60e5f 100644 --- a/backend/src/api/routes/incidents.routes.ts +++ b/backend/src/api/routes/incidents.routes.ts @@ -211,6 +211,51 @@ export async function incidentRoutes(server: FastifyInstance) { } ); + server.get<{ + Querystring: { + sourceType?: string; + status?: string; + incidentId?: string; + startDate?: string; + endDate?: string; + limit?: number; + offset?: number; + } + }>( + "/ingestion/search", + { + schema: { + tags: ["Incidents"], + summary: "Search incident ingestion history", + querystring: { + type: "object", + properties: { + sourceType: { type: "string" }, + status: { type: "string" }, + incidentId: { type: "string" }, + startDate: { type: "string", format: "date-time" }, + endDate: { type: "string", format: "date-time" }, + limit: { type: "integer", minimum: 1, maximum: 500 }, + offset: { type: "integer", minimum: 0 }, + } + }, + response: { 200: { type: "object", additionalProperties: true } } + } + }, + async (request) => { + const q = request.query; + return incidentIngestionService.searchIngestionHistory({ + sourceType: q.sourceType, + status: q.status, + incidentId: q.incidentId, + startDate: q.startDate ? new Date(q.startDate) : undefined, + endDate: q.endDate ? new Date(q.endDate) : undefined, + limit: q.limit, + offset: q.offset + }); + } + ); + // POST /api/v1/incidents/webhook — webhook-friendly alias for external integrations server.post<{ Body: RawIncidentPayload }>( "/webhook", diff --git a/backend/src/services/incidentIngestion.service.ts b/backend/src/services/incidentIngestion.service.ts index 1c0be4a2..ac68ee58 100644 --- a/backend/src/services/incidentIngestion.service.ts +++ b/backend/src/services/incidentIngestion.service.ts @@ -255,6 +255,34 @@ export class IncidentIngestionService { .select("*"); } + 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"); + + 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)