watchparty/server/test/partyStreamService.test.ts

// Exercises PartyStreamService's own caching logic without ever calling
// real Emby - CatalogService.negotiateStreamTarget and EmbyClient
// .proxyResponse are both mocked, same "mock the one real dependency, run
// everything else for real" approach as transcodeService.test.ts (which
// mocks CatalogService.getMediaSource for the same reason).
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import type { Request, Response } from 'express';
import { CatalogService } from '../src/services/CatalogService';
import { EmbyClient } from '../src/emby/EmbyClient';
import { MusicQueueService } from '../src/services/MusicQueueService';
import { PartyService } from '../src/services/PartyService';
import { PartyStreamService } from '../src/services/PartyStreamService';
import type { EmbySession } from '../src/types';

const FAKE_SESSION: EmbySession = {
  embyToken: 'tok-alice',
  userId: 'u-alice',
  userName: 'Alice',
  serverId: 'server1',
  createdUtc: Date.now()
};

function fakeReq(): Request {
  return { query: {} } as unknown as Request;
}

/** statusCode starts at 200, same default Express itself uses before
 *  anything calls .status() - proxyResponse (mocked in most tests here)
 *  is what would normally set it; a couple of tests set it directly to
 *  simulate a failed proxied response, matching how
 *  PartyStreamService.invalidateOnFailure actually reads it (right after
 *  EmbyClient.proxyResponse resolves, not via any other channel). */
function fakeRes(): Response {
  return { statusCode: 200 } as unknown as Response;
}

describe('PartyStreamService.getOrNegotiate', () => {
  it('negotiates once and reuses the cached target for a repeat call with the same desired options', async (t) => {
    let negotiateCount = 0;
    t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
      negotiateCount += 1;
      return { targetPath: '/Videos/movie1/master.m3u8?PlaySessionId=sess1', mediaSourceId: 'src1', playSessionId: 'sess1' };
    });

    const party = PartyService.create('u-alice', 'Alice', 'movie1', 'Un Film', '/img/movie1', null, 'video');
    const first = await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);
    const second = await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    assert.equal(negotiateCount, 1, 'the second call must reuse the cached target, not negotiate again');
    assert.equal(first, second);
    assert.deepEqual(PartyStreamService.getCachedPlaybackIds(party.id), { mediaSourceId: 'src1', playSessionId: 'sess1' }, 'cached alongside the target - see PlaystateService');

    PartyStreamService.endSession(party.id);
  });

  it('renegotiates when the host changes the audio/quality pick', async (t) => {
    let negotiateCount = 0;
    t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
      negotiateCount += 1;
      return { targetPath: `/Videos/movie2/master.m3u8?call=${negotiateCount}`, mediaSourceId: `src${negotiateCount}`, playSessionId: `sess${negotiateCount}` };
    });

    const party = PartyService.create('u-alice', 'Alice', 'movie2', 'Un Film 2', '/img/movie2', null, 'video');
    const first = await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    party.streamAudioIndex = '2';
    const second = await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    assert.equal(negotiateCount, 2);
    assert.notEqual(first, second);

    PartyStreamService.endSession(party.id);
  });

  it('renegotiates when the item changes (e.g. advanceQueue)', async (t) => {
    let negotiateCount = 0;
    t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
      negotiateCount += 1;
      return { targetPath: `/Videos/x/master.m3u8?call=${negotiateCount}`, mediaSourceId: `src${negotiateCount}`, playSessionId: `sess${negotiateCount}` };
    });

    const party = PartyService.create('u-alice', 'Alice', 'ep1', 'S01E01', '/img/ep1', [
      { itemId: 'ep1', itemName: 'S01E01', itemImageUrl: '/img/ep1' },
      { itemId: 'ep2', itemName: 'S01E02', itemImageUrl: '/img/ep2' }
    ], 'video');
    await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    PartyService.advanceQueue(party); // also calls PartyStreamService.endSession itself - see PartyService.ts
    await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    assert.equal(negotiateCount, 2, 'advancing to a new item must not reuse the old item\'s cached target');
  });

  it('music: reads the current queue track, never an audio/quality override (MusicBar has no such picker)', async (t) => {
    let capturedArgs: unknown[] = [];
    t.mock.method(CatalogService, 'negotiateStreamTarget', async (...args: unknown[]) => {
      capturedArgs = args;
      return { targetPath: '/Audio/t1/master.m3u8', mediaSourceId: 'src-t1', playSessionId: 'sess-t1' };
    });

    const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
    MusicQueueService.init(party, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 60_000_000 }, 'u-alice', 'Alice');

    await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    assert.deepEqual(capturedArgs, [FAKE_SESSION, 't1', true, '', '']);
  });

  it('music: fails cleanly (409, not a crash) when the queue is empty - nothing to stream', async () => {
    const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
    MusicQueueService.init(party, null, 'u-alice', 'Alice');

    await assert.rejects(
      () => PartyStreamService.getOrNegotiate(party, FAKE_SESSION),
      (err: Error & { status?: number }) => err.status === 409
    );
  });
});

describe('PartyStreamService failure recovery', () => {
  it('serveMasterPlaylist drops the cached session when the proxied response fails, so the next call renegotiates', async (t) => {
    let negotiateCount = 0;
    t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
      negotiateCount += 1;
      return { targetPath: `/Videos/movie3/master.m3u8?call=${negotiateCount}`, mediaSourceId: `src${negotiateCount}`, playSessionId: `sess${negotiateCount}` };
    });
    t.mock.method(EmbyClient, 'proxyResponse', async (_req: Request, res: Response) => {
      (res as unknown as { statusCode: number }).statusCode = 502; // simulate Emby having ended this session on its own
    });

    const party = PartyService.create('u-alice', 'Alice', 'movie3', 'Un Film 3', '/img/movie3', null, 'video');
    await PartyStreamService.serveMasterPlaylist(fakeReq(), fakeRes(), party, FAKE_SESSION);
    assert.equal(PartyStreamService.getCachedPlaybackIds(party.id), null, 'a failed response must drop the MediaSourceId/PlaySessionId too, not just the target path');
    await PartyStreamService.serveMasterPlaylist(fakeReq(), fakeRes(), party, FAKE_SESSION);

    assert.equal(negotiateCount, 2, 'a failed response must not be cached - the next request has to renegotiate');

    PartyStreamService.endSession(party.id);
  });

  it('a successful proxied response keeps the cache - the next call does not renegotiate', async (t) => {
    let negotiateCount = 0;
    t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
      negotiateCount += 1;
      return { targetPath: '/Videos/movie4/master.m3u8', mediaSourceId: 'src-movie4', playSessionId: 'sess-movie4' };
    });
    t.mock.method(EmbyClient, 'proxyResponse', async () => {}); // res.statusCode stays 200, the fakeRes() default

    const party = PartyService.create('u-alice', 'Alice', 'movie4', 'Un Film 4', '/img/movie4', null, 'video');
    await PartyStreamService.serveMasterPlaylist(fakeReq(), fakeRes(), party, FAKE_SESSION);
    await PartyStreamService.serveMasterPlaylist(fakeReq(), fakeRes(), party, FAKE_SESSION);

    assert.equal(negotiateCount, 1);

    PartyStreamService.endSession(party.id);
  });
});

describe('PartyStreamService lifecycle', () => {
  it('endSession is a harmless no-op for a party that never negotiated a stream', () => {
    assert.doesNotThrow(() => PartyStreamService.endSession('no-such-party'));
  });

  it('getCachedPlaybackIds is null for a party that never negotiated a stream - never negotiates one itself', () => {
    assert.equal(PartyStreamService.getCachedPlaybackIds('no-such-party'), null);
  });

  it('reapIdleSessions ends sessions past the idle cutoff, leaves a fresh one alone', async (t) => {
    let negotiateCount = 0;
    t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
      negotiateCount += 1;
      return { targetPath: `/Videos/x/master.m3u8?call=${negotiateCount}`, mediaSourceId: `src${negotiateCount}`, playSessionId: `sess${negotiateCount}` };
    });

    const party = PartyService.create('u-eve', 'Eve', 'movie5', 'Un Film 5', '/img/movie5', null, 'video');
    await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);

    // A generous timeout must not touch a session just negotiated.
    PartyStreamService.reapIdleSessions(30 * 60 * 1000);
    await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);
    assert.equal(negotiateCount, 1, 'a fresh session must not be reaped');

    // A *negative* timeout pushes the cutoff strictly into the future,
    // so it reaps regardless of exactly when lastAccessUtc was set -
    // unlike 0, which computes cutoff = Date.now() at reap time and can
    // tie with a lastAccessUtc set within the same millisecond (this
    // failed intermittently with 0 - Date.now()'s own resolution isn't
    // fine-grained enough to guarantee cutoff is strictly later). The
    // simplest way to exercise the sweep itself without a lastAccessUtc
    // setter (getOrNegotiate only ever returns the target string, not
    // the session object, unlike TranscodeService's own equivalent method).
    PartyStreamService.reapIdleSessions(-1);
    await PartyStreamService.getOrNegotiate(party, FAKE_SESSION);
    assert.equal(negotiateCount, 2, 'a past-the-future cutoff must reap even a just-touched session');

    PartyStreamService.endSession(party.id);
  });
});