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

package server

import (
	"context"
	"fmt"
	"time"

	"sync-live/gen"
	"sync-live/internal/auth"
	"sync-live/internal/config"
	"sync-live/internal/sfu/cdn"
	"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
	cdnP *cdn.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 NewServiceWithCDN(cfg *config.Config, cf *cloudflare.Provider, srsP *srs.Provider, cdnP *cdn.Provider, hub *roomHub) *Service {
	return &Service{cfg: cfg, cf: cf, srsP: srsP, cdnP: cdnP, 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
}