From 5d73d4d74dba11473a695d733bea992463a68e21 Mon Sep 17 00:00:00 2001 From: Victor Edeh Date: Sun, 30 Aug 2026 18:14:49 +0100 Subject: [PATCH] fix: scope stellar projections by network --- app/api/stellar/activity/route.ts | 25 +++-- lib/stellar/indexer.ts | 19 +++- models/StellarIndexedEvent.ts | 61 +++++++++- .../migrate-stellar-indexed-events-network.ts | 104 ++++++++++++++++++ 4 files changed, 187 insertions(+), 22 deletions(-) create mode 100644 scripts/migrate-stellar-indexed-events-network.ts diff --git a/app/api/stellar/activity/route.ts b/app/api/stellar/activity/route.ts index 06b18f4f..37a29238 100644 --- a/app/api/stellar/activity/route.ts +++ b/app/api/stellar/activity/route.ts @@ -167,18 +167,16 @@ export async function GET(request: Request) { } // 2. Query MongoDB for events - let eventQuery: any = {} - if (user.role === "admin") { - // Admins see all indexed events - eventQuery = {} - } else { - // Investors/Drivers see events involving their linked key - eventQuery = { - $or: [ - { sourceAccount: stellarPublicKey }, - { destinationAccount: stellarPublicKey } - ] - } + const eventQuery: any = { + network: config.network.toLowerCase(), + projectionStatus: "active", + } + if (user.role !== "admin") { + // Investors/Drivers see events involving their linked key on the configured Stellar network only. + eventQuery.$or = [ + { sourceAccount: stellarPublicKey }, + { destinationAccount: stellarPublicKey } + ] } const events = await StellarIndexedEvent.find(eventQuery) @@ -241,3 +239,6 @@ export async function GET(request: Request) { return NextResponse.json({ error: "Failed to fetch Stellar activity" }, { status: 500 }) } } + + + diff --git a/lib/stellar/indexer.ts b/lib/stellar/indexer.ts index 2d707a4d..08ddc849 100644 --- a/lib/stellar/indexer.ts +++ b/lib/stellar/indexer.ts @@ -21,7 +21,7 @@ * access. This is the default outside of the `production` NODE_ENV. */ -import type { StellarEventType, ChainMoveRecordType } from "@/models/StellarIndexedEvent" +import { buildStellarIndexedEventId, normalizeStellarIndexedNetwork, type StellarEventType, type ChainMoveRecordType } from "@/models/StellarIndexedEvent" import { getStellarConfig } from "@/lib/stellar/config" import crypto from "crypto" @@ -262,14 +262,20 @@ interface PersistResult { error?: string } -async function persistEvent(op: RawStellarOperation): Promise { +async function persistEvent(op: RawStellarOperation, network: string, projectionProvenance: "indexed" | "rebuilt_from_raw" = "indexed"): Promise { const { default: StellarIndexedEvent } = await import("@/models/StellarIndexedEvent") const chainMoveRecordType = mapEventToChainMoveRecord(op) const eventType = toEventType(op.type ?? "unknown") + const normalizedNetwork = normalizeStellarIndexedNetwork(network) + const operationId = op.id const doc = { - _id: op.id, + _id: buildStellarIndexedEventId(normalizedNetwork, operationId), + network: normalizedNetwork, + operationId, + projectionStatus: "active", + projectionProvenance, pagingToken: op.paging_token, eventType, sourceAccount: op.source_account, @@ -609,7 +615,7 @@ export function createStellarIndexer(options: StellarIndexerOptions = {}): Stell continue } - const result = await persistEvent(op) + const result = await persistEvent(op, network) if (result.inserted) { processed++ @@ -660,7 +666,7 @@ export function createStellarIndexer(options: StellarIndexerOptions = {}): Stell for (const failure of failures) { const op = failure.raw as RawStellarOperation - const result = await persistEvent(op) + const result = await persistEvent(op, network) await StellarDeadLetterEvent.findByIdAndUpdate(failure._id, { $inc: { replayCount: 1 }, $set: { lastReplayAt: new Date() }, @@ -723,3 +729,6 @@ export function createStellarIndexer(options: StellarIndexerOptions = {}): Stell return { sync, replayDeadLetters, health, streamId, isMock } } + + + diff --git a/models/StellarIndexedEvent.ts b/models/StellarIndexedEvent.ts index 2545b59d..edfb8741 100644 --- a/models/StellarIndexedEvent.ts +++ b/models/StellarIndexedEvent.ts @@ -2,9 +2,9 @@ import mongoose, { Document, Schema } from "mongoose" /** * Records each Stellar network event that has been ingested by the indexer. - * The `_id` field is used as the idempotency key — it is set to the Stellar - * event/operation/payment ID so a second attempt to insert the same event - * hits the unique index and is safely ignored. + * The `_id` field is used as the idempotency key. It includes the selected + * Stellar network plus the Horizon operation/payment ID so testnet and mainnet + * projections cannot collide. */ export type StellarEventType = | "payment" @@ -25,9 +25,30 @@ export type ChainMoveRecordType = | "contract_interaction" | "unclassified" +export type StellarIndexedEventStatus = "active" | "quarantined" +export type StellarIndexedEventProvenance = "indexed" | "rebuilt_from_raw" | "legacy_backfill" | "legacy_quarantine" + +export function normalizeStellarIndexedNetwork(network: string): string { + return network.trim().toLowerCase() +} + +export function buildStellarIndexedEventId(network: string, operationId: string): string { + return `${normalizeStellarIndexedNetwork(network)}:${operationId}` +} + export interface IStellarIndexedEvent { - /** Stellar event/operation/payment ID used as idempotency key. */ + /** Network-scoped idempotency key: `${network}:${operationId}`. */ _id: string + /** Immutable Stellar network provenance for this projected event. */ + network: string + /** Original Stellar event/operation/payment ID from Horizon. */ + operationId: string + /** Whether the projection is safe to serve from live activity queries. */ + projectionStatus: StellarIndexedEventStatus + /** Migration provenance for legacy projected rows. */ + projectionProvenance?: StellarIndexedEventProvenance + /** Reason a legacy or failed projection was quarantined. */ + quarantineReason?: string /** Stellar paging token / cursor at the time this event was indexed. */ pagingToken: string /** Raw Stellar event type string from Horizon. */ @@ -58,6 +79,31 @@ const StellarIndexedEventSchema = new Schema( type: String, required: true, }, + network: { + type: String, + required: true, + trim: true, + lowercase: true, + index: true, + }, + operationId: { + type: String, + required: true, + trim: true, + index: true, + }, + projectionStatus: { + type: String, + enum: ["active", "quarantined"], + required: true, + default: "active", + index: true, + }, + projectionProvenance: { + type: String, + enum: ["indexed", "rebuilt_from_raw", "legacy_backfill", "legacy_quarantine"], + }, + quarantineReason: { type: String }, pagingToken: { type: String, required: true, @@ -104,5 +150,10 @@ const StellarIndexedEventSchema = new Schema( { _id: false, timestamps: true }, ) +StellarIndexedEventSchema.index({ network: 1, operationId: 1 }, { unique: true }) +StellarIndexedEventSchema.index({ network: 1, projectionStatus: 1, stellarCreatedAt: -1, createdAt: -1 }) +StellarIndexedEventSchema.index({ network: 1, sourceAccount: 1, projectionStatus: 1 }) +StellarIndexedEventSchema.index({ network: 1, destinationAccount: 1, projectionStatus: 1 }, { sparse: true }) + export default (mongoose.models.StellarIndexedEvent || - mongoose.model("StellarIndexedEvent", StellarIndexedEventSchema)) as mongoose.Model<{ _id: any; [key: string]: any }>; \ No newline at end of file + mongoose.model("StellarIndexedEvent", StellarIndexedEventSchema)) as mongoose.Model<{ _id: any; [key: string]: any }>; diff --git a/scripts/migrate-stellar-indexed-events-network.ts b/scripts/migrate-stellar-indexed-events-network.ts new file mode 100644 index 00000000..21cc1ff7 --- /dev/null +++ b/scripts/migrate-stellar-indexed-events-network.ts @@ -0,0 +1,104 @@ +#!/usr/bin/env tsx + +import dbConnect from "@/lib/dbConnect" +import { getStellarConfig, parseStellarNetwork } from "@/lib/stellar/config" +import StellarIndexedEvent, { buildStellarIndexedEventId } from "@/models/StellarIndexedEvent" +import StellarRawEvent from "@/models/StellarRawEvent" + +interface Options { + apply: boolean + network?: string +} + +function parseArgs(argv: string[]): Options { + const options: Options = { apply: false } + for (let i = 0; i < argv.length; i++) { + const arg = argv[i] + if (arg === "--apply") options.apply = true + if (arg === "--network") options.network = argv[++i] + } + return options +} + +function inferOperationId(doc: any): string { + return String(doc.operationId || doc.raw?.id || doc._id) +} + +async function resolveLegacyNetwork(operationId: string, fallbackNetwork?: string): Promise<{ network?: string; reason?: string }> { + const rawMatches = await StellarRawEvent.find({ operationId }).select("network").lean() + const networks = [...new Set(rawMatches.map((match: any) => String(match.network || "").toLowerCase()).filter(Boolean))] + + if (networks.length === 1) return { network: networks[0] } + if (networks.length > 1) return { reason: `ambiguous raw provenance: ${networks.join(",")}` } + if (fallbackNetwork) return { network: parseStellarNetwork(fallbackNetwork) } + + return { reason: "legacy projection has no raw network provenance" } +} + +async function main() { + const options = parseArgs(process.argv.slice(2)) + const configuredNetwork = options.network ?? getStellarConfig().network + + await dbConnect() + + const legacyEvents = await StellarIndexedEvent.find({ + $or: [ + { network: { $exists: false } }, + { operationId: { $exists: false } }, + { _id: { $not: /^(testnet|mainnet):/ } }, + ], + }).lean() + + let backfilled = 0 + let quarantined = 0 + + for (const event of legacyEvents) { + const operationId = inferOperationId(event) + const resolution = await resolveLegacyNetwork(operationId, configuredNetwork) + + if (!resolution.network) { + quarantined++ + if (options.apply) { + await StellarIndexedEvent.updateOne( + { _id: event._id }, + { + $set: { + operationId, + projectionStatus: "quarantined", + projectionProvenance: "legacy_quarantine", + quarantineReason: resolution.reason, + }, + }, + ) + } + continue + } + + backfilled++ + if (options.apply) { + const nextId = buildStellarIndexedEventId(resolution.network, operationId) + await StellarIndexedEvent.replaceOne( + { _id: event._id }, + { + ...event, + _id: nextId, + network: resolution.network, + operationId, + projectionStatus: "active", + projectionProvenance: "legacy_backfill", + }, + { upsert: true }, + ) + if (nextId !== String(event._id)) { + await StellarIndexedEvent.deleteOne({ _id: event._id }) + } + } + } + + console.log(JSON.stringify({ scanned: legacyEvents.length, backfilled, quarantined, dryRun: !options.apply }, null, 2)) +} + +main().catch((error) => { + console.error(error instanceof Error ? error.message : error) + process.exit(1) +})