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

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();