feat: launch ingesters per enabled network in the bootstrap (#117)

.env.example

@@ -23,6 +23,13 @@ PRICE_CACHE_TTL=10
 # This drives all config.horizon / config.rpc / config.soroswap shortcut paths.
 STELLAR_NETWORK=testnet
 
+# Comma-separated list of networks the ingesters run for, e.g. "testnet,mainnet".
+# Each venue ingester (SDEX, AMM, Soroswap, Snapshot, Aquarius) is started once
+# per network listed here, each instance independently fault-isolated.
+# When unset (or nothing valid remains after parsing) it falls back to just
+# STELLAR_NETWORK, so a single-network node starts one instance of each ingester.
+ENABLED_NETWORKS=testnet
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Per-network Stellar configuration
 #

src/__tests__/aquariusIngester.test.ts

@@ -12,7 +12,14 @@ const mocks = vi.hoisted(() => ({
   },
 }))
 
-vi.mock('../config', () => ({ config: mocks.config }))
+vi.mock('../config', () => ({
+  config: mocks.config,
+  activeNetwork: 'testnet',
+  // startAquariusIngester / ingestAquariusPair resolve the per-network Aquarius
+  // block via getNetworkConfig(network); the mock returns the same shape for
+  // whichever network is asked for.
+  getNetworkConfig: () => mocks.config,
+}))
 vi.mock('../pairsRegistry', () => mocks.pairsRegistry)
 vi.mock('../db', () => ({ upsertPricePoints: vi.fn().mockResolvedValue(undefined) }))
 vi.mock('../webhookDispatcher', () => ({ dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined) }))

src/__tests__/enabledNetworks.test.ts

@@ -0,0 +1,51 @@
+import { describe, it, expect, afterEach, vi } from 'vitest'
+
+// activeNetwork is derived from STELLAR_NETWORK at config-module load; pin it so
+// the fallback path is deterministic regardless of the ambient env.
+vi.mock('../config', () => ({ activeNetwork: 'testnet' }))
+
+import { getEnabledNetworks } from '../network/enabledNetworks'
+
+const ORIGINAL = process.env.ENABLED_NETWORKS
+
+afterEach(() => {
+  if (ORIGINAL === undefined) delete process.env.ENABLED_NETWORKS
+  else process.env.ENABLED_NETWORKS = ORIGINAL
+})
+
+describe('getEnabledNetworks', () => {
+  it('falls back to the active network when ENABLED_NETWORKS is unset', () => {
+    delete process.env.ENABLED_NETWORKS
+    expect(getEnabledNetworks()).toEqual(['testnet'])
+  })
+
+  it('falls back to the active network when ENABLED_NETWORKS is blank', () => {
+    process.env.ENABLED_NETWORKS = '   '
+    expect(getEnabledNetworks()).toEqual(['testnet'])
+  })
+
+  it('parses an explicit multi-network list, preserving first-seen order', () => {
+    process.env.ENABLED_NETWORKS = 'mainnet,testnet'
+    expect(getEnabledNetworks()).toEqual(['mainnet', 'testnet'])
+  })
+
+  it('trims whitespace and lowercases each entry', () => {
+    process.env.ENABLED_NETWORKS = '  Testnet , MAINNET '
+    expect(getEnabledNetworks()).toEqual(['testnet', 'mainnet'])
+  })
+
+  it('de-duplicates repeated networks', () => {
+    process.env.ENABLED_NETWORKS = 'mainnet,mainnet,testnet,mainnet'
+    expect(getEnabledNetworks()).toEqual(['mainnet', 'testnet'])
+  })
+
+  it('filters out unrecognised network names and empty segments', () => {
+    process.env.ENABLED_NETWORKS = 'testnet,,futurenet,'
+    expect(getEnabledNetworks()).toEqual(['testnet'])
+  })
+
+  it('falls back to the active network when nothing valid remains', () => {
+    process.env.ENABLED_NETWORKS = 'futurenet,localnet'
+    expect(getEnabledNetworks()).toEqual(['testnet'])
+  })
+})

src/index.ts

@@ -8,7 +8,8 @@ if (!process.env.DIRECT_DATABASE_URL && process.env.DATABASE_URL) {
 import cors from '@fastify/cors'
 import compress from '@fastify/compress'
 import rateLimit from '@fastify/rate-limit'
-import { config } from './config'
+import { config, type NetworkName } from './config'
+import { getEnabledNetworks } from './network/enabledNetworks'
 import { redis } from './redis'
 import { pgPool } from './db'
 import { registerRESTRoutes } from './api/rest'
@@ -174,20 +175,26 @@ async function main() {
   }
 
   // ── Ingesters (run in background — infinite loops) ────────────────────────
-  // Each ingester is independently fault-isolated via restartIngester.
-  // A crash in the Soroswap ingester cannot take down SDEX or AMM.
-  console.log('[lens] Starting ingesters...')
-  const restartIngester = (name: string, fn: () => Promise<void>) => {
+  // Each ingester is independently fault-isolated via restartIngester, keyed by
+  // the (venue, network) pair: a crash in, say, the Soroswap ingester on
+  // mainnet only restarts that one instance and cannot affect SDEX/AMM or the
+  // ingesters running on testnet.
+  const restartIngester = (name: string, network: NetworkName, fn: () => Promise<void>) => {
     fn().catch(err => {
-      console.error(`[lens] ${name} ingester crashed, restarting in 10s:`, err.message)
-      setTimeout(() => restartIngester(name, fn), 10_000)
+      console.error(`[lens] ${name}/${network} ingester crashed, restarting in 10s:`, err.message)
+      setTimeout(() => restartIngester(name, network, fn), 10_000)
     })
   }
-  restartIngester('SDEX', startSDEXIngester)
-  restartIngester('AMM', startAMMIngester)
-  restartIngester('Soroswap', startSoroswapIngester)
-  restartIngester('Snapshot', startSnapshotIngester)
-  restartIngester('Aquarius', startAquariusIngester)
+
+  const enabledNetworks = getEnabledNetworks()
+  console.log(`[lens] Starting ingesters for network(s): ${enabledNetworks.join(', ')}`)
+  for (const network of enabledNetworks) {
+    restartIngester('SDEX', network, () => startSDEXIngester(network))
+    restartIngester('AMM', network, () => startAMMIngester(network))
+    restartIngester('Soroswap', network, () => startSoroswapIngester(network))
+    restartIngester('Snapshot', network, () => startSnapshotIngester(network))
+    restartIngester('Aquarius', network, () => startAquariusIngester(network))
+  }
 
   console.log(`[lens] Watching ${getActivePairs().length} pairs: ${getActivePairs().map(p => p.pairKey).join(', ')}`)
 }

src/ingest/venues/aquarius.ts

@@ -8,7 +8,7 @@
  * Issue: #103
  */
 
-import { config } from '../../config'
+import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config'
 import { getActivePairs } from '../../pairsRegistry'
 import { upsertPricePoints } from '../../db'
 import { dispatchPriceUpdate } from '../../webhookDispatcher'
@@ -52,8 +52,11 @@ export async function fetchAquariusPools(
   }
 }
 
-export async function ingestAquariusPair(pair: WatchedPair): Promise<void> {
-  const pools = await fetchAquariusPools(pair)
+export async function ingestAquariusPair(
+  pair: WatchedPair,
+  network: NetworkName = activeNetwork,
+): Promise<void> {
+  const pools = await fetchAquariusPools(pair, getNetworkConfig(network).aquarius.apiUrl)
   if (pools.length === 0) return
 
   const points = pools.flatMap(pool => {
@@ -98,16 +101,16 @@ async function sleep(ms: number) {
   return new Promise(r => setTimeout(r, ms))
 }
 
-export async function startAquariusIngester(): Promise<void> {
-  if (!config.aquarius.enabled) {
-    console.log('[aquarius] Aquarius is disabled on this network — ingester not started')
+export async function startAquariusIngester(network: NetworkName = activeNetwork): Promise<void> {
+  if (!getNetworkConfig(network).aquarius.enabled) {
+    console.log(`[aquarius] Aquarius is disabled on ${network} — ingester not started`)
     return
   }
 
-  console.log(`[aquarius] Starting Aquarius AMM ingester for ${getActivePairs().length} pairs`)
+  console.log(`[aquarius] Starting Aquarius AMM ingester for ${getActivePairs().length} pairs on ${network}`)
   while (true) {
     for (const pair of getActivePairs()) {
-      await ingestAquariusPair(pair)
+      await ingestAquariusPair(pair, network)
     }
     await sleep(config.indexer.pollIntervalMs)
   }

src/ingesters/snapshot.ts

@@ -1,5 +1,5 @@
 import { pgPool } from '../db'
-import { activeNetwork } from '../config'
+import { activeNetwork, type NetworkName } from '../config'
 import { getActivePairs } from '../pairsRegistry'
 import { price_snapshots_total } from '../metrics'
 
@@ -22,8 +22,15 @@ export function floorToMinute(date: Date): Date {
  * total base_volume traded during the minute that just closed. Pairs with no
  * price history yet are skipped (we don't fabricate a price). Returns the
  * number of rows inserted.
+ *
+ * Snapshot rows are network-scoped (the `(network, pair, ts)` primary key), so
+ * `network` selects both which `price_points` are read and which network the
+ * inserted `price_snapshots` rows are tagged with.
  */
-export async function appendSnapshots(now: Date = new Date()): Promise<number> {
+export async function appendSnapshots(
+  now: Date = new Date(),
+  network: NetworkName = activeNetwork,
+): Promise<number> {
   const pairs = getActivePairs()
   if (pairs.length === 0) return 0
 
@@ -53,7 +60,7 @@ export async function appendSnapshots(now: Date = new Date()): Promise<number> {
      FROM latest l
      LEFT JOIN vol v ON v.pair_key = l.pair_key
      ON CONFLICT (network, pair, ts) DO NOTHING`,
-    [pairKeys, ts, activeNetwork]
+    [pairKeys, ts, network]
   )
 
   const inserted = result.rowCount ?? 0
@@ -70,19 +77,19 @@ function sleep(ms: number) {
  * wall-clock minute boundary so snapshot timestamps land on :00 seconds and the
  * volume window cleanly covers the minute that just elapsed.
  */
-export async function startSnapshotIngester(): Promise<void> {
-  console.log(`[snapshot] Starting 1-minute snapshot ingester for ${getActivePairs().length} pairs`)
+export async function startSnapshotIngester(network: NetworkName = activeNetwork): Promise<void> {
+  console.log(`[snapshot] Starting 1-minute snapshot ingester for ${getActivePairs().length} pairs on ${network}`)
 
   while (true) {
     // Sleep until the next minute boundary.
     const msToNextMinute = 60_000 - (Date.now() % 60_000)
     await sleep(msToNextMinute)
 
     try {
-      const n = await appendSnapshots()
-      if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s)`)
+      const n = await appendSnapshots(new Date(), network)
+      if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s) on ${network}`)
     } catch (err) {
-      console.error('[snapshot] Error appending snapshots:', (err as Error).message)
+      console.error(`[snapshot] Error appending snapshots on ${network}:`, (err as Error).message)
     }
   }
 }

src/network/enabledNetworks.ts

@@ -0,0 +1,30 @@
+import { activeNetwork, type NetworkName } from '../config'
+
+const KNOWN_NETWORKS: readonly NetworkName[] = ['testnet', 'mainnet']
+
+/**
+ * The set of Stellar networks this process should ingest, driven by the
+ * comma-separated `ENABLED_NETWORKS` env var (e.g. `ENABLED_NETWORKS=testnet,mainnet`).
+ *
+ * Values are trimmed, lowercased, de-duplicated, and filtered to the known
+ * network names. When `ENABLED_NETWORKS` is unset — or nothing valid remains
+ * after filtering — this falls back to just the currently active network
+ * (`STELLAR_NETWORK`, via {@link activeNetwork}), so a single-network
+ * deployment starts exactly one instance of each ingester rather than one per
+ * network.
+ *
+ * Order follows first appearance in the env var; the fallback returns
+ * `[activeNetwork]`.
+ */
+export function getEnabledNetworks(): NetworkName[] {
+  const raw = process.env.ENABLED_NETWORKS
+  if (raw && raw.trim()) {
+    const parsed = raw
+      .split(',')
+      .map(s => s.trim().toLowerCase())
+      .filter((s): s is NetworkName => (KNOWN_NETWORKS as readonly string[]).includes(s))
+    const deduped = [...new Set(parsed)]
+    if (deduped.length > 0) return deduped
+  }
+  return [activeNetwork]
+}