feat(overlord): compute proposal's scores_total_value

package.json

@@ -47,7 +47,8 @@
     "ts-node": "^10.9.1",
     "turndown": "^7.2.0",
     "typescript": "^4.7.4",
-    "winston": "^3.8.2"
+    "winston": "^3.8.2",
+    "zod": "3.23.8"
   },
   "devDependencies": {
     "@snapshot-labs/eslint-config": "^0.1.0-beta.18",

src/constants.ts

@@ -2,7 +2,7 @@ export const CB = {
   FINAL: 1,
   PENDING_SYNC: 0, // Default db value, waiting from value from overlord
   PENDING_COMPUTE: -1,
-  PENDING_CLOSE: -2,
+  PENDING_FINAL: -2,
   INELIGIBLE: -10, // Payload format, can not compute
   ERROR_SYNC: -11 // Sync error from overlord, waiting for retry
 };

src/helpers/proposalsScoresValue.ts

@@ -0,0 +1,106 @@
+import { capture } from '@snapshot-labs/snapshot-sentry';
+import snapshot from '@snapshot-labs/snapshot.js';
+import { z } from 'zod';
+import db from './mysql';
+import { CB } from '../constants';
+
+type Proposal = {
+  id: string;
+  scoresState: string;
+  vpValueByStrategy: number[];
+  scoresByStrategy: number[][];
+};
+
+const REFRESH_INTERVAL = 10 * 1000;
+const BATCH_SIZE = 25;
+
+const proposalSchema = z
+  .object({
+    id: z.string(),
+    scoresState: z.string(),
+    vpValueByStrategy: z.array(z.number().finite()),
+    scoresByStrategy: z.array(z.array(z.number().finite()))
+  })
+  .refine(
+    data => {
+      if (data.scoresByStrategy.length === 0 || data.vpValueByStrategy.length === 0) {
+        return true;
+      }
+      // Ensure all scoresByStrategy arrays have the same length as vpValueByStrategy
+      return data.scoresByStrategy.every(
+        voteScores => voteScores.length === data.vpValueByStrategy.length
+      );
+    },
+    {
+      message: 'Array size mismatch: voteScores length does not match vpValueByStrategy length'
+    }
+  );
+
+async function getProposals(): Promise<Proposal[]> {
+  const query = `
+    SELECT id, scores_state, vp_value_by_strategy, scores_by_strategy
+    FROM proposals
+    WHERE cb = ?
+    ORDER BY created ASC
+    LIMIT ?
+  `;
+  const proposals = await db.queryAsync(query, [CB.PENDING_COMPUTE, BATCH_SIZE]);
+
+  return proposals.map((p: any) => ({
+    id: p.id,
+    scoresState: p.scores_state,
+    vpValueByStrategy: JSON.parse(p.vp_value_by_strategy),
+    scoresByStrategy: JSON.parse(p.scores_by_strategy)
+  }));
+}
+
+export function getScoresTotalValue(proposal: Proposal): number {
+  const { scoresByStrategy, vpValueByStrategy } = proposalSchema.parse(proposal);
+
+  return vpValueByStrategy.reduce((totalValue, strategyValue, strategyIndex) => {
+    const strategyTotal = scoresByStrategy.reduce(
+      (sum, voteScores) => sum + voteScores[strategyIndex],
+      0
+    );
+    return totalValue + strategyTotal * strategyValue;
+  }, 0);
+}
+
+async function refreshProposalsScoresTotalValue(proposals: Proposal[]) {
+  const query: string[] = [];
+  const params: any[] = [];
+
+  proposals.forEach(proposal => {
+    query.push('UPDATE proposals SET scores_total_value = ?, cb = ? WHERE id = ? LIMIT 1');
+
+    try {
+      const scoresTotalValue = getScoresTotalValue(proposal);
+      params.push(
+        scoresTotalValue,
+        proposal.scoresState === 'final' ? CB.FINAL : CB.PENDING_FINAL,
+        proposal.id
+      );
+    } catch (e) {
+      capture(e);
+      params.push(0, CB.INELIGIBLE, proposal.id);
+    }
+  });
+
+  if (query.length) {
+    await db.queryAsync(query.join(';'), params);
+  }
+}
+
+export default async function run() {
+  while (true) {
+    const proposals = await getProposals();
+
+    if (proposals.length) {
+      await refreshProposalsScoresTotalValue(proposals);
+    }
+
+    if (proposals.length < BATCH_SIZE) {
+      await snapshot.utils.sleep(REFRESH_INTERVAL);
+    }
+  }
+}

src/index.ts

@@ -6,6 +6,7 @@ import api from './api';
 import log from './helpers/log';
 import initMetrics from './helpers/metrics';
 import refreshModeration from './helpers/moderation';
+import refreshProposalsScoresValue from './helpers/proposalsScoresValue';
 import refreshProposalsVpValue from './helpers/proposalStrategiesValue';
 import rateLimit from './helpers/rateLimit';
 import shutter from './helpers/shutter';
@@ -22,6 +23,7 @@ async function startServer() {
   initLogger(app);
   refreshModeration();
   refreshProposalsVpValue();
+  refreshProposalsScoresValue();
 
   await initializeStrategies();
   refreshStrategies();

src/scores.ts

@@ -1,4 +1,5 @@
 import snapshot from '@snapshot-labs/snapshot.js';
+import { CB } from './constants';
 import { getVoteValue } from './helpers/entityValue';
 import log from './helpers/log';
 import db from './helpers/mysql';
@@ -77,7 +78,7 @@ async function updateVotesVp(votes: any[], vpState: string, proposalId: string)
   log.info(`[scores] updated votes vp, ${votesWithChange.length}/${votes.length} on ${proposalId}`);
 }
 
-async function updateProposalScores(proposalId: string, scores: any, votes: number) {
+async function updateProposalScores(proposal: any, scores: any, votes: number) {
   const ts = (Date.now() / 1e3).toFixed();
   const query = `
     UPDATE proposals
@@ -86,7 +87,8 @@ async function updateProposalScores(proposalId: string, scores: any, votes: numb
     scores_by_strategy = ?,
     scores_total = ?,
     scores_updated = ?,
-    votes = ?
+    votes = ?,
+    cb = ?
     WHERE id = ? LIMIT 1;
   `;
   await db.queryAsync(query, [
@@ -96,7 +98,8 @@ async function updateProposalScores(proposalId: string, scores: any, votes: numb
     scores.scores_total,
     ts,
     votes,
-    proposalId
+    proposal.cb === CB.PENDING_FINAL ? CB.PENDING_COMPUTE : proposal.cb,
+    proposal.id
   ]);
 }
 
@@ -184,7 +187,7 @@ export async function updateProposalAndVotes(proposalId: string, force = false)
     if (!isFinal) await updateVotesVp(votes, vpState, proposalId);
 
     // Store scores
-    await updateProposalScores(proposalId, results, votes.length);
+    await updateProposalScores(proposal, results, votes.length);
     log.info(
       `[scores] Proposal updated ${proposal.id}, ${proposal.space}, ${results.scores_state}, ${votes.length}`
     );

yarn.lock

@@ -6042,3 +6042,8 @@ yocto-queue@^0.1.0:
   version "0.1.0"
   resolved "https://registry.yarnpkg.com/yocto-queue/-/yocto-queue-0.1.0.tgz#0294eb3dee05028d31ee1a5fa2c556a6aaf10a1b"
   integrity sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==
+
+zod@3.23.8:
+  version "3.23.8"
+  resolved "https://registry.yarnpkg.com/zod/-/zod-3.23.8.tgz#e37b957b5d52079769fb8097099b592f0ef4067d"
+  integrity sha512-XBx9AXhXktjUqnepgTiE5flcKIYWi/rme0Eaj+5Y0lftuGBq+jyRu/md4WnuxqgP1ubdpNCsYEYPxrzVHD8d6g==