Rework progress observer to use websockets

api/src/auth.ts

@@ -33,7 +33,7 @@ const Jwt = t.Object({
 type Jwt = typeof Jwt.static;
 const validator = TypeCompiler.Compile(Jwt);
 
-export async function verifyJwt(bearer: string) {
+async function verifyJwt(bearer: string) {
 	// @ts-expect-error ts can't understand that there's two overload idk why
 	const { payload } = await jwtVerify(bearer, jwtSecret ?? jwks, {
 		issuer: process.env.JWT_ISSUER,

api/src/websockets.ts

@@ -1,6 +1,6 @@
 import type { TObject, TString } from "@sinclair/typebox";
 import Elysia, { type TSchema, t } from "elysia";
-import { verifyJwt } from "./auth";
+import { auth } from "./auth";
 import { updateProgress } from "./controllers/profiles/history";
 import { getOrCreateProfile } from "./controllers/profiles/profile";
 import { SeedHistory } from "./models/history";
@@ -8,7 +8,7 @@ import { SeedHistory } from "./models/history";
 const actionMap = {
 	ping: handler({
 		message(ws) {
-			ws.send({ response: "pong" });
+			ws.send({ action: "ping", response: "pong" });
 		},
 	}),
 	watch: handler({
@@ -20,55 +20,12 @@ const actionMap = {
 			const ret = await updateProgress(profilePk, [
 				{ ...body, playedDate: null },
 			]);
-			ws.send(ret);
+			ws.send({ action: "watch", ...ret });
 		},
 	}),
 };
 
-const baseWs = new Elysia()
-	.guard({
-		headers: t.Object(
-			{
-				authorization: t.Optional(t.TemplateLiteral("Bearer ${string}")),
-				"Sec-WebSocket-Protocol": t.Optional(
-					t.Array(
-						t.Union([t.Literal("kyoo"), t.TemplateLiteral("Bearer ${string}")]),
-					),
-				),
-			},
-			{ additionalProperties: true },
-		),
-	})
-	.resolve(
-		async ({
-			headers: { authorization, "Sec-WebSocket-Protocol": wsProtocol },
-			status,
-		}) => {
-			const auth =
-				authorization ??
-				(wsProtocol?.length === 2 &&
-				wsProtocol[0] === "kyoo" &&
-				wsProtocol[1].startsWith("Bearer ")
-					? wsProtocol[1]
-					: null);
-			const bearer = auth?.slice(7);
-			if (!bearer) {
-				return status(403, {
-					status: 403,
-					message: "No authorization header was found.",
-				});
-			}
-			try {
-				return await verifyJwt(bearer);
-			} catch (err) {
-				return status(403, {
-					status: 403,
-					message: "Invalid jwt. Verification vailed",
-					details: err,
-				});
-			}
-		},
-	);
+const baseWs = new Elysia().use(auth);
 
 export const appWs = baseWs.ws("/ws", {
 	body: t.Union(

auth/jwt.go

@@ -44,9 +44,17 @@ func (h *Handler) CreateJwt(c echo.Context) error {
 	var token string
 
 	if auth == "" {
-		c, _ := c.Request().Cookie("X-Bearer")
-		if c != nil {
-			token = c.Value
+		cookie, _ := c.Request().Cookie("X-Bearer")
+		if cookie != nil {
+			token = cookie.Value
+		} else {
+			protocol, ok := c.Request().Header["Sec-Websocket-Protocol"]
+			if ok &&
+				len(protocol) == 2 &&
+				protocol[0] == "kyoo" &&
+				strings.HasPrefix(protocol[1], "Bearer") {
+				token = protocol[1][len("Bearer "):]
+			}
 		}
 	} else if strings.HasPrefix(auth, "Bearer ") {
 		token = auth[len("Bearer "):]

front/bun.lock

@@ -1,5 +1,6 @@
 {
   "lockfileVersion": 1,
+  "configVersion": 0,
   "workspaces": {
     "": {
       "name": "kyoo",
@@ -49,6 +50,7 @@
         "react-native-web": "^0.21.2",
         "react-native-worklets": "0.5.1",
         "react-tooltip": "^5.29.1",
+        "react-use-websocket": "^4.13.0",
         "sweetalert2": "^11.26.3",
         "tsx": "^4.20.6",
         "uuid": "^13.0.0",
@@ -1352,6 +1354,8 @@
 
     "react-tooltip": ["react-tooltip@5.30.0", "", { "dependencies": { "@floating-ui/dom": "^1.6.1", "classnames": "^2.3.0" }, "peerDependencies": { "react": ">=16.14.0", "react-dom": ">=16.14.0" } }, "sha512-Yn8PfbgQ/wmqnL7oBpz1QiDaLKrzZMdSUUdk7nVeGTwzbxCAJiJzR4VSYW+eIO42F1INt57sPUmpgKv0KwJKtg=="],
 
+    "react-use-websocket": ["react-use-websocket@4.13.0", "", {}, "sha512-anMuVoV//g2N76Wxqvqjjo1X48r9Np3y1/gMl7arX84tAPXdy5R7sB5lO5hvCzQRYjqXwV8XMAiEBOUbyrZFrw=="],
+
     "regenerate": ["regenerate@1.4.2", "", {}, "sha512-zrceR/XhGYU/d/opr2EKO7aRHUeiBI8qjtfHqADTwZd6Szfy16la6kqD0MIUs5z5hx6AaKa+PixpPrR289+I0A=="],
 
     "regenerate-unicode-properties": ["regenerate-unicode-properties@10.2.2", "", { "dependencies": { "regenerate": "^1.4.2" } }, "sha512-m03P+zhBeQd1RGnYxrGyDAPpWX/epKirLrp8e3qevZdVkKtnCrjjWczIbYc8+xd6vcTStVlqfycTx1KR4LOr0g=="],

front/package.json

@@ -60,6 +60,7 @@
 		"react-native-web": "^0.21.2",
 		"react-native-worklets": "0.5.1",
 		"react-tooltip": "^5.29.1",
+		"react-use-websocket": "^4.13.0",
 		"sweetalert2": "^11.26.3",
 		"tsx": "^4.20.6",
 		"uuid": "^13.0.0",

front/src/query/websockets.ts

@@ -0,0 +1,28 @@
+import { useEffect } from "react";
+import useWebSocket from "react-use-websocket";
+import { useToken } from "~/providers/account-context";
+
+export const useWebsockets = ({
+	filterActions,
+}: {
+	filterActions: string[];
+}) => {
+	const { apiUrl, authToken } = useToken();
+	const ret = useWebSocket(`${apiUrl}/api/ws`, {
+		protocols: authToken ? ["kyoo", `Bearer ${authToken}`] : undefined,
+		filter: (msg) => filterActions.includes(msg.data.action),
+		share: true,
+		retryOnError: true,
+		heartbeat: {
+			message: `{ "action": "ping" }`,
+			returnMessage: `{ "response": "pong" }`,
+			interval: 25_000,
+		},
+	});
+
+	useEffect(() => {
+		console.log(ret.readyState);
+	}, [ret.readyState]);
+
+	return ret;
+};

front/src/ui/player/index.tsx

@@ -21,6 +21,7 @@ import { toggleFullscreen } from "./controls/misc";
 import { PlayModeContext } from "./controls/tracks-menu";
 import { useKeyboard } from "./keyboard";
 import { enhanceSubtitles } from "./subtitles";
+import { useProgressObserver } from "./progress-observer";
 
 const clientId = uuidv4();
 
@@ -113,6 +114,11 @@ export const Player = () => {
 		return true;
 	}, [data?.next, setSlug, setStart]);
 
+	useProgressObserver(
+		player,
+		data && entry ? { videoId: data.id, entryId: entry.id } : null,
+	);
+
 	useEvent(player, "onEnd", () => {
 		const hasNext = playNext();
 		if (!hasNext && data?.show) router.navigate(data.show.href);

front/src/ui/player/old/watch-status-observer.tsx

@@ -1,86 +0,0 @@
-/*
- * Kyoo - A portable and vast media library solution.
- * Copyright (c) Kyoo.
- *
- * See AUTHORS.md and LICENSE file in the project root for full license information.
- *
- * Kyoo is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * any later version.
- *
- * Kyoo is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with Kyoo. If not, see <https://www.gnu.org/licenses/>.
- */
-
-import { type MutationParam, useAccount, WatchStatusV } from "@kyoo/models";
-import { useMutation } from "@tanstack/react-query";
-import { useAtomValue } from "jotai";
-import { useAtomCallback } from "jotai/utils";
-import { useCallback, useEffect } from "react";
-import { playAtom, progressAtom } from "./old/statee";
-
-export const WatchStatusObserver = ({
-	type,
-	slug,
-	duration,
-}: {
-	type: "episode" | "movie";
-	slug: string;
-	duration: number;
-}) => {
-	const account = useAccount();
-	// const queryClient = useQueryClient();
-	const { mutate: _mutate } = useMutation<unknown, Error, MutationParam>({
-		mutationKey: [type, slug, "watchStatus"],
-		// onSettled: async () =>
-		// 	await queryClient.invalidateQueries({ queryKey: [type === "episode" ? "show" : type, slug] }),
-	});
-	const mutate = useCallback(
-		(type: string, slug: string, seconds: number) => {
-			if (seconds < 0 || duration <= 0) return;
-			_mutate({
-				method: "POST",
-				path: [type, slug, "watchStatus"],
-				params: {
-					status: WatchStatusV.Watching,
-					watchedTime: Math.round(seconds),
-					percent: Math.round((seconds / duration) * 100),
-				},
-			});
-		},
-		[_mutate, duration],
-	);
-	const readProgress = useAtomCallback(
-		useCallback((get) => {
-			const currCount = get(progressAtom);
-			return currCount;
-		}, []),
-	);
-
-	// update watch status every 10 seconds and on unmount.
-	useEffect(() => {
-		if (!account) return;
-		const timer = setInterval(() => {
-			mutate(type, slug, readProgress());
-		}, 10_000);
-		return () => {
-			clearInterval(timer);
-			mutate(type, slug, readProgress());
-		};
-	}, [account, type, slug, readProgress, mutate]);
-
-	// update watch status when play status change (and on mount).
-	const isPlaying = useAtomValue(playAtom);
-	// biome-ignore lint/correctness/useExhaustiveDependencies: Include isPlaying
-	useEffect(() => {
-		if (!account) return;
-		mutate(type, slug, readProgress());
-	}, [account, type, slug, isPlaying, readProgress, mutate]);
-	return null;
-};

front/src/ui/player/progress-observer.ts

@@ -0,0 +1,36 @@
+import { useCallback, useEffect } from "react";
+import { useEvent, type VideoPlayer } from "react-native-video";
+import { useWebsockets } from "~/query/websockets";
+
+export const useProgressObserver = (
+	player: VideoPlayer,
+	ids: { videoId: string; entryId: string } | null,
+) => {
+	const { sendJsonMessage } = useWebsockets({
+		filterActions: ["watch"],
+	});
+
+	const updateProgress = useCallback(() => {
+		if (
+			ids === null ||
+			Number.isNaN(player.currentTime) ||
+			Number.isNaN(player.duration) ||
+			!player.isPlaying
+		)
+			return;
+		sendJsonMessage({
+			action: "watch",
+			entry: ids.entryId,
+			videoId: ids.videoId,
+			percent: Math.round((player.currentTime / player.duration) * 100),
+			time: Math.round(player.currentTime),
+		});
+	}, [player, ids, sendJsonMessage]);
+
+	useEffect(() => {
+		const interval = setInterval(updateProgress, 5000);
+		return () => clearInterval(interval);
+	}, [updateProgress]);
+
+	useEvent(player, "onPlaybackStateChange", updateProgress);
+};