package server import ( "context" "fmt" "time" "sync-live/gen" "sync-live/internal/auth" "sync-live/internal/config" "sync-live/internal/sfu/cloudflare" "sync-live/internal/sfu/srs" ) // Service 实现 SyncLive 控制面逻辑:协调房间在各 SFU 后端的发布/订阅拓扑。 type Service struct { cfg *config.Config cf *cloudflare.Provider srsP *srs.Provider hub *roomHub auth *auth.Manager distributionManager DistributionManagerIface } // DistributionManagerIface abstracts distribution fanout for testability type DistributionManagerIface interface { Ensure(ctx context.Context, room, stream string) error } func (s *Service) SetDistributionManager(m DistributionManagerIface) { s.distributionManager = m } func NewService(cfg *config.Config, cf *cloudflare.Provider, srsP *srs.Provider, hub *roomHub) *Service { return &Service{cfg: cfg, cf: cf, srsP: srsP, hub: hub} } func (s *Service) SetAuth(a *auth.Manager) { s.auth = a } func (s *Service) checkAuth(ctx context.Context, obj, act string) error { if s.auth == nil { return nil } return s.auth.CheckGRPC(ctx, obj, act) } func backendName(k gen.BackendKind) string { switch k { case gen.BackendKind_BACKEND_KIND_CLOUDFLARE: return "cloudflare" case gen.BackendKind_BACKEND_KIND_SRS: return "srs" default: return "" } } func (s *Service) Backends() []*gen.BackendInfo { resp, _ := s.GetConfig(context.Background(), &gen.GetConfigRequest{}) if resp != nil { return resp.Backends } return nil } func (s *Service) GetConfig(ctx context.Context, req *gen.GetConfigRequest) (*gen.GetConfigResponse, error) { resp := &gen.GetConfigResponse{ Candidate: s.srsP.Candidate(), TokenRequired: s.cfg.TokenRequired, } for _, name := range s.cfg.ProviderList() { switch name { case "cloudflare": resp.Backends = append(resp.Backends, s.cf.BackendInfo()) case "srs": resp.Backends = append(resp.Backends, s.srsP.BackendInfo()) } } // SRS is the trunk resp.Trunk = s.srsP.BackendInfo() resp.Trunk.Primary = true // Build distributors list (capability matrix) for _, d := range s.cfg.DistributorList() { switch d { case "srs-hls": resp.Distributors = append(resp.Distributors, &gen.DistributorInfo{ Kind: gen.DistributionKind_DISTRIBUTION_KIND_SRS_HLS, Name: "SRS HLS", Configured: true, Enabled: true, }) case "cf": resp.Distributors = append(resp.Distributors, &gen.DistributorInfo{ Kind: gen.DistributionKind_DISTRIBUTION_KIND_CF_SFU, Name: "Cloudflare SFU", Configured: s.cf.Configured(), Enabled: s.cf.Configured(), }) case "cdn": resp.Distributors = append(resp.Distributors, &gen.DistributorInfo{ Kind: gen.DistributionKind_DISTRIBUTION_KIND_CDN, Name: s.cfg.CDNName, Configured: s.cfg.CDNEnabled(), Enabled: s.cfg.CDNEnabled(), }) } } return resp, nil } func (s *Service) ListRooms(ctx context.Context, req *gen.ListRoomsRequest) (*gen.ListRoomsResponse, error) { if err := s.checkAuth(ctx, "room", "list"); err != nil { return nil, err } return &gen.ListRoomsResponse{Rooms: s.hub.list()}, nil } func (s *Service) Publish(ctx context.Context, req *gen.PublishRequest) (*gen.PublishResponse, error) { if err := s.checkAuth(ctx, "room", "publish"); err != nil { return nil, err } room := req.GetRoom() if room == "" { return nil, fmt.Errorf("room required") } identity := req.GetIdentity() if identity == "" { identity = randomID() } // SRS is the only trunk: always publish to SRS regardless of requested backend stream := "live-" + room token, _ := signStreamToken(s.cfg.TokenSecret, stream, identity, "publish", 2*time.Hour) target := &gen.StreamTarget{ Backend: gen.BackendKind_BACKEND_KIND_SRS, Stream: stream, PublishToken: token, Url: fmt.Sprintf("/rtc/v1/whep/?app=%s&stream=%s", s.srsP.App(), stream), PublishedAt: time.Now().Unix(), } s.hub.setTarget(room, "srs", target) if s.distributionManager != nil { _ = s.distributionManager.Ensure(ctx, room, stream) } return &gen.PublishResponse{ Stream: stream, PublishToken: token, IceServers: []*gen.IceServer{{Urls: []string{s.cf.Stun()}}}, Target: target, }, nil } func (s *Service) Subscribe(ctx context.Context, req *gen.SubscribeRequest) (*gen.SubscribeResponse, error) { if err := s.checkAuth(ctx, "room", "subscribe"); err != nil { return nil, err } room := req.GetRoom() if room == "" { return nil, fmt.Errorf("room required") } // Always read from SRS trunk; backend param is kept for compatibility pub := s.findTarget(room, gen.BackendKind_BACKEND_KIND_SRS) if pub == nil { return nil, fmt.Errorf("room %q not live", room) } // If caller explicitly requested Cloudflare backend, still need viewer session via CF // Check requested backend to decide response shape reqBackend := req.GetBackend() if reqBackend == gen.BackendKind_BACKEND_KIND_CLOUDFLARE { if !s.cf.Configured() { return nil, fmt.Errorf("cloudflare realtime not configured") } // For CF distribution, viewer subscribes to publisher session // Publisher session is not trunk session; use trunk stream as identity viewerSession, err := s.cf.Client().CreateSession(room) if err != nil { return nil, fmt.Errorf("cf create viewer session: %w", err) } info, err := s.cf.Client().GetSession(viewerSession) if err != nil { return nil, fmt.Errorf("cf get viewer session: %w", err) } return &gen.SubscribeResponse{ SessionId: viewerSession, PublisherSessionId: pub.GetStream(), // trunk stream as publisher id for CF relay IceServers: iceServersToProto(info.IceServers), }, nil } return &gen.SubscribeResponse{ Stream: pub.GetStream(), IceServers: []*gen.IceServer{{Urls: []string{s.cf.Stun()}}}, }, nil } func (s *Service) StopStream(ctx context.Context, req *gen.StopStreamRequest) (*gen.StopStreamResponse, error) { if err := s.checkAuth(ctx, "room", "stop"); err != nil { return nil, err } room := req.GetRoom() if room == "" { return nil, fmt.Errorf("room and backend required") } // SRS trunk is the only target to remove; also clean CF sessions if any backend := backendName(req.GetBackend()) if backend == "" { backend = "srs" } if backend == "cloudflare" { for _, t := range s.hub.targets(room) { if t.GetBackend() == gen.BackendKind_BACKEND_KIND_CLOUDFLARE && t.GetSessionId() != "" { _ = s.cf.Client().DeleteSession(t.GetSessionId()) } } } // Always remove SRS trunk if present s.hub.removeTarget(room, "srs") if backend != "srs" { s.hub.removeTarget(room, backend) } // Also clear distributions s.hub.clearDistributions(room) return &gen.StopStreamResponse{Ok: true}, nil } func (s *Service) WatchRoom(req *gen.WatchRoomRequest, stream gen.SyncLive_WatchRoomServer) error { if err := s.checkAuth(stream.Context(), "room", "watch"); err != nil { return err } room := req.GetRoom() ch, unsub := s.hub.subscribe(room) defer unsub() if err := stream.Send(&gen.RoomEvent{Room: room, Targets: s.hub.targets(room), Distributions: s.hub.distributions(room)}); err != nil { return err } ticker := time.NewTicker(20 * time.Second) defer ticker.Stop() for { select { case <-stream.Context().Done(): return nil case msg := <-ch: targets, dists := decodeRoomEvent(msg) if targets == nil { targets = s.hub.targets(room) } if dists == nil { dists = s.hub.distributions(room) } if err := stream.Send(&gen.RoomEvent{Room: room, Targets: targets, Distributions: dists}); err != nil { return err } case <-ticker.C: if err := stream.Send(&gen.RoomEvent{Room: room, Targets: s.hub.targets(room), Distributions: s.hub.distributions(room)}); err != nil { return err } } } } func (s *Service) Login(ctx context.Context, req *gen.LoginRequest) (*gen.LoginResponse, error) { if s.auth == nil { return nil, fmt.Errorf("auth not configured") } if req.GetUsername() == "" || req.GetPassword() == "" { return nil, fmt.Errorf("username and password required") } token, user, err := s.auth.Login(req.GetUsername(), req.GetPassword()) if err != nil { return nil, fmt.Errorf("login failed: %w", err) } return &gen.LoginResponse{ Token: token, User: &gen.UserInfo{Username: user.Username, Role: user.Role, CreatedAt: user.CreatedAt}, ExpiresAt: time.Now().Add(s.auth.JWT.TTL()).Unix(), }, nil } func (s *Service) Register(ctx context.Context, req *gen.RegisterRequest) (*gen.RegisterResponse, error) { if s.auth == nil { return nil, fmt.Errorf("auth not configured") } if !s.cfg.AllowRegister { return nil, fmt.Errorf("registration disabled") } if req.GetUsername() == "" || req.GetPassword() == "" { return nil, fmt.Errorf("username and password required") } role := req.GetRole() if role == "" { role = auth.RoleViewer } if role == auth.RoleAdmin || role == auth.RolePublisher { u, _ := auth.FromContext(ctx) if u == nil || u.Role != auth.RoleAdmin { role = auth.RoleViewer } } token, user, err := s.auth.Register(req.GetUsername(), req.GetPassword(), role) if err != nil { return nil, fmt.Errorf("register failed: %w", err) } return &gen.RegisterResponse{ Token: token, User: &gen.UserInfo{Username: user.Username, Role: user.Role, CreatedAt: user.CreatedAt}, ExpiresAt: time.Now().Add(s.auth.JWT.TTL()).Unix(), }, nil } func (s *Service) GetMe(ctx context.Context, req *gen.GetMeRequest) (*gen.GetMeResponse, error) { if s.auth == nil { return nil, fmt.Errorf("auth not configured") } u, ok := auth.FromContext(ctx) if !ok || u.Username == "" { return nil, fmt.Errorf("unauthenticated") } if u.Token != "" { if claims, user, err := s.auth.VerifyToken(u.Token); err == nil { return &gen.GetMeResponse{ User: &gen.UserInfo{Username: user.Username, Role: user.Role, CreatedAt: user.CreatedAt}, ExpiresAt: claims.ExpiresAt.Unix(), IssuedAt: claims.IssuedAt.Unix(), }, nil } } return &gen.GetMeResponse{ User: &gen.UserInfo{Username: u.Username, Role: u.Role}, }, nil } func (s *Service) ListUsers(ctx context.Context, req *gen.ListUsersRequest) (*gen.ListUsersResponse, error) { if s.auth == nil { return nil, fmt.Errorf("auth not configured") } if err := s.checkAuth(ctx, "user", "list"); err != nil { return nil, err } users := s.auth.Store.List() out := make([]*gen.UserInfo, 0, len(users)) for _, u := range users { out = append(out, &gen.UserInfo{Username: u.Username, Role: u.Role, CreatedAt: u.CreatedAt}) } return &gen.ListUsersResponse{Users: out}, nil } func (s *Service) UpdateUserRole(ctx context.Context, req *gen.UpdateUserRoleRequest) (*gen.UpdateUserRoleResponse, error) { if s.auth == nil { return nil, fmt.Errorf("auth not configured") } if err := s.checkAuth(ctx, "user", "manage"); err != nil { return nil, err } if req.GetUsername() == "" || req.GetRole() == "" { return nil, fmt.Errorf("username and role required") } actor := "" if u, ok := authContextUser(ctx); ok { actor = u.Username } if actor != "" { if err := s.auth.UpdateUserRoleAs(actor, req.GetUsername(), req.GetRole()); err != nil { return nil, err } } else { if err := s.auth.UpdateUserRole(req.GetUsername(), req.GetRole()); err != nil { return nil, err } } u, _ := s.auth.Store.Get(req.GetUsername()) return &gen.UpdateUserRoleResponse{ Ok: true, User: &gen.UserInfo{Username: u.Username, Role: u.Role, CreatedAt: u.CreatedAt}, }, nil } func authContextUser(ctx context.Context) (*auth.AuthedUser, bool) { // helper to extract actor from gRPC/HTTP context (dual auth) if md, ok := authFromContext(ctx); ok { return md, true } return nil, false } func authFromContext(ctx context.Context) (*auth.AuthedUser, bool) { // try auth package context if u, ok := ctx.Value(auth.ContextUserKey).(*auth.AuthedUser); ok { return u, true } return nil, false } func (s *Service) findTarget(room string, kind gen.BackendKind) *gen.StreamTarget { for _, t := range s.hub.targets(room) { if t.GetBackend() == kind { return t } } return nil } func iceServersToProto(in []cloudflare.IceServer) []*gen.IceServer { out := make([]*gen.IceServer, 0, len(in)) for _, s := range in { out = append(out, &gen.IceServer{Urls: s.URLs, Username: s.Username, Credential: s.Credential}) } return out }