live-sfu-demo/internal/server/rooms.go

package server

import (
	"context"
	"database/sql"
	"encoding/json"
	"log"
	"sync"
	"time"

	"gospeak-live-sfu-demo/gen"
	"gospeak-live-sfu-demo/internal/db"
)

// roomHub 维护房间 -> 各后端分发目标（StreamTarget）的内存状态，
// 并向订阅者广播拓扑变化（实现「分流」状态实时可见）。
// 默认由 Turso 嵌入式 (libSQL file) 持久化。
type roomHub struct {
	mu    sync.RWMutex
	rooms map[string]*roomEntry
	db    *sql.DB
}

type roomEntry struct {
	targets map[string]*gen.StreamTarget // key: 后端名（cloudflare / srs）
	subs    map[chan []byte]struct{}
}

func newRoomHub() *roomHub {
	return &roomHub{rooms: map[string]*roomEntry{}}
}

// newRoomHubWithDB 创建带 Turso 嵌入式持久化的 hub，启动时自动从 DB 加载全量房间。
func newRoomHubWithDB(database *sql.DB) *roomHub {
	h := &roomHub{rooms: map[string]*roomEntry{}, db: database}
	if database != nil {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		all, err := db.LoadAll(ctx, database)
		if err != nil {
			log.Printf("[hub] load from turso embedded failed: %v", err)
		} else {
			for room, backends := range all {
				e := &roomEntry{targets: map[string]*gen.StreamTarget{}, subs: map[chan []byte]struct{}{}}
				for backend, t := range backends {
					e.targets[backend] = t
				}
				h.rooms[room] = e
			}
			if len(all) > 0 {
				log.Printf("[hub] loaded %d rooms from turso embedded", len(all))
			}
		}
	}
	return h
}

// DB 返回底层 *sql.DB，供优雅关闭。
func (h *roomHub) DB() *sql.DB { return h.db }

func (h *roomHub) get(name string) *roomEntry {
	e, ok := h.rooms[name]
	if !ok {
		e = &roomEntry{targets: map[string]*gen.StreamTarget{}, subs: map[chan []byte]struct{}{}}
		h.rooms[name] = e
	}
	return e
}

func (h *roomHub) setTarget(room, backend string, t *gen.StreamTarget) {
	h.mu.Lock()
	e := h.get(room)
	e.targets[backend] = t
	h.broadcast(room, e)
	h.mu.Unlock()

	if h.db != nil {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		if err := db.SaveTarget(ctx, h.db, room, backend, t); err != nil {
			log.Printf("[hub] save target %s/%s to turso embedded: %v", room, backend, err)
		}
	}
}

func (h *roomHub) removeTarget(room, backend string) {
	h.mu.Lock()
	e, ok := h.rooms[room]
	if !ok {
		h.mu.Unlock()
		return
	}
	delete(e.targets, backend)
	if len(e.targets) == 0 && len(e.subs) == 0 {
		delete(h.rooms, room)
		h.mu.Unlock()
		if h.db != nil {
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			if err := db.DeleteTarget(ctx, h.db, room, backend); err != nil {
				log.Printf("[hub] delete target %s/%s: %v", room, backend, err)
			}
		}
		return
	}
	h.broadcast(room, e)
	h.mu.Unlock()

	if h.db != nil {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		if err := db.DeleteTarget(ctx, h.db, room, backend); err != nil {
			log.Printf("[hub] delete target %s/%s: %v", room, backend, err)
		}
	}
}

func (h *roomHub) targets(room string) []*gen.StreamTarget {
	h.mu.RLock()
	defer h.mu.RUnlock()
	e, ok := h.rooms[room]
	if !ok {
		return nil
	}
	out := make([]*gen.StreamTarget, 0, len(e.targets))
	for _, t := range e.targets {
		out = append(out, t)
	}
	return out
}

func (h *roomHub) list() []*gen.Room {
	h.mu.RLock()
	defer h.mu.RUnlock()
	out := make([]*gen.Room, 0, len(h.rooms))
	for name, e := range h.rooms {
		room := &gen.Room{Name: name}
		for _, t := range e.targets {
			room.Targets = append(room.Targets, t)
		}
		out = append(out, room)
	}
	return out
}

func (h *roomHub) subscribe(room string) (chan []byte, func()) {
	h.mu.Lock()
	defer h.mu.Unlock()
	e := h.get(room)
	ch := make(chan []byte, 8)
	e.subs[ch] = struct{}{}
	return ch, func() {
		h.mu.Lock()
		defer h.mu.Unlock()
		if e2, ok := h.rooms[room]; ok {
			delete(e2.subs, ch)
		}
	}
}

func (h *roomHub) broadcast(room string, e *roomEntry) {
	payload, _ := json.Marshal(map[string]interface{}{
		"room":    room,
		"targets": e.targets,
	})
	for ch := range e.subs {
		select {
		case ch <- payload:
		default:
		}
	}
}