Maven/internal/voice/session.go

// voice/session.go — the live-client registry.
//
// One session per connected client. The registry is the seam the voicesink
// asks ("play this on the most-recently-active client") and the seam the
// server mutates ("client X is active at T"). The voicesink doesn't know
// the conn — this package hides it; voicesink holds *Sessions, calls
// PushToMostRecent, gets nil-id back when no one's around, and surfaces
// that as "no live voice channel" (delivery falls back to away channels if
// the routing table insists, drops if it doesn't).
//
// Concurrency: one mutex around the map; per-session last-active updates
// go through the same lock as add/remove. Adds are rare (one per client
// connect); updates are once per request (inbound PushToTalk refreshes
// last-active). The lock is held briefly — no audio bytes flow through
// sessions; the sink ships audio through the per-conn writeFrame call,
// which is short.
package voice

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

// Session — one connected client. The server writes to it via pushAudio;
// the registry reads lastActive via LastActive; the conn is closed when
// the client disconnects (serveConn returns) or the server shuts down.
type Session struct {
	ID         uint64
	Surface    Surface
	RemoteAddr string

	// lastActive — last time we heard from this client (request frame OR
	// Pong). PickRecent selects the max-lastActive session for routing.
	lastActive atomic.Int64 // unix nano

	mu     sync.Mutex
	conn   net.Conn
	closed bool
}

func (s *Session) setLastActive(t time.Time) {
	s.lastActive.Store(t.UnixNano())
}

// LastActive — int64 unixnano, atomic read. PickRecent compares these.
func (s *Session) LastActive() int64 { return s.lastActive.Load() }

// IsClosed reports whether the conn has been torn down. PickRecent skips
// closed sessions explicitly; an in-flight close race is fine — pushAudio
// returns an error and the sink reroutes (the same path as "no session").
func (s *Session) IsClosed() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.closed
}

// pushAudio writes one Push frame on this session's conn. Acquires the
// session lock so a concurrent close doesn't race with a write (the
// voicesink + the proactive-nudge pusher are goroutines separate from
// serveConn). Returns an error if the conn is closed.
func (s *Session) pushAudio(p AudioNudgePush) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed || s.conn == nil {
		return fmt.Errorf("voice: session %d closed: %w", s.ID, ErrNoSession)
	}
	params, err := json.Marshal(p)
	if err != nil {
		return fmt.Errorf("voice: marshal push params: %w", err)
	}
	return writeFrame(s.conn, Push{Kind: PushKindAudioNudge, Params: params})
}

func (s *Session) shutdown() {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return
	}
	s.closed = true
	if s.conn != nil {
		_ = s.conn.Close()
	}
}

// Sessions — the registry of live clients. Held by the Server AND by
// voicesink (same pointer; the daemon passes one to both). Mutex around the
// map; per-session conn writes use the session's own lock.
type Sessions struct {
	mu     sync.Mutex
	sess   map[uint64]*Session
	nextID uint64
}

func NewSessions() *Sessions {
	return &Sessions{sess: make(map[uint64]*Session)}
}

// Add registers a new client conn. Returns the Session (caller uses it to
// push; serveConn reads the next request from conn). Surface defaults to
// SurfacePCClient today (the reference client's surface; mTLS / passkey
// enrollment will populate this from the auth handshake instead). Removes
// the session on serveConn's return via Remove.
func (r *Sessions) Add(c net.Conn, surface Surface) *Session {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.nextID++
	s := &Session{
		ID:         r.nextID,
		Surface:    surface,
		RemoteAddr: c.RemoteAddr().String(),
		conn:       c,
	}
	s.setLastActive(time.Now())
	r.sess[s.ID] = s
	return s
}

// Remove drops a session (serveConn returned — client closed or read err).
func (r *Sessions) Remove(id uint64) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if s, ok := r.sess[id]; ok {
		s.shutdown()
		delete(r.sess, id)
	}
}

// Touch refreshes a session's lastActive to now (called on each request).
// No-op if the session is gone (a stale request after disconnect).
func (r *Sessions) Touch(id uint64, now time.Time) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if s, ok := r.sess[id]; ok {
		s.setLastActive(now)
	}
}

// Active returns the count of live sessions. For logging / readiness.
func (r *Sessions) Active() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.sess)
}

// PushToMostRecent — the voicesink's primary call. Delivers p to the
// session with the maximum lastActive; today the only Push kind. Returns
// ErrNoSession (wrapped) when no live session exists — the dispatcher /
// voicesink surfaces that as "voice channel not available, route away."
//
// We DO NOT broadcast; the spec's "most-recently-active client plays it"
// is enforced here. If the user has the phone and the laptop open, the
// one they used most recently plays; the other doesn't double-ring.
func (r *Sessions) PushToMostRecent(ctx context.Context, p AudioNudgePush) error {
	r.mu.Lock()
	var best *Session
	for _, s := range r.sess {
		if best == nil || s.LastActive() > best.LastActive() {
			best = s
		}
	}
	r.mu.Unlock()
	if best == nil {
		return fmt.Errorf("voice: push: %w", ErrNoSession)
	}
	return best.pushAudio(p)
}