// Package hub is the in-process authority for who is in a room: the
// participations of spec §2.4, the peer identity and limits of §4.2 and
// §4.4, and the resume window of §4.5. It holds no HTTP and no database;
// the handlers wrap it and the store answers standing for it.
//
// The hub is per process, which is fine on one host and would not
// survive a multi-host fleet; that is noted, not solved (spec §4.2).
package hub

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"log/slog"
	"sync"
	"time"
)

const (
	// MaxParticipations is what a room holds in any state: unjoined,
	// waiting, admitted, or held for resume (spec §4.2).
	MaxParticipations = 16
	// MaxWaiting is the knock queue; a ninth waiting join is refused
	// with full (spec §4.4).
	MaxWaiting = 8
	// MaxAdmitted is the mesh cap, checked at every transition into
	// the admitted state and never on a resume (spec §4.3).
	MaxAdmitted = 4
	// MaxStreamsPerPerson counts concurrent streams across all rooms.
	MaxStreamsPerPerson = 4

	// UnjoinedWindow is how long a participation lives without a join.
	UnjoinedWindow = 30 * time.Second
	// ResumeWindow is how long a detached participation keeps its
	// admission waiting for the browser to come back (spec §4.5).
	ResumeWindow = 30 * time.Second
	// TombstoneLife is how long an ended peer id is remembered so a
	// resume can be told why it ended (spec §4.5).
	TombstoneLife = 10 * time.Minute
	// RingSize is the replay ring: the last targeted messages held for
	// one participation, for the resume window only (spec §4.5).
	RingSize = 32
)

var (
	// ErrRoomFull is the sixteen-participation cap of spec §4.2, which a
	// handler answers with 503 before anything is allocated.
	ErrRoomFull = errors.New("hub: room holds sixteen participations")
	// ErrMeshFull is the mesh cap of four (spec §4.3), which is not a
	// failed request: the peer has been sent full and may retry with
	// join, so a handler answers 2xx and lets the event carry the news.
	ErrMeshFull       = errors.New("hub: four already admitted")
	ErrSessionHasPeer = errors.New("hub: this session already has a peer in the room")
	ErrPersonStreams  = errors.New("hub: this person has four streams open")
	ErrUnknownPeer    = errors.New("hub: no such peer")
	ErrNotYours       = errors.New("hub: peer is not bound to this session")
	ErrNotAdmitted    = errors.New("hub: peer is not admitted")
	ErrNotMember      = errors.New("hub: peer is not a member")
	ErrNotWaiting     = errors.New("hub: target is not waiting")
	ErrNotTargeted    = errors.New("hub: not a targeted message type")
	ErrStaleGen       = errors.New("hub: stale generation")
	ErrTooManyWaiting = errors.New("hub: eight already waiting")
)

type Config struct {
	// WaitingRoom is read at join time, outside the lock. Nil preserves the
	// hub’s original behavior: guests wait for admission.
	WaitingRoom func(context.Context, string) (bool, error)
	Now         func() time.Time
	Logger      *slog.Logger
	// StandingFor re-derives a person's kind and whether they still hold
	// entry authority for the room. ok=false means standing none.
	StandingFor func(ctx context.Context, room string, personID int64) (kind Kind, ok bool, err error)
	// Knocked says that someone has begun waiting. It exists because the
	// admission queue lives only here, in memory, and reaches members
	// over their own streams — so a member who is not in the room has
	// never had any way to learn that somebody is at the door.
	//
	// Called on a goroutine of its own, never under the lock: the app's
	// implementation reads the database and talks to Apple, and the hub's
	// mutex is held across every peer in every room. Nil is no
	// notification, which is every test and every deployment without a
	// push key.
	Knocked func(room string, personIDs []int64, name string)
}

// Participation is the hub's record that a peer is in a room. It is the
// authority for everything a peer does after entry (spec §2.4).
type Participation struct {
	Peer     PeerID
	Room     string
	Session  string
	PersonID int64
	Name     string
	Kind     Kind
	Admitted bool
	Joined   bool // has posted join (waiting or admitted)
	JoinedAt time.Time
	OpenedAt time.Time
	// Media is the peer's last report of its own microphone and camera
	// (§15.19), nil until it makes one. It lives and dies with the
	// participation, so a resume keeps it and a fresh join starts unknown.
	Media *MediaState

	sub         *Sub
	resumeUntil time.Time         // zero while a stream is attached
	seqOut      map[PeerID]uint64 // next seq per target (unused by the hub; kept by browsers)
	lastSeen    map[PeerID]uint64
	ring        []Message         // targeted messages delivered to this peer, newest last, cap RingSize
	evicted     map[PeerID]uint64 // highest seq per sender the ring has dropped
	// peerTag and reportToken are spec §15.26's: the random tag this
	// participation's reports are logged under, and the token its client
	// reports with (report.go).
	peerTag     string
	reportToken string
}

type pairKey struct{ a, b PeerID }

func keyFor(x, y PeerID) pairKey {
	if x < y {
		return pairKey{x, y}
	}
	return pairKey{y, x}
}

type room struct {
	sqid  string
	peers map[PeerID]*Participation
	gens  map[pairKey]uint64
	ended bool
	// call is the call instance now running here (report.go): replaced
	// when a participation opens in a room that held nobody.
	call *callInstance
}

// tombstone remembers why a participation ended, for ten minutes, so a
// stream that never received its lost event learns the reason on resume
// (spec §4.5). It carries the room and session it was bound to because
// the reason is account state — removed, standing, expired — and a peer
// id is published in rosters and is not a credential (spec §4.2).
type tombstone struct {
	room    string
	session string
	reason  Reason
	at      time.Time
}

type Hub struct {
	cfg  Config
	boot string

	mu         sync.Mutex
	rooms      map[string]*room
	bySession  map[string]map[string]PeerID // room -> session -> peer
	byPerson   map[int64]int                // open streams per person
	tombstones map[PeerID]tombstone
	reports    map[string]*reportGrant // report token -> grant (report.go)
}

func New(cfg Config) *Hub {
	if cfg.Now == nil {
		cfg.Now = time.Now
	}
	if cfg.Logger == nil {
		cfg.Logger = slog.Default()
	}
	return &Hub{
		cfg: cfg, boot: randomID(),
		rooms: map[string]*room{}, bySession: map[string]map[string]PeerID{},
		byPerson: map[int64]int{}, tombstones: map[PeerID]tombstone{},
		reports: map[string]*reportGrant{},
	}
}

func randomID() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

// Boot is the random id chosen when the process started. Every hello
// carries it so a browser can tell a restart from any other refusal
// (spec §4.2).
func (h *Hub) Boot() string { return h.boot }

func (h *Hub) roomLocked(sqid string) *room {
	r, ok := h.rooms[sqid]
	if !ok {
		r = &room{sqid: sqid, peers: map[PeerID]*Participation{}, gens: map[pairKey]uint64{}}
		h.rooms[sqid] = r
	}
	return r
}

type OpenResult struct {
	Peer     PeerID
	Sub      *Sub
	Existing PeerID
	// ReportToken is the client's authority to send quality and problem
	// reports (spec §15.26), carried in its hello.
	ReportToken string
}

// Open is a fresh stream: a new unjoined participation, subject to the
// room, session and person limits, all checked before allocation.
func (h *Hub) Open(sqid, session string, personID int64, name string, kind Kind) (OpenResult, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	r := h.rooms[sqid] // nil until this room holds someone
	if r != nil && r.ended {
		return OpenResult{}, ErrUnknownPeer
	}
	if existing, ok := h.bySession[sqid][session]; ok {
		// A participation whose resume window has passed is not a
		// conflict: §15.7 refuses its resume the moment the window
		// passes, so answering 409 "open in another tab" until the tick
		// caught up told a browser that had just been refused a resume
		// that it was competing with itself (Codex S2-2). It is ended
		// here, on the same terms the tick would have ended it on —
		// nothing can be listening for the lost event, and the room is
		// owed the departure.
		if p := h.lapsedLocked(r, existing); p != nil {
			wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer)
			delete(h.tombstones, p.Peer)
			h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
		} else {
			return OpenResult{Existing: existing}, ErrSessionHasPeer
		}
	}
	if r != nil && len(r.peers) >= MaxParticipations {
		return OpenResult{}, ErrRoomFull
	}
	if h.byPerson[personID] >= MaxStreamsPerPerson {
		return OpenResult{}, ErrPersonStreams
	}
	r = h.roomLocked(sqid)
	p := &Participation{
		Peer: PeerID(randomID()), Room: sqid, Session: session, PersonID: personID, Name: name, Kind: kind,
		OpenedAt: h.cfg.Now(), sub: newSub(), seqOut: map[PeerID]uint64{}, lastSeen: map[PeerID]uint64{},
	}
	h.enrolReportLocked(r, p) // before p joins r.peers: an empty room starts a new call

	r.peers[p.Peer] = p
	if h.bySession[sqid] == nil {
		h.bySession[sqid] = map[string]PeerID{}
	}
	h.bySession[sqid][session] = p.Peer
	h.byPerson[personID]++
	return OpenResult{Peer: p.Peer, Sub: p.sub, ReportToken: p.reportToken}, nil
}

// Find answers a peer only through its own room and session.
func (h *Hub) Find(sqid, session string, peer PeerID) (*Participation, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	p, ok := h.findLocked(sqid, session, peer)
	return p, ok
}

func (h *Hub) findLocked(sqid, session string, peer PeerID) (*Participation, bool) {
	r, ok := h.rooms[sqid]
	if !ok {
		return nil, false
	}
	p, ok := r.peers[peer]
	if !ok || p.Session != session {
		return nil, false
	}
	return p, true
}

// lapsedLocked is the participation this room holds under peer, when it
// is detached and past its resume window — the state §15.7 has already
// refused a resume in, and which only the tick used to clear. It is nil
// for a live one, and for a peer this room does not hold.
func (h *Hub) lapsedLocked(r *room, peer PeerID) *Participation {
	if r == nil {
		return nil
	}
	p, ok := r.peers[peer]
	if !ok {
		return nil
	}
	if p.sub != nil || p.resumeUntil.IsZero() || !h.cfg.Now().After(p.resumeUntil) {
		return nil
	}
	return p
}

// LiveForSession names the peer this session holds in the room, whether
// attached or within its resume window (spec §4.2).
//
// Past that window it holds nothing: the participation is still in the
// hub until the tick, but a resume of it is already refused (§15.7) and
// a fresh stream now ends it and takes its place (Open), so nothing that
// asks "does this session hold a peer here" may still be told yes.
func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	id, ok := h.bySession[sqid][session]
	if !ok || h.lapsedLocked(h.rooms[sqid], id) != nil {
		return "", false
	}
	return id, true
}

// LiveKindForSession is LiveForSession plus the kind that participation
// currently carries — the kind the last tick or the last request wrote,
// never a fresh standing read, which is the app's to do.
//
// It exists for the page authority of spec §2.5: a live participation
// stands in for standing only where §2.4 says the participation is now
// the authority, and that is an admitted **guest** whose grant lapsed
// (ruling 14). A member who lost the room is dropped by the next thing
// they do, so their participation must not let them keep the page.
func (h *Hub) LiveKindForSession(sqid, session string) (PeerID, Kind, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	id, ok := h.bySession[sqid][session]
	if !ok || h.lapsedLocked(h.rooms[sqid], id) != nil {
		return "", 0, false
	}
	p, ok := h.findLocked(sqid, session, id)
	if !ok {
		return "", 0, false
	}
	return id, p.Kind, true
}

// PeerNameFor answers target's display name exactly when the caller
// (this session, in this room) could go on to remove it: an admitted
// member participation, and target present in the same room and not
// the caller's own. This is Remove's own authority (admission.go),
// read here without mutating anything, so a confirm page can never
// name — or offer to remove — a peer the caller could not actually
// remove.
func (h *Hub) PeerNameFor(sqid, session string, target PeerID) (string, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	actor, ok := h.bySession[sqid][session]
	if !ok {
		return "", false
	}
	a, ok := h.findLocked(sqid, session, actor)
	if !ok || !a.Admitted || a.Kind != KindMember {
		return "", false
	}
	r := h.rooms[sqid]
	tp, ok := r.peers[target]
	if !ok || tp.Peer == a.Peer {
		return "", false
	}
	return tp.Name, true
}

// Detach starts the resume window for a participation whose stream
// closed. Its subscriber is closed; a resume attaches a new one.
//
// sub is the subscriber the caller was reading, and the detach happens
// only while that is still the participation's attachment. A stream
// handler calls this on every exit, including the exit a Resume from a
// second stream caused (lost{replaced}), and that displaced handler
// must not detach the replacement that took its place. A nil sub
// detaches whatever is attached, for a caller that holds no subscriber.
func (h *Hub) Detach(sqid string, peer PeerID, sub *Sub) {
	h.mu.Lock()
	defer h.mu.Unlock()
	r, ok := h.rooms[sqid]
	if !ok {
		return
	}
	p, ok := r.peers[peer]
	if !ok {
		return
	}
	if sub != nil && p.sub != sub {
		return
	}
	if p.sub != nil {
		p.sub.terminate(Message{Type: TypeLost, Reason: ReasonBuffer})
		p.sub = nil
	}
	p.resumeUntil = h.cfg.Now().Add(ResumeWindow)
}

// endLocked removes a participation for a terminal reason, closes its
// stream with lost{reason}, leaves a tombstone, and returns whether it
// was admitted or waiting so the caller can announce the departure.
func (h *Hub) endLocked(r *room, p *Participation, reason Reason) (wasAdmitted, wasWaiting bool) {
	wasAdmitted, wasWaiting = p.Admitted, p.Joined && !p.Admitted
	if p.sub != nil {
		p.sub.terminate(Message{Type: TypeLost, Reason: reason})
		p.sub = nil
	}
	delete(r.peers, p.Peer)
	h.endReportLocked(p)
	if h.bySession[r.sqid][p.Session] == p.Peer {
		delete(h.bySession[r.sqid], p.Session)
	}
	if h.byPerson[p.PersonID] > 0 {
		h.byPerson[p.PersonID]--
	}
	h.tombstones[p.Peer] = tombstone{room: r.sqid, session: p.Session, reason: reason, at: h.cfg.Now()}
	for k := range r.gens {
		if k.a == p.Peer || k.b == p.Peer {
			delete(r.gens, k)
		}
	}
	return
}

// standingCheck is one joined participation whose standing the tick must
// re-read, and the answer it came back with.
type standingCheck struct {
	room     string
	peer     PeerID
	personID int64
	kind     Kind
	ok       bool
}

// Tick runs the fifteen-second housekeeping: unjoined and lapsed
// participations are dropped, tombstones expire, and every joined
// participation's kind is re-derived from standing, applying the
// transition table of spec §2.4.
func (h *Hub) Tick(ctx context.Context) {
	h.mu.Lock()
	now := h.cfg.Now()
	for _, r := range h.rooms {
		for _, p := range r.peers {
			switch {
			case !p.Joined && p.sub != nil && now.Sub(p.OpenedAt) > UnjoinedWindow:
				h.endLocked(r, p, ReasonBuffer) // dropped quietly; the browser sees a closed stream
				delete(h.tombstones, p.Peer)
			case p.sub == nil && !p.resumeUntil.IsZero() && now.After(p.resumeUntil):
				wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer)
				delete(h.tombstones, p.Peer)
				h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
			}
		}
	}
	for id, t := range h.tombstones {
		if now.Sub(t.at) > TombstoneLife {
			delete(h.tombstones, id)
		}
	}
	h.sweepReportsLocked(now)
	var checks []standingCheck
	for _, r := range h.rooms {
		for _, p := range r.peers {
			if p.Joined {
				checks = append(checks, standingCheck{room: r.sqid, peer: p.Peer, personID: p.PersonID})
			}
		}
	}
	h.mu.Unlock()

	// Standing is a database read, so it happens with the lock down.
	answered := make([]standingCheck, 0, len(checks))
	for _, c := range checks {
		kind, ok, err := h.cfg.StandingFor(ctx, c.room, c.personID)
		if err != nil {
			h.cfg.Logger.Warn("hub: standing lookup failed on tick",
				"room", c.room, "peer", c.peer, "error", err)
			continue // this participation waits for the next tick
		}
		c.kind, c.ok = kind, ok
		answered = append(answered, c)
	}

	h.mu.Lock()
	defer h.mu.Unlock()
	for _, c := range answered {
		r, found := h.rooms[c.room]
		if !found {
			continue
		}
		p, found := r.peers[c.peer]
		if !found || !p.Joined {
			continue // it ended while standing was being read
		}
		switch {
		case !c.ok:
			// A member loses the room, and so does a waiting guest whose
			// grant lapsed; an admitted guest is untouched (ruling 14).
			h.standingLostLocked(r, p)
		case p.Admitted:
			h.setKindLocked(r, p, c.kind) // a change refreshes the rosters
		default:
			// Waiting. A promotion to member is admission under the cap,
			// exactly as an admit would be; a refusal leaves them waiting
			// as a member, free to retry with join.
			h.setKindLocked(r, p, c.kind)
			if c.kind == KindMember && !h.admitLocked(r, p) {
				h.sendLocked(p, Message{Type: TypeFull})
			}
		}
	}
}

// announceDepartureLocked tells the room that a participation is gone,
// in the shape its state earns: an admitted peer yields gone to admitted
// peers, a waiting one yields knock-gone to admitted members, an
// unjoined one yields nothing (spec §4.3).
func (h *Hub) announceDepartureLocked(r *room, p *Participation, wasAdmitted, wasWaiting bool) {
	switch {
	case wasAdmitted:
		h.broadcastLocked(r, Message{Type: TypeGone, From: p.Peer}, admittedPeers)
	case wasWaiting:
		h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers)
	}
}

type audience func(*Participation) bool

func admittedPeers(p *Participation) bool   { return p.Admitted }
func admittedMembers(p *Participation) bool { return p.Admitted && p.Kind == KindMember }

// sendLocked is one event to one participation, which is a no-op while
// its stream is detached and waiting to resume.
func (h *Hub) sendLocked(p *Participation, m Message) {
	if p.sub != nil {
		p.sub.send(m)
	}
}

func (h *Hub) broadcastLocked(r *room, m Message, who audience) {
	for _, q := range r.peers {
		if who(q) && q.sub != nil {
			q.sub.send(m)
		}
	}
}

// deliver is one event to one peer found by room and id, for the events
// the hub raises about the room. A targeted message never comes this
// way: Relay holds the recipient's participation already, because it
// must also ring the message (see relay.go).
func (h *Hub) deliver(sqid string, to PeerID, m Message) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.deliverLocked(sqid, to, m)
}

func (h *Hub) deliverLocked(sqid string, to PeerID, m Message) {
	r, ok := h.rooms[sqid]
	if !ok {
		return
	}
	p, ok := r.peers[to]
	if !ok {
		return
	}
	if p.sub != nil {
		p.sub.send(m)
	}
}