diff --git a/apps/nextjs/package.json b/apps/nextjs/package.json index ed76c58dc..a68484719 100644 --- a/apps/nextjs/package.json +++ b/apps/nextjs/package.json @@ -37,7 +37,6 @@ "@homarr/integrations": "workspace:^0.1.0", "@homarr/modals": "workspace:^0.1.0", "@homarr/modals-collection": "workspace:^0.1.0", - "@homarr/next-cache-handler": "workspace:^0.1.0", "@homarr/notifications": "workspace:^0.1.0", "@homarr/old-import": "workspace:^0.1.0", "@homarr/old-schema": "workspace:^0.1.0", diff --git a/packages/api/src/cache-invalidation.ts b/packages/api/src/cache-invalidation.ts deleted file mode 100644 index fa40d47ab..000000000 --- a/packages/api/src/cache-invalidation.ts +++ /dev/null @@ -1,51 +0,0 @@ -// ponytail: invalidation is wired now; `cacheTag()` on RSC read paths is Phase 1 follow-up. -// Until reads use `"use cache"` + `cacheTag()`, these calls affect only Next.js-internal -// cached segments (layout shells, static params) via cacheComponents, not cross-request DB queries. -// -// Two-tier invalidation: -// - `invalidateIntegrationCache`: for integration CONFIG changes (CRUD, permissions). -// Clears module cache + Next tags + Redis session store. -// - `invalidateIntegrationDataCache` (from integration-data-cache.ts): for integration DATA changes -// (widget interact mutations). Clears module cache only — pair with requestHandler.invalidateCache(). - -import { cacheTags } from "./cache-tags"; -import { invalidateIntegrationDataCache } from "./integration-data-cache"; - -// ponytail: revalidateTag throws outside Next.js request context (tests, tasks, websocket). -// Ceiling: if Next.js context detection changes, update the require/catch. -function isExpectedRevalidateError(error: unknown) { - const message = error instanceof Error ? error.message : ""; - return message.includes("static generation") || message.includes("was called outside a request scope"); -} - -function safeRevalidateTag(tag: string) { - try { - // eslint-disable-next-line @typescript-eslint/no-require-imports - const { revalidateTag } = require("next/cache") as typeof import("next/cache"); - revalidateTag(tag, "max"); - } catch (error) { - if (!isExpectedRevalidateError(error)) { - const message = error instanceof Error ? error.message : String(error); - console.warn(`[cache-invalidation] revalidateTag("${tag}") failed:`, message); - } - } -} - -export const invalidateBoardCache = (boardId: string, boardName?: string) => { - safeRevalidateTag(cacheTags.board(boardId)); - if (boardName) safeRevalidateTag(cacheTags.boardByName(boardName)); - safeRevalidateTag(cacheTags.boardList()); -}; - -export const invalidateIntegrationCache = (integrationId: string) => { - invalidateIntegrationDataCache(integrationId); - safeRevalidateTag(cacheTags.integration(integrationId)); -}; - -export const invalidateUserCache = (userId: string) => { - safeRevalidateTag(cacheTags.user(userId)); -}; - -export const invalidateServerSettingsCache = () => { - safeRevalidateTag(cacheTags.serverSettings()); -}; diff --git a/packages/api/src/cache-tags.spec.ts b/packages/api/src/cache-tags.spec.ts deleted file mode 100644 index eb7552d17..000000000 --- a/packages/api/src/cache-tags.spec.ts +++ /dev/null @@ -1,19 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { cacheTags } from "./cache-tags"; - -describe("cacheTags", () => { - it("produces stable tag shapes", () => { - expect(cacheTags.serverSettings()).toBe("server-settings"); - expect(cacheTags.board("abc")).toBe("board:abc"); - expect(cacheTags.boardByName("My Board")).toBe("board-name:MY BOARD"); - expect(cacheTags.boardList()).toBe("board-list"); - expect(cacheTags.user("u1")).toBe("user:u1"); - expect(cacheTags.integration("i1")).toBe("integration:i1"); - }); - - it("boardByName uppercases consistently", () => { - expect(cacheTags.boardByName("home")).toBe(cacheTags.boardByName("HOME")); - expect(cacheTags.boardByName("Home")).toBe(cacheTags.boardByName("hOmE")); - }); -}); diff --git a/packages/api/src/cache-tags.ts b/packages/api/src/cache-tags.ts deleted file mode 100644 index 0df0f7271..000000000 --- a/packages/api/src/cache-tags.ts +++ /dev/null @@ -1,8 +0,0 @@ -export const cacheTags = { - serverSettings: () => "server-settings" as const, - board: (boardId: string) => `board:${boardId}` as const, - boardByName: (name: string) => `board-name:${name.toUpperCase()}` as const, - boardList: () => "board-list" as const, - user: (userId: string) => `user:${userId}` as const, - integration: (integrationId: string) => `integration:${integrationId}` as const, -} as const; diff --git a/packages/api/src/router/board.ts b/packages/api/src/router/board.ts index 281d5f24a..88a1c42be 100644 --- a/packages/api/src/router/board.ts +++ b/packages/api/src/router/board.ts @@ -59,7 +59,6 @@ import { zodUnionFromArray } from "@homarr/validation/enums"; import type { BoardItemAdvancedOptions } from "@homarr/validation/shared"; import { sectionSchema, sharedItemSchema } from "@homarr/validation/shared"; -import { invalidateBoardCache, invalidateUserCache } from "../cache-invalidation"; import { createTRPCRouter, permissionRequiredProcedure, protectedProcedure, publicProcedure } from "../trpc"; import { throwIfActionForbiddenAsync } from "./board/board-access"; import { generateResponsiveGridFor } from "./board/grid-algorithm"; @@ -326,11 +325,8 @@ export const boardRouter = createTRPCRouter({ if (!user?.homeBoardId) { await ctx.db.update(users).set({ homeBoardId: boardId }).where(eq(users.id, ctx.session.user.id)); - invalidateUserCache(ctx.session.user.id); } - invalidateBoardCache(boardId, input.name); - return { boardId }; }), duplicateBoard: permissionRequiredProcedure @@ -559,8 +555,6 @@ export const boardRouter = createTRPCRouter({ }, }); - invalidateBoardCache(newBoardId, input.name); - return { boardId: newBoardId }; }), renameBoard: protectedProcedure @@ -579,17 +573,7 @@ export const boardRouter = createTRPCRouter({ await noBoardWithSimilarNameAsync(ctx.db, input.name, [input.id]); - const existingBoard = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.id), - columns: { name: true }, - }); - await ctx.db.update(boards).set({ name: input.name }).where(eq(boards.id, input.id)); - - invalidateBoardCache(input.id, input.name); - if (existingBoard && existingBoard.name !== input.name) { - invalidateBoardCache(input.id, existingBoard.name); - } }), changeBoardVisibility: protectedProcedure .meta({ @@ -604,12 +588,6 @@ export const boardRouter = createTRPCRouter({ .output(z.void()) .mutation(async ({ ctx, input }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "full"); - - const existingBoard = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.id), - columns: { name: true }, - }); - const boardSettings = await getServerSettingByKeyAsync(ctx.db, "board"); if ( @@ -626,8 +604,6 @@ export const boardRouter = createTRPCRouter({ .update(boards) .set({ isPublic: input.visibility === "public" }) .where(eq(boards.id, input.id)); - - invalidateBoardCache(input.id, existingBoard?.name); }), deleteBoard: protectedProcedure .meta({ @@ -643,22 +619,7 @@ export const boardRouter = createTRPCRouter({ .mutation(async ({ ctx, input }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "full"); - const existingBoard = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.id), - columns: { name: true }, - }); - - const affectedUsers = await ctx.db.query.users.findMany({ - columns: { id: true }, - where: eq(users.homeBoardId, input.id), - }); - await ctx.db.delete(boards).where(eq(boards.id, input.id)); - - for (const user of affectedUsers) { - invalidateUserCache(user.id); - } - invalidateBoardCache(input.id, existingBoard?.name); }), setHomeBoard: protectedProcedure .meta({ @@ -675,8 +636,6 @@ export const boardRouter = createTRPCRouter({ await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "view"); await ctx.db.update(users).set({ homeBoardId: input.id }).where(eq(users.id, ctx.session.user.id)); - - invalidateUserCache(ctx.session.user.id); }), setMobileHomeBoard: protectedProcedure .meta({ @@ -693,8 +652,6 @@ export const boardRouter = createTRPCRouter({ await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "view"); await ctx.db.update(users).set({ mobileHomeBoardId: input.id }).where(eq(users.id, ctx.session.user.id)); - - invalidateUserCache(ctx.session.user.id); }), getHomeBoard: publicProcedure.query(async ({ ctx }) => { const userId = ctx.session?.user.id; @@ -849,8 +806,6 @@ export const boardRouter = createTRPCRouter({ if (removedLayoutIds.length > 0) { await ctx.db.delete(layouts).where(inArray(layouts.id, removedLayoutIds)); } - - invalidateBoardCache(board.id, board.name); }), savePartialBoardSettings: protectedProcedure .meta({ @@ -866,11 +821,6 @@ export const boardRouter = createTRPCRouter({ .mutation(async ({ ctx, input }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "modify"); - const existingBoard = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.id), - columns: { name: true }, - }); - await ctx.db .update(boards) .set({ @@ -900,8 +850,6 @@ export const boardRouter = createTRPCRouter({ disableStatus: input.disableStatus, }) .where(eq(boards.id, input.id)); - - invalidateBoardCache(input.id, existingBoard?.name); }), saveBoard: protectedProcedure.input(boardSaveSchema).mutation(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "modify"); @@ -1318,8 +1266,6 @@ export const boardRouter = createTRPCRouter({ }); }, }); - - invalidateBoardCache(dbBoard.id, dbBoard.name); }), getBoardPermissions: protectedProcedure.input(byIdSchema).query(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.id), "full"); @@ -1396,16 +1342,6 @@ export const boardRouter = createTRPCRouter({ saveUserBoardPermissions: protectedProcedure.input(boardSavePermissionsSchema).mutation(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.entityId), "full"); - const board = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.entityId), - columns: { name: true }, - }); - - const existingUserPermissions = await ctx.db.query.boardUserPermissions.findMany({ - columns: { userId: true }, - where: eq(boardUserPermissions.boardId, input.entityId), - }); - await handleTransactionsAsync(ctx.db, { async handleAsync(db, schema) { await db.transaction(async (transaction) => { @@ -1441,31 +1377,10 @@ export const boardRouter = createTRPCRouter({ }); }, }); - - const userIds = [ - ...new Set([ - ...existingUserPermissions.map((permission) => permission.userId), - ...input.permissions.map((permission) => permission.principalId), - ]), - ]; - for (const userId of userIds) { - invalidateUserCache(userId); - } - invalidateBoardCache(input.entityId, board?.name); }), saveGroupBoardPermissions: protectedProcedure.input(boardSavePermissionsSchema).mutation(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(boards.id, input.entityId), "full"); - const board = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.entityId), - columns: { name: true }, - }); - - const existingGroupPermissions = await ctx.db.query.boardGroupPermissions.findMany({ - columns: { groupId: true }, - where: eq(boardGroupPermissions.boardId, input.entityId), - }); - await handleTransactionsAsync(ctx.db, { async handleAsync(db, schema) { await db.transaction(async (transaction) => { @@ -1503,23 +1418,6 @@ export const boardRouter = createTRPCRouter({ }); }, }); - - const groupIds = [ - ...new Set([ - ...existingGroupPermissions.map((permission) => permission.groupId), - ...input.permissions.map((permission) => permission.principalId), - ]), - ]; - if (groupIds.length > 0) { - const groupMembersForBoard = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: inArray(groupMembers.groupId, groupIds), - }); - for (const member of groupMembersForBoard) { - invalidateUserCache(member.userId); - } - } - invalidateBoardCache(input.entityId, board?.name); }), importOldmarrConfig: permissionRequiredProcedure .requiresPermission("board-create") @@ -1527,10 +1425,7 @@ export const boardRouter = createTRPCRouter({ .mutation(async ({ input, ctx }) => { const content = await input.file.text(); const oldmarr = oldmarrConfigSchema.parse(JSON.parse(content)); - const importedBoards = await importOldmarrAsync(ctx.db, oldmarr, input.configuration); - for (const board of importedBoards) { - invalidateBoardCache(board.id, board.name); - } + await importOldmarrAsync(ctx.db, oldmarr, input.configuration); }), addItem: protectedProcedure .meta({ @@ -1659,8 +1554,6 @@ export const boardRouter = createTRPCRouter({ .values(input.integrationIds.map((integrationId) => ({ itemId, integrationId }))); } - invalidateBoardCache(input.boardId, board.name); - return { itemId }; }), }); diff --git a/packages/api/src/router/group.ts b/packages/api/src/router/group.ts index 0041bce5d..7c00ff125 100644 --- a/packages/api/src/router/group.ts +++ b/packages/api/src/router/group.ts @@ -4,7 +4,7 @@ import { z } from "zod/v4"; import { canManageGroupMembersLocally, isGroupMembershipManagedLocally } from "@homarr/auth/server"; import { createId } from "@homarr/common"; import type { Database } from "@homarr/db"; -import { and, eq, handleTransactionsAsync, inArray, like, not } from "@homarr/db"; +import { and, eq, handleTransactionsAsync, like, not } from "@homarr/db"; import { getMaxGroupPositionAsync } from "@homarr/db/queries"; import { groupMembers, groupPermissions, groups, users } from "@homarr/db/schema"; import { everyoneGroup } from "@homarr/definitions"; @@ -19,7 +19,6 @@ import { } from "@homarr/validation/group"; import { createTRPCRouter, onboardingProcedure, permissionRequiredProcedure, protectedProcedure } from "../trpc"; -import { invalidateUserCache } from "../cache-invalidation"; import { nextOnboardingStepAsync } from "./onboard/onboard-queries"; export const groupRouter = createTRPCRouter({ @@ -232,14 +231,6 @@ export const groupRouter = createTRPCRouter({ name: input.name, }) .where(eq(groups.id, input.id)); - - const members = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: eq(groupMembers.groupId, input.id), - }); - for (const member of members) { - invalidateUserCache(member.userId); - } }), savePartialSettings: permissionRequiredProcedure .requiresPermission("admin") @@ -254,14 +245,6 @@ export const groupRouter = createTRPCRouter({ mobileHomeBoardId: input.settings.mobileHomeBoardId, }) .where(eq(groups.id, input.id)); - - const members = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: eq(groupMembers.groupId, input.id), - }); - for (const member of members) { - invalidateUserCache(member.userId); - } }), savePositions: permissionRequiredProcedure .requiresPermission("admin") @@ -285,14 +268,6 @@ export const groupRouter = createTRPCRouter({ }); }, }); - - const members = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: inArray(groupMembers.groupId, input.positions), - }); - for (const userId of new Set(members.map((member) => member.userId))) { - invalidateUserCache(userId); - } }), savePermissions: permissionRequiredProcedure .requiresPermission("admin") @@ -300,11 +275,6 @@ export const groupRouter = createTRPCRouter({ .mutation(async ({ input, ctx }) => { await throwIfGroupNotFoundAsync(ctx.db, input.groupId); - const members = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: eq(groupMembers.groupId, input.groupId), - }); - await ctx.db.delete(groupPermissions).where(eq(groupPermissions.groupId, input.groupId)); if (input.permissions.length > 0) { @@ -315,10 +285,6 @@ export const groupRouter = createTRPCRouter({ })), ); } - - for (const member of members) { - invalidateUserCache(member.userId); - } }), transferOwnership: permissionRequiredProcedure .requiresPermission("admin") @@ -327,22 +293,12 @@ export const groupRouter = createTRPCRouter({ await throwIfGroupNotFoundAsync(ctx.db, input.groupId); await throwIfGroupNameIsReservedAsync(ctx.db, input.groupId); - const group = await ctx.db.query.groups.findFirst({ - columns: { ownerId: true }, - where: eq(groups.id, input.groupId), - }); - await ctx.db .update(groups) .set({ ownerId: input.userId, }) .where(eq(groups.id, input.groupId)); - - invalidateUserCache(input.userId); - if (group?.ownerId && group.ownerId !== input.userId) { - invalidateUserCache(group.ownerId); - } }), deleteGroup: permissionRequiredProcedure .requiresPermission("admin") @@ -351,16 +307,7 @@ export const groupRouter = createTRPCRouter({ await throwIfGroupNotFoundAsync(ctx.db, input.id); await throwIfGroupNameIsReservedAsync(ctx.db, input.id); - const members = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: eq(groupMembers.groupId, input.id), - }); - await ctx.db.delete(groups).where(eq(groups.id, input.id)); - - for (const member of members) { - invalidateUserCache(member.userId); - } }), addMember: permissionRequiredProcedure .requiresPermission("admin") @@ -392,8 +339,6 @@ export const groupRouter = createTRPCRouter({ groupId: input.groupId, userId: input.userId, }); - - invalidateUserCache(input.userId); }), removeMember: permissionRequiredProcedure .requiresPermission("admin") @@ -406,8 +351,6 @@ export const groupRouter = createTRPCRouter({ await ctx.db .delete(groupMembers) .where(and(eq(groupMembers.groupId, input.groupId), eq(groupMembers.userId, input.userId))); - - invalidateUserCache(input.userId); }), }); diff --git a/packages/api/src/router/integration/integration-router.ts b/packages/api/src/router/integration/integration-router.ts index 3a86d3f7b..0fcf2acec 100644 --- a/packages/api/src/router/integration/integration-router.ts +++ b/packages/api/src/router/integration/integration-router.ts @@ -39,7 +39,6 @@ import { import { mediaRequestOptionsSchema, mediaRequestRequestSchema } from "@homarr/validation/widgets/media-request"; import { createOneIntegrationMiddleware } from "../../middlewares/integration"; -import { invalidateIntegrationCache, invalidateUserCache } from "../../cache-invalidation"; import { invalidateIntegrationDataCache } from "../../integration-data-cache"; import { createTRPCRouter, permissionRequiredProcedure, protectedProcedure, publicProcedure } from "../../trpc"; import { throwIfActionForbiddenAsync } from "./integration-access"; @@ -374,7 +373,7 @@ export const integrationRouter = createTRPCRouter({ }); await invalidateIntegrationCacheAsync(integrationId); - invalidateIntegrationCache(integrationId); + invalidateIntegrationDataCache(integrationId); }), update: protectedProcedure.input(integrationUpdateSchema).mutation(async ({ ctx, input }) => { await throwIfActionForbiddenAsync(ctx, eq(integrations.id, input.id), "full"); @@ -481,7 +480,7 @@ export const integrationRouter = createTRPCRouter({ // Invalidate all cached data for this integration so that widgets pick up the // new configuration immediately instead of serving stale (or errored) data. await invalidateIntegrationCacheAsync(input.id); - invalidateIntegrationCache(input.id); + invalidateIntegrationDataCache(input.id); }), delete: protectedProcedure .meta({ @@ -506,7 +505,7 @@ export const integrationRouter = createTRPCRouter({ // Clean up any cached data left behind by the deleted integration. await invalidateIntegrationCacheAsync(input.id); - invalidateIntegrationCache(input.id); + invalidateIntegrationDataCache(input.id); }), getIntegrationPermissions: protectedProcedure.input(byIdSchema).query(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(integrations.id, input.id), "full"); @@ -585,11 +584,6 @@ export const integrationRouter = createTRPCRouter({ .mutation(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(integrations.id, input.entityId), "full"); - const existingUserPermissions = await ctx.db.query.integrationUserPermissions.findMany({ - columns: { userId: true }, - where: eq(integrationUserPermissions.integrationId, input.entityId), - }); - await handleTransactionsAsync(ctx.db, { async handleAsync(db, schema) { await ctx.db.transaction(async (transaction) => { @@ -631,27 +625,13 @@ export const integrationRouter = createTRPCRouter({ }, }); - const userIds = [ - ...new Set([ - ...existingUserPermissions.map((permission) => permission.userId), - ...input.permissions.map((permission) => permission.principalId), - ]), - ]; - for (const userId of userIds) { - invalidateUserCache(userId); - } - invalidateIntegrationCache(input.entityId); + invalidateIntegrationDataCache(input.entityId); }), saveGroupIntegrationPermissions: protectedProcedure .input(integrationSavePermissionsSchema) .mutation(async ({ input, ctx }) => { await throwIfActionForbiddenAsync(ctx, eq(integrations.id, input.entityId), "full"); - const existingGroupPermissions = await ctx.db.query.integrationGroupPermissions.findMany({ - columns: { groupId: true }, - where: eq(integrationGroupPermissions.integrationId, input.entityId), - }); - await handleTransactionsAsync(ctx.db, { async handleAsync(db, schema) { await db.transaction(async (transaction) => { @@ -693,22 +673,7 @@ export const integrationRouter = createTRPCRouter({ }, }); - const groupIds = [ - ...new Set([ - ...existingGroupPermissions.map((permission) => permission.groupId), - ...input.permissions.map((permission) => permission.principalId), - ]), - ]; - if (groupIds.length > 0) { - const members = await ctx.db.query.groupMembers.findMany({ - columns: { userId: true }, - where: inArray(groupMembers.groupId, groupIds), - }); - for (const member of members) { - invalidateUserCache(member.userId); - } - } - invalidateIntegrationCache(input.entityId); + invalidateIntegrationDataCache(input.entityId); }), searchInIntegration: protectedProcedure .meta({ diff --git a/packages/api/src/router/section/section-router.ts b/packages/api/src/router/section/section-router.ts index 8a440a80c..81613f581 100644 --- a/packages/api/src/router/section/section-router.ts +++ b/packages/api/src/router/section/section-router.ts @@ -4,7 +4,6 @@ import { z } from "zod/v4"; import { and, eq } from "@homarr/db"; import { sectionCollapseStates, sections } from "@homarr/db/schema"; -import { invalidateUserCache } from "../../cache-invalidation"; import { createTRPCRouter, protectedProcedure } from "../../trpc"; export const sectionRouter = createTRPCRouter({ @@ -38,7 +37,6 @@ export const sectionRouter = createTRPCRouter({ userId: ctx.session.user.id, collapsed: input.collapsed, }); - invalidateUserCache(ctx.session.user.id); return; } @@ -50,7 +48,5 @@ export const sectionRouter = createTRPCRouter({ .where( and(eq(sectionCollapseStates.sectionId, section.id), eq(sectionCollapseStates.userId, ctx.session.user.id)), ); - - invalidateUserCache(ctx.session.user.id); }), }); diff --git a/packages/api/src/router/serverSettings.ts b/packages/api/src/router/serverSettings.ts index 308926475..9783e0ae9 100644 --- a/packages/api/src/router/serverSettings.ts +++ b/packages/api/src/router/serverSettings.ts @@ -14,7 +14,6 @@ import { defaultServerSettingsKeys } from "@homarr/server-settings"; import { settingsInitSchema } from "@homarr/validation/settings"; import { createTRPCRouter, onboardingProcedure, permissionRequiredProcedure, publicProcedure } from "../trpc"; -import { invalidateServerSettingsCache } from "../cache-invalidation"; import { nextOnboardingStepAsync } from "./onboard/onboard-queries"; const boardServerSettingsSchema = z.object({ @@ -92,8 +91,6 @@ export const serverSettingsRouter = createTRPCRouter({ await insertServerSettingByKeyAsync(ctx.db, "board", next); } - invalidateServerSettingsCache(); - return next; }), saveSettings: permissionRequiredProcedure @@ -110,8 +107,6 @@ export const serverSettingsRouter = createTRPCRouter({ ...current, ...input.value, } as ServerSettings[keyof ServerSettings]); - - invalidateServerSettingsCache(); }), initSettings: onboardingProcedure .requiresStep("settings") @@ -121,7 +116,5 @@ export const serverSettingsRouter = createTRPCRouter({ await updateServerSettingByKeyAsync(ctx.db, "analytics", { ...currentAnalytics, ...input.analytics }); await updateServerSettingByKeyAsync(ctx.db, "crawlingAndIndexing", input.crawlingAndIndexing); await nextOnboardingStepAsync(ctx.db, undefined); - - invalidateServerSettingsCache(); }), }); diff --git a/packages/api/src/router/user.ts b/packages/api/src/router/user.ts index 7a46a398a..a7e1735ee 100644 --- a/packages/api/src/router/user.ts +++ b/packages/api/src/router/user.ts @@ -31,7 +31,6 @@ import { } from "@homarr/validation/user"; import { convertIntersectionToZodObject } from "../schema-merger"; -import { invalidateUserCache } from "../cache-invalidation"; import { createTRPCRouter, isDemoMode, @@ -222,8 +221,6 @@ export const userRouter = createTRPCRouter({ image: input.image, }) .where(eq(users.id, input.userId)); - - invalidateUserCache(input.userId); }), getAll: permissionRequiredProcedure .requiresPermission("admin") @@ -455,8 +452,6 @@ export const userRouter = createTRPCRouter({ emailVerified: emailDirty === true ? null : undefined, }) .where(eq(users.id, input.id)); - - invalidateUserCache(input.id); }), delete: protectedProcedure .input(z.object({ userId: z.string() })) @@ -480,7 +475,6 @@ export const userRouter = createTRPCRouter({ } await ctx.db.delete(users).where(eq(users.id, input.userId)); - invalidateUserCache(input.userId); }), changePassword: protectedProcedure .input(convertIntersectionToZodObject(userChangePasswordApiSchema)) @@ -553,7 +547,6 @@ export const userRouter = createTRPCRouter({ password: hashedPassword, }) .where(eq(users.id, input.userId)); - invalidateUserCache(input.userId); }), changeHomeBoards: protectedProcedure .input(convertIntersectionToZodObject(userChangeHomeBoardsSchema.and(z.object({ userId: z.string() })))) @@ -605,8 +598,6 @@ export const userRouter = createTRPCRouter({ mobileHomeBoardId: input.mobileHomeBoardId, }) .where(eq(users.id, input.userId)); - - invalidateUserCache(input.userId); }), changeDefaultSearchEngine: protectedProcedure .input( @@ -662,8 +653,6 @@ export const userRouter = createTRPCRouter({ colorScheme: input.colorScheme, }) .where(eq(users.id, ctx.session.user.id)); - - invalidateUserCache(ctx.session.user.id); }), changeEnableRightClickOnWidgets: protectedProcedure .meta({ @@ -690,8 +679,6 @@ export const userRouter = createTRPCRouter({ enableRightClickOnWidgets: input.enableRightClickOnWidgets, }) .where(eq(users.id, input.id)); - - invalidateUserCache(input.id); }), changePingIconsEnabled: protectedProcedure .input(userPingIconsEnabledSchema.and(byIdSchema)) @@ -710,8 +697,6 @@ export const userRouter = createTRPCRouter({ pingIconsEnabled: input.pingIconsEnabled, }) .where(eq(users.id, input.id)); - - invalidateUserCache(input.id); }), changeDdgBangs: protectedProcedure .input(convertIntersectionToZodObject(userDdgBangsSchema.and(byIdSchema))) @@ -739,8 +724,6 @@ export const userRouter = createTRPCRouter({ ddgBangs: input.ddgBangs, }) .where(eq(users.id, input.id)); - - invalidateUserCache(input.id); }), changeFirstDayOfWeek: protectedProcedure .input(convertIntersectionToZodObject(userFirstDayOfWeekSchema.and(byIdSchema))) @@ -782,8 +765,6 @@ export const userRouter = createTRPCRouter({ firstDayOfWeek: input.firstDayOfWeek, }) .where(eq(users.id, input.id)); - - invalidateUserCache(input.id); }), completeTour: protectedProcedure .input(z.object({ tour: z.enum(["manage", "board"]) })) @@ -794,8 +775,6 @@ export const userRouter = createTRPCRouter({ } as const; await ctx.db.update(users).set(columnByTour[input.tour]).where(eq(users.id, ctx.session.user.id)); - - invalidateUserCache(ctx.session.user.id); }), resetTours: protectedProcedure.mutation(async ({ ctx }) => { await ctx.db @@ -805,8 +784,6 @@ export const userRouter = createTRPCRouter({ completedBoardTour: false, }) .where(eq(users.id, ctx.session.user.id)); - - invalidateUserCache(ctx.session.user.id); }), getTourStatus: protectedProcedure.query(async ({ ctx }) => { if (isDemoMode) { diff --git a/packages/api/src/router/user/change-search-preferences.ts b/packages/api/src/router/user/change-search-preferences.ts index e4c53eab9..41b69bfa4 100644 --- a/packages/api/src/router/user/change-search-preferences.ts +++ b/packages/api/src/router/user/change-search-preferences.ts @@ -8,8 +8,6 @@ import type { Database } from "@homarr/db"; import { users } from "@homarr/db/schema"; import { userChangeSearchPreferencesSchema } from "@homarr/validation/user"; -import { invalidateUserCache } from "../../cache-invalidation"; - export const changeSearchPreferencesInputSchema = userChangeSearchPreferencesSchema.and( z.object({ userId: z.string() }), ); @@ -53,6 +51,4 @@ export const changeSearchPreferencesAsync = async ( ddgBangs: input.ddgBangsEnabled, }) .where(eq(users.id, input.userId)); - - invalidateUserCache(input.userId); }; diff --git a/packages/api/src/router/widgets/options.ts b/packages/api/src/router/widgets/options.ts index 408bc2147..8dc5a0653 100644 --- a/packages/api/src/router/widgets/options.ts +++ b/packages/api/src/router/widgets/options.ts @@ -7,7 +7,6 @@ import { getServerSettingsAsync } from "@homarr/db/queries"; import { boards, items } from "@homarr/db/schema"; import type { WidgetOptionsSettings } from "../../../../widgets/src/definition"; -import { invalidateBoardCache } from "../../cache-invalidation"; import { createTRPCRouter, protectedProcedure, publicProcedure } from "../../trpc"; import { throwIfActionForbiddenAsync } from "../board/board-access"; @@ -52,11 +51,5 @@ export const optionsRouter = createTRPCRouter({ .update(items) .set({ options: SuperJSON.stringify(options) }) .where(eq(items.id, input.itemId)); - - const board = await ctx.db.query.boards.findFirst({ - where: eq(boards.id, input.boardId), - columns: { name: true }, - }); - invalidateBoardCache(input.boardId, board?.name); }), }); diff --git a/packages/next-cache-handler/package.json b/packages/next-cache-handler/package.json deleted file mode 100644 index 22ffd62a4..000000000 --- a/packages/next-cache-handler/package.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "name": "@homarr/next-cache-handler", - "version": "0.1.0", - "private": true, - "type": "module", - "exports": { - ".": "./src/index.ts" - }, - "scripts": { - "clean": "rm -rf .turbo node_modules", - "format": "oxfmt .", - "lint": "oxlint", - "typecheck": "tsc --noEmit" - }, - "dependencies": { - "ioredis": "catalog:", - "next": "catalog:" - }, - "devDependencies": { - "@homarr/tsconfig": "workspace:*", - "typescript": "catalog:", - "vitest": "catalog:" - } -} diff --git a/packages/next-cache-handler/src/index.ts b/packages/next-cache-handler/src/index.ts deleted file mode 100644 index 51cc30d39..000000000 --- a/packages/next-cache-handler/src/index.ts +++ /dev/null @@ -1,24 +0,0 @@ -import type { CacheHandler } from "next/dist/server/lib/cache-handlers/types"; - -import { RedisCacheHandler } from "./redis-handler"; - -let handlerPromise: Promise | undefined; - -const createHandlerAsync = async (): Promise => { - const useRedis = process.env.REDIS_HOST || process.env.REDIS_IS_EXTERNAL === "true"; - if (useRedis) return new RedisCacheHandler(); - - const { createDefaultCacheHandler } = await import("next/dist/server/lib/cache-handlers/default"); - return createDefaultCacheHandler(32 * 1024 * 1024); -}; - -const getHandler = (): Promise => (handlerPromise ??= createHandlerAsync()); - -export default { - get: async (...args: Parameters) => (await getHandler()).get(...args), - set: async (...args: Parameters) => (await getHandler()).set(...args), - refreshTags: async () => (await getHandler()).refreshTags(), - getExpiration: async (...args: Parameters) => - (await getHandler()).getExpiration(...args), - updateTags: async (...args: Parameters) => (await getHandler()).updateTags(...args), -} satisfies CacheHandler; diff --git a/packages/next-cache-handler/src/redis-handler.ts b/packages/next-cache-handler/src/redis-handler.ts deleted file mode 100644 index c0bbf4c87..000000000 --- a/packages/next-cache-handler/src/redis-handler.ts +++ /dev/null @@ -1,236 +0,0 @@ -import Redis from "ioredis"; - -import type { CacheHandler } from "next/dist/server/lib/cache-handlers/types"; - -type CacheEntry = NonNullable>>; - -type StoredEntry = { - tags: string[]; - stale: number; - timestamp: number; - expire: number; - revalidate: number; - body: string; -}; - -// eslint-disable-next-line no-underscore-dangle -const BUILD_ID = process.env.__NEXT_BUILD_ID ?? "dev"; -const TAG_TTL_MS = 7 * 24 * 60 * 60 * 1000; - -async function readStreamBody(stream: ReadableStream) { - const reader = stream.getReader(); - const chunks: Uint8Array[] = []; - let done = false; - while (!done) { - const result = await reader.read(); - done = result.done; - if (result.value) chunks.push(result.value); - } - - return new Uint8Array(Buffer.concat(chunks)); -} - -function entryFromStored(parsed: StoredEntry): CacheEntry { - const body = new Uint8Array(Buffer.from(parsed.body, "base64")); - return { - value: new ReadableStream({ - start(controller) { - controller.enqueue(body); - controller.close(); - }, - }), - tags: parsed.tags, - stale: parsed.stale, - timestamp: parsed.timestamp, - expire: parsed.expire, - revalidate: parsed.revalidate, - }; -} - -export class RedisCacheHandler implements CacheHandler { - private redis: Redis; - private pendingSets = new Map>(); - private tagCache = new Map(); - private connectPromise: Promise | undefined; - private loggedError = false; - - constructor() { - const tlsCa = process.env.REDIS_TLS_CA; - this.redis = new Redis({ - host: process.env.REDIS_HOST ?? "localhost", - port: Number(process.env.REDIS_PORT ?? 6379), - password: process.env.REDIS_PASSWORD || undefined, - username: process.env.REDIS_USERNAME || undefined, - db: Number(process.env.REDIS_CACHE_DB ?? 1), - tls: tlsCa ? { ca: tlsCa } : undefined, - lazyConnect: true, - retryStrategy: (times) => Math.min(times * 100, 3000), - }); - this.redis.on("error", () => { - if (!this.loggedError) { - console.warn("[next-cache-handler] Redis connection error, cache will degrade to always-miss"); - this.loggedError = true; - } - }); - } - - private entryKey(cacheKey: string) { - return `nextCache:v1:${BUILD_ID}:${cacheKey}`; - } - - private tagKey(tag: string) { - return `nextCache:tag:${tag}`; - } - - private ensureConnected() { - return (this.connectPromise ??= this.redis.connect().catch(() => {})); - } - - async get(cacheKey: string, softTags: string[]): Promise { - const pending = this.pendingSets.get(cacheKey); - if (pending) { - try { - const entry = await pending; - const [forCaller, forRest] = entry.value.tee(); - entry.value = forRest; - return { ...entry, value: forCaller }; - } catch { - return undefined; - } - } - - try { - await this.ensureConnected(); - const raw = await this.redis.getBuffer(this.entryKey(cacheKey)); - if (!raw) return undefined; - - const parsed = JSON.parse(raw.toString()) as StoredEntry; - - if (parsed.expire <= 0) { - await this.redis.del(this.entryKey(cacheKey)).catch(() => {}); - return undefined; - } - - const expiration = await this.getExpiration(parsed.tags.concat(softTags)); - if (expiration !== 0 && expiration > parsed.timestamp) { - return undefined; - } - - return entryFromStored(parsed); - } catch { - return undefined; - } - } - - async set(cacheKey: string, pendingEntry: Promise): Promise { - const safeEntry = pendingEntry.then((entry) => { - const [forCallers, forPersist] = entry.value.tee(); - entry.value = forCallers; - return { entry, forPersist }; - }); - const callerPromise = safeEntry.then(({ entry }) => entry); - this.pendingSets.set(cacheKey, callerPromise); - - try { - const { entry, forPersist } = await safeEntry; - const body = await readStreamBody(forPersist); - - const ttlMs = entry.expire * 1000; - if (ttlMs <= 0) return; - - const serialized = JSON.stringify({ - tags: entry.tags, - stale: entry.stale, - timestamp: entry.timestamp, - expire: entry.expire, - revalidate: entry.revalidate, - body: Buffer.from(body).toString("base64"), - }); - - await this.ensureConnected(); - await this.redis.set(this.entryKey(cacheKey), serialized, "PX", ttlMs); - } catch { - // Fail open - } finally { - if (this.pendingSets.get(cacheKey) === callerPromise) { - this.pendingSets.delete(cacheKey); - } - } - } - - async refreshTags(): Promise { - try { - await this.ensureConnected(); - this.tagCache.clear(); - let cursor = "0"; - do { - const [nextCursor, keys] = await this.redis.scan(cursor, "MATCH", "nextCache:tag:*", "COUNT", 100); - cursor = nextCursor; - if (keys.length === 0) continue; - - const values = await this.redis.mget(...keys); - for (let i = 0; i < keys.length; i++) { - const tag = keys[i]?.replace("nextCache:tag:", ""); - const val = values[i]; - if (tag && val) this.tagCache.set(tag, Number(val)); - } - } while (cursor !== "0"); - } catch { - // Fail open — stale tag cache is better than crash - } - } - - async getExpiration(tags: string[]): Promise { - if (tags.length === 0) return 0; - - const missingTags: string[] = []; - let maxTimestamp = 0; - - for (const tag of tags) { - const cached = this.tagCache.get(tag); - if (cached !== undefined) { - if (cached > maxTimestamp) maxTimestamp = cached; - } else { - missingTags.push(tag); - } - } - - if (missingTags.length > 0) { - try { - await this.ensureConnected(); - const values = await this.redis.mget(...missingTags.map((tag) => this.tagKey(tag))); - for (let i = 0; i < missingTags.length; i++) { - const val = values[i]; - const tag = missingTags[i]; - if (!val || !tag) continue; - const timestamp = Number(val); - this.tagCache.set(tag, timestamp); - if (timestamp > maxTimestamp) maxTimestamp = timestamp; - } - } catch { - // Fail open - } - } - - return maxTimestamp; - } - - async updateTags(tags: string[], durations?: { expire?: number }): Promise { - if (tags.length === 0) return; - try { - await this.ensureConnected(); - const now = Date.now(); - const expireMs = durations?.expire ? durations.expire * 1000 : TAG_TTL_MS; - const pipeline = this.redis.pipeline(); - for (const tag of tags) { - const key = this.tagKey(tag); - pipeline.set(key, String(now)); - pipeline.pexpire(key, expireMs); - this.tagCache.set(tag, now); - } - await pipeline.exec(); - } catch { - // Fail open - } - } -} diff --git a/packages/next-cache-handler/src/types.ts b/packages/next-cache-handler/src/types.ts deleted file mode 100644 index 6f0b91d9d..000000000 --- a/packages/next-cache-handler/src/types.ts +++ /dev/null @@ -1 +0,0 @@ -export type { CacheEntry, CacheHandler, Timestamp } from "next/dist/server/lib/cache-handlers/types"; diff --git a/packages/next-cache-handler/tsconfig.json b/packages/next-cache-handler/tsconfig.json deleted file mode 100644 index 82b695258..000000000 --- a/packages/next-cache-handler/tsconfig.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "extends": "@homarr/tsconfig/base.json", - "compilerOptions": { - "outDir": "dist", - "rootDir": "src" - }, - "include": ["src"] -} diff --git a/packages/redis/src/lib/channel-subscription-tracker.ts b/packages/redis/src/lib/channel-subscription-tracker.ts index 1866655cb..02adc6897 100644 --- a/packages/redis/src/lib/channel-subscription-tracker.ts +++ b/packages/redis/src/lib/channel-subscription-tracker.ts @@ -3,8 +3,7 @@ import { randomUUID } from "crypto"; import type { MaybePromise } from "@homarr/common/types"; import { createLogger } from "@homarr/core/infrastructure/logs"; -import { getSubscriberClient, usesMemoryFallback } from "./connection"; -import { memorySubscribe } from "./memory-channel"; +import { createRedisConnection } from "./connection"; const logger = createLogger({ module: "channelSubscriptionTracker" }); @@ -19,12 +18,7 @@ type SubscriptionCallback = (message: string) => MaybePromise; */ export class ChannelSubscriptionTracker { private static subscriptions = new Map>(); - private static redis: ReturnType | undefined; - private static getRedis() { - this.redis ??= getSubscriberClient(); - return this.redis; - } - private static memoryUnsubscribers = new Map void>(); + private static redis = createRedisConnection(); private static listenerActive = false; /** @@ -36,27 +30,19 @@ export class ChannelSubscriptionTracker { public static subscribe(channelName: string, callback: SubscriptionCallback) { logger.debug("Adding redis channel callback", { channel: channelName }); + // We only want to activate the listener once + if (!this.listenerActive) { + this.activateListener(); + this.listenerActive = true; + } + const channelSubscriptions = this.subscriptions.get(channelName) ?? new Map(); const id = randomUUID(); // If there are no subscriptions to the channel, subscribe to it if (channelSubscriptions.size === 0) { logger.debug("Subscribing to redis channel", { channel: channelName }); - if (usesMemoryFallback()) { - const unsub = memorySubscribe(channelName, (message) => { - this.dispatchMessage(channelName, message); - }); - this.memoryUnsubscribers.set(channelName, unsub); - } else { - if (!this.listenerActive) { - this.activateListener(); - this.listenerActive = true; - } - const redis = this.getRedis(); - if (redis) { - void redis.subscribe(channelName); - } - } + void this.redis.subscribe(channelName); } logger.debug("Adding redis channel callback", { channel: channelName, id }); @@ -79,44 +65,30 @@ export class ChannelSubscriptionTracker { } logger.debug("Unsubscribing from redis channel", { channel: channelName }); - if (usesMemoryFallback()) { - this.memoryUnsubscribers.get(channelName)?.(); - this.memoryUnsubscribers.delete(channelName); - } else { - const redis = this.getRedis(); - if (redis) { - void redis.unsubscribe(channelName); - } - } + void this.redis.unsubscribe(channelName); this.subscriptions.delete(channelName); }; } - private static dispatchMessage(channel: string, message: string) { - const channelSubscriptions = this.subscriptions.get(channel); - if (!channelSubscriptions) { - logger.warn("Received message on unknown channel", { channel }); - return; - } - - for (const [id, callback] of channelSubscriptions.entries()) { - // Don't log messages from the logging channel as it would create an infinite loop - if (channel !== "pubSub:logging") { - logger.debug("Calling subscription callback", { channel, id }); - } - void callback(message); - } - } - /** * Activates the listener for the redis client. */ private static activateListener() { logger.debug("Activating listener"); - const redis = this.getRedis(); - if (!redis) return; - redis.on("message", (channel, message) => { - this.dispatchMessage(channel, message); + this.redis.on("message", (channel, message) => { + const channelSubscriptions = this.subscriptions.get(channel); + if (!channelSubscriptions) { + logger.warn("Received message on unknown channel", { channel }); + return; + } + + for (const [id, callback] of channelSubscriptions.entries()) { + // Don't log messages from the logging channel as it would create an infinite loop + if (channel !== "pubSub:logging") { + logger.debug("Calling subscription callback", { channel, id }); + } + void callback(message); + } }); } } diff --git a/packages/redis/src/lib/channel.spec.ts b/packages/redis/src/lib/channel.spec.ts index ed80dc8df..0ec39a2a5 100644 --- a/packages/redis/src/lib/channel.spec.ts +++ b/packages/redis/src/lib/channel.spec.ts @@ -4,12 +4,9 @@ import { describe, expect, test, vi } from "vitest"; vi.mock("./connection", () => ({ createRedisConnection: () => null, - getDataClient: () => null, - getSubscriberClient: () => null, - usesMemoryFallback: () => true, })); -import { createLockChannel, createSubPubChannel } from "./channel"; +import { createLockChannel } from "./channel"; describe("createLockChannel", () => { test("degrades gracefully when Redis connections are disabled", async () => { @@ -22,31 +19,3 @@ describe("createLockChannel", () => { await expect(lock.releaseAsync(token)).resolves.toBeUndefined(); }); }); - -describe("createSubPubChannel with memory fallback", () => { - test("publish and subscribe deliver messages in-process", async () => { - const channel = createSubPubChannel<{ value: number }>("test-pubsub"); - - const received: { value: number }[] = []; - channel.subscribe((data) => { - received.push(data); - }); - - await channel.publishAsync({ value: 42 }); - - expect(received).toEqual([{ value: 42 }]); - await expect(channel.getLastDataAsync()).resolves.toEqual({ value: 42 }); - }); - - test("subscribe replays last persisted data", async () => { - const channel = createSubPubChannel("test-replay"); - await channel.publishAsync("hello"); - - const replayed: string[] = []; - channel.subscribe((data) => { - replayed.push(data); - }); - - expect(replayed).toEqual(["hello"]); - }); -}); diff --git a/packages/redis/src/lib/channel.ts b/packages/redis/src/lib/channel.ts index 4f9ead18d..20e5bbf96 100644 --- a/packages/redis/src/lib/channel.ts +++ b/packages/redis/src/lib/channel.ts @@ -3,13 +3,13 @@ import superjson from "superjson"; import { createId } from "@homarr/common"; import { ChannelSubscriptionTracker } from "./channel-subscription-tracker"; -import { getDataClient, usesMemoryFallback } from "./connection"; -import { memoryGetLast, memoryPublish } from "./memory-channel"; +import { createRedisConnection } from "./connection"; + +const publisher = createRedisConnection(); +const lastDataClient = createRedisConnection(); const LAST_DATA_TTL_SECONDS = 86_400; -const dataClient = getDataClient(); - /** * Creates a new pub/sub channel. * @param name name of the channel @@ -25,18 +25,11 @@ export const createSubPubChannel = (name: string, { persist }: { persist: */ subscribe: (callback: (data: TData) => void) => { if (persist) { - if (usesMemoryFallback()) { - const data = memoryGetLast(lastChannelName); + void lastDataClient.get(lastChannelName).then((data) => { if (data) { callback(superjson.parse(data)); } - } else if (dataClient) { - void dataClient.get(lastChannelName).then((data) => { - if (data) { - callback(superjson.parse(data)); - } - }); - } + }); } return ChannelSubscriptionTracker.subscribe(channelName, (message) => { callback(superjson.parse(message)); @@ -47,31 +40,22 @@ export const createSubPubChannel = (name: string, { persist }: { persist: * @param data data to be published */ publishAsync: async (data: TData) => { - const serialized = superjson.stringify(data); if (persist) { - if (usesMemoryFallback()) { - memoryPublish(lastChannelName, data); - } else if (dataClient) { - await dataClient.set(lastChannelName, serialized, "EX", LAST_DATA_TTL_SECONDS); - } - } - if (usesMemoryFallback()) { - memoryPublish(channelName, data); - } else if (dataClient) { - await dataClient.publish(channelName, serialized); + // Without an expiry these keys accumulate one payload per channel forever. + // getLastDataAsync already tolerates a miss, so expiring them is safe. + await lastDataClient.set(lastChannelName, superjson.stringify(data), "EX", LAST_DATA_TTL_SECONDS); } + await publisher.publish(channelName, superjson.stringify(data)); }, getLastDataAsync: async () => { - if (usesMemoryFallback()) { - const data = memoryGetLast(lastChannelName); - return data ? superjson.parse(data) : null; - } - const data = await dataClient?.get(lastChannelName); + const data = await lastDataClient.get(lastChannelName); return data ? superjson.parse(data) : null; }, }; }; +const getSetClient = createRedisConnection(); + /** * Creates a new redis channel for a list * @param name name of channel @@ -85,8 +69,7 @@ export const createListChannel = (name: string) => { * @returns an array of all items */ getAllAsync: async () => { - if (!dataClient) return []; - const items = await dataClient.lrange(listChannelName, 0, -1); + const items = await getSetClient.lrange(listChannelName, 0, -1); return items.map((item) => superjson.parse(item)); }, /** @@ -94,23 +77,20 @@ export const createListChannel = (name: string) => { * @param item item to remove */ removeAsync: async (item: TItem) => { - if (!dataClient) return; - await dataClient.lrem(listChannelName, 0, superjson.stringify(item)); + await getSetClient.lrem(listChannelName, 0, superjson.stringify(item)); }, /** * Clear all items from the channels list */ clearAsync: async () => { - if (!dataClient) return; - await dataClient.del(listChannelName); + await getSetClient.del(listChannelName); }, /** * Add an item to the channels list * @param item item to add */ addAsync: async (item: TItem) => { - if (!dataClient) return; - await dataClient.lpush(listChannelName, superjson.stringify(item)); + await getSetClient.lpush(listChannelName, superjson.stringify(item)); }, }; }; @@ -126,8 +106,7 @@ export const createGetSetChannel = (name: string) => { * @returns data or null if not found */ getAsync: async () => { - if (!dataClient) return null; - const data = await dataClient.get(name); + const data = await getSetClient.get(name); return data ? superjson.parse(data) : null; }, /** @@ -136,19 +115,17 @@ export const createGetSetChannel = (name: string) => { * @param options optional TTL in seconds */ setAsync: async (data: TData, options?: { ttlSeconds?: number }) => { - if (!dataClient) return; if (options?.ttlSeconds) { - await dataClient.set(name, superjson.stringify(data), "EX", options.ttlSeconds); + await getSetClient.set(name, superjson.stringify(data), "EX", options.ttlSeconds); return; } - await dataClient.set(name, superjson.stringify(data)); + await getSetClient.set(name, superjson.stringify(data)); }, /** * Remove data from the channel */ removeAsync: async () => { - if (!dataClient) return; - await dataClient.del(name); + await getSetClient.del(name); }, }; }; @@ -159,31 +136,21 @@ export const createGetSetChannel = (name: string) => { * The token prevents a process from releasing a lock that expired and was * acquired by another process in the meantime. */ -// ponytail: in-memory lock map for single-instance fallback; upgrade to redis-based lock when Redis is available -const memoryLocks = new Map(); - export const createLockChannel = (name: string) => { return { acquireAsync: async (ttlSeconds: number) => { const token = createId(); - if (!dataClient) { - const existing = memoryLocks.get(name); - if (existing && existing.expiresAt > Date.now()) return null; - memoryLocks.set(name, { token, expiresAt: Date.now() + ttlSeconds * 1000 }); - return token; - } + const client = getSetClient as typeof getSetClient | null; + if (!client) return token; - const result = await dataClient.set(name, token, "EX", ttlSeconds, "NX"); + const result = await client.set(name, token, "EX", ttlSeconds, "NX"); return result === "OK" ? token : null; }, releaseAsync: async (token: string) => { - if (!dataClient) { - const existing = memoryLocks.get(name); - if (existing?.token === token) memoryLocks.delete(name); - return; - } + const client = getSetClient as typeof getSetClient | null; + if (!client) return; - await dataClient.eval( + await client.eval( "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", 1, name, @@ -194,8 +161,9 @@ export const createLockChannel = (name: string) => { }; export const invalidateIntegrationCacheAsync = async (integrationId: string): Promise => { - if (!dataClient) return; - await dataClient.del(`session-store:${integrationId}`); + const client = getSetClient as typeof getSetClient | null; + if (!client) return; + await client.del(`session-store:${integrationId}`); }; /** @@ -203,12 +171,11 @@ export const invalidateIntegrationCacheAsync = async (integrationId: string): Pr */ export const createChannelEventHistoryOld = (channelName: string, maxElements = 15) => { const popElementsOverMaxAsync = async () => { - if (!dataClient) return; - const length = await dataClient.llen(channelName); + const length = await getSetClient.llen(channelName); if (length <= maxElements) { return; } - await dataClient.ltrim(channelName, 0, maxElements - 1); + await getSetClient.ltrim(channelName, 0, maxElements - 1); }; return { @@ -218,39 +185,33 @@ export const createChannelEventHistoryOld = (channelName: string, maxElem }); }, publishAndPushAsync: async (data: TData) => { - if (!dataClient) return; - await dataClient.publish(channelName, superjson.stringify(data)); - await dataClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); + await publisher.publish(channelName, superjson.stringify(data)); + await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); await popElementsOverMaxAsync(); }, pushAsync: async (data: TData) => { - if (!dataClient) return; - await dataClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); + await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); await popElementsOverMaxAsync(); }, clearAsync: async () => { - if (!dataClient) return; - await dataClient.del(channelName); + await getSetClient.del(channelName); }, getLastAsync: async () => { - if (!dataClient) return null; - const length = await dataClient.llen(channelName); - const data = await dataClient.lrange(channelName, length - 1, length); + const length = await getSetClient.llen(channelName); + const data = await getSetClient.lrange(channelName, length - 1, length); if (data.length !== 1) return null; // eslint-disable-next-line @typescript-eslint/no-non-null-assertion return superjson.parse<{ data: TData; timestamp: Date }>(data[0]!); }, getSliceAsync: async (startIndex: number, endIndex: number) => { - if (!dataClient) return []; - const range = await dataClient.lrange(channelName, startIndex, endIndex); + const range = await getSetClient.lrange(channelName, startIndex, endIndex); return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)); }, getSliceUntilTimeAsync: async (time: Date) => { - if (!dataClient) return []; - const length = await dataClient.llen(channelName); + const length = await getSetClient.llen(channelName); const items: TData[] = []; - const itemsInCollection = await dataClient.lrange(channelName, 0, length - 1); + const itemsInCollection = await getSetClient.lrange(channelName, 0, length - 1); for (let i = 0; i < length - 1; i++) { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion @@ -263,13 +224,14 @@ export const createChannelEventHistoryOld = (channelName: string, maxElem return items; }, getLengthAsync: async () => { - if (!dataClient) return 0; - return await dataClient.llen(channelName); + return await getSetClient.llen(channelName); }, name: channelName, }; }; +const queueClient = createRedisConnection(); + type WithId = TItem & { _id: string }; /** @@ -280,13 +242,11 @@ type WithId = TItem & { _id: string }; export const createQueueChannel = (name: string) => { const queueChannelName = `queue:${name}`; const getDataAsync = async () => { - if (!dataClient) return []; - const data = await dataClient.get(queueChannelName); + const data = await queueClient.get(queueChannelName); return data ? superjson.parse[]>(data) : []; }; const setDataAsync = async (data: WithId[]) => { - if (!dataClient) return; - await dataClient.set(queueChannelName, superjson.stringify(data)); + await queueClient.set(queueChannelName, superjson.stringify(data)); }; return { @@ -333,6 +293,5 @@ export const createQueueChannel = (name: string) => { }; export const handshakeAsync = async () => { - if (!dataClient) return; - await dataClient.hello(); + await getSetClient.hello(); }; diff --git a/packages/redis/src/lib/connection.ts b/packages/redis/src/lib/connection.ts index 7c2bd473a..ba945de33 100644 --- a/packages/redis/src/lib/connection.ts +++ b/packages/redis/src/lib/connection.ts @@ -1,39 +1,15 @@ import type { RedisClient } from "@homarr/core/infrastructure/redis"; import { createRedisClient } from "@homarr/core/infrastructure/redis"; -let dataClient: RedisClient | undefined; -let subscriberClient: RedisClient | undefined; - -const isRedisExplicitlyDisabled = () => - process.env.CI === "true" || - process.env.CI === "1" || - process.env.DISABLE_REDIS_LOGS === "true" || - process.env.DISABLE_REDIS_LOGS === "1"; - /** - * In-process pub/sub fallback for single-instance deployments, opted into via - * CI/DISABLE_REDIS_LOGS. This is a static, env-driven switch — it does not detect - * Redis unreachability at runtime; an unreachable Redis outside those flags still - * attempts a real connection (with ioredis's default retry/backoff), it does not - * fall back automatically. Never used when REDIS_IS_EXTERNAL=true (multi-instance - * requires real Redis). + * Creates a new Redis connection + * @returns redis client */ -export const usesMemoryFallback = (): boolean => - process.env.REDIS_IS_EXTERNAL !== "true" && isRedisExplicitlyDisabled(); +export const createRedisConnection = () => { + if (Boolean(process.env.CI) || Boolean(process.env.DISABLE_REDIS_LOGS)) { + // Return null if we are in CI as we don't want to connect to Redis + return null as unknown as RedisClient; + } -/** Shared client for SET/GET/PUBLISH and other non-subscribe Redis commands. */ -export const getDataClient = (): RedisClient | null => { - if (usesMemoryFallback()) return null; - dataClient ??= createRedisClient(); - return dataClient; + return createRedisClient(); }; - -/** Dedicated client for SUBSCRIBE — ioredis requires a separate connection for pub/sub. */ -export const getSubscriberClient = (): RedisClient | null => { - if (usesMemoryFallback()) return null; - subscriberClient ??= createRedisClient(); - return subscriberClient; -}; - -/** @deprecated Prefer getDataClient() — kept for test mocks. */ -export const createRedisConnection = () => getDataClient(); diff --git a/packages/redis/src/lib/memory-channel.ts b/packages/redis/src/lib/memory-channel.ts deleted file mode 100644 index 34b07d4dd..000000000 --- a/packages/redis/src/lib/memory-channel.ts +++ /dev/null @@ -1,34 +0,0 @@ -import EventEmitter from "node:events"; - -import superjson from "superjson"; - -const emitter = new EventEmitter(); -emitter.setMaxListeners(100); - -const lastData = new Map(); -const MAX_ENTRIES = 500; - -export const memoryPublish = (channel: string, data: unknown) => { - const serialized = superjson.stringify(data); - lastData.set(channel, { data: serialized, timestamp: Date.now() }); - - // ponytail: FIFO eviction by Map insertion order, not timestamp. - if (lastData.size > MAX_ENTRIES) { - const oldest = lastData.keys().next().value; - if (oldest) lastData.delete(oldest); - } - - emitter.emit(channel, serialized); -}; - -export const memorySubscribe = (channel: string, callback: (data: string) => void) => { - emitter.on(channel, callback); - return () => { - emitter.off(channel, callback); - }; -}; - -export const memoryGetLast = (channel: string): string | null => { - const entry = lastData.get(channel); - return entry?.data ?? null; -}; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f60231d28..0f31e95af 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -860,9 +860,6 @@ importers: '@homarr/modals-collection': specifier: workspace:^0.1.0 version: link:../../packages/modals-collection - '@homarr/next-cache-handler': - specifier: workspace:^0.1.0 - version: link:../../packages/next-cache-handler '@homarr/notifications': specifier: workspace:^0.1.0 version: link:../../packages/notifications @@ -2087,25 +2084,6 @@ importers: specifier: 'catalog:' version: 7.0.2 - packages/next-cache-handler: - dependencies: - ioredis: - specifier: 'catalog:' - version: 5.11.1(supports-color@10.2.2) - next: - specifier: 'catalog:' - version: 16.3.0-preview.9(@babel/core@7.29.0(supports-color@10.2.2))(@opentelemetry/api@1.9.0)(@playwright/test@1.61.1)(@types/node@24.13.3)(babel-plugin-react-compiler@1.0.0)(react-dom@19.2.7(react@19.2.7))(react@19.2.7)(sass@1.101.0) - devDependencies: - '@homarr/tsconfig': - specifier: workspace:* - version: link:../../tooling/typescript - typescript: - specifier: 'catalog:' - version: 7.0.2 - vitest: - specifier: 'catalog:' - version: 4.1.10(@opentelemetry/api@1.9.0)(@types/node@24.13.3)(@vitest/coverage-v8@4.1.10)(@vitest/ui@4.1.10)(jsdom@29.1.1(@noble/hashes@2.0.1))(msw@2.14.6(@types/node@24.13.3)(typescript@7.0.2))(vite@8.1.4(@types/node@24.13.3)(esbuild@0.28.1)(jiti@2.7.0)(sass@1.101.0)(sugarss@5.0.0(postcss@8.5.16))(terser@5.46.1)(tsx@4.21.0)(yaml@2.9.0)) - packages/notifications: dependencies: '@homarr/ui': @@ -21948,6 +21926,7 @@ snapshots: optionalDependencies: msw: 2.14.6(@types/node@24.13.3)(typescript@7.0.2) vite: 8.1.4(@types/node@24.13.3)(esbuild@0.28.1)(jiti@2.7.0)(sass@1.101.0)(sugarss@5.0.0(postcss@8.5.16))(terser@5.46.1)(tsx@4.21.0)(yaml@2.9.0) + optional: true '@vitest/pretty-format@4.1.10': dependencies: @@ -29824,6 +29803,7 @@ snapshots: terser: 5.46.1 tsx: 4.21.0 yaml: 2.9.0 + optional: true vitest@4.1.10(@opentelemetry/api@1.9.0)(@types/node@24.13.3)(@vitest/coverage-v8@4.1.10)(@vitest/ui@4.1.10)(jsdom@29.1.1(@noble/hashes@2.0.1))(msw@2.14.6(@types/node@24.13.3)(typescript@7.0.2))(vite@8.1.4(@types/node@24.13.3)(esbuild@0.28.1)(jiti@2.7.0)(sass@1.101.0)(sugarss@5.0.0(postcss@8.5.16))(terser@5.46.1)(tsx@4.20.4)(yaml@2.9.0)): dependencies: @@ -29886,6 +29866,7 @@ snapshots: jsdom: 29.1.1(@noble/hashes@2.0.1) transitivePeerDependencies: - msw + optional: true vue-component-type-helpers@3.3.4: {}