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

package server

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

	"sync-live/gen"
	"sync-live/internal/db"
)

// roomHub 维护房间 -> 各后端分发目标（StreamTarget）的内存状态，
// 并向订阅者广播拓扑变化（实现「分流」状态实时可见）。
// 默认由嵌入式 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）
	distributions map[string]*gen.DistributionTarget // key: distribution kind (srs-hls/cf/cdn)
	subs          map[chan []byte]struct{}
}

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

// newRoomHubWithDB 创建带嵌入式 libSQL 持久化的 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 embedded db failed: %v", err)
		} else {
			for room, backends := range all {
				e := &roomEntry{targets: map[string]*gen.StreamTarget{}, distributions: map[string]*gen.DistributionTarget{}, 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 embedded db", 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{}, distributions: map[string]*gen.DistributionTarget{}, 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{}, distributions: map[string]*gen.DistributionTarget{}, 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 embedded db: %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)
		}
		for _, d := range e.distributions {
			room.Distributions = append(room.Distributions, d)
		}
		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,
		"distributions": e.distributions,
	})
	for ch := range e.subs {
		select {
		case ch <- payload:
		default:
		}
	}
}

// broadcastEvent 向房间订阅者广播一个带类型的通用事件（playlist / playback / presence 等）。
// 复用与拓扑相同的订阅通道，前端在同一个 SSE 连接上按 event type 分发处理。
func (h *roomHub) broadcastEvent(room, eventType string, payload interface{}) {
	h.mu.RLock()
	e, ok := h.rooms[room]
	if !ok {
		h.mu.RUnlock()
		return
	}
	envelope, _ := json.Marshal(map[string]interface{}{
		"type":    eventType,
		"room":    room,
		"payload": payload,
	})
	for ch := range e.subs {
		select {
		case ch <- envelope:
		default:
		}
	}
	h.mu.RUnlock()
}

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

func (h *roomHub) setDistribution(room string, d *gen.DistributionTarget) {
	h.mu.Lock()
	e := h.get(room)
	if e.distributions == nil {
		e.distributions = map[string]*gen.DistributionTarget{}
	}
	key := d.GetKind().String()
	e.distributions[key] = d
	h.broadcast(room, e)
	h.mu.Unlock()
}

func (h *roomHub) clearDistributions(room string) {
	h.mu.Lock()
	if e, ok := h.rooms[room]; ok {
		e.distributions = map[string]*gen.DistributionTarget{}
		h.broadcast(room, e)
	}
	h.mu.Unlock()
}

func (h *roomHub) removeRoom(name string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	delete(h.rooms, name)
}