player is ended update

internal/controller/controller.go

@@ -26,6 +26,7 @@ type iRoomService interface {
 	UpdateIsReady(context.Context, *service.UpdateIsReadyParams) (service.UpdateIsReadyResponse, error)
 	UpdateIsMuted(context.Context, *service.UpdateIsMutedParams) (service.UpdateIsMutedResponse, error)
 	ReorderPlaylist(context.Context, *service.ReorderPlaylistParams) (service.ReorderPlaylistResponse, error)
+	EndVideo(context.Context, *service.EndVideoParams) (service.EndVideoResponse, error)
 }
 
 type controller struct {
@@ -44,6 +45,7 @@ func NewController(roomService iRoomService, logger *slog.Logger) *controller {
 		},
 		roomService: roomService,
 		logger:      logger,
+		wsmux:       nil,
 	}
 	c.wsmux = c.getWSRouter()
 

internal/controller/helper.controller.go

@@ -11,9 +11,9 @@ import (
 	"github.com/sharetube/server/internal/service"
 )
 
-type EmptyStruct struct{}
+type EmptyInput struct{}
 
-func (es *EmptyStruct) UnmarshalJSON([]byte) error {
+func (es *EmptyInput) UnmarshalJSON([]byte) error {
 	return nil
 }
 

internal/controller/ws-handler.controller.go

@@ -15,14 +15,13 @@ type Output struct {
 	Payload any    `json:"payload"`
 }
 
-func (c controller) handleAlive(ctx context.Context, conn *websocket.Conn, input EmptyStruct) error {
+func (c controller) handleAlive(_ context.Context, _ *websocket.Conn, _ EmptyInput) error {
 	return nil
 }
 
 type UpdatePlayerStateInput struct {
 	VideoId      int     `json:"video_id"`
 	IsPlaying    bool    `json:"is_playing"`
-	IsEnded      bool    `json:"is_ended"`
 	CurrentTime  int     `json:"current_time"`
 	PlaybackRate float64 `json:"playback_rate"`
 	UpdatedAt    int     `json:"updated_at"`
@@ -36,7 +35,6 @@ func (c controller) handleUpdatePlayerState(ctx context.Context, conn *websocket
 		SenderConn:   conn,
 		VideoId:      input.VideoId,
 		IsPlaying:    input.IsPlaying,
-		IsEnded:      input.IsEnded,
 		CurrentTime:  input.CurrentTime,
 		PlaybackRate: input.PlaybackRate,
 		UpdatedAt:    input.UpdatedAt,
@@ -128,6 +126,39 @@ func (c controller) handleAddVideo(ctx context.Context, _ *websocket.Conn, input
 	return nil
 }
 
+func (c controller) handleEndVideo(ctx context.Context, _ *websocket.Conn, _ EmptyInput) error {
+	roomId := c.getRoomIdFromCtx(ctx)
+	memberId := c.getMemberIdFromCtx(ctx)
+
+	addVideoResponse, err := c.roomService.EndVideo(ctx, &service.EndVideoParams{
+		SenderId: memberId,
+		RoomId:   roomId,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to add video: %w", err)
+	}
+
+	if addVideoResponse.Player != nil {
+		if err := c.broadcastPlayerVideoUpdated(ctx,
+			addVideoResponse.Conns,
+			addVideoResponse.Player,
+			addVideoResponse.Playlist,
+			addVideoResponse.Members,
+		); err != nil {
+			return fmt.Errorf("failed to broadcast player updated: %w", err)
+		}
+	} else {
+		if err := c.broadcast(ctx, addVideoResponse.Conns, &Output{
+			Type:    "VIDEO_ENDED",
+			Payload: nil,
+		}); err != nil {
+			return fmt.Errorf("failed to broadcast video ended: %w", err)
+		}
+	}
+
+	return nil
+}
+
 type RemoveMemberInput struct {
 	MemberId uuid.UUID `json:"member_id"`
 }

internal/controller/ws-router.controller.go

@@ -23,6 +23,7 @@ func (c controller) getWSRouter() *wsrouter.WSRouter {
 	// video
 	wsrouter.Handle(mux, "ALIVE", c.handleAlive)
 	wsrouter.Handle(mux, "ADD_VIDEO", c.handleAddVideo)
+	wsrouter.Handle(mux, "END_VIDEO", c.handleEndVideo)
 	wsrouter.Handle(mux, "REMOVE_VIDEO", c.handleRemoveVideo)
 	wsrouter.Handle(mux, "REORDER_PLAYLIST", c.handleReorderPlaylist)
 

internal/repository/room/error.go

@@ -7,6 +7,7 @@ var (
 	ErrMemberListNotFound      = errors.New("memberlist not found")
 	ErrTokenAlreadyExists      = errors.New("auth token already exists")
 	ErrPlayerNotFound          = errors.New("player not found")
+	ErrIsVideoEndedNotFound    = errors.New("is video ended not found")
 	ErrPlaylistNotFound        = errors.New("playlist not found")
 	ErrVideoNotFound           = errors.New("video not found")
 	ErrLastVideoNotFound       = errors.New("last video not found")

internal/repository/room/player.go

@@ -5,7 +5,6 @@ import "time"
 type Player struct {
 	IsPlaying       bool
 	WaitingForReady bool
-	IsEnded         bool
 	CurrentTime     int
 	PlaybackRate    float64
 	UpdatedAt       int
@@ -14,7 +13,6 @@ type Player struct {
 type SetPlayerParams struct {
 	IsPlaying       bool
 	WaitingForReady bool
-	IsEnded         bool
 	CurrentTime     int
 	PlaybackRate    float64
 	UpdatedAt       int
@@ -25,3 +23,13 @@ type ExpirePlayerParams struct {
 	RoomId   string
 	ExpireAt time.Time
 }
+
+type SetIsVideoEndedParams struct {
+	RoomId       string
+	IsVideoEnded bool
+}
+
+type ExpireIsVideoEndedParams struct {
+	RoomId   string
+	ExpireAt time.Time
+}

internal/repository/room/redis/player.go

@@ -10,7 +10,6 @@ import (
 const (
 	isPlayingKey       = "is_playing"
 	waitingForReadyKey = "waiting_for_ready"
-	isEndedKey         = "is_ended"
 	currentTimeKey     = "current_time"
 	playbackRateKey    = "playback_rate"
 	updatedAtKey       = "updated_at"
@@ -20,14 +19,50 @@ func (r repo) getPlayerKey(roomId string) string {
 	return fmt.Sprintf("room:%s:player", roomId)
 }
 
+func (r repo) getIsVideoEndedKey(roomId string) string {
+	return fmt.Sprintf("room:%s:video-ended", roomId)
+}
+
+func (r repo) SetIsVideoEnded(ctx context.Context, params *room.SetIsVideoEndedParams) error {
+	pipe := r.rc.TxPipeline()
+
+	isVideoEndedKey := r.getIsVideoEndedKey(params.RoomId)
+	pipe.Set(ctx, isVideoEndedKey, params.IsVideoEnded, r.maxExpireDuration)
+	pipe.Expire(ctx, isVideoEndedKey, r.maxExpireDuration)
+
+	return r.executePipe(ctx, pipe)
+}
+
+func (r repo) GetIsVideoEnded(ctx context.Context, roomId string) (bool, error) {
+	isVideoEndedKey := r.getIsVideoEndedKey(roomId)
+	res, err := r.rc.Get(ctx, isVideoEndedKey).Bool()
+	if err != nil {
+		return false, err
+	}
+
+	return res, nil
+}
+
+func (r repo) ExpireIsVideoEnded(ctx context.Context, params *room.ExpireIsVideoEndedParams) error {
+	res, err := r.rc.ExpireAt(ctx, r.getIsVideoEndedKey(params.RoomId), params.ExpireAt).Result()
+	if err != nil {
+		return err
+	}
+
+	if !res {
+		return room.ErrIsVideoEndedNotFound
+	}
+
+	return nil
+}
+
 func (r repo) SetPlayer(ctx context.Context, params *room.SetPlayerParams) error {
 	pipe := r.rc.TxPipeline()
 
 	playerKey := r.getPlayerKey(params.RoomId)
 	pipe.HSet(ctx, playerKey, map[string]any{
 		isPlayingKey:       params.IsPlaying,
 		waitingForReadyKey: params.WaitingForReady,
-		isEndedKey:         params.IsEnded,
 		currentTimeKey:     params.CurrentTime,
 		playbackRateKey:    params.PlaybackRate,
 		updatedAtKey:       params.UpdatedAt,
@@ -63,7 +98,6 @@ func (r repo) GetPlayer(ctx context.Context, roomId string) (room.Player, error)
 	return room.Player{
 		IsPlaying:       r.fieldToBool(playerMap[isPlayingKey]),
 		WaitingForReady: r.fieldToBool(playerMap[waitingForReadyKey]),
-		IsEnded:         r.fieldToBool(playerMap[isEndedKey]),
 		CurrentTime:     r.fieldToInt(playerMap[currentTimeKey]),
 		PlaybackRate:    r.fieldToFload64(playerMap[playbackRateKey]),
 		UpdatedAt:       r.fieldToInt(playerMap[updatedAtKey]),
@@ -127,10 +161,6 @@ func (r repo) UpdatePlayerWaitingForReady(ctx context.Context, roomId string, wa
 	return r.updatePlayerValue(ctx, roomId, waitingForReadyKey, waitingForReady)
 }
 
-func (r repo) UpdatePlayerIsEnded(ctx context.Context, roomId string, isEnded bool) error {
-	return r.updatePlayerValue(ctx, roomId, isEndedKey, isEnded)
-}
-
 func (r repo) UpdatePlayerCurrentTime(ctx context.Context, roomId string, currentTime int) error {
 	return r.updatePlayerValue(ctx, roomId, currentTimeKey, currentTime)
 }