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

package server

import (
	"bytes"
	"embed"
	"io/fs"
	"log"
	"net"
	"net/http"
	"time"

	"google.golang.org/grpc"
	"gospeak-live-sfu-demo/gen"
	"gospeak-live-sfu-demo/internal/auth"
	"gospeak-live-sfu-demo/internal/config"
	"gospeak-live-sfu-demo/internal/db"
	"gospeak-live-sfu-demo/internal/sfu/cloudflare"
	"gospeak-live-sfu-demo/internal/sfu/srs"
)

//go:embed static
var staticFS embed.FS

// Server 聚合控制面（gRPC + JSON 网关）、媒体面反向代理与静态 UI，并集成登录与 Casbin 鉴权。
type Server struct {
	cfg      *config.Config
	svc      *Service
	hub      *roomHub
	srsProxy http.Handler
	cfProxy  http.Handler
	auth     *auth.Manager
}

func New(cfg *config.Config) *Server {
	var hub *roomHub
	if cfg.DatabaseURL != "" {
		if dbConn, err := db.Open(cfg.DatabaseURL); err != nil {
			log.Printf("[warn] turso db open failed (%v), falling back to memory hub: %s", err, cfg.DatabaseURL)
			hub = newRoomHub()
		} else {
			hub = newRoomHubWithDB(dbConn)
			log.Printf("[db] turso enabled: dsn=%s remote=%v", cfg.DSN(), cfg.IsRemoteTurso())
		}
	} else {
		hub = newRoomHub()
	}
	cf := cloudflare.NewProvider(cfg.CFAppID, cfg.CFAppSecret, cfg.CFBaseURL, cfg.CFStunURL)
	srsP := srs.NewProvider(cfg.SRSBaseURL, cfg.SRSApp, cfg.SRSSecret, cfg.SRSCandidate)
	svc := NewService(cfg, cf, srsP, hub)
	s := &Server{cfg: cfg, svc: svc, hub: hub}
	s.srsProxy = s.srsProxyHandler()
	s.cfProxy = s.cfProxyHandler()
	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)
		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
	}
	return s
}

func NewWithHub(cfg *config.Config, hub *roomHub) *Server {
	cf := cloudflare.NewProvider(cfg.CFAppID, cfg.CFAppSecret, cfg.CFBaseURL, cfg.CFStunURL)
	srsP := srs.NewProvider(cfg.SRSBaseURL, cfg.SRSApp, cfg.SRSSecret, cfg.SRSCandidate)
	svc := NewService(cfg, cf, srsP, hub)
	s := &Server{cfg: cfg, svc: svc, hub: hub}
	s.srsProxy = s.srsProxyHandler()
	s.cfProxy = s.cfProxyHandler()
	mgr, err := auth.NewManager(cfg.JWTSecret, cfg.JWTTTL, cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile)
	if err == nil {
		s.auth = mgr
		svc.auth = mgr
	}
	return s
}

func (s *Server) Auth() *auth.Manager { return s.auth }

func (s *Server) StartGRPC() error {
	lis, err := net.Listen("tcp", ":"+s.cfg.GRPCPort)
	if err != nil {
		return err
	}
	var opts []grpc.ServerOption
	if s.auth != nil {
		opts = append(opts,
			grpc.UnaryInterceptor(s.auth.UnaryAuthInterceptor()),
			grpc.StreamInterceptor(s.auth.StreamAuthInterceptor()),
		)
	}
	gs := grpc.NewServer(opts...)
	gen.RegisterLiveSFUServer(gs, NewGRPCServer(s.svc))
	go func() { _ = gs.Serve(lis) }()
	return nil
}

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	sub, _ := fs.Sub(staticFS, "static")
	mux.Handle("GET /", s.fileServer("static/index.html"))
	mux.Handle("GET /publish", s.fileServer("static/publish.html"))
	mux.Handle("GET /watch", s.fileServer("static/watch.html"))
	mux.Handle("GET /login", s.fileServer("static/login.html"))
	mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServer(http.FS(sub))))

	mux.HandleFunc("POST /api/auth/login", s.handleLogin)
	mux.HandleFunc("POST /api/auth/register", s.handleRegister)
	mux.HandleFunc("POST /api/auth/logout", s.handleLogout)
	mux.Handle("GET /api/auth/me", s.authWrap(http.HandlerFunc(s.handleMe), "user", "list", true))
	mux.Handle("GET /api/auth/users", s.authWrap(http.HandlerFunc(s.handleListUsers), "user", "list", true))
	mux.Handle("POST /api/auth/users/role", s.authWrap(http.HandlerFunc(s.handleUpdateRole), "user", "manage", true))
	mux.Handle("GET /api/auth/check", s.authWrap(http.HandlerFunc(s.handleAuthCheck), "config", "read", false))

	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/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 /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false))
	mux.Handle("POST /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false))
	mux.Handle("PUT /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false))
	mux.Handle("DELETE /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false))

	mux.Handle("GET /api/cf/", s.authWrap(s.cfProxy, "room", "watch", false))
	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))
	return mux
}

func (s *Server) authWrap(next http.Handler, obj, act string, needAuth bool) http.Handler {
	if s.auth == nil {
		return next
	}
	return s.auth.AuthorizeMiddleware(obj, act, needAuth)(next)
}

func (s *Server) Close() error {
	if s.hub != nil && s.hub.DB() != nil {
		return s.hub.DB().Close()
	}
	return nil
}

func (s *Server) fileServer(name string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		data, err := staticFS.ReadFile(name)
		if err != nil {
			http.Error(w, "not found", http.StatusNotFound)
			return
		}
		http.ServeContent(w, r, name, time.Time{}, bytes.NewReader(data))
	}
}