Merge branch 'master' into feat-pull-overriding-strategies-from-strategies-list

Author: wa0x6e

docs/cb-flow.md

@@ -8,6 +8,7 @@
 | -1 | `PENDING_COMPUTE` | Strategy values synced. Waiting for `scores_total_value` computation | Default. Waiting for `vp_value` computation |
 | -2 | `PENDING_FINAL` | Score value computed, but proposal still active | Value computed, but `vp_state` not final |
 | 1 | `FINAL` | Fully computed | Fully computed |
+| -3 | `PENDING_DELETE` | N/A | Proposal deleted, vote awaiting async cleanup |
 | -10 | `INELIGIBLE` | Invalid payload format, cannot compute (permanent) | Invalid data, cannot compute (permanent) |
 | -11 | `ERROR_SYNC` | Overlord sync failed, will be retried | N/A |
 
@@ -54,14 +55,23 @@ flowchart TD
     C -->|No| E["cb = -2<br>PENDING_FINAL"]
     E -->|New vote on same proposal<br>triggers scores.ts<br>vp recalculated| B
 
+    B -->|delete-proposal| H["cb = -3<br>PENDING_DELETE"]
+    D -->|delete-proposal| H
+    E -->|delete-proposal| H
+    G -->|delete-proposal| H
+    H -->|deleteProposalVotes.ts<br>async cleanup| I([Vote Deleted])
+
     classDef user fill:#4a90d9,color:#fff
     classDef async fill:#e8833a,color:#fff
 
     class A user
     class F,C async
-    class B,D,E,G default
+    class B,D,E,G,H default
+    class I default
 
     linkStyle 0 stroke:#4a90d9
     linkStyle 1,2,3,4,5 stroke:#e8833a
     linkStyle 6 stroke:#4a90d9
+    linkStyle 7,8,9,10 stroke:#4a90d9
+    linkStyle 11 stroke:#e8833a
 ```

jest.config.ts

@@ -4,6 +4,7 @@
  */
 
 export default {
+  roots: ['<rootDir>/test'],
   clearMocks: true,
   collectCoverage: true,
   collectCoverageFrom: ['./src/**'],

package.json

@@ -3,7 +3,7 @@
   "version": "0.1.3",
   "license": "MIT",
   "scripts": {
-    "lint": "eslint src/ --ext .ts",
+    "lint": "eslint src/ test/ --ext .ts",
     "lint:fix": "yarn lint --fix",
     "typecheck": "tsc --noEmit",
     "build": "tsc",

src/constants.ts

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

src/helpers/deleteProposalVotes.ts

@@ -0,0 +1,80 @@
+import snapshot from '@snapshot-labs/snapshot.js';
+import log from './log';
+import db from './mysql';
+import { CB } from '../constants';
+
+type Vote = {
+  id: string;
+  voter: string;
+  space: string;
+};
+
+const REFRESH_INTERVAL = 60 * 1000;
+const BATCH_SIZE = 100;
+
+async function getVotesPendingDeletion(): Promise<Vote[]> {
+  return db.queryAsync(`SELECT id, voter, space FROM votes WHERE cb = ? LIMIT ?`, [
+    CB.PENDING_DELETE,
+    BATCH_SIZE
+  ]);
+}
+
+async function processVotes(votes: Vote[]) {
+  const query: string[] = [];
+  const params: (string | number | string[])[] = [];
+
+  // Update spaces vote_count
+  const grouped = new Map<string, number>();
+  for (const vote of votes) {
+    grouped.set(vote.space, (grouped.get(vote.space) || 0) + 1);
+  }
+  for (const [space, count] of grouped) {
+    query.push('UPDATE spaces SET vote_count = GREATEST(vote_count - ?, 0) WHERE id = ?');
+    params.push(count, space);
+  }
+
+  // Delete votes
+  query.push('DELETE FROM votes WHERE id IN (?) AND cb = ?');
+  params.push(
+    votes.map(v => v.id),
+    CB.PENDING_DELETE
+  );
+
+  // Refresh leaderboard from remaining votes (idempotent)
+  const pairs = new Set(votes.map(v => `${v.voter}:${v.space}`));
+  if (pairs.size) {
+    const pairPlaceholders = Array.from(pairs)
+      .map(() => '(?, ?)')
+      .join(', ');
+    query.push(
+      `UPDATE leaderboard l
+        SET vote_count = (SELECT COUNT(*) FROM votes v WHERE v.voter = l.user AND v.space = l.space),
+            vp_value = COALESCE((
+              SELECT SUM(v.vp_value) FROM votes v WHERE v.voter = l.user AND v.space = l.space
+            ), 0)
+        WHERE (l.user, l.space) IN (${pairPlaceholders})`
+    );
+    for (const pair of pairs) {
+      const [voter, space] = pair.split(':');
+      params.push(voter, space);
+    }
+  }
+
+  await db.queryAsync(query.join(';'), params);
+}
+
+export default async function run() {
+  while (true) {
+    const votes = await getVotesPendingDeletion();
+
+    if (votes.length) {
+      log.info(`[deleteProposalVotes] ${votes.length} votes to delete`);
+      await processVotes(votes);
+    }
+
+    if (votes.length < BATCH_SIZE) {
+      log.info('[deleteProposalVotes] sleeping');
+      await snapshot.utils.sleep(REFRESH_INTERVAL);
+    }
+  }
+}

src/helpers/shutter.ts

@@ -6,6 +6,7 @@ import express from 'express';
 import log from './log';
 import db from './mysql';
 import { getIp, jsonParse, jsonRpcRequest, rpcError, rpcSuccess } from './utils';
+import { CB } from '../constants';
 import { updateProposalAndVotes } from '../scores';
 
 init().then(() => log.info('[shutter] init'));
@@ -58,8 +59,8 @@ export async function setProposalKey(params) {
     const ts = (Date.now() / 1e3).toFixed();
     if (!proposal || ts < proposal.end) return false;
 
-    query = 'SELECT id, choice FROM votes WHERE proposal = ?';
-    const votes = await db.queryAsync(query, [proposal.id]);
+    query = 'SELECT id, choice FROM votes WHERE proposal = ? AND cb != ?';
+    const votes = await db.queryAsync(query, [proposal.id, CB.PENDING_DELETE]);
 
     const sqlParams: string[] = [];
     let sqlQuery = '';

src/helpers/votesVpValue.ts

@@ -9,6 +9,8 @@ type ProposalVpValues = Map<string, { cb: number; vpValueByStrategy: number[] }>
 
 type Datum = {
   id: string;
+  voter: string;
+  space: string;
   vpState: string;
   vpByStrategy: number[];
   vpValueByStrategy: number[];
@@ -22,6 +24,8 @@ const PROPOSALS_BATCH_SIZE = 50000;
 const datumSchema = z
   .object({
     id: z.string(),
+    voter: z.string(),
+    space: z.string(),
     vpState: z.string(),
     vpValueByStrategy: z.array(z.number().finite()),
     vpByStrategy: z.array(z.number().finite())
@@ -31,17 +35,26 @@ const datumSchema = z
   });
 
 async function getPendingVotes(): Promise<
-  { id: string; proposal: string; vpState: string; vpByStrategy: number[] }[]
+  {
+    id: string;
+    voter: string;
+    space: string;
+    proposal: string;
+    vpState: string;
+    vpByStrategy: number[];
+  }[]
 > {
   const query = `
-    SELECT id, proposal, vp_state, vp_by_strategy
+    SELECT id, voter, space, proposal, vp_state, vp_by_strategy
     FROM votes
     WHERE cb = ?
     LIMIT ?`;
   const results = await db.queryAsync(query, [CB.PENDING_COMPUTE, DEFAULT_BATCH_SIZE]);
 
   return results.map((r: any) => ({
     id: r.id,
+    voter: r.voter,
+    space: r.space,
     proposal: r.proposal,
     vpState: r.vp_state,
     vpByStrategy: JSON.parse(r.vp_by_strategy)
@@ -88,6 +101,7 @@ async function refreshVotesVpValues(data: Datum[]) {
   const ids: string[] = [];
   const vpValues: Map<string, number> = new Map();
   const cbValues: Map<string, number> = new Map();
+  const leaderboardPairs: Set<string> = new Set();
 
   for (const datum of data) {
     if (datum.proposalCb === CB.INELIGIBLE) {
@@ -110,6 +124,8 @@ async function refreshVotesVpValues(data: Datum[]) {
         validatedDatum.id,
         validatedDatum.vpState === 'final' ? CB.FINAL : CB.PENDING_FINAL
       );
+
+      leaderboardPairs.add(`${validatedDatum.voter}:${validatedDatum.space}`);
     } catch (e) {
       capture(e);
       ids.push(datum.id);
@@ -132,8 +148,30 @@ async function refreshVotesVpValues(data: Datum[]) {
     cbParams.push(id, cbValues.get(id)!);
   }
 
-  const query = `UPDATE votes SET vp_value = CASE ${vpCases} END, cb = CASE ${cbCases} END WHERE id IN (${placeholders})`;
-  await db.queryAsync(query, [...vpParams, ...cbParams, ...ids]);
+  const queries: string[] = [
+    `UPDATE votes SET vp_value = CASE ${vpCases} END, cb = CASE ${cbCases} END WHERE id IN (${placeholders}) AND cb = ?`
+  ];
+  const params: (number | string)[] = [...vpParams, ...cbParams, ...ids, CB.PENDING_COMPUTE];
+
+  // Refresh leaderboard vp_value using SUM from votes table (idempotent)
+  if (leaderboardPairs.size) {
+    const pairPlaceholders = Array.from(leaderboardPairs)
+      .map(() => '(?, ?)')
+      .join(', ');
+    queries.push(
+      `UPDATE leaderboard l
+        SET vp_value = COALESCE((
+          SELECT SUM(v.vp_value) FROM votes v WHERE v.voter = l.user AND v.space = l.space
+        ), 0)
+        WHERE (l.user, l.space) IN (${pairPlaceholders})`
+    );
+    for (const pair of leaderboardPairs) {
+      const [voter, space] = pair.split(':');
+      params.push(voter, space);
+    }
+  }
+
+  await db.queryAsync(queries.join(';'), params);
 }
 
 async function processBatch(proposalVpValues: ProposalVpValues): Promise<number> {
@@ -146,6 +184,8 @@ async function processBatch(proposalVpValues: ProposalVpValues): Promise<number>
       const proposal = proposalVpValues.get(v.proposal)!;
       return {
         id: v.id,
+        voter: v.voter,
+        space: v.space,
         vpState: v.vpState,
         vpByStrategy: v.vpByStrategy,
         vpValueByStrategy: proposal.vpValueByStrategy,

src/index.ts

@@ -3,6 +3,7 @@ import { fallbackLogger, initLogger } from '@snapshot-labs/snapshot-sentry';
 import cors from 'cors';
 import express from 'express';
 import api from './api';
+import deleteProposalVotes from './helpers/deleteProposalVotes';
 import log from './helpers/log';
 import initMetrics from './helpers/metrics';
 import refreshModeration from './helpers/moderation';
@@ -26,6 +27,7 @@ async function startServer() {
   refreshProposalsVpValue();
   refreshProposalsScoresValue();
   refreshVotesVpValue();
+  deleteProposalVotes();
 
   await initializeStrategies();
   refreshStrategies();

src/scores.ts

@@ -62,14 +62,15 @@ async function updateVotesVp(votes: any[], vpState: string, proposalId: string)
     votesInPage.forEach((vote: any) => {
       query += `UPDATE votes
       SET vp = ?, vp_by_strategy = ?, vp_state = ?, vp_value = ?, cb = ?
-      WHERE id = ? AND proposal = ? LIMIT 1; `;
+      WHERE id = ? AND proposal = ? AND cb != ? LIMIT 1; `;
       params.push(vote.balance);
       params.push(JSON.stringify(vote.scores));
       params.push(vpState);
       params.push(vote.vp_value);
       params.push(CB.PENDING_COMPUTE);
       params.push(vote.id);
       params.push(proposalId);
+      params.push(CB.PENDING_DELETE);
     });
     await db.queryAsync(query, params);
     if (i) await snapshot.utils.sleep(200);

src/writer/alias.ts

@@ -10,9 +10,18 @@ export async function verify(message): Promise<any> {
     log.warn(`[writer] Wrong alias format ${JSON.stringify(schemaIsValid)}`);
     return Promise.reject('wrong alias format');
   }
-  if (message.from === msg.payload.alias) {
+  if (message.address === msg.payload.alias) {
     return Promise.reject('alias cannot be the same as the address');
   }
+
+  const [existing] = await db.queryAsync(
+    'SELECT address FROM aliases WHERE alias = ? AND address != ? LIMIT 1',
+    [msg.payload.alias, message.address]
+  );
+  if (existing) {
+    return Promise.reject('alias is already linked to another address');
+  }
+
   return true;
 }
 
@@ -25,5 +34,8 @@ export async function action(message, ipfs, receipt, id): Promise<void> {
     alias: msg.payload.alias,
     created: msg.timestamp
   };
-  await db.queryAsync('INSERT INTO aliases SET ?', params);
+  await db.queryAsync(
+    'INSERT INTO aliases SET ? ON DUPLICATE KEY UPDATE id = VALUES(id), ipfs = VALUES(ipfs), created = VALUES(created)',
+    params
+  );
 }