move GET /participant/:address/reward-transfers to platform stats

stats/lib/handler.js

@@ -9,7 +9,6 @@ import {
   fetchMonthlyParticipants,
   fetchParticipantChangeRates,
   fetchParticipantScheduledRewards,
-  fetchParticipantRewardTransfers,
   fetchRetrievalSuccessRate,
   fetchDealSummary
 } from './stats-fetchers.js'
@@ -96,8 +95,6 @@ const handler = async (req, res, pgPools) => {
     await respond(fetchParticipantChangeRates)
   } else if (req.method === 'GET' && segs[0] === 'participant' && segs[1] && segs[2] === 'scheduled-rewards') {
     await respond(fetchParticipantScheduledRewards, segs[1])
-  } else if (req.method === 'GET' && segs[0] === 'participant' && segs[1] && segs[2] === 'reward-transfers') {
-    await respond(fetchParticipantRewardTransfers, segs[1])
   } else if (req.method === 'GET' && url === '/miners/retrieval-success-rate/summary') {
     await respond(fetchMinersRSRSummary)
   } else if (await handlePlatformRoutes(req, res, pgPools)) {

stats/lib/platform-routes.js

@@ -5,39 +5,54 @@ import {
   fetchDailyRewardTransfers,
   fetchTopEarningParticipants,
   fetchParticipantsWithTopMeasurements,
-  fetchDailyStationAcceptedMeasurementCount
+  fetchDailyStationAcceptedMeasurementCount,
+  fetchParticipantRewardTransfers
 } from './platform-stats-fetchers.js'
 
-const createRespondWithFetchFn = (pathname, searchParams, res) => (pgPool, fetchFn) => {
-  return getStatsWithFilterAndCaching(
-    pathname,
-    {},
-    searchParams,
-    res,
-    pgPool,
-    fetchFn
-  )
-}
+/** @typedef {import('@filecoin-station/spark-stats-db').PgPools} PgPools */
+/** @typedef {import('./typings.js').DateRangeFilter} DateRangeFilter */
+
+/**
+ * @param {string} pathname
+ * @param {URLSearchParams} searchParams
+ * @param {import('node:http').ServerResponse} res
+ * @param {PgPools} pgPools
+ * @returns {(fetchFn: (pgPools: PgPools, filter: DateRangeFilter, pathVariables: object) => Promise<any>, pathParams?: object) => Promise<void>}
+ */
+const createRespondWithFetchFn =
+(pathname, searchParams, res, pgPools) =>
+  (fetchFn, pathParams) => {
+    return getStatsWithFilterAndCaching(
+      pathname,
+      pathParams,
+      searchParams,
+      res,
+      pgPools,
+      fetchFn
+    )
+  }
 
 export const handlePlatformRoutes = async (req, res, pgPools) => {
   // Caveat! `new URL('//foo', 'http://127.0.0.1')` would produce "http://foo/" - not what we want!
   const { pathname, searchParams } = new URL(`http://127.0.0.1${req.url}`)
   const segs = pathname.split('/').filter(Boolean)
   const url = `/${segs.join('/')}`
-  const respond = createRespondWithFetchFn(pathname, searchParams, res)
+  const respond = createRespondWithFetchFn(pathname, searchParams, res, pgPools)
 
   if (req.method === 'GET' && url === '/stations/daily') {
-    await respond(pgPools.evaluate, fetchDailyStationCount)
+    await respond(fetchDailyStationCount)
   } else if (req.method === 'GET' && url === '/stations/monthly') {
-    await respond(pgPools.evaluate, fetchMonthlyStationCount)
+    await respond(fetchMonthlyStationCount)
   } else if (req.method === 'GET' && url === '/measurements/daily') {
-    await respond(pgPools.evaluate, fetchDailyStationAcceptedMeasurementCount)
+    await respond(fetchDailyStationAcceptedMeasurementCount)
   } else if (req.method === 'GET' && url === '/participants/top-measurements') {
-    await respond(pgPools.evaluate, fetchParticipantsWithTopMeasurements)
+    await respond(fetchParticipantsWithTopMeasurements)
   } else if (req.method === 'GET' && url === '/participants/top-earning') {
-    await respond(pgPools.stats, fetchTopEarningParticipants)
+    await respond(fetchTopEarningParticipants)
   } else if (req.method === 'GET' && url === '/transfers/daily') {
-    await respond(pgPools.stats, fetchDailyRewardTransfers)
+    await respond(fetchDailyRewardTransfers)
+  } else if (req.method === 'GET' && segs[0] === 'participant' && segs[1] && segs[2] === 'reward-transfers') {
+    await respond(fetchParticipantRewardTransfers, segs[1])
   } else {
     return false
   }

stats/lib/platform-stats-fetchers.js

@@ -6,40 +6,40 @@ import {
   yesterday
 } from './request-helpers.js'
 
-/** @typedef {import('@filecoin-station/spark-stats-db').Queryable} Queryable */
+/** @typedef {import('@filecoin-station/spark-stats-db').PgPools} PgPools */
 
 /**
- * @param {Queryable} pgPool
+ * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter
  */
-export const fetchDailyStationCount = async (pgPool, filter) => {
+export const fetchDailyStationCount = async (pgPools, filter) => {
   return await getDailyDistinctCount({
-    pgPool,
+    pgPool: pgPools.evaluate,
     table: 'daily_stations',
     column: 'station_id',
     filter
   })
 }
 
 /**
- * @param {Queryable} pgPool
+ * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter
  */
-export const fetchMonthlyStationCount = async (pgPool, filter) => {
+export const fetchMonthlyStationCount = async (pgPools, filter) => {
   return await getMonthlyDistinctCount({
-    pgPool,
+    pgPool: pgPools.evaluate,
     table: 'daily_stations',
     column: 'station_id',
     filter
   })
 }
 
 /**
- * @param {Queryable} pgPool
+ * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter
  */
-export const fetchDailyStationAcceptedMeasurementCount = async (pgPool, filter) => {
-  const { rows } = await pgPool.query(`
+export const fetchDailyStationAcceptedMeasurementCount = async (pgPools, filter) => {
+  const { rows } = await pgPools.evaluate.query(`
     SELECT day::TEXT, SUM(accepted_measurement_count) as accepted_measurement_count
     FROM daily_stations
     WHERE day >= $1 AND day <= $2
@@ -50,23 +50,23 @@ export const fetchDailyStationAcceptedMeasurementCount = async (pgPool, filter)
 }
 
 /**
- * @param {Queryable} pgPool
+ * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter
  */
-export const fetchParticipantsWithTopMeasurements = async (pgPool, filter) => {
+export const fetchParticipantsWithTopMeasurements = async (pgPools, filter) => {
   assert(filter.to === filter.from, 400, 'Multi-day queries are not supported for this endpoint')
   assert(filter.to === yesterday(), 400, 'filter.to must be set to yesterday, other values are not supported yet')
   // Ignore the filter for this query
   // Get the top measurement stations from the Materialized View
-  return (await pgPool.query('SELECT * FROM top_measurement_participants_yesterday_mv')).rows
+  return (await pgPools.evaluate.query('SELECT * FROM top_measurement_participants_yesterday_mv')).rows
 }
 
 /**
- * @param {Queryable} pgPool
+ * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter
  */
-export const fetchDailyRewardTransfers = async (pgPool, filter) => {
-  const { rows } = await pgPool.query(`
+export const fetchDailyRewardTransfers = async (pgPools, filter) => {
+  const { rows } = await pgPools.stats.query(`
     SELECT day::TEXT, SUM(amount) as amount
     FROM daily_reward_transfers
     WHERE day >= $1 AND day <= $2
@@ -77,15 +77,15 @@ export const fetchDailyRewardTransfers = async (pgPool, filter) => {
 }
 
 /**
- * @param {Queryable} pgPool
+ * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter
  */
-export const fetchTopEarningParticipants = async (pgPool, filter) => {
+export const fetchTopEarningParticipants = async (pgPools, filter) => {
   // The query combines "transfers until filter.to" with "latest scheduled rewards as of today".
   // As a result, it produces incorrect result if `to` is different from `now()`.
   // See https://github.com/filecoin-station/spark-stats/pull/170#discussion_r1664080395
   assert(filter.to === today(), 400, 'filter.to must be today, other values are not supported')
-  const { rows } = await pgPool.query(`
+  const { rows } = await pgPools.stats.query(`
     WITH latest_scheduled_rewards AS (
       SELECT DISTINCT ON (participant_address) participant_address, scheduled_rewards
       FROM daily_scheduled_rewards
@@ -103,3 +103,17 @@ export const fetchTopEarningParticipants = async (pgPool, filter) => {
   `, [filter.from, filter.to])
   return rows
 }
+
+/**
+ * @param {PgPools} pgPools
+ * @param {import('./typings.js').DateRangeFilter} filter
+ * @param {string} address
+ */
+export const fetchParticipantRewardTransfers = async (pgPools, { from, to }, address) => {
+  const { rows } = await pgPools.stats.query(`
+    SELECT day::TEXT, amount
+    FROM daily_reward_transfers
+    WHERE to_address = $1 AND day >= $2 AND day <= $3
+  `, [address, from, to])
+  return rows
+}

stats/lib/stats-fetchers.js

@@ -166,20 +166,6 @@ export const fetchParticipantScheduledRewards = async (pgPools, { from, to }, ad
   return rows
 }
 
-/**
- * @param {PgPools} pgPools
- * @param {import('./typings.js').DateRangeFilter} filter
- * @param {string} address
- */
-export const fetchParticipantRewardTransfers = async (pgPools, { from, to }, address) => {
-  const { rows } = await pgPools.stats.query(`
-    SELECT day::TEXT, amount
-    FROM daily_reward_transfers
-    WHERE to_address = $1 AND day >= $2 AND day <= $3
-  `, [address, from, to])
-  return rows
-}
-
 /**
  * @param {PgPools} pgPools
  * @param {import('./typings.js').DateRangeFilter} filter

stats/test/handler.test.js

@@ -530,7 +530,7 @@ describe('HTTP request handler', () => {
 
 /**
  *
- * @param {import('../lib/platform-stats-fetchers.js').Queryable} pgPool
+ * @param {import('../../db/typings.js').Queryable} pgPool
  * @param {object} data
  * @param {string} data.day
  * @param {string} [data.minerId]