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])
}