Use @fastify/fastify-postgres. CheckerNetwork/roadmap#220

db/package.json

@@ -13,7 +13,7 @@
   },
   "dependencies": {
     "@filecoin-station/spark-evaluate": "^1.2.0",
-    "pg": "^8.14.0",
+    "pg": "^8.13.3",
     "postgrator": "^8.0.0"
   },
   "standard": {

stats/lib/app.js

@@ -35,8 +35,6 @@ export const createApp = async ({
   await app.register(fastifyPostgres, {
     connectionString: EVALUATE_DB_URL,
     name: 'evaluate',
-
-
   })
 
   app.register(cors, {

stats/lib/platform-routes.js

@@ -21,28 +21,27 @@ export const addPlatformRoutes = (app) => {
     app.addHook('onSend', filterOnSendHook)
 
     app.get('/stations/daily', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchDailyStationCount(request.server.pg.evaluate, request.filter))
     })
     app.get('/stations/monthly', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchMonthlyStationCount(request.server.pg.evaluate, request.filter))
+      reply.send(await fetchMonthlyStationCount(request.server.pg, request.filter))
     })
     app.get('/stations/desktop/daily', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchDailyDesktopUsers(request.server.pg.stats, request.filter))
+      reply.send(await fetchDailyDesktopUsers(request.server.pg, request.filter))
     })
     app.get('/measurements/daily', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchDailyStationMeasurementCounts(request.server.pg.evaluate, request.filter))
+      reply.send(await fetchDailyStationMeasurementCounts(request.server.pg, request.filter))
     })
     app.get('/participants/top-measurements', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchParticipantsWithTopMeasurements(request.server.pg.evaluate, request.filter))
+      reply.send(await fetchParticipantsWithTopMeasurements(request.server.pg, request.filter))
     })
     app.get('/participants/top-earning', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchTopEarningParticipants(request.server.pg.stats, request.filter))    })
+      reply.send(await fetchTopEarningParticipants(request.server.pg, request.filter))    })
 
     app.get('/participants/accumulative/daily', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchAccumulativeDailyParticipantCount(request.server.pg.evaluate, request.filter))
+      reply.send(await fetchAccumulativeDailyParticipantCount(request.server.pg, request.filter))
     })
     app.get('/transfers/daily', async (/** @type {RequestWithFilter} */ request, reply) => {
-      reply.send(await fetchDailyRewardTransfers(request.server.pg.stats, request.filter))
+      reply.send(await fetchDailyRewardTransfers(request.server.pg, request.filter))
     })
   })
 

stats/lib/platform-stats-fetchers.js

@@ -1,16 +1,24 @@
 import assert from 'http-assert'
 import { today, yesterday } from './request-helpers.js'
 
-/** @typedef {import('./typings.js').DateRangeFilter} DateRangeFilter */
+/** 
+  @typedef {import('./typings.js').DateRangeFilter} DateRangeFilter 
+  @typedef {import('@filecoin-station/spark-stats-db').Queryable} Queryable 
+  @typedef {import('./typings.js').FastifyPg} FastifyPg 
+*/
+
+/**
+ * @param {FastifyPg} pg - Fastify pg object with database connections
+ */
 
 const ONE_DAY = 24 * 60 * 60 * 1000
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDailyStationCount = async (pg, filter) => {
-  const { rows } = await pg.query(`
+  const { rows } = await pg.evaluate.query(`
     SELECT day::TEXT, station_count
     FROM daily_platform_stats
     WHERE day >= $1 AND day <= $2
@@ -20,11 +28,11 @@ export const fetchDailyStationCount = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchMonthlyStationCount = async (pg, filter) => {
-  const { rows } = await pg.query(`
+  const { rows } = await pg.evaluate.query(`
     SELECT month::TEXT, station_count
     FROM monthly_active_station_count
     WHERE
@@ -36,11 +44,11 @@ export const fetchMonthlyStationCount = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDailyStationMeasurementCounts = async (pg, filter) => {
-  const { rows } = await pg.query(`
+  const { rows } = await pg.evaluate.query(`
     SELECT day::TEXT, accepted_measurement_count, total_measurement_count
     FROM daily_platform_stats
     WHERE day >= $1 AND day <= $2
@@ -50,22 +58,22 @@ export const fetchDailyStationMeasurementCounts = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchParticipantsWithTopMeasurements = async (pg, 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 pg.query(`
+  return (await pg.evaluate.query(`
     SELECT day::TEXT, participant_address, station_count, accepted_measurement_count, inet_group_count
     FROM top_measurement_participants_yesterday_mv
   `)).rows
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDailyRewardTransfers = async (pg, filter) => {
@@ -74,7 +82,7 @@ export const fetchDailyRewardTransfers = async (pg, filter) => {
     400,
     'Date range must be 31 days max'
   )
-  const { rows } = await pg.query(`
+  const { rows } = await pg.stats.query(`
     SELECT day::TEXT, to_address, amount
     FROM daily_reward_transfers
     WHERE day >= $1 AND day <= $2
@@ -99,11 +107,11 @@ export const fetchDailyRewardTransfers = async (pg, filter) => {
 }
 
 /**
- *   @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchAccumulativeDailyParticipantCount = async (pg, filter) => {
-  const { rows } = await pg.query(`
+  const { rows } = await pg.evaluate.query(`
     WITH first_appearance AS (
       SELECT participant_id, MIN(day) as day
       FROM daily_participants
@@ -126,15 +134,15 @@ export const fetchAccumulativeDailyParticipantCount = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchTopEarningParticipants = async (pg, 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 pg.query(`
+  const { rows } = await pg.stats.query(`
     WITH latest_scheduled_rewards AS (
       SELECT DISTINCT ON (participant_address) participant_address, scheduled_rewards
       FROM daily_scheduled_rewards
@@ -154,10 +162,10 @@ export const fetchTopEarningParticipants = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
-*/
+ * @param {FastifyPg} pg 
+ */
 export const fetchParticipantsSummary = async (pg) => {
-  const { rows } = await pg.query(`
+  const { rows } = await pg.evaluate.query(`
     SELECT COUNT(DISTINCT participant_id) FROM daily_participants
   `)
   return {
@@ -166,11 +174,11 @@ export const fetchParticipantsSummary = async (pg) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDailyDesktopUsers = async (pg, filter) => {
-  const { rows } = await pg.query(`
+  const { rows } = await pg.stats.query(`
     SELECT
       day::TEXT,
       user_count
@@ -180,4 +188,4 @@ export const fetchDailyDesktopUsers = async (pg, filter) => {
   [filter.from, filter.to])
 
   return rows
-}
+}

stats/lib/routes.js

@@ -25,6 +25,8 @@ import {
 
 
 
+
+
 export const addRoutes = (app, SPARK_API_BASE_URL) => {
   app.register(async app => {
     app.addHook('preHandler', filterPreHandlerHook)

stats/lib/stats-fetchers.js

@@ -1,7 +1,13 @@
+/**  
+   @typedef {import('@filecoin-station/spark-stats-db').Queryable} Queryable 
+   @typedef {import('./typings.js').FastifyPg} FastifyPg 
+ */
+
 /**
- * @param {Object} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter & {nonZero?: 'true'}} filter
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  */
+
 export const fetchRetrievalSuccessRate = async (pg, filter) => {
   // Fetch the "day" (DATE) as a string (TEXT) to prevent node-postgres for converting it into
   // a JavaScript Date with a timezone, as that could change the date one day forward or back.
@@ -35,7 +41,7 @@ export const fetchRetrievalSuccessRate = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {any} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDailyDealStats = async (pg, filter) => {
@@ -62,7 +68,7 @@ export const fetchDailyDealStats = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDealSummary = async (pg, filter) => {
@@ -114,7 +120,7 @@ export const fetchMonthlyParticipants = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchParticipantChangeRates = async (pg, filter) => {
@@ -182,7 +188,7 @@ export const fetchParticipantChangeRates = async (pg, filter) => {
 }
 
 /**
- * @param {Object} pg 
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  * @param {string} address
  */
@@ -196,7 +202,7 @@ export const fetchParticipantScheduledRewards = async (pg, { from, to }, address
 }
 
 /**
- * @param {Object} pg
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  * @param {string} address
  */
@@ -211,7 +217,7 @@ export const fetchParticipantRewardTransfers = async (pg, { from, to }, address)
 
 /**
  * Fetches the retrieval stats summary for all miners for given date range.
- * @param {Object} pg 
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchMinersRSRSummary = async (pg, filter) => {
@@ -245,7 +251,7 @@ export const fetchMinersRSRSummary = async (pg, filter) => {
 
 /**
  * Fetches the retrieval stats summary for a single miner for given date range.
- * @param {Object} pg 
+ * @param {FastifyPg} pg - Fastify pg object with database connections
  * @param {import('./typings.js').DateRangeFilter} filter
  * @param {string} minerId
  */
@@ -301,7 +307,7 @@ export const fetchDailyRetrievalResultCodes = async (pg, filter) => {
 
 /**
  * Fetches daily global retrieval time statistics
- * @param {Object} pg 
+ * @param {FastifyPg} pg 
  * @param {import('./typings.js').DateRangeFilter} filter
  */
 export const fetchDailyRetrievalTimings = async (pg, filter) => {
@@ -322,7 +328,7 @@ export const fetchDailyRetrievalTimings = async (pg, filter) => {
 
 /**
  * Fetches per miner daily retrieval time statistics
- * @param {Object} pg 
+ * @param {{stats: Queryable, evaluate: Queryable}} pg
  * @param {import('./typings.js').DateRangeFilter} filter
  * @param {string} minerId
  */
@@ -346,9 +352,10 @@ export const fetchDailyMinerRetrievalTimings = async (pg, { from, to }, minerId)
 
 /**
  * Fetches retrieval time statistics summary for all miners for given date range.
- * @param {Object} pg 
+ /**
+ * @param {FastifyPg} pg
  * @param {import('./typings.js').DateRangeFilter} filter
- */
+*/
 export const fetchMinersTimingsSummary = async (pg, { from, to }) => {
   const { rows } = await pg.evaluate.query(`
     SELECT
@@ -366,10 +373,10 @@ export const fetchMinersTimingsSummary = async (pg, { from, to }) => {
 }
 
 /**
- * Fetches the retrieval stats summary for all clients for given date range.
- * @param {Object} pg
- * @param {import('./typings.js').DateRangeFilter} filter
- */
+* Fetches the retrieval stats summary for all clients for given date range.
+* @param {FastifyPg} pg
+* @param {import('./typings.js').DateRangeFilter} filter
+*/
 export const fetchClientsRSRSummary = async (pg, filter) => {
   const { rows } = await pg.evaluate.query(`
     SELECT 
@@ -399,7 +406,7 @@ export const fetchClientsRSRSummary = async (pg, filter) => {
 
 /**
  * Fetches the retrieval stats summary for a single client for given date range.
- * @param {Object} pg
+ * @param {{stats: Queryable, evaluate: Queryable}} pg
  * @param {import('./typings.js').DateRangeFilter} filter
  * @param {string} clientId
  */

stats/lib/typings.d.ts

@@ -1,10 +1,16 @@
 import { FastifyRequest } from 'fastify'
+import { PostgresDb } from '@fastify/postgres'
+import { Queryable } from '@filecoin-station/spark-stats-db'
+
+
+export type FastifyPg = PostgresDb & Record<string, PostgresDb>
 
 export interface DateRangeFilter {
   from: string;
   to: string;
 }
 
+
 export type RequestWithFilter = FastifyRequest<{
   Querystring: { from: string, to: string }
 }>