feat: #655 implement jellyfin media server

.vscode/settings.json

@@ -8,7 +8,14 @@
   "typescript.tsdk": "node_modules\\typescript\\lib",
   "js/ts.implicitProjectConfig.experimentalDecorators": true,
   "prettier.configPath": "./tooling/prettier/index.mjs",
-  "cSpell.words": ["cqmin", "homarr", "superjson", "trpc", "Umami"],
+  "cSpell.words": [
+    "cqmin",
+    "homarr",
+    "jellyfin",
+    "superjson",
+    "trpc",
+    "Umami"
+  ],
   "i18n-ally.dirStructure": "auto",
   "i18n-ally.enabledFrameworks": ["next-international"],
   "i18n-ally.localesPaths": ["./packages/translation/src/lang/"],

apps/nextjs/src/components/board/sections/content.tsx

@@ -206,7 +206,7 @@ const ItemMenu = ({
   return (
     <Menu withinPortal withArrow position="right-start" arrowPosition="center">
       <Menu.Target>
-        <ActionIcon variant="transparent" pos="absolute" top={offset} right={offset} style={{ zIndex: 1 }}>
+        <ActionIcon variant="transparent" pos="absolute" top={offset} right={offset} style={{ zIndex: 2 }}>
           <IconDotsVertical />
         </ActionIcon>
       </Menu.Target>

apps/tasks/src/jobs.ts

@@ -1,6 +1,7 @@
 import { iconsUpdaterJob } from "~/jobs/icons-updater";
 import { smartHomeEntityStateJob } from "~/jobs/integrations/home-assistant";
 import { analyticsJob } from "./jobs/analytics";
+import { mediaServerJob } from "./jobs/integrations/media-server";
 import { pingJob } from "./jobs/ping";
 import { queuesJob } from "./jobs/queue";
 import { createJobGroup } from "./lib/cron-job/group";
@@ -11,6 +12,7 @@ export const jobs = createJobGroup({
   iconsUpdater: iconsUpdaterJob,
   ping: pingJob,
   smartHomeEntityState: smartHomeEntityStateJob,
+  mediaServer: mediaServerJob,
 
   // This job is used to process queues.
   queues: queuesJob,

apps/tasks/src/jobs/integrations/media-server.ts

@@ -0,0 +1,46 @@
+import { decryptSecret } from "@homarr/common";
+import { db, eq } from "@homarr/db";
+import { items } from "@homarr/db/schema/sqlite";
+import { JellyfinIntegration } from "@homarr/integrations";
+import { createCacheChannel } from "@homarr/redis";
+
+import { EVERY_5_SECONDS } from "~/lib/cron-job/constants";
+import { createCronJob } from "~/lib/cron-job/creator";
+
+export const mediaServerJob = createCronJob(EVERY_5_SECONDS).withCallback(async () => {
+  const itemsForIntegration = await db.query.items.findMany({
+    where: eq(items.kind, "mediaServer"),
+    with: {
+      integrations: {
+        with: {
+          integration: {
+            with: {
+              secrets: {
+                columns: {
+                  kind: true,
+                  value: true,
+                },
+              },
+            },
+          },
+        },
+      },
+    },
+  });
+
+  for (const itemForIntegration of itemsForIntegration) {
+    for (const integration of itemForIntegration.integrations) {
+      const jellyfinIntegration = new JellyfinIntegration({
+        ...integration.integration,
+        decryptedSecrets: integration.integration.secrets.map((secret) => ({
+          ...secret,
+          value: decryptSecret(secret.value),
+        })),
+      });
+      const streamSessions = await jellyfinIntegration.getCurrentSessionsAsync();
+      const cache = createCacheChannel(`media-server:${integration.integrationId}`);
+      await cache.setAsync(streamSessions);
+      await cache.publishAsync(streamSessions);
+    }
+  }
+});

apps/tasks/src/lib/cron-job/creator.ts

@@ -11,6 +11,7 @@ export const createCronJob = (cronExpression: string, options: CreateCronJobOpti
   return {
     withCallback: (callback: () => MaybePromise<void>) => {
       const catchingCallbackAsync = async () => {
+        logger.info(`Running Cron job with expression ${cronExpression}...`);
         try {
           await callback();
         } catch (error) {

packages/api/src/router/widgets/index.ts

@@ -1,6 +1,7 @@
 import { createTRPCRouter } from "../../trpc";
 import { appRouter } from "./app";
 import { dnsHoleRouter } from "./dns-hole";
+import { mediaServerRouter } from "./media-server";
 import { notebookRouter } from "./notebook";
 import { smartHomeRouter } from "./smart-home";
 import { weatherRouter } from "./weather";
@@ -11,4 +12,5 @@ export const widgetRouter = createTRPCRouter({
   app: appRouter,
   dnsHole: dnsHoleRouter,
   smartHome: smartHomeRouter,
+  mediaServer: mediaServerRouter,
 });

packages/api/src/router/widgets/media-server.ts

@@ -0,0 +1,36 @@
+import { observable } from "@trpc/server/observable";
+
+import { createCacheChannel } from "@homarr/redis";
+
+import type { StreamSession } from "../../../../integrations/src/interfaces/media-server/session";
+import { createManyIntegrationMiddleware } from "../../middlewares/integration";
+import { createTRPCRouter, publicProcedure } from "../../trpc";
+
+export const mediaServerRouter = createTRPCRouter({
+  getCurrentStreams: publicProcedure
+    .unstable_concat(createManyIntegrationMiddleware("jellyfin", "plex"))
+    .query(async ({ ctx }) => {
+      const data = await Promise.all(
+        ctx.integrations.map(async (integration) => {
+          const cache = createCacheChannel<StreamSession[]>(`media-server:${integration.id}`);
+          return await cache.getAsync();
+        }),
+      );
+      return data;
+    }),
+  subcribeToCurrentStreams: publicProcedure
+    .unstable_concat(createManyIntegrationMiddleware("jellyfin", "plex"))
+    .subscription(({ ctx }) => {
+      return observable<{ integrationId: string; data: StreamSession[] }>((emit) => {
+        for (const integration of ctx.integrations) {
+          const cache = createCacheChannel<StreamSession[]>(`media-server:${integration.id}`);
+          void cache.subscribeAsync((sessions) => {
+            emit.next({
+              integrationId: integration.id,
+              data: sessions,
+            });
+          });
+        }
+      });
+    }),
+});

packages/definitions/src/widget.ts

@@ -8,5 +8,6 @@ export const widgetKinds = [
   "dnsHoleSummary",
   "smartHome-entityState",
   "smartHome-executeAutomation",
+  "mediaServer",
 ] as const;
 export type WidgetKind = (typeof widgetKinds)[number];

packages/integrations/package.json

@@ -22,10 +22,11 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
+    "@homarr/common": "workspace:^0.1.0",
     "@homarr/definitions": "workspace:^0.1.0",
     "@homarr/log": "workspace:^0.1.0",
     "@homarr/validation": "workspace:^0.1.0",
-    "@homarr/common": "workspace:^0.1.0"
+    "@jellyfin/sdk": "^0.10.0"
   },
   "devDependencies": {
     "@homarr/eslint-config": "workspace:^0.2.0",

packages/integrations/src/index.ts

@@ -1,2 +1,3 @@
 export { PiHoleIntegration } from "./pi-hole/pi-hole-integration";
 export { HomeAssistantIntegration } from "./homeassistant/homeassistant-integration";
+export { JellyfinIntegration } from "./jellyfin/jellyfin-integration";

packages/integrations/src/interfaces/media-server/session.ts

@@ -0,0 +1,17 @@
+export interface StreamSession {
+  sessionId: string;
+  sessionName: string;
+  user: {
+    userId: string;
+    username: string;
+    profilePictureUrl: URL | null;
+  };
+  currentlyPlaying: {
+    type: "audio" | "video" | "tv" | "movie";
+    name: string;
+    seasonName: string | undefined;
+    episodeName?: string | null;
+    albumName?: string | null;
+    episodeCount?: number | null;
+  } | null;
+}

packages/integrations/src/jellyfin/jellyfin-integration.ts

@@ -0,0 +1,57 @@
+import { Jellyfin } from "@jellyfin/sdk";
+import { getSessionApi } from "@jellyfin/sdk/lib/utils/api/session-api";
+
+import { Integration } from "../base/integration";
+import type { StreamSession } from "../interfaces/media-server/session";
+
+const jellyfin = new Jellyfin({
+  clientInfo: {
+    name: "Homarr",
+    version: "0.0.1",
+  },
+  deviceInfo: {
+    name: "Homarr",
+    id: "homarr",
+  },
+});
+
+export class JellyfinIntegration extends Integration {
+  async getCurrentSessionsAsync(): Promise<StreamSession[]> {
+    const apiKey = this.getSecretValue("apiKey");
+    const api = jellyfin.createApi(this.integration.url, apiKey);
+    const sessionApi = getSessionApi(api);
+    const sessions = await sessionApi.getSessions();
+
+    if (sessions.status !== 200) {
+      throw new Error(
+        `Jellyfin server ${this.integration.url} returned a non successful status code: ${sessions.status}`,
+      );
+    }
+
+    return sessions.data.map((sessionInfo): StreamSession => {
+      let nowPlaying: StreamSession["currentlyPlaying"] | null = null;
+
+      if (sessionInfo.NowPlayingItem) {
+        nowPlaying = {
+          type: "tv",
+          name: sessionInfo.NowPlayingItem.Name ?? "",
+          seasonName: sessionInfo.NowPlayingItem.SeasonName ?? "",
+          episodeName: sessionInfo.NowPlayingItem.EpisodeTitle,
+          albumName: sessionInfo.NowPlayingItem.Album ?? "",
+          episodeCount: sessionInfo.NowPlayingItem.EpisodeCount,
+        };
+      }
+
+      return {
+        sessionId: `${sessionInfo.Id}`,
+        sessionName: `${sessionInfo.Client} (${sessionInfo.DeviceName})`,
+        user: {
+          profilePictureUrl: null,
+          userId: `${sessionInfo.UserId}`,
+          username: `${sessionInfo.UserName}`,
+        },
+        currentlyPlaying: nowPlaying,
+      };
+    });
+  }
+}

packages/redis/src/lib/channel.ts

@@ -144,6 +144,20 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
 
       return parsedData;
     },
+    subscribeAsync: async (callback: (data: TData) => void) => {
+      await subscriber.subscribe(cacheChannelName);
+      subscriber.on("message", (channel, message) => {
+        if (channel !== cacheChannelName) {
+          logger.warn("received message on " + channel + " but was looking for " + cacheChannelName);
+          return;
+        }
+        callback(superjson.parse(message));
+        logger.info("sent message on" + cacheChannelName + "!");
+      });
+    },
+    publishAsync: async (data: TData) => {
+      await publisher.publish(cacheChannelName, superjson.stringify(data));
+    },
     /**
      * Invalidate the cache channels data.
      */

packages/translation/src/lang/en.ts

@@ -927,6 +927,7 @@ export default {
         },
         noIntegration: "No integration selected",
       },
+      option: {},
     },
     video: {
       name: "Video Stream",
@@ -951,6 +952,11 @@ export default {
         forYoutubeUseIframe: "For YouTube videos use the iframe option",
       },
     },
+    mediaServer: {
+      name: "Current media server streams",
+      description: "Show the current streams on your media servers",
+      option: {},
+    },
   },
   widgetPreview: {
     toggle: {

packages/widgets/package.json

@@ -55,6 +55,7 @@
     "clsx": "^2.1.1",
     "dayjs": "^1.11.11",
     "next": "^14.2.4",
+    "mantine-react-table": "2.0.0-beta.5",
     "react": "^18.3.1",
     "video.js": "^8.12.0"
   },

packages/widgets/src/index.tsx

@@ -11,6 +11,7 @@ import type { WidgetComponentProps } from "./definition";
 import * as dnsHoleSummary from "./dns-hole/summary";
 import * as iframe from "./iframe";
 import type { WidgetImportRecord } from "./import";
+import * as mediaServer from "./media-server";
 import * as notebook from "./notebook";
 import * as smartHomeEntityState from "./smart-home/entity-state";
 import * as smartHomeExecuteAutomation from "./smart-home/execute-automation";
@@ -33,6 +34,7 @@ export const widgetImports = {
   dnsHoleSummary,
   "smartHome-entityState": smartHomeEntityState,
   "smartHome-executeAutomation": smartHomeExecuteAutomation,
+  mediaServer,
 } satisfies WidgetImportRecord;
 
 export type WidgetImports = typeof widgetImports;