Compare commits

...

2 Commits

Author SHA1 Message Date
Gauvain
19f3b6af09 feat(home): refresh Continue Watching on UserDataChanged
develop already refreshes the home library sections on LibraryChanged, but
nothing reacted to UserDataChanged, so "Continue Watching" / "Next Up" only
updated on the 60s interval or screen refocus after finishing an episode.

Subscribe to UserDataChanged and invalidate just the progression-based
sections (resume / next up / TV hero), debounced to coalesce the burst a
single finished item emits. Works on phone and TV (handled in the global
provider). Recently-added and suggestions are intentionally left untouched.
2026-05-31 23:42:51 +02:00
Gauvain
791a6db692 refactor(websocket): add subscribe() pub/sub API
The provider only kept the most recent message in `lastMessage`, so two
messages arriving in the same tick were coalesced and one was lost. Add a
`subscribe(messageType, handler)` registry that delivers every message to
its handlers without coalescing, and move the provider's own `Play` and
`LibraryChanged` handling onto it.

`lastMessage` is kept (now deprecated) for `useWebsockets`, which still
consumes it for GeneralCommand handling.
2026-05-31 23:41:52 +02:00

View File

@@ -28,6 +28,20 @@ const LIBRARY_CHANGE_QUERY_KEYS = [
["episodes"], ["episodes"],
] as const; ] as const;
// Query keys that depend on per-user playback state (resume position, played
// status, favorites) and should be refreshed when the server reports a
// `UserDataChanged`. Scoped to the progression-based sections so finishing an
// episode does not pointlessly refetch "recently added" or suggestions.
const USER_DATA_CHANGE_QUERY_KEYS = [
["home", "continueAndNextUp"],
["home", "resumeItems"],
["home", "nextUp-all"],
["home", "heroItems"],
["resumeItems"],
["nextUp-all"],
["nextUp"],
] as const;
interface WebSocketMessage { interface WebSocketMessage {
MessageType: string; MessageType: string;
Data: any; Data: any;
@@ -38,10 +52,30 @@ interface WebSocketProviderProps {
children: ReactNode; children: ReactNode;
} }
/**
* Handler invoked for every message of a given `MessageType`. Receives the
* message `Data` payload and the full message.
*/
type WebSocketMessageHandler = (data: any, message: WebSocketMessage) => void;
interface WebSocketContextType { interface WebSocketContextType {
ws: WebSocket | null; ws: WebSocket | null;
isConnected: boolean; isConnected: boolean;
/**
* @deprecated Prefer `subscribe`. `lastMessage` only keeps the most recent
* message, so bursts arriving in the same tick are coalesced and lost. Kept
* for `useWebsockets` (GeneralCommand handling) until it is migrated.
*/
lastMessage: WebSocketMessage | null; lastMessage: WebSocketMessage | null;
/**
* Subscribe to a given message type. The handler is called synchronously for
* every matching message (no coalescing, unlike `lastMessage`). Returns an
* unsubscribe function to call on cleanup.
*/
subscribe: (
messageType: string,
handler: WebSocketMessageHandler,
) => () => void;
sendMessage: (message: any) => void; sendMessage: (message: any) => void;
clearLastMessage: () => void; clearLastMessage: () => void;
} }
@@ -63,6 +97,43 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
const libraryChangeDebounceRef = useRef<ReturnType<typeof setTimeout> | null>( const libraryChangeDebounceRef = useRef<ReturnType<typeof setTimeout> | null>(
null, null,
); );
const userDataChangeDebounceRef = useRef<ReturnType<
typeof setTimeout
> | null>(null);
// Pub/sub registry: messageType -> set of handlers. Stored in a ref so
// subscribing/dispatching never triggers a re-render.
const listenersRef = useRef<Map<string, Set<WebSocketMessageHandler>>>(
new Map(),
);
const subscribe = useCallback(
(messageType: string, handler: WebSocketMessageHandler) => {
const listeners = listenersRef.current;
let handlers = listeners.get(messageType);
if (!handlers) {
handlers = new Set();
listeners.set(messageType, handlers);
}
handlers.add(handler);
return () => {
handlers?.delete(handler);
if (handlers && handlers.size === 0) {
listeners.delete(messageType);
}
};
},
[],
);
const dispatchMessage = useCallback((message: WebSocketMessage) => {
const handlers = listenersRef.current.get(message.MessageType);
if (!handlers || handlers.size === 0) return;
// Copy to tolerate handlers that unsubscribe during dispatch.
for (const handler of [...handlers]) {
handler(message.Data, message);
}
}, []);
const connectWebSocket = useCallback(() => { const connectWebSocket = useCallback(() => {
if (!deviceId || !api?.accessToken || !isNetworkConnected) { if (!deviceId || !api?.accessToken || !isNetworkConnected) {
@@ -113,7 +184,10 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
newWebSocket.onmessage = (e) => { newWebSocket.onmessage = (e) => {
try { try {
const message = JSON.parse(e.data); const message = JSON.parse(e.data);
setLastMessage(message); // Store the last message in context // Legacy single-slot state, still consumed by useWebsockets.
setLastMessage(message);
// Pub/sub: deliver to every subscriber without coalescing.
dispatchMessage(message);
} catch (error) { } catch (error) {
console.error("Error parsing WebSocket message:", error); console.error("Error parsing WebSocket message:", error);
} }
@@ -126,7 +200,7 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
} }
newWebSocket.close(); newWebSocket.close();
}; };
}, [api, deviceId, isNetworkConnected]); }, [api, deviceId, isNetworkConnected, dispatchMessage]);
const handleLibraryChanged = useCallback( const handleLibraryChanged = useCallback(
(data: any) => { (data: any) => {
@@ -157,22 +231,49 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
[queryClient], [queryClient],
); );
useEffect(() => { const handleUserDataChanged = useCallback(
if (!lastMessage) { (data: any) => {
// Jellyfin sends UserDataChanged when playback position, played status
// or favorites change (e.g. finishing an episode). Only the
// progression-based home sections care about it.
if (!((data?.UserDataList?.length ?? 0) > 0)) {
return; return;
} }
if (lastMessage.MessageType === "Play") {
handlePlayCommand(lastMessage.Data); // Finishing an item can emit several UserDataChanged messages, so
} else if (lastMessage.MessageType === "LibraryChanged") { // debounce to invalidate the affected sections only once.
handleLibraryChanged(lastMessage.Data); if (userDataChangeDebounceRef.current) {
clearTimeout(userDataChangeDebounceRef.current);
} }
}, [lastMessage, router, handleLibraryChanged]); userDataChangeDebounceRef.current = setTimeout(() => {
for (const queryKey of USER_DATA_CHANGE_QUERY_KEYS) {
queryClient.invalidateQueries({ queryKey: [...queryKey] });
}
}, 800);
},
[queryClient],
);
// Refresh library-dependent queries when the server reports a change.
useEffect(
() => subscribe("LibraryChanged", handleLibraryChanged),
[subscribe, handleLibraryChanged],
);
// Refresh "Continue Watching" / "Next Up" when playback state changes.
useEffect(
() => subscribe("UserDataChanged", handleUserDataChanged),
[subscribe, handleUserDataChanged],
);
useEffect(() => { useEffect(() => {
return () => { return () => {
if (libraryChangeDebounceRef.current) { if (libraryChangeDebounceRef.current) {
clearTimeout(libraryChangeDebounceRef.current); clearTimeout(libraryChangeDebounceRef.current);
} }
if (userDataChangeDebounceRef.current) {
clearTimeout(userDataChangeDebounceRef.current);
}
}; };
}, []); }, []);
@@ -200,6 +301,12 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
[router], [router],
); );
// Server-initiated "Play me this item" remote command.
useEffect(
() => subscribe("Play", handlePlayCommand),
[subscribe, handlePlayCommand],
);
useEffect(() => { useEffect(() => {
const cleanup = connectWebSocket(); const cleanup = connectWebSocket();
return cleanup; return cleanup;
@@ -267,7 +374,14 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
}, []); }, []);
return ( return (
<WebSocketContext.Provider <WebSocketContext.Provider
value={{ ws, isConnected, lastMessage, sendMessage, clearLastMessage }} value={{
ws,
isConnected,
lastMessage,
subscribe,
sendMessage,
clearLastMessage,
}}
> >
{children} {children}
</WebSocketContext.Provider> </WebSocketContext.Provider>