mirror of
https://github.com/Such-Software/wownerogue.git
synced 2026-10-23 09:08:32 +02:00
fix: resolve security/correctness review + AAA graphics pass
Backend (Tier 0/1/3): single-payout double-pay -> needs_review; admin-retry tx_hash guard; UNIQUE index on games.payment_id (migration 023) + atomic claim; reconnect no longer resets block-timeout; completeGame keys on user_id+dungeon_seed (not socket_id); token-gated /api/user/:socketId/* endpoints; payout min/max enforcement; orphaned-active-game boot recovery; sessionManager cleanup FK fix; XFF rightmost-hop; global uncaughtException/unhandledRejection guard; server binds 127.0.0.1; provably-fair client-seed (HMAC) + RNG [0,1) bound; difficulty preset ranges reach the digger; chat no longer leaks full socket.id; removed dangerous-keyword SQL theater. Match mode (Tier 2, still MATCH_ENABLED-gated): fixed match_queue/match_move crash paths + scheduler dead-code + leaked ceiling timer; pot collected before start with abort+refund; BigInt/basis-point fee math, capped payout; no-address winner claimable; winnerId==placement[0]; finalize schema + payout-first isolation; reconnect re-key with ownership; refunds on all cancel paths; real countdown; map/timer cleanup. +16 tests. Graphics: new overlay FX layer (torch light, vignette, embers, sparkle/burst/flash/shake, self-suspending rAF); gameplay juice; atmospheric screens; CSS design-token chrome with framed modals; richer Kenney tile renderer; imported Kenney dungeon/interior sheets + fx particles. Verification: 45 suites / 312 tests pass; 40 changed JS files parse; 20 server modules load clean; no regressions.
This commit is contained in:
+309
-104
@@ -2,21 +2,25 @@
|
||||
* MatchQueue — persisted, per-economy queue for match mode.
|
||||
*
|
||||
* Responsibilities:
|
||||
* • Join/leave a queue for each economy (free, credits_prestige, crypto_race).
|
||||
* • Atomically deduct credits or tickets into escrow on join; refund on leave.
|
||||
* • List queued players for a given economy.
|
||||
* • Drain a full economy bucket into a match at block time.
|
||||
* • Enqueue/leave a queue for each economy (free, credits_prestige, crypto_race).
|
||||
* • Atomically deduct credits or tickets into escrow on join; REFUND on EVERY cancellation
|
||||
* path (leave, stale cleanup, drain abort, shutdown, boot recovery), always writing a
|
||||
* ledger row — never a status flip alone.
|
||||
* • Drain a full economy bucket into a match at block time (memory only mutated after the DB
|
||||
* commit succeeds, so a failed drain leaves the queue intact).
|
||||
*
|
||||
* The queue is backed by `match_queue_entries` so state survives server restarts and
|
||||
* disconnects. In-memory maps are rebuilt on initialization.
|
||||
* Backed by `match_queue_entries` so state survives restarts. In-memory maps are rebuilt on
|
||||
* initialize(), which also refunds abandoned queue rows and recovers in-flight `matches`.
|
||||
*
|
||||
* This module has no Socket.IO coupling; MatchManager owns transport.
|
||||
* No Socket.IO coupling; MatchManager owns transport.
|
||||
*/
|
||||
|
||||
const { normalizeError } = require('../utils/errors');
|
||||
|
||||
const ECONOMIES = Object.freeze(['free', 'credits_prestige', 'crypto_race']);
|
||||
const MATCH_ECONOMY_SET = new Set(ECONOMIES);
|
||||
// race_entry_transactions.reason is CHECK-constrained; map any cancellation reason into it.
|
||||
const RACE_LEDGER_REASONS = new Set(['queue_leave', 'match_cancel', 'refund']);
|
||||
|
||||
class MatchQueue {
|
||||
constructor({ db, gameModeManager = null, debugManager = null } = {}) {
|
||||
@@ -37,23 +41,12 @@ class MatchQueue {
|
||||
async initialize() {
|
||||
if (!this.enabled || !this.db) return;
|
||||
|
||||
// Cancel any stale matched-but-not-started entries (server crashed during drain).
|
||||
await this.db.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET status = 'cancelled', match_id = NULL
|
||||
WHERE status = 'matched'
|
||||
AND created_at < NOW() - INTERVAL '5 minutes'
|
||||
`);
|
||||
// MP-H3: refund + cancel abandoned entries and recover in-flight matches BEFORE
|
||||
// rebuilding memory, so escrow is never silently lost across a restart/crash.
|
||||
await this._recoverStaleQueueEntries();
|
||||
await this._recoverAbandonedMatches();
|
||||
|
||||
// Refund queued-but-stale entries older than 24 hours.
|
||||
await this.db.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET status = 'cancelled'
|
||||
WHERE status = 'queued'
|
||||
AND created_at < NOW() - INTERVAL '24 hours'
|
||||
`);
|
||||
|
||||
// Rebuild in-memory state.
|
||||
// Rebuild in-memory state from the surviving 'queued' rows.
|
||||
const result = await this.db.query(`
|
||||
SELECT id, user_id, economy, socket_id, session_token, created_at
|
||||
FROM match_queue_entries
|
||||
@@ -75,6 +68,98 @@ class MatchQueue {
|
||||
this._log(`[MatchQueue] Initialized: ${this.length('free')} free, ${this.length('credits_prestige')} prestige, ${this.length('crypto_race')} crypto`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Refund + cancel queue rows abandoned by a crash: 'matched' rows that never became a
|
||||
* running match (older than 5 min), and 'queued' rows older than 24h. Each refund is
|
||||
* written to the ledger; the row is only cancelled in the same transaction as its refund.
|
||||
*/
|
||||
async _recoverStaleQueueEntries() {
|
||||
let stale = [];
|
||||
try {
|
||||
const res = await this.db.query(`
|
||||
SELECT user_id, economy FROM match_queue_entries
|
||||
WHERE (status = 'matched' AND created_at < NOW() - INTERVAL '5 minutes')
|
||||
OR (status = 'queued' AND created_at < NOW() - INTERVAL '24 hours')
|
||||
`);
|
||||
stale = res.rows || [];
|
||||
} catch (err) {
|
||||
this._log('[MatchQueue] stale scan error', err.message);
|
||||
return;
|
||||
}
|
||||
for (const row of stale) {
|
||||
try {
|
||||
await this.db.withTransaction(async (client) => {
|
||||
const upd = await client.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET status = 'cancelled', match_id = NULL
|
||||
WHERE user_id = $1 AND economy = $2 AND status IN ('queued', 'matched')
|
||||
RETURNING id
|
||||
`, [row.user_id, row.economy]);
|
||||
if (upd.rowCount > 0) {
|
||||
await this._applyRefund(client, row.user_id, row.economy, 'match_cancel');
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
this._log('[MatchQueue] stale recovery error', err.message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot recovery for in-flight `matches` rows left 'starting'/'active' by a crash. If a
|
||||
* payout was already recorded for the match, let it stand (finalize the status); otherwise
|
||||
* cancel the match and refund every entrant so escrow is never stranded.
|
||||
*/
|
||||
async _recoverAbandonedMatches() {
|
||||
let rows = [];
|
||||
try {
|
||||
const res = await this.db.query(`SELECT id, economy FROM matches WHERE status IN ('starting', 'active')`);
|
||||
rows = res.rows || [];
|
||||
} catch (err) {
|
||||
// matches table may not exist in a minimal test DB — non-fatal.
|
||||
this._log('[MatchQueue] abandoned match scan error', err.message);
|
||||
return;
|
||||
}
|
||||
for (const m of rows) {
|
||||
try {
|
||||
await this.db.withTransaction(async (client) => {
|
||||
// If a payout already exists, the race effectively completed — don't refund
|
||||
// (that would double-pay the winner); just finalize the match status so the
|
||||
// batcher/retry can settle the payout.
|
||||
const pay = await client.query(`SELECT id FROM payouts WHERE match_id = $1 LIMIT 1`, [m.id]);
|
||||
if (pay.rows.length > 0) {
|
||||
await client.query(`
|
||||
UPDATE matches SET status = 'finished', ended_at = COALESCE(ended_at, NOW())
|
||||
WHERE id = $1 AND status IN ('starting', 'active')
|
||||
`, [m.id]);
|
||||
return;
|
||||
}
|
||||
|
||||
const cancel = await client.query(`
|
||||
UPDATE matches SET status = 'cancelled', ended_at = NOW()
|
||||
WHERE id = $1 AND status IN ('starting', 'active')
|
||||
RETURNING id
|
||||
`, [m.id]);
|
||||
if (cancel.rowCount === 0) return; // already handled by another process
|
||||
|
||||
const ent = await client.query(`SELECT user_id FROM match_entrants WHERE match_id = $1`, [m.id]);
|
||||
for (const e of ent.rows || []) {
|
||||
if (m.economy === 'crypto_race' || m.economy === 'credits_prestige') {
|
||||
await this._applyRefund(client, e.user_id, m.economy, 'match_cancel');
|
||||
}
|
||||
}
|
||||
await client.query(`
|
||||
UPDATE match_queue_entries SET status = 'cancelled'
|
||||
WHERE match_id = $1 AND status IN ('queued', 'matched')
|
||||
`, [m.id]);
|
||||
});
|
||||
this._log(`[MatchQueue] recovered abandoned match ${m.id} (${m.economy})`);
|
||||
} catch (err) {
|
||||
this._log('[MatchQueue] abandoned match recovery error', err.message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_validateEconomy(economy) {
|
||||
return MATCH_ECONOMY_SET.has(economy) ? economy : null;
|
||||
}
|
||||
@@ -92,33 +177,53 @@ class MatchQueue {
|
||||
return this._queues[economy].slice();
|
||||
}
|
||||
|
||||
/**
|
||||
* socketHandlers-facing API: enqueue a resolved session into a match queue. Identity comes
|
||||
* from the (server-resolved) session; the economy is explicit.
|
||||
* @param {{userId:number, socketId:string, sessionToken?:string, economy?:string}} session
|
||||
* @param {{economy?:string}} [opts]
|
||||
*/
|
||||
async enqueue(session, opts = {}) {
|
||||
const economy = (opts && opts.economy) || session?.economy;
|
||||
return this.join({
|
||||
userId: session?.userId,
|
||||
socketId: session?.socketId,
|
||||
sessionToken: session?.sessionToken,
|
||||
economy
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Join a queue. For paid economies the cost is held in escrow.
|
||||
* @param {object} entry
|
||||
* @param {number} entry.userId
|
||||
* @param {string} entry.socketId
|
||||
* @param {string} entry.sessionToken
|
||||
* @param {string} entry.economy
|
||||
* @param {object} [entry.appearance]
|
||||
* @returns {Promise<{success:boolean, reason?:string, position?:number}>}
|
||||
*/
|
||||
async join(entry) {
|
||||
if (!this.enabled) return { success: false, reason: 'match_disabled' };
|
||||
if (!entry || entry.userId == null) return { success: false, reason: 'invalid_user' };
|
||||
const economy = this._validateEconomy(entry.economy);
|
||||
if (!economy) return { success: false, reason: 'invalid_economy' };
|
||||
|
||||
// session_token is NOT NULL in the schema; coerce a missing token to a stable-ish
|
||||
// string (reconnect matches on userId anyway).
|
||||
const normalized = {
|
||||
...entry,
|
||||
economy,
|
||||
sessionToken: entry.sessionToken != null ? String(entry.sessionToken) : String(entry.userId)
|
||||
};
|
||||
|
||||
try {
|
||||
if (economy === 'free') {
|
||||
return await this._joinFree(entry);
|
||||
return await this._joinFree(normalized);
|
||||
}
|
||||
if (economy === 'credits_prestige') {
|
||||
return await this._joinCreditsPrestige(entry);
|
||||
return await this._joinCreditsPrestige(normalized);
|
||||
}
|
||||
return await this._joinCryptoRace(entry);
|
||||
return await this._joinCryptoRace(normalized);
|
||||
} catch (err) {
|
||||
const normalized = normalizeError(err, 'Failed to join match queue');
|
||||
this._log('[MatchQueue] join error:', normalized.message);
|
||||
return { success: false, reason: normalized.safeMessage || 'join_failed' };
|
||||
const normErr = normalizeError(err, 'Failed to join match queue');
|
||||
this._log('[MatchQueue] join error:', normErr.message);
|
||||
return { success: false, reason: normErr.safeMessage || 'join_failed' };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -165,7 +270,7 @@ class MatchQueue {
|
||||
|
||||
const balanceAfter = creditResult.rows[0].credits;
|
||||
|
||||
// 2. Log credit transaction.
|
||||
// 2. Log credit transaction (escrow).
|
||||
await client.query(`
|
||||
INSERT INTO credit_transactions (user_id, amount, reason, balance_after, transaction_type, metadata)
|
||||
VALUES ($1, $2, 'match_queue_join', $3, 'match', $4)
|
||||
@@ -186,7 +291,7 @@ class MatchQueue {
|
||||
|
||||
async _joinCryptoRace(entry) {
|
||||
await this.db.withTransaction(async (client) => {
|
||||
// 1. Consume one race entry ticket atomically.
|
||||
// 1. Consume one race entry ticket atomically (escrow).
|
||||
const ticketResult = await client.query(`
|
||||
UPDATE users
|
||||
SET race_entries = race_entries - 1
|
||||
@@ -200,7 +305,7 @@ class MatchQueue {
|
||||
|
||||
const balanceAfter = ticketResult.rows[0].race_entries;
|
||||
|
||||
// 2. Log ticket transaction (escrow).
|
||||
// 2. Log ticket transaction.
|
||||
await client.query(`
|
||||
INSERT INTO race_entry_transactions (user_id, delta, balance_after, reason, metadata)
|
||||
VALUES ($1, -1, $2, 'queue_join', $3)
|
||||
@@ -232,20 +337,95 @@ class MatchQueue {
|
||||
}
|
||||
|
||||
/**
|
||||
* Leave a queue before a match starts. Refund credits/tickets.
|
||||
* @param {number} userId
|
||||
* @param {string} economy
|
||||
* @returns {Promise<{success:boolean, reason?:string}>}
|
||||
* Apply a refund for one entrant within an open transaction. Amounts are DB-derived
|
||||
* (MP-H4): the credit refund reads the exact deducted amount from the ledger, so it is
|
||||
* correct even after a restart cleared the in-memory entry. Always writes a ledger row.
|
||||
*/
|
||||
async leave(userId, economy) {
|
||||
async _applyRefund(client, userId, economy, ledgerReason = 'queue_leave') {
|
||||
if (economy === 'credits_prestige') {
|
||||
const amount = await this._creditsJoinAmount(client, userId);
|
||||
if (amount <= 0) return 0;
|
||||
const r = await client.query(`
|
||||
UPDATE users SET credits = credits + $1 WHERE id = $2 RETURNING credits
|
||||
`, [amount, userId]);
|
||||
const bal = r.rows[0]?.credits ?? 0;
|
||||
await client.query(`
|
||||
INSERT INTO credit_transactions (user_id, amount, reason, balance_after, transaction_type, metadata)
|
||||
VALUES ($1, $2, 'match_queue_refund', $3, 'match', $4)
|
||||
`, [userId, amount, bal, JSON.stringify({ economy, refunded: true, reason: ledgerReason })]);
|
||||
return amount;
|
||||
}
|
||||
if (economy === 'crypto_race') {
|
||||
const reason = RACE_LEDGER_REASONS.has(ledgerReason) ? ledgerReason : 'queue_leave';
|
||||
const r = await client.query(`
|
||||
UPDATE users SET race_entries = race_entries + 1 WHERE id = $1 RETURNING race_entries
|
||||
`, [userId]);
|
||||
const bal = r.rows[0]?.race_entries ?? 0;
|
||||
await client.query(`
|
||||
INSERT INTO race_entry_transactions (user_id, delta, balance_after, reason, metadata)
|
||||
VALUES ($1, 1, $2, $3, $4)
|
||||
`, [userId, bal, reason, JSON.stringify({ economy, refunded: true, source: ledgerReason })]);
|
||||
return 1;
|
||||
}
|
||||
return 0; // free: nothing to refund
|
||||
}
|
||||
|
||||
/**
|
||||
* The exact credit amount deducted at join, read from the ledger so refunds survive
|
||||
* restarts (MP-H4). Falls back to the configured cost if no ledger row is found.
|
||||
*/
|
||||
async _creditsJoinAmount(client, userId) {
|
||||
try {
|
||||
const r = await client.query(`
|
||||
SELECT amount FROM credit_transactions
|
||||
WHERE user_id = $1 AND reason = 'match_queue_join'
|
||||
AND (metadata->>'economy') = 'credits_prestige'
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT 1
|
||||
`, [userId]);
|
||||
if (r.rows.length > 0) {
|
||||
const amt = Math.abs(parseInt(r.rows[0].amount, 10));
|
||||
if (Number.isFinite(amt) && amt > 0) return amt;
|
||||
}
|
||||
} catch (_) { /* fall through to configured cost */ }
|
||||
return this._creditsCost();
|
||||
}
|
||||
|
||||
/**
|
||||
* Leave a queue before a match starts, refunding credits/tickets. Accepts either a resolved
|
||||
* session object ({ userId, economy }) — the socketHandlers form — or the legacy
|
||||
* (userId, economy) pair. When no economy is supplied, leaves every economy the user is in.
|
||||
*/
|
||||
async leave(sessionOrUserId, maybeEconomy) {
|
||||
if (!this.enabled) return { success: false, reason: 'match_disabled' };
|
||||
|
||||
let userId, economy;
|
||||
if (sessionOrUserId && typeof sessionOrUserId === 'object') {
|
||||
userId = sessionOrUserId.userId;
|
||||
economy = sessionOrUserId.economy || maybeEconomy;
|
||||
} else {
|
||||
userId = sessionOrUserId;
|
||||
economy = maybeEconomy;
|
||||
}
|
||||
if (userId == null) return { success: false, reason: 'invalid_user' };
|
||||
|
||||
// No economy given → leave every economy the user is queued in.
|
||||
if (!economy) {
|
||||
const results = [];
|
||||
for (const eco of ECONOMIES) {
|
||||
results.push(await this._leaveOne(userId, eco));
|
||||
}
|
||||
const ok = results.some(r => r && r.success);
|
||||
return { success: ok, reason: ok ? undefined : 'not_queued' };
|
||||
}
|
||||
|
||||
economy = this._validateEconomy(economy);
|
||||
if (!economy) return { success: false, reason: 'invalid_economy' };
|
||||
return this._leaveOne(userId, economy);
|
||||
}
|
||||
|
||||
async _leaveOne(userId, economy) {
|
||||
try {
|
||||
const memIdx = this._queues[economy].findIndex(e => e.userId === userId);
|
||||
const entry = memIdx >= 0 ? this._queues[economy][memIdx] : null;
|
||||
|
||||
await this.db.withTransaction(async (client) => {
|
||||
const result = await client.query(`
|
||||
DELETE FROM match_queue_entries
|
||||
@@ -257,43 +437,64 @@ class MatchQueue {
|
||||
throw Object.assign(new Error('Not in queue'), { code: 'NOT_QUEUED' });
|
||||
}
|
||||
|
||||
if (economy === 'credits_prestige' && entry?.cost) {
|
||||
const cost = entry.cost;
|
||||
const creditResult = await client.query(`
|
||||
UPDATE users SET credits = credits + $1 WHERE id = $2 RETURNING credits
|
||||
`, [cost, userId]);
|
||||
await client.query(`
|
||||
INSERT INTO credit_transactions (user_id, amount, reason, balance_after, transaction_type, metadata)
|
||||
VALUES ($1, $2, 'match_queue_leave', $3, 'match', $4)
|
||||
`, [userId, cost, creditResult.rows[0].credits, JSON.stringify({ economy: 'credits_prestige', refunded: true })]);
|
||||
}
|
||||
|
||||
if (economy === 'crypto_race') {
|
||||
const ticketResult = await client.query(`
|
||||
UPDATE users SET race_entries = race_entries + 1 WHERE id = $1 RETURNING race_entries
|
||||
`, [userId]);
|
||||
await client.query(`
|
||||
INSERT INTO race_entry_transactions (user_id, delta, balance_after, reason, metadata)
|
||||
VALUES ($1, 1, $2, 'queue_leave', $3)
|
||||
`, [userId, ticketResult.rows[0].race_entries, JSON.stringify({ economy: 'crypto_race', refunded: true })]);
|
||||
}
|
||||
// MP-H4: refund is DB-derived inside _applyRefund, not read from volatile memory.
|
||||
await this._applyRefund(client, userId, economy, 'queue_leave');
|
||||
});
|
||||
|
||||
const memIdx = this._queues[economy].findIndex(e => e.userId === userId);
|
||||
if (memIdx >= 0) this._queues[economy].splice(memIdx, 1);
|
||||
this._log(`[MatchQueue] leave user=${userId} economy=${economy}`);
|
||||
return { success: true };
|
||||
} catch (err) {
|
||||
const normalized = normalizeError(err, 'Failed to leave match queue');
|
||||
if (normalized.code === 'NOT_QUEUED') return { success: false, reason: 'not_queued' };
|
||||
return { success: false, reason: normalized.safeMessage || 'leave_failed' };
|
||||
const norm = normalizeError(err, 'Failed to leave match queue');
|
||||
if (norm.code === 'NOT_QUEUED') {
|
||||
// Keep memory consistent even if the DB row was already gone.
|
||||
const memIdx = this._queues[economy].findIndex(e => e.userId === userId);
|
||||
if (memIdx >= 0) this._queues[economy].splice(memIdx, 1);
|
||||
return { success: false, reason: 'not_queued' };
|
||||
}
|
||||
return { success: false, reason: norm.safeMessage || 'leave_failed' };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drain queued players for an economy into a match. Marks queue entries 'matched' and
|
||||
* returns the set of entrants. The caller creates the MatchRoom and starts it.
|
||||
* @param {string} economy
|
||||
* @param {number} maxPlayers
|
||||
* Refund a set of drained entrants (used when a race is aborted after drain marked them
|
||||
* 'matched', e.g. pot-collection failure or a <2-player defensive drain). Each refund is
|
||||
* written to the ledger and the queue row is cancelled in the same transaction, only when
|
||||
* the row was actually still queued/matched (so it can never double-refund).
|
||||
*/
|
||||
async refundEntries(entries, economy, reason = 'match_cancel') {
|
||||
if (!this.db || !Array.isArray(entries) || entries.length === 0) return;
|
||||
economy = this._validateEconomy(economy);
|
||||
if (!economy) return;
|
||||
|
||||
for (const e of entries) {
|
||||
const userId = e?.userId;
|
||||
if (userId == null) continue;
|
||||
try {
|
||||
await this.db.withTransaction(async (client) => {
|
||||
const upd = await client.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET status = 'cancelled', match_id = NULL
|
||||
WHERE user_id = $1 AND economy = $2 AND status IN ('queued', 'matched')
|
||||
RETURNING id
|
||||
`, [userId, economy]);
|
||||
if (upd.rowCount > 0) {
|
||||
await this._applyRefund(client, userId, economy, reason);
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
this._log('[MatchQueue] refundEntries error', err.message);
|
||||
}
|
||||
const idx = this._queues[economy].findIndex(q => q.userId === userId);
|
||||
if (idx >= 0) this._queues[economy].splice(idx, 1);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drain queued players for an economy into a match. Marks the DB rows 'matched' and returns
|
||||
* the entrants. The in-memory queue is only mutated AFTER the DB commit succeeds, so a
|
||||
* failed drain leaves the queue intact for the next block.
|
||||
* @returns {Promise<{entries:Array}|null>}
|
||||
*/
|
||||
async drain(economy, maxPlayers) {
|
||||
@@ -301,80 +502,84 @@ class MatchQueue {
|
||||
if (!economy || this._queues[economy].length < 2) return null;
|
||||
|
||||
const count = Math.min(maxPlayers, this._queues[economy].length);
|
||||
const drained = this._queues[economy].splice(0, count);
|
||||
const userIds = drained.map(e => e.userId);
|
||||
const candidates = this._queues[economy].slice(0, count);
|
||||
const userIds = candidates.map(e => e.userId);
|
||||
|
||||
await this.db.withTransaction(async (client) => {
|
||||
const result = await client.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET status = 'matched'
|
||||
WHERE user_id = ANY($1::int[]) AND economy = $2 AND status = 'queued'
|
||||
RETURNING id, user_id
|
||||
`, [userIds, economy]);
|
||||
try {
|
||||
await this.db.withTransaction(async (client) => {
|
||||
const result = await client.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET status = 'matched', matched_at = NOW()
|
||||
WHERE user_id = ANY($1::int[]) AND economy = $2 AND status = 'queued'
|
||||
RETURNING id, user_id
|
||||
`, [userIds, economy]);
|
||||
|
||||
if (result.rowCount !== userIds.length) {
|
||||
// Should never happen if memory state is in sync; abort.
|
||||
throw Object.assign(
|
||||
new Error(`Queue drain mismatch: expected ${userIds.length}, got ${result.rowCount}`),
|
||||
{ code: 'DRAIN_MISMATCH' }
|
||||
);
|
||||
}
|
||||
});
|
||||
if (result.rowCount !== userIds.length) {
|
||||
// Memory/DB out of sync — abort without mutating memory; retry next block.
|
||||
throw Object.assign(
|
||||
new Error(`Queue drain mismatch: expected ${userIds.length}, got ${result.rowCount}`),
|
||||
{ code: 'DRAIN_MISMATCH' }
|
||||
);
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
this._log('[MatchQueue] drain error', err.message);
|
||||
return null;
|
||||
}
|
||||
|
||||
this._log(`[MatchQueue] drained ${drained.length} players for ${economy}`);
|
||||
return { entries: drained };
|
||||
// DB commit succeeded — now remove them from the in-memory queue.
|
||||
const idSet = new Set(userIds);
|
||||
this._queues[economy] = this._queues[economy].filter(e => !idSet.has(e.userId));
|
||||
this._log(`[MatchQueue] drained ${candidates.length} players for ${economy}`);
|
||||
return { entries: candidates };
|
||||
}
|
||||
|
||||
/**
|
||||
* Reattach an in-memory entry after a socket reconnect. The DB row is keyed on user_id +
|
||||
* economy, so we just update the socket_id in memory and DB.
|
||||
* @param {number} userId
|
||||
* @param {string} economy
|
||||
* @param {string} newSocketId
|
||||
* @param {string} sessionToken
|
||||
* @returns {Promise<{inQueue:boolean, position?:number}>}
|
||||
* Reattach an in-memory entry after a socket reconnect while still IN THE QUEUE (not in a
|
||||
* running match). The DB row is keyed on user_id + economy.
|
||||
*/
|
||||
async reattach(userId, economy, newSocketId, sessionToken) {
|
||||
economy = this._validateEconomy(economy);
|
||||
if (!economy) return { inQueue: false };
|
||||
|
||||
const token = sessionToken != null ? String(sessionToken) : String(userId);
|
||||
const result = await this.db.query(`
|
||||
UPDATE match_queue_entries
|
||||
SET socket_id = $1, session_token = $2
|
||||
WHERE user_id = $3 AND economy = $4 AND status = 'queued'
|
||||
RETURNING id, created_at
|
||||
`, [newSocketId, sessionToken, userId, economy]);
|
||||
`, [newSocketId, token, userId, economy]);
|
||||
|
||||
if (result.rowCount === 0) return { inQueue: false };
|
||||
|
||||
const memIdx = this._queues[economy].findIndex(e => e.userId === userId);
|
||||
if (memIdx >= 0) {
|
||||
this._queues[economy][memIdx].socketId = newSocketId;
|
||||
this._queues[economy][memIdx].sessionToken = sessionToken;
|
||||
this._queues[economy][memIdx].sessionToken = token;
|
||||
return { inQueue: true, position: memIdx + 1 };
|
||||
}
|
||||
|
||||
// Memory was stale; rebuild this one from DB row.
|
||||
const row = result.rows[0];
|
||||
this._addToMemory(economy, {
|
||||
queueEntryId: row.id,
|
||||
userId,
|
||||
socketId: newSocketId,
|
||||
sessionToken,
|
||||
sessionToken: token,
|
||||
createdAt: new Date(row.created_at).getTime()
|
||||
});
|
||||
return { inQueue: true, position: this._queues[economy].findIndex(e => e.userId === userId) + 1 };
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel all queues (e.g. server shutdown). Refunds paid entries.
|
||||
* Cancel all queues (e.g. server shutdown). Refunds paid entries via _leaveOne (ledger row
|
||||
* written for each).
|
||||
*/
|
||||
async shutdown() {
|
||||
if (!this.enabled || !this.db) return;
|
||||
for (const economy of ECONOMIES) {
|
||||
const entries = this._queues[economy].slice();
|
||||
for (const e of entries) {
|
||||
await this.leave(e.userId, economy).catch(() => {});
|
||||
await this._leaveOne(e.userId, economy).catch(() => {});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user