feat: single-owner game state, transactional lifecycle, durable money on server stop

5 files changed, 206 insertions, 26 deletions

diff --git a/internal/net/conn.go b/internal/net/conn.go
index 2fff0bc..6d78d8f 100644
--- a/internal/net/conn.go
+++ b/internal/net/conn.go
@@ -6,12 +6,18 @@ import (
 	"io"
 	"net"
 	"strings"
+	"time"
 )
 
 type Conn interface {
 	ReadMessage() (string, error)
 	io.Writer
 	io.Closer
+	// SetWriteDeadline is optional for transports: sessions that support it
+	// enforce per-write deadlines so a stalled peer surfaces as a timeout
+	// instead of a hung writer goroutine. Implementations without deadline
+	// support return nil (deadlines are best-effort).
+	SetWriteDeadline(t time.Time) error
 	SetEcho(bool) error
 }
 
@@ -70,6 +76,10 @@ func (t *tcpConn) Write(b []byte) (int, error) {
 	return t.conn.Write(b)
 }
 
+func (t *tcpConn) SetWriteDeadline(ts time.Time) error {
+	return t.conn.SetWriteDeadline(ts)
+}
+
 func (t *tcpConn) Close() error {
 	return t.conn.Close()
 }
diff --git a/internal/net/server.go b/internal/net/server.go
index 9b5be18..b0407fc 100644
--- a/internal/net/server.go
+++ b/internal/net/server.go
@@ -1,6 +1,7 @@
 package net
 
 import (
+	"context"
 	"crypto/rand"
 	"crypto/rsa"
 	"crypto/tls"
@@ -73,6 +74,19 @@ type Session struct {
 	promptVisible     bool
 	writeMu           sync.Mutex
 
+	// Outbound queueing. Transport sessions (telnet/TLS/websocket) are put
+	// into queued mode by startWriter before their input goroutine begins:
+	// Write* then enqueues into outbox, and the dedicated writer goroutine
+	// performs the network I/O with a per-message write deadline. A stalled
+	// or slow client therefore blocks at most its own queue, never the
+	// world; an overflowing queue disconnects only that client. Sessions
+	// without a writer (tests, synthetic conns) keep the legacy synchronous
+	// path so message construction is shared and behavior identical.
+	outbox     chan string
+	writerOnce sync.Once
+	closeOnce  sync.Once
+	writeDone  chan struct{}
+
 	// seq is a monotonically increasing sequence number assigned when the
 	// session is added to a Hub. Hub.AllSessions and Hub.PlayersInRoom sort
 	// sessions by seq so per-tick consumers iterate players in a stable,
@@ -97,6 +111,10 @@ type Server struct {
 	httpsTLSConfig *tls.Config
 	hub            *Hub
 	handler        func(*Session, string)
+
+	done         chan struct{}
+	shutdownOnce sync.Once
+	listenErr    error
 }
 
 type Hub struct {
@@ -107,6 +125,64 @@ type Hub struct {
 	onDisconnect func(*Session)
 }
 
+// writeQueueBounds bound the outbound queue: maxQueuedMessages before a
+// client is disconnected for falling behind, writeDeadline per message, and
+// closeFlushTimeout to drain the queue when the session is closing.
+const (
+	maxQueuedMessages = 512
+	writeDeadline     = 5 * time.Second
+	closeFlushTimeout = 2 * time.Second
+)
+
+// startWriter puts the session into queued mode: writes are enqueued and a
+// dedicated goroutine drains the queue to the network with per-message write
+// deadlines. The goroutine closes the transport after the queue drains, so a
+// Close() during shutdown flushes pending output before the socket drops.
+// Must be called before the session's input goroutine starts.
+func (sess *Session) StartWriter() {
+	sess.writerOnce.Do(func() {
+		sess.outbox = make(chan string, maxQueuedMessages)
+		sess.writeDone = make(chan struct{})
+		go func() {
+			defer close(sess.writeDone)
+			for msg := range sess.outbox {
+				if dl, ok := sess.Conn.(interface {
+					SetWriteDeadline(time.Time) error
+				}); ok {
+					_ = dl.SetWriteDeadline(time.Now().Add(writeDeadline))
+				}
+				if _, err := sess.Conn.Write([]byte(msg)); err != nil {
+					log.Printf("session write error: %v", err)
+					// Abandon the queue and drop the transport; the input
+					// goroutine notices the closed connection on its next
+					// read and performs normal removal.
+					sess.Conn.Close()
+					return
+				}
+			}
+			// Queue closed and drained: flush any final bytes and close.
+			if dl, ok := sess.Conn.(interface {
+				SetWriteDeadline(time.Time) error
+			}); ok {
+				_ = dl.SetWriteDeadline(time.Now().Add(closeFlushTimeout))
+			}
+			sess.Conn.Close()
+		}()
+	})
+}
+
+// enqueue appends msg to the outbound queue. When the client's queue is full
+// the session is disconnected instead of blocking the caller: the world must
+// never wait on a single slow client.
+func (sess *Session) enqueue(msg string) {
+	select {
+	case sess.outbox <- msg:
+	default:
+		log.Printf("session write queue overflow; disconnecting slow client")
+		sess.Close()
+	}
+}
+
 func NewHub() *Hub {
 	return &Hub{
 		sessions: make(map[*Session]bool),
@@ -145,24 +221,34 @@ func (h *Hub) Remove(s *Session) {
 		}
 		return
 	}
-	h.hardRemoveLocked(s)
+	callback := h.hardRemoveLocked(s)
 	h.mu.Unlock()
+	// The removal callback runs outside the Hub mutex: it performs saves
+	// and game-state cleanup, and must never serialize behind (or hold up)
+	// Hub membership operations.
+	if callback != nil {
+		callback(s)
+	}
 }
 
 func (h *Hub) HardRemove(s *Session) {
 	h.mu.Lock()
-	defer h.mu.Unlock()
-	h.hardRemoveLocked(s)
+	callback := h.hardRemoveLocked(s)
+	h.mu.Unlock()
+	if callback != nil {
+		callback(s)
+	}
 }
 
-func (h *Hub) hardRemoveLocked(s *Session) {
+// hardRemoveLocked removes the session from all membership structures and
+// returns the removal callback for the caller to invoke AFTER releasing the
+// Hub mutex (nil when none is registered).
+func (h *Hub) hardRemoveLocked(s *Session) func(*Session) {
 	delete(h.sessions, s)
 	for _, room := range h.rooms {
 		delete(room, s)
 	}
-	if h.onRemove != nil {
-		h.onRemove(s)
-	}
+	return h.onRemove
 }
 
 func (h *Hub) EnterRoom(s *Session, roomID int) {
@@ -187,6 +273,28 @@ func (h *Hub) leaveRoomLocked(s *Session) {
 	}
 }
 
+// RekeyRooms remaps Hub room membership buckets per idMap (oldID -> newID).
+// Uses delete-all-then-set-all so a swap (A<->B) cannot clobber an entry
+// mid-rekey. Sessions' own player state is migrated by the game layer; this
+// only moves the membership buckets.
+func (h *Hub) RekeyRooms(idMap map[int]int) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if len(idMap) == 0 {
+		return
+	}
+	moved := make(map[int]map[*Session]bool, len(h.rooms))
+	for id, set := range h.rooms {
+		if newID, ok := idMap[id]; ok && newID != id {
+			moved[newID] = set
+			delete(h.rooms, id)
+		}
+	}
+	for newID, set := range moved {
+		h.rooms[newID] = set
+	}
+}
+
 func (h *Hub) AllSessions() []*Session {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -215,6 +323,7 @@ func NewServer(cfg *config.Config) (*Server, error) {
 	s := &Server{
 		config: cfg,
 		hub:    NewHub(),
+		done:   make(chan struct{}),
 	}
 
 	if cfg.Telnet.Enabled {
@@ -337,7 +446,38 @@ func (s *Server) ListenAndServe(handler func(*Session, string)) error {
 		return fmt.Errorf("no listeners configured")
 	}
 
-	select {}
+	// Block until Shutdown is called (or a fatal listener error occurs).
+	<-s.done
+	return s.listenErr
+}
+
+// Shutdown stops all transports: HTTP(S) servers drain, listeners close, and
+// every session is closed (its outbound queue drains first). ListenAndServe
+// returns after Shutdown completes. It is safe to call more than once.
+func (s *Server) Shutdown() {
+	s.shutdownOnce.Do(func() {
+		if s.httpServer != nil {
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			defer cancel()
+			_ = s.httpServer.Shutdown(ctx)
+		}
+		if s.httpsServer != nil {
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			defer cancel()
+			_ = s.httpsServer.Shutdown(ctx)
+		}
+		if s.telnetLn != nil {
+			s.telnetLn.Close()
+		}
+		if s.telnetTLSLn != nil {
+			s.telnetTLSLn.Close()
+		}
+		for _, sess := range s.hub.AllSessions() {
+			sess.Close()
+		}
+		close(s.done)
+	})
+	<-s.done
 }
 
 func (s *Server) serveTelnet() {
@@ -399,7 +539,7 @@ func (s *Server) handleSession(sess *Session, handler func(*Session, string)) {
 		sess.Close()
 	}()
 
-	sess.Conn.Write([]byte("\033[2J\033[H"))
+	sess.Write("\033[2J\033[H")
 
 	sess.Write(WelcomeBanner())
 	sess.Write("What's your account name? ")
@@ -439,15 +579,26 @@ func (sess *Session) ClearPrompt() {
 	sess.promptVisible = false
 }
 
+// dispatch delivers a fully-formatted message to the transport: enqueued in
+// queued mode (the writer goroutine owns the network), written synchronously
+// in direct mode (tests, synthetic conns). Caller holds writeMu.
+func (sess *Session) dispatch(msg string) {
+	if sess.outbox != nil {
+		sess.enqueue(msg)
+		return
+	}
+	sess.Conn.Write([]byte(msg))
+}
+
 // Write sends raw text with no wrapping. Used for prompts and partial lines.
 func (sess *Session) Write(msg string) {
 	sess.writeMu.Lock()
 	defer sess.writeMu.Unlock()
 	if sess.promptVisible {
-		sess.Conn.Write([]byte("\r\n"))
+		sess.dispatch("\r\n")
 		sess.promptVisible = false
 	}
-	sess.Conn.Write([]byte(msg))
+	sess.dispatch(msg)
 }
 
 func (sess *Session) WriteLine(msg string) {
@@ -455,10 +606,10 @@ func (sess *Session) WriteLine(msg string) {
 	sess.writeMu.Lock()
 	defer sess.writeMu.Unlock()
 	if sess.promptVisible {
-		sess.Conn.Write([]byte("\r\n"))
+		sess.dispatch("\r\n")
 		sess.promptVisible = false
 	}
-	sess.Conn.Write([]byte(msg + "\r\n"))
+	sess.dispatch(msg + "\r\n")
 }
 
 func (sess *Session) WriteLines(lines ...string) {
@@ -466,10 +617,10 @@ func (sess *Session) WriteLines(lines ...string) {
 	defer sess.writeMu.Unlock()
 	for i, l := range lines {
 		if i == 0 && sess.promptVisible {
-			sess.Conn.Write([]byte("\r\n"))
+			sess.dispatch("\r\n")
 			sess.promptVisible = false
 		}
-		sess.Conn.Write([]byte(sess.wrap(l) + "\r\n"))
+		sess.dispatch(sess.wrap(l) + "\r\n")
 	}
 }
 
@@ -478,10 +629,10 @@ func (sess *Session) Writef(format string, args ...interface{}) {
 	sess.writeMu.Lock()
 	defer sess.writeMu.Unlock()
 	if sess.promptVisible {
-		sess.Conn.Write([]byte("\r\n"))
+		sess.dispatch("\r\n")
 		sess.promptVisible = false
 	}
-	sess.Conn.Write([]byte(msg))
+	sess.dispatch(msg)
 }
 
 func (sess *Session) WritePrompt(prompt string) {
@@ -490,19 +641,26 @@ func (sess *Session) WritePrompt(prompt string) {
 	if sess.promptVisible {
 		return
 	}
-	sess.Conn.Write([]byte(prompt))
+	sess.dispatch(prompt)
 	sess.promptVisible = true
 }
 
 func (sess *Session) Reprompt(prompt string) {
 	sess.writeMu.Lock()
 	defer sess.writeMu.Unlock()
-	sess.Conn.Write([]byte(prompt))
+	sess.dispatch(prompt)
 	sess.promptVisible = true
 }
 
+// Close stops the session. In queued mode it closes the queue so the writer
+// goroutine drains pending messages and then closes the transport; in direct
+// mode it closes the transport immediately. Idempotent.
 func (sess *Session) Close() error {
-	sess.writeMu.Lock()
-	defer sess.writeMu.Unlock()
-	return sess.Conn.Close()
+	if sess.outbox != nil {
+		sess.closeOnce.Do(func() { close(sess.outbox) })
+		return nil
+	}
+	var err error
+	sess.closeOnce.Do(func() { err = sess.Conn.Close() })
+	return err
 }
diff --git a/internal/net/server_test.go b/internal/net/server_test.go
index b106c53..c91c701 100644
--- a/internal/net/server_test.go
+++ b/internal/net/server_test.go
@@ -3,6 +3,7 @@ package net
 import (
 	"sync"
 	"testing"
+	"time"
 )
 
 func TestHubConcurrency(t *testing.T) {
@@ -38,10 +39,12 @@ func TestHubConcurrency(t *testing.T) {
 
 type testConn struct{ data []byte }
 
-func (c *testConn) ReadMessage() (string, error) { return "", nil }
-func (c *testConn) Write(b []byte) (int, error)  { c.data = append(c.data, b...); return len(b), nil }
-func (c *testConn) Close() error                 { return nil }
-func (c *testConn) SetEcho(bool) error           { return nil }
+func (c *testConn) ReadMessage() (string, error)     { return "", nil }
+func (c *testConn) Write(b []byte) (int, error)      { c.data = append(c.data, b...); return len(b), nil }
+func (c *testConn) SetWriteDeadline(time.Time) error { return nil }
+
+func (c *testConn) Close() error       { return nil }
+func (c *testConn) SetEcho(bool) error { return nil }
 
 func TestSessionWriteConcurrency(t *testing.T) {
 	conn := &testConn{}
diff --git a/internal/net/web.go b/internal/net/web.go
index e573aba..233928f 100644
--- a/internal/net/web.go
+++ b/internal/net/web.go
@@ -64,6 +64,7 @@ func newHTTPMux(s *Server) *http.ServeMux {
 			Conn:  newWSConn(ws),
 			State: StateAccountName,
 		}
+		sess.StartWriter()
 		s.hub.Add(sess)
 		go s.handleSession(sess, s.handler)
 	})
diff --git a/internal/net/wsconn.go b/internal/net/wsconn.go
index 4990c0b..34663ba 100644
--- a/internal/net/wsconn.go
+++ b/internal/net/wsconn.go
@@ -1,6 +1,8 @@
 package net
 
 import (
+	"time"
+
 	"github.com/gorilla/websocket"
 )
 
@@ -25,6 +27,12 @@ func (w *wsConn) Write(b []byte) (int, error) {
 	return len(b), err
 }
 
+// SetWriteDeadline delegates to the underlying websocket connection, which
+// supports deadlines natively.
+func (w *wsConn) SetWriteDeadline(ts time.Time) error {
+	return w.conn.SetWriteDeadline(ts)
+}
+
 func (w *wsConn) Close() error {
 	return w.conn.Close()
 }