try to fix this spaghetti of a code

This commit is contained in:
Frederik Spirgi
2026-09-09 10:55:40 +02:00
parent 2194ea4f0d
commit 101116472a
26 changed files with 1938 additions and 0 deletions
+21
View File
@@ -0,0 +1,21 @@
const path = require('node:path');
function configuration(overrides = {}) {
const config = {
port: Number(process.env.PORT || 42069),
host: process.env.HOST || '127.0.0.1',
databasePath: process.env.DATABASE_PATH || path.join(__dirname, '..', 'users.sqlite'),
secret: process.env.TOKEN_SECRET,
origin: process.env.PUBLIC_ORIGIN || null,
trustProxy: process.env.TRUST_PROXY ? process.env.TRUST_PROXY.split(',').map(value => value.trim()) : false,
sessionSeconds: 1800, bcryptRounds: 12, ante: 10,
intermissionMs: 30000, turnMs: 30000, disconnectMs: 10000,
heartbeatMs: 30000, maxConnections: 200, maxConnectionsPerIp: 10,
authLimit: 20, authWindowMs: 15 * 60 * 1000, messageLimit: 30,
now: Date.now, ...overrides
};
if (typeof config.secret !== 'string' || config.secret.length < 32) throw new Error('TOKEN_SECRET must contain at least 32 characters');
if (!Number.isInteger(config.port) || config.port < 0 || config.port > 65535) throw new Error('Invalid PORT');
if (config.origin && new URL(config.origin).origin !== config.origin) throw new Error('PUBLIC_ORIGIN must be an origin without a trailing slash');
return config;
}
module.exports = {configuration};
+19
View File
@@ -0,0 +1,19 @@
const crypto = require('node:crypto');
class AppError extends Error {
constructor(code, message, status = 400) { super(message); this.code = code; this.status = status; }
}
class KeyedQueue {
constructor() { this.tails = new Map(); }
run(keys, task) {
keys = [...new Set(keys)].sort();
const previous = keys.map(key => this.tails.get(key) || Promise.resolve());
const result = Promise.all(previous).then(task);
const tail = result.catch(() => {});
for (const key of keys) this.tails.set(key, tail);
tail.then(() => { for (const key of keys) if (this.tails.get(key) === tail) this.tails.delete(key); });
return result;
}
async drain() { await Promise.all([...this.tails.values()]); }
}
const hash = value => crypto.createHash('sha256').update(value).digest('hex');
module.exports = {AppError, KeyedQueue, hash};
+20
View File
@@ -0,0 +1,20 @@
const crypto = require('node:crypto');
// Claim ownership inside a SQLite transaction, including replacement of a dead PID.
// This avoids races between two processes attempting crash recovery at once.
async function acquireDatabaseLock(db) {
const token = crypto.randomUUID();
await db.exec('BEGIN IMMEDIATE');
try {
await db.exec('CREATE TABLE IF NOT EXISTS runtime_owner (id INTEGER PRIMARY KEY CHECK(id = 1), pid INTEGER NOT NULL, token TEXT NOT NULL)');
const owner = await db.get('SELECT pid FROM runtime_owner WHERE id = 1');
if (owner) {
if (!Number.isInteger(owner.pid) || owner.pid <= 0) throw new Error('Invalid database owner');
try {process.kill(owner.pid, 0); throw new Error(`Database is in use by process ${owner.pid}`);}
catch (error) {if (error.code !== 'ESRCH') throw error;}
}
await db.run('INSERT INTO runtime_owner(id,pid,token) VALUES(1,?,?) ON CONFLICT(id) DO UPDATE SET pid=excluded.pid,token=excluded.token', process.pid, token);
await db.exec('COMMIT');
} catch (error) {await db.exec('ROLLBACK'); throw error;}
return () => db.run('DELETE FROM runtime_owner WHERE id = 1 AND token = ?', token);
}
module.exports = {acquireDatabaseLock};
+143
View File
@@ -0,0 +1,143 @@
const crypto = require('node:crypto');
const {createDeck, evaluateBestHand, calculatePayouts} = require('./poker');
const {AppError} = require('./core');
const dict = () => Object.create(null);
const live = game => game.players.filter(player => game.hands[player] && !game.folded[player]);
const canAct = (game, player) => live(game).includes(player) && game.stacks[player] > 0;
const callAmount = (game, player) => Math.max(0, game.currentBet - (game.playerBets[player] || 0));
const canRaise = (game, player) => game.actedAt[player] == null || game.currentBet - game.actedAt[player] >= game.minRaise;
const turn = game => game.phase === 'betting' ? game.players[game.currentPlayerIndex] : null;
function intermission(players, balances, config, previous = null) {
return {
id: crypto.randomUUID(), players: [...players], stacks: {...balances}, hands: dict(), folded: dict(),
contributions: dict(), playerBets: dict(), actedAt: dict(), communityCards: [], deck: [],
phase: 'intermission', street: 'intermission', round: previous?.round || 0,
dealerIndex: previous?.dealerIndex ?? -1, currentPlayerIndex: -1,
pot: 0, currentBet: 0, minRaise: config.ante, turnEndsAt: null,
nextRoundStartsAt: config.now() + config.intermissionMs, lastAction: 'Waiting for the next hand.',
winner: null, showdown: null, payouts: dict(), departed: []
};
}
function deal(players, balances, config, previous) {
const game = intermission(players, balances, config, previous);
game.round++;
game.dealerIndex = (game.dealerIndex + 1) % players.length;
game.nextRoundStartsAt = null;
const funded = players.filter(player => balances[player] >= config.ante);
if (funded.length < 2) {
game.phase = 'waiting'; game.lastAction = `Waiting for two players with at least ${config.ante} chips.`;
return game;
}
game.deck = (config.createDeck || createDeck)();
game.phase = 'betting'; game.street = 'preflop';
for (const player of players) {
game.contributions[player] = 0; game.playerBets[player] = 0; game.actedAt[player] = null;
game.folded[player] = !funded.includes(player);
if (!game.folded[player]) {
game.hands[player] = [game.deck.pop(), game.deck.pop()];
pay(game, player, config.ante);
// The ante is not a street bet.
game.playerBets[player] = 0;
}
}
game.lastAction = `Round ${game.round}: each player paid ${config.ante} chips.`;
advance(game, config, game.dealerIndex);
return game;
}
function pay(game, player, amount) {
game.stacks[player] -= amount;
game.playerBets[player] += amount;
game.contributions[player] += amount;
game.pot += amount;
}
function settle(game, config) {
const active = live(game);
const evaluations = active.map(player => ({player, hand: game.hands[player], result: active.length === 1
? {score: [0], name: 'Last remaining hand', comboCards: []}
: evaluateBestHand([...game.hands[player], ...game.communityCards])}));
const seats = game.players.slice(game.dealerIndex + 1).concat(game.players.slice(0, game.dealerIndex + 1));
game.payouts = calculatePayouts(game.contributions, evaluations, seats);
for (const [player, amount] of Object.entries(game.payouts)) game.stacks[player] += amount;
const distributed = Object.values(game.payouts).reduce((sum, amount) => sum + amount, 0);
if (distributed !== game.pot) throw new Error('Pot conservation failed');
game.lastAction = `Distributed ${game.pot} chips: ${Object.entries(game.payouts).map(([player, amount]) => `${player} +${amount}`).join(', ')}.`;
game.pot = 0; game.phase = 'intermission'; game.currentPlayerIndex = -1;
game.turnEndsAt = null; game.nextRoundStartsAt = config.now() + config.intermissionMs;
game.winner = Object.keys(game.payouts).join(', ');
game.showdown = active.length > 1 ? evaluations : [];
}
function advance(game, config, from = game.currentPlayerIndex) {
if (live(game).length <= 1) { settle(game, config); return; }
const solvent = live(game).filter(player => game.stacks[player] > 0);
const pending = solvent.filter(player => game.actedAt[player] == null || callAmount(game, player) > 0);
// A sole solvent hand cannot bet into players who are already all-in.
if (pending.length && !(solvent.length === 1 && callAmount(game, solvent[0]) === 0)) {
for (let offset = 1; offset <= game.players.length; offset++) {
const index = (from + offset + game.players.length) % game.players.length;
if (!pending.includes(game.players[index])) continue;
game.currentPlayerIndex = index;
game.turnEndsAt = config.now() + config.turnMs;
return;
}
}
if (game.street === 'river') { settle(game, config); return; }
const next = {preflop: 'flop', flop: 'turn', turn: 'river'}[game.street];
const count = next === 'flop' ? 3 : 1;
for (let i = 0; i < count; i++) game.communityCards.push(game.deck.pop());
game.street = next; game.currentBet = 0; game.minRaise = config.ante;
for (const player of game.players) { game.playerBets[player] = 0; game.actedAt[player] = null; }
advance(game, config, game.dealerIndex);
}
function action(game, player, kind, amount, config) {
if (game.phase !== 'betting') throw new AppError('WRONG_PHASE', 'No hand is accepting actions', 409);
if (turn(game) !== player || !canAct(game, player)) throw new AppError('NOT_YOUR_TURN', 'Wait for your turn', 409);
if (kind === 'timeout') kind = callAmount(game, player) ? 'fold' : 'hit';
if (kind === 'fold') {
game.folded[player] = true; game.lastAction = `${player} folded.`;
} else if (kind === 'hit' || kind === 'bet') {
const call = callAmount(game, player);
if (kind === 'hit') amount = Math.min(call, game.stacks[player]);
if (!Number.isSafeInteger(amount) || amount < 0 || amount > game.stacks[player]) throw new AppError('INVALID_BET', 'Bet must be whole chips within your stack');
const allIn = amount === game.stacks[player];
const target = game.playerBets[player] + amount;
if (amount < call && !allIn) throw new AppError('INVALID_BET', `Call at least ${call} chips or go all-in`);
if (target > game.currentBet) {
if (!canRaise(game, player)) throw new AppError('RAISE_NOT_REOPENED', 'A short all-in has not reopened your raise');
if (live(game).filter(other => other !== player && game.stacks[other] > 0).length === 0) throw new AppError('INVALID_BET', 'No opponent can match a raise');
const raise = target - game.currentBet;
if (raise < game.minRaise && !allIn) throw new AppError('MINIMUM_RAISE', `A full raise must add at least ${game.minRaise} chips above the table bet`);
if (raise >= game.minRaise) game.minRaise = raise;
game.currentBet = target;
}
pay(game, player, amount);
game.actedAt[player] = game.currentBet;
game.lastAction = `${player} ${amount ? `paid ${amount} chips` : 'checked'}.`;
} else throw new AppError('INVALID_ACTION', 'Unknown poker action');
advance(game, config);
}
function leave(game, player, config) {
if (!game || game.phase !== 'betting' || !game.players.includes(player)) return;
game.departed.push(player); game.folded[player] = true;
if (turn(game) === player || live(game).length <= 1) advance(game, config);
else {
const solvent = live(game).filter(other => game.stacks[other] > 0);
if (!solvent.length || (solvent.length === 1 && callAmount(game, solvent[0]) === 0)) advance(game, config);
}
}
function state(game, username, lobbyId, config) {
return {
type: 'poker:state', lobbyId, username, players: game.players, round: game.round,
pot: game.pot, currentBet: game.currentBet, bet: game.currentBet,
communityCards: game.communityCards, hand: game.hands[username] || [],
currentPlayer: turn(game), isYourTurn: turn(game) === username,
queue: game.phase === 'betting' ? game.players.slice(game.currentPlayerIndex).concat(game.players.slice(0, game.currentPlayerIndex)).filter(player => canAct(game, player)) : [],
folded: game.folded, street: game.street, phase: game.phase, lastAction: game.lastAction,
yourContribution: game.contributions[username] || 0, yourStack: game.stacks[username] || 0,
isSpectator: !game.hands[username] || Boolean(game.folded[username]), callAmount: callAmount(game, username),
minimumStake: config.ante, minimumRaise: game.minRaise, canRaise: canRaise(game, username),
nextRoundStartsAt: game.nextRoundStartsAt, turnEndsAt: game.turnEndsAt,
winner: game.winner, showdown: game.showdown, payouts: game.payouts
};
}
module.exports = {intermission, deal, action, leave, state, turn, callAmount};
+89
View File
@@ -0,0 +1,89 @@
const express = require('express');
const cookieParser = require('cookie-parser');
const path = require('node:path');
const swaggerUi = require('swagger-ui-express');
const {AppError} = require('./core');
const {tokenFrom} = require('./transport');
const {validate} = require('../public/protocol');
function createHttpApp(repository, sessions, lobbies, config) {
const app = express();
app.disable('x-powered-by'); app.set('trust proxy', config.trustProxy);
app.use(express.json({limit: '16kb'})); app.use(cookieParser());
app.use((req, res, next) => {
res.set('X-Content-Type-Options', 'nosniff');
res.set('Referrer-Policy', 'same-origin');
if (!['GET', 'HEAD', 'OPTIONS'].includes(req.method) && req.headers.origin) {
const origin = config.origin || `${req.protocol}://${req.get('host')}`;
if (req.headers.origin !== origin) return res.status(403).json({code: 'ORIGIN_DENIED', error: 'Origin denied'});
}
next();
});
const attempts = new Map();
function rateLimit(req, res, next) {
const now = config.now();
for (const [key, value] of attempts) if (value.until <= now) attempts.delete(key);
const key = req.ip;
if (!attempts.has(key)) {
if (attempts.size >= 10000) return res.status(503).json({code: 'BUSY', error: 'Try again later'});
attempts.set(key, {count: 0, until: now + config.authWindowMs});
}
const record = attempts.get(key);
if (++record.count > config.authLimit) {
res.set('Retry-After', String(Math.ceil((record.until - now) / 1000)));
return res.status(429).json({code: 'RATE_LIMITED', error: 'Too many authentication attempts'});
}
next();
}
const cookie = req => ({httpOnly: true, sameSite: 'strict', secure: req.secure, path: '/', maxAge: config.sessionSeconds * 1000});
const authorized = handler => async (req, res) => sessions.authorized(tokenFrom(req), user => handler(req, res, user));
for (const method of ['register', 'login']) app.post(`/${method}`, rateLimit, async (req, res) => {
const result = await sessions[method](req.body);
res.cookie('authorization', result.token, cookie(req)); res.set('Cache-Control', 'no-store'); res.json(result);
});
app.post('/verify', authorized((req, res, user) => res.json({valid: true, user})));
app.delete('/logout', authorized(async (req, res, user) => {
const lobby = lobbies.membership(user.username);
if (lobby) await lobbies.command(user.username, {type: 'lobby:leave', lobbyId: lobby.id, requestId: require('node:crypto').randomUUID()});
await sessions.revoke(user.username);
const options = cookie(req); delete options.maxAge;
res.clearCookie('authorization', options); res.json({message: 'Logged out'});
}));
app.get('/balance', authorized(async (req, res, user) => res.json({balance: (await repository.user(user.username)).money})));
app.get('/leaderboard', authorized(async (req, res, user) => {
const users = await repository.users(); const index = users.findIndex(entry => entry.username === user.username);
res.json({leaderboard: users.slice(0, 10), self: {...users[index], position: index + 1}});
}));
for (const [route, type] of [['/lobby/create', 'lobby:create'], ['/lobby/join', 'lobby:join'], ['/lobby/leave', 'lobby:leave'], ['/poker/start', 'poker:start'], ['/poker/action', 'poker:action']]) {
app.post(route, authorized(async (req, res, user) => {
const message = {...req.body, type};
const invalid = validate(message);
if (invalid) throw new AppError('INVALID_MESSAGE', invalid);
res.json(await lobbies.command(user.username, message));
}));
app.get(route, (req, res) => res.set('Allow', 'POST').status(405).json({code: 'METHOD_NOT_ALLOWED', error: 'Use POST with a requestId'}));
}
app.get('/lobby/public', authorized((req, res) => res.json({lobbies: lobbies.publicList()})));
app.get('/lobby/players', authorized((req, res, user) => {
const lobby = lobbies.get(req.query.lobbyId);
if (!lobby?.players.includes(user.username)) throw new AppError('NOT_SEATED', 'Join this lobby first', 403);
res.json({players: lobby.players});
}));
const publicPath = path.join(__dirname, '..', 'public');
app.get(['/', '/home.html', '/poker.html'], async (req, res) => {
if (!await sessions.verify(tokenFrom(req))) return res.redirect('/login');
res.set('Cache-Control', 'no-store');
res.sendFile(path.join(publicPath, req.path === '/poker.html' ? 'poker.html' : 'home.html'));
});
app.get('/login', (req, res) => res.sendFile(path.join(publicPath, 'login.html')));
app.use('/swagger-ui', swaggerUi.serve, swaggerUi.setup(require('../swagger.json')));
app.use(express.static(publicPath, {index: false}));
app.use((req, res) => res.status(404).json({code: 'NOT_FOUND', error: 'Not found'}));
app.use((error, req, res, next) => {
if (res.headersSent) return next(error);
const status = error.status >= 400 && error.status < 600 ? error.status : 500;
if (status === 500) console.error('Request failed:', error);
res.status(status).json({code: error.code || (status === 500 ? 'INTERNAL_ERROR' : 'INVALID_BODY'), error: status === 500 ? 'Internal server error' : error.message});
});
return app;
}
module.exports = {createHttpApp};
+121
View File
@@ -0,0 +1,121 @@
const crypto = require('node:crypto');
const {EventEmitter} = require('node:events');
const {KeyedQueue, AppError, hash} = require('./core');
const engine = require('./engine');
function createLobbyService(repository, config) {
const lobbies = new Map();
const queue = new KeyedQueue();
const events = new EventEmitter();
const timers = new Map();
let closing = false;
function membership(username) { return [...lobbies.values()].find(lobby => lobby.players.includes(username)); }
function reserved(username) { return [...lobbies.values()].find(lobby => lobby.game?.phase === 'betting' && lobby.game.players.includes(username)); }
async function balances(players) { return Object.fromEntries(await Promise.all(players.map(async player => [player, (await repository.user(player)).money]))); }
function schedule(lobby) {
clearTimeout(timers.get(lobby.id)); timers.delete(lobby.id);
if (closing || !lobby.players.length || !lobby.game) return;
const deadline = lobby.game.phase === 'betting' ? lobby.game.turnEndsAt : lobby.game.nextRoundStartsAt;
if (!deadline) return;
const id = lobby.game.id;
const expectedPhase = lobby.game.phase;
const revision = lobby.revision;
const timer = setTimeout(() => {
command(null, {type: 'timer', lobbyId: lobby.id, handId: id, deadline, phase: expectedPhase, revision, requestId: `timer-${id}-${revision}`})
.catch(error => {
// Failed transactions leave state untouched. Retry the same deadline/idempotency key.
events.emit('failure', error);
if (!closing && lobbies.get(lobby.id) === lobby) {
const retry = setTimeout(() => schedule(lobby), 1000); retry.unref(); timers.set(lobby.id, retry);
}
});
}, Math.max(1, deadline - config.now()));
timer.unref(); timers.set(lobby.id, timer);
}
async function command(username, message) {
if (closing) throw new AppError('SHUTTING_DOWN', 'Server is shutting down', 503);
const membershipMutation = ['lobby:create', 'lobby:join', 'lobby:leave'].includes(message.type);
return queue.run([...(membershipMutation ? ['membership'] : []), ...(username ? [`account:${username}`] : []),
...(message.lobbyId ? [`lobby:${message.lobbyId}`] : [])], async () => {
const id = `${username || 'system'}:${message.requestId}`;
const fingerprint = hash(JSON.stringify(message));
const prior = await repository.command(id);
if (prior) {
if (prior.fingerprint !== fingerprint) throw new AppError('IDEMPOTENCY_CONFLICT', 'Request ID was reused', 409);
return JSON.parse(prior.result);
}
let old = lobbies.get(message.lobbyId);
if (message.type === 'lobby:create') {
if (membership(username) || reserved(username)) throw new AppError('ALREADY_SEATED', 'Leave your table and wait for its hand to settle first', 409);
old = {id: crypto.randomBytes(8).toString('hex'), host: username, players: [], private: message.private || false, game: null};
}
if (!old) throw new AppError('LOBBY_NOT_FOUND', 'Lobby not found', 404);
const lobby = structuredClone(old);
const baseline = Object.assign(Object.create(null), old.game?.stacks || {});
if (lobby.game) {
for (const key of ['stacks', 'hands', 'folded', 'contributions', 'playerBets', 'actedAt', 'payouts']) lobby.game[key] = Object.assign(Object.create(null), lobby.game[key]);
}
if (message.type === 'lobby:create' || message.type === 'lobby:join') {
const current = membership(username) || reserved(username);
if (current && current.id !== lobby.id) throw new AppError('ALREADY_SEATED', 'Leave your table and wait for its hand to settle first', 409);
if (!lobby.players.includes(username)) {
if (lobby.game?.phase === 'betting' && lobby.game.players.includes(username)) throw new AppError('HAND_PENDING', 'Rejoin after your previous hand settles', 409);
if (lobby.players.length >= 10) throw new AppError('LOBBY_FULL', 'Lobby is full', 409);
lobby.players.push(username);
if (lobby.game) {
const balance = (await repository.user(username)).money;
lobby.game.stacks[username] = balance; baseline[username] = balance;
}
}
if (lobby.game?.phase === 'waiting' && (Object.values(await balances(lobby.players)).filter(value => value >= config.ante).length >= 2)) {
lobby.game = engine.intermission(lobby.players, await balances(lobby.players), config, lobby.game);
}
} else if (message.type === 'lobby:leave') {
if (!lobby.players.includes(username)) throw new AppError('NOT_SEATED', 'You are not in this lobby', 409);
lobby.players = lobby.players.filter(player => player !== username);
engine.leave(lobby.game, username, config);
if (lobby.host === username) lobby.host = lobby.players[0] || null;
} else if (message.type === 'poker:start') {
if (lobby.host !== username) throw new AppError('HOST_REQUIRED', 'Only the host can start', 403);
if (lobby.game) throw new AppError('GAME_STARTED', 'Game already started', 409);
if (lobby.players.length < 2) throw new AppError('NOT_ENOUGH_PLAYERS', 'At least two players are required', 409);
lobby.game = engine.intermission(lobby.players, await balances(lobby.players), config);
Object.assign(baseline, lobby.game.stacks);
} else if (message.type === 'poker:action') {
if (!lobby.players.includes(username)) throw new AppError('NOT_SEATED', 'Join this lobby first', 403);
if (!lobby.game) throw new AppError('WRONG_PHASE', 'Host must start the game', 409);
engine.action(lobby.game, username, message.action, message.payload?.amount, config);
} else if (message.type === 'timer') {
if (message.revision !== undefined && message.revision !== lobby.revision) return {ignored: true};
if (!lobby.game || lobby.game.id !== message.handId || lobby.game.phase !== message.phase) return {ignored: true};
const deadline = lobby.game.phase === 'betting' ? lobby.game.turnEndsAt : lobby.game.nextRoundStartsAt;
if (deadline !== message.deadline || deadline > config.now()) return {ignored: true};
if (lobby.game.phase === 'betting') engine.action(lobby.game, engine.turn(lobby.game), 'timeout', undefined, config);
else {
const fresh = await balances(lobby.players);
Object.assign(baseline, fresh);
lobby.game = engine.deal(lobby.players, fresh, config, lobby.game);
}
} else throw new AppError('INVALID_MESSAGE', 'Unsupported command');
const deltas = Object.fromEntries(Object.entries(lobby.game?.stacks || {}).map(([player, stack]) => [player, stack - (baseline[player] ?? stack)]));
const difference = Object.values(deltas).reduce((sum, amount) => sum + amount, 0) + (lobby.game?.pot || 0) - (old.game?.pot || 0);
if (difference !== 0) throw new Error('Command violates chip conservation');
lobby.revision = (old.revision || 0) + 1;
const result = {lobbyId: lobby.id, username, type: message.type, requestId: message.requestId};
await repository.commit({id, fingerprint, lobby, deltas, result, reason: message.type});
if (lobby.players.length) lobbies.set(lobby.id, lobby); else lobbies.delete(lobby.id);
schedule(lobby);
events.emit('changed', lobby, username, message.type);
return result;
});
}
return {command, membership, events,
get: id => lobbies.get(id),
publicList: () => [...lobbies.values()].filter(lobby => !lobby.private).map(lobby => ({lobbyId: lobby.id, host: lobby.host, players: lobby.players.length, gameStarted: Boolean(lobby.game)})),
state: (lobby, username) => lobby.game ? engine.state(lobby.game, username, lobby.id, config) : null,
lobbyState: lobby => ({type: 'lobby:state', lobbyId: lobby.id, host: lobby.host, players: lobby.players, isPublic: !lobby.private,
canStart: lobby.players.length >= 2 && !lobby.game, gameStarted: Boolean(lobby.game), gameUrl: `/poker.html?lobbyId=${lobby.id}`}),
async close() { closing = true; for (const timer of timers.values()) clearTimeout(timer); await queue.drain(); }
};
}
module.exports = {createLobbyService};
+106
View File
@@ -0,0 +1,106 @@
const sqlite3 = require('sqlite3');
const {open} = require('sqlite');
const {KeyedQueue, AppError} = require('./core');
async function createRepository(filename, {beforeCommit = async () => {}} = {}) {
const db = await open({filename, driver: sqlite3.Database});
const queue = new KeyedQueue();
let release;
try {
await db.exec('PRAGMA foreign_keys = ON; PRAGMA busy_timeout = 5000; PRAGMA journal_mode = WAL; PRAGMA synchronous = FULL;');
release = await require('./database-lock').acquireDatabaseLock(db);
} catch (error) {await db.close(); throw error;}
async function transaction(work) {
return queue.run(['database'], async () => {
await db.exec('BEGIN IMMEDIATE');
try {
const result = await work(db);
await beforeCommit();
await db.exec('COMMIT');
return result;
} catch (error) { await db.exec('ROLLBACK'); throw error; }
});
}
try {
const version = await db.get('PRAGMA user_version');
const oldUsers = await db.get("SELECT name FROM sqlite_master WHERE type='table' AND name='users'");
if (version.user_version === 0 && oldUsers && filename !== ':memory:') {
await db.run('VACUUM INTO ?', `${filename}.${Date.now()}.backup`);
}
await transaction(async tx => {
const {user_version: version} = await tx.get('PRAGMA user_version');
if (version > 1) throw new Error('Database schema is newer than this application');
if (version === 1) return;
const old = await tx.get("SELECT name FROM sqlite_master WHERE type='table' AND name='users'");
if (old) await tx.exec('ALTER TABLE users RENAME TO legacy_users');
await tx.exec(`
CREATE TABLE users (
username TEXT PRIMARY KEY COLLATE NOCASE,
password_hash TEXT NOT NULL,
session_hash TEXT,
money INTEGER NOT NULL DEFAULT 200 CHECK(typeof(money) = 'integer' AND money >= 0 AND money <= 9007199254740991)
);
CREATE TABLE lobbies (id TEXT PRIMARY KEY, state TEXT NOT NULL);
CREATE TABLE commands (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, result TEXT NOT NULL, created_at INTEGER NOT NULL);
CREATE TABLE ledger (id INTEGER PRIMARY KEY, command_id TEXT NOT NULL, username TEXT NOT NULL REFERENCES users(username), delta INTEGER NOT NULL, reason TEXT NOT NULL);
CREATE TABLE migrations (id TEXT PRIMARY KEY, applied_at INTEGER NOT NULL);
`);
if (old) {
// Fails and rolls back rather than silently rounding or losing invalid balances.
await tx.exec('INSERT INTO users(username, password_hash, money) SELECT username, password_hash, money FROM legacy_users; DROP TABLE legacy_users;');
}
await tx.exec('PRAGMA user_version = 1');
});
} catch (error) { await release(); await db.close(); throw error; }
const read = work => queue.run(['database'], () => work(db));
return {
transaction,
user: username => read(tx => tx.get('SELECT username, password_hash AS passwordHash, session_hash AS sessionHash, money FROM users WHERE username = ?', username)),
users: () => read(tx => tx.all('SELECT username, money FROM users ORDER BY money DESC, username ASC')),
async register(username, passwordHash, sessionHash) {
try { await transaction(tx => tx.run('INSERT INTO users(username,password_hash,session_hash) VALUES(?,?,?)', username, passwordHash, sessionHash)); }
catch (error) { if (error.code === 'SQLITE_CONSTRAINT') throw new AppError('USERNAME_EXISTS', 'Username already exists', 409); throw error; }
},
setSession: (username, sessionHash) => transaction(tx => tx.run('UPDATE users SET session_hash = ? WHERE username = ?', sessionHash, username)),
async recover() {
return transaction(async tx => {
const rows = await tx.all('SELECT id,state FROM lobbies');
for (const row of rows) {
const lobby = JSON.parse(row.state);
if (lobby.game?.phase !== 'betting') continue;
for (const [username, amount] of Object.entries(lobby.game.contributions)) {
if (!amount) continue;
await tx.run('UPDATE users SET money = money + ? WHERE username = ?', amount, username);
await tx.run('INSERT INTO ledger(command_id,username,delta,reason) VALUES(?,?,?,?)', `recovery:${lobby.game.id}`, username, amount, 'Interrupted hand refund');
}
}
await tx.run('DELETE FROM lobbies');
});
},
command: id => read(tx => tx.get('SELECT fingerprint,result FROM commands WHERE id = ?', id)),
async commit({id, fingerprint, lobby, deltas, result, reason}) {
return transaction(async tx => {
const prior = await tx.get('SELECT fingerprint,result FROM commands WHERE id = ?', id);
if (prior) {
if (prior.fingerprint !== fingerprint) throw new AppError('IDEMPOTENCY_CONFLICT', 'Request ID was reused for a different command', 409);
return {duplicate: true, result: JSON.parse(prior.result)};
}
for (const [username, amount] of Object.entries(deltas)) {
if (!Number.isSafeInteger(amount)) throw new Error('Invalid ledger delta');
if (!amount) continue;
const updated = await tx.run('UPDATE users SET money = money + ? WHERE username = ? AND money + ? >= 0', amount, username, amount);
if (updated.changes !== 1) throw new AppError('INSUFFICIENT_CHIPS', 'Insufficient chips', 409);
await tx.run('INSERT INTO ledger(command_id,username,delta,reason) VALUES(?,?,?,?)', id, username, amount, reason);
}
if (lobby.players.length) await tx.run('INSERT INTO lobbies(id,state) VALUES(?,?) ON CONFLICT(id) DO UPDATE SET state=excluded.state', lobby.id, JSON.stringify(lobby));
else await tx.run('DELETE FROM lobbies WHERE id = ?', lobby.id);
await tx.run('INSERT INTO commands(id,fingerprint,result,created_at) VALUES(?,?,?,?)', id, fingerprint, JSON.stringify(result), Date.now());
return {duplicate: false, result};
});
},
totals: () => read(async tx => ({users: (await tx.get('SELECT COALESCE(SUM(money),0) AS total FROM users')).total,
escrow: (await tx.all('SELECT state FROM lobbies')).reduce((sum, row) => sum + (JSON.parse(row.state).game?.pot || 0), 0)})),
close: async () => { await queue.drain(); await release(); await db.close(); }
};
}
module.exports = {createRepository};
+67
View File
@@ -0,0 +1,67 @@
const jwt = require('jsonwebtoken');
const bcrypt = require('bcrypt');
const crypto = require('node:crypto');
const {EventEmitter} = require('node:events');
const {AppError, KeyedQueue, hash} = require('./core');
const {schemas, matches} = require('../public/protocol');
function createSessionService(repository, config) {
const events = new EventEmitter();
const queue = new KeyedQueue();
let dummyHash;
function credentials(body) {
if (!matches(body, schemas.credentials) || Buffer.byteLength(body.password, 'utf8') > 72) throw new AppError('INVALID_CREDENTIALS', 'Username must be 1-40 characters; password must be 1-72 UTF-8 bytes');
const username = body.username.trim().toLowerCase();
if (!username) throw new AppError('INVALID_CREDENTIALS', 'Username is required');
return {username, password: body.password};
}
function issue(username) { return jwt.sign({username}, config.secret, {algorithm: 'HS256', expiresIn: config.sessionSeconds, jwtid: crypto.randomUUID()}); }
async function verify(token) {
try {
if (typeof token !== 'string') return null;
const decoded = jwt.verify(token, config.secret, {algorithms: ['HS256']});
if (typeof decoded.username !== 'string') return null;
const user = await repository.user(decoded.username);
if (!user?.sessionHash || user.sessionHash !== hash(token)) return null;
return {username: user.username, exp: decoded.exp};
} catch (error) {
if (error instanceof jwt.JsonWebTokenError || error instanceof jwt.TokenExpiredError || error instanceof jwt.NotBeforeError) return null;
throw error;
}
}
return {events, verify,
async register(body) {
const {username, password} = credentials(body);
return queue.run([username], async () => {
const passwordHash = await bcrypt.hash(password, config.bcryptRounds);
const token = issue(username);
await repository.register(username, passwordHash, hash(token));
return {username, token};
});
},
async login(body) {
const {username, password} = credentials(body);
return queue.run([username], async () => {
const user = await repository.user(username);
dummyHash ||= bcrypt.hash('invalid-account-placeholder', config.bcryptRounds);
const valid = await bcrypt.compare(password, user?.passwordHash || await dummyHash);
if (!user || !valid) throw new AppError('INVALID_CREDENTIALS', 'Invalid credentials', 401);
const token = issue(user.username);
await repository.setSession(user.username, hash(token));
events.emit('revoked', user.username);
return {username: user.username, token};
});
},
async authorized(token, work) {
const decoded = typeof token === 'string' ? jwt.decode(token) : null;
if (typeof decoded?.username !== 'string') throw new AppError('UNAUTHORIZED', 'Please sign in', 401);
return queue.run([decoded.username], async () => {
const user = await verify(token);
if (!user) throw new AppError('UNAUTHORIZED', 'Session expired or revoked', 401);
return work(user);
});
},
async revoke(username) { await repository.setSession(username, null); events.emit('revoked', username); },
close: () => queue.drain()
};
}
module.exports = {createSessionService};
+122
View File
@@ -0,0 +1,122 @@
const {WebSocketServer, WebSocket} = require('ws');
const crypto = require('node:crypto');
const {AppError} = require('./core');
const {validate} = require('../public/protocol');
function parseCookies(header = '') {
const result = Object.create(null);
for (const entry of header.split(';')) {
const index = entry.indexOf('='); if (index < 0) continue;
try { result[entry.slice(0, index).trim()] = decodeURIComponent(entry.slice(index + 1).trim()); } catch { /* Ignore malformed cookies. */ }
}
return result;
}
function tokenFrom(request) {
const bearer = /^Bearer ([^ ]+)$/i.exec(request.headers.authorization || '');
return bearer?.[1] || request.cookies?.authorization || parseCookies(request.headers.cookie).authorization;
}
function connectTransport(server, sessions, lobbies, config) {
const wss = new WebSocketServer({noServer: true, maxPayload: 16 * 1024});
const connections = new Map();
const disconnects = new Map();
let stopping = false;
function send(socket, data) {
if (socket.readyState !== WebSocket.OPEN) return;
if (socket.bufferedAmount > 256 * 1024) { socket.close(1013, 'Client is too slow'); return; }
socket.send(JSON.stringify(data));
}
function snapshot(socket) {
const lobby = lobbies.get(socket.lobbyId);
if (!lobby?.players.includes(socket.user.username)) return;
send(socket, lobbies.lobbyState(lobby));
if (lobby.game) send(socket, lobbies.state(lobby, socket.user.username));
}
function disconnected(username) {
if (stopping || [...wss.clients].some(socket => socket.user?.username === username && socket.readyState === WebSocket.OPEN)) return;
clearTimeout(disconnects.get(username));
const timer = setTimeout(() => {
disconnects.delete(username);
if ([...wss.clients].some(socket => socket.user?.username === username && socket.readyState === WebSocket.OPEN)) return;
const lobby = lobbies.membership(username);
if (lobby) lobbies.command(username, {type: 'lobby:leave', lobbyId: lobby.id, requestId: crypto.randomUUID()}).catch(error => {
console.error('Disconnect leave failed:', error.message);
disconnected(username);
});
}, config.disconnectMs);
timer.unref(); disconnects.set(username, timer);
}
server.on('upgrade', (request, socket, head) => {
const ip = request.socket.remoteAddress;
const reject = status => { socket.end(`HTTP/1.1 ${status}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`); };
if (stopping) return reject('503 Service Unavailable');
if (request.url !== '/') return reject('404 Not Found');
const expectedOrigin = config.origin || `${request.socket.encrypted ? 'https' : 'http'}://${request.headers.host}`;
if (request.headers.origin ? request.headers.origin !== expectedOrigin : !request.headers.authorization) return reject('403 Forbidden');
if ([...connections.values()].reduce((sum, count) => sum + count, 0) >= config.maxConnections || (connections.get(ip) || 0) >= config.maxConnectionsPerIp) return reject('429 Too Many Requests');
connections.set(ip, (connections.get(ip) || 0) + 1);
let released = false;
const release = () => { if (released) return; released = true; const count = connections.get(ip) - 1; if (count) connections.set(ip, count); else connections.delete(ip); };
socket.once('close', release);
socket.on('error', () => {});
const token = tokenFrom(request);
sessions.authorized(token, user => {
if (socket.destroyed || stopping) return socket.destroy();
wss.handleUpgrade(request, socket, head, ws => {
ws.user = user; ws.token = token; ws.alive = true; ws.pending = 0;
ws.windowAt = config.now(); ws.messages = 0;
clearTimeout(disconnects.get(user.username)); disconnects.delete(user.username);
ws.lobbyId = lobbies.membership(user.username)?.id || null;
ws.on('error', error => console.error('WebSocket error:', error.message));
ws.on('pong', () => { ws.alive = true; });
const expiry = setTimeout(() => ws.close(1008, 'Session expired'), Math.max(1, user.exp * 1000 - config.now()));
expiry.unref();
ws.on('close', () => { clearTimeout(expiry); release(); disconnected(user.username); });
ws.on('message', raw => {
if (config.now() - ws.windowAt >= 1000) { ws.windowAt = config.now(); ws.messages = 0; }
if (++ws.messages > config.messageLimit || ws.pending >= 8) { ws.close(1008, 'Message limit exceeded'); return; }
let message;
try { message = JSON.parse(raw.toString()); } catch { send(ws, {type: 'error', code: 'INVALID_JSON', message: 'Invalid JSON'}); return; }
const invalid = validate(message);
if (invalid) { send(ws, {type: 'error', code: 'INVALID_MESSAGE', message: invalid, requestId: message?.requestId}); return; }
ws.pending++;
sessions.authorized(token, async current => {
if (message.type === 'poker:join') {
const lobby = lobbies.get(message.lobbyId);
if (!lobby?.players.includes(current.username)) throw new AppError('NOT_SEATED', 'You are not in this lobby', 403);
ws.lobbyId = lobby.id; snapshot(ws); return;
}
const result = await lobbies.command(current.username, message);
send(ws, {...result, type: 'command:ack', commandType: result.type});
}).catch(error => {
send(ws, {type: 'error', code: error.code || 'INTERNAL_ERROR', message: error.status ? error.message : 'Command failed; retry with the same request ID', requestId: message.requestId});
if (error.status === 401) ws.close(1008, 'Session expired or revoked');
if (!error.status) console.error('Socket command failed:', error);
}).finally(() => ws.pending--);
});
// Handlers exist before readiness is announced.
send(ws, {type: 'auth:success', username: user.username}); snapshot(ws);
});
}).catch(error => { if (error.status === 401) reject('401 Unauthorized'); else {console.error('Upgrade failed:', error.message); reject('500 Internal Server Error');} });
});
const onChanged = (lobby, username, type) => {
for (const socket of wss.clients) {
if (socket.user?.username === username && ['lobby:create', 'lobby:join'].includes(type)) socket.lobbyId = lobby.id;
if (socket.lobbyId !== lobby.id) continue;
if (!lobby.players.includes(socket.user.username)) {
socket.lobbyId = null; send(socket, {type: 'lobby:left', lobbyId: lobby.id});
} else snapshot(socket);
}
};
const onRevoked = username => { for (const socket of wss.clients) if (socket.user?.username === username) socket.close(1008, 'Session expired or revoked'); };
lobbies.events.on('changed', onChanged); sessions.events.on('revoked', onRevoked);
const heartbeat = setInterval(() => {
for (const socket of wss.clients) { if (!socket.alive) socket.terminate(); else { socket.alive = false; socket.ping(); } }
}, config.heartbeatMs);
heartbeat.unref();
return {wss, async close() {
stopping = true; clearInterval(heartbeat); for (const timer of disconnects.values()) clearTimeout(timer);
lobbies.events.off('changed', onChanged); sessions.events.off('revoked', onRevoked);
for (const socket of wss.clients) socket.terminate();
await new Promise(resolve => wss.close(resolve));
}};
}
module.exports = {connectTransport, tokenFrom, parseCookies};