diff --git a/internal/storage/db.go b/internal/storage/db.go index 9cf9ca8..38db698 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -94,6 +94,9 @@ func runMigrations(d *sql.DB) error { addColumnIfMissing(d, "post_log", "url_canonical", "TEXT") addColumnIfMissing(d, "post_log", "forced", "INTEGER NOT NULL DEFAULT 0") addColumnIfMissing(d, "round_robin_state", "last_channel", "TEXT") + // Occupancy of a shared table. Rows written before the casino went multiplayer + // are solo games and read as NULL, which is exactly what they are. + addColumnIfMissing(d, "game_live_hands", "table_id", "TEXT") // FTS5 virtual tables don't support IF NOT EXISTS reliably. // Check sqlite_master before creating. diff --git a/internal/storage/games.go b/internal/storage/games.go index ec0018a..7cf6f14 100644 --- a/internal/storage/games.go +++ b/internal/storage/games.go @@ -563,6 +563,12 @@ type LiveHand struct { State []byte Seed1 uint64 Seed2 uint64 + // TableID is set when the player is sitting at a shared table instead of playing + // alone. The cards are then in game_tables and State here is empty: this row is + // the occupancy claim and nothing else. One row per player either way, which is + // the point — the primary key that stops a second solo hand is the same one that + // stops a second seat. + TableID string } // StartLiveHand seats a *new* hand, and refuses if the player is already in one. @@ -609,16 +615,17 @@ func LoadLiveHand(user string) (LiveHand, error) { var h LiveHand var state string var s1, s2 int64 + var tableID sql.NullString err := Get().QueryRow( - `SELECT game, state, seed1, seed2 FROM game_live_hands WHERE matrix_user = ?`, user, - ).Scan(&h.Game, &state, &s1, &s2) + `SELECT game, state, seed1, seed2, table_id FROM game_live_hands WHERE matrix_user = ?`, user, + ).Scan(&h.Game, &state, &s1, &s2, &tableID) if errors.Is(err, sql.ErrNoRows) { return LiveHand{}, ErrNoLiveHand } if err != nil { return LiveHand{}, fmt.Errorf("games: load live hand: %w", err) } - h.State, h.Seed1, h.Seed2 = []byte(state), uint64(s1), uint64(s2) + h.State, h.Seed1, h.Seed2, h.TableID = []byte(state), uint64(s1), uint64(s2), tableID.String return h, nil } diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 377fa24..6d9a3dc 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -239,9 +239,103 @@ CREATE TABLE IF NOT EXISTS game_live_hands ( state TEXT NOT NULL, -- JSON: the engine's State seed1 INTEGER NOT NULL, -- carried to the audit log when it settles seed2 INTEGER NOT NULL, + -- Set when the player is sitting at a shared table rather than playing alone. + -- The engine state then lives in game_tables.state, not here, and this row is + -- purely the occupancy claim: its PRIMARY KEY is what stops one player being + -- in two games at once, and it is the row the cash-out check reads. Making + -- game_seats a second uniqueness domain instead would be a split brain — see + -- the comment on game_seats. + table_id TEXT, updated_at INTEGER NOT NULL ); +-- --------------------------------------------------------------------------- +-- Shared tables: the casino with more than one person at it. +-- --------------------------------------------------------------------------- + +-- A table other people can sit at. The state column is the engine's State, +-- exactly as game_live_hands holds it for a solo game — one blob for the whole +-- felt, because a pot is not divisible into per-player rows. +-- +-- version is the concurrency authority, and the mutex in the web layer is only +-- an optimisation on top of it. Every state write is a conditional UPDATE +-- against the version the writer read; zero rows affected means somebody moved +-- first. This has to live in the database rather than in a mutex map because a +-- mutex does not survive a redeploy — during a drain, two processes hold two +-- different mutexes over the same row and both believe they are alone. +CREATE TABLE IF NOT EXISTS game_tables ( + id TEXT PRIMARY KEY, + game TEXT NOT NULL, -- 'holdem' | 'uno' | 'blackjack' + tier TEXT NOT NULL, -- the stake, as that game names it + state TEXT NOT NULL, -- JSON: the engine's State + seed1 INTEGER NOT NULL, + seed2 INTEGER NOT NULL, + phase TEXT NOT NULL, -- the engine's phase, lifted out so the lobby can read it + hand_no INTEGER NOT NULL DEFAULT 0, -- with id, the identity of one hand: the payout key + version INTEGER NOT NULL DEFAULT 0, + -- Unix seconds by which the seat to act must act, or 0 for no clock. The turn + -- clock scans this. It is set only when the turn lands on a human: bots resolve + -- inside ApplyMove and are never waited for. + deadline INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_game_tables_due ON game_tables(deadline) WHERE deadline > 0; +CREATE INDEX IF NOT EXISTS idx_game_tables_lobby ON game_tables(game, updated_at DESC); + +-- Who is sitting where. A seat with no matrix_user is a bot. +-- +-- This is deliberately *not* a uniqueness domain for players: there is no unique +-- index on matrix_user, and there must not be one. Occupancy is decided by +-- game_live_hands' primary key, which already stops a player being in two games, +-- already makes a double-clicked join a 409, and is already what the cash-out +-- check reads. A second domain that could disagree with the first would silently +-- switch all three off — the worst of them being a player who cashes out to zero +-- while sitting at a poker table with chips in the pot. +-- +-- staked is what the player brought to the table and has not yet taken home. It +-- is the chip-conservation anchor: the chips are off their game_chips stack and +-- inside the table blob, where the idle reaper cannot see them. +CREATE TABLE IF NOT EXISTS game_seats ( + table_id TEXT NOT NULL, + seat INTEGER NOT NULL, + matrix_user TEXT, -- NULL for a bot + name TEXT NOT NULL, + staked INTEGER NOT NULL DEFAULT 0, + -- Set once a human's clock has run out on them. An absent human is not a bot, + -- but the bot loop has to be allowed past their seat or a table with three + -- ghosts spends a minute an orbit folding air. They come back the moment they act. + away INTEGER NOT NULL DEFAULT 0, + last_seen INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (table_id, seat) +); +CREATE INDEX IF NOT EXISTS idx_game_seats_user ON game_seats(matrix_user) WHERE matrix_user IS NOT NULL; + +-- There is no payout ledger here, and its absence is deliberate — the design +-- called for one and the money model made it unnecessary. Chips cross into a +-- table when a player sits down and back out when they get up; a hand ending +-- moves the pot *within* the state blob and credits nobody's game_chips row. So +-- there is no money write to make idempotent: a settle is a state write, +-- conditional on the version, and a replayed one affects zero rows and rolls +-- back. See the header of internal/storage/tables.go. + +-- Chat on the felt. Messages only — no typing indicators, which is the one thing +-- that would have justified a socket. It does not mirror into Matrix. +-- +-- hand_no is kept against every line for a reason: at a table of real people, +-- collusion looks like chat, and the only way to ever answer that question is to +-- be able to read what was said during the hand it was said in. +CREATE TABLE IF NOT EXISTS game_chat ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + table_id TEXT NOT NULL, + hand_no INTEGER NOT NULL, + matrix_user TEXT, -- NULL when the house is talking + name TEXT NOT NULL, + body TEXT NOT NULL, + said_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_game_chat_table ON game_chat(table_id, id); + -- The trivia bank: questions pulled from the Open Trivia Database ahead of time, -- so that asking one is a local read. -- diff --git a/internal/storage/tables.go b/internal/storage/tables.go new file mode 100644 index 0000000..e795f6f --- /dev/null +++ b/internal/storage/tables.go @@ -0,0 +1,684 @@ +package storage + +import ( + "crypto/rand" + "database/sql" + "encoding/base64" + "errors" + "fmt" + "strings" +) + +// Shared tables: the casino with more than one person at it. +// +// The money model is the thing to understand first, because everything else +// follows from it, and it is *not* the one the plan sketched. +// +// **Chips cross into a table when you sit down, and back out when you get up. +// Nothing in between touches game_chips at all.** Your buy-in leaves your stack +// and becomes a stack inside the engine's own state; antes, bets, pots and +// payouts are all moves within that blob; and getting up is the single write +// that turns what is in front of you back into chips. +// +// This is how hold'em already worked as a solo session, and generalising it is +// what makes a pot safe. The obvious alternative — settle each hand by crediting +// every winner's game_chips row — puts a real money write on the end of every +// hand, and a crash between the credit and the state write pays the winner twice. +// Here a settle credits nobody: it is a state write, conditional on the version, +// and a replay of it affects zero rows and rolls back. The pot cannot be paid +// twice because it is never *paid* at all, only moved. +// +// So the money invariant across a table's whole life is: +// +// sum(seat stacks in the blob) + pot == sum(game_seats.staked) - rake taken +// +// and the only two statements that move chips across the border are the stake in +// SitDown and the award in LeaveTable — each of them one transaction, each of +// them carrying the state write that justifies it. +// +// The other half is concurrency. A table has two writers where a solo game had +// one: an HTTP move, and a turn clock acting for whoever walked away. The version +// column is the authority — every state write is conditional on the version its +// writer read — and the striped mutex in the web layer is only an optimisation on +// top. Correctness has to live in the database, because a mutex does not survive +// a redeploy: during a drain, two processes hold two different mutexes over the +// same row and both of them believe they are alone. + +var ( + // ErrStaleTable means somebody else wrote the table first. The caller's read is + // out of date; it must reload and decide again. This is not an error condition + // so much as the normal outcome of a race, and it is what a 409 is made of. + ErrStaleTable = errors.New("games: the table moved on") + // ErrNoSuchTable means there is no table with that id. + ErrNoSuchTable = errors.New("games: no such table") + // ErrSeatTaken means somebody sat down there between the read and the write. + ErrSeatTaken = errors.New("games: that seat is taken") +) + +// Table is a felt other people can sit at. +// +// State is the engine's State, serialized whole — the same blob game_live_hands +// holds for a solo game, and for the same reason: the deck is in it, so it never +// leaves the server, and a hand survives a redeploy. +type Table struct { + ID string + Game string + Tier string + State []byte + Seed1 uint64 + Seed2 uint64 + Phase string + // HandNo, with the id, is the identity of one hand. It is what the audit trail + // keys on now that a seed no longer reproduces a hand: at a shared table the + // cards fall the way they do because of the order the others acted, not just + // the seed, so "deal it again from seed1/seed2" stopped being a true story the + // moment there was a second player. + HandNo int64 + // Version is the concurrency authority. Read it, write against it, and a write + // that finds it moved is a write that lost the race. + Version int64 + // Deadline is the unix second by which the seat to act has to act, or 0 for no + // clock at all. Only a *human* to act sets one: bots resolve inside ApplyMove + // and there is nobody to wait for. + Deadline int64 + CreatedAt int64 + UpdatedAt int64 +} + +// Seat is one chair. A seat with no MatrixUser is a bot, which is what makes solo +// play just "a table nobody else has joined yet" rather than a second mode. +type Seat struct { + Seat int + MatrixUser string // "" for a bot + Name string + // Staked is what this player brought and has not yet taken home. The chips are + // off their game_chips stack and inside the table blob, where the idle reaper + // cannot see them — so this is the row that says they exist. + Staked int64 + Away bool + LastSeen int64 +} + +// Bot reports whether nobody is sitting here. +func (s Seat) Bot() bool { return s.MatrixUser == "" } + +// NewTableID mints a table id. Short enough to put in a URL, random enough that +// nobody guesses their way onto somebody else's felt. +func NewTableID() (string, error) { + b := make([]byte, 9) + if _, err := rand.Read(b); err != nil { + return "", fmt.Errorf("games: mint table id: %w", err) + } + return base64.RawURLEncoding.EncodeToString(b), nil +} + +// OpenTable creates a table and seats it — bots in every chair nobody has taken. +func OpenTable(t Table, seats []Seat) error { + now := nowUnix() + tx, err := Get().Begin() + if err != nil { + return fmt.Errorf("games: begin open table: %w", err) + } + defer tx.Rollback() //nolint:errcheck // no-op once committed + + if _, err := tx.Exec( + `INSERT INTO game_tables (id, game, tier, state, seed1, seed2, phase, hand_no, version, deadline, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)`, + t.ID, t.Game, t.Tier, string(t.State), int64(t.Seed1), int64(t.Seed2), + t.Phase, t.HandNo, t.Deadline, now, now, + ); err != nil { + return fmt.Errorf("games: open table: %w", err) + } + for _, s := range seats { + if err := upsertSeat(tx, t.ID, s, now); err != nil { + return err + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("games: commit open table: %w", err) + } + return nil +} + +// upsertSeat writes a seat row inside an open transaction, bots included. +func upsertSeat(tx *sql.Tx, tableID string, s Seat, now int64) error { + var user any + if s.MatrixUser != "" { + user = s.MatrixUser + } + if _, err := tx.Exec( + `INSERT INTO game_seats (table_id, seat, matrix_user, name, staked, away, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(table_id, seat) DO UPDATE SET + matrix_user = excluded.matrix_user, name = excluded.name, + staked = excluded.staked, away = excluded.away, last_seen = excluded.last_seen`, + tableID, s.Seat, user, s.Name, s.Staked, boolInt(s.Away), now, + ); err != nil { + return fmt.Errorf("games: seat: %w", err) + } + return nil +} + +func boolInt(b bool) int64 { + if b { + return 1 + } + return 0 +} + +// LoadTable reads a table and everyone at it. +func LoadTable(id string) (Table, []Seat, error) { + var t Table + var state string + var s1, s2 int64 + err := Get().QueryRow( + `SELECT id, game, tier, state, seed1, seed2, phase, hand_no, version, deadline, created_at, updated_at + FROM game_tables WHERE id = ?`, id, + ).Scan(&t.ID, &t.Game, &t.Tier, &state, &s1, &s2, &t.Phase, &t.HandNo, &t.Version, &t.Deadline, &t.CreatedAt, &t.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return Table{}, nil, ErrNoSuchTable + } + if err != nil { + return Table{}, nil, fmt.Errorf("games: load table: %w", err) + } + t.State, t.Seed1, t.Seed2 = []byte(state), uint64(s1), uint64(s2) + + seats, err := tableSeats(id) + if err != nil { + return Table{}, nil, err + } + return t, seats, nil +} + +// tableSeats reads the chairs, in seat order. +func tableSeats(id string) ([]Seat, error) { + rows, err := Get().Query( + `SELECT seat, matrix_user, name, staked, away, last_seen + FROM game_seats WHERE table_id = ? ORDER BY seat`, id) + if err != nil { + return nil, fmt.Errorf("games: table seats: %w", err) + } + defer rows.Close() + + var out []Seat + for rows.Next() { + var s Seat + var user sql.NullString + var away int64 + if err := rows.Scan(&s.Seat, &user, &s.Name, &s.Staked, &away, &s.LastSeen); err != nil { + return nil, fmt.Errorf("games: scan seat: %w", err) + } + s.MatrixUser, s.Away = user.String, away != 0 + out = append(out, s) + } + return out, rows.Err() +} + +// TableSummary is a table as the lobby lists it: enough to decide whether to sit +// down, and nothing that would give away a card. +type TableSummary struct { + ID string `json:"id"` + Game string `json:"game"` + Tier string `json:"tier"` + Phase string `json:"phase"` + Humans int `json:"humans"` + Seats int `json:"seats"` + UpdatedAt int64 `json:"updated_at"` +} + +// LobbyTables lists the live tables, most recently played first. A game of "" is +// all of them. +func LobbyTables(game string, limit int) ([]TableSummary, error) { + if limit <= 0 { + limit = 50 + } + q := `SELECT t.id, t.game, t.tier, t.phase, t.updated_at, + COUNT(s.seat), + COUNT(s.matrix_user) + FROM game_tables t + LEFT JOIN game_seats s ON s.table_id = t.id` + args := []any{} + if game != "" { + q += ` WHERE t.game = ?` + args = append(args, game) + } + q += ` GROUP BY t.id ORDER BY t.updated_at DESC LIMIT ?` + args = append(args, limit) + + rows, err := Get().Query(q, args...) + if err != nil { + return nil, fmt.Errorf("games: lobby: %w", err) + } + defer rows.Close() + + var out []TableSummary + for rows.Next() { + var s TableSummary + if err := rows.Scan(&s.ID, &s.Game, &s.Tier, &s.Phase, &s.UpdatedAt, &s.Seats, &s.Humans); err != nil { + return nil, fmt.Errorf("games: scan lobby row: %w", err) + } + out = append(out, s) + } + return out, rows.Err() +} + +// ---- writing a table back -------------------------------------------------- + +// TableCommit is one write-back of a shared table. +// +// There is no payout field, and its absence is the design. A hand ending at a +// shared table moves chips *within* the blob — the pot becomes somebody's stack — +// so settling one credits nobody and mints nothing. What it writes is the state, +// the audit rows, and whatever the seats now look like. The version makes it +// exactly-once: a settle that runs twice loses the race with itself. +type TableCommit struct { + // Table carries the new state and the version that was *read*. The write is + // conditional on that version and bumps it. + Table Table + // Seats to rewrite — a stack that changed hands, a seat that came back from + // away. Seats not named here are left alone. + Seats []Seat + // Audit is the per-seat record of a hand that just ended. Empty mid-hand. + Audit []Hand +} + +// CommitTable writes a table back, and the hand it just finished with it, in one +// transaction. +// +// The version check is the whole safety property, and it is why this can be +// called by the turn clock and an HTTP move at the same instant without either +// having to trust the other. Whoever gets there first bumps the version; the +// loser's UPDATE matches zero rows, the transaction rolls back, and it comes back +// ErrStaleTable with nothing written — no half-settled hand, no audit row for a +// hand that did not happen. +// +// Nothing in here may call Get().Exec. The pool runs at MaxOpenConns(1), so a +// bare Exec inside an open transaction waits forever for the connection that this +// transaction is holding — and takes the news app down with it. See CommitHand. +func CommitTable(c TableCommit) error { + now := nowUnix() + + tx, err := Get().Begin() + if err != nil { + return fmt.Errorf("games: begin commit table: %w", err) + } + defer tx.Rollback() //nolint:errcheck // no-op once committed + + if err := saveTable(tx, c.Table, now); err != nil { + return err + } + for _, s := range c.Seats { + if err := upsertSeat(tx, c.Table.ID, s, now); err != nil { + return err + } + } + for _, h := range c.Audit { + if err := recordHand(tx, h, now); err != nil { + return err + } + // Playing a hand is the most deliberate thing a player does. Keep the reaper + // off them — their chips are inside the table blob, where it cannot see them + // anyway, but their game_chips row is what it reads. + if h.MatrixUser == "" { + continue + } + if _, err := tx.Exec( + `UPDATE game_chips SET last_played = ?, updated_at = ? WHERE matrix_user = ?`, + now, now, h.MatrixUser, + ); err != nil { + return fmt.Errorf("games: touch session: %w", err) + } + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("games: commit table: %w", err) + } + return nil +} + +// saveTable is the conditional state write: it lands only if the version is still +// the one the caller read. +func saveTable(tx *sql.Tx, t Table, now int64) error { + res, err := tx.Exec( + `UPDATE game_tables SET state = ?, phase = ?, hand_no = ?, seed1 = ?, seed2 = ?, + deadline = ?, version = version + 1, updated_at = ? + WHERE id = ? AND version = ?`, + string(t.State), t.Phase, t.HandNo, int64(t.Seed1), int64(t.Seed2), + t.Deadline, now, t.ID, t.Version, + ) + if err != nil { + return fmt.Errorf("games: save table: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrStaleTable + } + return nil +} + +// ---- sitting down and getting up ------------------------------------------- + +// Sit is one player taking one chair, with the table state that has them in it. +type Sit struct { + Table Table // the new state, and the version that was read + Seat Seat // MatrixUser, Name and the chair; Staked is the buy-in + BuyIn int64 +} + +// SitDown moves a player's chips onto a table and puts them in a seat — the first +// of the only two statements in the casino that cross the chip/table border. +// +// It is one transaction and every step of it can refuse: +// +// - the chips leave in the same statement that checks they are there, so two +// joins fired at once cannot spend the same chip; +// - the occupancy claim is game_live_hands' primary key, exactly as it is for a +// solo hand, so a player cannot be at two tables (or at a table and in a solo +// game) at once, and a double-clicked Join is a 409; +// - the seat is taken only if a bot is sitting in it, so two players racing for +// the last chair cannot both win; +// - and the state write is conditional on the version, so the engine's idea of +// who is at the table cannot drift from the seat rows. +// +// Any of those failing rolls back the buy-in with it. +func SitDown(s Sit) error { + if s.BuyIn <= 0 { + return ErrBadAmount + } + now := nowUnix() + + tx, err := Get().Begin() + if err != nil { + return fmt.Errorf("games: begin sit: %w", err) + } + defer tx.Rollback() //nolint:errcheck // no-op once committed + + res, err := tx.Exec( + `UPDATE game_chips SET chips = chips - ?, last_played = ?, updated_at = ? + WHERE matrix_user = ? AND chips >= ?`, + s.BuyIn, now, now, s.Seat.MatrixUser, s.BuyIn, + ) + if err != nil { + return fmt.Errorf("games: stake buy-in: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrInsufficientChips + } + + // The occupancy claim. The state column is empty on purpose: the cards live in + // game_tables, and this row exists to be a primary key. + res, err = tx.Exec( + `INSERT INTO game_live_hands (matrix_user, game, state, seed1, seed2, table_id, updated_at) + VALUES (?, ?, '', 0, 0, ?, ?) + ON CONFLICT(matrix_user) DO NOTHING`, + s.Seat.MatrixUser, s.Table.Game, s.Table.ID, now, + ) + if err != nil { + return fmt.Errorf("games: claim seat: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrHandInProgress + } + + // Take the chair, but only out of a bot's hands. + res, err = tx.Exec( + `UPDATE game_seats SET matrix_user = ?, name = ?, staked = ?, away = 0, last_seen = ? + WHERE table_id = ? AND seat = ? AND matrix_user IS NULL`, + s.Seat.MatrixUser, s.Seat.Name, s.BuyIn, now, s.Table.ID, s.Seat.Seat, + ) + if err != nil { + return fmt.Errorf("games: take seat: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrSeatTaken + } + + if err := saveTable(tx, s.Table, now); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("games: commit sit: %w", err) + } + return nil +} + +// Leave is one player getting up, with the table state that no longer has them. +type Leave struct { + Table Table // the new state, and the version that was read + Seat int + MatrixUser string + // Bot is who takes the chair over. A table always has a full complement, so + // getting up hands the seat back to the house rather than leaving a hole. + Bot string + // Amount is what is in front of them: everything they are taking home. It may + // be more than they brought, or nothing at all. + Amount int64 + // Audit, if the leaving itself settles something worth recording. + Audit []Hand +} + +// LeaveTable turns what is in front of a player back into chips — the second and +// last statement that crosses the chip/table border. +// +// One transaction, and the state write is in it. As two statements this is a +// double-pay waiting to happen: award 1,240 chips, fail the state write, and the +// player reloads to find their seat still there with 1,240 in front of it. They +// get up again, and again. +func LeaveTable(l Leave) error { + now := nowUnix() + + tx, err := Get().Begin() + if err != nil { + return fmt.Errorf("games: begin leave: %w", err) + } + defer tx.Rollback() //nolint:errcheck // no-op once committed + + if err := saveTable(tx, l.Table, now); err != nil { + return err + } + if err := upsertSeat(tx, l.Table.ID, Seat{Seat: l.Seat, Name: l.Bot}, now); err != nil { + return err + } + if _, err := tx.Exec( + `DELETE FROM game_live_hands WHERE matrix_user = ? AND table_id = ?`, + l.MatrixUser, l.Table.ID, + ); err != nil { + return fmt.Errorf("games: release seat claim: %w", err) + } + if err := award(tx, l.MatrixUser, l.Amount, now); err != nil { + return err + } + for _, h := range l.Audit { + if err := recordHand(tx, h, now); err != nil { + return err + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("games: commit leave: %w", err) + } + return nil +} + +// TableOf reports which table a player is sitting at, if any. Read off the +// occupancy claim, so it agrees with the cash-out check by construction. +func TableOf(user string) (string, error) { + var id sql.NullString + err := Get().QueryRow( + `SELECT table_id FROM game_live_hands WHERE matrix_user = ?`, user, + ).Scan(&id) + if errors.Is(err, sql.ErrNoRows) { + return "", ErrNoLiveHand + } + if err != nil { + return "", fmt.Errorf("games: table of: %w", err) + } + return id.String, nil +} + +// CloseTable deletes a table nobody is sitting at. Called when the last human +// gets up: a felt with six bots on it and nobody watching is not a game, it is a +// row that the lobby would advertise forever. +func CloseTable(id string) error { + tx, err := Get().Begin() + if err != nil { + return fmt.Errorf("games: begin close table: %w", err) + } + defer tx.Rollback() //nolint:errcheck // no-op once committed + + var humans int + if err := tx.QueryRow( + `SELECT COUNT(*) FROM game_seats WHERE table_id = ? AND matrix_user IS NOT NULL`, id, + ).Scan(&humans); err != nil { + return fmt.Errorf("games: count humans: %w", err) + } + if humans > 0 { + return nil // somebody is still playing; the table stays + } + for _, q := range []string{ + `DELETE FROM game_seats WHERE table_id = ?`, + `DELETE FROM game_chat WHERE table_id = ?`, + `DELETE FROM game_tables WHERE id = ?`, + } { + if _, err := tx.Exec(q, id); err != nil { + return fmt.Errorf("games: close table: %w", err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("games: commit close table: %w", err) + } + return nil +} + +// ---- the turn clock -------------------------------------------------------- + +// TableRef is a table the clock has found expired: which one, and at what version +// it was seen. +// +// The version is the point. The clock acts only if it is *still* that version by +// the time it takes the lock, because otherwise: Bob's raise lands in the same +// second his clock expires, the action passes to Cara, and the clock — still +// holding its scan-time belief that the seat to act has run out of time — folds +// Cara, who had twenty-five seconds left. That is a one-second window that recurs +// on every single turn of every hand. +type TableRef struct { + ID string + Version int64 +} + +// DueTables lists the tables whose clock has run out. +// +// It closes the rows before returning, and it must: the caller is about to take a +// table lock and open a transaction, and holding a *sql.Rows across that means +// holding the only connection in the pool while waiting for it. That is not a +// slow query, it is a deadlock. +func DueTables(now int64) ([]TableRef, error) { + rows, err := Get().Query( + `SELECT id, version FROM game_tables WHERE deadline > 0 AND deadline <= ?`, now) + if err != nil { + return nil, fmt.Errorf("games: due tables: %w", err) + } + defer rows.Close() + + var out []TableRef + for rows.Next() { + var r TableRef + if err := rows.Scan(&r.ID, &r.Version); err != nil { + return nil, fmt.Errorf("games: scan due table: %w", err) + } + out = append(out, r) + } + return out, rows.Err() +} + +// PushDeadlines shoves every live clock out by a grace period. Called once on +// boot: a deploy takes a table's clock with it, and without this the first tick +// after a restart wakes up to find every deadline in the casino already expired +// and auto-folds all of them at once. +func PushDeadlines(grace int64) error { + if _, err := Get().Exec( + `UPDATE game_tables SET deadline = ? WHERE deadline > 0 AND deadline < ?`, + nowUnix()+grace, nowUnix()+grace, + ); err != nil { + return fmt.Errorf("games: push deadlines: %w", err) + } + return nil +} + +// ---- chat ------------------------------------------------------------------ + +// ChatLine is one thing somebody said at the felt. +type ChatLine struct { + ID int64 `json:"id"` + HandNo int64 `json:"hand_no"` + Name string `json:"name"` + Body string `json:"body"` + SaidAt int64 `json:"said_at"` + // Mine is filled in by the web layer, per reader. It is not in the database. + Mine bool `json:"mine,omitempty"` +} + +// MaxChatLen is where a message stops. Long enough for a table read, short enough +// that nobody pastes a novel onto the felt. +const MaxChatLen = 240 + +// Say records a line of chat and returns it. Its hand_no is stamped from the +// table, which is what makes the log answer the only question chat at a money +// table ever really raises: what was said, during which hand. +func Say(tableID, user, name, body string) (ChatLine, error) { + body = strings.TrimSpace(body) + if body == "" { + return ChatLine{}, ErrBadAmount + } + if len(body) > MaxChatLen { + body = body[:MaxChatLen] + } + var handNo int64 + if err := Get().QueryRow(`SELECT hand_no FROM game_tables WHERE id = ?`, tableID).Scan(&handNo); errors.Is(err, sql.ErrNoRows) { + return ChatLine{}, ErrNoSuchTable + } else if err != nil { + return ChatLine{}, fmt.Errorf("games: chat hand no: %w", err) + } + + now := nowUnix() + res, err := Get().Exec( + `INSERT INTO game_chat (table_id, hand_no, matrix_user, name, body, said_at) + VALUES (?, ?, ?, ?, ?, ?)`, + tableID, handNo, user, name, body, now, + ) + if err != nil { + return ChatLine{}, fmt.Errorf("games: say: %w", err) + } + id, _ := res.LastInsertId() + return ChatLine{ID: id, HandNo: handNo, Name: name, Body: body, SaidAt: now}, nil +} + +// Chat reads the last few lines said at a table, oldest first. +func Chat(tableID string, limit int) ([]ChatLine, error) { + if limit <= 0 || limit > 200 { + limit = 50 + } + rows, err := Get().Query( + `SELECT id, hand_no, name, body, said_at FROM game_chat + WHERE table_id = ? ORDER BY id DESC LIMIT ?`, tableID, limit) + if err != nil { + return nil, fmt.Errorf("games: chat: %w", err) + } + defer rows.Close() + + var out []ChatLine + for rows.Next() { + var c ChatLine + if err := rows.Scan(&c.ID, &c.HandNo, &c.Name, &c.Body, &c.SaidAt); err != nil { + return nil, fmt.Errorf("games: scan chat: %w", err) + } + out = append(out, c) + } + if err := rows.Err(); err != nil { + return nil, err + } + // Read newest-first so the LIMIT takes the right end; hand them back in the + // order they were said. + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, nil +} diff --git a/internal/storage/tables_test.go b/internal/storage/tables_test.go new file mode 100644 index 0000000..2478e3a --- /dev/null +++ b/internal/storage/tables_test.go @@ -0,0 +1,271 @@ +package storage + +import ( + "errors" + "testing" +) + +// openTestTable stands up a table with a full ring of bots and returns it. Six +// seats, because that is hold'em's ring and the most seats any game here has. +func openTestTable(t *testing.T, id, game string) Table { + t.Helper() + tbl := Table{ + ID: id, Game: game, Tier: "1-2", State: []byte(`{}`), + Seed1: 1, Seed2: 2, Phase: "betting", HandNo: 1, + } + seats := make([]Seat, 6) + for i := range seats { + seats[i] = Seat{Seat: i, Name: "bot"} + } + if err := OpenTable(tbl, seats); err != nil { + t.Fatal(err) + } + return tbl +} + +// reload reads a table back and fails the test if it is gone. +func reload(t *testing.T, id string) (Table, []Seat) { + t.Helper() + tbl, seats, err := LoadTable(id) + if err != nil { + t.Fatal(err) + } + return tbl, seats +} + +func TestOpenTable_SeatsAreAllBots(t *testing.T) { + setupTestDB(t) + openTestTable(t, "t1", "holdem") + + _, seats := reload(t, "t1") + if len(seats) != 6 { + t.Fatalf("want 6 seats, got %d", len(seats)) + } + for _, s := range seats { + if !s.Bot() { + t.Errorf("seat %d should be a bot", s.Seat) + } + } +} + +func TestSitDown_MovesChipsOntoTheTable(t *testing.T) { + setupTestDB(t) + tbl := openTestTable(t, "t1", "holdem") + fund(t, player, 5000) + + if err := SitDown(Sit{ + Table: tbl, + Seat: Seat{Seat: 2, MatrixUser: player, Name: "reala"}, + BuyIn: 1000, + }); err != nil { + t.Fatal(err) + } + + // The chips are off the stack... + if got := chipsOf(t, player); got != 4000 { + t.Errorf("stack: want 4000, got %d", got) + } + // ...and onto the seat. + _, seats := reload(t, "t1") + seat := seats[2] + if seat.MatrixUser != player || seat.Staked != 1000 { + t.Errorf("seat 2: want reala staked 1000, got %q staked %d", seat.MatrixUser, seat.Staked) + } + // The occupancy claim points at the table. + id, err := TableOf(player) + if err != nil || id != "t1" { + t.Errorf("TableOf: want t1, got %q err %v", id, err) + } +} + +func TestSitDown_CannotTakeATakenSeat(t *testing.T) { + setupTestDB(t) + tbl := openTestTable(t, "t1", "holdem") + fund(t, player, 5000) + fund(t, "@bob:parodia.dev", 5000) + + if err := SitDown(Sit{Table: tbl, Seat: Seat{Seat: 2, MatrixUser: player, Name: "reala"}, BuyIn: 1000}); err != nil { + t.Fatal(err) + } + tbl2, _ := reload(t, "t1") + err := SitDown(Sit{Table: tbl2, Seat: Seat{Seat: 2, MatrixUser: "@bob:parodia.dev", Name: "bob"}, BuyIn: 1000}) + if !errors.Is(err, ErrSeatTaken) { + t.Fatalf("want ErrSeatTaken, got %v", err) + } + // Bob's chips did not move. + if got := chipsOf(t, "@bob:parodia.dev"); got != 5000 { + t.Errorf("bob's stack should be untouched, got %d", got) + } +} + +func TestSitDown_CannotSitAtTwoTables(t *testing.T) { + setupTestDB(t) + tbl1 := openTestTable(t, "t1", "holdem") + tbl2 := openTestTable(t, "t2", "holdem") + fund(t, player, 5000) + + if err := SitDown(Sit{Table: tbl1, Seat: Seat{Seat: 0, MatrixUser: player, Name: "reala"}, BuyIn: 1000}); err != nil { + t.Fatal(err) + } + err := SitDown(Sit{Table: tbl2, Seat: Seat{Seat: 0, MatrixUser: player, Name: "reala"}, BuyIn: 1000}) + if !errors.Is(err, ErrHandInProgress) { + t.Fatalf("want ErrHandInProgress, got %v", err) + } + // The buy-in for the second table rolled back. + if got := chipsOf(t, player); got != 4000 { + t.Errorf("only the first buy-in should have left the stack, got %d", got) + } +} + +func TestSitDown_InsufficientChipsRollsBack(t *testing.T) { + setupTestDB(t) + tbl := openTestTable(t, "t1", "holdem") + fund(t, player, 500) + + err := SitDown(Sit{Table: tbl, Seat: Seat{Seat: 0, MatrixUser: player, Name: "reala"}, BuyIn: 1000}) + if !errors.Is(err, ErrInsufficientChips) { + t.Fatalf("want ErrInsufficientChips, got %v", err) + } + if _, err := TableOf(player); !errors.Is(err, ErrNoLiveHand) { + t.Errorf("no seat should have been claimed, got %v", err) + } + _, seats := reload(t, "t1") + if seats[0].MatrixUser != "" { + t.Errorf("seat should still be a bot") + } +} + +func TestLeaveTable_BringsChipsHome(t *testing.T) { + setupTestDB(t) + tbl := openTestTable(t, "t1", "holdem") + fund(t, player, 5000) + if err := SitDown(Sit{Table: tbl, Seat: Seat{Seat: 0, MatrixUser: player, Name: "reala"}, BuyIn: 1000}); err != nil { + t.Fatal(err) + } + + tbl2, _ := reload(t, "t1") + // Got up with 1,240 — up on the session. + if err := LeaveTable(Leave{Table: tbl2, Seat: 0, MatrixUser: player, Bot: "bot", Amount: 1240}); err != nil { + t.Fatal(err) + } + if got := chipsOf(t, player); got != 5240 { + t.Errorf("want 5240 back on the stack, got %d", got) + } + if _, err := TableOf(player); !errors.Is(err, ErrNoLiveHand) { + t.Errorf("seat claim should be gone, got %v", err) + } + _, seats := reload(t, "t1") + if seats[0].MatrixUser != "" { + t.Errorf("a bot should have taken the empty chair") + } +} + +func TestSaveTable_VersionGuardsTheWrite(t *testing.T) { + setupTestDB(t) + openTestTable(t, "t1", "holdem") + + a, _ := reload(t, "t1") // both read version 0 + b, _ := reload(t, "t1") + + a.State = []byte(`{"a":1}`) + if err := CommitTable(TableCommit{Table: a}); err != nil { + t.Fatalf("first write should win: %v", err) + } + b.State = []byte(`{"b":2}`) + if err := CommitTable(TableCommit{Table: b}); !errors.Is(err, ErrStaleTable) { + t.Fatalf("second write should be stale, got %v", err) + } + + after, _ := reload(t, "t1") + if string(after.State) != `{"a":1}` { + t.Errorf("the winning write should stand, got %s", after.State) + } + if after.Version != 1 { + t.Errorf("version should have bumped once, got %d", after.Version) + } +} + +func TestDueTables_OnlyExpiredClocks(t *testing.T) { + setupTestDB(t) + now := nowUnix() + + past := openTestTable(t, "past", "holdem") + past.Deadline = now - 5 + if err := CommitTable(TableCommit{Table: past}); err != nil { + t.Fatal(err) + } + future := openTestTable(t, "future", "holdem") + future.Deadline = now + 60 + if err := CommitTable(TableCommit{Table: future}); err != nil { + t.Fatal(err) + } + openTestTable(t, "noclock", "holdem") // deadline 0 + + due, err := DueTables(now) + if err != nil { + t.Fatal(err) + } + if len(due) != 1 || due[0].ID != "past" { + t.Fatalf("only the past-due table should show, got %+v", due) + } +} + +func TestChat_KeepsTheHandItWasSaidDuring(t *testing.T) { + setupTestDB(t) + openTestTable(t, "t1", "holdem") // hand_no 1 + + if _, err := Say("t1", player, "reala", "nice hand"); err != nil { + t.Fatal(err) + } + // The table moves to the next hand. + tbl2, _ := reload(t, "t1") + tbl2.HandNo = 2 + if err := CommitTable(TableCommit{Table: tbl2}); err != nil { + t.Fatal(err) + } + if _, err := Say("t1", player, "reala", "and another"); err != nil { + t.Fatal(err) + } + + lines, err := Chat("t1", 50) + if err != nil { + t.Fatal(err) + } + if len(lines) != 2 { + t.Fatalf("want 2 lines, got %d", len(lines)) + } + if lines[0].Body != "nice hand" || lines[0].HandNo != 1 { + t.Errorf("first line should be hand 1: %+v", lines[0]) + } + if lines[1].HandNo != 2 { + t.Errorf("second line should be hand 2: %+v", lines[1]) + } +} + +func TestCloseTable_KeepsATableWithAHumanAtIt(t *testing.T) { + setupTestDB(t) + tbl := openTestTable(t, "t1", "holdem") + fund(t, player, 5000) + if err := SitDown(Sit{Table: tbl, Seat: Seat{Seat: 0, MatrixUser: player, Name: "reala"}, BuyIn: 1000}); err != nil { + t.Fatal(err) + } + + if err := CloseTable("t1"); err != nil { + t.Fatal(err) + } + if _, _, err := LoadTable("t1"); err != nil { + t.Errorf("a table with a human should survive close, got %v", err) + } + + // Everyone leaves; now it goes. + tbl2, _ := reload(t, "t1") + if err := LeaveTable(Leave{Table: tbl2, Seat: 0, MatrixUser: player, Bot: "bot", Amount: 1000}); err != nil { + t.Fatal(err) + } + if err := CloseTable("t1"); err != nil { + t.Fatal(err) + } + if _, _, err := LoadTable("t1"); !errors.Is(err, ErrNoSuchTable) { + t.Errorf("an all-bot table should close, got %v", err) + } +} diff --git a/internal/web/games_hub.go b/internal/web/games_hub.go new file mode 100644 index 0000000..8478d87 --- /dev/null +++ b/internal/web/games_hub.go @@ -0,0 +1,117 @@ +package web + +import ( + "sync" + "sync/atomic" +) + +// The SSE hub: how a move one player makes reaches the phones of everyone else +// at the felt. +// +// It is in-memory and it is intentionally dumb. It holds no game state and makes +// no decisions — it is a fan-out of opaque byte frames, keyed by table id. The +// authority is always the database; a frame is a nudge that says "the table at +// this version changed, come and look", and a subscriber that misses one (a +// dropped send, a reconnect) refetches the table, which is authoritative anyway. +// So a lost frame is a cosmetic hiccup, never a wrong balance. +// +// Two rules hold it together, and both are load-bearing: +// +// 1. **Sends are non-blocking.** A subscriber's channel is buffered, and a send +// that would block is dropped, not waited on. The publish happens under the +// table lock (which is what orders frames correctly for free), so a blocking +// send would hold that lock while one phone on a train stalls — and the turn +// clock behind that lock stalls with it, for the whole casino. A dropped frame +// costs that one subscriber a refetch; a held lock costs everyone the room. +// +// 2. **The publisher never touches the database.** The hub is reached only after +// the DB work is done and the connection released. Holding a *sql.Rows or a tx +// open for the life of a stream would hold the one connection in the pool +// forever, and a single subscriber would brick the whole application. + +// hubFrame is what goes down the wire: an opaque payload the browser knows how to +// read (a JSON table view), tagged with the version it represents so a subscriber +// can tell a frame it already has from one it missed. +type hubFrame struct { + Version int64 + Data []byte +} + +// tableSub is one open EventSource: a buffered channel and the id that lets the +// subscriber unregister itself when the stream closes. +type tableSub struct { + id int64 + ch chan hubFrame +} + +// gamesHub fans table frames out to whoever is watching each table. +type gamesHub struct { + mu sync.Mutex + tables map[string]map[int64]*tableSub + nextID atomic.Int64 +} + +func newGamesHub() *gamesHub { + return &gamesHub{tables: make(map[string]map[int64]*tableSub)} +} + +// subChanBuffer is how many frames a slow subscriber can fall behind before the +// hub starts dropping theirs. A few is plenty: a subscriber that far behind is +// going to refetch the authoritative table anyway, so buffering more just delays +// that with staler frames. +const subChanBuffer = 8 + +// subscribe registers a new watcher of a table and returns its channel plus the +// unsubscribe to defer. The channel is buffered so a publish never blocks on a +// reader that is mid-write to its socket. +func (h *gamesHub) subscribe(tableID string) (<-chan hubFrame, func()) { + sub := &tableSub{id: h.nextID.Add(1), ch: make(chan hubFrame, subChanBuffer)} + + h.mu.Lock() + subs := h.tables[tableID] + if subs == nil { + subs = make(map[int64]*tableSub) + h.tables[tableID] = subs + } + subs[sub.id] = sub + h.mu.Unlock() + + return sub.ch, func() { h.unsubscribe(tableID, sub.id) } +} + +func (h *gamesHub) unsubscribe(tableID string, id int64) { + h.mu.Lock() + defer h.mu.Unlock() + subs := h.tables[tableID] + if subs == nil { + return + } + delete(subs, id) + if len(subs) == 0 { + delete(h.tables, tableID) + } +} + +// publish pushes a frame to everyone watching a table, dropping it for any +// subscriber whose buffer is full rather than waiting on them. See rule 1: this +// is called under the table lock, so it must never block. +func (h *gamesHub) publish(tableID string, f hubFrame) { + h.mu.Lock() + defer h.mu.Unlock() + for _, sub := range h.tables[tableID] { + select { + case sub.ch <- f: + default: + // Full buffer: this subscriber is behind. Dropping is correct — they will + // refetch the authoritative table when they next read a version gap. + } + } +} + +// watchers reports how many streams are open on a table. Used by the caller that +// decides whether a frame is worth rendering at all. +func (h *gamesHub) watchers(tableID string) int { + h.mu.Lock() + defer h.mu.Unlock() + return len(h.tables[tableID]) +} diff --git a/internal/web/games_hub_test.go b/internal/web/games_hub_test.go new file mode 100644 index 0000000..b88ffeb --- /dev/null +++ b/internal/web/games_hub_test.go @@ -0,0 +1,94 @@ +package web + +import ( + "sync" + "testing" +) + +func TestHub_DeliversToSubscribers(t *testing.T) { + h := newGamesHub() + ch, done := h.subscribe("t1") + defer done() + + h.publish("t1", hubFrame{Version: 3, Data: []byte("hi")}) + f := <-ch + if f.Version != 3 || string(f.Data) != "hi" { + t.Fatalf("got %+v", f) + } +} + +func TestHub_OnlyToTheRightTable(t *testing.T) { + h := newGamesHub() + ch1, d1 := h.subscribe("t1") + defer d1() + ch2, d2 := h.subscribe("t2") + defer d2() + + h.publish("t1", hubFrame{Version: 1}) + select { + case <-ch2: + t.Fatal("t2 should not have received t1's frame") + default: + } + if f := <-ch1; f.Version != 1 { + t.Fatalf("t1 got %+v", f) + } +} + +// TestHub_PublishNeverBlocks is the load-bearing property: a subscriber that +// never reads must not be able to hold up a publish, because publish happens +// under the table lock and a blocked publish stalls the turn clock for everyone. +func TestHub_PublishNeverBlocks(t *testing.T) { + h := newGamesHub() + _, done := h.subscribe("t1") // never read from + defer done() + + // Far more than the buffer. If any of these blocked, the test would hang. + blocked := make(chan struct{}) + go func() { + for i := 0; i < subChanBuffer*10; i++ { + h.publish("t1", hubFrame{Version: int64(i)}) + } + close(blocked) + }() + <-blocked +} + +func TestHub_UnsubscribeStopsDelivery(t *testing.T) { + h := newGamesHub() + ch, done := h.subscribe("t1") + done() + + if h.watchers("t1") != 0 { + t.Fatalf("watchers should be 0 after unsubscribe, got %d", h.watchers("t1")) + } + h.publish("t1", hubFrame{Version: 1}) + select { + case _, ok := <-ch: + if ok { + t.Fatal("a frame arrived after unsubscribe") + } + default: + } +} + +func TestHub_ConcurrentSubscribers(t *testing.T) { + h := newGamesHub() + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ch, done := h.subscribe("t1") + defer done() + <-ch + }() + } + // Let them all register, then flood so every one of them reads at least one. + for h.watchers("t1") < 50 { + } + for i := 0; i < subChanBuffer; i++ { + h.publish("t1", hubFrame{Version: int64(i)}) + } + wg.Wait() +} diff --git a/internal/web/games_lock.go b/internal/web/games_lock.go new file mode 100644 index 0000000..e795beb --- /dev/null +++ b/internal/web/games_lock.go @@ -0,0 +1,58 @@ +package web + +import ( + "hash/fnv" + "sync" +) + +// The striped table lock, and why it is only ever an optimisation. +// +// The database's version column is the real concurrency authority: every write +// to a table is conditional on the version the writer read, so two writers that +// race produce one winner and one ErrStaleTable no matter what happens in +// memory. This lock exists purely to make the loser lose *before* it does the +// work, rather than after — it serialises the read-modify-write on a table so the +// common case doesn't burn an engine step and a marshal only to be told it was +// stale. +// +// It is a fixed array hashed on table id, never a map you can delete from, and +// that is deliberate. A map of mutexes keyed by table id, cleaned up when a table +// empties, will hand two goroutines two different mutex objects for the same +// table across a delete-and-recreate — which is no lock at all. A fixed array has +// no lifecycle: the same id always hashes to the same mutex, forever. The only +// cost is that two unrelated tables can collide onto one stripe and briefly wait +// on each other, which is harmless. +// +// A redeploy is the case that proves the version column has to be the authority: +// during a drain two processes are running, each with its own array, so a table +// is "locked" by two mutexes that know nothing about each other. The version +// column is the only thing both processes share, and it is what keeps them +// correct while the mutexes are useless. + +// lockStripes is how many mutexes the array holds. A power of two so the mask is +// clean; large enough that collisions between live tables are rare. +const lockStripes = 256 + +type stripedLocks struct { + m [lockStripes]sync.Mutex +} + +func newStripedLocks() *stripedLocks { return &stripedLocks{} } + +// forTable returns the mutex a given table hashes onto. The same id always +// returns the same mutex. +func (s *stripedLocks) forTable(id string) *sync.Mutex { + h := fnv.New32a() + _, _ = h.Write([]byte(id)) + return &s.m[h.Sum32()&(lockStripes-1)] +} + +// withTable runs fn while holding the table's stripe. The lock is released when +// fn returns — it never spans a network read or an SSE send, only the +// read-modify-write against the database. +func (s *stripedLocks) withTable(id string, fn func() error) error { + mu := s.forTable(id) + mu.Lock() + defer mu.Unlock() + return fn() +} diff --git a/internal/web/server.go b/internal/web/server.go index 532a903..ab52b20 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -76,6 +76,13 @@ type Server struct { metricsMu sync.Mutex saltDay int64 salt [16]byte + + // The shared-table machinery. hub fans SSE frames out to the phones at a felt; + // tableLocks is the striped optimisation over the DB's version column (see + // games_table.go). Both are nil-safe to construct always: they cost nothing + // until a table is opened. + hub *gamesHub + tableLocks *stripedLocks } // New builds the server. Templates are parsed once at startup. Each page gets @@ -136,7 +143,7 @@ func New(cfg config.WebConfig, sources []config.SourceConfig, postingEnabled boo live = append(live, ch) } - s := &Server{cfg: cfg, sources: infos, postingEnabled: postingEnabled, tpls: tpls, adminSubs: adminSubs, adv: adv, advPost: advPost, channels: live} + s := &Server{cfg: cfg, sources: infos, postingEnabled: postingEnabled, tpls: tpls, adminSubs: adminSubs, adv: adv, advPost: advPost, channels: live, hub: newGamesHub(), tableLocks: newStripedLocks()} // Optional OIDC sign-in (Authentik). Discovery is a network call; if the // provider is unreachable at boot we log and serve anonymously rather than