Merge pull request #52 from Balatro-Multiplayer/queue-fixes

Queue fixes and cron job

Author: robotzurg

src/events/interactionCreate.ts

@@ -158,14 +158,10 @@ export default {
                     : 'You left the queue.',
               })
 
-              await updateQueueMessage()
-
               // Delete the message after 10 seconds
               setTimeout(async () => {
                 await interaction.deleteReply(reply.id).catch(() => {})
               }, 10000)
-            } else {
-              await updateQueueMessage()
             }
           } finally {
             // Always remove user from processing set
@@ -491,8 +487,6 @@ export default {
             content: message,
           })
 
-          await updateQueueMessage()
-
           // Delete the message after 10 seconds
           setTimeout(async () => {
             await interaction.deleteReply(reply.id).catch(() => {})

src/index.ts

@@ -6,6 +6,7 @@ import {
   runDecayTick,
   updateMatchCountCronJob,
   deleteExpiredStrikesCronJob,
+  updateQueueMessageCronJob,
 } from './utils/cronJobs'
 import { app } from './api/app'
 import { client } from './client'
@@ -70,6 +71,9 @@ void updateMatchCountCronJob().catch((error) =>
 void deleteExpiredStrikesCronJob().catch((error) =>
   console.error('[STRIKES CRON ERROR]', error),
 )
+void updateQueueMessageCronJob().catch((error) =>
+  console.error('[QUEUE MESSAGE CRON ERROR]', error),
+)
 
 // Start API server
 const port = Number(process.env.PORT) || 4931

src/utils/cronJobs.ts

@@ -14,7 +14,11 @@ import {
   removeUserFromQueue,
   updateCurrentEloRangeForUser,
 } from './queryDB'
-import { createMatch, timeSpentInQueue } from './queueHelpers'
+import {
+  createMatch,
+  timeSpentInQueue,
+  updateQueueMessage,
+} from './queueHelpers'
 import { updateMatchCountChannel } from './matchHelpers'
 import { pool } from '../db'
 import * as fs from 'fs'
@@ -159,21 +163,44 @@ export async function incrementEloCronJobAllQueues() {
         if (bestPair.length === 2) {
           const matchupUsers = bestPair.map((u) => u.userId)
 
-          // remove users from all queues they are in
-          // Wait for all removal operations to complete before creating match
-          await Promise.all(
-            matchupUsers.map(async (userId) => {
-              const userQueues = await getUserQueues(userId)
-              await Promise.all(
-                userQueues.map((queue) =>
-                  removeUserFromQueue(queue.id, userId),
-                ),
-              )
-            }),
-          )
+          // Check with a lock to ensure that only one match is created at a time
+          const dbClient = await pool.connect()
+          try {
+            await dbClient.query('BEGIN')
+
+            // Lock the rows and check if all users are still in queue
+            const checkResult = await dbClient.query(
+              `SELECT user_id FROM queue_users
+               WHERE user_id = ANY($1::varchar[])
+               AND queue_join_time IS NOT NULL
+               FOR UPDATE`,
+              [matchupUsers],
+            )
+
+            // If not all users are still available, skip this match
+            if (checkResult.rowCount !== matchupUsers.length) {
+              await dbClient.query('ROLLBACK')
+              dbClient.release()
+              continue
+            }
+
+            // Remove users from queue (set queue_join_time to NULL)
+            await dbClient.query(
+              `UPDATE queue_users SET queue_join_time = NULL
+               WHERE user_id = ANY($1::varchar[])`,
+              [matchupUsers],
+            )
 
-          // queue them for the match
-          await createMatch(matchupUsers, bestPair[0].queueId)
+            await dbClient.query('COMMIT')
+            dbClient.release()
+
+            // Now create the match (users are already removed from queue)
+            await createMatch(matchupUsers, bestPair[0].queueId)
+          } catch (err) {
+            await dbClient.query('ROLLBACK')
+            dbClient.release()
+            console.error('Error creating match in incrementEloCronJob:', err)
+          }
         }
       }
     } catch (err) {
@@ -258,3 +285,14 @@ export async function deleteExpiredStrikesCronJob() {
     2 * 60 * 60 * 1000,
   ) // every 2 hours
 }
+
+// update queue message every 2 seconds
+export async function updateQueueMessageCronJob() {
+  setInterval(async () => {
+    try {
+      await updateQueueMessage()
+    } catch (err) {
+      console.error('Error updating queue message:', err)
+    }
+  }, 2 * 1000) // every 2 seconds
+}

src/utils/queueHelpers.ts

@@ -287,17 +287,18 @@ export async function joinQueues(
     // Batch upsert all queue joins
     for (const qId of selectedQueueIds) {
       const queueId = parseInt(qId)
-      const queue = queueMap.get(queueId)
+      const queue: Queues = queueMap.get(queueId)
       if (!queue) continue
 
       await createQueueUser(userId, queueId)
       await client.query(
         `UPDATE queue_users
          SET queue_join_time = NOW(),
              elo = COALESCE(elo, $3),
-             peak_elo = COALESCE(peak_elo, $3)
+             peak_elo = COALESCE(peak_elo, $3),
+             current_elo_range = $4
          WHERE user_id = $1 AND queue_id = $2`,
-        [userId, queueId, queue.default_elo],
+        [userId, queueId, queue.default_elo, queue.elo_search_start],
       )
     }
 
@@ -543,8 +544,6 @@ export async function createMatch(
     } catch (err) {}
   }
 
-  await updateQueueMessage()
-
   // Wait 2 seconds for channel to fully propagate in Discord's API
   await new Promise((resolve) => setTimeout(resolve, 2000))