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 } // createRoom 确保房间占位存在(发布前即出现在列表)。 func (h *roomHub) createRoom(name string) { h.mu.Lock() if _, ok := h.rooms[name]; !ok { h.rooms[name] = &roomEntry{targets: map[string]*gen.StreamTarget{}, subs: map[chan []byte]struct{}{}} } h.mu.Unlock() } 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: } } }