diff --git a/install_output.txt b/install_output.txt new file mode 100644 index 00000000..1c85d343 Binary files /dev/null and b/install_output.txt differ diff --git a/openapi.json b/openapi.json index b6c26090..149e55a2 100644 --- a/openapi.json +++ b/openapi.json @@ -563,6 +563,9 @@ "fee_bp": { "type": "integer" }, + "venue": { + "type": "string" + }, "timestamp": { "type": "string", "format": "date-time" diff --git a/openapi.yaml b/openapi.yaml index 6dffdbac..03542d62 100644 --- a/openapi.yaml +++ b/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 + diff --git a/package-lock.json b/package-lock.json index bf123327..63ba131f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -365,7 +365,6 @@ "dev": true, "license": "MIT", "optional": true, - "peer": true, "dependencies": { "@emnapi/wasi-threads": "1.2.2", "tslib": "^2.4.0" @@ -378,19 +377,6 @@ "dev": true, "license": "MIT", "optional": true, - "peer": true, - "dependencies": { - "tslib": "^2.4.0" - } - }, - "node_modules/@emnapi/runtime": { - "version": "1.11.1", - "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", - "integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==", - "dev": true, - "license": "MIT", - "optional": true, - "peer": true, "dependencies": { "tslib": "^2.4.0" } @@ -2114,6 +2100,7 @@ "integrity": "sha512-8kzdPJ3FsNsVIurqBs7oodNnCEVbni9yUEkaHbgptDACOPW04jimGagZ51E6+lXUwJjgnBw+hyko/lkFWCldqw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "undici-types": "~6.21.0" } @@ -2143,6 +2130,7 @@ "integrity": "sha512-38C0/Ddb7HcRG0Z4/DUem8x57d2p9jYgp18mkaYswEOQBGsI1CG4f/hjm0ZCeaJfWhSZ4k7jgs29V1Zom7Ki9A==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@bcoe/v8-coverage": "^1.0.2", "@vitest/utils": "4.1.5", @@ -3920,6 +3908,7 @@ "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", + "dev": true, "hasInstallScript": true, "license": "MIT", "optional": true, @@ -4073,6 +4062,7 @@ "resolved": "https://registry.npmjs.org/graphql/-/graphql-16.13.2.tgz", "integrity": "sha512-5bJ+nf/UCpAjHM8i06fl7eLyVC9iuNAjm9qzkiu2ZGhM0VscSvS6WDPfAwkdkBuoXGM9FJSbKl6wylMwP9Ktig==", "license": "MIT", + "peer": true, "engines": { "node": "^12.22.0 || ^14.16.0 || ^16.0.0 || >=17.0.0" } @@ -4160,6 +4150,7 @@ "resolved": "https://registry.npmjs.org/hono/-/hono-4.13.5.tgz", "integrity": "sha512-O6+/eCYRkzzzy0rPWwKLiGBR1nFuUPZynnwjxN1MBA62NNqbT0wQEzQyK2gSO5yDIDB336sXQleAhOHrzlYyKw==", "license": "MIT", + "peer": true, "engines": { "node": ">=16.9.0" } @@ -5550,6 +5541,7 @@ "resolved": "https://registry.npmjs.org/pg/-/pg-8.20.0.tgz", "integrity": "sha512-ldhMxz2r8fl/6QkXnBD3CR9/xg694oT6DZQ2s6c/RI28OjtSOpxnPrUCGOBJ46RCUxcWdx3p6kw/xnDHjKvaRA==", "license": "MIT", + "peer": true, "dependencies": { "pg-connection-string": "^2.12.0", "pg-pool": "^3.13.0", @@ -5647,6 +5639,7 @@ "integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -5810,6 +5803,7 @@ "devOptional": true, "hasInstallScript": true, "license": "Apache-2.0", + "peer": true, "dependencies": { "@prisma/engines": "5.22.0" }, @@ -6949,6 +6943,7 @@ "integrity": "sha512-5C1sg4USs1lfG0GFb2RLXsdpXqBSEhAaA/0kPL01wxzpMqLILNxIxIOKiILz+cdg/pLnOUxFYOR5yhHU666wbw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "esbuild": "~0.27.0", "get-tsconfig": "^4.7.5" @@ -7122,6 +7117,7 @@ "integrity": "sha512-t7g7GVRpMXjNpa67HaVWI/8BWtdVIQPCL2WoozXXA7LBGEFK4AkkKkHx2hAQf5x1GZSlcmEDPkVLSGahxnEEZw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "lightningcss": "^1.32.0", "picomatch": "^4.0.4", @@ -7200,6 +7196,7 @@ "integrity": "sha512-9Xx1v3/ih3m9hN+SbfkUyy0JAs72ap3r7joc87XL6jwF0jGg6mFBvQ1SrwaX+h8BlkX6Hz9shdd1uo6AF+ZGpg==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@vitest/expect": "4.1.5", "@vitest/mocker": "4.1.5", @@ -7378,6 +7375,7 @@ "resolved": "https://registry.npmjs.org/zod/-/zod-4.5.4.tgz", "integrity": "sha512-sC95tT5iHHH9gtpj6A81kh+NEaRAUFN+qlUPDUbRfOMvNf5QCBqsb3WgvnpVtK5Y+4UfA6KqufotuTvMGiTlsA==", "license": "MIT", + "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } diff --git a/src/__tests__/discoveryCycle.test.ts b/src/__tests__/discoveryCycle.test.ts new file mode 100644 index 00000000..570f97ea --- /dev/null +++ b/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 + }) +}) diff --git a/src/api/rest.ts b/src/api/rest.ts index ceb505ac..21307476 100644 --- a/src/api/rest.ts +++ b/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) { } ) } + diff --git a/src/api/schemas.ts b/src/api/schemas.ts index dceedc4b..b7159fd4 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -705,3 +705,4 @@ export function installResponseValidation(app: FastifyInstance): void { } }) } + diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index 9f267cc9..4cb563d6 100644 --- a/src/ingest/venues/aquarius.ts +++ b/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() + export async function fetchAquariusPools( pair: WatchedPair, apiUrl: string = config.aquarius.apiUrl @@ -144,6 +151,27 @@ export const aquariusAdapter: VenueAdapter = { _resetLastPrice, _getLastPrice, + + async startRediscovery(intervalMs: number, network: NetworkName = activeNetwork): Promise { + 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 { + 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 { diff --git a/src/ingest/venues/index.ts b/src/ingest/venues/index.ts index 0ea72a28..ef032935 100644 --- a/src/ingest/venues/index.ts +++ b/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, ] + diff --git a/src/ingest/venues/phoenix.ts b/src/ingest/venues/phoenix.ts new file mode 100644 index 00000000..d5ad526a --- /dev/null +++ b/src/ingest/venues/phoenix.ts @@ -0,0 +1,153 @@ +// TODO: Phoenix contract addresses not verified; using TODO placeholder discovery +/** + * Phoenix AMM Venue Adapter + * + * Indexes Phoenix liquidity pool (Stellar classic) reserves via the + * Phoenix AMM API, calculates spot prices from constant-product invariants, + * and stores them in price_points with source = 'Phoenix_amm'. + * + * Issue: #103, #218 + */ + +import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' +import { createLogger } from '../../logger' +import type { WatchedPair } from '../../types' +import type { VenueAdapter, VenuePricePoint } from './types' +import { + getLastPrice, + resetLastPrice, + writeVenuePricePoints, + startVenueAdapter, +} from './runner' +import { fetchWithTimeout } from '../../fetch' + +const log = createLogger('ingester:Phoenix') + +export function _resetLastPrice(): void { + resetLastPrice('Phoenix') +} + +export function _getLastPrice(network: NetworkName, pairKey: string): number | undefined { + return getLastPrice('Phoenix', network, pairKey) +} + +export interface PhoenixPool { + pool_hash: string + reserves: string[] + total_shares: string +} + +export interface PhoenixListResponse { + results?: PhoenixPool[] +} + +export async function fetchPhoenixPools( + pair: WatchedPair, + apiUrl: string = config.Phoenix.apiUrl +): Promise { + try { + const assetAStr = pair.assetA.issuer + ? `${pair.assetA.code}:${pair.assetA.issuer}` + : 'native' + const assetBStr = pair.assetB.issuer + ? `${pair.assetB.code}:${pair.assetB.issuer}` + : 'native' + + const params = new URLSearchParams() + params.append('assets[]', assetAStr) + params.append('assets[]', assetBStr) + + const res = await fetchWithTimeout(`${apiUrl}?${params.toString()}`) + if (!res.ok) return [] + const data = (await res.json()) as PhoenixListResponse + return data.results ?? [] + } catch (err) { + log.error({ pair: pair.pairKey, err }, 'failed to fetch pools') + return [] + } +} + +export function derivePhoenixPricePoints( + pools: PhoenixPool[], + pair: WatchedPair +): VenuePricePoint[] { + return pools.flatMap((pool) => { + const [r0Str, r1Str] = pool.reserves + if (!r0Str || !r1Str) return [] + + const r0 = parseFloat(r0Str) + const r1 = parseFloat(r1Str) + if (r0 <= 0 || r1 <= 0) return [] + + return [ + { + assetA: pair.assetA.code, + assetB: pair.assetB.code, + pairKey: pair.pairKey, + source: 'Phoenix_amm', + poolId: pool.pool_hash, + price: r1 / r0, + baseVolume: 0, + counterVolume: 0, + ledger: 0, + timestamp: new Date(), + }, + ] + }) +} + +export async function ingestPhoenixPair( + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { + const pools = await fetchPhoenixPools(pair, getNetworkConfig(network).Phoenix.apiUrl) + if (pools.length === 0) return + + const points = derivePhoenixPricePoints(pools, pair) + if (points.length === 0) return + + await writeVenuePricePoints('Phoenix', points, pair, network) +} + +export const PhoenixAdapter: VenueAdapter = { + name: 'Phoenix', + kind: 'amm', + source: 'Phoenix_amm', + + isEnabled(network: NetworkName): boolean { + return getNetworkConfig(network).Phoenix.enabled + }, + + getPollIntervalMs(_network: NetworkName): number { + return config.indexer.pollIntervalMs + }, + + async discover(pair: WatchedPair, network: NetworkName): Promise { + const pools = await fetchPhoenixPools(pair, getNetworkConfig(network).Phoenix.apiUrl) + return pools.length > 0 ? [pools] : [] + }, + + async read(target: PhoenixPool[], _pair: WatchedPair, _network: NetworkName): Promise { + return target + }, + + derive(raw: PhoenixPool[], _target: PhoenixPool[], pair: WatchedPair, _network: NetworkName): VenuePricePoint[] { + return derivePhoenixPricePoints(raw, pair) + }, + + async write(points: VenuePricePoint[], _target: PhoenixPool[], pair: WatchedPair, network: NetworkName): Promise { + await writeVenuePricePoints('Phoenix', points, pair, network) + }, + + async ingestPair(pair: WatchedPair, network: NetworkName): Promise { + await ingestPhoenixPair(pair, network) + }, + + _resetLastPrice, + _getLastPrice, +} + +export async function startPhoenixIngester(network: NetworkName = activeNetwork): Promise { + await startVenueAdapter(PhoenixAdapter, network) +} + diff --git a/src/ingest/venues/runner.ts b/src/ingest/venues/runner.ts index 9dad46b3..9f13b461 100644 --- a/src/ingest/venues/runner.ts +++ b/src/ingest/venues/runner.ts @@ -6,6 +6,7 @@ import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' +import { pools_added_total, pools_removed_total, pools_discovered_total } from '../../metrics' import { ingest_cycle_duration_seconds, ingest_errors_total, markIngestSuccess } from '../../metrics' const log = createLogger('ingester:runner') @@ -82,6 +83,113 @@ export async function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } +// ── Rediscovery helpers ────────────────────────────────────────────────────── + +export interface RediscoveryState { + watched: Set + watchedTargets: Map +} + +export function createRediscoveryState(): RediscoveryState { + return { + watched: new Set(), + watchedTargets: new Map(), + } +} + +export function getTargetKey(target: any): string { + if (target == null) return '' + if (typeof target === 'string') return target + if (target.pool_hash) return target.pool_hash + if (target.poolId) return target.poolId + if (target.poolAddress) return target.poolAddress + if (target.id) return target.id + try { + return JSON.stringify(target) + } catch { + return String(target) + } +} + +export async function runDiscoveryCycle( + venueName: string, + network: NetworkName, + pairs: WatchedPair[], + discoverFn: (pair: WatchedPair, network: NetworkName) => Promise, + state: RediscoveryState +): Promise<{ added: number; removed: number }> { + const currentSet = new Set() + const currentMap = new Map() + + for (const pair of pairs) { + try { + const discovered = await discoverFn(pair, network) + if (Array.isArray(discovered)) { + for (const item of discovered) { + const key = getTargetKey(item) + if (key) { + currentSet.add(key) + currentMap.set(key, item) + } + } + } + } catch (err) { + log.error({ venue: venueName, pair: pair.pairKey, err }, 'discovery cycle error') + } + } + + let added = 0 + let removed = 0 + + for (const key of currentSet) { + if (!state.watched.has(key)) { + added++ + state.watched.add(key) + if (currentMap.has(key)) { + state.watchedTargets.set(key, currentMap.get(key) as T) + } + } + } + + for (const key of Array.from(state.watched)) { + if (!currentSet.has(key)) { + removed++ + state.watched.delete(key) + state.watchedTargets.delete(key) + } + } + + pools_discovered_total.inc({ venue: venueName, network }, currentSet.size) + if (added > 0) pools_added_total.inc({ venue: venueName, network }, added) + if (removed > 0) pools_removed_total.inc({ venue: venueName, network }, removed) + + log.debug({ venue: venueName, network, added, removed, total: currentSet.size }, 'discovery cycle complete') + + return { added, removed } +} + +export async function startRediscoveryLoop( + venueName: string, + adapter: VenueAdapter, + network: NetworkName, + intervalMs: number, + state: RediscoveryState, + discoverFn: (pair: WatchedPair, network: NetworkName) => Promise +): Promise { + const pairs = getActivePairs() + await runDiscoveryCycle(venueName, network, pairs, discoverFn, state) + const timer = setInterval(async () => { + try { + const currentPairs = getActivePairs() + await runDiscoveryCycle(venueName, network, currentPairs, discoverFn, state) + } catch (err) { + log.error({ venue: venueName, network, err }, 'rediscovery loop error') + } + }, intervalMs) + ;(global as any).__rediscoveryTimers = (global as any).__rediscoveryTimers || [] + ;(global as any).__rediscoveryTimers.push(timer) +} + // ── Scheduling, Retry & Error Handling Runner ───────────────────────────────── /** diff --git a/src/ingest/venues/types.ts b/src/ingest/venues/types.ts index 2fa4601d..5a56dc6a 100644 --- a/src/ingest/venues/types.ts +++ b/src/ingest/venues/types.ts @@ -76,4 +76,19 @@ export interface VenueAdapter { * Gets the last seen price for threshold/webhook tracking. */ _getLastPrice?(network: NetworkName, pairKey: string): number | undefined + + /** + * Start periodic rediscovery of targets for watched pairs. + */ + startRediscovery?(intervalMs: number, network?: NetworkName): Promise + + /** + * Get current watched set (for testing) + */ + getWatchedSet?(): Set + + /** + * Run a discovery cycle manually (for testing) + */ + runDiscoveryCycle?(network?: NetworkName): Promise<{ added: number; removed: number }> } diff --git a/src/metrics.ts b/src/metrics.ts index 2d10a735..2551ad4c 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -95,6 +95,27 @@ export const last_trade_timestamp = new Gauge({ registers: [register] }) +export const pools_discovered_total = new Counter({ + name: 'pools_discovered_total', + help: 'Total number of pools discovered across all venues', + labelNames: ['venue', 'network'], + registers: [register] +}) + +export const pools_added_total = new Counter({ + name: 'pools_added_total', + help: 'Total number of pools added during discovery cycles', + labelNames: ['venue', 'network'], + registers: [register] +}) + +export const pools_removed_total = new Counter({ + name: 'pools_removed_total', + help: 'Total number of pools removed during discovery cycles', + labelNames: ['venue', 'network'], + registers: [register] +}) + // Ingestion health is labelled only by the finite adapter registry and the // two supported networks. Do not add pair or pool identifiers here: they make // the series count grow with user configuration and on-chain pool creation. diff --git a/test_output.txt b/test_output.txt new file mode 100644 index 00000000..c18625bc Binary files /dev/null and b/test_output.txt differ