fix: scope Stellar projected events by network

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 })
   }
 }
+
+
+

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<PersistResult> {
+async function persistEvent(op: RawStellarOperation, network: string, projectionProvenance: "indexed" | "rebuilt_from_raw" = "indexed"): Promise<PersistResult> {
   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 }
 }
+
+
+

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<IStellarIndexedEvent>(
       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<IStellarIndexedEvent>(
   { _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<IStellarIndexedEvent>("StellarIndexedEvent", StellarIndexedEventSchema)) as mongoose.Model<{ _id: any; [key: string]: any }>;
+  mongoose.model<IStellarIndexedEvent>("StellarIndexedEvent", StellarIndexedEventSchema)) as mongoose.Model<{ _id: any; [key: string]: any }>;

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)
+})