Fix support big integers. Closes #13

bin/spark-rewards.js

@@ -3,6 +3,7 @@ import http from 'node:http'
 import { once } from 'node:events'
 import { createHandler } from '../index.js'
 import Redis from 'ioredis'
+import Redlock from 'redlock'
 
 const {
   PORT: port = 8000,
@@ -31,8 +32,9 @@ const redis = new Redis({
   password: redisUrlParsed.password,
   family: 6 // required for upstash
 })
+const redlock = new Redlock([redis])
 
-const handler = await createHandler({ logger, redis, signerAddresses })
+const handler = await createHandler({ logger, redis, redlock, signerAddresses })
 const server = http.createServer(handler)
 server.listen(port, host)
 await once(server, 'listening')

index.js

@@ -9,11 +9,11 @@ const maxScore = BigInt(1e15)
 // https://github.com/filecoin-station/spark-impact-evaluator/blob/fd64313a96957fcb3d5fda0d334245601676bb73/test/Spark.t.sol#L11C39-L11C65
 const roundReward = 456621004566210048n
 
-const handler = async (req, res, redis, signerAddresses) => {
+const handler = async (req, res, redis, redlock, signerAddresses) => {
   if (req.method === 'POST' && req.url === '/scores') {
-    await handleIncreaseScores(req, res, redis, signerAddresses)
+    await handleIncreaseScores(req, res, redis, signerAddresses, redlock)
   } else if (req.method === 'POST' && req.url === '/paid') {
-    await handlePaidScheduledRewards(req, res, redis, signerAddresses)
+    await handlePaidScheduledRewards(req, res, redis, signerAddresses, redlock)
   } else if (req.method === 'GET' && req.url === '/scheduled-rewards') {
     await handleGetAllScheduledRewards(res, redis)
   } else if (req.method === 'GET' && req.url.startsWith('/scheduled-rewards/')) {
@@ -56,8 +56,9 @@ const addLogJSON = (tx, obj) => {
   tx.ltrim('log', -(3 * 24 * 5000 * 30), -1)
 }
 
-async function handleIncreaseScores (req, res, redis, signerAddresses) {
+async function handleIncreaseScores (req, res, redis, signerAddresses, redlock) {
   const body = JSON.parse(await getRawBody(req, { limit: '1mb' }))
+  const timestamp = new Date()
 
   httpAssert(
     typeof body === 'object' && body !== null,
@@ -112,38 +113,59 @@ async function handleIncreaseScores (req, res, redis, signerAddresses) {
     body.scores.splice(index, 1)
   }
 
-  const timestamp = new Date()
-  const tx = redis.multi()
-  for (let i = 0; i < body.participants.length; i++) {
-    const address = body.participants[i]
-    const score = body.scores[i]
-    const scheduledRewards = (BigInt(score) * roundReward) / maxScore
-    tx.hincrby('rewards', address, scheduledRewards)
-    addLogJSON(tx, {
-      timestamp,
-      address,
-      score,
-      scheduledRewardsDelta: String(scheduledRewards)
+  if (body.participants.length === 0) {
+    return json(res, {})
+  }
+
+  const scheduledRewardsDelta = body.scores.map(score => {
+    return (BigInt(score) * roundReward) / maxScore
+  })
+  let updatedRewards
+
+  const lock = await redlock.lock('lock:rewards', 5000)
+  try {
+    const currentRewards = (await redis.hmget('rewards', ...body.participants)).map(amount => {
+      return BigInt(amount || '0')
     })
+    updatedRewards = body.scores.map((_, i) => {
+      return currentRewards[i] + scheduledRewardsDelta[i]
+    })
+
+    const tx = redis.multi()
+    tx.hset(
+      'rewards',
+      Object.fromEntries(body.participants.map((address, i) => ([
+        address,
+        String(updatedRewards[i])
+      ])))
+    )
+    for (let i = 0; i < body.participants.length; i++) {
+      addLogJSON(tx, {
+        timestamp,
+        address: body.participants[i],
+        score: body.scores[i],
+        scheduledRewardsDelta: String(scheduledRewardsDelta[i])
+      })
+    }
+    await tx.exec()
+  } finally {
+    await lock.unlock()
   }
-  const results = await tx.exec()
 
   json(
     res,
     Object.fromEntries(
       body.participants.map((address, i) => [
         address,
-        // Every 3rd entry is from `hincrby`, which returns the new value.
-        // Inside the array there are two fields, the 2nd containing the
-        // new value.
-        String(results[i * 3][1])
+        String(updatedRewards[i])
       ])
     )
   )
 }
 
-async function handlePaidScheduledRewards (req, res, redis, signerAddresses) {
+async function handlePaidScheduledRewards (req, res, redis, signerAddresses, redlock) {
   const body = JSON.parse(await getRawBody(req, { limit: '1mb' }))
+  const timestamp = new Date()
 
   httpAssert(
     typeof body === 'object' && body !== null,
@@ -190,29 +212,47 @@ async function handlePaidScheduledRewards (req, res, redis, signerAddresses) {
     signerAddresses
   )
 
-  const timestamp = new Date()
-  const tx = redis.multi()
-  for (let i = 0; i < body.participants.length; i++) {
-    const address = body.participants[i]
-    const amount = body.rewards[i]
-    tx.hincrby('rewards', address, BigInt(amount) * -1n)
-    addLogJSON(tx, {
-      timestamp,
-      address,
-      scheduledRewardsDelta: String(BigInt(amount) * -1n)
+  if (body.participants.length === 0) {
+    return json(res, {})
+  }
+
+  let updatedRewards
+
+  const lock = await redlock.lock('lock:rewards', 5000)
+  try {
+    const currentRewards = (await redis.hmget('rewards', ...body.participants)).map(amount => {
+      return BigInt(amount || '0')
     })
+    updatedRewards = body.rewards.map((amount, i) => {
+      return currentRewards[i] - BigInt(amount)
+    })
+
+    const tx = redis.multi()
+    tx.hset(
+      'rewards',
+      Object.fromEntries(body.participants.map((address, i) => ([
+        address,
+        String(updatedRewards[i])
+      ])))
+    )
+    for (let i = 0; i < body.participants.length; i++) {
+      addLogJSON(tx, {
+        timestamp,
+        address: body.participants[i],
+        scheduledRewardsDelta: String(BigInt(body.rewards[i]) * -1n)
+      })
+    }
+    await tx.exec()
+  } finally {
+    await lock.unlock()
   }
-  const updated = await tx.exec()
 
   json(
     res,
     Object.fromEntries(
       body.participants.map((address, i) => [
         address,
-        // Every 3rd entry is from `hincrby`, which returns the new value.
-        // Inside the array there are two fields, the 2nd containing the
-        // new value.
-        String(updated[i * 3][1])
+        String(updatedRewards[i])
       ])
     )
   )
@@ -250,15 +290,16 @@ const errorHandler = (res, err, logger) => {
   }
 }
 
-export const createHandler = async ({ logger, redis, signerAddresses }) => {
+export const createHandler = async ({ logger, redis, signerAddresses, redlock }) => {
   assert(logger, '.logger required')
   assert(redis, '.redis required')
+  assert(redlock, '.redlock required')
   assert(signerAddresses, '.signerAddresses required')
 
   return (req, res) => {
     const start = new Date()
     logger.request(`${req.method} ${req.url} ...`)
-    handler(req, res, redis, signerAddresses)
+    handler(req, res, redis, redlock, signerAddresses)
       .catch(err => errorHandler(res, err, logger))
       .then(() => {
         logger.request(

package.json

@@ -12,7 +12,8 @@
     "http-assert": "^1.5.0",
     "http-responders": "^2.2.0",
     "ioredis": "^5.4.1",
-    "raw-body": "^3.0.0"
+    "raw-body": "^3.0.0",
+    "redlock": "^4.2.0"
   },
   "devDependencies": {
     "standard": "^17.1.2"

test.js

@@ -5,6 +5,7 @@ import Redis from 'ioredis'
 import { once } from 'node:events'
 import assert from 'node:assert/strict'
 import * as ethers from 'ethers'
+import Redlock from 'redlock'
 
 let signer
 let server
@@ -24,6 +25,7 @@ test.before(async () => {
   const handler = await createHandler({
     logger,
     redis,
+    redlock: new Redlock([redis]),
     signerAddresses: [await signer.getAddress()]
   })
   server = http.createServer(handler)
@@ -189,6 +191,26 @@ suite('scheduled rewards', () => {
       ])
     }
   })
+  test('big integer', async t => {
+    const participants = [
+      '0x000000000000000000000000000000000000dE12'
+    ]
+    const scores = [
+      '1000000000000000000000000000'
+    ]
+    const res = await fetch(`${api}/scores`, {
+      method: 'POST',
+      body: JSON.stringify({
+        participants,
+        scores,
+        signature: await sign(participants, scores)
+      })
+    })
+    assert.strictEqual(res.status, 200)
+    assert.deepStrictEqual(await res.json(), {
+      '0x000000000000000000000000000000000000dE12': '456621004566210048000000000000'
+    })
+  })
   test('paid rewards', async t => {
     {
       const participants = ['0x000000000000000000000000000000000000dEa2']
@@ -210,7 +232,8 @@ suite('scheduled rewards', () => {
       const res = await fetch(`${api}/scheduled-rewards`)
       assert.deepEqual(await res.json(), {
         '0x000000000000000000000000000000000000dEa2': '0',
-        '0x000000000000000000000000000000000000dEa7': '45662'
+        '0x000000000000000000000000000000000000dEa7': '45662',
+        '0x000000000000000000000000000000000000dE12': '456621004566210048000000000000'
       })
     }
     {
@@ -236,6 +259,11 @@ suite('scheduled rewards', () => {
           score: '10',
           scheduledRewardsDelta: '4566'
         },
+        {
+          address: '0x000000000000000000000000000000000000dE12',
+          score: '1000000000000000000000000000',
+          scheduledRewardsDelta: '456621004566210048000000000000'
+        },
         {
           address: '0x000000000000000000000000000000000000dEa2',
           scheduledRewardsDelta: '-9132'