KoalaSync/extension/offline-media-intent.test.mjs

import { describe, expect, it } from 'vitest';
import { EVENTS, MAX_MEDIA_TIME } from '../shared/constants.js';
import { canonicalMediaStateFromRoomData } from './canonical-media-state.js';
import {
    discardQueuedMediaIntents,
    drainQueuedBatch,
    enqueueQueuedEvent,
    hasQueuedMediaIntent,
    materializeMediaIntent,
    maxQueuedSequence,
    normalizePersistedEventQueue,
    queuedEntryWireCount,
    queuedMediaIntentCount,
    queuedWireCount,
    reconcileQueuedRoomIntent,
    reserveLatestMediaIntentSequence
} from './offline-media-intent.js';

const roomId = 'room-a';
const media = (event, data, queue = []) => enqueueQueuedEvent(queue, event, data, { roomId }).queue;

function reserve(queue, sequence) {
    return reserveLatestMediaIntentSequence(queue, roomId, sequence).queue;
}

describe('offline media intent coalescing', () => {
    it('normalizes an empty or roomless persisted queue to empty', () => {
        expect(normalizePersistedEventQueue([], roomId)).toEqual([]);
        expect(normalizePersistedEventQueue([{ event: EVENTS.PLAY, data: { seq: 1 } }], null)).toEqual([]);
    });

    it('creates a PLAY intent and reserves ordered legacy SEEK + PLAY frames', () => {
        let queue = media(EVENTS.PLAY, { currentTime: 10, seq: 5, actionTimestamp: 100 });
        queue = reserve(queue, 6);
        expect(queue).toHaveLength(1);
        expect(materializeMediaIntent(queue[0])).toEqual([
            { event: EVENTS.SEEK, data: { seq: 5, currentTime: 10, targetTime: 10 } },
            { event: EVENTS.PLAY, data: { seq: 6, actionTimestamp: 100, currentTime: 10 } }
        ]);
    });

    it('collapses repeated PLAY and SEEK while preserving final state and monotonic sequences', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
        queue = media(EVENTS.PLAY, { currentTime: 20, seq: 3 }, queue);
        queue = media(EVENTS.SEEK, { targetTime: 30, seq: 4 }, queue);
        expect(queue).toHaveLength(1);
        expect(queue[0].intent).toMatchObject({
            playbackState: 'playing',
            currentTime: 30,
            latestEvent: EVENTS.SEEK,
            previousSeq: 3,
            latestSeq: 4,
            sourceEventCount: 3
        });
        expect(materializeMediaIntent(queue[0]).map(frame => frame.data.seq)).toEqual([3, 4]);
    });

    it('drops a regressing queued sequence instead of emitting stale ordering', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 10 }), 11);
        const result = enqueueQueuedEvent(queue, EVENTS.PAUSE, { currentTime: 20, seq: 9 }, { roomId });
        expect(result.droppedStale).toBe(1);
        expect(result.queue).toEqual(queue);
        expect(maxQueuedSequence(result.queue)).toBe(11);
    });

    it('merges PLAY -> SEEK -> PAUSE into paused at the final position', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 500, seq: 1 }), 2);
        queue = media(EVENTS.SEEK, { targetTime: 600, seq: 3 }, queue);
        queue = media(EVENTS.PAUSE, { currentTime: 605, seq: 4 }, queue);
        expect(materializeMediaIntent(queue[0])).toEqual([
            { event: EVENTS.SEEK, data: { seq: 3, currentTime: 605, targetTime: 605 } },
            { event: EVENTS.PAUSE, data: { seq: 4, currentTime: 605 } }
        ]);
    });

    it('merges PAUSE -> SEEK -> PLAY into playing at the final position', () => {
        let queue = reserve(media(EVENTS.PAUSE, { currentTime: 100, seq: 10 }), 11);
        queue = media(EVENTS.SEEK, { targetTime: 200, seq: 12 }, queue);
        queue = media(EVENTS.PLAY, { currentTime: 205, seq: 13 }, queue);
        expect(materializeMediaIntent(queue[0]).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PLAY]);
        expect(queue[0].intent).toMatchObject({ playbackState: 'playing', currentTime: 205 });
    });

    it('clamps finite positions and rejects a SEEK with no trustworthy position', () => {
        expect(media(EVENTS.SEEK, { targetTime: NaN, seq: 1 })).toEqual([]);
        expect(media(EVENTS.SEEK, { targetTime: -50, seq: 1 })[0].intent.currentTime).toBe(0);
        expect(media(EVENTS.SEEK, { targetTime: MAX_MEDIA_TIME + 50, seq: 1 })[0].intent.currentTime).toBe(MAX_MEDIA_TIME);
    });

    it('keeps title metadata bounded and never retains arbitrary payload fields', () => {
        const queue = media(EVENTS.PAUSE, {
            currentTime: 10,
            seq: 1,
            mediaTitle: 'x'.repeat(200),
            password: 'secret',
            chatKey: 'secret'
        });
        expect(queue[0].intent.mediaTitle).toHaveLength(100);
        expect(JSON.stringify(queue)).not.toContain('password');
        expect(JSON.stringify(queue)).not.toContain('chatKey');
    });

    it('keeps a thousand-event media burst at one logical entry', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 0, seq: 1 }), 2);
        for (let index = 1; index <= 1000; index++) {
            queue = media(EVENTS.SEEK, { targetTime: index, seq: index + 2 }, queue);
        }
        queue = media(EVENTS.PAUSE, { currentTime: 1000, seq: 1003 }, queue);
        expect(queue).toHaveLength(1);
        expect(queue[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 1000, sourceEventCount: 1002 });
    });

    it('treats retained coordination events as barriers', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 50, seq: 3 }, { roomId }).queue;
        queue = media(EVENTS.SEEK, { targetTime: 100, seq: 4 }, queue);
        queue = media(EVENTS.PAUSE, { currentTime: 120, seq: 5 }, queue);
        expect(queue).toHaveLength(3);
        expect(queue[0].kind).toBe('media-intent');
        expect(queue[1].event).toBe(EVENTS.FORCE_SYNC_PREPARE);
        expect(queue[2].kind).toBe('media-intent');
        expect(queue[2].intent).toMatchObject({ playbackState: 'paused', currentTime: 120 });
    });

    it('does not persist stale liveness and command ACK frames as ordering barriers', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
        for (const event of [EVENTS.PING, EVENTS.PONG, EVENTS.PEER_STATUS, EVENTS.EVENT_ACK]) {
            const result = enqueueQueuedEvent(queue, event, { seq: 99 }, { roomId });
            expect(result.droppedStale).toBe(1);
            queue = result.queue;
        }
        queue = media(EVENTS.PAUSE, { currentTime: 20, seq: 3 }, queue);
        expect(queue).toHaveLength(1);
        expect(queue[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 20 });
    });

    it('preserves the bounded logical queue cap', () => {
        let queue = [];
        for (let index = 0; index < 60; index++) {
            queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { index }, { roomId }).queue;
        }
        expect(queue).toHaveLength(50);
        expect(queue[0].data.index).toBe(10);
        expect(queue.at(-1).data.index).toBe(59);
    });

    it('migrates old raw media entries without crossing transactional barriers', () => {
        const restored = normalizePersistedEventQueue([
            { event: EVENTS.PLAY, data: { currentTime: 10, seq: 1 } },
            { event: EVENTS.SEEK, data: { targetTime: 20, seq: 2 } },
            { event: EVENTS.PAUSE, data: { currentTime: 25, seq: 3 } },
            { event: EVENTS.EPISODE_LOBBY, data: { expectedTitle: 'S01E02' } },
            { event: EVENTS.PLAY, data: { currentTime: 30, seq: 4 } }
        ], roomId);
        expect(restored).toHaveLength(3);
        expect(restored[0].intent).toMatchObject({ playbackState: 'paused', currentTime: 25 });
        expect(restored[1].event).toBe(EVENTS.EPISODE_LOBBY);
        expect(restored[2].intent).toMatchObject({ playbackState: 'playing', currentTime: 30 });
        expect(maxQueuedSequence(restored)).toBe(4);
    });

    it('enforces the cap while restoring consecutive persisted media intents', () => {
        const persisted = [10, 20, 30].map((currentTime, index) => ({
            kind: 'media-intent',
            roomId,
            intent: {
                playbackState: 'paused',
                currentTime,
                latestEvent: EVENTS.PAUSE,
                previousSeq: null,
                latestSeq: index + 1,
                actionTimestamp: index + 1,
                mediaTitle: null,
                sourceEventCount: 1
            }
        }));
        const restored = normalizePersistedEventQueue(persisted, roomId, 2);
        expect(restored).toHaveLength(2);
        expect(restored.map(entry => entry.intent.currentTime)).toEqual([20, 30]);
    });

    it('drops room-scoped barriers from another room and unknown persisted events', () => {
        const restored = normalizePersistedEventQueue([
            { kind: 'event', roomId: 'room-b', event: EVENTS.FORCE_SYNC_EXECUTE, data: { seq: 1 } },
            { kind: 'event', roomId, event: 'unexpected_event', data: { secret: 'nope' } },
            { kind: 'event', roomId, event: EVENTS.EPISODE_READY, data: { seq: 2 } }
        ], roomId);
        expect(restored).toEqual([{
            kind: 'event', roomId, event: EVENTS.EPISODE_READY, data: { seq: 2 }
        }]);
    });

    it('discards stale-room intent without affecting the new room', () => {
        const queue = reserve(media(EVENTS.PAUSE, { currentTime: 500, seq: 1 }), 2);
        expect(hasQueuedMediaIntent(queue, roomId)).toBe(true);
        expect(discardQueuedMediaIntents(queue, roomId)).toEqual([]);
        expect(hasQueuedMediaIntent(queue, 'room-b')).toBe(false);
    });

    it('drops queued shared intent after controller role loss so canonical recovery can proceed', () => {
        let queue = reserve(media(EVENTS.PAUSE, { currentTime: 1200, seq: 1 }), 2);
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 1300, seq: 3 }, { roomId }).queue;
        const result = reconcileQueuedRoomIntent(queue, { roomId, canControl: false });
        expect(result.queue).toEqual([]);
        expect(result.discarded).toBe(2);
        expect(result.hasPendingLocalIntent).toBe(false);
    });

    it('keeps an active Episode Lobby authoritative over queued media and Force Sync', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 100, seq: 1 }), 2);
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 500, seq: 3 }, { roomId }).queue;
        queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { title: 'S01E02' }, { roomId }).queue;
        const result = reconcileQueuedRoomIntent(queue, { roomId, activeLobby: true });
        expect(result.queue).toEqual([{
            kind: 'event',
            roomId,
            event: EVENTS.EPISODE_READY,
            data: { title: 'S01E02' }
        }]);
        expect(result.hasPendingLocalIntent).toBe(false);
    });

    it('drops stale queued Episode Lobby coordination when ROOM_DATA has an authoritative lobby', () => {
        let queue = enqueueQueuedEvent([], EVENTS.EPISODE_LOBBY, { expectedTitle: 'S02E01' }, { roomId }).queue;
        queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { title: 'S02E01' }, { roomId }).queue;
        queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_LOBBY_CANCEL, {}, { roomId }).queue;
        const result = reconcileQueuedRoomIntent(queue, {
            roomId,
            activeLobby: true,
            authoritativeLobby: true
        });
        expect(result.queue).toEqual([]);
        expect(result.discarded).toBe(3);
    });

    it('does not let intentional solo mode retain future room-driving intent', () => {
        const queue = reserve(media(EVENTS.SEEK, { targetTime: 600, playbackState: 'paused', seq: 1 }), 2);
        const result = reconcileQueuedRoomIntent(queue, { roomId, desynced: true });
        expect(result.queue).toEqual([]);
        expect(result.hasPendingLocalIntent).toBe(false);
    });

    it('materializes legacy events normally when an old relay has no media-state capability', () => {
        expect(canonicalMediaStateFromRoomData({
            roomId,
            capabilities: ['host-control', 'chat-v1']
        })).toEqual({ status: 'unsupported', mediaState: null });
        const queue = reserve(media(EVENTS.PAUSE, { currentTime: 75, seq: 20 }), 21);
        expect(materializeMediaIntent(queue[0]).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PAUSE]);
        expect(materializeMediaIntent(queue[0]).every(frame =>
            frame.data.mediaState === undefined && frame.data.revision === undefined
        )).toBe(true);
    });
});

describe('offline media intent drain', () => {
    it('counts actual wire frames and never splits an intent at a batch boundary', async () => {
        let queue = enqueueQueuedEvent([], EVENTS.EPISODE_READY, { seq: 1 }, { roomId }).queue;
        queue = reserve(media(EVENTS.PAUSE, { currentTime: 50, seq: 2 }, queue), 3);
        const sent = [];
        const first = await drainQueuedBatch(queue, {
            roomId,
            maxWireEvents: 2,
            sendFrame: async frame => { sent.push(frame); return true; }
        });
        expect(first.sentWireEvents).toBe(1);
        expect(first.queue).toHaveLength(1);
        expect(sent.map(frame => frame.event)).toEqual([EVENTS.EPISODE_READY]);

        const second = await drainQueuedBatch(first.queue, {
            roomId,
            maxWireEvents: 2,
            sendFrame: async frame => { sent.push(frame); return true; }
        });
        expect(second.sentWireEvents).toBe(2);
        expect(second.queue).toEqual([]);
        expect(sent.slice(1).map(frame => frame.event)).toEqual([EVENTS.SEEK, EVENTS.PAUSE]);
    });

    it('retains the whole logical intent after a partial send failure', async () => {
        const queue = reserve(media(EVENTS.PLAY, { currentTime: 90, seq: 5, actionTimestamp: 500 }), 6);
        let calls = 0;
        const result = await drainQueuedBatch(queue, {
            roomId,
            maxWireEvents: 10,
            sendFrame: async () => ++calls === 1
        });
        expect(result.status).toBe('send_failed');
        expect(result.sentWireEvents).toBe(1);
        expect(result.queue).toEqual(queue);
        expect(materializeMediaIntent(result.queue[0]).map(frame => frame.data.seq)).toEqual([5, 6]);
        expect(materializeMediaIntent(result.queue[0]).map(frame => frame.data.actionTimestamp))
            .toEqual([undefined, 500]);
    });

    it('drops stale-room intent during drain and preserves unrelated events', async () => {
        let queue = reserve(media(EVENTS.PAUSE, { currentTime: 40, seq: 1 }), 2);
        queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { seq: 3 }, { roomId }).queue;
        const sent = [];
        const result = await drainQueuedBatch(queue, {
            roomId: 'room-b',
            maxWireEvents: 10,
            sendFrame: async frame => { sent.push(frame); return true; }
        });
        expect(result.droppedStaleIntents).toBe(2);
        expect(sent).toEqual([]);
    });

    it('repairs regressing sequences across malformed persisted intent entries', () => {
        const restored = normalizePersistedEventQueue([
            {
                kind: 'media-intent',
                roomId,
                intent: {
                    playbackState: 'playing', currentTime: 10, latestEvent: EVENTS.PLAY,
                    previousSeq: 99, latestSeq: 100, actionTimestamp: 1, mediaTitle: null, sourceEventCount: 1
                }
            },
            {
                kind: 'media-intent',
                roomId,
                intent: {
                    playbackState: 'paused', currentTime: 20, latestEvent: EVENTS.PAUSE,
                    previousSeq: 49, latestSeq: 50, actionTimestamp: 2, mediaTitle: null, sourceEventCount: 1
                }
            }
        ], roomId);
        expect(materializeMediaIntent(restored[0]).map(frame => frame.data.seq)).toEqual([99, 100]);
        expect(materializeMediaIntent(restored[1]).map(frame => frame.data.seq)).toEqual([101, 102]);
        expect(maxQueuedSequence(restored)).toBe(102);
    });

    it('preserves a valid legacy single-frame intent during sequence repair', () => {
        const restored = normalizePersistedEventQueue([{
            kind: 'media-intent',
            roomId,
            intent: {
                playbackState: 'paused', currentTime: 20, latestEvent: EVENTS.PAUSE,
                previousSeq: null, latestSeq: 50, actionTimestamp: 2, mediaTitle: null, sourceEventCount: 1
            }
        }], roomId);
        expect(materializeMediaIntent(restored[0])).toEqual([{
            event: EVENTS.PAUSE,
            data: { seq: 50, actionTimestamp: 2, currentTime: 20 }
        }]);
    });

    it('evicts a complete Force Sync transaction instead of orphaning EXECUTE at the cap', () => {
        let queue = enqueueQueuedEvent([], EVENTS.FORCE_SYNC_PREPARE, { targetTime: 100, seq: 1 }, { roomId }).queue;
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_EXECUTE, { seq: 2 }, { roomId }).queue;
        for (let index = 0; index < 49; index++) {
            queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_ACK, { seq: index + 3 }, { roomId }).queue;
        }
        expect(queue).toHaveLength(49);
        expect(queue.some(entry => entry.event === EVENTS.FORCE_SYNC_PREPARE)).toBe(false);
        expect(queue.some(entry => entry.event === EVENTS.FORCE_SYNC_EXECUTE)).toBe(false);
    });

    it('keeps adjacent Force Sync PREPARE and EXECUTE in the same replay batch', async () => {
        let queue = [];
        for (let index = 0; index < 9; index++) {
            queue = enqueueQueuedEvent(queue, EVENTS.EPISODE_READY, { seq: index + 1 }, { roomId }).queue;
        }
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { targetTime: 100, seq: 10 }, { roomId }).queue;
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_EXECUTE, { seq: 11 }, { roomId }).queue;
        const sent = [];
        const result = await drainQueuedBatch(queue, {
            roomId,
            maxWireEvents: 10,
            sendFrame: async frame => { sent.push(frame.event); return true; }
        });
        expect(sent).toEqual(Array(9).fill(EVENTS.EPISODE_READY));
        expect(result.queue.map(entry => entry.event)).toEqual([
            EVENTS.FORCE_SYNC_PREPARE,
            EVENTS.FORCE_SYNC_EXECUTE
        ]);
    });

    it('retains the full Force Sync transaction if EXECUTE replay fails', async () => {
        let queue = enqueueQueuedEvent([], EVENTS.FORCE_SYNC_PREPARE, { targetTime: 100, seq: 1 }, { roomId }).queue;
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_EXECUTE, { seq: 2 }, { roomId }).queue;
        const result = await drainQueuedBatch(queue, {
            roomId,
            maxWireEvents: 10,
            sendFrame: async frame => frame.event !== EVENTS.FORCE_SYNC_EXECUTE
        });
        expect(result.status).toBe('send_failed');
        expect(result.sentWireEvents).toBe(1);
        expect(result.queue).toEqual(queue);
    });

    it('reports logical and actual-wire queue sizes separately', () => {
        let queue = reserve(media(EVENTS.PLAY, { currentTime: 10, seq: 1 }), 2);
        queue = enqueueQueuedEvent(queue, EVENTS.FORCE_SYNC_PREPARE, { seq: 3 }, { roomId }).queue;
        expect(queuedMediaIntentCount(queue, roomId)).toBe(1);
        expect(queuedEntryWireCount(queue[0])).toBe(2);
        expect(queuedWireCount(queue)).toBe(3);
    });
});