Compare commits

...

3 Commits

Author SHA1 Message Date
Gauvino
1d962255ca fix(websocket): harden dispatch and drop provider-level router hook
Address CodeRabbit review on the pub/sub provider:

- Wrap each subscriber call in dispatchMessage with try/catch so one
  throwing handler can't abort the rest of the fan-out, nor get
  misreported as a parse failure by the outer onmessage catch.
- Replace the provider-level useAppRouter() hook with expo-router's
  static router for the Play command, per the repo guideline
  (useRouter at provider level can trigger tab switches). The Play
  navigation hardcodes offline:"false", so the offline-aware wrapper
  added nothing here.
2026-06-01 12:13:03 +02:00
Gauvain
075712e064 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-06-01 12:04:54 +02:00
Gauvain
43d3efbaf0 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-06-01 12:04:53 +02:00

View File

@@ -1,4 +1,5 @@
import { getSessionApi } from "@jellyfin/sdk/lib/utils/api"; import { getSessionApi } from "@jellyfin/sdk/lib/utils/api";
import { router } from "expo-router";
import { useAtomValue } from "jotai"; import { useAtomValue } from "jotai";
import { import {
createContext, createContext,
@@ -11,7 +12,6 @@ import {
useState, useState,
} from "react"; } from "react";
import { AppState, type AppStateStatus } from "react-native"; import { AppState, type AppStateStatus } from "react-native";
import useRouter from "@/hooks/useAppRouter";
import { useNetworkAwareQueryClient } from "@/hooks/useNetworkAwareQueryClient"; import { useNetworkAwareQueryClient } from "@/hooks/useNetworkAwareQueryClient";
import { apiAtom, getOrSetDeviceId } from "@/providers/JellyfinProvider"; import { apiAtom, getOrSetDeviceId } from "@/providers/JellyfinProvider";
import { useNetworkStatus } from "@/providers/NetworkStatusProvider"; import { useNetworkStatus } from "@/providers/NetworkStatusProvider";
@@ -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;
} }
@@ -54,7 +88,6 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
const [ws, setWs] = useState<WebSocket | null>(null); const [ws, setWs] = useState<WebSocket | null>(null);
const [isConnected, setIsConnected] = useState(false); const [isConnected, setIsConnected] = useState(false);
const [lastMessage, setLastMessage] = useState<WebSocketMessage | null>(null); const [lastMessage, setLastMessage] = useState<WebSocketMessage | null>(null);
const router = useRouter();
const queryClient = useNetworkAwareQueryClient(); const queryClient = useNetworkAwareQueryClient();
const deviceId = useMemo(() => { const deviceId = useMemo(() => {
return getOrSetDeviceId(); return getOrSetDeviceId();
@@ -63,6 +96,52 @@ 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]) {
// Isolate each handler so one throwing subscriber can't abort the rest
// (and isn't misreported as a parse failure by the outer onmessage catch).
try {
handler(message.Data, message);
} catch (error) {
console.error(
`Error handling WebSocket message type "${message.MessageType}":`,
error,
);
}
}
}, []);
const connectWebSocket = useCallback(() => { const connectWebSocket = useCallback(() => {
if (!deviceId || !api?.accessToken || !isNetworkConnected) { if (!deviceId || !api?.accessToken || !isNetworkConnected) {
@@ -113,7 +192,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 +208,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,47 +239,77 @@ export const WebSocketProvider = ({ children }: WebSocketProviderProps) => {
[queryClient], [queryClient],
); );
useEffect(() => { const handleUserDataChanged = useCallback(
if (!lastMessage) { (data: any) => {
return; // Jellyfin sends UserDataChanged when playback position, played status
} // or favorites change (e.g. finishing an episode). Only the
if (lastMessage.MessageType === "Play") { // progression-based home sections care about it.
handlePlayCommand(lastMessage.Data); if (!((data?.UserDataList?.length ?? 0) > 0)) {
} else if (lastMessage.MessageType === "LibraryChanged") { return;
handleLibraryChanged(lastMessage.Data); }
}
}, [lastMessage, router, handleLibraryChanged]); // Finishing an item can emit several UserDataChanged messages, so
// debounce to invalidate the affected sections only once.
if (userDataChangeDebounceRef.current) {
clearTimeout(userDataChangeDebounceRef.current);
}
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);
}
}; };
}, []); }, []);
const handlePlayCommand = useCallback( const handlePlayCommand = useCallback((data: any) => {
(data: any) => { if (!data?.ItemIds?.length) {
if (!data?.ItemIds?.length) { return;
return; }
}
const itemId = data.ItemIds[0]; const itemId = data.ItemIds[0];
router.push({ router.push({
pathname: "/(auth)/player/direct-player", pathname: "/(auth)/player/direct-player",
params: { params: {
itemId: itemId, itemId: itemId,
playCommand: data.PlayCommand || "PlayNow", playCommand: data.PlayCommand || "PlayNow",
audioIndex: data.AudioStreamIndex?.toString(), audioIndex: data.AudioStreamIndex?.toString(),
subtitleIndex: data.SubtitleStreamIndex?.toString(), subtitleIndex: data.SubtitleStreamIndex?.toString(),
mediaSourceId: data.MediaSourceId || "", mediaSourceId: data.MediaSourceId || "",
bitrateValue: "", bitrateValue: "",
offline: "false", offline: "false",
}, },
}); });
}, }, []);
[router],
// Server-initiated "Play me this item" remote command.
useEffect(
() => subscribe("Play", handlePlayCommand),
[subscribe, handlePlayCommand],
); );
useEffect(() => { useEffect(() => {
@@ -267,7 +379,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>