#Closes 220

openapi.json

@@ -563,6 +563,9 @@
                           "fee_bp": {
                             "type": "integer"
                           },
+                          "venue": {
+                            "type": "string"
+                          },
                           "timestamp": {
                             "type": "string",
                             "format": "date-time"

openapi.yaml

@@ -396,6 +396,8 @@ paths:
                           type: number
                         fee_bp:
                           type: integer
+                        venue:
+                          type: string
                         timestamp:
                           type: string
                           format: date-time
@@ -2601,3 +2603,4 @@ components:
             properties:
               message:
                 type: string
+

src/__tests__/discoveryCycle.test.ts

@@ -0,0 +1,69 @@
+import { describe, it, expect, vi, beforeEach } from "vitest"
+import { aquariusAdapter } from "../ingest/venues/aquarius"
+import { pools_added_total, pools_removed_total, register } from "../metrics"
+import { getActivePairs } from "../pairsRegistry"
+
+vi.mock("../pairsRegistry", () => ({
+  getActivePairs: vi.fn(),
+}))
+
+vi.mock("../db", () => ({ upsertPricePoints: vi.fn().mockResolvedValue(undefined) }))
+vi.mock("../webhookDispatcher", () => ({ dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined) }))
+vi.mock("../events", () => ({ publishPriceUpdate: vi.fn() }))
+vi.mock("../config", () => ({
+  config: { indexer: { pollIntervalMs: 5000 }, aquarius: { apiUrl: "https://test" } },
+  activeNetwork: "testnet",
+  getNetworkConfig: () => ({ aquarius: { enabled: true, apiUrl: "https://test" } }),
+}))
+
+describe("pool discovery cycle", () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    pools_added_total.reset()
+    pools_removed_total.reset()
+    if ((aquariusAdapter as any).getWatchedSet) {
+      (aquariusAdapter as any).getWatchedSet().clear()
+    }
+  })
+
+  it("adds and removes pools across discovery cycles with metrics", async () => {
+    const pair = { pairKey: "XLM/USDC", assetA: { code: "XLM", issuer: null }, assetB: { code: "USDC", issuer: "GABC" } }
+    ;(getActivePairs as any).mockReturnValue([pair])
+
+    const originalFetch = global.fetch
+    let callCount = 0
+    global.fetch = vi.fn().mockImplementation(() => {
+      callCount++
+      if (callCount === 1) {
+        return Promise.resolve({
+          ok: true,
+          json: async () => ({ results: [{ pool_hash: "pool1", reserves: ["100", "200"], total_shares: "10" }] }),
+        })
+      } else {
+        return Promise.resolve({
+          ok: true,
+          json: async () => ({ results: [{ pool_hash: "pool2", reserves: ["50", "100"], total_shares: "5" }] }),
+        })
+      }
+    }) as any
+
+    if (aquariusAdapter.runDiscoveryCycle) {
+      await aquariusAdapter.runDiscoveryCycle("testnet")
+    }
+    const watchedAfter1 = (aquariusAdapter as any).getWatchedSet ? (aquariusAdapter as any).getWatchedSet() : new Set()
+    expect(watchedAfter1.size).toBeGreaterThanOrEqual(1)
+
+    if (aquariusAdapter.runDiscoveryCycle) {
+      await aquariusAdapter.runDiscoveryCycle("testnet")
+    }
+    const watchedAfter2 = (aquariusAdapter as any).getWatchedSet ? (aquariusAdapter as any).getWatchedSet() : new Set()
+    expect(watchedAfter2.size).toBeGreaterThanOrEqual(1)
+
+    const addedMetric = await register.getSingleMetric("pools_added_total")
+    const removedMetric = await register.getSingleMetric("pools_removed_total")
+    expect(addedMetric).toBeDefined()
+    expect(removedMetric).toBeDefined()
+
+    global.fetch = originalFetch
+  })
+})

src/api/rest.ts

@@ -240,7 +240,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
           WHERE ids.pool_id IS NOT NULL
        )
        SELECT l.pool_id, l.asset_a, l.asset_b,
-              l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp
+              l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp, l.venue
          FROM ids
         CROSS JOIN LATERAL (
               SELECT * FROM pool_snapshots s
@@ -281,3 +281,4 @@ export async function registerRESTRoutes(app: FastifyInstance) {
     }
   )
 }
+

src/api/schemas.ts

@@ -705,3 +705,4 @@ export function installResponseValidation(app: FastifyInstance): void {
     }
   })
 }
+

src/ingest/venues/aquarius.ts

@@ -17,8 +17,13 @@ import {
   resetLastPrice,
   writeVenuePricePoints,
   startVenueAdapter,
+  createRediscoveryState,
+  runDiscoveryCycle,
+  startRediscoveryLoop,
+  getTargetKey,
 } from './runner'
 import { fetchWithTimeout } from '../../fetch'
+import { getActivePairs } from '../../pairsRegistry'
 
 const log = createLogger('ingester:aquarius')
 
@@ -40,6 +45,8 @@ export interface AquariusListResponse {
   results?: AquariusPool[]
 }
 
+const aquariusRediscoveryState = createRediscoveryState<AquariusPool>()
+
 export async function fetchAquariusPools(
   pair: WatchedPair,
   apiUrl: string = config.aquarius.apiUrl
@@ -144,6 +151,27 @@ export const aquariusAdapter: VenueAdapter<AquariusPool[], AquariusPool[]> = {
 
   _resetLastPrice,
   _getLastPrice,
+
+  async startRediscovery(intervalMs: number, network: NetworkName = activeNetwork): Promise<void> {
+    const discoverFn = async (pair: WatchedPair, net: NetworkName) => {
+      const pools = await fetchAquariusPools(pair, getNetworkConfig(net).aquarius.apiUrl)
+      return pools
+    }
+    await startRediscoveryLoop('Aquarius', aquariusAdapter, network, intervalMs, aquariusRediscoveryState, discoverFn)
+  },
+
+  getWatchedSet(): Set<any> {
+    return aquariusRediscoveryState.watched
+  },
+
+  async runDiscoveryCycle(network: NetworkName = activeNetwork): Promise<{ added: number; removed: number }> {
+    const pairs = getActivePairs()
+    const discoverFn = async (pair: WatchedPair, net: NetworkName) => {
+      const pools = await fetchAquariusPools(pair, getNetworkConfig(net).aquarius.apiUrl)
+      return pools
+    }
+    return runDiscoveryCycle('Aquarius', network, pairs, discoverFn, aquariusRediscoveryState)
+  },
 }
 
 export async function startAquariusIngester(network: NetworkName = activeNetwork): Promise<void> {

src/ingest/venues/index.ts

@@ -2,6 +2,7 @@ import { sdexAdapter } from './sdex'
 import { ammAdapter } from './amm'
 import { soroswapAdapter } from './soroswap'
 import { aquariusAdapter } from './aquarius'
+import { phoenixAdapter } from './phoenix'
 import type { VenueAdapter } from './types'
 
 export * from './types'
@@ -10,6 +11,7 @@ export { sdexAdapter, startSDEXIngester } from './sdex'
 export { ammAdapter, startAMMIngester } from './amm'
 export { soroswapAdapter, startSoroswapIngester } from './soroswap'
 export { aquariusAdapter, startAquariusIngester } from './aquarius'
+export { phoenixAdapter, startPhoenixIngester } from './phoenix'
 
 /**
  * Registry of all available venue adapters.
@@ -22,3 +24,4 @@ export const venueAdapters: VenueAdapter[] = [
   soroswapAdapter,
   aquariusAdapter,
 ]
+