veza/veza-backend-api/internal/services/live_stream_service.go

package services

import (
	"context"
	"errors"
	"time"

	"github.com/google/uuid"

	"veza-backend-api/internal/models"
	"veza-backend-api/internal/repositories"
)

// LiveStreamService handles live stream business logic
type LiveStreamService struct {
	repo     repositories.LiveStreamRepository
	roomRepo *repositories.RoomRepository
}

// NewLiveStreamService creates a new LiveStreamService
func NewLiveStreamService(repo repositories.LiveStreamRepository, roomRepo *repositories.RoomRepository) *LiveStreamService {
	return &LiveStreamService{repo: repo, roomRepo: roomRepo}
}

// List returns all live streams, optionally filtered by is_live
func (s *LiveStreamService) List(ctx context.Context, isLive *bool) ([]*models.LiveStream, error) {
	return s.repo.List(ctx, isLive)
}

// Get returns a single live stream by ID
func (s *LiveStreamService) Get(ctx context.Context, id uuid.UUID) (*models.LiveStream, error) {
	return s.repo.GetByID(ctx, id)
}

// Create creates a new live stream for a user and a chat room for the stream (GL3-01)
func (s *LiveStreamService) Create(ctx context.Context, userID uuid.UUID, stream *models.LiveStream) (*models.LiveStream, error) {
	if stream.Title == "" {
		return nil, errors.New("title is required")
	}
	stream.UserID = userID
	stream.StreamKey = uuid.New().String()
	if stream.StreamerName == "" {
		stream.StreamerName = "Streamer"
	}
	if err := s.repo.Create(ctx, stream); err != nil {
		return nil, err
	}
	// Create chat room for live stream (stream_id = room_id for JoinConversation)
	if s.roomRepo != nil {
		room := &models.Room{
			ID:        stream.ID,
			Name:      stream.Title,
			Type:      "live_stream",
			IsPrivate: false,
			CreatedBy: userID,
		}
		if err := s.roomRepo.Create(ctx, room); err != nil {
			// Non-fatal: stream created, chat room creation failed (e.g. room already exists)
			_ = err
		}
	}
	return stream, nil
}

// Update updates an existing live stream (with ownership check)
func (s *LiveStreamService) Update(ctx context.Context, id, userID uuid.UUID, stream *models.LiveStream) (*models.LiveStream, error) {
	existing, err := s.repo.GetByID(ctx, id)
	if err != nil {
		return nil, err
	}
	if existing.UserID != userID {
		return nil, errors.New("live stream not found")
	}
	stream.ID = id
	stream.UserID = userID
	if err := s.repo.Update(ctx, stream); err != nil {
		return nil, err
	}
	return stream, nil
}

// Delete deletes a live stream (with ownership check)
func (s *LiveStreamService) Delete(ctx context.Context, id, userID uuid.UUID) error {
	existing, err := s.repo.GetByID(ctx, id)
	if err != nil {
		return err
	}
	if existing.UserID != userID {
		return errors.New("live stream not found")
	}
	return s.repo.Delete(ctx, id)
}

// ListByUser returns all streams for a user (including stream_key)
func (s *LiveStreamService) ListByUser(ctx context.Context, userID uuid.UUID) ([]*models.LiveStream, error) {
	return s.repo.ListByUserID(ctx, userID)
}

// RegenerateStreamKey generates a new stream key for a stream (ownership check)
func (s *LiveStreamService) RegenerateStreamKey(ctx context.Context, id, userID uuid.UUID) (string, error) {
	existing, err := s.repo.GetByID(ctx, id)
	if err != nil {
		return "", err
	}
	if existing.UserID != userID {
		return "", errors.New("stream not found")
	}
	existing.StreamKey = uuid.New().String()
	if err := s.repo.Update(ctx, existing); err != nil {
		return "", err
	}
	return existing.StreamKey, nil
}

// SetIsLive updates is_live, started_at and ended_at for a stream (called by stream server callbacks)
func (s *LiveStreamService) SetIsLive(ctx context.Context, id uuid.UUID, isLive bool) error {
	stream, err := s.repo.GetByID(ctx, id)
	if err != nil {
		return err
	}
	stream.IsLive = isLive
	if isLive {
		now := time.Now()
		stream.StartedAt = &now
		stream.EndedAt = nil
	} else {
		now := time.Now()
		stream.EndedAt = &now
	}
	return s.repo.Update(ctx, stream)
}

// UpdateViewerCount adjusts viewer_count by delta (called by stream server for ListenerJoined/ListenerLeft)
func (s *LiveStreamService) UpdateViewerCount(ctx context.Context, id uuid.UUID, delta int) error {
	stream, err := s.repo.GetByID(ctx, id)
	if err != nil {
		return err
	}
	newCount := stream.ViewerCount + delta
	if newCount < 0 {
		newCount = 0
	}
	stream.ViewerCount = newCount
	return s.repo.Update(ctx, stream)
}

// GetByStreamKey returns a stream by its stream_key (for RTMP callback validation)
func (s *LiveStreamService) GetByStreamKey(ctx context.Context, streamKey string) (*models.LiveStream, error) {
	return s.repo.GetByStreamKey(ctx, streamKey)
}

// UpdateStreamURL sets the HLS stream URL (called by Nginx-RTMP on_publish callback)
func (s *LiveStreamService) UpdateStreamURL(ctx context.Context, id uuid.UUID, streamURL string) error {
	stream, err := s.repo.GetByID(ctx, id)
	if err != nil {
		return err
	}
	stream.StreamURL = streamURL
	return s.repo.Update(ctx, stream)
}