Pete/internal/web/games_table.go

package web

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"time"

	"pete/internal/storage"
)

// The runtime's public surface: the lobby a player finds a table in, the stream
// that keeps their felt live while other people play on it, and the chat that
// runs along the rail. None of this knows poker from UNO — it is keyed on table
// id and moves opaque frames — which is what lets it serve every shared table.

// ---- the lobby -------------------------------------------------------------

// handleHoldemLobby and handleUnoLobby list the tables of their game with a seat
// going spare. A table with every chair taken is not shown, because a lobby you
// cannot join from is just a list; bots keep every open table populated, so there
// is always something to sit down at.
func (s *Server) handleHoldemLobby(w http.ResponseWriter, r *http.Request) {
	s.lobby(w, r, gameHoldem)
}

func (s *Server) handleUnoLobby(w http.ResponseWriter, r *http.Request) {
	s.lobby(w, r, gameUno)
}

func (s *Server) lobby(w http.ResponseWriter, r *http.Request, game string) {
	if _, ok := s.player(w, r); !ok {
		return
	}
	tables, err := storage.LobbyTables(game, 50)
	if err != nil {
		slog.Error("games: lobby", "game", game, "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	open := make([]storage.TableSummary, 0, len(tables))
	for _, t := range tables {
		if t.Humans < t.Seats {
			open = append(open, t)
		}
	}
	writeJSON(w, map[string]any{"tables": open})
}

// ---- the live stream -------------------------------------------------------

// streamPing is how often the stream sends a comment down an idle connection.
// EventSource reconnects itself, but a proxy will hang up a stream that has gone
// quiet, and a heartbeat well under the usual idle timeout keeps it open.
const streamPing = 25 * time.Second

// handleHoldemStream is the player's live view of their table: a Server-Sent
// Events stream that carries a nudge every time the table changes and every line
// of chat as it is said.
//
// It obeys the one rule that keeps a stream from bricking the app (rule from
// games_hub.go): it touches the database exactly once, at the top, to find which
// table the player is at. After that it only ever reads its channel. Holding a
// query open for the life of a stream would hold the single pooled connection for
// the life of a stream, and one idle subscriber would take the whole site down.
func (s *Server) handleTableStream(w http.ResponseWriter, r *http.Request) {
	user, ok := s.player(w, r)
	if !ok {
		return
	}
	tableID, err := storage.TableOf(user)
	if errors.Is(err, storage.ErrNoLiveHand) {
		writeJSONStatus(w, http.StatusConflict, map[string]string{"error": "you're not at a table"})
		return
	}
	if err != nil {
		slog.Error("games: stream table of", "user", user, "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no") // tell nginx-likes not to buffer us
	w.WriteHeader(http.StatusOK)

	ch, unsubscribe := s.hub.subscribe(tableID)
	defer unsubscribe()

	// Nudge the client to fetch straight away, so a stream that opens after a change
	// already missed does not sit blank until the next one.
	fmt.Fprint(w, "event: sync\ndata: {}\n\n")
	flusher.Flush()

	ping := time.NewTicker(streamPing)
	defer ping.Stop()
	ctx := r.Context()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ping.C:
			fmt.Fprint(w, ": ping\n\n")
			flusher.Flush()
		case f, open := <-ch:
			if !open {
				return
			}
			fmt.Fprintf(w, "data: %s\n\n", f.Data)
			flusher.Flush()
		}
	}
}

// ---- chat ------------------------------------------------------------------

// handleHoldemChat reads the recent rail of a player's table, oldest first, with
// their own lines flagged so the felt can lay them out on the right.
func (s *Server) handleTableChat(w http.ResponseWriter, r *http.Request) {
	user, ok := s.player(w, r)
	if !ok {
		return
	}
	tableID, err := storage.TableOf(user)
	if errors.Is(err, storage.ErrNoLiveHand) {
		writeJSON(w, map[string]any{"chat": []storage.ChatLine{}})
		return
	}
	if err != nil {
		slog.Error("games: chat table of", "user", user, "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	lines, err := storage.Chat(tableID, 50)
	if err != nil {
		slog.Error("games: chat", "user", user, "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	for i := range lines {
		lines[i].Mine = false // filled below; the DB does not know who is asking
	}
	name := s.displayName(r, user)
	for i := range lines {
		lines[i].Mine = lines[i].Name == name
	}
	writeJSON(w, map[string]any{"chat": lines})
}

// handleHoldemSay records a line of chat and fans it to the table. The line is
// stamped with the hand it was said during — the one question chat at a money
// table ever really raises — and pushed to every open stream so it lands on the
// rail in real time.
func (s *Server) handleTableSay(w http.ResponseWriter, r *http.Request) {
	user, ok := s.player(w, r)
	if !ok {
		return
	}
	var req struct {
		Body string `json:"body"`
	}
	if err := decodeJSON(r, &req); err != nil {
		http.Error(w, "bad json", http.StatusBadRequest)
		return
	}
	tableID, err := storage.TableOf(user)
	if errors.Is(err, storage.ErrNoLiveHand) {
		writeJSONStatus(w, http.StatusConflict, map[string]string{"error": "you're not at a table"})
		return
	}
	if err != nil {
		slog.Error("games: say table of", "user", user, "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	name := s.displayName(r, user)
	line, err := storage.Say(tableID, user, name, req.Body)
	if errors.Is(err, storage.ErrBadAmount) {
		writeJSONStatus(w, http.StatusBadRequest, map[string]string{"error": "say something"})
		return
	}
	if err != nil {
		slog.Error("games: say", "user", user, "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	s.publishChat(tableID, line)
	line.Mine = true
	writeJSON(w, line)
}

// publishChat fans a chat line to everyone watching a table. Like publishTable it
// is a non-blocking send after the database work is done — the line is already
// saved, this only delivers it — so a slow subscriber costs itself a missed line
// (which its next chat fetch recovers), never the lock.
func (s *Server) publishChat(tableID string, line storage.ChatLine) {
	if s.hub.watchers(tableID) == 0 {
		return
	}
	data, err := json.Marshal(map[string]any{"type": "chat", "line": line})
	if err != nil {
		return
	}
	s.hub.publish(tableID, hubFrame{Data: data})
}

// ---- the abandoned-table reaper --------------------------------------------

// runTableReaper cashes out tables that everyone has walked away from, on the
// same slow timer as the session reaper. It is the seated-player counterpart to
// that loop: a walked-away poker player's chips are inside a table blob, where
// the session reaper (which reads the game_chips stack) cannot see them, so
// without this they would sit in limbo until the player happened to come back.
func (s *Server) runTableReaper(ctx context.Context) {
	ticker := time.NewTicker(reaperInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.reapAbandonedTables()
		}
	}
}

// reapAbandonedTables finds the tables nobody is coming back to and closes each,
// sending every seated human home with whatever is in front of them.
func (s *Server) reapAbandonedTables() {
	cutoff := time.Now().Unix() - int64(storage.SessionIdleAfter.Seconds())
	refs, err := storage.AbandonedTables(cutoff)
	if err != nil {
		slog.Error("games: abandoned tables", "err", err)
		return
	}
	for _, ref := range refs {
		s.reapTable(ref)
	}
}

// reapTable cashes out and closes one abandoned table, under its lock and only if
// it is still the version the scan saw — so a player who wandered back to the
// felt in the same instant keeps their seat and their chips.
func (s *Server) reapTable(ref storage.TableRef) {
	err := s.tableLocks.withTable(ref.ID, func() error {
		t, seats, err := storage.LoadTable(ref.ID)
		if errors.Is(err, storage.ErrNoSuchTable) {
			return nil
		}
		if err != nil {
			return err
		}
		if t.Version != ref.Version {
			return nil // somebody came back between the scan and here
		}
		game := s.games()[t.Game]
		if game == nil {
			slog.Error("games: reaper over unknown game", "game", t.Game, "table", t.ID)
			return nil
		}
		stacks, err := game.stacks(t.State)
		if err != nil {
			return err
		}
		var humans []storage.ReapSeat
		for _, seat := range seats {
			if seat.MatrixUser == "" {
				continue
			}
			amount := int64(0)
			if seat.Seat >= 0 && seat.Seat < len(stacks) {
				amount = stacks[seat.Seat]
			}
			humans = append(humans, storage.ReapSeat{
				Seat: seat.Seat, MatrixUser: seat.MatrixUser, Amount: amount,
			})
		}
		if err := storage.ReapTable(storage.Reap{TableID: t.ID, Version: t.Version, Humans: humans}); err != nil {
			if errors.Is(err, storage.ErrStaleTable) {
				return nil
			}
			return err
		}
		slog.Info("games: reaped abandoned table", "table", t.ID, "humans", len(humans))
		return nil
	})
	if err != nil {
		slog.Error("games: reap table", "table", ref.ID, "err", err)
	}
}