feat: add top earnings participant endpoint

README.md

@@ -32,6 +32,10 @@ Base URL: http://stats.filspark.com/
 
   http://stats.filspark.com/participants/change-rates
 
+- `GET /participants/top-earning`
+
+  http://stats.filspark.com/participants/top-earning
+
 - `GET /participant/:address/scheduled-rewards?address=<address>&from=<day>&to=<day>`
 
   http://stats.filspark.com/participant/0x000000000000000000000000000000000000dEaD/scheduled-rewards

stats/lib/platform-routes.js

@@ -3,6 +3,7 @@ import {
   fetchDailyStationCount,
   fetchMonthlyStationCount,
   fetchDailyRewardTransfers,
+  fetchTopEarningParticipants,
   fetchParticipantsWithTopMeasurements,
   fetchDailyStationAcceptedMeasurementCount
 } from './platform-stats-fetchers.js'
@@ -33,6 +34,8 @@ export const handlePlatformRoutes = async (req, res, pgPools) => {
     await respond(pgPools.evaluate, fetchDailyStationAcceptedMeasurementCount)
   } else if (req.method === 'GET' && url === '/participants/top-measurements') {
     await respond(pgPools.evaluate, fetchParticipantsWithTopMeasurements)
+  } else if (req.method === 'GET' && url === '/participants/top-earning') {
+    await respond(pgPools.stats, fetchTopEarningParticipants)
   } else if (req.method === 'GET' && url === '/transfers/daily') {
     await respond(pgPools.stats, fetchDailyRewardTransfers)
   } else {

stats/lib/platform-stats-fetchers.js

@@ -1,5 +1,5 @@
 import assert from 'http-assert'
-import { getDailyDistinctCount, getMonthlyDistinctCount } from './request-helpers.js'
+import { getDailyDistinctCount, getMonthlyDistinctCount, today } from './request-helpers.js'
 
 /** @typedef {import('@filecoin-station/spark-stats-db').Queryable} Queryable */
 
@@ -71,3 +71,31 @@ export const fetchDailyRewardTransfers = async (pgPool, filter) => {
   `, [filter.from, filter.to])
   return rows
 }
+
+/**
+ * @param {Queryable} pgPool
+ * @param {import('./typings.js').DateRangeFilter} filter
+ */
+export const fetchTopEarningParticipants = async (pgPool, 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(`
+    WITH latest_scheduled_rewards AS (
+      SELECT DISTINCT ON (participant_address) participant_address, scheduled_rewards
+      FROM daily_scheduled_rewards
+      ORDER BY participant_address, day DESC
+    )
+    SELECT
+      COALESCE(drt.to_address, lsr.participant_address) as participant_address,
+      COALESCE(SUM(drt.amount), 0) + COALESCE(lsr.scheduled_rewards, 0) as total_rewards
+    FROM daily_reward_transfers drt
+    FULL OUTER JOIN latest_scheduled_rewards lsr
+      ON drt.to_address = lsr.participant_address
+    WHERE (drt.day >= $1 AND drt.day <= $2) OR drt.day IS NULL
+    GROUP BY COALESCE(drt.to_address, lsr.participant_address), lsr.scheduled_rewards
+    ORDER BY total_rewards DESC
+  `, [filter.from, filter.to])
+  return rows
+}

stats/lib/request-helpers.js

@@ -5,7 +5,7 @@ import pg from 'pg'
 
 /** @typedef {import('@filecoin-station/spark-stats-db').Queryable} Queryable */
 
-const getDayAsISOString = (d) => d.toISOString().split('T')[0]
+export const getDayAsISOString = (d) => d.toISOString().split('T')[0]
 
 export const today = () => getDayAsISOString(new Date())
 

stats/test/platform-routes.test.js

@@ -6,6 +6,7 @@ import { getPgPools } from '@filecoin-station/spark-stats-db'
 
 import { assertResponseStatus, getPort } from './test-helpers.js'
 import { createHandler } from '../lib/handler.js'
+import { getDayAsISOString, today } from '../lib/request-helpers.js'
 
 const STATION_STATS = { stationId: 'station1', participantAddress: 'f1abcdef', inetGroup: 'group1' }
 
@@ -47,6 +48,7 @@ describe('Platform Routes HTTP request handler', () => {
     await pgPools.evaluate.query('REFRESH MATERIALIZED VIEW top_measurement_participants_yesterday_mv')
 
     await pgPools.stats.query('DELETE FROM daily_reward_transfers')
+    await pgPools.stats.query('DELETE FROM daily_scheduled_rewards')
   })
 
   describe('GET /stations/daily', () => {
@@ -247,6 +249,105 @@ describe('Platform Routes HTTP request handler', () => {
       ])
     })
   })
+
+  describe('GET /participants/top-earning', () => {
+    const yesterdayDate = new Date()
+    yesterdayDate.setDate(yesterdayDate.getDate() - 1)
+    const yesterday = getDayAsISOString(yesterdayDate)
+    console.log('yesterday', yesterday)
+
+    const oneWeekAgoDate = new Date()
+    oneWeekAgoDate.setDate(oneWeekAgoDate.getDate() - 7)
+    const oneWeekAgo = getDayAsISOString(oneWeekAgoDate)
+    console.log('oneWeekAgo', oneWeekAgo)
+
+    const setupScheduledRewardsData = async () => {
+      await pgPools.stats.query(`
+        INSERT INTO daily_scheduled_rewards (day, participant_address, scheduled_rewards)
+        VALUES 
+          ('${yesterday}', 'address1', 10),
+          ('${yesterday}', 'address2', 20),
+          ('${yesterday}', 'address3', 30),
+          ('${today()}', 'address1', 15),
+          ('${today()}', 'address2', 25),
+          ('${today()}', 'address3', 35)
+      `)
+    }
+    it('returns top earning participants for the given date range', async () => {
+      // First two dates should be ignored
+      await givenDailyRewardTransferMetrics(pgPools.stats, '2024-01-09', [
+        { toAddress: 'address1', amount: 100, lastCheckedBlock: 1 },
+        { toAddress: 'address2', amount: 100, lastCheckedBlock: 1 },
+        { toAddress: 'address3', amount: 100, lastCheckedBlock: 1 }
+      ])
+      await givenDailyRewardTransferMetrics(pgPools.stats, '2024-01-10', [
+        { toAddress: 'address1', amount: 100, lastCheckedBlock: 1 }
+      ])
+
+      // These should be included in the results
+      await givenDailyRewardTransferMetrics(pgPools.stats, oneWeekAgo, [
+        { toAddress: 'address2', amount: 150, lastCheckedBlock: 1 },
+        { toAddress: 'address1', amount: 50, lastCheckedBlock: 1 }
+      ])
+      await givenDailyRewardTransferMetrics(pgPools.stats, today(), [
+        { toAddress: 'address3', amount: 200, lastCheckedBlock: 1 },
+        { toAddress: 'address2', amount: 100, lastCheckedBlock: 1 }
+      ])
+
+      // Set up scheduled rewards data
+      await setupScheduledRewardsData()
+
+      const res = await fetch(
+        new URL(
+          `/participants/top-earning?from=${oneWeekAgo}&to=${today()}`,
+          baseUrl
+        ), {
+          redirect: 'manual'
+        }
+      )
+      await assertResponseStatus(res, 200)
+      const topEarners = await res.json()
+      assert.deepStrictEqual(topEarners, [
+        { participant_address: 'address2', total_rewards: '275' },
+        { participant_address: 'address3', total_rewards: '235' },
+        { participant_address: 'address1', total_rewards: '65' }
+      ])
+    })
+    it('returns top earning participants for the given date range with no existing reward transfers', async () => {
+      await setupScheduledRewardsData()
+
+      await givenDailyRewardTransferMetrics(pgPools.stats, today(), [
+        { toAddress: 'address1', amount: 100, lastCheckedBlock: 1 }
+      ])
+
+      const res = await fetch(
+        new URL(
+          `/participants/top-earning?from=${oneWeekAgo}&to=${today()}`,
+          baseUrl
+        ), {
+          redirect: 'manual'
+        }
+      )
+      await assertResponseStatus(res, 200)
+      const topEarners = await res.json()
+      assert.deepStrictEqual(topEarners, [
+        { participant_address: 'address1', total_rewards: '115' },
+        { participant_address: 'address3', total_rewards: '35' },
+        { participant_address: 'address2', total_rewards: '25' }
+      ])
+    })
+    it('returns 400 if the date range end is not today', async () => {
+      const res = await fetch(
+        new URL(
+          `/participants/top-earning?from=${oneWeekAgo}&to=${yesterday}`,
+          baseUrl
+        ), {
+          redirect: 'manual'
+        }
+      )
+      await assertResponseStatus(res, 400)
+    })
+  })
 })
 
 const givenDailyStationMetrics = async (pgPoolEvaluate, day, stationStats) => {