| new file mode 100644 |
| index 0000000..5cc7f8b |
| --- /dev/null |
| +++ b/internal/meet/hub/hub.go |
| @@ -0,0 +1,341 @@ |
| +// 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 = errors.New("hub: room holds sixteen participations") |
| + 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") |
| + ErrStaleGen = errors.New("hub: stale generation") |
| + ErrTooManyWaiting = errors.New("hub: eight already waiting") |
| +) |
| + |
| +type Config struct { |
| + 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) |
| +} |
| + |
| +// 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 |
| + |
| + 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 |
| +} |
| + |
| +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 |
| +} |
| + |
| +type tombstone struct { |
| + 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 |
| +} |
| + |
| +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{}, |
| + } |
| +} |
| + |
| +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 |
| +} |
| + |
| +// 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 { |
| + 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{}, |
| + } |
| + 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}, 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 |
| +} |
| + |
| +// LiveForSession names the peer this session holds in the room, whether |
| +// attached or within its resume window (spec §4.2). |
| +func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + id, ok := h.bySession[sqid][session] |
| + return id, ok |
| +} |
| + |
| +// Detach starts the resume window for a participation whose stream |
| +// closed. Its subscriber is closed; a resume attaches a new one. |
| +func (h *Hub) Detach(sqid string, peer PeerID) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return |
| + } |
| + p, ok := r.peers[peer] |
| + if !ok { |
| + 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) |
| + 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{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 |
| +} |
| + |
| +// Tick runs the fifteen-second housekeeping: unjoined and lapsed |
| +// participations are dropped, tombstones expire. Task 2 adds kind |
| +// re-derivation and promotion. |
| +func (h *Hub) Tick(ctx context.Context) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + 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) |
| + } |
| + } |
| +} |
| + |
| +// 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 } |
| + |
| +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 the one place an event reaches a subscriber; targeted |
| +// messages are also kept in the recipient's replay ring (Task 3). |
| +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) |
| + } |
| +} |
amadan