kyuno/watchparty
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

Replace PlayingItems/Progress with direct UserData writes
All checks were successful
Publier l'image Docker / test (push) Successful in 21s
Details
Publier l'image Docker / publish (push) Successful in 13s
Details

Browse source 
Report: "still don't see my Emby profile updated - watched items
should be marked read, partially-watched ones should show in Resume".
After two rounds of fixing PlayingItems/Progress's request shape
(missing MediaSourceId, then missing PlaySessionId), this pointed at
something more fundamental: this app deliberately never calls the
matching PlayingItems "Start"/"Stop" - a documented v1 scope cut - and
a bare Progress report with no corresponding Start may simply never
reach the resume-position/watched-history data real clients read,
regardless of how correctly-shaped the request is.

Confirmed directly against the real server that this data lives
somewhere else entirely, trivially readable/writable with no playback
session required at all: every item Emby returns already carries the
requesting user's own UserData embedded (PlaybackPositionTicks/Played/
PlayCount/IsFavorite/...) - the exact thing Continue Watching and the
watched checkmark actually read - via a plain
POST /Users/{userId}/Items/{itemId}/UserData.

Switched entirely to that instead of layering a third fix onto
PlayingItems:
- PlaystateService.updatePosition (replacing reportProgress, same
  ~20s sweep in index.ts) updates just PlaybackPositionTicks.
- New PlaystateService.markPlayed sets Played: true and resets
  PlaybackPositionTicks to 0 - called once per connected member (with
  a token) when the party moves past an item for good:
  PartyService.advanceQueue (next episode) and
  MusicQueueService.playNext (a track naturally finishing, or a
  majority vote-skip - both go through this one method). Addresses
  the other half of the report ("watched items marked as watched").
- Both go through one private fetch-merge-write helper: GET the item
  first, apply only the field(s) being changed to a copy of its
  current UserData, POST the whole result back. Emby's spec documents
  the body as the complete UserItemDataDto with nothing indicating
  merge-not-replace semantics, and this is the one write in the whole
  app where guessing wrong risks actually erasing a user's own real
  data (a favorite, a rating) - so never a bare partial POST.

Reverted the MediaSourceId/PlaySessionId plumbing rounds 1-2 added
(CatalogService.negotiateStreamTarget back to returning a bare target
path, PartyStreamService no longer caches either) - nothing needs them
now that UserData doesn't require an active PlayingItems session.

Not yet confirmed end to end against a real server (no safe way to
fire a real, state-mutating test call without one).

Tests: PlaystateService's fetch-merge-write behavior in detail (a
favorite/rating/play-count survives a position-only update byte for
byte, a Played-only update doesn't disturb IsFavorite, a failed read
skips the write entirely, an item with no UserData yet starts from an
empty object), its request/outcome logging including that the token
never leaks into a log line, and that advanceQueue/playNext call
markPlayed for the outgoing item only, only for connected+tokened
members. catalogService/partyStreamService tests reverted to their
simpler pre-existing shape. Full typecheck/test/build green
(107/107 tests). docs/architecture.md and README.md updated.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Nicolas 2026-09-14 23:53:59 +02:00
parent 939beb6fa9
commit b167e321e7
14 changed files with 538 additions and 348 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
README.md
View file

@ -108,15 +108,21 @@ attempt at a Blu-ray-style subtitle as the thing to double-check here.
unconfirmed (the ffmpeg direct-stream URL, the Playlists endpoint, the
Emby-playstate sync endpoint) were all checked directly against Emby's
own real OpenAPI spec. Two matched exactly. The third - playstate sync
- took two rounds to actually fix: `PositionTicks`/`IsPaused` were
being sent as a JSON body when Emby expects query parameters, alongside
a required `MediaSourceId` that wasn't being sent at all (round 1); a
real request logged after that shipped still failed, this time with
Emby's own "Value cannot be null (Parameter 'key')" - traced to a
missing `PlaySessionId` (round 2, from new logging added specifically
to catch exactly this). Both fixed, though round 2 hasn't yet been
confirmed against a real server the way round 1's own bad request was
- see docs/architecture.md's "Sync watch-party progress back to Emby"
- took three rounds: two request-shape bugs against Emby's
`PlayingItems/Progress` check-in endpoint (a missing `MediaSourceId`,
then a missing `PlaySessionId`), then a further report ("still don't
see this in my Emby profile") that questioned the whole approach - a
bare `Progress` report with no matching "Start" call may just never
reach the resume-position/watched-history data real clients read,
regardless of request shape. Switched entirely to a plain
`POST /Users/{userId}/Items/{itemId}/UserData` instead - confirmed
directly against the real server that this is exactly the data model
"Continue Watching" and the watched checkmark read, updated with no
playback session required at all - and added marking an item Played
(the other half of the same report) when the party moves past it.
Not yet confirmed end to end against a real server (no safe way to
fire a real, state-mutating test call without one) - see
docs/architecture.md's "Sync watch-party progress back to Emby"
section for the full story. The one-shared-stream-per-party work is a
different kind of unverified: its own Emby API calls were already live,
confirmed code, just newly shared instead of per-viewer - what's untested is behavior
@ -272,7 +278,8 @@ Then put it behind your existing reverse proxy like everything else
- [x] Browse Emby playlists and launch one as a pre-queued party (video
queue or the participative music queue) — see docs/architecture.md
- [x] Sync watch-party progress back to Emby's own PlaystateService, per
member's own account — see docs/architecture.md
member's own account, including marking an item watched once the
party moves past it — see docs/architecture.md
- [x] One shared Emby stream per party instead of one per viewer — see
docs/architecture.md
- [x] Text subtitle choice joins burn-in as a host-only, party-wide pick

204
docs/architecture.md
View file

@ -1989,106 +1989,122 @@ thrown - this runs on a plain `setInterval` (~20s, alongside
with no request to fail and no business ever disrupting real playback
over a background sync hiccup.
**TODO(verify) resolved - and the original guess was wrong.** Checked
directly against Emby's own real OpenAPI spec (swagger.emby.media,
pointed at a real server) after a "watch status" bug report suggested
this wasn't actually working: `POST /Users/{UserId}/PlayingItems/{Id}
/Progress` does exist at that exact path, but the spec declares
`PositionTicks`/`IsPaused` as *query string* parameters, not body
fields - and a third parameter, `MediaSourceId`, is *required* and
wasn't being sent at all. The request body schema this endpoint does
list (`Api.OnPlaybackProgress`) covers a handful of unrelated fields
(`PlaylistIndex`/`Shuffle`/`SleepTimerMode`/...), none of them these -
so the original guess was sending `PositionTicks`/`IsPaused` somewhere
Emby never reads them, while omitting a parameter Emby requires just
to accept the call at all. In practice this almost certainly meant
every progress report was failing outright, silently (best-effort by
design - see above - so nothing user-visible ever broke loudly; only
`[watchparty:emby]`'s own status+body capture would have shown it,
and no one had gone looking until this bug report).
**Round 1, TODO(verify) resolved - and the original guess was wrong.**
Checked directly against Emby's own real OpenAPI spec
(swagger.emby.media, pointed at a real server) after a "watch status"
bug report suggested this wasn't actually working: `POST /Users
/{UserId}/PlayingItems/{Id}/Progress` does exist at that exact path,
but the spec declares `PositionTicks`/`IsPaused` as *query string*
parameters, not body fields - and a third parameter, `MediaSourceId`,
is *required* and wasn't being sent at all. Fixed: sent as query
parameters instead, `MediaSourceId` resolved from the same
`PlaybackInfo` negotiation `CatalogService.negotiateStreamTarget`
already does for the party's shared stream. While checking this, the
app's other two Emby-API assumptions (the ffmpeg direct-stream URL,
the `/Playlists/{Id}/Items` endpoint) were re-checked against the same
real spec too - both already correct.
Fixed: `reportProgress` now sends `MediaSourceId`/`PositionTicks`/
`IsPaused` as query parameters, no body at all. `MediaSourceId` comes
from `PartyStreamService`'s own cache - `CatalogService
.negotiateStreamTarget` already resolves a real one for every party's
shared stream negotiation (see "One shared Emby stream per party"
below), so `PartyStreamService` now caches it alongside the target
path and exposes it read-only via a new `getCachedMediaSourceId
(partyId)`. Deliberately passive - it never triggers a negotiation of
its own - so a party where no one has started playback yet this
session simply has nothing to report; `index.ts`'s sweep skips it, the
same way it already skips a party with nothing playing at all, rather
than paying for an extra PlaybackInfo round trip on every single tick
just to get a value the party's already-negotiated stream produces for
free.
**Round 2, caught immediately from new logging added specifically to
catch exactly this** (every outgoing request + its outcome, under
`[watchparty:playstate]` - see `Logger.playstate`'s own comment for why
this needed to be enough on its own to answer "is this actually
reaching Emby, and with what"): a real request logged after round 1
shipped still came back `400`, now `Value cannot be null. (Parameter
'key')` - a bare .NET `Dictionary` null-key exception, and *not* the
earlier problem (`MediaSourceId` was present and well-formed).
`PlaySessionId` was the standout candidate - how Emby/Jellyfin clients
normally correlate a report with the specific playback session it
belongs to - and sending it (resolved the same way `MediaSourceId`
was) fixed this particular error.
While checking this, the app's other two Emby-API assumptions (the
ffmpeg direct-stream URL in `TranscodeService.buildSourceUrl`, and the
`/Playlists/{Id}/Items` endpoint in `CatalogService`) were re-checked
against the same real spec - both match with no missing required
parameter, consistent with them already having been tried for real
successfully.
**Round 3: a further report questioned the whole approach, not just
its request shape** - "still don't see this in my Emby profile /
Continue Watching" - and that turned out to be the right instinct.
This app deliberately never calls the matching `POST .../PlayingItems
/{Id}` ("Start") or `DELETE .../PlayingItems/{Id}` ("Stop") - a
documented v1 scope cut (see the original framing of this section,
still true) - and a bare `Progress` report with no corresponding
"Start" may simply never reach the resume-position/watched-history
data real clients read, regardless of how correctly-shaped the request
is by that point. Confirmed directly against the real server that this
data lives somewhere else entirely, trivially reachable without any
session lifecycle at all: every item Emby returns already carries the
requesting user's own `UserData` embedded
(`PlaybackPositionTicks`/`Played`/`PlayCount`/`IsFavorite`/...) - the
exact thing "Continue Watching" and the watched checkmark actually
read - and a plain `POST /Users/{UserId}/Items/{ItemId}/UserData`
updates it directly.
**Made the sweep's own traffic visible, not just its failures** - asked
for directly, once fixing the request shape raised the obvious
follow-up question ("is it actually being sent correctly now?").
`reportProgress` now logs the exact outgoing request (`-> POST
/Users/{userId}/PlayingItems/{itemId}/Progress?MediaSourceId=...&...`)
right before sending it, and its outcome right after (`<- 200 ok` /
`<- 400 FAILED` / `<- threw ...`), all under the same
`[watchparty:playstate]` prefix - so `docker logs | grep
watchparty:playstate` answers "what did this send, and did Emby accept
it" on its own, without cross-referencing `EmbyClient`'s own generic
`[watchparty:emby]` failure log (which still fires too, independently,
with the fuller error body). Safe to log the full path here, unlike
`Logger.stream`'s own deliberately-partial logging - this query string
is only ever `MediaSourceId`/`PlaySessionId`/`PositionTicks`/`IsPaused`,
never a token; the real `embyToken` travels as a header, never in the
URL, and the tests confirm it never ends up in these lines either.
**Switched to that entirely** rather than layering a third fix onto
`PlayingItems` - `PlaystateService.updatePosition` (periodic, ~20s,
replacing `reportProgress`) and a new `PlaystateService.markPlayed`
(see below) both go through one private `updateUserData` helper.
Simpler than the endpoint it replaces (one JSON body, no query-string
parameter shape to get wrong) and more direct (writes the exact data
model the UI reads, rather than a live-session side-channel whose
effect on that data model was the very thing in doubt). The
`MediaSourceId`/`PlaySessionId` plumbing rounds 1-2 added
(`CatalogService.negotiateStreamTarget` returning them,
`PartyStreamService` caching them) was removed along with it -
`negotiateStreamTarget` is back to returning a bare target path, since
nothing needs those two values for anything else.
**Round 2, caught immediately from that new logging**: a real request
logged after the fix above shipped still came back `400`, now `Value
cannot be null. (Parameter 'key')` - a bare .NET `Dictionary` null-key
exception, and *not* the earlier problem (the logged request's
`MediaSourceId` was present and well-formed). This exact message had
already burned this app once before, in `CatalogService
.negotiateStreamTarget`'s hand-built fallback URL (see its own
TODO(verify) comment) - diagnosed there as Emby indexing some internal
state by a parameter the request hadn't supplied. `PlaySessionId` was
the standout remaining candidate: it's how Emby/Jellyfin clients
normally correlate a progress report with the specific playback/
transcode session it belongs to, and evidently something in Emby's real
handler keys a lookup on it even though the OpenAPI spec doesn't mark
it required. Now sent too, resolved the exact same way `MediaSourceId`
already was: `CatalogService.negotiateStreamTarget` already reads
`info.PlaySessionId` from the very same `PlaybackInfo` response (it
already used it in the hand-built fallback URL) - its return type
(`NegotiatedStreamTarget`) just needed to surface it, so
`PartyStreamService` could cache it alongside `mediaSourceId` and
expose both together from one accessor,
`getCachedPlaybackIds(partyId)`, replacing the round-1 `getCached
MediaSourceId`. Not independently confirmed against a real server yet
(no way to safely fire a real, state-mutating test call without one) -
flagged here the same way every other real-server-dependent guess in
this app is, but grounded in this app's own prior, confirmed diagnosis
of the identical error message rather than a fresh guess.
**Fetch-merge-write, deliberately, never a bare partial POST.** Emby's
OpenAPI spec documents the `UserData` endpoint's body as the *complete*
`UserItemDataDto`, with nothing suggesting a partial object merges
rather than replaces - and this is the one write in the whole app where
guessing wrong risks actually *erasing* a user's own real data (a
favorite, a rating), not just silently doing nothing the way the
`Progress`-endpoint bugs did. So every call to `updateUserData` first
`GET`s the item (confirmed directly against the real server that Emby
always returns it with the requesting user's current `UserData`
embedded), applies just the one or two fields being changed to a copy
of exactly that, and `POST`s the whole result back - correct regardless
of which way the real merge-vs-replace semantics actually go, at the
cost of one extra read before every write.
**Marking items Played, addressing the other half of the same
report** ("les éléments visionnés marqué comme lu") - not attempted in
any earlier round. `PlaystateService.markPlayed` sets `Played: true`
and resets `PlaybackPositionTicks` to `0`, called once per currently-
connected member (with a token) at the exact moment the party moves
past an item for good: `PartyService.advanceQueue` (next episode, for
the *outgoing* item, captured before the party's own state moves on)
and `MusicQueueService.playNext` (a track naturally finishing, or a
majority vote-skip - both go through this one method, so both get the
same treatment; `music.current` captured as `outgoing` before it's
reassigned). Fire-and-forget, same as the periodic position sync - a
member's bad token or a transient Emby hiccup must never block the
party from actually advancing. This is a deliberate heuristic ("the
party moved on from it" = "done with it"), not a real "did they watch
the whole thing" check - this app has no reliable way to know that
beyond what it already tracks, and most real streaming clients treat a
skip as a play too. Precision beyond that (e.g. only marking Played
past some percentage watched) would need every queued item's own
duration tracked everywhere it's queued, which nothing currently does -
left as a documented gap, not built speculatively.
Verified: full typecheck/test/build on both projects. Unit tests cover
`getCurrentPlaybackState` (video reads the host, music reads the shared
queue, both branches, plus the empty-queue null case),
`setMemberEmbyToken`, `PlaystateService.reportProgress`'s corrected
query-string request shape including `PlaySessionId` (and that a thrown
failure never propagates), its own request/outcome logging (success, a
non-ok status, and a thrown error each produce a distinct, greppable
line, and the token itself never appears in one), and
`PartyStreamService.getCachedPlaybackIds`'s
cache-hit/cache-miss/dropped-on-failure behavior for both ids together.
`EmbyClient.fetch`'s optional body-encoding capability stays generic
plumbing (its own test no longer claims PlaystateService as the reason
it exists) - no current call site needs it any more, but it costs
nothing to keep for whatever future caller does. No client changes at
all - this is a pure server-side background process with nothing for
the browser to render or react to.
`getCurrentPlaybackState`, `setMemberEmbyToken`,
`PlaystateService.updatePosition`/`markPlayed`'s fetch-merge-write
behavior in detail - a favorite/rating/play-count this app never asked
to change survives a position update byte for byte, a `Played`-only
update via `markPlayed` doesn't disturb `IsFavorite`, a failed *read*
skips the write entirely rather than posting incomplete data, an item
with no `UserData` at all yet starts cleanly from an empty object - its
own request/outcome logging (success, a non-ok status on either call,
and a thrown error each produce a distinct, greppable line, and the
token itself never appears in one), and that
`PartyService.advanceQueue`/`MusicQueueService.playNext` call
`markPlayed` for the *outgoing* item only, only for members who are
both connected and have a known token, mirroring the exact same guard
`index.ts`'s periodic sweep already uses. `EmbyClient.fetch`'s optional
body-encoding capability (added for this feature's very first attempt)
stays generic plumbing - still exercised by this feature's own
`UserData` POST, so unlike after round 2 it's not just kept around for
a hypothetical future caller. No client changes at all - this is a pure
server-side background/event-driven process with nothing for the
browser to render or react to.
## One shared Emby stream per party (not one per viewer)

21
server/src/Logger.ts
View file

@ -66,16 +66,17 @@ export class Logger {
console.log(`[watchparty:transcode] ${new Date().toISOString()} ${message}`);
}
/** The periodic Emby-playstate sweep (index.ts / PlaystateService) -
* logs every outgoing Progress request (method + path + query - see
* PlaystateService.reportProgress) and its outcome, not just
* failures: this is a best-effort background sync with nothing else
* watching it, so `docker logs | grep watchparty:playstate` needs to
* be enough on its own to see exactly what was sent and whether Emby
* accepted it, without cross-referencing `emby` above. Safe to log
* the full path here (unlike `stream` above) - the query string is
* only ever MediaSourceId/PositionTicks/IsPaused, never a token; the
* real embyToken travels as a header, not a URL. */
/** The party watch-state sync (index.ts's periodic sweep, and
* PartyService.advanceQueue/MusicQueueService.playNext's own item-
* transition calls - see PlaystateService) - logs every outgoing
* request (method + path, and the POST body for the UserData write)
* and its outcome, not just failures: this is a best-effort
* background sync with nothing else watching it, so `docker logs |
* grep watchparty:playstate` needs to be enough on its own to see
* exactly what was sent and whether Emby accepted it, without cross-
* referencing `emby` above. Safe to log in full here (unlike `stream`
* above) - the real embyToken travels as a header, never logged, and
* PlaystateService's own tests confirm it never leaks into these lines. */
static playstate(message: string): void {
console.log(`[watchparty:playstate] ${new Date().toISOString()} ${message}`);
}

24
server/src/index.ts
View file

@ -503,27 +503,23 @@ setInterval(() => TranscodeService.reapIdleSessions(30 * 60 * 1000), 5 * 60 * 10
// destroy branch (and an item/track change) ending it immediately.
setInterval(() => PartyStreamService.reapIdleSessions(30 * 60 * 1000), 5 * 60 * 1000);
// Pushes every connected member's own progress to Emby, under their own
// account - see docs/architecture.md's "Sync watch-party progress back
// to Emby" section for the full design and its deliberate v1 scope
// (periodic Progress reports only, no explicit start/stop). Best-effort:
// PlaystateService.reportProgress swallows its own failures, so one bad
// Pushes every connected member's own resume position to Emby (their own
// UserData for the item, under their own account) - see
// docs/architecture.md's "Sync watch-party progress back to Emby"
// section for the full history of why this writes UserData directly
// rather than Emby's PlayingItems/Progress check-in endpoint. Best-effort:
// PlaystateService.updatePosition swallows its own failures, so one bad
// token or a transient Emby hiccup never stops the rest of this sweep.
// MediaSourceId/PlaySessionId come from PartyStreamService's own cache,
// read-only (getCachedPlaybackIds never negotiates) - both matter to the
// real Progress endpoint in practice (see PlaystateService.ts), and
// skipping a party entirely when nothing's cached yet (no one has hit
// play this session) is simpler and cheaper than this sweep ever
// negotiating a stream of its own just to get them.
// Marking an item Played (once the party moves past it) is handled
// separately, at the source of that transition - see
// PartyService.advanceQueue and MusicQueueService.playNext.
setInterval(() => {
for (const party of PartyService.listActive()) {
const state = PartyService.getCurrentPlaybackState(party);
if (!state) continue; // nothing currently playing (e.g. an empty music queue)
const playbackIds = PartyStreamService.getCachedPlaybackIds(party.id);
if (!playbackIds) continue; // nothing negotiated yet this session - Progress needs it
for (const member of party.members.values()) {
if (!member.connection || !member.embyToken) continue;
PlaystateService.reportProgress(member.embyToken, member.userId, state.itemId, playbackIds.mediaSourceId, playbackIds.playSessionId, state.positionTicks, state.isPaused);
PlaystateService.updatePosition(member.embyToken, member.userId, state.itemId, state.positionTicks);
}
}
}, 20 * 1000);

28
server/src/services/CatalogService.ts
View file

@ -13,13 +13,6 @@ import type { EmbyItem, EmbySession, ItemDto, LibraryDto, MediaType, PlaybackOpt
// actually stream. "Voir sur Emby" always works though.
const WATCH_PARTY_TYPES = new Set(['Movie', 'Episode', 'Video', 'MusicVideo', 'Audio']);
/** negotiateStreamTarget's return shape - see its own comment. */
export interface NegotiatedStreamTarget {
targetPath: string;
mediaSourceId: string;
playSessionId: string;
}
// Emby's /Views can return all sorts of things beyond actual media
// libraries - boxset collections, live TV, photos, books, a mixed-content
// folder with no CollectionType at all - none of which this app has any
@ -310,23 +303,8 @@ class CatalogServiceClass {
* encoding option this app has no visibility into, sidesteps the
* entire class of failure.
*
* Returns `mediaSourceId`/`playSessionId` alongside the target path,
* not just the path - PartyStreamService caches all three, and
* PlaystateService's own Progress report needs both too: MediaSourceId
* is a required query parameter there (confirmed against Emby's real
* OpenAPI spec), and PlaySessionId - though only optional per that
* same spec - turned out to matter in practice too: a real report sent
* without one hit Emby's own "Value cannot be null (Parameter 'key')"
* (a bare .NET Dictionary null-key exception) - the exact error this
* app already ran into once before, in the stream-negotiation fallback
* URL above, for the same underlying reason (see that TODO(verify)
* comment) - so Emby's real Progress handler evidently keys some
* internal lookup (an active transcode/session entry) on it even
* though the spec doesn't mark it required. Resolving both here, once
* per negotiation, means the playstate sweep never has to make its own
* extra PlaybackInfo round trip just to get them.
*/
async negotiateStreamTarget(session: EmbySession, itemId: string, isAudio: boolean, audioStreamIndex: string, maxStreamingBitrate: string): Promise<NegotiatedStreamTarget> {
async negotiateStreamTarget(session: EmbySession, itemId: string, isAudio: boolean, audioStreamIndex: string, maxStreamingBitrate: string): Promise<string> {
// Applied to the negotiation itself, not just the final URL - a pick
// could otherwise silently do nothing if Emby's negotiation didn't
// apply the query-string override the same way its own web client's
@ -355,7 +333,7 @@ class CatalogServiceClass {
const params = new URLSearchParams(urlQuery || '');
this.applyPlaybackOverrides(params, audioStreamIndex, maxStreamingBitrate);
this.logStreamRequest(itemId, 'negotiated', params);
return { targetPath: `${urlPath}?${params}`, mediaSourceId: source.Id || '', playSessionId: info.PlaySessionId || '' };
return `${urlPath}?${params}`;
}
const params = new URLSearchParams({
@ -367,7 +345,7 @@ class CatalogServiceClass {
});
this.applyPlaybackOverrides(params, audioStreamIndex, maxStreamingBitrate);
this.logStreamRequest(itemId, 'fallback', params);
return { targetPath: `/${isAudio ? 'Audio' : 'Videos'}/${itemId}/master.m3u8?${params}`, mediaSourceId: source.Id || '', playSessionId: info.PlaySessionId || '' };
return `/${isAudio ? 'Audio' : 'Videos'}/${itemId}/master.m3u8?${params}`;
}
/**

17
server/src/services/MusicQueueService.ts
View file

@ -2,6 +2,7 @@
import { Logger } from '../Logger';
import { PartyStreamService } from './PartyStreamService';
import { PlaystateService } from './PlaystateService';
import type { AppServer, MusicCurrentPublic, MusicPublicState, MusicTrack, MusicTrackInput, Party } from '../types';
/**
@ -164,12 +165,28 @@ class MusicQueueServiceClass {
private playNext(party: Party): void {
const music = party.music!;
const outgoing = music.current; // the track finishing/being skipped, if any - see the markPlayed loop below
this.clearTimer(party);
music.skipVotes.clear();
// Whatever was previously negotiated (see PartyStreamService) was for
// the outgoing track - never valid for whatever plays next, whether
// that's a new track or nothing at all.
PartyStreamService.endSession(party.id);
// The queue moving past `outgoing` - naturally finishing, or a
// majority vote-skip, either way - is this app's own signal that
// it's "done" with that track, for every member connected right now.
// Fire-and-forget, same reasoning as PartyService.advanceQueue's own
// identical loop for video - see PlaystateService.markPlayed. `if
// (outgoing)` skips this on the very first call for a party (the
// queue was empty, nothing outgoing to mark).
if (outgoing) {
for (const member of party.members.values()) {
if (!member.connection || !member.embyToken) continue;
PlaystateService.markPlayed(member.embyToken, member.userId, outgoing.track.itemId);
}
}
const next = music.queue.shift();
if (!next) {
music.current = null;

14
server/src/services/PartyService.ts
View file

@ -5,6 +5,7 @@ import { Logger } from '../Logger';
import { MusicQueueService } from './MusicQueueService';
import { TranscodeService } from './TranscodeService';
import { PartyStreamService } from './PartyStreamService';
import { PlaystateService } from './PlaystateService';
import type { LeaveResult, MediaType, Party, PartyMember, PartyPublicState, QueueItem } from '../types';
/**
@ -193,6 +194,7 @@ class PartyServiceClass {
if (party.queueIndex >= party.queue.length - 1) {
return false;
}
const outgoingItemId = party.itemId; // whatever was just playing - see the markPlayed loop below
party.queueIndex += 1;
const next = party.queue[party.queueIndex]!;
party.itemId = next.itemId;
@ -203,6 +205,18 @@ class PartyServiceClass {
member.isPaused = false;
member.lastUpdatedUtc = Date.now();
}
// The party moving on from `outgoingItemId` is this app's own signal
// that it's "done" - see PlaystateService.markPlayed's own comment
// for why that's a heuristic, not a real "watched it all" check, and
// why every currently-connected member gets their own account
// updated, not just the host. Fire-and-forget, same as the periodic
// position sync in index.ts - a member's bad token or a transient
// Emby hiccup here must never block the party from actually moving
// on to the next item.
for (const member of party.members.values()) {
if (!member.connection || !member.embyToken) continue;
PlaystateService.markPlayed(member.embyToken, member.userId, outgoingItemId);
}
// A new item's own MediaStreams indices have nothing to do with the
// old one's - carrying a burn-in or text subtitle pick across would
// show whatever track happens to share that index on the new item,

32
server/src/services/PartyStreamService.ts
View file

@ -11,15 +11,6 @@ interface StreamSession {
audioStreamIndex: string;
maxStreamingBitrate: string;
targetPath: string;
/** Alongside targetPath, from the same negotiation - see
* CatalogService.negotiateStreamTarget's own comment. Surfaced
* read-only, together, via getCachedPlaybackIds() below for
* PlaystateService's own Progress report, which needs both too (see
* that method's own comment for why) but must never trigger a
* negotiation of its own just to get them - see index.ts's periodic
* sweep. */
mediaSourceId: string;
playSessionId: string;
lastAccessUtc: number;
}
@ -67,30 +58,11 @@ class PartyStreamServiceClass {
return existing.targetPath;
}
const { targetPath, mediaSourceId, playSessionId } = await CatalogService.negotiateStreamTarget(session, itemId, isAudio, audioStreamIndex, maxStreamingBitrate);
this.sessions.set(party.id, { itemId, audioStreamIndex, maxStreamingBitrate, targetPath, mediaSourceId, playSessionId, lastAccessUtc: Date.now() });
const targetPath = await CatalogService.negotiateStreamTarget(session, itemId, isAudio, audioStreamIndex, maxStreamingBitrate);
this.sessions.set(party.id, { itemId, audioStreamIndex, maxStreamingBitrate, targetPath, lastAccessUtc: Date.now() });
return targetPath;
}
/**
* Read-only: the MediaSourceId/PlaySessionId behind this party's
* cached stream, or null if nothing has been negotiated yet this
* session (e.g. the party just started and no one has hit play).
* Deliberately does NOT call getOrNegotiate/touch lastAccessUtc - this
* is a passive lookup for index.ts's periodic playstate sweep, not
* real streaming traffic, and must never itself keep an otherwise-idle
* session alive (see reapIdleSessions) or trigger a fresh Emby
* negotiation on its own timer tick. Returns both together, from the
* same cache entry, rather than two separate lookups - they always
* come from the same negotiation, and PlaystateService needs both at
* once anyway (see its own comment for why PlaySessionId turned out to
* matter too, not just MediaSourceId).
*/
getCachedPlaybackIds(partyId: string): { mediaSourceId: string; playSessionId: string } | null {
const session = this.sessions.get(partyId);
return session ? { mediaSourceId: session.mediaSourceId, playSessionId: session.playSessionId } : null;
}
/** GET /api/catalog/party-stream/:partyId/master.m3u8 - the caller
* (index.ts) has already checked the requester is a member of `party`. */
async serveMasterPlaylist(req: Request, res: Response, party: Party, session: EmbySession): Promise<void> {

182
server/src/services/PlaystateService.ts
View file

@ -3,78 +3,136 @@
import { EmbyClient } from '../emby/EmbyClient';
import { Logger } from '../Logger';
/** The subset of Emby's own UserItemDataDto this app ever reads/writes -
* see updateUserData's own comment for why this is always read back in
* full before being changed, never posted as a bare partial object. */
interface EmbyUserData {
PlaybackPositionTicks?: number;
PlayCount?: number;
IsFavorite?: boolean;
Played?: boolean;
PlayedPercentage?: number;
LastPlayedDate?: string;
Rating?: number;
}
/**
* Reports a party member's own playback progress back to Emby, under
* Writes a party member's own watch/listen state back to Emby, under
* *their own* account - the one piece of this app that ever writes
* anything to Emby rather than just reading/proxying it (see
* docs/architecture.md's "Sync watch-party progress back to Emby"
* section for the full design and its deliberate v1 scope). Called from
* index.ts's periodic sweep, never from a request/response cycle - see
* that file for who actually decides *when* and *with what* to call
* this.
* section for the fuller history - this class went through two rounds
* of chasing the wrong endpoint before landing here).
*
* Emby's own "PlaystateService" (the controller name in both Emby's and
* Jellyfin's server source) is what exposes this route.
*
* Endpoint/param shape CONFIRMED against Emby's own real OpenAPI spec
* (swagger.emby.media, pointed at a real server) - the original guess
* here was wrong in a way that plausibly broke this entire feature:
* PositionTicks/IsPaused were sent as a JSON body, but Emby declares
* them (and a REQUIRED MediaSourceId this app wasn't sending at all) as
* *query string* parameters instead - the request body schema
* (Api.OnPlaybackProgress) covers a handful of unrelated fields
* (PlaylistIndex/Shuffle/SleepTimerMode/...), none of them these. A
* missing required query parameter almost certainly meant every one of
* these calls was failing outright.
*
* Round 2, from a real request logged after that fix shipped: still a
* 400, now `Value cannot be null. (Parameter 'key')` - a bare .NET
* `Dictionary` null-key exception, and *not* the earlier problem
* (MediaSourceId was present and well-formed in the logged request).
* This exact message already burned this app once before, in
* CatalogService.negotiateStreamTarget's hand-built fallback URL (see
* its own TODO(verify) comment) - diagnosed there as Emby's server
* indexing some internal state by a param this app hadn't supplied.
* `PlaySessionId` is the standout candidate: it's how Emby/Jellyfin
* clients normally correlate a progress report with the specific
* playback/transcode session it belongs to, and evidently something on
* Emby's side keys a lookup on it even though the OpenAPI spec doesn't
* mark it required. Now sent too - see PartyStreamService
* .getCachedPlaybackIds, which resolves it the exact same way
* MediaSourceId already was (both come from the one PlaybackInfo
* negotiation CatalogService.negotiateStreamTarget already does for the
* party's shared stream).
* v1 tried Emby's `PlaystateService` "PlayingItems" check-in endpoints
* (`POST .../Progress`) - the standard shape a live media client uses to
* report ongoing playback. Two real request-shape bugs there were found
* and fixed (a missing required `MediaSourceId`, then a missing
* `PlaySessionId` that crashed Emby's own handler with "Value cannot be
* null"), but a further report ("still don't see this in my Emby
* profile / Continue Watching") pointed at something more fundamental:
* this app deliberately never calls the matching `POST .../PlayingItems
* /{Id}` ("Start") or `DELETE .../PlayingItems/{Id}` ("Stop") - a
* documented, deliberate v1 scope cut (see the same architecture.md
* section) - and Progress reports with no corresponding "Start" may
* simply never reach the resume-position/watched-history data real
* clients read, regardless of how correctly-shaped the request is.
* Confirmed against the real server that this data lives somewhere
* else entirely and is trivially readable/writable directly: every
* item Emby returns already carries an embedded `UserData` object
* (`PlaybackPositionTicks`/`Played`/`PlayCount`/`IsFavorite`/...) - the
* exact thing "Continue Watching" and the watched checkmark actually
* read - and Emby exposes a plain `POST /Users/{UserId}/Items/{ItemId}
* /UserData` to update it directly, no active playback session
* required at all. Switched to that entirely instead of layering a
* third fix onto the PlayingItems approach - simpler (one JSON body,
* no query-string parameter shape to get wrong) and more direct (writes
* the exact data model the UI reads, rather than a live-session
* side-channel whose effect on that data model was the very thing in
* doubt).
*/
class PlaystateServiceClass {
/**
* Best-effort: any failure (Emby unreachable, a stale/expired token,
* a 4xx/5xx) is caught and logged here, never thrown - this runs on a
* timer with no request to fail, and one member's bad token or one
* transient Emby hiccup must never stop the rest of the sweep, let
* alone touch real playback. `mediaSourceId`/`playSessionId` come from
* PartyStreamService's own cache (see index.ts's sweep) - resolving
* fresh ones here would mean an extra PlaybackInfo round trip on every
* tick, for values the party's already-negotiated stream already
* produced for free.
*
* Every call logs the exact outgoing request (method, path, query -
* everything actually sent, minus the token itself, which travels as
* a header, never in this) and its outcome - see Logger.playstate's
* own comment for why this needs to be enough on its own, greppable,
* to answer "is this actually reaching Emby, and with what" without
* digging through EmbyClient's own generic failure log.
* Called periodically (~20s, see index.ts's sweep) while a member is
* connected to a party with something playing: updates just their own
* resume position for `itemId`, leaving `Played`/`IsFavorite`/`Rating`/
* everything else exactly as Emby already has it - see updateUserData
* below for why this never touches those fields, and never posts a
* bare partial object to get there.
*/
async reportProgress(embyToken: string, userId: string, itemId: string, mediaSourceId: string, playSessionId: string, positionTicks: number, isPaused: boolean): Promise<void> {
const params = new URLSearchParams({
MediaSourceId: mediaSourceId,
PlaySessionId: playSessionId,
PositionTicks: String(Math.round(positionTicks)),
IsPaused: String(isPaused)
});
const path = `/Users/${userId}/PlayingItems/${itemId}/Progress?${params}`;
Logger.playstate(`-> POST ${path}`);
async updatePosition(embyToken: string, userId: string, itemId: string, positionTicks: number): Promise<void> {
await this.updateUserData(embyToken, userId, itemId, (current) => ({
...current,
PlaybackPositionTicks: Math.round(positionTicks)
}));
}
/**
* Called once per member when the party moves past `itemId` for good -
* video's `PartyService.advanceQueue` (next episode) and music's
* `MusicQueueService.playNext` (a track naturally finishing, or a
* majority vote-skip) both call this for the *outgoing* item, for
* every member connected at that moment. Marks it Played and clears
* the resume position, the same way finishing something in Emby's own
* web client leaves it. A deliberate heuristic, not a real "did they
* actually watch the whole thing" check: this app has no reliable way
* to know that beyond "the party moved on from it" - which real
* streaming clients broadly treat as good enough too (a skip still
* counts as a play in most of them). Precision beyond that - e.g. only
* marking Played past some percentage watched - would need this app
* to also track each item's own duration everywhere it's queued,
* which nothing currently does; left as a documented gap rather than
* built speculatively.
*/
async markPlayed(embyToken: string, userId: string, itemId: string): Promise<void> {
await this.updateUserData(embyToken, userId, itemId, (current) => ({
...current,
PlaybackPositionTicks: 0,
Played: true
}));
}
/**
* Fetch-merge-write, always - deliberately never a bare partial POST.
* Emby's own OpenAPI spec documents this endpoint's body as the
* *complete* UserItemDataDto, with nothing in it suggesting a partial
* object merges rather than replaces - and this is the one write in
* this entire app where guessing wrong risks actually *erasing* a
* user's own real data (a favorite, a rating), not just silently
* doing nothing the way the earlier Progress-endpoint bugs did. So:
* GET the item first (confirmed directly against the real server that
* Emby always returns it with the requesting user's current UserData
* embedded), apply `mutate` to a copy of exactly that, POST the whole
* result back - correct regardless of which way the real merge-vs-
* replace semantics actually go, at the cost of one extra read before
* every write.
*
* Best-effort throughout: any failure (Emby unreachable, a stale/
* expired token, a 4xx/5xx on either call) is caught and logged here,
* never thrown - `updatePosition` runs on a timer with no request to
* fail, and `markPlayed` runs from inside a queue-advance/track-skip
* path that must never let one member's bad token or a transient Emby
* hiccup block the rest of the party from moving on. Every call logs
* the exact outgoing requests and their outcome - see Logger
* .playstate's own comment for why this needs to be enough on its
* own, greppable, to answer "is this actually reaching Emby, and with
* what" without digging through EmbyClient's own generic failure log.
*/
private async updateUserData(embyToken: string, userId: string, itemId: string, mutate: (current: EmbyUserData) => EmbyUserData): Promise<void> {
const itemPath = `/Users/${userId}/Items/${itemId}`;
Logger.playstate(`-> GET ${itemPath} (reading current UserData before merging - see PlaystateService.updateUserData)`);
try {
const res = await EmbyClient.fetch(path, embyToken, { method: 'POST' });
const itemRes = await EmbyClient.fetch(itemPath, embyToken);
if (!itemRes.ok) {
Logger.playstate(`<- ${itemRes.status} FAILED userId=${userId} itemId=${itemId} (could not read current UserData, update skipped)`);
return;
}
const item = (await itemRes.json()) as { UserData?: EmbyUserData };
const updated = mutate(item.UserData || {});
const updatePath = `/Users/${userId}/Items/${itemId}/UserData`;
Logger.playstate(`-> POST ${updatePath} body=${JSON.stringify(updated)}`);
const res = await EmbyClient.fetch(updatePath, embyToken, { method: 'POST', body: updated });
Logger.playstate(`<- ${res.status} ${res.ok ? 'ok' : 'FAILED'} userId=${userId} itemId=${itemId}`);
} catch (err) {
Logger.playstate(`<- threw userId=${userId} itemId=${itemId}: ${(err as Error).message}`);

26
server/test/catalogService.test.ts
View file

@ -47,40 +47,36 @@ describe('CatalogService', () => {
PlaySessionId: 'sess1'
}));
const { targetPath, mediaSourceId, playSessionId } = await CatalogService.negotiateStreamTarget(session, 'item1', false, '2', '');
const target = await CatalogService.negotiateStreamTarget(session, 'item1', false, '2', '');
assert.ok(targetPath.includes('AudioStreamIndex=2'));
assert.ok(!targetPath.includes('Subtitle'), 'this method never touches subtitle params at all');
assert.equal(mediaSourceId, 'src1', 'PlaystateService\'s own Progress report needs this too - see PartyStreamService');
assert.equal(playSessionId, 'sess1', 'so does this - see PlaystateService\'s own "Value cannot be null" comment');
assert.ok(target.includes('AudioStreamIndex=2'));
assert.ok(!target.includes('Subtitle'), 'this method never touches subtitle params at all');
});
it('negotiateStreamTarget\'s hand-built fallback picks /Audio/ (and drops VideoCodec) when isAudio is true', async (t) => {
t.mock.method(EmbyClient, 'fetch', async () => jsonResponse({ MediaSources: [{ Id: 'src2' }], PlaySessionId: 'sess2' }));
const { targetPath } = await CatalogService.negotiateStreamTarget(session, 'track1', true, '', '');
const target = await CatalogService.negotiateStreamTarget(session, 'track1', true, '', '');
assert.ok(targetPath.startsWith('/Audio/track1/master.m3u8'));
assert.ok(!targetPath.includes('VideoCodec'));
assert.ok(target.startsWith('/Audio/track1/master.m3u8'));
assert.ok(!target.includes('VideoCodec'));
});
it('negotiateStreamTarget\'s hand-built fallback picks /Videos/ (with VideoCodec) when isAudio is false', async (t) => {
t.mock.method(EmbyClient, 'fetch', async () => jsonResponse({ MediaSources: [{ Id: 'src3' }], PlaySessionId: 'sess3' }));
const { targetPath, mediaSourceId, playSessionId } = await CatalogService.negotiateStreamTarget(session, 'movie1', false, '', '');
const target = await CatalogService.negotiateStreamTarget(session, 'movie1', false, '', '');
assert.ok(targetPath.startsWith('/Videos/movie1/master.m3u8'));
assert.ok(targetPath.includes('VideoCodec=h264'));
assert.equal(mediaSourceId, 'src3');
assert.equal(playSessionId, 'sess3', 'the hand-built fallback branch must surface this too, not just the negotiated-TranscodingUrl one');
assert.ok(target.startsWith('/Videos/movie1/master.m3u8'));
assert.ok(target.includes('VideoCodec=h264'));
});
it('negotiateStreamTarget applies a quality (MaxStreamingBitrate) pick the same way as an audio pick', async (t) => {
t.mock.method(EmbyClient, 'fetch', async () => jsonResponse({ MediaSources: [{ Id: 'src4' }], PlaySessionId: 'sess4' }));
const { targetPath } = await CatalogService.negotiateStreamTarget(session, 'movie2', false, '', '4000000');
const target = await CatalogService.negotiateStreamTarget(session, 'movie2', false, '', '4000000');
assert.ok(targetPath.includes('MaxStreamingBitrate=4000000'));
assert.ok(target.includes('MaxStreamingBitrate=4000000'));
});
it('negotiateStreamTarget fails cleanly (not a crash) when Emby has no media source for this item', async (t) => {

59
server/test/musicQueueService.test.ts
View file

@ -9,7 +9,8 @@ import { describe, it, mock } from 'node:test';
import assert from 'node:assert/strict';
import { setTimeout as sleep } from 'node:timers/promises';
import { MusicQueueService } from '../src/services/MusicQueueService';
import type { MusicState, Party } from '../src/types';
import { PlaystateService } from '../src/services/PlaystateService';
import type { AppSocket, Party, PartyMember, MusicState } from '../src/types';
function fakeParty(id: string): Party {
return {
@ -32,6 +33,26 @@ function fakeParty(id: string): Party {
};
}
/** A connected member with a known token - the two things
* playNext's own markPlayed loop (and index.ts's periodic sweep) both
* gate on. Not used by most tests in this file at all: `fakeParty`'s
* own `members` map starts empty, and MusicQueueService's methods never
* populate it themselves (that's PartyService.join's job) - so unless a
* test explicitly adds one via this, playNext's markPlayed loop simply
* has nothing to iterate, same as an empty party in production. */
function fakeMember(userId: string, userName: string, embyToken: string): PartyMember {
return {
userId,
userName,
connection: {} as unknown as AppSocket,
deviceId: null,
embyToken,
isPaused: false,
positionTicks: 0,
lastUpdatedUtc: Date.now()
};
}
function music(party: Party): MusicState {
return party.music as MusicState;
}
@ -68,6 +89,42 @@ describe('MusicQueueService', () => {
assert.equal(music(party).skipVotes.size, 0, 'skipping clears the tally for the new track');
});
it('a skip marks the OUTGOING track Played for every connected member with a token, under their own account', (t) => {
const calls: [string, string, string][] = [];
t.mock.method(PlaystateService, 'markPlayed', async (embyToken: string, userId: string, itemId: string) => {
calls.push([embyToken, userId, itemId]);
});
const party = fakeParty('p3b');
party.members.set('u-alice', fakeMember('u-alice', 'Alice', 'tok-alice'));
party.members.set('u-bob', fakeMember('u-bob', 'Bob', 'tok-bob'));
MusicQueueService.init(party, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 10_000_000 }, 'u-alice', 'Alice');
MusicQueueService.addTrack(party, { itemId: 't2', itemName: 'Track Two', itemImageUrl: '/img/t2', durationTicks: 30_000_000 }, 'u-bob', 'Bob');
assert.equal(calls.length, 0, 'starting the very first track has nothing outgoing to mark');
// connectedMemberCount=2 needs a strict majority (2) to actually
// skip - same arithmetic as the "needs a strict majority" test above.
MusicQueueService.voteSkip(party, 'u-alice', 'Alice', 2);
assert.equal(calls.length, 0, 'one vote of two is not a majority yet - nothing outgoing to mark');
MusicQueueService.voteSkip(party, 'u-bob', 'Bob', 2);
assert.deepEqual(calls.sort(), [['tok-alice', 'u-alice', 't1'], ['tok-bob', 'u-bob', 't1']].sort(), 'marks t1 (outgoing), not t2 (just started)');
});
it('never marks a member Played if they have no connection or no token yet', (t) => {
let called = false;
t.mock.method(PlaystateService, 'markPlayed', async () => { called = true; });
const party = fakeParty('p3c');
party.members.set('u-alice', { ...fakeMember('u-alice', 'Alice', 'tok-alice'), connection: null });
MusicQueueService.init(party, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 10_000_000 }, 'u-alice', 'Alice');
MusicQueueService.addTrack(party, { itemId: 't2', itemName: 'Track Two', itemImageUrl: '/img/t2', durationTicks: 30_000_000 }, 'u-bob', 'Bob');
MusicQueueService.voteSkip(party, 'u-alice', 'Alice', 1);
assert.equal(called, false);
});
it('pause freezes elapsed time; resume continues it from where it froze', async () => {
const party = fakeParty('p4');
MusicQueueService.init(party, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 600_000_000 }, 'u-alice', 'Alice');

42
server/test/partyService.test.ts
View file

@ -2,6 +2,8 @@ import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import { PartyService } from '../src/services/PartyService';
import { MusicQueueService } from '../src/services/MusicQueueService';
import { PlaystateService } from '../src/services/PlaystateService';
import type { AppSocket } from '../src/types';
describe('PartyService', () => {
it('create() sets mediaType and seeds the host as the first member', () => {
@ -45,6 +47,46 @@ describe('PartyService', () => {
assert.equal(party.textSubtitleIndex, null);
});
it('advanceQueue marks the outgoing item Played for every connected member with a token, under their own account - see PlaystateService', (t) => {
const calls: [string, string, string][] = [];
t.mock.method(PlaystateService, 'markPlayed', async (embyToken: string, userId: string, itemId: string) => {
calls.push([embyToken, userId, itemId]);
});
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');
PartyService.join(party, 'u-bob', 'Bob');
PartyService.setMemberEmbyToken(party, 'u-alice', 'tok-alice');
PartyService.setMemberEmbyToken(party, 'u-bob', 'tok-bob');
// A real socket connection isn't needed for anything else this test
// checks - just enough to pass advanceQueue's own "connected" guard,
// same one the periodic sweep in index.ts already uses.
party.members.get('u-alice')!.connection = {} as unknown as AppSocket;
party.members.get('u-bob')!.connection = {} as unknown as AppSocket;
PartyService.advanceQueue(party);
assert.deepEqual(calls.sort(), [['tok-alice', 'u-alice', 'ep1'], ['tok-bob', 'u-bob', 'ep1']].sort(), 'marks the OUTGOING item (ep1), not the new one just moved to');
});
it('advanceQueue never marks a member Played if they have no connection or no token yet', (t) => {
let called = false;
t.mock.method(PlaystateService, 'markPlayed', async () => { called = true; });
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');
// Host never got a connection/token set at all in this test - same as
// a REST-created party nobody has actually joined a socket to yet.
PartyService.advanceQueue(party);
assert.equal(called, false);
});
it('leave(): a non-host leaving neither destroys the party nor hands off', () => {
const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
PartyService.join(party, 'u-bob', 'Bob');

20
server/test/partyStreamService.test.ts
View file

@ -40,7 +40,7 @@ describe('PartyStreamService.getOrNegotiate', () => {
let negotiateCount = 0;
t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
negotiateCount += 1;
return { targetPath: '/Videos/movie1/master.m3u8?PlaySessionId=sess1', mediaSourceId: 'src1', playSessionId: 'sess1' };
return '/Videos/movie1/master.m3u8?PlaySessionId=sess1';
});
const party = PartyService.create('u-alice', 'Alice', 'movie1', 'Un Film', '/img/movie1', null, 'video');
@ -49,7 +49,6 @@ describe('PartyStreamService.getOrNegotiate', () => {
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);
});
@ -58,7 +57,7 @@ describe('PartyStreamService.getOrNegotiate', () => {
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}` };
return `/Videos/movie2/master.m3u8?call=${negotiateCount}`;
});
const party = PartyService.create('u-alice', 'Alice', 'movie2', 'Un Film 2', '/img/movie2', null, 'video');
@ -77,7 +76,7 @@ describe('PartyStreamService.getOrNegotiate', () => {
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}` };
return `/Videos/x/master.m3u8?call=${negotiateCount}`;
});
const party = PartyService.create('u-alice', 'Alice', 'ep1', 'S01E01', '/img/ep1', [
@ -96,7 +95,7 @@ describe('PartyStreamService.getOrNegotiate', () => {
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' };
return '/Audio/t1/master.m3u8';
});
const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
@ -123,7 +122,7 @@ describe('PartyStreamService failure recovery', () => {
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}` };
return `/Videos/movie3/master.m3u8?call=${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
@ -131,7 +130,6 @@ describe('PartyStreamService failure recovery', () => {
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');
@ -143,7 +141,7 @@ describe('PartyStreamService failure recovery', () => {
let negotiateCount = 0;
t.mock.method(CatalogService, 'negotiateStreamTarget', async () => {
negotiateCount += 1;
return { targetPath: '/Videos/movie4/master.m3u8', mediaSourceId: 'src-movie4', playSessionId: 'sess-movie4' };
return '/Videos/movie4/master.m3u8';
});
t.mock.method(EmbyClient, 'proxyResponse', async () => {}); // res.statusCode stays 200, the fakeRes() default
@ -162,15 +160,11 @@ describe('PartyStreamService lifecycle', () => {
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}` };
return `/Videos/x/master.m3u8?call=${negotiateCount}`;
});
const party = PartyService.create('u-eve', 'Eve', 'movie5', 'Un Film 5', '/img/movie5', null, 'video');

190
server/test/playstateService.test.ts
View file

@ -1,116 +1,158 @@
// PlaystateService.reportProgress - mocks EmbyClient.fetch (not the
// global fetch it wraps), same isolation style as catalogService.test.ts,
// since this is about what PlaystateService itself sends, not
// EmbyClient's own plumbing (see embyClient.test.ts for that).
// PlaystateService.updatePosition/markPlayed - mocks EmbyClient.fetch
// (not the global fetch it wraps), same isolation style as
// catalogService.test.ts, since this is about what PlaystateService
// itself sends/merges, not EmbyClient's own plumbing (see
// embyClient.test.ts for that).
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import { PlaystateService } from '../src/services/PlaystateService';
import { EmbyClient } from '../src/emby/EmbyClient';
describe('PlaystateService.reportProgress', () => {
it('POSTs to this user\'s own PlayingItems/Progress with their own token, params in the query string not a body', async (t) => {
let capturedPath = '';
let capturedToken = '';
let capturedOptions: unknown;
/** What a real `GET /Users/{userId}/Items/{itemId}` comes back with -
* confirmed directly against a real server that the requesting user's
* own UserData is always embedded right on the item like this. */
function itemResponse(userData: Record<string, unknown> | undefined): Response {
return new Response(JSON.stringify({ Id: 'item1', UserData: userData }), { status: 200 });
}
describe('PlaystateService.updatePosition', () => {
it('GETs the item, then POSTs UserData with only PlaybackPositionTicks changed - everything else survives untouched', async (t) => {
const capturedCalls: { path: string; token: string; options: unknown }[] = [];
t.mock.method(EmbyClient, 'fetch', async (path: string, token: string, options: unknown) => {
capturedPath = path;
capturedToken = token;
capturedOptions = options;
capturedCalls.push({ path, token, options });
if (capturedCalls.length === 1) {
return itemResponse({ PlaybackPositionTicks: 0, Played: false, IsFavorite: true, Rating: 9.5, PlayCount: 3 });
}
return new Response('{}', { status: 200 });
});
await PlaystateService.reportProgress('tok-alice', 'u-alice', 'item1', 'src1', 'sess1', 12_345_678, false);
await PlaystateService.updatePosition('tok-alice', 'u-alice', 'item1', 12_345_678);
// Query params, not a JSON body - confirmed against Emby's real
// OpenAPI spec (see PlaystateService.ts's own comment): the
// original body-based guess was missing a required MediaSourceId
// entirely and would have failed outright. PlaySessionId came next
// (round 2 - see the same comment): optional per the spec, but a
// real request without one still hit Emby's own "Value cannot be
// null (Parameter 'key')".
const [path, query] = capturedPath.split('?');
assert.equal(path, '/Users/u-alice/PlayingItems/item1/Progress');
assert.equal(new URLSearchParams(query).get('MediaSourceId'), 'src1');
assert.equal(new URLSearchParams(query).get('PlaySessionId'), 'sess1');
assert.equal(new URLSearchParams(query).get('PositionTicks'), '12345678');
assert.equal(new URLSearchParams(query).get('IsPaused'), 'false');
assert.equal(capturedToken, 'tok-alice');
assert.deepEqual(capturedOptions, { method: 'POST' }, 'no body at all - the real body schema (Api.OnPlaybackProgress) has none of these fields');
assert.equal(capturedCalls.length, 2);
assert.equal(capturedCalls[0]!.path, '/Users/u-alice/Items/item1');
assert.equal(capturedCalls[0]!.token, 'tok-alice');
assert.equal(capturedCalls[0]!.options, undefined, 'the GET is a plain fetch, no method/body');
assert.equal(capturedCalls[1]!.path, '/Users/u-alice/Items/item1/UserData');
assert.equal(capturedCalls[1]!.token, 'tok-alice');
const posted = capturedCalls[1]!.options as { method: string; body: Record<string, unknown> };
assert.equal(posted.method, 'POST');
// The whole point of fetch-merge-write: a favorite/rating/play count
// this app never asked to change must come back out exactly as read -
// guessing wrong here risks actually erasing real user data, not just
// silently no-op'ing the way the old Progress-endpoint bugs did.
assert.equal(posted.body.IsFavorite, true);
assert.equal(posted.body.Rating, 9.5);
assert.equal(posted.body.PlayCount, 3);
assert.equal(posted.body.Played, false, 'Played is preserved, not touched by a position-only update');
assert.equal(posted.body.PlaybackPositionTicks, 12345678, 'only this field actually changes');
});
it('rounds a fractional positionTicks (Emby expects an integer)', async (t) => {
let capturedPath = '';
t.mock.method(EmbyClient, 'fetch', async (path: string) => {
capturedPath = path;
return new Response('{}', { status: 200 });
let posted: Record<string, unknown> | undefined;
t.mock.method(EmbyClient, 'fetch', async (path: string, _token: string, options?: { body?: Record<string, unknown> }) => {
if (path.endsWith('/UserData')) { posted = options?.body; return new Response('{}', { status: 200 }); }
return itemResponse({});
});
await PlaystateService.reportProgress('tok', 'u1', 'item1', 'src1', 'sess1', 100.7, true);
await PlaystateService.updatePosition('tok', 'u1', 'item1', 100.7);
const query = capturedPath.split('?')[1];
assert.equal(new URLSearchParams(query).get('PositionTicks'), '101');
assert.equal(posted?.PlaybackPositionTicks, 101);
});
it('swallows a failure instead of throwing - this runs on a timer with nothing to catch it', async (t) => {
it('has no existing UserData at all (a never-touched item) - still works, starting from an empty object', async (t) => {
let posted: Record<string, unknown> | undefined;
t.mock.method(EmbyClient, 'fetch', async (path: string, _token: string, options?: { body?: Record<string, unknown> }) => {
if (path.endsWith('/UserData')) { posted = options?.body; return new Response('{}', { status: 200 }); }
return itemResponse(undefined);
});
await assert.doesNotReject(() => PlaystateService.updatePosition('tok', 'u1', 'item1', 5000));
assert.equal(posted?.PlaybackPositionTicks, 5000);
});
it('skips the write entirely if the GET fails - never posts with missing/corrupt data', async (t) => {
let postCalled = false;
t.mock.method(EmbyClient, 'fetch', async (path: string) => {
if (path.endsWith('/UserData')) postCalled = true;
return new Response('nope', { status: 404 });
});
await assert.doesNotReject(() => PlaystateService.updatePosition('tok', 'u1', 'item1', 0));
assert.equal(postCalled, false, 'a failed read must not be followed by a write built on nothing');
});
it('swallows a thrown error instead of rejecting - this runs on a timer with nothing to catch it', async (t) => {
t.mock.method(EmbyClient, 'fetch', async () => { throw new Error('Emby unreachable'); });
await assert.doesNotReject(() => PlaystateService.reportProgress('tok', 'u1', 'item1', 'src1', 'sess1', 0, false));
await assert.doesNotReject(() => PlaystateService.updatePosition('tok', 'u1', 'item1', 0));
});
});
describe('PlaystateService.markPlayed', () => {
it('POSTs Played: true and resets PlaybackPositionTicks to 0, preserving other fields', async (t) => {
let posted: Record<string, unknown> | undefined;
t.mock.method(EmbyClient, 'fetch', async (path: string, _token: string, options?: { body?: Record<string, unknown> }) => {
if (path.endsWith('/UserData')) { posted = options?.body; return new Response('{}', { status: 200 }); }
return itemResponse({ PlaybackPositionTicks: 3_000_000, Played: false, IsFavorite: true });
});
await PlaystateService.markPlayed('tok', 'u1', 'item1');
assert.equal(posted?.Played, true);
assert.equal(posted?.PlaybackPositionTicks, 0);
assert.equal(posted?.IsFavorite, true, 'a favorite must survive being marked played');
});
// Logging: the whole point is being able to answer "what did this
// actually send, and did Emby accept it" from `docker logs` alone -
// see Logger.playstate's own comment.
it('logs the exact outgoing request, never the token itself', async (t) => {
it('swallows a thrown error instead of rejecting - this runs from an item-advance path that must never stall on it', async (t) => {
t.mock.method(EmbyClient, 'fetch', async () => { throw new Error('Emby unreachable'); });
await assert.doesNotReject(() => PlaystateService.markPlayed('tok', 'u1', 'item1'));
});
});
describe('PlaystateService logging', () => {
it('logs both outgoing requests and the final outcome, never the token itself', async (t) => {
const logged: string[] = [];
t.mock.method(console, 'log', (line: string) => { logged.push(line); });
t.mock.method(EmbyClient, 'fetch', async () => new Response('{}', { status: 200 }));
t.mock.method(EmbyClient, 'fetch', async (path: string) => {
if (path.endsWith('/UserData')) return new Response('{}', { status: 200 });
return itemResponse({});
});
await PlaystateService.reportProgress('super-secret-token', 'u-alice', 'item1', 'src1', 'sess1', 12_345_678, false);
await PlaystateService.updatePosition('super-secret-token', 'u-alice', 'item1', 5_000_000);
const requestLine = logged.find((l) => l.includes('[watchparty:playstate]') && l.includes('POST'));
assert.ok(requestLine, 'the request itself must be logged');
assert.ok(requestLine!.includes('/Users/u-alice/PlayingItems/item1/Progress'));
assert.ok(requestLine!.includes('MediaSourceId=src1'));
assert.ok(requestLine!.includes('PlaySessionId=sess1'));
assert.ok(requestLine!.includes('PositionTicks=12345678'));
assert.ok(requestLine!.includes('IsPaused=false'));
assert.ok(!requestLine!.includes('super-secret-token'), 'the token travels as a header, never logged into the request line');
const lines = logged.filter((l) => l.includes('[watchparty:playstate]'));
assert.ok(lines.some((l) => l.includes('GET') && l.includes('/Users/u-alice/Items/item1')), 'the read must be logged');
assert.ok(lines.some((l) => l.includes('POST') && l.includes('/Users/u-alice/Items/item1/UserData')), 'the write must be logged');
assert.ok(lines.some((l) => l.includes('200') && l.includes('ok')), 'the outcome must be logged');
assert.ok(!lines.some((l) => l.includes('super-secret-token')), 'the token travels as a header, never logged');
});
it('logs a successful response as ok, with its real HTTP status', async (t) => {
it('logs a non-ok write as FAILED, with its real HTTP status', async (t) => {
const logged: string[] = [];
t.mock.method(console, 'log', (line: string) => { logged.push(line); });
t.mock.method(EmbyClient, 'fetch', async () => new Response(null, { status: 204 })); // 204 is a null-body status - a non-null body here would throw
t.mock.method(EmbyClient, 'fetch', async (path: string) => {
if (path.endsWith('/UserData')) return new Response('bad request', { status: 400 });
return itemResponse({});
});
await PlaystateService.reportProgress('tok', 'u1', 'item1', 'src1', 'sess1', 0, false);
const outcomeLine = logged.find((l) => l.includes('[watchparty:playstate]') && l.includes('204'));
assert.ok(outcomeLine, 'the real status code must be logged');
assert.ok(outcomeLine!.includes('ok'));
assert.ok(!outcomeLine!.includes('FAILED'));
});
it('logs a non-ok response as FAILED, with its real HTTP status - the exact case both wrong request shapes hit for real', async (t) => {
const logged: string[] = [];
t.mock.method(console, 'log', (line: string) => { logged.push(line); });
t.mock.method(EmbyClient, 'fetch', async () => new Response("Value cannot be null. (Parameter 'key')", { status: 400 }));
await PlaystateService.reportProgress('tok', 'u1', 'item1', 'src1', 'sess1', 0, false);
await PlaystateService.updatePosition('tok', 'u1', 'item1', 0);
const outcomeLine = logged.find((l) => l.includes('[watchparty:playstate]') && l.includes('400'));
assert.ok(outcomeLine, 'the real status code must be logged');
assert.ok(outcomeLine);
assert.ok(outcomeLine!.includes('FAILED'));
});
it('logs a thrown error (Emby unreachable) distinctly from a non-ok HTTP response', async (t) => {
it('logs a failed read distinctly, without ever attempting/logging the write', async (t) => {
const logged: string[] = [];
t.mock.method(console, 'log', (line: string) => { logged.push(line); });
t.mock.method(EmbyClient, 'fetch', async () => { throw new Error('Emby unreachable'); });
t.mock.method(EmbyClient, 'fetch', async () => new Response('not found', { status: 404 }));
await PlaystateService.reportProgress('tok', 'u1', 'item1', 'src1', 'sess1', 0, false);
await PlaystateService.updatePosition('tok', 'u1', 'item1', 0);
const outcomeLine = logged.find((l) => l.includes('[watchparty:playstate]') && l.includes('threw'));
assert.ok(outcomeLine, 'a thrown error must still be logged, not just swallowed silently');
assert.ok(outcomeLine!.includes('Emby unreachable'));
const lines = logged.filter((l) => l.includes('[watchparty:playstate]'));
assert.ok(lines.some((l) => l.includes('404') && l.includes('FAILED')));
assert.ok(!lines.some((l) => l.includes('POST')), 'no write should even be attempted, let alone logged, once the read failed');
});
});