feat(prices): gate price source on USE_PRICE_SERVICE and stop serving table-backed prices

.env.example

@@ -21,6 +21,10 @@ YPRICE_API_X_SIGNER=
 YPRICE_API_X_SIGNATURE=
 PRICE_SERVICE_API_KEY=
 PRICE_SERVICE_URL=https://prices.yearn.dev
+# When true, indexer reads prices from yearn-prices and skips Postgres price table R/W.
+# Default false. Toggling requires a process restart — not a live flip.
+# When true, set PRICE_SERVICE_API_KEY.
+USE_PRICE_SERVICE=false
 
 REDIS_HOST=
 REDIS_PORT=

packages/ingest/extract/waveydb.ts

@@ -5,6 +5,7 @@ import { Processor } from 'lib/processor'
 import { Price, PriceSchema } from 'lib/types'
 import batchx from 'lib/batchx'
 import { getAddress } from 'viem'
+import { usePriceService } from '../prices'
 
 const db = new Pool({
   host: process.env.WAVEYDB_HOST,
@@ -32,6 +33,9 @@ export class WaveyDbExtractor implements Processor {
   }
 
   async extractPrices() {
+    // Service mode decommissions the price table; skip these writes entirely.
+    if (usePriceService()) return
+
     const result = await db.query(`
       SELECT 
         chain_id as "chainId",

packages/ingest/index.ts

@@ -11,6 +11,7 @@ import { Processor, ProcessorPool } from 'lib/processor'
 import { cache, chains, abisConfig, crons as cronsConfig, mq, sentry } from 'lib'
 import db from './db'
 import { camelToSnake } from 'lib/strings'
+import { usePriceService } from './prices'
 
 const exportsProcessor = (filePath: string): boolean => {
   const fileContent = fs.readFileSync(filePath, 'utf8')
@@ -79,7 +80,7 @@ function up() {
     abis,
   ]).then(() => {
 
-    console.log('🐒 ingest up')
+    console.log('🐒 ingest up', `USE_PRICE_SERVICE=${usePriceService()}`)
 
   }).catch(error => fatal('up', error))
 }

packages/ingest/prices.service.spec.ts

@@ -0,0 +1,78 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mqAdd = vi.fn()
+
+vi.mock('lib', () => ({
+  mq: { add: mqAdd, job: { load: { price: { name: 'price' } } } }
+}))
+
+vi.mock('lib/blocks', () => ({
+  getBlockTime: vi.fn(async () => 1700000000n),
+  getBlockNumber: vi.fn(async () => 1n)
+}))
+
+vi.mock('lib/cache', () => ({
+  cache: { wrap: (_key: string, fn: () => Promise<unknown>) => fn() }
+}))
+
+// `lib`'s barrel re-exports `../ingest/prices` (circular), and vitest.setup.ts
+// imports `lib` before this file's vi.mock calls run. Reset + re-import
+// dynamically so ./prices re-resolves against the mocks declared above.
+vi.resetModules()
+const { fetchErc20PriceUsd } = await import('./prices')
+
+const WETH = '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2' as const
+const CHAIN_ID = 137 // polygon — not in the on-chain `lens` map
+
+describe('fetchErc20PriceUsd (USE_PRICE_SERVICE=true)', () => {
+  const originalUsePriceService = process.env.USE_PRICE_SERVICE
+  const originalApiKey = process.env.PRICE_SERVICE_API_KEY
+
+  beforeEach(() => {
+    process.env.USE_PRICE_SERVICE = 'true'
+    process.env.PRICE_SERVICE_API_KEY = 'test-key'
+    mqAdd.mockReset()
+  })
+
+  afterEach(() => {
+    if (originalUsePriceService === undefined) delete process.env.USE_PRICE_SERVICE
+    else process.env.USE_PRICE_SERVICE = originalUsePriceService
+    if (originalApiKey === undefined) delete process.env.PRICE_SERVICE_API_KEY
+    else process.env.PRICE_SERVICE_API_KEY = originalApiKey
+    vi.unstubAllGlobals()
+  })
+
+  it('returns the price service value and never enqueues', async () => {
+    const fetchMock = vi.fn(async () => ({
+      ok: true,
+      json: async () => ({
+        coins: {
+          ['polygon:0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2']: {
+            symbol: 'WETH',
+            prices: [{ timestamp: 1700000000, price: 2, confidence: 1, source: 'defillama' }]
+          }
+        }
+      })
+    }))
+    vi.stubGlobal('fetch', fetchMock)
+
+    const { priceSource, priceUsd } = await fetchErc20PriceUsd(CHAIN_ID, WETH, 1n)
+
+    expect(priceSource).to.equal('priceservice')
+    expect(priceUsd).to.equal(2)
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+    expect(mqAdd).not.toHaveBeenCalled()
+  })
+
+  it('returns na when the service has nothing — no fallback, no enqueue', async () => {
+    const fetchMock = vi.fn(async () => ({ ok: false }))
+    vi.stubGlobal('fetch', fetchMock)
+
+    const { priceSource, priceUsd } = await fetchErc20PriceUsd(CHAIN_ID, WETH, 1n)
+
+    expect(priceSource).to.equal('na')
+    expect(priceUsd).to.equal(0)
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+    expect(mqAdd).not.toHaveBeenCalled()
+  })
+})

packages/ingest/prices.ts

@@ -15,6 +15,13 @@ export const lens = {
   [arbitrum.id]: '0x043518AB266485dC085a1DB095B8d9C2Fc78E9b9' as `0x${string}`
 }
 
+const LATEST_CACHE_TTL_MS = 30_000
+
+/** When true, indexer reads prices from yearn-prices and skips the Postgres price table. */
+export function usePriceService(): boolean {
+  return (process.env.USE_PRICE_SERVICE || '').trim().toLowerCase() === 'true'
+}
+
 export async function fetchErc20PriceUsd(chainId: number, token: `0x${string}`, blockNumber?: bigint, latest = false): Promise<{ priceUsd: number, priceSource: string }>{
   token = getAddress(token)
 
@@ -23,91 +30,116 @@ export async function fetchErc20PriceUsd(chainId: number, token: `0x${string}`,
     latest = true
   }
 
-  return cache.wrap(`fetchErc20PriceUsd:${chainId}:${token}:${blockNumber}`, async () => {
-    return await __fetchErc20PriceUsd(chainId, token, blockNumber!, latest)
-  }, 30_000)
+  return cache.wrap(
+    `fetchErc20PriceUsd:${chainId}:${token}:${blockNumber}`,
+    async () => __fetchErc20PriceUsd(chainId, token, blockNumber!, latest),
+    LATEST_CACHE_TTL_MS
+  )
 }
 
 async function __fetchErc20PriceUsd(chainId: number, token: `0x${string}`, blockNumber: bigint, latest = false) {
+  // USE_PRICE_SERVICE=true: price service is the only source — no table read/write,
+  // no fallbacks. Unknown price when the service has nothing.
+  if (usePriceService()) {
+    return (await fetchPriceServiceUsd(chainId, token, blockNumber)) ?? unknownPrice(chainId, token, blockNumber)
+  }
+  return __fetchErc20PriceUsdFromTable(chainId, token, blockNumber, latest)
+}
+
+/** Legacy path: read/write the Postgres price table (USE_PRICE_SERVICE=false, default). */
+async function __fetchErc20PriceUsdFromTable(chainId: number, token: `0x${string}`, blockNumber: bigint, latest = false) {
   let result: Price | undefined
 
-  if(latest) {
+  if (latest) {
     result = await fetchYDaemonPriceUsd(chainId, token, blockNumber)
-    await mq.add(mq.job.load.price, result)
-    if(result) return result
+    if (result) {
+      await mq.add(mq.job.load.price, result)
+      return result
+    }
   }
 
   result = await fetchDbPriceUsd(chainId, token, blockNumber)
-  if(result) return result
+  if (result) return result
 
   result = await fetchLensPriceUsd(chainId, token, blockNumber)
-  if(result) {
+  if (result) {
     await mq.add(mq.job.load.price, result)
     return result
   }
 
-  if(JSON.parse(process.env.YPRICE_ENABLED || 'false')) {
+  if (JSON.parse(process.env.YPRICE_ENABLED || 'false')) {
     result = await fetchYPriceUsd(chainId, token, blockNumber)
-    if(result) {
+    if (result) {
       await mq.add(mq.job.load.price, result)
       return result
     }
   }
 
-  if(!result) {
-    result = await fetchPriceServiceUsd(chainId, token, blockNumber)
-    if(result) {
-      await mq.add(mq.job.load.price, result)
-      return result
-    }
+  result = await fetchPriceServiceUsd(chainId, token, blockNumber)
+  if (result) {
+    await mq.add(mq.job.load.price, result)
+    return result
   }
 
   console.warn('🚨', 'no price', chainId, token, blockNumber)
-  const empty = { chainId, address: token, priceUsd: 0, priceSource: 'na', blockNumber, blockTime: await getBlockTime(chainId, blockNumber) }
+  const empty = await unknownPrice(chainId, token, blockNumber)
   await mq.add(mq.job.load.price, empty)
   return empty
 }
 
-const PRICE_SERVICE_CHAIN_NAMES: Record<number, string> = {
-  1: 'ethereum', 10: 'optimism', 100: 'xdai', 137: 'polygon',
+async function unknownPrice(chainId: number, token: `0x${string}`, blockNumber: bigint): Promise<Price> {
+  return {
+    chainId,
+    address: token,
+    priceUsd: 0,
+    priceSource: 'na',
+    blockNumber,
+    blockTime: await getBlockTime(chainId, blockNumber)
+  }
+}
+
+/** Must match price-service CHAIN_ID_TO_NAME (gnosis, not xdai). */
+export const PRICE_SERVICE_CHAIN_NAMES: Record<number, string> = {
+  1: 'ethereum', 10: 'optimism', 100: 'gnosis', 137: 'polygon',
   146: 'sonic', 250: 'fantom', 8453: 'base', 42161: 'arbitrum',
   80094: 'berachain', 747474: 'katana',
 }
 
 const PRICE_SERVICE_DEFAULT_URL = 'https://prices.yearn.dev'
 
 async function fetchPriceServiceUsd(chainId: number, token: `0x${string}`, blockNumber: bigint) {
-  if(!process.env.PRICE_SERVICE_API_KEY) return undefined
+  if (!process.env.PRICE_SERVICE_API_KEY) return undefined
   const chainName = PRICE_SERVICE_CHAIN_NAMES[chainId]
-  if(!chainName) return undefined
+  if (!chainName) return undefined
 
   const baseUrl = process.env.PRICE_SERVICE_URL || PRICE_SERVICE_DEFAULT_URL
 
   try {
     const blockTime = await getBlockTime(chainId, blockNumber)
     const coinId = `${chainName}:${token.toLowerCase()}`
     const coins = encodeURIComponent(JSON.stringify({ [coinId]: [Number(blockTime)] }))
-    const url = `${baseUrl}/api/prices/batchHistorical?source=defillama&coins=${coins}`
+    // No source= filter: service uses its default priority (defillama → … → enso).
+    const url = `${baseUrl}/api/prices/batchHistorical?coins=${coins}`
 
     const response = await fetch(url, {
       headers: { Authorization: `Bearer ${process.env.PRICE_SERVICE_API_KEY}` }
     })
-    if(!response.ok) return undefined
+    if (!response.ok) return undefined
 
     const data = await response.json() as { coins: Record<string, { symbol: string; prices: { timestamp: number; price: number; confidence: number; source: string }[] }> }
     const coinData = data.coins[coinId]
     const priceUsd = coinData?.prices?.[0]?.price
-    if(!priceUsd) return undefined
+    if (!priceUsd) return undefined
 
     return PriceSchema.parse({ chainId, address: token, priceUsd, priceSource: 'priceservice', blockNumber, blockTime })
-  } catch(error) {
+  } catch {
     console.warn('🚨', 'price service failed', chainId, token, blockNumber)
     return undefined
   }
 }
 
 async function fetchYPriceUsd(chainId: number, token: `0x${string}`, blockNumber: bigint) {
-  if(!process.env.YPRICE_API) return undefined
+  if (!process.env.YPRICE_API) return undefined
 
   try {
     const url = `${process.env.YPRICE_API}/get_price/${chainId}/${token}?block=${blockNumber}`
@@ -119,7 +151,7 @@ async function fetchYPriceUsd(chainId: number, token: `0x${string}`, blockNumber
     })
 
     const priceUsd = Number(await result.json())
-    if(priceUsd === 0) return undefined
+    if (priceUsd === 0) return undefined
 
     return PriceSchema.parse({
       chainId,
@@ -130,7 +162,7 @@ async function fetchYPriceUsd(chainId: number, token: `0x${string}`, blockNumber
       blockTime: await getBlockTime(chainId, blockNumber)
     })
 
-  } catch(error) {
+  } catch {
     console.warn('🚨', 'yprice failed', chainId, token, blockNumber)
     return undefined
   }
@@ -148,23 +180,23 @@ async function fetchDbPriceUsd(chainId: number, token: `0x${string}`, blockNumbe
     FROM price WHERE chain_id = $1 AND address = $2 AND block_number = $3`,
     [chainId, getAddress(token), blockNumber]
   )
-  if(result.rows.length === 0) return undefined
+  if (result.rows.length === 0) return undefined
   return PriceSchema.parse(result.rows[0])
 }
 
 async function fetchLensPriceUsd(chainId: number, token: `0x${string}`, blockNumber: bigint) {
-  if(!(chainId in lens)) return undefined
+  if (!(chainId in lens)) return undefined
 
   try {
     const priceUSDC = await rpcs.next(chainId, blockNumber).readContract({
       address: lens[chainId as keyof typeof lens],
       functionName: 'getPriceUsdcRecommended',
-      args: [ token ],
+      args: [token],
       abi: parseAbi(['function getPriceUsdcRecommended(address tokenAddress) view returns (uint256)']),
       blockNumber
     }) as bigint
 
-    if(priceUSDC === 0n) return undefined
+    if (priceUSDC === 0n) return undefined
 
     return PriceSchema.parse({
       chainId,
@@ -175,14 +207,14 @@ async function fetchLensPriceUsd(chainId: number, token: `0x${string}`, blockNum
       blockTime: await getBlockTime(chainId, blockNumber)
     })
 
-  } catch(error) {
+  } catch (error) {
     console.warn('🚨', 'lens price failed', error)
     return undefined
   }
 }
 
 async function fetchAllYDaemonPrices() {
-  if(!process.env.YDAEMON_API) throw new Error('!YDAEMON_API')
+  if (!process.env.YDAEMON_API) throw new Error('!YDAEMON_API')
   return cache.wrap('fetchAllYDaemonPrices', async () => {
     const url = `${process.env.YDAEMON_API}/prices/all?humanized=true`
     const result = await fetch(url)
@@ -212,7 +244,7 @@ async function fetchYDaemonPriceUsd(chainId: number, token: `0x${string}`, block
   try {
     const prices = await fetchAllYDaemonPrices()
     const price = prices[chainId.toString()]?.[token.toLowerCase()] || 0
-    if(isNaN(price)) return undefined
+    if (isNaN(price)) return undefined
     return PriceSchema.parse({
       chainId,
       address: token,
@@ -221,7 +253,7 @@ async function fetchYDaemonPriceUsd(chainId: number, token: `0x${string}`, block
       blockNumber,
       blockTime: await getBlockTime(chainId, blockNumber)
     })
-  } catch(error) {
+  } catch (error) {
     console.warn('🚨', 'ydaemon price failed', error)
     return undefined
   }

packages/ingest/prices.unit.spec.ts

@@ -0,0 +1,30 @@
+import { expect } from 'chai'
+import { PRICE_SERVICE_CHAIN_NAMES, usePriceService } from './prices'
+
+describe('prices helpers', () => {
+  const originalUsePriceService = process.env.USE_PRICE_SERVICE
+
+  afterEach(() => {
+    if (originalUsePriceService === undefined) delete process.env.USE_PRICE_SERVICE
+    else process.env.USE_PRICE_SERVICE = originalUsePriceService
+  })
+
+  it('defaults USE_PRICE_SERVICE to false', () => {
+    delete process.env.USE_PRICE_SERVICE
+    expect(usePriceService()).to.equal(false)
+  })
+
+  it('accepts true in any case', () => {
+    process.env.USE_PRICE_SERVICE = 'TRUE'
+    expect(usePriceService()).to.equal(true)
+  })
+
+  it('treats non-true values as false', () => {
+    process.env.USE_PRICE_SERVICE = '1'
+    expect(usePriceService()).to.equal(false)
+  })
+
+  it('maps chain 100 to gnosis (not xdai)', () => {
+    expect(PRICE_SERVICE_CHAIN_NAMES[100]).to.equal('gnosis')
+  })
+})