functions/src/resolve-market.ts

@@ -1,6 +1,6 @@
 import * as admin from 'firebase-admin'
 import { z } from 'zod'
-import { mapValues, groupBy, sumBy } from 'lodash'
+import { mapValues, groupBy, sumBy, uniqBy } from 'lodash'
 
 import {
   Contract,
@@ -15,14 +15,14 @@ import {
   getValues,
   isProd,
   log,
-  payUser,
+  payUsers,
+  payUsersMultipleTransactions,
   revalidateStaticProps,
 } from './utils'
 import {
   getLoanPayouts,
   getPayouts,
   groupPayoutsByUser,
-  Payout,
 } from '../../common/payouts'
 import { isAdmin, isManifoldId } from '../../common/envs/constants'
 import { removeUndefinedProps } from '../../common/util/object'
@@ -131,8 +131,12 @@ export const resolvemarket = newEndpoint(opts, async (req, auth) => {
     (doc) => doc.data() as LiquidityProvision
   )
 
-  const { payouts, creatorPayout, liquidityPayouts, collectedFees } =
+  const {
-    getPayouts(
+    payouts: traderPayouts,
+    creatorPayout,
+    liquidityPayouts,
+    collectedFees,
+  } = getPayouts(
     outcome,
     contract,
     bets,
@@ -156,30 +160,43 @@ export const resolvemarket = newEndpoint(opts, async (req, auth) => {
     subsidyPool: 0,
   }
 
-  await contractDoc.update(updatedContract)
-
-  console.log('contract ', contractId, 'resolved to:', outcome)
-
   const openBets = bets.filter((b) => !b.isSold && !b.sale)
   const loanPayouts = getLoanPayouts(openBets)
 
+  const payouts = [
+    { userId: creatorId, payout: creatorPayout, deposit: creatorPayout },
+    ...liquidityPayouts.map((p) => ({ ...p, deposit: p.payout })),
+    ...traderPayouts,
+    ...loanPayouts,
+  ]
+
   if (!isProd())
     console.log(
-      'payouts:',
+      'trader payouts:',
-      payouts,
+      traderPayouts,
       'creator payout:',
       creatorPayout,
-      'liquidity payout:'
+      'liquidity payout:',
+      liquidityPayouts,
+      'loan payouts:',
+      loanPayouts
     )
 
-  if (creatorPayout)
+  const userCount = uniqBy(payouts, 'userId').length
-    await processPayouts([{ userId: creatorId, payout: creatorPayout }], true)
 
-  await processPayouts(liquidityPayouts, true)
+  if (userCount <= 499) {
+    await firestore.runTransaction(async (transaction) => {
+      payUsers(transaction, payouts)
+      transaction.update(contractDoc, updatedContract)
+    })
+  } else {
+    await payUsersMultipleTransactions(payouts)
+    await contractDoc.update(updatedContract)
+  }
+
+  console.log('contract ', contractId, 'resolved to:', outcome)
 
-  await processPayouts([...payouts, ...loanPayouts])
   await undoUniqueBettorRewardsIfCancelResolution(contract, outcome)
-
   await revalidateStaticProps(getContractPath(contract))
 
   const userPayoutsWithoutLoans = groupPayoutsByUser(payouts)
@@ -211,18 +228,6 @@ export const resolvemarket = newEndpoint(opts, async (req, auth) => {
   return updatedContract
 })
 
-const processPayouts = async (payouts: Payout[], isDeposit = false) => {
-  const userPayouts = groupPayoutsByUser(payouts)
-
-  const payoutPromises = Object.entries(userPayouts).map(([userId, payout]) =>
-    payUser(userId, payout, isDeposit)
-  )
-
-  return await Promise.all(payoutPromises)
-    .catch((e) => ({ status: 'error', message: e }))
-    .then(() => ({ status: 'success' }))
-}
-
 function getResolutionParams(contract: Contract, body: string) {
   const { outcomeType } = contract
 

functions/src/utils.ts

@@ -1,7 +1,8 @@
 import * as admin from 'firebase-admin'
 import fetch from 'node-fetch'
+import { FieldValue, Transaction } from 'firebase-admin/firestore'
+import { chunk, groupBy, mapValues, sumBy } from 'lodash'
 
-import { chunk } from 'lodash'
 import { Contract } from '../../common/contract'
 import { PrivateUser, User } from '../../common/user'
 import { Group } from '../../common/group'
@@ -128,38 +129,29 @@ export const getUserByUsername = async (username: string) => {
   return snap.empty ? undefined : (snap.docs[0].data() as User)
 }
 
+const firestore = admin.firestore()
+
 const updateUserBalance = (
+  transaction: Transaction,
   userId: string,
-  delta: number,
+  balanceDelta: number,
-  isDeposit = false
+  depositDelta: number
 ) => {
-  const firestore = admin.firestore()
-  return firestore.runTransaction(async (transaction) => {
   const userDoc = firestore.doc(`users/${userId}`)
-    const userSnap = await transaction.get(userDoc)
-    if (!userSnap.exists) return
-    const user = userSnap.data() as User
 
-    const newUserBalance = user.balance + delta
+  // Note: Balance is allowed to go negative.
-
+  transaction.update(userDoc, {
-    // if (newUserBalance < 0)
+    balance: FieldValue.increment(balanceDelta),
-    //   throw new Error(
+    totalDeposits: FieldValue.increment(depositDelta),
-    //     `User (${userId}) balance cannot be negative: ${newUserBalance}`
-    //   )
-
-    if (isDeposit) {
-      const newTotalDeposits = (user.totalDeposits || 0) + delta
-      transaction.update(userDoc, { totalDeposits: newTotalDeposits })
-    }
-
-    transaction.update(userDoc, { balance: newUserBalance })
   })
 }
 
 export const payUser = (userId: string, payout: number, isDeposit = false) => {
   if (!isFinite(payout)) throw new Error('Payout is not finite: ' + payout)
 
-  return updateUserBalance(userId, payout, isDeposit)
+  return firestore.runTransaction(async (transaction) => {
+    updateUserBalance(transaction, userId, payout, isDeposit ? payout : 0)
+  })
 }
 
 export const chargeUser = (
@@ -170,7 +162,67 @@ export const chargeUser = (
   if (!isFinite(charge) || charge <= 0)
     throw new Error('User charge is not positive: ' + charge)
 
-  return updateUserBalance(userId, -charge, isAnte)
+  return payUser(userId, -charge, isAnte)
+}
+
+const checkAndMergePayouts = (
+  payouts: {
+    userId: string
+    payout: number
+    deposit?: number
+  }[]
+) => {
+  for (const { payout, deposit } of payouts) {
+    if (!isFinite(payout)) {
+      throw new Error('Payout is not finite: ' + payout)
+    }
+    if (deposit !== undefined && !isFinite(deposit)) {
+      throw new Error('Deposit is not finite: ' + deposit)
+    }
+  }
+
+  const groupedPayouts = groupBy(payouts, 'userId')
+  return Object.values(
+    mapValues(groupedPayouts, (payouts, userId) => ({
+      userId,
+      payout: sumBy(payouts, 'payout'),
+      deposit: sumBy(payouts, (p) => p.deposit ?? 0),
+    }))
+  )
+}
+
+// Max 500 users in one transaction.
+export const payUsers = (
+  transaction: Transaction,
+  payouts: {
+    userId: string
+    payout: number
+    deposit?: number
+  }[]
+) => {
+  const mergedPayouts = checkAndMergePayouts(payouts)
+  for (const { userId, payout, deposit } of mergedPayouts) {
+    updateUserBalance(transaction, userId, payout, deposit)
+  }
+}
+
+export const payUsersMultipleTransactions = async (
+  payouts: {
+    userId: string
+    payout: number
+    deposit?: number
+  }[]
+) => {
+  const mergedPayouts = checkAndMergePayouts(payouts)
+  const payoutChunks = chunk(mergedPayouts, 500)
+
+  for (const payoutChunk of payoutChunks) {
+    await firestore.runTransaction(async (transaction) => {
+      for (const { userId, payout, deposit } of payoutChunk) {
+        updateUserBalance(transaction, userId, payout, deposit)
+      }
+    })
+  }
 }
 
 export const getContractPath = (contract: Contract) => {