feat: add jellyfin integration (#672)

* feat: #655 implement jellyfin media server

* fix: table overflow

* feat: pr feedback

* refactor: format

* refactor: merge existing code

* fix: code smells

* refactor: format commit

.vscode/settings.json

@@ -11,10 +11,11 @@
   "cSpell.words": [
     "cqmin",
     "homarr",
-    "Sonarr",
+    "jellyfin",
     "superjson",
     "trpc",
-    "Umami"
+    "Umami",
+    "Sonarr"
   ],
   "i18n-ally.dirStructure": "auto",
   "i18n-ally.enabledFrameworks": ["next-international"],

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

@@ -206,8 +206,8 @@ 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 }}>
-          <IconDotsVertical />
+        <ActionIcon variant="default" radius={"xl"} pos="absolute" top={offset} right={offset} style={{ zIndex: 10 }}>
+          <IconDotsVertical size={"1rem"} />
         </ActionIcon>
       </Menu.Target>
       <Menu.Dropdown miw={128}>

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

@@ -1,5 +1,5 @@
 import type { CalendarEvent } from "@homarr/integrations/types";
-import { createItemWithIntegrationChannel } from "@homarr/redis";
+import { createItemAndIntegrationChannel } from "@homarr/redis";
 
 import { createManyIntegrationOfOneItemMiddleware } from "../../middlewares/integration";
 import { createTRPCRouter, publicProcedure } from "../../trpc";
@@ -10,10 +10,8 @@ export const calendarRouter = createTRPCRouter({
     .query(async ({ ctx }) => {
       return await Promise.all(
         ctx.integrations.flatMap(async (integration) => {
-          for (const item of integration.items) {
-            const cache = createItemWithIntegrationChannel<CalendarEvent[]>(item.itemId, integration.id);
-            return await cache.getAsync();
-          }
+          const cache = createItemAndIntegrationChannel<CalendarEvent[]>("calendar", integration.id);
+          return await cache.getAsync();
         }),
       );
     }),

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

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

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

@@ -0,0 +1,39 @@
+import { observable } from "@trpc/server/observable";
+
+import type { StreamSession } from "@homarr/integrations";
+import { createItemAndIntegrationChannel } from "@homarr/redis";
+
+import { createManyIntegrationMiddleware } from "../../middlewares/integration";
+import { createTRPCRouter, publicProcedure } from "../../trpc";
+
+export const mediaServerRouter = createTRPCRouter({
+  getCurrentStreams: publicProcedure
+    .unstable_concat(createManyIntegrationMiddleware("jellyfin", "plex"))
+    .query(async ({ ctx }) => {
+      return await Promise.all(
+        ctx.integrations.map(async (integration) => {
+          const channel = createItemAndIntegrationChannel<StreamSession[]>("mediaServer", integration.id);
+          const data = await channel.getAsync();
+          return {
+            integrationId: integration.id,
+            sessions: data?.data ?? [],
+          };
+        }),
+      );
+    }),
+  subscribeToCurrentStreams: publicProcedure
+    .unstable_concat(createManyIntegrationMiddleware("jellyfin", "plex"))
+    .subscription(({ ctx }) => {
+      return observable<{ integrationId: string; data: StreamSession[] }>((emit) => {
+        for (const integration of ctx.integrations) {
+          const channel = createItemAndIntegrationChannel<StreamSession[]>("mediaServer", integration.id);
+          void channel.subscribeAsync((sessions) => {
+            emit.next({
+              integrationId: integration.id,
+              data: sessions,
+            });
+          });
+        }
+      });
+    }),
+});

packages/cron-jobs/src/index.ts

@@ -2,6 +2,7 @@ import { analyticsJob } from "./jobs/analytics";
 import { iconsUpdaterJob } from "./jobs/icons-updater";
 import { smartHomeEntityStateJob } from "./jobs/integrations/home-assistant";
 import { mediaOrganizerJob } from "./jobs/integrations/media-organizer";
+import { mediaServerJob } from "./jobs/integrations/media-server";
 import { pingJob } from "./jobs/ping";
 import { createCronJobGroup } from "./lib";
 
@@ -10,6 +11,7 @@ export const jobGroup = createCronJobGroup({
   iconsUpdater: iconsUpdaterJob,
   ping: pingJob,
   smartHomeEntityState: smartHomeEntityStateJob,
+  mediaServer: mediaServerJob,
   mediaOrganizer: mediaOrganizerJob,
 });
 

packages/cron-jobs/src/jobs/integrations/media-organizer.ts

@@ -7,7 +7,7 @@ import { db, eq } from "@homarr/db";
 import { items } from "@homarr/db/schema/sqlite";
 import { SonarrIntegration } from "@homarr/integrations";
 import type { CalendarEvent } from "@homarr/integrations/types";
-import { createItemWithIntegrationChannel } from "@homarr/redis";
+import { createItemAndIntegrationChannel } from "@homarr/redis";
 
 // This import is done that way to avoid circular dependencies.
 import type { WidgetComponentProps } from "../../../../widgets";
@@ -50,7 +50,7 @@ export const mediaOrganizerJob = createCronJob("mediaOrganizer", EVERY_MINUTE).w
       });
       const events = await sonarr.getCalendarEventsAsync(start, end);
 
-      const cache = createItemWithIntegrationChannel<CalendarEvent[]>(itemForIntegration.id, integration.integrationId);
+      const cache = createItemAndIntegrationChannel<CalendarEvent[]>("calendar", integration.integrationId);
       await cache.setAsync(events);
     }
   }

packages/cron-jobs/src/jobs/integrations/media-server.ts

@@ -0,0 +1,45 @@
+import { decryptSecret } from "@homarr/common";
+import { EVERY_5_SECONDS } from "@homarr/cron-jobs-core/expressions";
+import { db, eq } from "@homarr/db";
+import { items } from "@homarr/db/schema/sqlite";
+import { JellyfinIntegration } from "@homarr/integrations";
+import { createItemAndIntegrationChannel } from "@homarr/redis";
+
+import { createCronJob } from "../../lib";
+
+export const mediaServerJob = createCronJob("mediaServer", 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 channel = createItemAndIntegrationChannel("mediaServer", integration.integrationId);
+      await channel.publishAndUpdateLastStateAsync(streamSessions);
+    }
+  }
+});

packages/definitions/src/widget.ts

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

packages/integrations/package.json

@@ -23,10 +23,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",
     "@homarr/translation": "workspace:^0.1.0"
   },
   "devDependencies": {

packages/integrations/src/base/creator.ts

@@ -1,6 +1,7 @@
 import type { IntegrationKind } from "@homarr/definitions";
 
 import { HomeAssistantIntegration } from "../homeassistant/homeassistant-integration";
+import { JellyfinIntegration } from "../jellyfin/jellyfin-integration";
 import { SonarrIntegration } from "../media-organizer/sonarr/sonarr-integration";
 import { PiHoleIntegration } from "../pi-hole/pi-hole-integration";
 import type { IntegrationInput } from "./integration";
@@ -11,6 +12,8 @@ export const integrationCreatorByKind = (kind: IntegrationKind, integration: Int
       return new PiHoleIntegration(integration);
     case "homeAssistant":
       return new HomeAssistantIntegration(integration);
+    case "jellyfin":
+      return new JellyfinIntegration(integration);
     case "sonarr":
       return new SonarrIntegration(integration);
     default:

packages/integrations/src/index.ts

@@ -1,8 +1,12 @@
 // General integrations
 export { PiHoleIntegration } from "./pi-hole/pi-hole-integration";
 export { HomeAssistantIntegration } from "./homeassistant/homeassistant-integration";
+export { JellyfinIntegration } from "./jellyfin/jellyfin-integration";
 export { SonarrIntegration } from "./media-organizer/sonarr/sonarr-integration";
 
+// Types
+export type { StreamSession } from "./interfaces/media-server/session";
+
 // Helpers
 export { IntegrationTestConnectionError } from "./base/test-connection-error";
 export { integrationCreatorByKind } from "./base/creator";

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: string | 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,68 @@
+import { Jellyfin } from "@jellyfin/sdk";
+import { getSessionApi } from "@jellyfin/sdk/lib/utils/api/session-api";
+import { getSystemApi } from "@jellyfin/sdk/lib/utils/api/system-api";
+
+import { Integration } from "../base/integration";
+import type { StreamSession } from "../interfaces/media-server/session";
+
+export class JellyfinIntegration extends Integration {
+  private readonly jellyfin: Jellyfin = new Jellyfin({
+    clientInfo: {
+      name: "Homarr",
+      version: "0.0.1",
+    },
+    deviceInfo: {
+      name: "Homarr",
+      id: "homarr",
+    },
+  });
+
+  public async testConnectionAsync(): Promise<void> {
+    const api = this.getApi();
+    const systemApi = getSystemApi(api);
+    await systemApi.getPingSystem();
+  }
+
+  public async getCurrentSessionsAsync(): Promise<StreamSession[]> {
+    const api = this.getApi();
+    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: `${this.integration.url}/Users/${sessionInfo.UserId}/Images/Primary`,
+          userId: sessionInfo.UserId ?? "",
+          username: sessionInfo.UserName ?? "",
+        },
+        currentlyPlaying: nowPlaying,
+      };
+    });
+  }
+
+  private getApi() {
+    const apiKey = this.getSecretValue("apiKey");
+    return this.jellyfin.createApi(this.integration.url, apiKey);
+  }
+}

packages/redis/package.json

@@ -25,7 +25,8 @@
     "superjson": "2.2.1",
     "@homarr/log": "workspace:^",
     "@homarr/db": "workspace:^",
-    "@homarr/common": "workspace:^"
+    "@homarr/common": "workspace:^",
+    "@homarr/definitions": "workspace:^"
   },
   "devDependencies": {
     "@homarr/eslint-config": "workspace:^0.2.0",

packages/redis/src/index.ts

@@ -1,6 +1,6 @@
 import { createListChannel, createQueueChannel, createSubPubChannel } from "./lib/channel";
 
-export { createCacheChannel, createItemWithIntegrationChannel } from "./lib/channel";
+export { createCacheChannel, createItemAndIntegrationChannel } from "./lib/channel";
 
 export const exampleChannel = createSubPubChannel<{ message: string }>("example");
 export const pingChannel = createSubPubChannel<{ url: string; statusCode: number } | { url: string; error: string }>(

packages/redis/src/lib/channel.ts

@@ -1,6 +1,7 @@
 import superjson from "superjson";
 
 import { createId } from "@homarr/db";
+import type { WidgetKind } from "@homarr/definitions";
 import { logger } from "@homarr/log";
 
 import { createRedisConnection } from "./connection";
@@ -168,8 +169,35 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
   };
 };
 
-export const createItemWithIntegrationChannel = <T>(itemId: string, integrationId: string) =>
-  createCacheChannel<T>(`item:${itemId}:integration:${integrationId}`);
+export const createItemAndIntegrationChannel = <TData>(kind: WidgetKind, integrationId: string) => {
+  const channelName = `item:${kind}:integration:${integrationId}`;
+  return {
+    subscribeAsync: async (callback: (data: TData) => void) => {
+      await subscriber.subscribe(channelName);
+      subscriber.on("message", (channel, message) => {
+        if (channel !== channelName) {
+          logger.warn(`received message on ${channel} channel but was looking for ${channelName}`);
+          return;
+        }
+        callback(superjson.parse(message));
+        logger.debug(`sent message on ${channelName}`);
+      });
+    },
+    publishAndUpdateLastStateAsync: async (data: TData) => {
+      await publisher.publish(channelName, superjson.stringify(data));
+      await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() }));
+    },
+    setAsync: async (data: TData) => {
+      await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() }));
+    },
+    getAsync: async () => {
+      const data = await getSetClient.get(channelName);
+      if (!data) return null;
+
+      return superjson.parse<{ data: TData; timestamp: Date }>(data);
+    },
+  };
+};
 
 const queueClient = createRedisConnection();