perf: drop the cache layers that measured at zero
Ablation (PHASE2-FINDINGS.md) attributed the memory win to the serverExternalPackages line and the integrations barrel diet. These three measured at exactly zero and were most of the file churn, so they go: next-cache-handler: orphaned. cacheHandlers and cacheComponents are both already gone from next.config.ts and nothing imported the package — it was reachable only through a stale dependency entry. cache-tags / cache-invalidation: there is no "use cache", cacheTag, cacheLife or unstable_cache anywhere in app source, so revalidateTag was firing on tags nothing registers, with no handler to receive them, inside a catch that swallowed the resulting out-of-scope error. Removing the calls also removed the SELECTs that existed only to supply a board name or member list to them, so board.ts, group.ts, user.ts, serverSettings.ts, section-router.ts and change-search-preferences.ts return to the merge-base exactly. The live in-process integration data cache is untouched. Redis 6 -> 2 client consolidation: no measurable effect at idle (139.3 vs 139.2 MiB) or under load. Reverted, but the pubSub:last:* TTL is kept — that one is an unbounded-growth fix, not a consolidation. 23 files, -883 lines. Diff vs merge-base: 100 files/3336 insertions -> 82 files/2484. pnpm test 144/144 files, 1795 tests, 0 failures; typecheck and lint clean; app and Docker image build.
This commit is contained in:
23 files changed
+90
-883
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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());
|
||||
};
|
||||
@@ -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"));
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
@@ -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 };
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -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);
|
||||
}),
|
||||
});
|
||||
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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);
|
||||
}),
|
||||
});
|
||||
@@ -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();
|
||||
}),
|
||||
});
|
||||
@@ -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) {
|
||||
|
||||
@@ -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);
|
||||
};
|
||||
@@ -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);
|
||||
}),
|
||||
});
|
||||
@@ -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:"
|
||||
}
|
||||
}
|
||||
@@ -1,24 +0,0 @@
|
||||
import type { CacheHandler } from "next/dist/server/lib/cache-handlers/types";
|
||||
|
||||
import { RedisCacheHandler } from "./redis-handler";
|
||||
|
||||
let handlerPromise: Promise<CacheHandler> | undefined;
|
||||
|
||||
const createHandlerAsync = async (): Promise<CacheHandler> => {
|
||||
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<CacheHandler> => (handlerPromise ??= createHandlerAsync());
|
||||
|
||||
export default {
|
||||
get: async (...args: Parameters<CacheHandler["get"]>) => (await getHandler()).get(...args),
|
||||
set: async (...args: Parameters<CacheHandler["set"]>) => (await getHandler()).set(...args),
|
||||
refreshTags: async () => (await getHandler()).refreshTags(),
|
||||
getExpiration: async (...args: Parameters<CacheHandler["getExpiration"]>) =>
|
||||
(await getHandler()).getExpiration(...args),
|
||||
updateTags: async (...args: Parameters<CacheHandler["updateTags"]>) => (await getHandler()).updateTags(...args),
|
||||
} satisfies CacheHandler;
|
||||
@@ -1,236 +0,0 @@
|
||||
import Redis from "ioredis";
|
||||
|
||||
import type { CacheHandler } from "next/dist/server/lib/cache-handlers/types";
|
||||
|
||||
type CacheEntry = NonNullable<Awaited<ReturnType<CacheHandler["get"]>>>;
|
||||
|
||||
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<Uint8Array>) {
|
||||
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<string, Promise<CacheEntry>>();
|
||||
private tagCache = new Map<string, number>();
|
||||
private connectPromise: Promise<void> | 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<CacheEntry | undefined> {
|
||||
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<CacheEntry>): Promise<void> {
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
export type { CacheEntry, CacheHandler, Timestamp } from "next/dist/server/lib/cache-handlers/types";
|
||||
@@ -1,8 +0,0 @@
|
||||
{
|
||||
"extends": "@homarr/tsconfig/base.json",
|
||||
"compilerOptions": {
|
||||
"outDir": "dist",
|
||||
"rootDir": "src"
|
||||
},
|
||||
"include": ["src"]
|
||||
}
|
||||
@@ -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<void>;
|
||||
*/
|
||||
export class ChannelSubscriptionTracker {
|
||||
private static subscriptions = new Map<string, Map<string, SubscriptionCallback>>();
|
||||
private static redis: ReturnType<typeof getSubscriberClient> | undefined;
|
||||
private static getRedis() {
|
||||
this.redis ??= getSubscriberClient();
|
||||
return this.redis;
|
||||
}
|
||||
private static memoryUnsubscribers = new Map<string, () => 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<string, SubscriptionCallback>();
|
||||
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);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<string>("test-replay");
|
||||
await channel.publishAsync("hello");
|
||||
|
||||
const replayed: string[] = [];
|
||||
channel.subscribe((data) => {
|
||||
replayed.push(data);
|
||||
});
|
||||
|
||||
expect(replayed).toEqual(["hello"]);
|
||||
});
|
||||
});
|
||||
@@ -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 = <TData>(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 = <TData>(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<TData>(data) : null;
|
||||
}
|
||||
const data = await dataClient?.get(lastChannelName);
|
||||
const data = await lastDataClient.get(lastChannelName);
|
||||
return data ? superjson.parse<TData>(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 = <TItem>(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<TItem>(item));
|
||||
},
|
||||
/**
|
||||
@@ -94,23 +77,20 @@ export const createListChannel = <TItem>(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 = <TData>(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<TData>(data) : null;
|
||||
},
|
||||
/**
|
||||
@@ -136,19 +115,17 @@ export const createGetSetChannel = <TData>(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 = <TData>(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<string, { token: string; expiresAt: number }>();
|
||||
|
||||
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<void> => {
|
||||
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 = <TData>(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 = <TData>(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 = <TData>(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> = TItem & { _id: string };
|
||||
|
||||
/**
|
||||
@@ -280,13 +242,11 @@ type WithId<TItem> = TItem & { _id: string };
|
||||
export const createQueueChannel = <TItem>(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<WithId<TItem>[]>(data) : [];
|
||||
};
|
||||
const setDataAsync = async (data: WithId<TItem>[]) => {
|
||||
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 = <TItem>(name: string) => {
|
||||
};
|
||||
|
||||
export const handshakeAsync = async () => {
|
||||
if (!dataClient) return;
|
||||
await dataClient.hello();
|
||||
await getSetClient.hello();
|
||||
};
|
||||
@@ -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();
|
||||
@@ -1,34 +0,0 @@
|
||||
import EventEmitter from "node:events";
|
||||
|
||||
import superjson from "superjson";
|
||||
|
||||
const emitter = new EventEmitter();
|
||||
emitter.setMaxListeners(100);
|
||||
|
||||
const lastData = new Map<string, { data: string; timestamp: number }>();
|
||||
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;
|
||||
};
|
||||
Generated
+3
-22
@@ -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: {}
|
||||
|
||||
|
||||
Reference in new issue
Block a user