refactor: add request handlers for centralized cached requests (#1504)

* feat: add object base64 hash method

* chore: add script to add package

* feat: add request-handler package

* wip: add request handlers for all jobs and widget api procedures

* wip: remove errors shown in logs, add missing decryption for secrets in cached-request-job-handler

* wip: highly improve request handler, add request handlers for calendar, media-server, indexer-manager and more, add support for multiple inputs from job handler creator

* refactor: move media-server requests to request-handler, add invalidation logic for dns-hole and media requests

* refactor: remove unused integration item middleware

* feat: add invalidation to switch entity action of smart-home

* fix: lint issues

* chore: use integration-kind-by-category instead of union for request-handlers

* fix: build not working for tasks and websocket

* refactor: add more logs

* refactor: readd timestamp logic for diconnect status

* fix: lint and typecheck issue

* chore: address pull request feedback

packages/request-handler/src/lib/cached-integration-request-handler.ts

@@ -0,0 +1,42 @@
+import type { Duration } from "dayjs/plugin/duration";
+
+import type { Modify } from "@homarr/common/types";
+import type { Integration, IntegrationSecret } from "@homarr/db/schema/sqlite";
+import type { IntegrationKind } from "@homarr/definitions";
+import { createIntegrationOptionsChannel } from "@homarr/redis";
+
+import { createCachedRequestHandler } from "./cached-request-handler";
+
+type IntegrationOfKind<TKind extends IntegrationKind> = Omit<Integration, "kind"> & {
+  kind: TKind;
+  decryptedSecrets: Modify<Pick<IntegrationSecret, "kind" | "value">, { value: string }>[];
+};
+
+interface Options<TData, TKind extends IntegrationKind, TInput extends Record<string, unknown>> {
+  // Unique key for this request handler
+  queryKey: string;
+  requestAsync: (integration: IntegrationOfKind<TKind>, input: TInput) => Promise<TData>;
+  cacheDuration: Duration;
+}
+
+export const createCachedIntegrationRequestHandler = <
+  TData,
+  TKind extends IntegrationKind,
+  TInput extends Record<string, unknown>,
+>(
+  options: Options<TData, TKind, TInput>,
+) => {
+  return {
+    handler: (integration: IntegrationOfKind<TKind>, itemOptions: TInput) =>
+      createCachedRequestHandler({
+        queryKey: options.queryKey,
+        requestAsync: async (input: { options: TInput; integration: IntegrationOfKind<TKind> }) => {
+          return await options.requestAsync(input.integration, input.options);
+        },
+        cacheDuration: options.cacheDuration,
+        createRedisChannel(input, options) {
+          return createIntegrationOptionsChannel<TData>(input.integration.id, options.queryKey, input.options);
+        },
+      }).handler({ options: itemOptions, integration }),
+  };
+};

packages/request-handler/src/lib/cached-request-handler.ts

@@ -0,0 +1,74 @@
+import dayjs from "dayjs";
+import type { Duration } from "dayjs/plugin/duration";
+
+import { logger } from "@homarr/log";
+import type { createChannelWithLatestAndEvents } from "@homarr/redis";
+
+interface Options<TData, TInput extends Record<string, unknown>> {
+  // Unique key for this request handler
+  queryKey: string;
+  requestAsync: (input: TInput) => Promise<TData>;
+  createRedisChannel: (
+    input: TInput,
+    options: Options<TData, TInput>,
+  ) => ReturnType<typeof createChannelWithLatestAndEvents<TData>>;
+  cacheDuration: Duration;
+}
+
+export const createCachedRequestHandler = <TData, TInput extends Record<string, unknown>>(
+  options: Options<TData, TInput>,
+) => {
+  return {
+    handler: (input: TInput) => {
+      const channel = options.createRedisChannel(input, options);
+
+      return {
+        async getCachedOrUpdatedDataAsync({ forceUpdate = false }) {
+          const requestNewDataAsync = async () => {
+            const data = await options.requestAsync(input);
+            await channel.publishAndUpdateLastStateAsync(data);
+            return {
+              data,
+              timestamp: new Date(),
+            };
+          };
+
+          if (forceUpdate) {
+            logger.debug(
+              `Cached request handler forced update for channel='${channel.name}' queryKey='${options.queryKey}'`,
+            );
+            return await requestNewDataAsync();
+          }
+
+          const channelData = await channel.getAsync();
+
+          const shouldRequestNewData =
+            !channelData ||
+            dayjs().diff(channelData.timestamp, "milliseconds") > options.cacheDuration.asMilliseconds();
+
+          if (shouldRequestNewData) {
+            logger.debug(
+              `Cached request handler cache miss for channel='${channel.name}' queryKey='${options.queryKey}' reason='${!channelData ? "no data" : "cache expired"}'`,
+            );
+            return await requestNewDataAsync();
+          }
+
+          logger.debug(
+            `Cached request handler cache hit for channel='${channel.name}' queryKey='${options.queryKey}' expiresAt='${dayjs(channelData.timestamp).add(options.cacheDuration).toISOString()}'`,
+          );
+
+          return channelData;
+        },
+        async invalidateAsync() {
+          logger.debug(
+            `Cached request handler invalidating cache channel='${channel.name}' queryKey='${options.queryKey}'`,
+          );
+          await this.getCachedOrUpdatedDataAsync({ forceUpdate: true });
+        },
+        subscribe(callback: (data: TData) => void) {
+          return channel.subscribe(callback);
+        },
+      };
+    },
+  };
+};

packages/request-handler/src/lib/cached-request-integration-job-handler.ts

@@ -0,0 +1,103 @@
+import SuperJSON from "superjson";
+
+import { hashObjectBase64, Stopwatch } from "@homarr/common";
+import { decryptSecret } from "@homarr/common/server";
+import type { MaybeArray } from "@homarr/common/types";
+import { db } from "@homarr/db";
+import { getItemsWithIntegrationsAsync } from "@homarr/db/queries";
+import type { WidgetKind } from "@homarr/definitions";
+import { logger } from "@homarr/log";
+
+// This imports are done that way to avoid circular dependencies.
+import type { inferSupportedIntegrationsStrict } from "../../../widgets/src";
+import { reduceWidgetOptionsWithDefaultValues } from "../../../widgets/src";
+import type { WidgetComponentProps } from "../../../widgets/src/definition";
+import type { createCachedIntegrationRequestHandler } from "./cached-integration-request-handler";
+
+export const createRequestIntegrationJobHandler = <
+  TWidgetKind extends WidgetKind,
+  TIntegrationKind extends inferSupportedIntegrationsStrict<TWidgetKind>,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  THandler extends ReturnType<typeof createCachedIntegrationRequestHandler<any, TIntegrationKind, any>>["handler"],
+>(
+  handler: THandler,
+  {
+    widgetKinds,
+    getInput,
+  }: {
+    widgetKinds: TWidgetKind[];
+    getInput: {
+      [key in TWidgetKind]: (options: WidgetComponentProps<key>["options"]) => MaybeArray<Parameters<THandler>[1]>;
+    };
+  },
+) => {
+  return async () => {
+    const itemsForIntegration = await getItemsWithIntegrationsAsync(db, {
+      kinds: widgetKinds,
+    });
+
+    logger.debug(
+      `Found items for integration widgetKinds='${widgetKinds.join(",")}' count=${itemsForIntegration.length}`,
+    );
+
+    const distinctIntegrations: {
+      integrationId: string;
+      inputHash: string;
+      integration: (typeof itemsForIntegration)[number]["integrations"][number]["integration"];
+      input: Parameters<THandler>[1];
+    }[] = [];
+
+    for (const itemForIntegration of itemsForIntegration) {
+      const oneOrMultipleInputs = getInput[itemForIntegration.kind](
+        reduceWidgetOptionsWithDefaultValues(
+          itemForIntegration.kind,
+          SuperJSON.parse(itemForIntegration.options),
+        ) as never,
+      );
+      for (const { integration } of itemForIntegration.integrations) {
+        const inputArray = Array.isArray(oneOrMultipleInputs) ? oneOrMultipleInputs : [oneOrMultipleInputs];
+
+        for (const input of inputArray) {
+          const inputHash = hashObjectBase64(input);
+          if (
+            distinctIntegrations.some(
+              (distinctIntegration) =>
+                distinctIntegration.integrationId === integration.id && distinctIntegration.inputHash === inputHash,
+            )
+          ) {
+            continue;
+          }
+
+          distinctIntegrations.push({ integrationId: integration.id, inputHash, integration, input });
+        }
+      }
+    }
+
+    for (const { integrationId, integration, input, inputHash } of distinctIntegrations) {
+      try {
+        const decryptedSecrets = integration.secrets.map((secret) => ({
+          ...secret,
+          value: decryptSecret(secret.value),
+        }));
+
+        const innerHandler = handler(
+          {
+            ...integration,
+            kind: integration.kind as TIntegrationKind,
+            decryptedSecrets,
+          },
+          input,
+        );
+        const stopWatch = new Stopwatch();
+        await innerHandler.getCachedOrUpdatedDataAsync({ forceUpdate: true });
+        logger.debug(
+          `Ran integration job integration=${integrationId} inputHash='${inputHash}' elapsed=${stopWatch.getElapsedInHumanWords()}`,
+        );
+      } catch (error) {
+        logger.error(
+          `Failed to run integration job integration=${integrationId} inputHash='${inputHash}' error=${error as string}`,
+        );
+      }
+    }
+  };
+};