feat(media-transcoding): add periodic live updates (#4166)

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

@@ -1,4 +1,7 @@
+import { observable } from "@trpc/server/observable";
+
 import { getIntegrationKindsByCategory } from "@homarr/definitions";
+import type { MediaTranscoding } from "@homarr/request-handler/media-transcoding";
 import { mediaTranscodingRequestHandler } from "@homarr/request-handler/media-transcoding";
 import { paginatedSchema } from "@homarr/validation/common";
 
@@ -15,7 +18,7 @@ export const mediaTranscodingRouter = createTRPCRouter({
     .input(paginatedSchema.pick({ page: true, pageSize: true }))
     .query(async ({ ctx, input }) => {
       const innerHandler = mediaTranscodingRequestHandler.handler(ctx.integration, {
-        pageOffset: input.page,
+        pageOffset: (input.page - 1) * input.pageSize,
         pageSize: input.pageSize,
       });
       const { data } = await innerHandler.getCachedOrUpdatedDataAsync({ forceUpdate: false });
@@ -25,4 +28,19 @@ export const mediaTranscodingRouter = createTRPCRouter({
         data,
       };
     }),
+  subscribeData: publicProcedure
+    .concat(createIndexerManagerIntegrationMiddleware("query"))
+    .input(paginatedSchema.pick({ page: true, pageSize: true }))
+    .subscription(({ ctx, input }) => {
+      return observable<{ integrationId: string; data: MediaTranscoding }>((emit) => {
+        const innerHandler = mediaTranscodingRequestHandler.handler(ctx.integration, {
+          pageOffset: (input.page - 1) * input.pageSize,
+          pageSize: input.pageSize,
+        });
+        const unsubscribe = innerHandler.subscribe((data) => {
+          emit.next({ integrationId: input.integrationId, data });
+        });
+        return unsubscribe;
+      });
+    }),
 });

packages/request-handler/src/media-transcoding.ts

@@ -7,7 +7,7 @@ import type { TdarrQueue, TdarrStatistics, TdarrWorker } from "@homarr/integrati
 import { createCachedIntegrationRequestHandler } from "./lib/cached-integration-request-handler";
 
 export const mediaTranscodingRequestHandler = createCachedIntegrationRequestHandler<
-  { queue: TdarrQueue; workers: TdarrWorker[]; statistics: TdarrStatistics },
+  MediaTranscoding,
   IntegrationKindByCategory<"mediaTranscoding">,
   { pageOffset: number; pageSize: number }
 >({
@@ -22,3 +22,9 @@ export const mediaTranscodingRequestHandler = createCachedIntegrationRequestHand
     };
   },
 });
+
+export interface MediaTranscoding {
+  queue: TdarrQueue;
+  workers: TdarrWorker[];
+  statistics: TdarrStatistics;
+}

packages/widgets/src/media-transcoding/component.tsx

@@ -30,18 +30,23 @@ export default function MediaTranscodingWidget({
 }: WidgetComponentProps<"mediaTranscoding">) {
   const [queuePage, setQueuePage] = useState(1);
   const queuePageSize = 10;
-  const [transcodingData] = clientApi.widget.mediaTranscoding.getDataAsync.useSuspenseQuery(
-    {
-      integrationId: integrationIds[0] ?? "",
-      pageSize: queuePageSize,
-      page: queuePage,
+  const input = {
+    integrationId: integrationIds[0] ?? "",
+    pageSize: queuePageSize,
+    page: queuePage,
+  };
+  const [transcodingData] = clientApi.widget.mediaTranscoding.getDataAsync.useSuspenseQuery(input, {
+    refetchOnMount: false,
+    refetchOnWindowFocus: false,
+    refetchOnReconnect: false,
+  });
+
+  const utils = clientApi.useUtils();
+  clientApi.widget.mediaTranscoding.subscribeData.useSubscription(input, {
+    onData(data) {
+      utils.widget.mediaTranscoding.getDataAsync.setData(input, data);
     },
-    {
-      refetchOnMount: false,
-      refetchOnWindowFocus: false,
-      refetchOnReconnect: false,
-    },
-  );
+  });
 
   const [view, setView] = useState<View>(options.defaultView);
   const totalQueuePages = Math.ceil((transcodingData.data.queue.totalCount || 1) / queuePageSize);