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

package server

import (
	"context"
	"fmt"
	"time"

	"gospeak-live-sfu-demo/gen"
	"gospeak-live-sfu-demo/internal/auth"
	"gospeak-live-sfu-demo/internal/config"
	"gospeak-live-sfu-demo/internal/sfu/cloudflare"
	"gospeak-live-sfu-demo/internal/sfu/srs"
)

// Service 实现 LiveSFU 控制面逻辑：协调房间在各 SFU 后端的发布/订阅拓扑。
type Service struct {
	cfg  *config.Config
	cf   *cloudflare.Provider
	srsP *srs.Provider
	hub  *roomHub
	auth *auth.Manager
}

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) 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())
		}
	}
	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")
	}
	backend := backendName(req.GetBackend())
	if backend == "" {
		return nil, fmt.Errorf("unknown backend")
	}
	identity := req.GetIdentity()
	if identity == "" {
		identity = randomID()
	}
	switch backend {
	case "cloudflare":
		if !s.cf.Configured() {
			return nil, fmt.Errorf("cloudflare realtime not configured (set CF_APP_ID / CF_APP_SECRET)")
		}
		sessionID, err := s.cf.Client().CreateSession(room)
		if err != nil {
			return nil, fmt.Errorf("cf create session: %w", err)
		}
		info, err := s.cf.Client().GetSession(sessionID)
		if err != nil {
			return nil, fmt.Errorf("cf get session: %w", err)
		}
		target := &gen.StreamTarget{
			Backend:     gen.BackendKind_BACKEND_KIND_CLOUDFLARE,
			SessionId:   sessionID,
			PublishedAt: time.Now().Unix(),
		}
		s.hub.setTarget(room, "cloudflare", target)
		return &gen.PublishResponse{
			SessionId:   sessionID,
			IceServers:  iceServersToProto(info.IceServers),
			Target:      target,
		}, nil
	case "srs":
		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)
		return &gen.PublishResponse{
			Stream:       stream,
			PublishToken: token,
			IceServers:   []*gen.IceServer{{Urls: []string{s.cf.Stun()}}},
			Target:       target,
		}, nil
	}
	return nil, fmt.Errorf("unsupported backend")
}

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")
	}
	backend := backendName(req.GetBackend())
	if backend == "" {
		return nil, fmt.Errorf("unknown backend")
	}
	pub := s.findTarget(room, req.GetBackend())
	if pub == nil {
		return nil, fmt.Errorf("room %q not live on %s", room, backend)
	}
	switch backend {
	case "cloudflare":
		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.GetSessionId(),
			IceServers:         iceServersToProto(info.IceServers),
		}, nil
	case "srs":
		return &gen.SubscribeResponse{
			Stream:     pub.GetStream(),
			IceServers: []*gen.IceServer{{Urls: []string{s.cf.Stun()}}},
		}, nil
	}
	return nil, fmt.Errorf("unsupported backend")
}

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()
	backend := backendName(req.GetBackend())
	if room == "" || backend == "" {
		return nil, fmt.Errorf("room and backend required")
	}
	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())
			}
		}
	}
	s.hub.removeTarget(room, backend)
	return &gen.StopStreamResponse{Ok: true}, nil
}

func (s *Service) WatchRoom(req *gen.WatchRoomRequest, stream gen.LiveSFU_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)}); err != nil {
		return err
	}
	ticker := time.NewTicker(20 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-stream.Context().Done():
			return nil
		case msg := <-ch:
			if err := stream.Send(&gen.RoomEvent{Room: room, Targets: decodeTargets(msg)}); err != nil {
				return err
			}
		case <-ticker.C:
			if err := stream.Send(&gen.RoomEvent{Room: room, Targets: s.hub.targets(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")
	}
	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 (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
}