feat(sync): add canonical room media state

server/README.md

@@ -104,5 +104,6 @@ For focused server checks, see `scripts/test-server-ops.mjs`, `scripts/test-serv
 | File | Purpose |
 |---|---|
 | `index.js` | Express + Socket.IO server: room management, relay loop, graceful shutdown |
+| `media-state.js` | Canonical room media clock, control updates, and projected ROOM_DATA snapshots |
 | `rate-limiter.js` | Connection, event, health, and auth rate limiting with 6 functions + cleanup intervals |
 | `ops.js` | Health endpoint helpers, metrics payload builder, auth validation |

server/index.js

@@ -4,8 +4,13 @@ import { fileURLToPath } from 'url';
 import { Server } from 'socket.io';
 import crypto from 'crypto';
 import dotenv from 'dotenv';
-import { EVENTS, OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION, CONTROL_MODES, CAPABILITIES } from '../shared/constants.js';
+import { EVENTS, OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION, CONTROL_MODES, CAPABILITIES, MAX_MEDIA_TIME } from '../shared/constants.js';
 import { createChatEnvelope } from './chat.js';
+import {
+    commitForceSyncMediaState,
+    snapshotMediaState,
+    updateMediaStateFromControl
+} from './media-state.js';
 import {
     buildHealthPayload,
     checkCooldown,
@@ -178,7 +183,8 @@ const SERVER_CAPABILITIES = [
     CAPABILITIES.HOST_CONTROL,
     CAPABILITIES.CO_HOST,
     CAPABILITIES.CHAT,
-    CAPABILITIES.CHAT_V1
+    CAPABILITIES.CHAT_V1,
+    CAPABILITIES.MEDIA_STATE_V1
 ];
 
 function normalizeClientCapabilities(value) {
@@ -283,6 +289,7 @@ function removePeerFromRoom(socketId, roomId, reason) {
         // H-1: a leaving initiator strands the room's force-sync — release the
         // slot so a future controller's PREPARE can take over cleanly.
         if (room.forceSyncInitiator === peerId) room.forceSyncInitiator = null;
+        if (room.forceSyncTarget?.initiatorPeerId === peerId) room.forceSyncTarget = null;
         if (room.hostPeerId === peerId) {
             // Owner left → reassign owner + fall back to 'everyone' so the room is
             // never stuck locked, and reset the controller set to just the new owner.
@@ -445,7 +452,13 @@ io.on('connection', (socket) => {
                             // controller's FORCE_SYNC_EXECUTE through the host-only gate — without
                             // it, demoting a co-host mid-force-sync would drop their EXECUTE and
                             // leave every peer stuck paused.
-                            forceSyncInitiator: null
+                            forceSyncInitiator: null,
+                            // Canonical Media State v1 is lazy, room-local and absent until
+                            // the first accepted command establishes a trustworthy position.
+                            mediaState: null,
+                            // PREPARE is choreography, not stable room intent. Retain its
+                            // validated target only so the matching EXECUTE can commit it.
+                            forceSyncTarget: null
                         };
                         rooms.set(roomId, room);
                         createdByMe = true;
@@ -528,6 +541,7 @@ io.on('connection', (socket) => {
             peerToSocket.set(peerId, socket.id);
 
             socket.to(roomId).emit(EVENTS.PEER_STATUS, { peerId, username: username || null, tabTitle: tabTitle || null, mediaTitle: mediaTitle || null, status: 'joined' });
+            const snapshotAt = Date.now();
             socket.emit(EVENTS.ROOM_DATA, {
                 roomId,
                 peers: Array.from(room.peers).map(sid => room.peerData.get(sid)),
@@ -535,6 +549,7 @@ io.on('connection', (socket) => {
                 hostPeerId: room.hostPeerId || null,
                 controlMode: room.controlMode || CONTROL_MODES.EVERYONE,
                 controllers: room.controllers ? Array.from(room.controllers) : [],
+                mediaState: snapshotMediaState(room.mediaState, snapshotAt),
                 capabilities: SERVER_CAPABILITIES
             });
             log('ROOM', `Peer ${peerId} joined: ${roomId.substring(0, 3)}***`);
@@ -617,7 +632,7 @@ io.on('connection', (socket) => {
                         tabTitle:      data.tabTitle      === null ? null : (data.tabTitle      !== undefined ? (clamp(data.tabTitle, 100)  ?? existing.tabTitle)      : existing.tabTitle),
                         mediaTitle:    data.mediaTitle    === null ? null : (data.mediaTitle    !== undefined ? (clamp(data.mediaTitle, 100) ?? existing.mediaTitle)   : existing.mediaTitle),
                         playbackState: data.playbackState !== undefined ? (validState(data.playbackState) ?? existing.playbackState) : existing.playbackState,
-                        currentTime:   data.currentTime   === null ? null : (data.currentTime   !== undefined ? (clampNum(data.currentTime, 0, 86400) ?? existing.currentTime)   : existing.currentTime),
+                        currentTime:   data.currentTime   === null ? null : (data.currentTime   !== undefined ? (clampNum(data.currentTime, 0, MAX_MEDIA_TIME) ?? existing.currentTime)   : existing.currentTime),
                         volume:        data.volume        !== undefined ? (clampNum(data.volume, 0, 1) ?? existing.volume)                 : existing.volume,
                         muted:         data.muted         !== undefined ? (validBool(data.muted) ?? existing.muted)                       : existing.muted,
                         desynced:      data.desynced      !== undefined ? (validBool(data.desynced) === true) : (existing.desynced || false),
@@ -628,8 +643,8 @@ io.on('connection', (socket) => {
                     const relayPayload = {
                         senderId:        mapping.peerId,
                         seq:             clampNum(data.seq, 0, Number.MAX_SAFE_INTEGER),
-                        currentTime:     data.currentTime === null ? null : clampNum(data.currentTime, 0, 86400),
-                        targetTime:      clampNum(data.targetTime, 0, 86400),
+                        currentTime:     data.currentTime === null ? null : clampNum(data.currentTime, 0, MAX_MEDIA_TIME),
+                        targetTime:      clampNum(data.targetTime, 0, MAX_MEDIA_TIME),
                         playbackState:   validState(data.playbackState),
                         username:        clamp(data.username, 30),
                         tabTitle:        data.tabTitle === null ? null : clamp(data.tabTitle, 100),
@@ -645,6 +660,32 @@ io.on('connection', (socket) => {
                     };
                     // Strip undefined keys for clean wire format
                     Object.keys(relayPayload).forEach(k => relayPayload[k] === undefined && delete relayPayload[k]);
+
+                    // Canonical Media State v1: mutate only after rate limiting,
+                    // room mapping, Host Control authorization and sanitization.
+                    // Heartbeats remain observational and never enter this path.
+                    const mediaStateNow = Date.now();
+                    updateMediaStateFromControl(room, eventName, relayPayload, mapping.peerId, {
+                        now: mediaStateNow,
+                        senderPlaybackState: existing.playbackState
+                    });
+                    if (eventName === EVENTS.FORCE_SYNC_PREPARE) {
+                        room.forceSyncTarget = Number.isFinite(relayPayload.targetTime)
+                            ? { initiatorPeerId: mapping.peerId, targetTime: relayPayload.targetTime }
+                            : null;
+                    } else if (eventName === EVENTS.FORCE_SYNC_EXECUTE) {
+                        const forceSyncTarget = room.forceSyncTarget;
+                        if (forceSyncTarget?.initiatorPeerId === mapping.peerId) {
+                            commitForceSyncMediaState(
+                                room,
+                                forceSyncTarget.targetTime,
+                                mapping.peerId,
+                                mediaStateNow
+                            );
+                        }
+                        room.forceSyncTarget = null;
+                    }
+
                     socket.to(mapping.roomId).emit(eventName, relayPayload);
 
                     // --- Side-effects: Server-side Episode Lobby Tracking ---

server/media-state.js

@@ -0,0 +1,87 @@
+import { EVENTS, MAX_MEDIA_TIME } from '../shared/constants.js';
+
+function clampMediaTime(value) {
+    if (typeof value !== 'number' || !Number.isFinite(value)) return null;
+    return Math.max(0, Math.min(MAX_MEDIA_TIME, value));
+}
+
+export function effectiveMediaPosition(mediaState, now = Date.now()) {
+    if (!mediaState) return null;
+    const currentTime = clampMediaTime(mediaState.currentTime);
+    if (currentTime === null) return null;
+    if (mediaState.playbackState !== 'playing') return currentTime;
+    const updatedAt = typeof mediaState.updatedAt === 'number' && Number.isFinite(mediaState.updatedAt)
+        ? mediaState.updatedAt
+        : now;
+    return clampMediaTime(currentTime + Math.max(0, now - updatedAt) / 1000);
+}
+
+export function snapshotMediaState(mediaState, now = Date.now()) {
+    if (!mediaState) return null;
+    const currentTime = effectiveMediaPosition(mediaState, now);
+    if (currentTime === null
+        || !Number.isSafeInteger(mediaState.revision)
+        || mediaState.revision < 1
+        || (mediaState.playbackState !== 'playing' && mediaState.playbackState !== 'paused')) {
+        return null;
+    }
+    return {
+        revision: mediaState.revision,
+        playbackState: mediaState.playbackState,
+        currentTime,
+        updatedBy: mediaState.updatedBy
+    };
+}
+
+function commitMediaState(room, playbackState, currentTime, updatedBy, now) {
+    const normalizedTime = clampMediaTime(currentTime);
+    if (normalizedTime === null
+        || (playbackState !== 'playing' && playbackState !== 'paused')
+        || typeof updatedBy !== 'string'
+        || !updatedBy) {
+        return false;
+    }
+    room.mediaState = {
+        revision: (room.mediaState?.revision || 0) + 1,
+        playbackState,
+        currentTime: normalizedTime,
+        updatedAt: now,
+        updatedBy
+    };
+    return true;
+}
+
+export function updateMediaStateFromControl(room, eventName, payload, senderPeerId, {
+    now = Date.now(),
+    senderPlaybackState = null
+} = {}) {
+    if (!room || !payload || typeof payload !== 'object') return false;
+
+    if (eventName === EVENTS.PLAY || eventName === EVENTS.PAUSE) {
+        const eventPosition = clampMediaTime(payload.currentTime);
+        const currentTime = eventPosition ?? effectiveMediaPosition(room.mediaState, now);
+        if (currentTime === null) return false;
+        return commitMediaState(
+            room,
+            eventName === EVENTS.PLAY ? 'playing' : 'paused',
+            currentTime,
+            senderPeerId,
+            now
+        );
+    }
+
+    if (eventName === EVENTS.SEEK) {
+        const targetTime = clampMediaTime(payload.targetTime) ?? clampMediaTime(payload.currentTime);
+        const playbackState = payload.playbackState === 'playing' || payload.playbackState === 'paused'
+            ? payload.playbackState
+            : (room.mediaState?.playbackState || senderPlaybackState);
+        if (targetTime === null) return false;
+        return commitMediaState(room, playbackState, targetTime, senderPeerId, now);
+    }
+
+    return false;
+}
+
+export function commitForceSyncMediaState(room, targetTime, senderPeerId, now = Date.now()) {
+    return commitMediaState(room, 'playing', targetTime, senderPeerId, now);
+}

server/media-state.test.mjs

@@ -0,0 +1,97 @@
+import { describe, expect, it } from 'vitest';
+import { EVENTS, MAX_MEDIA_TIME } from '../shared/constants.js';
+import {
+    commitForceSyncMediaState,
+    effectiveMediaPosition,
+    snapshotMediaState,
+    updateMediaStateFromControl
+} from './media-state.js';
+
+function room(mediaState = null) {
+    return { mediaState };
+}
+
+describe('canonical media state', () => {
+    it('starts null and snapshots null', () => {
+        expect(snapshotMediaState(null, 1000)).toBeNull();
+    });
+
+    it('projects playing state and clamps elapsed time', () => {
+        const state = { revision: 1, playbackState: 'playing', currentTime: 10, updatedAt: 1000, updatedBy: 'a' };
+        expect(effectiveMediaPosition(state, 3500)).toBe(12.5);
+        expect(effectiveMediaPosition(state, 500)).toBe(10);
+        expect(effectiveMediaPosition({ ...state, currentTime: MAX_MEDIA_TIME }, 3500)).toBe(MAX_MEDIA_TIME);
+    });
+
+    it('keeps paused state fixed', () => {
+        const state = { revision: 1, playbackState: 'paused', currentTime: 10, updatedAt: 1000, updatedBy: 'a' };
+        expect(effectiveMediaPosition(state, 601000)).toBe(10);
+        expect(snapshotMediaState(state, 601000)).toEqual({
+            revision: 1,
+            playbackState: 'paused',
+            currentTime: 10,
+            updatedBy: 'a'
+        });
+    });
+
+    it('initializes and updates PLAY with server-owned revisions', () => {
+        const target = room();
+        expect(updateMediaStateFromControl(target, EVENTS.PLAY, { currentTime: 12, revision: 999 }, 'a', { now: 1000 })).toBe(true);
+        expect(target.mediaState).toEqual({ revision: 1, playbackState: 'playing', currentTime: 12, updatedAt: 1000, updatedBy: 'a' });
+        expect(updateMediaStateFromControl(target, EVENTS.PLAY, { currentTime: 20 }, 'b', { now: 2000 })).toBe(true);
+        expect(target.mediaState.revision).toBe(2);
+        expect(target.mediaState.updatedBy).toBe('b');
+    });
+
+    it('does not invent an initial PLAY position and preserves a known effective position', () => {
+        const target = room();
+        expect(updateMediaStateFromControl(target, EVENTS.PLAY, {}, 'a', { now: 1000 })).toBe(false);
+        expect(target.mediaState).toBeNull();
+        target.mediaState = { revision: 1, playbackState: 'playing', currentTime: 5, updatedAt: 1000, updatedBy: 'a' };
+        expect(updateMediaStateFromControl(target, EVENTS.PLAY, {}, 'a', { now: 3000 })).toBe(true);
+        expect(target.mediaState.currentTime).toBe(7);
+    });
+
+    it('freezes PAUSE at its event or effective canonical position', () => {
+        const target = room({ revision: 1, playbackState: 'playing', currentTime: 10, updatedAt: 1000, updatedBy: 'a' });
+        expect(updateMediaStateFromControl(target, EVENTS.PAUSE, {}, 'a', { now: 4000 })).toBe(true);
+        expect(target.mediaState).toEqual({ revision: 2, playbackState: 'paused', currentTime: 13, updatedAt: 4000, updatedBy: 'a' });
+        expect(effectiveMediaPosition(target.mediaState, 9000)).toBe(13);
+    });
+
+    it('uses SEEK targetTime, preserves playback state, and lets the second controller win', () => {
+        const target = room({ revision: 3, playbackState: 'playing', currentTime: 10, updatedAt: 1000, updatedBy: 'a' });
+        expect(updateMediaStateFromControl(target, EVENTS.SEEK, { currentTime: 50, targetTime: 100 }, 'a', { now: 2000 })).toBe(true);
+        expect(target.mediaState).toMatchObject({ revision: 4, playbackState: 'playing', currentTime: 100, updatedBy: 'a' });
+        expect(updateMediaStateFromControl(target, EVENTS.SEEK, { targetTime: 200 }, 'b', { now: 2001 })).toBe(true);
+        expect(target.mediaState).toMatchObject({ revision: 5, currentTime: 200, updatedBy: 'b' });
+    });
+
+    it('uses an observed sender state only to establish an otherwise ambiguous first SEEK', () => {
+        const target = room();
+        expect(updateMediaStateFromControl(target, EVENTS.SEEK, { targetTime: 50 }, 'a', { now: 1000 })).toBe(false);
+        expect(updateMediaStateFromControl(target, EVENTS.SEEK, { targetTime: 50 }, 'a', { now: 1000, senderPlaybackState: 'paused' })).toBe(true);
+        expect(target.mediaState).toMatchObject({ revision: 1, playbackState: 'paused', currentTime: 50 });
+    });
+
+    it('rejects non-finite/missing controls without corruption and clamps existing protocol bounds', () => {
+        const original = { revision: 2, playbackState: 'paused', currentTime: 30, updatedAt: 1000, updatedBy: 'a' };
+        for (const payload of [{ targetTime: NaN }, { targetTime: Infinity }, { targetTime: '50' }, {}]) {
+            const target = room({ ...original });
+            expect(updateMediaStateFromControl(target, EVENTS.SEEK, payload, 'b', { now: 2000 })).toBe(false);
+            expect(target.mediaState).toEqual(original);
+        }
+        const low = room({ ...original });
+        updateMediaStateFromControl(low, EVENTS.SEEK, { targetTime: -5 }, 'b', { now: 2000 });
+        expect(low.mediaState.currentTime).toBe(0);
+        const high = room({ ...original });
+        updateMediaStateFromControl(high, EVENTS.SEEK, { targetTime: MAX_MEDIA_TIME + 5 }, 'b', { now: 2000 });
+        expect(high.mediaState.currentTime).toBe(MAX_MEDIA_TIME);
+    });
+
+    it('commits Force Sync only at execute time', () => {
+        const target = room({ revision: 4, playbackState: 'paused', currentTime: 90, updatedAt: 1000, updatedBy: 'a' });
+        expect(commitForceSyncMediaState(target, 500, 'b', 2000)).toBe(true);
+        expect(target.mediaState).toEqual({ revision: 5, playbackState: 'playing', currentTime: 500, updatedAt: 2000, updatedBy: 'b' });
+    });
+});