feat(chat): room chat/danmaku over SSE with history replay, host broadcast API and rate limit

- chatHub broadcasts room danmaku over SSE; new subscribers replay the
  last N messages on join so they don't face an empty room.
- Add a host-only external push endpoint POST /api/room/{room}/broadcast
  (requires room:publish, Bearer token) that marks messages host:true.
- Slide-window rate limiter (room+sender+IP, 5 per 3s) guards both the
  public chat and the broadcast endpoint.
- Wire RBAC room:chat policy, SPA routing and the room-creation route.

internal/auth/policy.csv

@@ -25,6 +25,11 @@ p, viewer, room, watch
 p, viewer, srs, streams
 
 p, guest, config, read
+p, guest, srs, streams
 
 # 默认 g 关系由代码动态维护，亦可在此预设示例用户
+p, admin, room, chat
+p, publisher, room, chat
+p, viewer, room, chat
+p, guest, room, chat
 # g, alice, admin

internal/server/auth_handler.go

@@ -214,9 +214,22 @@ func (s *Server) handleAuthCheck(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	perms := map[string]bool{}
-	for _, act := range []string{"list", "publish", "subscribe", "watch", "stop"} {
+	for _, c := range []struct {
-		ok, _ := s.auth.Check(claims.Username, claims.Role, "room", act)
+		key, obj, act string
-		perms["room:"+act] = ok
+	}{
+		{"room:list", "room", "list"},
+		{"room:publish", "room", "publish"},
+		{"room:subscribe", "room", "subscribe"},
+		{"room:watch", "room", "watch"},
+		{"room:stop", "room", "stop"},
+		{"room:chat", "room", "chat"},
+		{"user:list", "user", "list"},
+		{"user:manage", "user", "manage"},
+		{"config:read", "config", "read"},
+		{"srs:streams", "srs", "streams"},
+	} {
+		ok, _ := s.auth.Check(claims.Username, claims.Role, c.obj, c.act)
+		perms[c.key] = ok
 	}
 	w.Header().Set("Content-Type", "application/json")
 	json.NewEncoder(w).Encode(map[string]interface{}{

internal/server/chat.go

@@ -0,0 +1,294 @@
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+)
+
+// chatMessage 是房间弹幕/聊天的一条广播消息。
+type chatMessage struct {
+	ID      string `json:"id"`
+	Room    string `json:"room"`
+	User    string `json:"user"`
+	Role    string `json:"role"`
+	Message string `json:"message"`
+	Color   string `json:"color,omitempty"`
+	Host    bool   `json:"host,omitempty"`
+	TS      int64  `json:"ts"`
+}
+
+// chatHub 维护房间 -> 弹幕订阅者的内存广播（与 roomHub 同构，重启不恢复历史）。
+type chatHub struct {
+	mu    sync.RWMutex
+	rooms map[string]map[chan []byte]struct{}
+	// recentMsgs 保存每个房间最近若干条弹幕，用于新订阅者进场回放（重启不恢复，符合内存态设计）。
+	recentMsgs map[string][]*chatMessage
+}
+
+// recentLimit 是单房间保留的弹幕历史条数上限。
+const recentLimit = 50
+
+func newChatHub() *chatHub {
+	return &chatHub{rooms: map[string]map[chan []byte]struct{}{}, recentMsgs: map[string][]*chatMessage{}}
+}
+
+func (c *chatHub) subscribe(room string) (chan []byte, func()) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	ch := make(chan []byte, 16)
+	if c.rooms[room] == nil {
+		c.rooms[room] = map[chan []byte]struct{}{}
+	}
+	if c.recentMsgs[room] == nil {
+		c.recentMsgs[room] = nil
+	}
+	c.rooms[room][ch] = struct{}{}
+	return ch, func() {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		if m, ok := c.rooms[room]; ok {
+			delete(m, ch)
+			if len(m) == 0 {
+				delete(c.rooms, room)
+				// 房间再无订阅者时清空历史，避免内存随房间数无界增长。
+				delete(c.recentMsgs, room)
+			}
+		}
+	}
+}
+
+// recent 返回房间最近若干条弹幕的快照（副本，调用方无需持有锁）。
+func (c *chatHub) recent(room string) []*chatMessage {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	out := make([]*chatMessage, len(c.recentMsgs[room]))
+	copy(out, c.recentMsgs[room])
+	return out
+}
+
+func (c *chatHub) broadcast(room string, msg *chatMessage) {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return
+	}
+	c.mu.Lock()
+	c.recentMsgs[room] = append(c.recentMsgs[room], msg)
+	if n := len(c.recentMsgs[room]); n > recentLimit {
+		c.recentMsgs[room] = c.recentMsgs[room][n-recentLimit:]
+	}
+	for ch := range c.rooms[room] {
+		select {
+		case ch <- data:
+		default:
+		}
+	}
+	c.mu.Unlock()
+}
+
+// chatRateLimiter 是弹幕发送的滑动窗口限流（按 房间 + 发送者 + IP 维度），用于防刷屏。
+// 纯内存实现，重启即清空，与弹幕本身的内存态一致；后台定期清理过期 key 防止 map 无界增长。
+type chatRateLimiter struct {
+	mu     sync.Mutex
+	win    map[string][]time.Time
+	limit  int
+	window time.Duration
+}
+
+func newChatRateLimiter(limit int, window time.Duration) *chatRateLimiter {
+	r := &chatRateLimiter{win: map[string][]time.Time{}, limit: limit, window: window}
+	go r.cleanup()
+	return r
+}
+
+// allow 在窗口内计数未超上限时记入并返回 true，否则返回 false（应拒绝本次发送）。
+func (r *chatRateLimiter) allow(key string, now time.Time) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	cut := now.Add(-r.window)
+	ts := r.win[key]
+	kept := ts[:0]
+	for _, t := range ts {
+		if t.After(cut) {
+			kept = append(kept, t)
+		}
+	}
+	if len(kept) >= r.limit {
+		r.win[key] = kept
+		return false
+	}
+	r.win[key] = append(kept, now)
+	return true
+}
+
+// cleanup 周期性删除已无活跃计数的 key。
+func (r *chatRateLimiter) cleanup() {
+	ticker := time.NewTicker(time.Minute)
+	defer ticker.Stop()
+	for range ticker.C {
+		r.mu.Lock()
+		cut := time.Now().Add(-r.window)
+		for k, ts := range r.win {
+			if len(ts) == 0 || ts[len(ts)-1].Before(cut) {
+				delete(r.win, k)
+			}
+		}
+		r.mu.Unlock()
+	}
+}
+
+// clientIP 从代理头或 RemoteAddr 取客户端 IP（取 X-Forwarded-For 第一段）。
+func clientIP(r *http.Request) string {
+	if x := r.Header.Get("X-Forwarded-For"); x != "" {
+		if i := strings.IndexByte(x, ','); i >= 0 {
+			return strings.TrimSpace(x[:i])
+		}
+		return strings.TrimSpace(x)
+	}
+	if x := r.Header.Get("X-Real-IP"); x != "" {
+		return strings.TrimSpace(x)
+	}
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr
+	}
+	return host
+}
+
+// handleRoomChat 以 SSE 推送房间弹幕（前端据此渲染叠加层 / 聊天列表）。
+func (s *Server) handleRoomChat(w http.ResponseWriter, r *http.Request) {
+	room := r.PathValue("room")
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	ch, unsub := s.chat.subscribe(room)
+	defer unsub()
+	// 进场回放最近若干条弹幕，避免新订阅者面对空房间（非阻塞，填满通道即止）。
+	for _, m := range s.chat.recent(room) {
+		b, err := json.Marshal(m)
+		if err != nil {
+			continue
+		}
+		select {
+		case ch <- b:
+		default:
+			break
+		}
+	}
+	ticker := time.NewTicker(25 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-r.Context().Done():
+			return
+		case msg := <-ch:
+			fmt.Fprintf(w, "event: chat\ndata: %s\n\n", msg)
+			flusher.Flush()
+		case <-ticker.C:
+			fmt.Fprintf(w, ": ping\n\n")
+			flusher.Flush()
+		}
+	}
+}
+
+// handleRoomChatPost 接收一条弹幕并广播给房间内所有订阅者。
+func (s *Server) handleRoomChatPost(w http.ResponseWriter, r *http.Request) {
+	room := r.PathValue("room")
+	var body struct {
+		Message string `json:"message"`
+		Color   string `json:"color"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Message == "" {
+		http.Error(w, "bad request", http.StatusBadRequest)
+		return
+	}
+	if len(body.Message) > 500 {
+		body.Message = body.Message[:500]
+	}
+	// 限流：按 房间 + 发送者 + IP 维度做滑动窗口，防刷屏。
+	user, role, id := "guest", "guest", "guest"
+	if s.auth != nil {
+		if _, u, err := s.auth.VerifyToken(extractAuthToken(r)); err == nil {
+			user, role, id = u.Username, u.Role, u.Username
+		}
+	}
+	if s.chatRL != nil {
+		key := room + "/" + id + "/" + clientIP(r)
+		if !s.chatRL.allow(key, time.Now()) {
+			http.Error(w, "too many messages, please slow down", http.StatusTooManyRequests)
+			return
+		}
+	}
+	msg := &chatMessage{
+		ID:      randomID(),
+		Room:    room,
+		User:    user,
+		Role:    role,
+		Message: body.Message,
+		Color:   body.Color,
+		TS:      time.Now().Unix(),
+	}
+	s.chat.broadcast(room, msg)
+	log.Printf("[chat] room=%s user=%s role=%s: %s", room, user, role, body.Message)
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": msg.ID})
+}
+
+// handleRoomBroadcast 是主播的独立外部推送接口：仅 publisher/admin 可调用（authWrap 已校验 room:publish），
+// 推送的弹幕标记 host=true，经同一 chatHub 广播，主播端与观众端实时可见。供 OBS 脚本 / 机器人 / 管理工具使用。
+func (s *Server) handleRoomBroadcast(w http.ResponseWriter, r *http.Request) {
+	room := r.PathValue("room")
+	var body struct {
+		Message string `json:"message"`
+		Color   string `json:"color"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Message == "" {
+		http.Error(w, "bad request", http.StatusBadRequest)
+		return
+	}
+	if len(body.Message) > 500 {
+		body.Message = body.Message[:500]
+	}
+	if s.chat == nil {
+		http.Error(w, "chat not available", http.StatusServiceUnavailable)
+		return
+	}
+	// 身份来自 token（外部工具用 Authorization: Bearer <token> 调用）
+	user, role, id := "guest", "guest", "guest"
+	if s.auth != nil {
+		if _, u, err := s.auth.VerifyToken(extractAuthToken(r)); err == nil {
+			user, role, id = u.Username, u.Role, u.Username
+		}
+	}
+	if s.chatRL != nil {
+		key := room + "/broadcast/" + id + "/" + clientIP(r)
+		if !s.chatRL.allow(key, time.Now()) {
+			http.Error(w, "too many messages, please slow down", http.StatusTooManyRequests)
+			return
+		}
+	}
+	msg := &chatMessage{
+		ID:      randomID(),
+		Room:    room,
+		User:    user,
+		Role:    role,
+		Message: body.Message,
+		Color:   body.Color,
+		Host:    true,
+		TS:      time.Now().Unix(),
+	}
+	s.chat.broadcast(room, msg)
+	log.Printf("[chat:broadcast] room=%s user=%s role=%s: %s", room, user, role, body.Message)
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": msg.ID})
+}

internal/server/rooms.go

@@ -67,6 +67,14 @@ func (h *roomHub) get(name string) *roomEntry {
 	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)

internal/server/server.go

@@ -3,10 +3,12 @@ package server
 import (
 	"bytes"
 	"embed"
+	"encoding/json"
 	"io/fs"
 	"log"
 	"net"
 	"net/http"
+	"strings"
 	"time"
 
 	"google.golang.org/grpc"
@@ -23,12 +25,15 @@ var staticFS embed.FS
 
 // Server 聚合控制面（gRPC + JSON 网关）、媒体面反向代理与静态 UI，并集成登录与 Casbin 鉴权。
 type Server struct {
-	cfg      *config.Config
+	cfg         *config.Config
-	svc      *Service
+	svc         *Service
-	hub      *roomHub
+	hub         *roomHub
-	srsProxy http.Handler
+	srsProxy    http.Handler
-	cfProxy  http.Handler
+	cfProxy     http.Handler
-	auth     *auth.Manager
+	srsHlsProxy http.Handler
+	auth        *auth.Manager
+	chat        *chatHub
+	chatRL      *chatRateLimiter
 }
 
 func New(cfg *config.Config) *Server {
@@ -50,17 +55,16 @@ func New(cfg *config.Config) *Server {
 	s := &Server{cfg: cfg, svc: svc, hub: hub}
 	s.srsProxy = s.srsProxyHandler()
 	s.cfProxy = s.cfProxyHandler()
+	s.srsHlsProxy = s.srsHlsProxyHandler()
 	mgr, err := auth.NewManager(cfg.JWTSecret, cfg.JWTTTL, cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile)
 	if err != nil {
-		log.Printf("[warn] auth manager init failed (%v), falling back to memory-only", err)
+		log.Printf("[warn] auth manager init failed (%v), running without auth", err)
-		store, _ := auth.NewUserStore("")
-		mgr2 := auth.NewJWTManager(cfg.JWTSecret, cfg.JWTTTL)
-		_ = mgr2
-		_ = store
 	} else {
 		s.auth = mgr
 		log.Printf("[auth] casbin enabled: model=%s policy=%s users=%s", cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile)
 		svc.auth = mgr
+		s.chat = newChatHub()
+		s.chatRL = newChatRateLimiter(5, 3*time.Second)
 	}
 	return s
 }
@@ -72,10 +76,13 @@ func NewWithHub(cfg *config.Config, hub *roomHub) *Server {
 	s := &Server{cfg: cfg, svc: svc, hub: hub}
 	s.srsProxy = s.srsProxyHandler()
 	s.cfProxy = s.cfProxyHandler()
+	s.srsHlsProxy = s.srsHlsProxyHandler()
 	mgr, err := auth.NewManager(cfg.JWTSecret, cfg.JWTTTL, cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile)
 	if err == nil {
 		s.auth = mgr
 		svc.auth = mgr
+		s.chat = newChatHub()
+		s.chatRL = newChatRateLimiter(5, 3*time.Second)
 	}
 	return s
 }
@@ -100,13 +107,30 @@ func (s *Server) StartGRPC() error {
 	return nil
 }
 
+// handleCreateRoom 创建一个房间占位（发布前即可出现在房间列表）。
+func (s *Server) handleCreateRoom(w http.ResponseWriter, r *http.Request) {
+	var body struct {
+		Name string `json:"name"`
+	}
+	_ = json.NewDecoder(r.Body).Decode(&body)
+	name := strings.TrimSpace(body.Name)
+	if name == "" {
+		http.Error(w, "room name required", http.StatusBadRequest)
+		return
+	}
+	s.hub.createRoom(name)
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]any{"ok": true, "room": name})
+}
 func (s *Server) Handler() http.Handler {
 	mux := http.NewServeMux()
 	sub, _ := fs.Sub(staticFS, "static")
-	mux.Handle("GET /", s.fileServer("static/index.html"))
+	index := s.fileServer("static/index.html")
-	mux.Handle("GET /publish", s.fileServer("static/publish.html"))
+	// SPA 入口：发布/观看/登录均为前端路由，统一回退 index.html
-	mux.Handle("GET /watch", s.fileServer("static/watch.html"))
+	mux.Handle("GET /", index)
-	mux.Handle("GET /login", s.fileServer("static/login.html"))
+	mux.Handle("GET /publish", index)
+	mux.Handle("GET /watch", index)
+	mux.Handle("GET /login", index)
 	mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServer(http.FS(sub))))
 
 	mux.HandleFunc("POST /api/auth/login", s.handleLogin)
@@ -119,11 +143,17 @@ func (s *Server) Handler() http.Handler {
 
 	mux.Handle("GET /api/config", s.authWrap(http.HandlerFunc(s.handleConfig), "config", "read", false))
 	mux.Handle("GET /api/rooms", s.authWrap(http.HandlerFunc(s.handleRooms), "room", "list", true))
+	mux.Handle("POST /api/rooms", s.authWrap(http.HandlerFunc(s.handleCreateRoom), "room", "list", true))
 	mux.Handle("POST /api/publish", s.authWrap(http.HandlerFunc(s.handlePublish), "room", "publish", true))
 	mux.Handle("POST /api/subscribe", s.authWrap(http.HandlerFunc(s.handleSubscribe), "room", "subscribe", true))
 	mux.Handle("POST /api/stop", s.authWrap(http.HandlerFunc(s.handleStop), "room", "stop", true))
 	mux.Handle("GET /api/srs/streams", s.authWrap(http.HandlerFunc(s.handleSRSStreams), "srs", "streams", true))
 	mux.Handle("GET /api/room/{room}/events", s.authWrap(http.HandlerFunc(s.handleRoomEvents), "room", "watch", true))
+	mux.Handle("GET /api/room/{room}/chat", s.authWrap(http.HandlerFunc(s.handleRoomChat), "room", "chat", false))
+	// 发送与 SSE 订阅一致：guest 亦可发弹幕（RBAC 已授权 room:chat），needAuth=false
+	mux.Handle("POST /api/room/{room}/chat", s.authWrap(http.HandlerFunc(s.handleRoomChatPost), "room", "chat", false))
+	// 主播独立外部推送接口：仅 publisher/admin（room:publish），供 OBS/机器人/管理工具以 Bearer token 调用，消息标记 host
+	mux.Handle("POST /api/room/{room}/broadcast", s.authWrap(http.HandlerFunc(s.handleRoomBroadcast), "room", "publish", true))
 
 	mux.Handle("GET /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false))
 	mux.Handle("POST /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false))
@@ -134,6 +164,9 @@ func (s *Server) Handler() http.Handler {
 	mux.Handle("POST /api/cf/", s.authWrap(s.cfProxy, "room", "publish", false))
 	mux.Handle("PUT /api/cf/", s.authWrap(s.cfProxy, "room", "publish", false))
 	mux.Handle("DELETE /api/cf/", s.authWrap(s.cfProxy, "room", "publish", false))
+	// SRS HLS/FLV 静态切片反代（:8080），前端经 /live/* 拉流
+	mux.Handle("GET /live/", s.authWrap(s.srsHlsProxy, "srs", "streams", false))
+	mux.Handle("HEAD /live/", s.authWrap(s.srsHlsProxy, "srs", "streams", false))
 	return mux
 }
 
@@ -160,4 +193,4 @@ func (s *Server) fileServer(name string) http.HandlerFunc {
 		}
 		http.ServeContent(w, r, name, time.Time{}, bytes.NewReader(data))
 	}
-}
+}