feat: Add network discriminator to all models + Redis key prefix

prisma/schema.prisma

@@ -11,6 +11,7 @@ datasource db {
 
 model PricePoint {
   id            String   @default(uuid())
+  network       String
   assetA        String   @map("asset_a")
   assetB        String   @map("asset_b")
   pairKey       String   @map("pair_key")
@@ -24,13 +25,14 @@ model PricePoint {
   eventId       String?  @map("event_id")
 
   @@id([id, timestamp])
-  @@index([pairKey, timestamp(sort: Desc)])
-  @@index([pairKey, source, timestamp(sort: Desc)])
+  @@index([network, pairKey, timestamp(sort: Desc)])
+  @@index([network, pairKey, source, timestamp(sort: Desc)])
   @@map("price_points")
 }
 
 model PoolSnapshot {
   id          String   @default(uuid())
+  network     String
   poolId      String   @map("pool_id")
   assetA      String   @map("asset_a")
   assetB      String   @map("asset_b")
@@ -43,12 +45,13 @@ model PoolSnapshot {
   timestamp   DateTime
 
   @@id([id, timestamp])
-  @@index([poolId, timestamp(sort: Desc)])
+  @@index([network, poolId, timestamp(sort: Desc)])
   @@map("pool_snapshots")
 }
 
 model PriceAggregate {
   pairKey     String   @map("pair_key")
+  network     String
   window      String
   bucket      DateTime
   vwap        Decimal  @db.Decimal(36, 18)
@@ -63,44 +66,50 @@ model PriceAggregate {
   highPrice   Decimal? @map("high_price") @db.Decimal(36, 18)
   lowPrice    Decimal? @map("low_price") @db.Decimal(36, 18)
 
-  @@id([pairKey, window, bucket])
+  @@id([network, pairKey, window, bucket])
   @@map("price_aggregates")
 }
 
 model PriceSnapshot {
   pair   String
+  network String
   ts     DateTime
   price  Decimal  @db.Decimal(36, 18)
   volume Decimal  @default(0) @db.Decimal(36, 7)
 
-  @@id([pair, ts])
-  @@index([pair, ts])
+  @@id([network, pair, ts])
+  @@index([network, pair, ts])
   @@map("price_snapshots")
 }
 
 model IndexerState {
-  id              String    @id
+  id              String
+  network         String
   lastCursor      String?   @map("last_cursor")
   lastLedger      Int?      @map("last_ledger")
   lastProcessedAt DateTime? @map("last_processed_at")
   updatedAt       DateTime  @default(now()) @updatedAt @map("updated_at")
 
+  @@id([network, id])
   @@map("indexer_state")
 }
 
 model PairConfig {
-  pairKey   String   @id @map("pair_key")
+  pairKey   String   @map("pair_key")
+  network   String
   assetACode   String   @map("asset_a_code")
   assetAIssuer String?  @map("asset_a_issuer")
   assetBCode   String   @map("asset_b_code")
   assetBIssuer String?  @map("asset_b_issuer")
   addedAt   DateTime @default(now()) @map("added_at")
 
+  @@id([network, pairKey])
   @@map("pair_configs")
 }
 
 model Webhook {
   id        String   @id @default(uuid())
+  network   String
   url       String
   assetA    String   @map("asset_a")
   assetB    String   @map("asset_b")
@@ -109,7 +118,7 @@ model Webhook {
   secret    String
   createdAt DateTime @default(now()) @map("created_at")
 
-  @@index([assetA, assetB])
+  @@index([network, assetA, assetB])
   @@map("webhooks")
 }
 

sql/schema.sql

@@ -1,6 +1,7 @@
 -- Raw price points from SDEX trades and AMM swaps
 CREATE TABLE IF NOT EXISTS price_points (
   id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
+  network TEXT NOT NULL,
   asset_a TEXT NOT NULL,
   asset_b TEXT NOT NULL,
   pair_key TEXT NOT NULL,
@@ -14,13 +15,14 @@ CREATE TABLE IF NOT EXISTS price_points (
   event_id TEXT
 );
 
-CREATE INDEX IF NOT EXISTS idx_price_points_pair_time ON price_points (pair_key, timestamp DESC);
-CREATE INDEX IF NOT EXISTS idx_price_points_pair_source_time ON price_points (pair_key, source, timestamp DESC);
-CREATE INDEX IF NOT EXISTS idx_price_points_pool_time ON price_points (pool_id, timestamp DESC) WHERE pool_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_price_points_pair_time ON price_points (network, pair_key, timestamp DESC);
+CREATE INDEX IF NOT EXISTS idx_price_points_pair_source_time ON price_points (network, pair_key, source, timestamp DESC);
+CREATE INDEX IF NOT EXISTS idx_price_points_pool_time ON price_points (network, pool_id, timestamp DESC) WHERE pool_id IS NOT NULL;
 
 -- AMM pool reserve snapshots
 CREATE TABLE IF NOT EXISTS pool_snapshots (
   id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
+  network TEXT NOT NULL,
   pool_id TEXT NOT NULL,
   asset_a TEXT NOT NULL,
   asset_b TEXT NOT NULL,
@@ -33,12 +35,13 @@ CREATE TABLE IF NOT EXISTS pool_snapshots (
   timestamp TIMESTAMPTZ NOT NULL
 );
 
-CREATE INDEX IF NOT EXISTS idx_pool_snapshots_pool_time ON pool_snapshots (pool_id, timestamp DESC);
-CREATE INDEX IF NOT EXISTS idx_pool_snapshots_assets_time ON pool_snapshots (asset_a, asset_b, timestamp DESC);
+CREATE INDEX IF NOT EXISTS idx_pool_snapshots_pool_time ON pool_snapshots (network, pool_id, timestamp DESC);
+CREATE INDEX IF NOT EXISTS idx_pool_snapshots_assets_time ON pool_snapshots (network, asset_a, asset_b, timestamp DESC);
 
 -- Pre-computed VWAP aggregates
 CREATE TABLE IF NOT EXISTS price_aggregates (
   pair_key TEXT NOT NULL,
+  network TEXT NOT NULL,
   "window" TEXT NOT NULL CHECK ("window" IN ('1m', '5m', '1h', '24h')),
   bucket TIMESTAMPTZ NOT NULL,
   vwap NUMERIC(36, 18) NOT NULL,
@@ -52,7 +55,7 @@ CREATE TABLE IF NOT EXISTS price_aggregates (
   close_price NUMERIC(36, 18),
   high_price NUMERIC(36, 18),
   low_price NUMERIC(36, 18),
-  PRIMARY KEY (pair_key, "window", bucket)
+  PRIMARY KEY (network, pair_key, "window", bucket)
 );
 
 -- 1-minute price snapshot ring buffer.
@@ -61,21 +64,24 @@ CREATE TABLE IF NOT EXISTS price_aggregates (
 -- backtests, audit trails) without paying the cost of scanning raw price_points.
 CREATE TABLE IF NOT EXISTS price_snapshots (
   pair TEXT NOT NULL,
+  network TEXT NOT NULL,
   ts TIMESTAMPTZ NOT NULL,
   price NUMERIC(36, 18) NOT NULL,
   volume NUMERIC(36, 7) NOT NULL DEFAULT 0,
-  PRIMARY KEY (pair, ts)
+  PRIMARY KEY (network, pair, ts)
 );
 
-CREATE INDEX IF NOT EXISTS idx_price_snapshots_pair_ts ON price_snapshots (pair, ts);
+CREATE INDEX IF NOT EXISTS idx_price_snapshots_pair_ts ON price_snapshots (network, pair, ts);
 
 -- Indexer cursor state
 CREATE TABLE IF NOT EXISTS indexer_state (
-  id TEXT PRIMARY KEY,
+  id TEXT NOT NULL,
+  network TEXT NOT NULL,
   last_cursor TEXT,
   last_ledger INTEGER,
   last_processed_at TIMESTAMPTZ,
-  updated_at TIMESTAMPTZ DEFAULT NOW()
+  updated_at TIMESTAMPTZ DEFAULT NOW(),
+  PRIMARY KEY (network, id)
 );
 
 -- API keys for authenticated, rate-quota'd access.

src/__tests__/bestRoute.test.ts

@@ -25,6 +25,7 @@ vi.mock('@stellar/stellar-sdk', () => {
       vi.fn(function(code, issuer) { return { code, issuer } }),
       { native: vi.fn(() => 'native') }
     ),
+    Networks: { PUBLIC: 'PUBLIC', TESTNET: 'TESTNET' },
     __mockCall: callFn
   }
 })

src/__tests__/snapshotIngester.test.ts

@@ -82,6 +82,6 @@ describe('appendSnapshots', () => {
 
     await appendSnapshots()
 
-    expect(mockQuery.mock.calls[0][0]).toMatch(/ON CONFLICT \(pair, ts\) DO NOTHING/)
+    expect(mockQuery.mock.calls[0][0]).toMatch(/ON CONFLICT \(network, pair, ts\) DO NOTHING/)
   })
 })

src/__tests__/snapshotRetention.test.ts

@@ -15,6 +15,7 @@ vi.mock('bullmq', () => ({
 }))
 
 import { pruneOldSnapshots, SNAPSHOT_RETENTION_DAYS } from '../jobs/snapshotRetention'
+import { activeNetwork } from '../config'
 
 describe('pruneOldSnapshots', () => {
   beforeEach(() => {
@@ -28,9 +29,9 @@ describe('pruneOldSnapshots', () => {
 
     expect(SNAPSHOT_RETENTION_DAYS).toBe(30)
     expect(pruned).toBe(5)
-    expect(mockQuery.mock.calls[0][1]).toEqual([30])
+    expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 30])
     expect(mockQuery.mock.calls[0][0]).toMatch(/DELETE FROM price_snapshots/)
-    expect(mockQuery.mock.calls[0][0]).toMatch(/ts < NOW\(\) - \(\$1 \|\| ' days'\)::interval/)
+    expect(mockQuery.mock.calls[0][0]).toMatch(/ts < NOW\(\) - \(\$2 \|\| ' days'\)::interval/)
   })
 
   it('honors a custom retention window', async () => {
@@ -39,7 +40,7 @@ describe('pruneOldSnapshots', () => {
     const pruned = await pruneOldSnapshots(7)
 
     expect(pruned).toBe(0)
-    expect(mockQuery.mock.calls[0][1]).toEqual([7])
+    expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 7])
   })
 
   it('returns 0 when rowCount is null', async () => {

src/api/history.ts

@@ -1,5 +1,6 @@
 import type { FastifyInstance } from 'fastify'
 import { pgPool } from '../db'
+import { activeNetwork } from '../config'
 
 /** Supported aggregation intervals → bucket width in seconds. */
 export const HISTORY_INTERVAL_SECONDS: Record<string, number> = {
@@ -37,12 +38,12 @@ export async function queryHistory(
        (array_agg(price::float ORDER BY ts DESC))[1] AS price,
        SUM(volume::float) AS volume
      FROM price_snapshots
-     WHERE pair = $2
+     WHERE pair = $2 AND network = $5
        AND ts >= $3
        AND ts <= $4
      GROUP BY floor(EXTRACT(EPOCH FROM ts) / $1)
      ORDER BY bucket ASC`,
-    [intervalSecs, pair, from, to]
+    [intervalSecs, pair, from, to, activeNetwork]
   )
 
   return result.rows.map(r => ({

src/db.ts

@@ -1,7 +1,7 @@
 import { PrismaClient } from '@prisma/client'
 import { Pool } from 'pg'
 import { db_query_duration_seconds } from './metrics'
-import { config } from './config'
+import { config, activeNetwork } from './config'
 
 // Prisma for schema management + simple queries
 const globalForPrisma = globalThis as unknown as { prisma?: PrismaClient }
@@ -37,6 +37,7 @@ export async function upsertPricePoints(points: {
   if (points.length === 0) return 0
   const result = await prisma.pricePoint.createMany({
     data: points.map(p => ({
+      network: activeNetwork,
       assetA: p.assetA,
       assetB: p.assetB,
       pairKey: p.pairKey,
@@ -55,14 +56,14 @@ export async function upsertPricePoints(points: {
 }
 
 export async function getIndexerCursor(id: string): Promise<string | null> {
-  const state = await prisma.indexerState.findUnique({ where: { id } })
+  const state = await prisma.indexerState.findUnique({ where: { network_id: { network: activeNetwork, id } } })
   return state?.lastCursor ?? null
 }
 
 export async function setIndexerCursor(id: string, cursor: string, ledger?: number): Promise<void> {
   await prisma.indexerState.upsert({
-    where: { id },
-    create: { id, lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() },
+    where: { network_id: { network: activeNetwork, id } },
+    create: { id, network: activeNetwork, lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() },
     update: { lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() },
   })
 }

src/ingesters/amm.ts

@@ -1,6 +1,6 @@
 import { Horizon } from '@stellar/stellar-sdk'
 import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp } from '../metrics'
-import { config } from '../config'
+import { config, activeNetwork } from '../config'
 import { getActivePairs } from '../pairsRegistry'
 import { upsertPricePoints, getIndexerCursor, setIndexerCursor, prisma } from '../db'
 import { dispatchPriceUpdate } from '../webhookDispatcher'
@@ -52,6 +52,7 @@ export async function snapshotPool(pool: any, pair: WatchedPair): Promise<void>
 
     await prisma.poolSnapshot.create({
       data: {
+        network: activeNetwork,
         poolId: pool.id,
         assetA: pair.assetA.code,
         assetB: pair.assetB.code,

src/ingesters/snapshot.ts

@@ -1,4 +1,5 @@
 import { pgPool } from '../db'
+import { activeNetwork } from '../config'
 import { getActivePairs } from '../pairsRegistry'
 import { price_snapshots_total } from '../metrics'
 
@@ -36,23 +37,23 @@ export async function appendSnapshots(now: Date = new Date()): Promise<number> {
     `WITH latest AS (
        SELECT DISTINCT ON (pair_key) pair_key, price::numeric AS price
        FROM price_points
-       WHERE pair_key = ANY($1)
+       WHERE pair_key = ANY($1) AND network = $3
        ORDER BY pair_key, timestamp DESC
      ),
      vol AS (
        SELECT pair_key, SUM(base_volume::numeric) AS volume
        FROM price_points
-       WHERE pair_key = ANY($1)
+       WHERE pair_key = ANY($1) AND network = $3
          AND timestamp >= $2
          AND timestamp < $2 + INTERVAL '1 minute'
        GROUP BY pair_key
      )
-     INSERT INTO price_snapshots (pair, ts, price, volume)
-     SELECT l.pair_key, $2, l.price, COALESCE(v.volume, 0)
+     INSERT INTO price_snapshots (network, pair, ts, price, volume)
+     SELECT $3, l.pair_key, $2, l.price, COALESCE(v.volume, 0)
      FROM latest l
      LEFT JOIN vol v ON v.pair_key = l.pair_key
-     ON CONFLICT (pair, ts) DO NOTHING`,
-    [pairKeys, ts]
+     ON CONFLICT (network, pair, ts) DO NOTHING`,
+    [pairKeys, ts, activeNetwork]
   )
 
   const inserted = result.rowCount ?? 0

src/jobs/aggregateRefresh.ts

@@ -1,11 +1,11 @@
 import { Queue, Worker } from 'bullmq'
-import { config } from '../config'
+import { config, activeNetwork } from '../config'
 import { pgPool, prisma } from '../db'
 import { setCachedPrice } from '../redis'
 import { calculateVWAP, calculateOHLCV, getAggregatedPrice } from '../aggregator/vwap'
 import { getBestRoute } from '../aggregator/bestRoute'
 
-const QUEUE_NAME = 'aggregate-refresh'
+const QUEUE_NAME = `${activeNetwork}:aggregate-refresh`
 
 function redisConnection() {
   const url = process.env.REDIS_URL
@@ -60,9 +60,9 @@ export function startAggregateWorker() {
           if (vwap === 0) continue
 
           await prisma.priceAggregate.upsert({
-            where: { pairKey_window_bucket: { pairKey, window: w.key, bucket } },
+            where: { network_pairKey_window_bucket: { network: activeNetwork, pairKey, window: w.key, bucket } },
             create: {
-              pairKey, window: w.key, bucket,
+              network: activeNetwork, pairKey, window: w.key, bucket,
               vwap, sdexVwap: sdexVwap || null, ammVwap: ammVwap || null,
               volume: ohlcv.volume, tradeCount: ohlcv.tradeCount,
               openPrice: ohlcv.open || null, closePrice: ohlcv.close || null,