diff --git a/internal/api/handlers/watch_together.go b/internal/api/handlers/watch_together.go index 607d591d..69486025 100644 --- a/internal/api/handlers/watch_together.go +++ b/internal/api/handlers/watch_together.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "sync" + "sync/atomic" "time" "github.com/Silo-Server/silo-server/internal/access" @@ -110,9 +111,16 @@ type watchTogetherPingMessage struct { ClientSentAt string `json:"client_sent_at"` } +// watchTogetherRoomConn serializes every write to the underlying gorilla +// connection. gorilla/websocket does not support concurrent writers, and room +// broadcasts arrive from other members' goroutines, so all writes — including +// pong/error replies from the read loop — must go through this wrapper. type watchTogetherRoomConn struct { conn *websocket.Conn writeMu sync.Mutex + // pingSentAtNano is the send time of the most recent protocol-level ping, + // used to measure round-trip latency when the pong arrives. + pingSentAtNano atomic.Int64 } func (c *watchTogetherRoomConn) WriteJSON(v any) error { @@ -130,9 +138,35 @@ func (c *watchTogetherRoomConn) Close() error { func (c *watchTogetherRoomConn) WritePing() error { c.writeMu.Lock() defer c.writeMu.Unlock() + c.pingSentAtNano.Store(time.Now().UnixNano()) return writeWebSocketControl(c.conn, websocket.PingMessage, nil) } +// TakePingSentAt returns and clears the send time of the last unanswered +// protocol-level ping. +func (c *watchTogetherRoomConn) TakePingSentAt() time.Time { + nano := c.pingSentAtNano.Swap(0) + if nano == 0 { + return time.Time{} + } + return time.Unix(0, nano) +} + +func (c *watchTogetherRoomConn) WriteError(code, message string) { + _ = c.WriteJSON(map[string]string{ + "type": "error", + "code": code, + "message": message, + }) +} + +func (c *watchTogetherRoomConn) writeRoomClosed(reason string) { + _ = c.WriteJSON(map[string]string{ + "type": "room_closed", + "reason": reason, + }) +} + func NewWatchTogetherHandler( service *watchtogether.Service, scopeResolver WatchTogetherScopeResolver, @@ -653,19 +687,33 @@ func (h *WatchTogetherHandler) HandleRoomWebSocket(w http.ResponseWriter, r *htt reg, snapshot, err := h.Service.Connect(ctx, roomID, claims.UserID, profileID, realtimeConn) if err != nil { + // Terminal failures use room_closed so clients stop reconnecting + // instead of retrying a room that will never come back. if errors.Is(err, watchtogether.ErrRoomNotFound) { - writeWebSocketError(conn, "not_found", "Room not found") + realtimeConn.writeRoomClosed("not_found") } else if errors.Is(err, watchtogether.ErrRoomClosed) { - writeWebSocketError(conn, "gone", "Room is no longer active") + realtimeConn.writeRoomClosed("ended") } else { - writeWebSocketError(conn, "internal_error", "Failed to connect room socket") + realtimeConn.WriteError("internal_error", "Failed to connect room socket") } return } defer h.Service.Disconnect(reg, false) configureWebSocket(conn) + // Measure round-trip latency from protocol-level ping/pong on the server + // clock; client-reported timestamps are subject to clock skew and cannot + // be trusted for command scheduling. + conn.SetPongHandler(func(string) error { + _ = conn.SetReadDeadline(time.Now().Add(wsPingInterval + wsPongTimeout)) + if sentAt := realtimeConn.TakePingSentAt(); !sentAt.IsZero() { + _ = h.Service.HandlePingForConnection(ctx, reg, claims.UserID, profileID, time.Since(sentAt).Milliseconds()) + } + return nil + }) startWebSocketPingLoop(ctx, realtimeConn.WritePing) + // Prime an RTT sample right away instead of waiting for the first tick. + _ = realtimeConn.WritePing() if err := realtimeConn.WriteJSON(map[string]any{ "type": "snapshot", @@ -680,15 +728,15 @@ func (h *WatchTogetherHandler) HandleRoomWebSocket(w http.ResponseWriter, r *htt return } - if err := h.handleRoomClientMessage(ctx, conn, reg, claims.UserID, profileID, data); err != nil { - writeWebSocketError(conn, "bad_request", err.Error()) + if err := h.handleRoomClientMessage(ctx, realtimeConn, reg, claims.UserID, profileID, data); err != nil { + realtimeConn.WriteError("bad_request", err.Error()) } } } func (h *WatchTogetherHandler) handleRoomClientMessage( ctx context.Context, - conn *websocket.Conn, + rc *watchTogetherRoomConn, reg *watchtogether.Registration, userID int, profileID string, @@ -768,16 +816,11 @@ func (h *WatchTogetherHandler) handleRoomClientMessage( if err := json.Unmarshal(data, &msg); err != nil { return err } - clientSentAt, err := time.Parse(time.RFC3339Nano, msg.ClientSentAt) - if err != nil { - return errors.New("client_sent_at must be RFC3339Nano") - } + // The echoed timestamps only serve the client's clock-offset estimate; + // latency for command scheduling is measured server-side from + // protocol-level ping/pong. now := time.Now().UTC() - pingMS := now.Sub(clientSentAt.UTC()).Milliseconds() - if pingMS > 0 { - _ = h.Service.HandlePingForConnection(ctx, reg, userID, profileID, pingMS) - } - return writeWebSocketJSON(conn, map[string]string{ + return rc.WriteJSON(map[string]string{ "type": "pong", "client_sent_at": msg.ClientSentAt, "server_received_at": now.Format(time.RFC3339Nano), diff --git a/internal/api/router.go b/internal/api/router.go index 792ea911..c45a839c 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -816,9 +816,9 @@ func NewRouter(deps Dependencies) chi.Router { watchtogether.NewRepository(deps.DB), deps.SessionMgr, deps.FileRepo, - playbackHandler.CommandDispatcher, watchtogether.NewCatalogSelectionResolver(detailSvc), watchtogether.NewSuggestionRepository(deps.DB), + watchtogether.NewProfileNameResolver(deps.UserStoreProvider), ), viewerResolver, roomTokenService, diff --git a/internal/watchtogether/profile_names.go b/internal/watchtogether/profile_names.go new file mode 100644 index 00000000..0b7d098d --- /dev/null +++ b/internal/watchtogether/profile_names.go @@ -0,0 +1,42 @@ +package watchtogether + +import ( + "context" + "strings" + + "github.com/Silo-Server/silo-server/internal/userstore" +) + +const fallbackMemberName = "Guest" + +// ProfileNameResolver resolves a profile's display name for room member lists. +type ProfileNameResolver interface { + ProfileDisplayName(ctx context.Context, userID int, profileID string) string +} + +type userStoreProfileNames struct { + provider userstore.UserStoreProvider +} + +// NewProfileNameResolver adapts a UserStoreProvider into a ProfileNameResolver. +func NewProfileNameResolver(provider userstore.UserStoreProvider) ProfileNameResolver { + if provider == nil { + return nil + } + return &userStoreProfileNames{provider: provider} +} + +func (r *userStoreProfileNames) ProfileDisplayName(ctx context.Context, userID int, profileID string) string { + if r == nil || r.provider == nil { + return fallbackMemberName + } + store, err := r.provider.ForUser(ctx, userID) + if err != nil || store == nil { + return fallbackMemberName + } + profile, err := store.GetProfile(ctx, profileID) + if err != nil || profile == nil || strings.TrimSpace(profile.Name) == "" { + return fallbackMemberName + } + return profile.Name +} diff --git a/internal/watchtogether/repository.go b/internal/watchtogether/repository.go index f260da38..d10f6e32 100644 --- a/internal/watchtogether/repository.go +++ b/internal/watchtogether/repository.go @@ -152,6 +152,37 @@ func (r *Repository) UpdateAnchor( ) } +// ListIdleRoomIDs returns rooms that are still open but have seen no +// playback-anchor activity since the cutoff, so the janitor can close them. +func (r *Repository) ListIdleRoomIDs(ctx context.Context, cutoff time.Time, limit int) ([]string, error) { + if r == nil || r.pool == nil { + return nil, fmt.Errorf("watch together repository unavailable") + } + + const query = ` + SELECT id FROM watch_together_rooms + WHERE phase <> 'ended' + AND GREATEST(anchor_updated_at, created_at) < $1 + LIMIT $2 + ` + + rows, err := r.pool.Query(ctx, query, cutoff.UTC(), limit) + if err != nil { + return nil, fmt.Errorf("list idle watch together rooms: %w", err) + } + defer rows.Close() + + var roomIDs []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan idle watch together room id: %w", err) + } + roomIDs = append(roomIDs, id) + } + return roomIDs, rows.Err() +} + func (r *Repository) CloseRoom(ctx context.Context, roomID string, closedAt time.Time) (*Room, error) { const query = ` UPDATE watch_together_rooms diff --git a/internal/watchtogether/service.go b/internal/watchtogether/service.go index df95b88f..f1253cb7 100644 --- a/internal/watchtogether/service.go +++ b/internal/watchtogether/service.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "math" + "sort" "strings" "sync" "time" @@ -32,6 +33,21 @@ var ( const ( defaultTransportLead = 500 * time.Millisecond minTransportLead = 350 * time.Millisecond + // maxTransportLead bounds how far in the future transport commands may be + // scheduled, so a single member with a huge (or bogus) measured latency + // cannot stall the whole room. + maxTransportLead = 5 * time.Second + // maxBufferingAnchorDriftSeconds bounds how far a buffering member's + // reported position may move the shared room anchor. + maxBufferingAnchorDriftSeconds = 5.0 + // waitingResumeDeadline is how long a room stays in the waiting state + // before stragglers are skipped and playback resumes for everyone ready. + waitingResumeDeadline = 30 * time.Second + // roomIdleTTL is how long a room may go without any playback-anchor + // activity before the janitor closes it. + roomIdleTTL = 24 * time.Hour + // janitorInterval is how often idle rooms are swept. + janitorInterval = 10 * time.Minute ) type RoomConnection interface { @@ -57,6 +73,7 @@ type RoomStore interface { expectedGeneration int64, ) (*Room, error) CloseRoom(ctx context.Context, roomID string, closedAt time.Time) (*Room, error) + ListIdleRoomIDs(ctx context.Context, cutoff time.Time, limit int) ([]string, error) UpdateSelection( ctx context.Context, roomID string, @@ -81,14 +98,6 @@ type MediaFileLookup interface { GetByID(ctx context.Context, id int) (*models.MediaFile, error) } -type RoomCommandDispatcher interface { - DispatchToSession( - command playback.CommandEnvelope, - deadline time.Duration, - fallback func(), - ) playback.CommandDispatchResult -} - type WatchTogetherSelectionResolver interface { ResolveSelection(ctx context.Context, userID int, profileID string, input SelectItemInput) (*ResolvedSelection, error) } @@ -102,18 +111,27 @@ type Registration struct { type memberState struct { userID int profileID string + displayName string sessionID string connection RoomConnection isReady bool isBuffering bool - ignoreWait bool - lastPingMS int64 + // ignoreWait excludes a member from room-wide readiness barriers. It is + // set when the member fails to become ready before waitingResumeDeadline + // and cleared once they attach or report ready again. + ignoreWait bool + lastPingMS int64 } type liveRoom struct { room Room members map[string]*memberState hostCloseTimer *time.Timer + waitingTimer *time.Timer + // waitingEpoch identifies the current waiting period; deadline callbacks + // carry the epoch they were armed for so a stale timer cannot act on a + // newer waiting period. + waitingEpoch int64 } type snapshotDispatch struct { @@ -132,11 +150,13 @@ type Service struct { suggestions SuggestionStore sessions RoomSessionLookup files MediaFileLookup - dispatcher RoomCommandDispatcher selectionResolver WatchTogetherSelectionResolver + profileNames ProfileNameResolver hostDisconnectTTL time.Duration now func() time.Time + janitorStop chan struct{} + mu sync.Mutex rooms map[string]*liveRoom } @@ -145,22 +165,37 @@ func NewService( repo RoomStore, sessions RoomSessionLookup, files MediaFileLookup, - dispatcher RoomCommandDispatcher, selectionResolver WatchTogetherSelectionResolver, suggestions SuggestionStore, + profileNames ProfileNameResolver, ) *Service { - return &Service{ + s := &Service{ repo: repo, suggestions: suggestions, sessions: sessions, files: files, - dispatcher: dispatcher, selectionResolver: selectionResolver, + profileNames: profileNames, hostDisconnectTTL: 15 * time.Second, now: func() time.Time { return time.Now().UTC() }, - rooms: make(map[string]*liveRoom), + janitorStop: make(chan struct{}), + rooms: make(map[string]*liveRoom), + } + go s.runJanitor() + return s +} + +// Close stops the service's background maintenance loop. +func (s *Service) Close() { + if s == nil || s.janitorStop == nil { + return + } + select { + case <-s.janitorStop: + default: + close(s.janitorStop) } } @@ -256,6 +291,10 @@ func (s *Service) Connect( } memberKey := buildMemberKey(userID, profileID) + displayName := fallbackMemberName + if s.profileNames != nil { + displayName = s.profileNames.ProfileDisplayName(ctx, userID, profileID) + } s.mu.Lock() current := live.members[memberKey] @@ -267,6 +306,7 @@ func (s *Service) Connect( previousConn = current.connection } current.connection = conn + current.displayName = displayName if room.HostUserID == userID && room.HostProfileID == profileID && live.hostCloseTimer != nil { live.hostCloseTimer.Stop() @@ -289,7 +329,6 @@ func (s *Service) Disconnect(reg *Registration, explicitLeave bool) { return } - var dispatches []snapshotDispatch s.mu.Lock() live := s.rooms[reg.roomID] if live == nil { @@ -321,290 +360,35 @@ func (s *Service) Disconnect(reg *Registration, explicitLeave bool) { hostUserID := live.room.HostUserID hostProfileID := live.room.HostProfileID live.hostCloseTimer = time.AfterFunc(s.hostDisconnectTTL, func() { - _ = s.CloseRoom(context.Background(), roomID, hostUserID, hostProfileID) + s.closeIfHostStillDisconnected(roomID, hostUserID, hostProfileID) }) } - dispatches = s.prepareSnapshotDispatchesLocked(live) - s.mu.Unlock() - s.runDispatches(dispatches) -} - -func (s *Service) AttachSession( - ctx context.Context, - roomID string, - userID int, - profileID string, - sessionID string, -) (Snapshot, error) { - room, live, err := s.getOrLoadLiveRoom(ctx, roomID) - if err != nil { - return Snapshot{}, err - } - - session, err := s.sessions.GetSession(sessionID) - if err != nil { - return Snapshot{}, err - } - if session.UserID != userID || session.ProfileID != profileID { - return Snapshot{}, ErrSessionMismatch - } - if err := s.validateSessionContent(ctx, room, session); err != nil { - return Snapshot{}, err - } - - s.mu.Lock() - member := live.members[buildMemberKey(userID, profileID)] - if member == nil || member.connection == nil { - s.mu.Unlock() - return Snapshot{}, ErrRoomForbidden - } - member.sessionID = sessionID - member.isReady = false - member.isBuffering = live.room.Phase == RoomPhasePlaying - - var commandDispatches []commandDispatch - if live.room.Phase == RoomPhasePlaying { - if live.room.PlaybackState == RoomPlaybackStatePlaying && s.activeParticipantCountLocked(live) > 1 { - position := s.expectedPositionLocked(live) - commandDispatches, _ = s.enterWaitingLocked(live, position, true) - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) - if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot := s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } - s.mu.Unlock() - return Snapshot{}, updateErr - } - live.room = *persisted - } else { - commandDispatches = s.syncMemberToRoomLocked(live, sessionID) - } - } - - snapshot := s.buildSnapshotLocked(live, userID, profileID) - dispatches := s.prepareSnapshotDispatchesLocked(live) - s.mu.Unlock() - - s.runDispatches(dispatches) - s.runCommandDispatches(commandDispatches) - return snapshot, nil -} - -func (s *Service) HandleTransportRequest( - ctx context.Context, - roomID string, - userID int, - profileID string, - request TransportRequest, -) (Snapshot, error) { - _, live, err := s.getOrLoadLiveRoom(ctx, roomID) - if err != nil { - return Snapshot{}, err - } - - s.mu.Lock() - member := live.members[buildMemberKey(userID, profileID)] - if member == nil || member.connection == nil || member.sessionID == "" { - s.mu.Unlock() - return Snapshot{}, ErrConnectionNotAttached - } - if err := s.ensureTransportAllowedLocked(live, userID, profileID, request.Action); err != nil { - s.mu.Unlock() - return Snapshot{}, err - } - - position := live.room.AnchorPositionSeconds - if request.PositionSeconds != nil { - position = math.Max(0, *request.PositionSeconds) - } else if !live.room.IsPaused { - position = s.expectedPositionLocked(live) - } - - now := s.now() - live.room.AnchorPositionSeconds = position - live.room.AnchorUpdatedAt = now - commandDispatches := []commandDispatch(nil) - executeAt := now.Add(s.highestPingLocked(live)) - switch request.Action { - case TransportActionPlay: - live.room.ResumeOnReady = true - live.room.IsPaused = false - live.room.PlaybackState = RoomPlaybackStatePlaying - commandDispatches = s.transportCommandDispatchesLocked( - live, - TransportActionPlay, - position, - executeAt, - ) - case TransportActionPause: - live.room.ResumeOnReady = false - live.room.IsPaused = true - live.room.PlaybackState = RoomPlaybackStatePaused - commandDispatches = s.transportCommandDispatchesLocked( - live, - TransportActionPause, - position, - executeAt, - ) - case TransportActionSeek: - live.room.ResumeOnReady = !request.IsPaused - live.room.IsPaused = true - live.room.PlaybackState = RoomPlaybackStateWaiting - s.resetMemberReadinessLocked(live, false) - commandDispatches = s.transportCommandDispatchesLocked( - live, - TransportActionSeek, - position, - executeAt, - ) - default: - s.mu.Unlock() - return Snapshot{}, ErrTransportNotAllowed - } - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) - if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - } - s.mu.Unlock() - if errors.Is(updateErr, ErrRoomStateConflict) { - return s.Snapshot(ctx, roomID, userID, profileID) - } - return Snapshot{}, updateErr - } - live.room = *persisted - snapshot := s.buildSnapshotLocked(live, userID, profileID) - dispatches := s.prepareSnapshotDispatchesLocked(live) - s.mu.Unlock() - - s.runDispatches(dispatches) - s.runCommandDispatches(commandDispatches) - return snapshot, nil -} - -func (s *Service) HandleStateReport( - ctx context.Context, - roomID string, - userID int, - profileID string, - report StateReport, -) (Snapshot, error) { - _, live, err := s.getOrLoadLiveRoom(ctx, roomID) - if err != nil { - return Snapshot{}, err - } - - var dispatches []snapshotDispatch - var correctionDispatches []commandDispatch - - s.mu.Lock() - member := live.members[buildMemberKey(userID, profileID)] - if member == nil || member.connection == nil { - s.mu.Unlock() - return Snapshot{}, ErrRoomForbidden - } - if member.sessionID == "" || member.sessionID != report.SessionID { - s.mu.Unlock() - return Snapshot{}, ErrConnectionNotAttached - } - - isHost := userID == live.room.HostUserID && profileID == live.room.HostProfileID - expected := s.expectedPositionLocked(live) - pauseMismatch := report.IsPaused != live.room.IsPaused - drift := math.Abs(report.PositionSeconds - expected) - - snapshot := s.buildSnapshotLocked(live, userID, profileID) - if isHost && (pauseMismatch || drift > 1.5) { - live.room.AnchorPositionSeconds = math.Max(0, report.PositionSeconds) - live.room.IsPaused = report.IsPaused - live.room.AnchorUpdatedAt = s.now() - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) - if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot = s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } - s.mu.Unlock() - return Snapshot{}, updateErr - } - live.room = *persisted - snapshot = s.buildSnapshotLocked(live, userID, profileID) + // A departing member may have been the last participant the room was + // waiting on; re-evaluate readiness so the others aren't stuck. + dispatches, commandDispatches := s.maybeResumeFromWaitingLocked(context.Background(), live, false) + if dispatches == nil { dispatches = s.prepareSnapshotDispatchesLocked(live) - } else if !isHost && (pauseMismatch || drift > 1.0) { - correctionDispatches = s.targetedCommandDispatchesLocked(live, report.SessionID, TransportCommand{ - CommandID: uuid.NewString(), - SelectionRevision: live.room.SelectionRevision, - Action: func() TransportAction { - if live.room.PlaybackState == RoomPlaybackStatePlaying { - return TransportActionPlay - } - return TransportActionPause - }(), - PositionSeconds: math.Max(0, expectedPosition(live.room, s.now())), - ExecuteAt: s.now().Add(s.highestPingLocked(live)).UTC().Format(time.RFC3339Nano), - IssuedAt: s.now().UTC().Format(time.RFC3339Nano), - PlaybackState: live.room.PlaybackState, - }) } s.mu.Unlock() - if isHost && (pauseMismatch || drift > 1.5) { - s.runDispatches(dispatches) - return snapshot, nil - } + s.runDispatches(dispatches) + s.runCommandDispatches(commandDispatches) +} - if len(correctionDispatches) > 0 { - s.runCommandDispatches(correctionDispatches) +// closeIfHostStillDisconnected closes a room after the host-disconnect grace +// period, unless the host reconnected while the timer was in flight +// (Timer.Stop cannot recall a callback that already fired). +func (s *Service) closeIfHostStillDisconnected(roomID string, hostUserID int, hostProfileID string) { + s.mu.Lock() + live := s.rooms[roomID] + // A nil live room means nobody (host included) is connected; closing is + // safe. Only a live room with the host re-connected aborts the close. + if live != nil && s.hostConnectedLocked(live) { + s.mu.Unlock() + return } - - return snapshot, nil + s.mu.Unlock() + _ = s.CloseRoom(context.Background(), roomID, hostUserID, hostProfileID) } func (s *Service) AttachSessionForConnection( @@ -642,6 +426,7 @@ func (s *Service) AttachSessionForConnection( } member.sessionID = sessionID member.isReady = false + member.ignoreWait = false member.isBuffering = live.room.Phase == RoomPhasePlaying var commandDispatches []commandDispatch @@ -649,32 +434,16 @@ func (s *Service) AttachSessionForConnection( if live.room.PlaybackState == RoomPlaybackStatePlaying && s.activeParticipantCountLocked(live) > 1 { position := s.expectedPositionLocked(live) commandDispatches, _ = s.enterWaitingLocked(live, position, true) - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) - if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, reg.roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot := s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } + conflict, err := s.persistAnchorLocked(ctx, live) + if err != nil { s.mu.Unlock() - return Snapshot{}, updateErr + return Snapshot{}, err + } + if conflict { + snapshot := s.buildSnapshotLocked(live, userID, profileID) + s.mu.Unlock() + return snapshot, nil } - live.room = *persisted } else { commandDispatches = s.syncMemberToRoomLocked(live, sessionID) } @@ -726,13 +495,14 @@ func (s *Service) HandleTransportRequestForConnection( now := s.now() live.room.AnchorPositionSeconds = position live.room.AnchorUpdatedAt = now - commandDispatches := []commandDispatch(nil) + var commandDispatches []commandDispatch executeAt := now.Add(s.highestPingLocked(live)) switch request.Action { case TransportActionPlay: live.room.ResumeOnReady = true live.room.IsPaused = false live.room.PlaybackState = RoomPlaybackStatePlaying + s.disarmWaitingDeadlineLocked(live) commandDispatches = s.transportCommandDispatchesLocked( live, TransportActionPlay, @@ -743,6 +513,7 @@ func (s *Service) HandleTransportRequestForConnection( live.room.ResumeOnReady = false live.room.IsPaused = true live.room.PlaybackState = RoomPlaybackStatePaused + s.disarmWaitingDeadlineLocked(live) commandDispatches = s.transportCommandDispatchesLocked( live, TransportActionPause, @@ -754,6 +525,7 @@ func (s *Service) HandleTransportRequestForConnection( live.room.IsPaused = true live.room.PlaybackState = RoomPlaybackStateWaiting s.resetMemberReadinessLocked(live, false) + s.armWaitingDeadlineLocked(live) commandDispatches = s.transportCommandDispatchesLocked( live, TransportActionSeek, @@ -764,32 +536,16 @@ func (s *Service) HandleTransportRequestForConnection( s.mu.Unlock() return Snapshot{}, ErrTransportNotAllowed } - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) + conflict, updateErr := s.persistAnchorLocked(ctx, live) if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, reg.roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - } s.mu.Unlock() - if errors.Is(updateErr, ErrRoomStateConflict) { - return s.Snapshot(ctx, reg.roomID, userID, profileID) - } return Snapshot{}, updateErr } - live.room = *persisted + if conflict { + snapshot := s.buildSnapshotLocked(live, userID, profileID) + s.mu.Unlock() + return snapshot, nil + } snapshot := s.buildSnapshotLocked(live, userID, profileID) dispatches := s.prepareSnapshotDispatchesLocked(live) s.mu.Unlock() @@ -839,33 +595,16 @@ func (s *Service) HandleStateReportForConnection( live.room.AnchorPositionSeconds = math.Max(0, report.PositionSeconds) live.room.IsPaused = report.IsPaused live.room.AnchorUpdatedAt = s.now() - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) + conflict, updateErr := s.persistAnchorLocked(ctx, live) if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, reg.roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot = s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } s.mu.Unlock() return Snapshot{}, updateErr } - live.room = *persisted snapshot = s.buildSnapshotLocked(live, userID, profileID) + if conflict { + s.mu.Unlock() + return snapshot, nil + } dispatches = s.prepareSnapshotDispatchesLocked(live) } else if !isHost && (pauseMismatch || drift > 1.0) { correctionDispatches = s.targetedCommandDispatchesLocked(live, report.SessionID, TransportCommand{ @@ -928,61 +667,13 @@ func (s *Service) HandleReadyForConnection( member.isReady = true member.isBuffering = false + member.ignoreWait = false + + dispatches, commandDispatches = s.maybeResumeFromWaitingLocked(ctx, live, false) snapshot := s.buildSnapshotLocked(live, userID, profileID) - if live.room.PlaybackState != RoomPlaybackStateWaiting || !s.allParticipantsReadyLocked(live) { + if dispatches == nil { dispatches = s.prepareSnapshotDispatchesLocked(live) - s.mu.Unlock() - s.runDispatches(dispatches) - return snapshot, nil } - - live.room.AnchorPositionSeconds = math.Max(0, live.room.AnchorPositionSeconds) - live.room.AnchorUpdatedAt = s.now() - action := TransportActionPause - if live.room.ResumeOnReady { - live.room.IsPaused = false - live.room.PlaybackState = RoomPlaybackStatePlaying - action = TransportActionPlay - } else { - live.room.IsPaused = true - live.room.PlaybackState = RoomPlaybackStatePaused - action = TransportActionPause - } - - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) - if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, reg.roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot = s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } - s.mu.Unlock() - return Snapshot{}, updateErr - } - live.room = *persisted - snapshot = s.buildSnapshotLocked(live, userID, profileID) - dispatches = s.prepareSnapshotDispatchesLocked(live) - commandDispatches = s.transportCommandDispatchesLocked( - live, - action, - live.room.AnchorPositionSeconds, - s.now().Add(s.highestPingLocked(live)), - ) s.mu.Unlock() s.runDispatches(dispatches) @@ -1022,45 +713,36 @@ func (s *Service) HandleBufferingForConnection( member.isBuffering = true member.isReady = false - if live.room.Phase != RoomPhasePlaying { + // Members already excluded from the readiness barrier must not drag the + // whole room back into waiting while they catch up. + if live.room.Phase != RoomPhasePlaying || member.ignoreWait { snapshot := s.buildSnapshotLocked(live, userID, profileID) s.mu.Unlock() return snapshot, nil } if live.room.PlaybackState != RoomPlaybackStateWaiting { + // Bound how far a single member's report can move the shared anchor. + position := math.Max(0, report.PositionSeconds) + expected := math.Max(0, s.expectedPositionLocked(live)) + if math.Abs(position-expected) > maxBufferingAnchorDriftSeconds { + position = expected + } commandDispatches, _ = s.enterWaitingLocked( live, - math.Max(0, report.PositionSeconds), + position, live.room.PlaybackState == RoomPlaybackStatePlaying || live.room.ResumeOnReady, ) - - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdateAnchor( - ctx, - live.room.ID, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorUpdatedAt, - live.room.Generation, - expectedGeneration, - ) + conflict, updateErr := s.persistAnchorLocked(ctx, live) if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, reg.roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot := s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } s.mu.Unlock() return Snapshot{}, updateErr } - live.room = *persisted + if conflict { + snapshot := s.buildSnapshotLocked(live, userID, profileID) + s.mu.Unlock() + return snapshot, nil + } } snapshot := s.buildSnapshotLocked(live, userID, profileID) @@ -1094,6 +776,9 @@ func (s *Service) HandlePingForConnection( return ErrRoomForbidden } if pingMS > 0 { + if maxMS := maxTransportLead.Milliseconds(); pingMS > maxMS { + pingMS = maxMS + } member.lastPingMS = pingMS } return nil @@ -1122,29 +807,18 @@ func (s *Service) UpdatePolicy( } live.room.GuestControlPolicy = policy - expectedGeneration := live.room.Generation - live.room.Generation++ - persisted, updateErr := s.repo.UpdatePolicy( - ctx, - roomID, - policy, - live.room.Generation, - expectedGeneration, - ) + conflict, updateErr := s.persistRoomChangeLocked(ctx, live, func(room Room, expectedGeneration int64) (*Room, error) { + return s.repo.UpdatePolicy(ctx, roomID, room.GuestControlPolicy, room.Generation, expectedGeneration) + }) if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot := s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil - } s.mu.Unlock() return Snapshot{}, updateErr } - live.room = *persisted snapshot := s.buildSnapshotLocked(live, userID, profileID) + if conflict { + s.mu.Unlock() + return snapshot, nil + } dispatches := s.prepareSnapshotDispatchesLocked(live) s.mu.Unlock() @@ -1202,44 +876,49 @@ func (s *Service) SelectItem( live.room.AnchorPositionSeconds = 0 live.room.IsPaused = true live.room.AnchorUpdatedAt = now - s.resetMemberReadinessLocked(live, false) - expectedGeneration := live.room.Generation live.room.SelectionRevision++ - live.room.Generation++ - - persisted, updateErr := s.repo.UpdateSelection( - ctx, - roomID, - SelectItemInput{ - ContentID: resolved.ContentID, - FileID: resolved.FileID, - LibraryID: resolved.LibraryID, - }, - live.room.Phase, - live.room.PlaybackState, - live.room.ResumeOnReady, - live.room.AnchorPositionSeconds, - live.room.IsPaused, - live.room.AnchorUpdatedAt, - live.room.SelectionRevision, - live.room.Generation, - expectedGeneration, - ) - if updateErr != nil { - if errors.Is(updateErr, ErrRoomStateConflict) { - if refreshed, refreshErr := s.repo.GetRoomByID(ctx, roomID); refreshErr == nil && refreshed != nil { - live.room = *refreshed - } - snapshot := s.buildSnapshotLocked(live, userID, profileID) - s.mu.Unlock() - return snapshot, nil + // Sessions attached for the previous selection are stale: readiness for + // the new content must come from a fresh attach, not an old session. + for _, member := range live.members { + if member == nil { + continue } + member.sessionID = "" + member.isReady = false + member.isBuffering = false + member.ignoreWait = false + } + s.disarmWaitingDeadlineLocked(live) + + conflict, updateErr := s.persistRoomChangeLocked(ctx, live, func(room Room, expectedGeneration int64) (*Room, error) { + return s.repo.UpdateSelection( + ctx, + roomID, + SelectItemInput{ + ContentID: resolved.ContentID, + FileID: resolved.FileID, + LibraryID: resolved.LibraryID, + }, + room.Phase, + room.PlaybackState, + room.ResumeOnReady, + room.AnchorPositionSeconds, + room.IsPaused, + room.AnchorUpdatedAt, + room.SelectionRevision, + room.Generation, + expectedGeneration, + ) + }) + if updateErr != nil { s.mu.Unlock() return Snapshot{}, updateErr } - - live.room = *persisted snapshot := s.buildSnapshotLocked(live, userID, profileID) + if conflict { + s.mu.Unlock() + return snapshot, nil + } dispatches := s.prepareSnapshotDispatchesLocked(live) s.mu.Unlock() @@ -1288,6 +967,9 @@ func (s *Service) CloseRoom(ctx context.Context, roomID string, userID int, prof if live.hostCloseTimer != nil { live.hostCloseTimer.Stop() } + if live.waitingTimer != nil { + live.waitingTimer.Stop() + } delete(s.rooms, roomID) } s.mu.Unlock() @@ -1307,6 +989,229 @@ func (s *Service) loadRoom(ctx context.Context, load func() (*Room, error)) (*Ro return room, nil } +// persistRoomChangeLocked persists a caller-applied mutation of live.room. +// It bumps the room generation, releases s.mu for the database round-trip, +// then re-acquires it and reconciles live.room with the persisted row. It +// must be called with s.mu held and always returns with s.mu held. +// +// When it returns conflict=true the caller lost an optimistic-concurrency +// race: live.room has been refreshed from the database and the caller should +// rebuild its snapshot from it and skip dispatching transport commands. +func (s *Service) persistRoomChangeLocked( + ctx context.Context, + live *liveRoom, + persist func(room Room, expectedGeneration int64) (*Room, error), +) (conflict bool, err error) { + expected := live.room.Generation + live.room.Generation++ + roomCopy := live.room + + s.mu.Unlock() + persisted, persistErr := persist(roomCopy, expected) + var refreshed *Room + if errors.Is(persistErr, ErrRoomStateConflict) { + refreshed, _ = s.repo.GetRoomByID(ctx, roomCopy.ID) + } + s.mu.Lock() + + if persistErr != nil { + // This writer's optimistic increment never landed; undo it so a + // failed write cannot leave a phantom generation that makes every + // later CAS conflict. Concurrent writers' stacked increments are + // preserved because each writer undoes exactly its own. + live.room.Generation-- + if errors.Is(persistErr, ErrRoomStateConflict) { + // Adopt the database row only if it is at least as new as the + // local copy — a concurrent writer may have advanced live.room + // while the lock was released, and a stale refresh must not + // overwrite that newer state. + if refreshed != nil && refreshed.Generation >= live.room.Generation { + live.room = *refreshed + } + return true, nil + } + return false, persistErr + } + // A concurrent writer may have advanced the local copy while the lock was + // released; never regress it to an older persisted generation. + if persisted != nil && persisted.Generation >= live.room.Generation { + live.room = *persisted + } + return false, nil +} + +// persistAnchorLocked persists the room's anchor/playback-state fields via +// persistRoomChangeLocked. Must be called with s.mu held; returns with it held. +func (s *Service) persistAnchorLocked(ctx context.Context, live *liveRoom) (bool, error) { + return s.persistRoomChangeLocked(ctx, live, func(room Room, expectedGeneration int64) (*Room, error) { + return s.repo.UpdateAnchor( + ctx, + room.ID, + room.AnchorPositionSeconds, + room.IsPaused, + room.PlaybackState, + room.ResumeOnReady, + room.AnchorUpdatedAt, + room.Generation, + expectedGeneration, + ) + }) +} + +func (s *Service) armWaitingDeadlineLocked(live *liveRoom) { + if live.waitingTimer != nil { + live.waitingTimer.Stop() + } + live.waitingEpoch++ + epoch := live.waitingEpoch + roomID := live.room.ID + live.waitingTimer = time.AfterFunc(waitingResumeDeadline, func() { + s.handleWaitingDeadline(roomID, epoch) + }) +} + +func (s *Service) disarmWaitingDeadlineLocked(live *liveRoom) { + if live.waitingTimer != nil { + live.waitingTimer.Stop() + live.waitingTimer = nil + } + live.waitingEpoch++ +} + +// handleWaitingDeadline fires when a waiting period outlives +// waitingResumeDeadline: members that never became ready stop blocking the +// readiness barrier (ignoreWait) and playback resumes for everyone else. +func (s *Service) handleWaitingDeadline(roomID string, epoch int64) { + s.mu.Lock() + live := s.rooms[roomID] + if live == nil || live.waitingEpoch != epoch || live.room.PlaybackState != RoomPlaybackStateWaiting { + s.mu.Unlock() + return + } + for _, member := range live.members { + if member == nil || member.connection == nil || member.sessionID == "" { + continue + } + if !member.isReady { + member.ignoreWait = true + } + } + dispatches, commandDispatches := s.maybeResumeFromWaitingLocked(context.Background(), live, true) + s.mu.Unlock() + s.runDispatches(dispatches) + s.runCommandDispatches(commandDispatches) +} + +// maybeResumeFromWaitingLocked leaves the waiting state once every remaining +// participant is ready (or unconditionally when force is set), persists the +// transition, and prepares the resulting snapshot and transport dispatches. +// It returns (nil, nil) when the room is not ready to resume. Must be called +// with s.mu held; the lock is temporarily released for persistence. +func (s *Service) maybeResumeFromWaitingLocked( + ctx context.Context, + live *liveRoom, + force bool, +) ([]snapshotDispatch, []commandDispatch) { + if live.room.Phase != RoomPhasePlaying || live.room.PlaybackState != RoomPlaybackStateWaiting { + return nil, nil + } + if !force && !s.allParticipantsReadyLocked(live) { + return nil, nil + } + + saved := live.room + live.room.AnchorPositionSeconds = math.Max(0, live.room.AnchorPositionSeconds) + live.room.AnchorUpdatedAt = s.now() + action := TransportActionPause + if live.room.ResumeOnReady { + live.room.IsPaused = false + live.room.PlaybackState = RoomPlaybackStatePlaying + action = TransportActionPlay + } else { + live.room.IsPaused = true + live.room.PlaybackState = RoomPlaybackStatePaused + } + + conflict, err := s.persistAnchorLocked(ctx, live) + if err != nil { + // The transition never landed in the database. Restore the waiting + // state so snapshots keep matching persisted reality, and re-arm the + // deadline so the resume is retried instead of silently dropped. + live.room.AnchorPositionSeconds = saved.AnchorPositionSeconds + live.room.AnchorUpdatedAt = saved.AnchorUpdatedAt + live.room.IsPaused = saved.IsPaused + live.room.PlaybackState = saved.PlaybackState + s.armWaitingDeadlineLocked(live) + return s.prepareSnapshotDispatchesLocked(live), nil + } + if conflict { + // live.room now reflects the database row that won the race; if it is + // still waiting the armed deadline keeps covering it. + return s.prepareSnapshotDispatchesLocked(live), nil + } + s.disarmWaitingDeadlineLocked(live) + commandDispatches := s.transportCommandDispatchesLocked( + live, + action, + live.room.AnchorPositionSeconds, + s.now().Add(s.highestPingLocked(live)), + ) + return s.prepareSnapshotDispatchesLocked(live), commandDispatches +} + +func (s *Service) runJanitor() { + ticker := time.NewTicker(janitorInterval) + defer ticker.Stop() + for { + select { + case <-s.janitorStop: + return + case <-ticker.C: + s.sweepIdleRooms() + } + } +} + +// sweepIdleRooms evicts live rooms with no connected members from memory and +// closes rooms whose playback anchor has been idle for longer than +// roomIdleTTL, so abandoned rooms do not accumulate forever. +func (s *Service) sweepIdleRooms() { + s.mu.Lock() + for roomID, live := range s.rooms { + if live == nil || s.connectedMemberCountLocked(live) == 0 { + // Room state is fully persisted; it reloads on next access. Any + // pending host-close timer keeps working from the database. + if live != nil && live.waitingTimer != nil { + live.waitingTimer.Stop() + } + delete(s.rooms, roomID) + } + } + s.mu.Unlock() + + if s.repo == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cutoff := s.now().Add(-roomIdleTTL) + roomIDs, err := s.repo.ListIdleRoomIDs(ctx, cutoff, 100) + if err != nil { + return + } + closedAt := s.now() + for _, roomID := range roomIDs { + s.mu.Lock() + live := s.rooms[roomID] + hasMembers := live != nil && s.connectedMemberCountLocked(live) > 0 + s.mu.Unlock() + if hasMembers { + continue + } + _, _ = s.repo.CloseRoom(ctx, roomID, closedAt) + } +} + func (s *Service) getOrLoadLiveRoom(ctx context.Context, roomID string) (*Room, *liveRoom, error) { s.mu.Lock() if live := s.rooms[roomID]; live != nil { @@ -1391,6 +1296,30 @@ func (s *Service) buildSnapshotLocked(live *liveRoom, userID int, profileID stri invitePath = fmt.Sprintf("/rooms/join?token=%s", live.room.JoinToken) } + members := make([]MemberSummary, 0, len(live.members)) + for _, m := range live.members { + if m == nil || m.connection == nil { + continue + } + members = append(members, MemberSummary{ + UserID: m.userID, + ProfileID: m.profileID, + DisplayName: m.displayName, + IsHost: m.userID == live.room.HostUserID && m.profileID == live.room.HostProfileID, + IsSelf: m.userID == userID && m.profileID == profileID, + Connected: true, + }) + } + sort.Slice(members, func(i, j int) bool { + if members[i].IsHost != members[j].IsHost { + return members[i].IsHost + } + if members[i].DisplayName != members[j].DisplayName { + return members[i].DisplayName < members[j].DisplayName + } + return members[i].ProfileID < members[j].ProfileID + }) + return Snapshot{ RoomID: live.room.ID, Phase: live.room.Phase, @@ -1424,6 +1353,7 @@ func (s *Service) buildSnapshotLocked(live *liveRoom, userID int, profileID stri return member.sessionID }(), InvitePath: invitePath, + Members: members, } } @@ -1485,36 +1415,6 @@ func (s *Service) hostConnectedLocked(live *liveRoom) bool { return member != nil && member.connection != nil } -func (s *Service) sessionIDsLocked(live *liveRoom) []string { - var sessionIDs []string - for _, member := range live.members { - if member == nil || member.sessionID == "" { - continue - } - sessionIDs = append(sessionIDs, member.sessionID) - } - return sessionIDs -} - -func (s *Service) hasHostAttachedSessionLocked(live *liveRoom) bool { - member := live.members[buildMemberKey(live.room.HostUserID, live.room.HostProfileID)] - return member != nil && member.sessionID != "" -} - -func withoutSessionID(sessionIDs []string, excluded string) []string { - if excluded == "" || len(sessionIDs) == 0 { - return sessionIDs - } - filtered := make([]string, 0, len(sessionIDs)) - for _, sessionID := range sessionIDs { - if sessionID == excluded { - continue - } - filtered = append(filtered, sessionID) - } - return filtered -} - func (s *Service) expectedPositionLocked(live *liveRoom) float64 { return expectedPosition(live.room, s.now()) } @@ -1556,6 +1456,10 @@ func (s *Service) allParticipantsReadyLocked(live *liveRoom) bool { return participants > 0 } +// highestPingLocked returns the scheduling lead for transport commands: the +// worst measured round-trip time across participants, bounded to +// [minTransportLead, maxTransportLead] so one member's bogus latency cannot +// stall the room. func (s *Service) highestPingLocked(live *liveRoom) time.Duration { highest := defaultTransportLead for _, member := range live.members { @@ -1565,7 +1469,7 @@ func (s *Service) highestPingLocked(live *liveRoom) time.Duration { if member.lastPingMS <= 0 { continue } - delay := time.Duration(member.lastPingMS*2) * time.Millisecond + delay := time.Duration(member.lastPingMS) * time.Millisecond if delay > highest { highest = delay } @@ -1573,6 +1477,9 @@ func (s *Service) highestPingLocked(live *liveRoom) time.Duration { if highest < minTransportLead { return minTransportLead } + if highest > maxTransportLead { + return maxTransportLead + } return highest } @@ -1656,6 +1563,7 @@ func (s *Service) enterWaitingLocked(live *liveRoom, positionSeconds float64, re if s.activeParticipantCountLocked(live) == 0 { return nil, false } + s.armWaitingDeadlineLocked(live) executeAt := s.now().Add(s.highestPingLocked(live)) return s.transportCommandDispatchesLocked( live, diff --git a/internal/watchtogether/service_test.go b/internal/watchtogether/service_test.go index 6ac67b64..3e6037c6 100644 --- a/internal/watchtogether/service_test.go +++ b/internal/watchtogether/service_test.go @@ -12,6 +12,9 @@ import ( type stubRepo struct { room Room + // anchorErr, when set, is returned from UpdateAnchor to simulate a + // database failure. + anchorErr error } func (s *stubRepo) CreateRoom(_ context.Context, room Room) (*Room, error) { @@ -31,6 +34,9 @@ func (s *stubRepo) GetRoomByJoinToken(context.Context, string) (*Room, error) { room := s.room return &room, nil } +func (s *stubRepo) ListIdleRoomIDs(context.Context, time.Time, int) ([]string, error) { + return nil, nil +} func (s *stubRepo) UpdatePolicy(_ context.Context, _ string, policy GuestControlPolicy, generation int64, expectedGeneration int64) (*Room, error) { if s.room.Generation != expectedGeneration { return nil, ErrRoomStateConflict @@ -51,6 +57,9 @@ func (s *stubRepo) UpdateAnchor( generation int64, expectedGeneration int64, ) (*Room, error) { + if s.anchorErr != nil { + return nil, s.anchorErr + } if s.room.Generation != expectedGeneration { return nil, ErrRoomStateConflict } @@ -125,27 +134,6 @@ func (s *stubFiles) GetByID(context.Context, int) (*models.MediaFile, error) { return &cp, nil } -type dispatchedCommand struct { - sessionID string - name playback.CommandName -} - -type stubDispatcher struct { - commands []dispatchedCommand -} - -func (s *stubDispatcher) DispatchToSession( - command playback.CommandEnvelope, - _ time.Duration, - _ func(), -) playback.CommandDispatchResult { - s.commands = append(s.commands, dispatchedCommand{ - sessionID: command.SessionID, - name: command.Name, - }) - return playback.CommandDispatchResult{} -} - type stubConn struct{} func (stubConn) WriteJSON(any) error { return nil } @@ -204,8 +192,8 @@ func baseRoom(now time.Time) Room { } } -func newServiceForTest(now time.Time, repo *stubRepo, sessions *stubSessions, files *stubFiles, dispatcher *stubDispatcher, resolver WatchTogetherSelectionResolver) *Service { - service := NewService(repo, sessions, files, dispatcher, resolver, nil) +func newServiceForTest(now time.Time, repo *stubRepo, sessions *stubSessions, files *stubFiles, resolver WatchTogetherSelectionResolver) *Service { + service := NewService(repo, sessions, files, resolver, nil, nil) service.hostDisconnectTTL = time.Hour service.now = func() time.Time { return now } service.rooms[repo.room.ID] = &liveRoom{ @@ -215,6 +203,14 @@ func newServiceForTest(now time.Time, repo *stubRepo, sessions *stubSessions, fi return service } +func registrationFor(roomID string, userID int, profileID string, conn RoomConnection) *Registration { + return &Registration{ + roomID: roomID, + memberKey: buildMemberKey(userID, profileID), + connection: conn, + } +} + func stringPtr(value string) *string { return &value } @@ -223,14 +219,12 @@ func TestGuestPlayPausePolicyStillRejectsGuestSeek(t *testing.T) { now := time.Date(2026, 4, 9, 12, 0, 20, 0, time.UTC) repo := &stubRepo{room: baseRoom(now)} repo.room.GuestControlPolicy = GuestControlPolicyGuestPlayPause - dispatcher := &stubDispatcher{} conn := &recordingConn{} service := newServiceForTest( now, repo, &stubSessions{}, &stubFiles{file: &models.MediaFile{ID: 42, ContentID: "movie-1"}}, - dispatcher, nil, ) service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = &memberState{ @@ -241,20 +235,20 @@ func TestGuestPlayPausePolicyStillRejectsGuestSeek(t *testing.T) { } position := 120.0 - _, err := service.HandleTransportRequest(context.Background(), repo.room.ID, 8, "guest", TransportRequest{ + reg := registrationFor(repo.room.ID, 8, "guest", conn) + _, err := service.HandleTransportRequestForConnection(context.Background(), reg, 8, "guest", TransportRequest{ Action: TransportActionSeek, PositionSeconds: &position, IsPaused: false, }) if !errors.Is(err, ErrTransportNotAllowed) { - t.Fatalf("HandleTransportRequest(guest seek) error = %v, want ErrTransportNotAllowed", err) + t.Fatalf("HandleTransportRequestForConnection(guest seek) error = %v, want ErrTransportNotAllowed", err) } } func TestGuestDriftTriggersCorrection(t *testing.T) { now := time.Date(2026, 4, 9, 12, 0, 20, 0, time.UTC) repo := &stubRepo{room: baseRoom(now)} - dispatcher := &stubDispatcher{} conn := &recordingConn{} service := newServiceForTest( now, @@ -266,7 +260,6 @@ func TestGuestDriftTriggersCorrection(t *testing.T) { MediaFileID: 42, }}, &stubFiles{file: &models.MediaFile{ID: 42, ContentID: "movie-1"}}, - dispatcher, nil, ) service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = &memberState{ @@ -276,13 +269,14 @@ func TestGuestDriftTriggersCorrection(t *testing.T) { connection: conn, } - _, err := service.HandleStateReport(context.Background(), repo.room.ID, 8, "guest", StateReport{ + reg := registrationFor(repo.room.ID, 8, "guest", conn) + _, err := service.HandleStateReportForConnection(context.Background(), reg, 8, "guest", StateReport{ SessionID: "session-1", PositionSeconds: 2, IsPaused: false, }) if err != nil { - t.Fatalf("HandleStateReport() error = %v", err) + t.Fatalf("HandleStateReportForConnection() error = %v", err) } if len(conn.payloads) == 0 { @@ -297,7 +291,6 @@ func TestHostAttachKeepsRoomSelectionAnchor(t *testing.T) { repo.room.IsPaused = true repo.room.AnchorUpdatedAt = now repo.room.Generation = 1 - dispatcher := &stubDispatcher{} conn := &recordingConn{} service := newServiceForTest( now, @@ -311,7 +304,6 @@ func TestHostAttachKeepsRoomSelectionAnchor(t *testing.T) { IsPaused: false, }}, &stubFiles{file: &models.MediaFile{ID: 42, ContentID: "movie-1"}}, - dispatcher, nil, ) service.rooms[repo.room.ID].members[buildMemberKey(7, "host")] = &memberState{ @@ -320,9 +312,10 @@ func TestHostAttachKeepsRoomSelectionAnchor(t *testing.T) { connection: conn, } - snapshot, err := service.AttachSession(context.Background(), repo.room.ID, 7, "host", "session-1") + reg := registrationFor(repo.room.ID, 7, "host", conn) + snapshot, err := service.AttachSessionForConnection(context.Background(), reg, 7, "host", "session-1") if err != nil { - t.Fatalf("AttachSession() error = %v", err) + t.Fatalf("AttachSessionForConnection() error = %v", err) } if snapshot.AnchorPositionSeconds != 0 { @@ -346,7 +339,6 @@ func TestHostAttachKeepsRoomSelectionAnchorEvenWhenGuestAttached(t *testing.T) { repo.room.IsPaused = true repo.room.AnchorUpdatedAt = now repo.room.Generation = 1 - dispatcher := &stubDispatcher{} service := newServiceForTest( now, repo, @@ -359,7 +351,6 @@ func TestHostAttachKeepsRoomSelectionAnchorEvenWhenGuestAttached(t *testing.T) { IsPaused: false, }}, &stubFiles{file: &models.MediaFile{ID: 42, ContentID: "movie-1"}}, - dispatcher, nil, ) service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = &memberState{ @@ -374,9 +365,10 @@ func TestHostAttachKeepsRoomSelectionAnchorEvenWhenGuestAttached(t *testing.T) { connection: stubConn{}, } - snapshot, err := service.AttachSession(context.Background(), repo.room.ID, 7, "host", "host-session") + reg := registrationFor(repo.room.ID, 7, "host", stubConn{}) + snapshot, err := service.AttachSessionForConnection(context.Background(), reg, 7, "host", "host-session") if err != nil { - t.Fatalf("AttachSession() error = %v", err) + t.Fatalf("AttachSessionForConnection() error = %v", err) } if snapshot.AnchorPositionSeconds != 0 { @@ -395,7 +387,6 @@ func TestAttachSessionAcceptsEpisodeContentID(t *testing.T) { repo.room.IsPaused = true repo.room.AnchorUpdatedAt = now repo.room.Generation = 1 - dispatcher := &stubDispatcher{} service := newServiceForTest( now, repo, @@ -412,7 +403,6 @@ func TestAttachSessionAcceptsEpisodeContentID(t *testing.T) { ContentID: "series-1", EpisodeID: "episode-19", }}, - dispatcher, nil, ) service.rooms[repo.room.ID].members[buildMemberKey(7, "host")] = &memberState{ @@ -421,9 +411,10 @@ func TestAttachSessionAcceptsEpisodeContentID(t *testing.T) { connection: stubConn{}, } - snapshot, err := service.AttachSession(context.Background(), repo.room.ID, 7, "host", "host-session") + reg := registrationFor(repo.room.ID, 7, "host", stubConn{}) + snapshot, err := service.AttachSessionForConnection(context.Background(), reg, 7, "host", "host-session") if err != nil { - t.Fatalf("AttachSession() error = %v", err) + t.Fatalf("AttachSessionForConnection() error = %v", err) } if snapshot.AttachedSessionID != "host-session" { @@ -437,7 +428,7 @@ func TestAttachSessionAcceptsEpisodeContentID(t *testing.T) { func TestCreateRoomStartsInLobbyWithoutSelection(t *testing.T) { now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) repo := &stubRepo{} - service := NewService(repo, &stubSessions{}, &stubFiles{}, &stubDispatcher{}, nil, nil) + service := NewService(repo, &stubSessions{}, &stubFiles{}, nil, nil, nil) service.now = func() time.Time { return now } room, err := service.CreateRoom(context.Background(), CreateRoomInput{ @@ -480,7 +471,6 @@ func TestHostCanSelectItemFromLobby(t *testing.T) { repo, &stubSessions{}, &stubFiles{}, - &stubDispatcher{}, &stubSelectionResolver{resolved: &ResolvedSelection{ ContentID: "movie-2", FileID: intPtr(55), @@ -515,6 +505,41 @@ func TestHostCanSelectItemFromLobby(t *testing.T) { } } +func TestSelectItemClearsStaleMemberSessions(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) + repo := &stubRepo{room: baseRoom(now)} + service := newServiceForTest( + now, + repo, + &stubSessions{}, + &stubFiles{}, + &stubSelectionResolver{resolved: &ResolvedSelection{ContentID: "movie-2"}}, + ) + guest := &memberState{ + userID: 8, + profileID: "guest", + sessionID: "old-session", + isReady: true, + ignoreWait: true, + connection: stubConn{}, + } + service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = guest + + _, err := service.SelectItem(context.Background(), "room-1", 7, "host", SelectItemInput{ + ContentID: "movie-2", + }) + if err != nil { + t.Fatalf("SelectItem() error = %v", err) + } + + if guest.sessionID != "" { + t.Fatalf("guest session = %q, want cleared", guest.sessionID) + } + if guest.isReady || guest.ignoreWait { + t.Fatal("guest readiness flags should reset on new selection") + } +} + func TestGuestCannotSelectItem(t *testing.T) { now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) repo := &stubRepo{room: baseRoom(now)} @@ -523,7 +548,6 @@ func TestGuestCannotSelectItem(t *testing.T) { repo, &stubSessions{}, &stubFiles{}, - &stubDispatcher{}, &stubSelectionResolver{resolved: &ResolvedSelection{ContentID: "movie-2"}}, ) @@ -543,7 +567,6 @@ func TestSelectItemRejectsInvalidSelection(t *testing.T) { repo, &stubSessions{}, &stubFiles{}, - &stubDispatcher{}, &stubSelectionResolver{err: ErrInvalidSelection}, ) @@ -559,7 +582,6 @@ func TestAttachSessionEnforcesSelectedFileID(t *testing.T) { now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) repo := &stubRepo{room: baseRoom(now)} repo.room.SelectedFileID = intPtr(99) - dispatcher := &stubDispatcher{} service := newServiceForTest( now, repo, @@ -570,7 +592,6 @@ func TestAttachSessionEnforcesSelectedFileID(t *testing.T) { MediaFileID: 42, }}, &stubFiles{file: &models.MediaFile{ID: 42, ContentID: "movie-1"}}, - dispatcher, nil, ) service.rooms[repo.room.ID].members[buildMemberKey(7, "host")] = &memberState{ @@ -579,9 +600,171 @@ func TestAttachSessionEnforcesSelectedFileID(t *testing.T) { connection: stubConn{}, } - _, err := service.AttachSession(context.Background(), repo.room.ID, 7, "host", "host-session") + reg := registrationFor(repo.room.ID, 7, "host", stubConn{}) + _, err := service.AttachSessionForConnection(context.Background(), reg, 7, "host", "host-session") if !errors.Is(err, ErrSessionMismatch) { - t.Fatalf("AttachSession() error = %v, want ErrSessionMismatch", err) + t.Fatalf("AttachSessionForConnection() error = %v, want ErrSessionMismatch", err) + } +} + +func TestDisconnectOfLastUnreadyMemberResumesWaitingRoom(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) + repo := &stubRepo{room: baseRoom(now)} + repo.room.PlaybackState = RoomPlaybackStateWaiting + repo.room.IsPaused = true + repo.room.ResumeOnReady = true + service := newServiceForTest(now, repo, &stubSessions{}, &stubFiles{}, nil) + + hostConn := &recordingConn{} + service.rooms[repo.room.ID].members[buildMemberKey(7, "host")] = &memberState{ + userID: 7, + profileID: "host", + sessionID: "host-session", + isReady: true, + connection: hostConn, + } + guestConn := &recordingConn{} + service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = &memberState{ + userID: 8, + profileID: "guest", + sessionID: "guest-session", + isReady: false, + connection: guestConn, + } + + service.Disconnect(registrationFor(repo.room.ID, 8, "guest", guestConn), false) + + if repo.room.PlaybackState != RoomPlaybackStatePlaying { + t.Fatalf("playback state = %q, want %q", repo.room.PlaybackState, RoomPlaybackStatePlaying) + } + foundCommand := false + for _, payload := range hostConn.payloads { + if payload["type"] == "transport_command" { + foundCommand = true + } + } + if !foundCommand { + t.Fatal("expected a resume transport command for the remaining member") + } +} + +func TestBufferingReportCannotTeleportRoomAnchor(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) + repo := &stubRepo{room: baseRoom(now)} + service := newServiceForTest(now, repo, &stubSessions{}, &stubFiles{}, nil) + guestConn := &recordingConn{} + service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = &memberState{ + userID: 8, + profileID: "guest", + sessionID: "guest-session", + connection: guestConn, + } + + // Anchor was 10s, 10s ago and playing: expected position is ~20s. A + // report claiming 500s must be clamped back to the expected position. + reg := registrationFor(repo.room.ID, 8, "guest", guestConn) + snapshot, err := service.HandleBufferingForConnection(context.Background(), reg, 8, "guest", StateReport{ + SessionID: "guest-session", + PositionSeconds: 500, + IsPaused: false, + }) + if err != nil { + t.Fatalf("HandleBufferingForConnection() error = %v", err) + } + if snapshot.PlaybackState != RoomPlaybackStateWaiting { + t.Fatalf("playback state = %q, want %q", snapshot.PlaybackState, RoomPlaybackStateWaiting) + } + if snapshot.AnchorPositionSeconds > 20.001 || snapshot.AnchorPositionSeconds < 19.999 { + t.Fatalf("anchor = %v, want ~20 (clamped)", snapshot.AnchorPositionSeconds) + } +} + +func TestPingIsClampedToMaxTransportLead(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) + repo := &stubRepo{room: baseRoom(now)} + service := newServiceForTest(now, repo, &stubSessions{}, &stubFiles{}, nil) + guestConn := &recordingConn{} + member := &memberState{ + userID: 8, + profileID: "guest", + sessionID: "guest-session", + connection: guestConn, + } + service.rooms[repo.room.ID].members[buildMemberKey(8, "guest")] = member + + reg := registrationFor(repo.room.ID, 8, "guest", guestConn) + if err := service.HandlePingForConnection(context.Background(), reg, 8, "guest", 3_600_000); err != nil { + t.Fatalf("HandlePingForConnection() error = %v", err) + } + if member.lastPingMS != maxTransportLead.Milliseconds() { + t.Fatalf("lastPingMS = %d, want %d", member.lastPingMS, maxTransportLead.Milliseconds()) + } +} + +func TestReadyPersistFailureKeepsWaitingState(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) + repo := &stubRepo{room: baseRoom(now)} + repo.room.PlaybackState = RoomPlaybackStateWaiting + repo.room.IsPaused = true + repo.room.ResumeOnReady = true + repo.anchorErr = errors.New("database unavailable") + service := newServiceForTest(now, repo, &stubSessions{}, &stubFiles{}, nil) + + hostConn := &recordingConn{} + service.rooms[repo.room.ID].members[buildMemberKey(7, "host")] = &memberState{ + userID: 7, + profileID: "host", + sessionID: "host-session", + connection: hostConn, + } + + reg := registrationFor(repo.room.ID, 7, "host", hostConn) + snapshot, err := service.HandleReadyForConnection(context.Background(), reg, 7, "host", StateReport{ + SessionID: "host-session", + }) + if err != nil { + t.Fatalf("HandleReadyForConnection() error = %v", err) + } + + if snapshot.PlaybackState != RoomPlaybackStateWaiting { + t.Fatalf("playback state = %q, want %q (resume must not be announced when persistence failed)", + snapshot.PlaybackState, RoomPlaybackStateWaiting) + } + live := service.rooms[repo.room.ID] + if live.room.Generation != repo.room.Generation { + t.Fatalf("live generation = %d, want %d (failed write must not leave a phantom generation)", + live.room.Generation, repo.room.Generation) + } + if live.waitingTimer == nil { + t.Fatal("waiting deadline should stay armed so the resume is retried") + } +} + +func TestStaleLiveConflictAdoptsDatabaseRow(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 20, 0, time.UTC) + repo := &stubRepo{room: baseRoom(now)} + service := newServiceForTest(now, repo, &stubSessions{}, &stubFiles{}, nil) + // The database row has moved ahead of the cached live copy. + repo.room.Generation = 5 + + hostConn := &recordingConn{} + service.rooms[repo.room.ID].members[buildMemberKey(7, "host")] = &memberState{ + userID: 7, + profileID: "host", + sessionID: "host-session", + connection: hostConn, + } + + reg := registrationFor(repo.room.ID, 7, "host", hostConn) + snapshot, err := service.HandleTransportRequestForConnection(context.Background(), reg, 7, "host", TransportRequest{ + Action: TransportActionPause, + }) + if err != nil { + t.Fatalf("HandleTransportRequestForConnection() error = %v", err) + } + + if snapshot.Generation != 5 { + t.Fatalf("snapshot generation = %d, want 5 (conflict must adopt the newer database row)", snapshot.Generation) } } diff --git a/internal/watchtogether/types.go b/internal/watchtogether/types.go index d88101d8..04b458f8 100644 --- a/internal/watchtogether/types.go +++ b/internal/watchtogether/types.go @@ -74,6 +74,16 @@ type Room struct { ClosedAt *time.Time } +// MemberSummary describes one connected room member in a snapshot. +type MemberSummary struct { + UserID int `json:"user_id"` + ProfileID string `json:"profile_id"` + DisplayName string `json:"display_name"` + IsHost bool `json:"is_host"` + IsSelf bool `json:"is_self"` + Connected bool `json:"connected"` +} + type Snapshot struct { RoomID string `json:"room_id"` Phase RoomPhase `json:"phase"` @@ -97,6 +107,7 @@ type Snapshot struct { SelfIgnoreWait bool `json:"self_ignore_wait"` AttachedSessionID string `json:"attached_session_id,omitempty"` InvitePath string `json:"invite_path,omitempty"` + Members []MemberSummary `json:"members,omitempty"` } type RoomJoinResult struct { diff --git a/web/src/App.tsx b/web/src/App.tsx index 9eb55094..527d56b7 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -142,8 +142,21 @@ function ScrollRestorationManager() { return null; } +/** + * Builds a guard redirect target (e.g. "/login") that preserves the current + * location so the user returns to it after authenticating. + */ +function guardRedirectTarget(base: string, location: ReturnType): string { + const destination = `${location.pathname}${location.search}`; + if (destination === "/" || destination === "") { + return base; + } + return `${base}?redirect=${encodeURIComponent(destination)}`; +} + function RequireAuth({ children }: { children: ReactNode }) { const { user, loading, setupLoading } = useAuth(); + const location = useLocation(); if (loading || setupLoading) { return (
@@ -152,7 +165,7 @@ function RequireAuth({ children }: { children: ReactNode }) {
); } - if (!user) return ; + if (!user) return ; return <>{children}; } @@ -172,7 +185,8 @@ function SetupGate({ children }: { children: ReactNode }) { function RequireProfile({ children }: { children: ReactNode }) { const { profile } = useAuth(); - if (!profile) return ; + const location = useLocation(); + if (!profile) return ; return <>{children}; } diff --git a/web/src/components/watchtogether/ConnectionStatusDot.tsx b/web/src/components/watchtogether/ConnectionStatusDot.tsx new file mode 100644 index 00000000..156c0b22 --- /dev/null +++ b/web/src/components/watchtogether/ConnectionStatusDot.tsx @@ -0,0 +1,32 @@ +export type WatchTogetherConnectionState = "disconnected" | "connecting" | "connected"; + +const connectionLabels: Record = { + connected: "Connected", + connecting: "Connecting…", + disconnected: "Disconnected", +}; + +/** + * Shared watch-together connection status label. Pair with + * `ConnectionStatusDot` so every surface uses the same palette + vocabulary. + */ +export function ConnectionStateLabel({ state }: { state: WatchTogetherConnectionState }) { + return <>{connectionLabels[state] ?? connectionLabels.disconnected}; +} + +/** Shared watch-together connection indicator dot. */ +export function ConnectionStatusDot({ + state, + className = "h-2 w-2", +}: { + state: WatchTogetherConnectionState; + className?: string; +}) { + const color = + state === "connected" + ? "bg-emerald-400" + : state === "connecting" + ? "animate-pulse bg-amber-300" + : "bg-red-400"; + return