fix: address v2 stack review findings
This commit is contained in:
1 parent
53b518960c
commit
263702210a
48 files changed
+1235
-530
No files matched your search
@@ -5,3 +5,4 @@ export {
|
||||
serializeCustomWidgetDefinition,
|
||||
} from "./router/custom-widget/stored-definition";
|
||||
export { configureCustomWidgetSourceFromRequest } from "./router/custom-widget/secret-persistence";
|
||||
export { invalidateCustomWidgetResponseCache } from "./router/custom-widget/request-executor";
|
||||
@@ -5,6 +5,7 @@ import { z } from "zod/v4";
|
||||
import { constructIntegrationPermissions } from "@homarr/auth/shared";
|
||||
import { createId } from "@homarr/common";
|
||||
import { decryptSecret, encryptSecret } from "@homarr/common/server";
|
||||
import { fetchWithTrustedCertificatesAsync } from "@homarr/core/infrastructure/http";
|
||||
import { and, asc, desc, eq, handleTransactionsAsync, inArray } from "@homarr/db";
|
||||
import type { Database } from "@homarr/db";
|
||||
import {
|
||||
@@ -33,7 +34,7 @@ import { fetchOpenRouterGenerationTelemetryAsync } from "../assistant-generation
|
||||
import { env } from "../env";
|
||||
import { orderMessagesByParent } from "../assistant-message-order";
|
||||
import { verifyAssistantGenerationAccessToken } from "../assistant-generation-access";
|
||||
import { createTRPCRouter, isDemoMode, permissionRequiredProcedure, protectedProcedure } from "../trpc";
|
||||
import { createTRPCRouter, isDemoMode, isDemoReadOnly, permissionRequiredProcedure, protectedProcedure } from "../trpc";
|
||||
import { boardRouter, getHomeIdBoardAsync } from "./board";
|
||||
|
||||
const adminProcedure = permissionRequiredProcedure.requiresPermission("admin");
|
||||
@@ -206,9 +207,9 @@ const fetchModelsAsync = async (configuration: AssistantConfiguration) => {
|
||||
}
|
||||
|
||||
const endpoint = `${configuration.baseUrl}${configuration.modelDiscoveryPath}`;
|
||||
let response: Response;
|
||||
let response: Awaited<ReturnType<typeof fetchWithTrustedCertificatesAsync>>;
|
||||
try {
|
||||
response = await fetch(endpoint, {
|
||||
response = await fetchWithTrustedCertificatesAsync(endpoint, {
|
||||
headers: getProviderHeaders(configuration),
|
||||
signal: AbortSignal.timeout(15_000),
|
||||
});
|
||||
@@ -454,7 +455,7 @@ const addFeedbackToMessageContent = (serializedContent: string, type: "positive"
|
||||
export const assistantRouter = createTRPCRouter({
|
||||
getAvailability: protectedProcedure.query(async ({ ctx }) => {
|
||||
if (isDemoMode) {
|
||||
return { enabled: true };
|
||||
return { enabled: !isDemoReadOnly };
|
||||
}
|
||||
const configuration = await getConfigurationAsync(ctx.db);
|
||||
const requiresApiKey = configuration ? assistantProviderRequiresApiKey(configuration.provider) : false;
|
||||
|
||||
@@ -64,7 +64,7 @@ export const previewActionProcedures = {
|
||||
kind: "action",
|
||||
});
|
||||
if (response.ok && request.invalidates?.length) {
|
||||
invalidateCustomWidgetResponseCache(
|
||||
await invalidateCustomWidgetResponseCache(
|
||||
request.invalidates.map((requestId) => `custom-jsx:preview:${session.id}:${requestId}:`),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -19,7 +19,7 @@ export const previewQueryProcedures = {
|
||||
.input(z.object({ sessionId: z.string().min(1) }))
|
||||
.mutation(async ({ ctx, input }) => {
|
||||
const session = await getPreviewSession(input.sessionId, ctx.session.user.id);
|
||||
invalidateCustomWidgetResponseCache([`custom-jsx:preview:${session.id}:`]);
|
||||
await invalidateCustomWidgetResponseCache([`custom-jsx:preview:${session.id}:`]);
|
||||
}),
|
||||
|
||||
previewQuery: permissionRequiredProcedure
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
import type { CustomWidgetHttpRequest, CustomWidgetHttpResponse } from "@homarr/custom-widgets/server";
|
||||
|
||||
import { toTrpcError } from "./domain-error";
|
||||
import { executeWithCustomWidgetResponseCache, invalidateSharedCustomWidgetResponseCache } from "./response-cache";
|
||||
|
||||
export type {
|
||||
CustomWidgetAuthConfig,
|
||||
@@ -74,14 +75,18 @@ export async function resolveAndValidateHost(hostname: string, scope: CustomWidg
|
||||
|
||||
export async function executeCustomWidgetRequest(input: CustomWidgetHttpRequest): Promise<CustomWidgetHttpResponse> {
|
||||
try {
|
||||
return await executeDomainRequest({
|
||||
...input,
|
||||
logError: (event) => logger.error("Custom widget request failed", event),
|
||||
});
|
||||
return await executeWithCustomWidgetResponseCache(input, (request) =>
|
||||
executeDomainRequest({
|
||||
...request,
|
||||
logError: (event) => logger.error("Custom widget request failed", event),
|
||||
}),
|
||||
);
|
||||
} catch (error) {
|
||||
toTrpcError(error);
|
||||
}
|
||||
}
|
||||
|
||||
export const invalidateCustomWidgetResponseCache = (prefixes: readonly string[]) =>
|
||||
export async function invalidateCustomWidgetResponseCache(prefixes: readonly string[]) {
|
||||
invalidateDomainResponseCache(prefixes);
|
||||
await invalidateSharedCustomWidgetResponseCache(prefixes);
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
import { Buffer } from "node:buffer";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
import { createLogger } from "@homarr/core/infrastructure/logs";
|
||||
import { createRedisClient } from "@homarr/core/infrastructure/redis";
|
||||
import type { RedisClient } from "@homarr/core/infrastructure/redis";
|
||||
import type { CustomWidgetHttpRequest, CustomWidgetHttpResponse } from "@homarr/custom-widgets/server";
|
||||
|
||||
const ENTRY_PREFIX = "custom-widget:response-cache:v2:entry:";
|
||||
const GENERATION_KEY = "custom-widget:response-cache:v2:generation";
|
||||
const LRU_KEY = "custom-widget:response-cache:v2:lru";
|
||||
const SIZE_KEY = "custom-widget:response-cache:v2:sizes";
|
||||
const TOTAL_BYTES_KEY = "custom-widget:response-cache:v2:bytes";
|
||||
const MAX_ENTRIES = 1_000;
|
||||
const MAX_BYTES = 64 * 1024 * 1024;
|
||||
|
||||
const STORE_SCRIPT = `
|
||||
local previousSize = tonumber(redis.call('HGET', KEYS[2], ARGV[1]) or '0')
|
||||
if previousSize > 0 then redis.call('DECRBY', KEYS[3], previousSize) end
|
||||
redis.call('SET', ARGV[1], ARGV[2], 'PX', ARGV[3])
|
||||
redis.call('ZADD', KEYS[1], ARGV[4], ARGV[1])
|
||||
redis.call('HSET', KEYS[2], ARGV[1], ARGV[5])
|
||||
local totalBytes = redis.call('INCRBY', KEYS[3], ARGV[5])
|
||||
while redis.call('ZCARD', KEYS[1]) > tonumber(ARGV[6]) or totalBytes > tonumber(ARGV[7]) do
|
||||
local evicted = redis.call('ZPOPMIN', KEYS[1], 1)
|
||||
if #evicted == 0 then break end
|
||||
local evictedKey = evicted[1]
|
||||
local evictedSize = tonumber(redis.call('HGET', KEYS[2], evictedKey) or '0')
|
||||
redis.call('DEL', evictedKey)
|
||||
redis.call('HDEL', KEYS[2], evictedKey)
|
||||
if evictedSize > 0 then totalBytes = redis.call('DECRBY', KEYS[3], evictedSize) end
|
||||
end
|
||||
return totalBytes
|
||||
`;
|
||||
|
||||
const logger = createLogger({ module: "custom-widget:response-cache" });
|
||||
const inFlight = new Map<string, Promise<CustomWidgetHttpResponse>>();
|
||||
let redis: RedisClient | undefined;
|
||||
|
||||
const useSharedCache = () => process.env.NODE_ENV !== "test";
|
||||
|
||||
const getRedis = () => {
|
||||
redis ??= createRedisClient();
|
||||
return redis;
|
||||
};
|
||||
|
||||
const getGeneration = async (client: RedisClient) => Number((await client.get(GENERATION_KEY)) ?? 0);
|
||||
|
||||
const getEntryKey = (generation: number, cacheKey: string) => {
|
||||
const digest = createHash("sha256").update(cacheKey).digest("hex");
|
||||
return `${ENTRY_PREFIX}${generation}:${digest}`;
|
||||
};
|
||||
|
||||
const readCachedResponse = async (client: RedisClient, entryKey: string) => {
|
||||
const serialized = await client.get(entryKey);
|
||||
if (serialized === null) return undefined;
|
||||
await client.zadd(LRU_KEY, Date.now(), entryKey);
|
||||
try {
|
||||
return JSON.parse(serialized) as CustomWidgetHttpResponse;
|
||||
} catch {
|
||||
await client.del(entryKey);
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
const storeCachedResponse = async (
|
||||
client: RedisClient,
|
||||
entryKey: string,
|
||||
response: CustomWidgetHttpResponse,
|
||||
ttlSeconds: number,
|
||||
) => {
|
||||
const serialized = JSON.stringify(response);
|
||||
const size = Buffer.byteLength(serialized, "utf8");
|
||||
await client.eval(
|
||||
STORE_SCRIPT,
|
||||
3,
|
||||
LRU_KEY,
|
||||
SIZE_KEY,
|
||||
TOTAL_BYTES_KEY,
|
||||
entryKey,
|
||||
serialized,
|
||||
ttlSeconds * 1_000,
|
||||
Date.now(),
|
||||
size,
|
||||
MAX_ENTRIES,
|
||||
MAX_BYTES,
|
||||
);
|
||||
};
|
||||
|
||||
export async function executeWithCustomWidgetResponseCache(
|
||||
input: CustomWidgetHttpRequest,
|
||||
execute: (input: CustomWidgetHttpRequest) => Promise<CustomWidgetHttpResponse>,
|
||||
): Promise<CustomWidgetHttpResponse> {
|
||||
const cacheKey = input.kind === "query" ? input.cacheKey : undefined;
|
||||
if (!cacheKey) return execute(input);
|
||||
if (!useSharedCache()) return execute(input);
|
||||
|
||||
let client: RedisClient;
|
||||
let generation: number;
|
||||
try {
|
||||
client = getRedis();
|
||||
generation = await getGeneration(client);
|
||||
if ((input.cacheTtlSeconds ?? 0) > 0) {
|
||||
const cached = await readCachedResponse(client, getEntryKey(generation, cacheKey));
|
||||
if (cached) return cached;
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error("Shared custom widget response cache is unavailable", {
|
||||
errorName: error instanceof Error ? error.name : "UnknownError",
|
||||
});
|
||||
return execute({ ...input, cacheKey: undefined });
|
||||
}
|
||||
|
||||
const pendingKey = `${generation}:${cacheKey}`;
|
||||
const pending = inFlight.get(pendingKey);
|
||||
if (pending) return pending;
|
||||
|
||||
let request: Promise<CustomWidgetHttpResponse>;
|
||||
request = execute({ ...input, cacheKey: undefined })
|
||||
.then(async (response) => {
|
||||
const ttlSeconds = input.cacheTtlSeconds ?? 0;
|
||||
if (!response.ok || ttlSeconds <= 0) return response;
|
||||
try {
|
||||
const currentGeneration = await getGeneration(client);
|
||||
if (currentGeneration === generation) {
|
||||
await storeCachedResponse(client, getEntryKey(generation, cacheKey), response, ttlSeconds);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error("Failed to store a custom widget response", {
|
||||
errorName: error instanceof Error ? error.name : "UnknownError",
|
||||
});
|
||||
}
|
||||
return response;
|
||||
})
|
||||
.finally(() => {
|
||||
if (inFlight.get(pendingKey) === request) inFlight.delete(pendingKey);
|
||||
});
|
||||
inFlight.set(pendingKey, request);
|
||||
return request;
|
||||
}
|
||||
|
||||
export async function invalidateSharedCustomWidgetResponseCache(prefixes: readonly string[]) {
|
||||
if (prefixes.length === 0 || !useSharedCache()) return;
|
||||
for (const key of inFlight.keys()) {
|
||||
if (prefixes.some((prefix) => key.includes(prefix))) inFlight.delete(key);
|
||||
}
|
||||
try {
|
||||
await getRedis().incr(GENERATION_KEY);
|
||||
} catch (error) {
|
||||
logger.error("Failed to invalidate the shared custom widget response cache", {
|
||||
errorName: error instanceof Error ? error.name : "UnknownError",
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -118,8 +118,8 @@ describe("Custom JSX v2 board router", () => {
|
||||
});
|
||||
|
||||
test("refreshes the server response cache for a placed widget", async () => {
|
||||
const { db, itemId } = await setup();
|
||||
const caller = customApiRouter.createCaller({ db, deviceType: undefined, session: null });
|
||||
const { db, itemId, ownerId } = await setup();
|
||||
const caller = customApiRouter.createCaller({ db, deviceType: undefined, session: createSession(ownerId) });
|
||||
|
||||
await caller.refresh({ itemId });
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ import { TRPCError } from "@trpc/server";
|
||||
import { parse as parseSuperJson } from "superjson";
|
||||
import { z } from "zod/v4";
|
||||
|
||||
import { mapWithConcurrency } from "@homarr/common";
|
||||
import { decryptSecret } from "@homarr/common/server";
|
||||
import { eq } from "@homarr/db";
|
||||
import { boards, customWidgetDefinitions, items, legacyCustomWidgetDefinitions } from "@homarr/db/schema";
|
||||
@@ -13,6 +14,7 @@ import {
|
||||
validateCustomWidgetOptions,
|
||||
} from "@homarr/custom-widgets/core";
|
||||
import type { CustomJsxRequest, CustomWidgetSource } from "@homarr/custom-widgets/core";
|
||||
import { CUSTOM_WIDGET_USER_ITEM_CONCURRENCY_LIMIT } from "@homarr/custom-widgets/server";
|
||||
|
||||
import { createTRPCRouter, protectedProcedure, publicProcedure } from "../../trpc";
|
||||
import { throwIfActionForbiddenAsync } from "../board/board-access";
|
||||
@@ -40,7 +42,7 @@ interface CustomWidgetItemOptions {
|
||||
refreshInterval?: number;
|
||||
}
|
||||
|
||||
type RouterContext = Parameters<typeof throwIfActionForbiddenAsync>[0];
|
||||
type RouterContext = Parameters<typeof throwIfActionForbiddenAsync>[0] & { ipAddress?: string };
|
||||
type ResolvedDefinition = Awaited<ReturnType<typeof resolvePlacedDefinitionAsync>>;
|
||||
|
||||
const parseItemOptions = (raw: string): CustomWidgetItemOptions => {
|
||||
@@ -154,6 +156,7 @@ const withRequestLimit = async <T>(
|
||||
const release = await acquireCustomWidgetRequestLimit({
|
||||
category: request.kind === "query" ? "query" : request.method === "DELETE" ? "delete" : "action",
|
||||
userId: ctx.session?.user.id,
|
||||
anonymousId: ctx.ipAddress,
|
||||
itemId: resolved.item.id,
|
||||
definitionId: resolved.stored.id,
|
||||
});
|
||||
@@ -194,9 +197,9 @@ const executeRequest = async (
|
||||
};
|
||||
|
||||
export const customApiRouter = createTRPCRouter({
|
||||
refresh: publicProcedure.input(itemInputSchema).mutation(async ({ ctx, input }) => {
|
||||
refresh: protectedProcedure.input(itemInputSchema).mutation(async ({ ctx, input }) => {
|
||||
const resolved = await resolvePlacedDefinitionAsync(ctx, input.itemId);
|
||||
invalidateCustomWidgetResponseCache([
|
||||
await invalidateCustomWidgetResponseCache([
|
||||
`custom-jsx:${resolved.item.id}:${getCustomWidgetCacheVersion(resolved.stored)}:`,
|
||||
]);
|
||||
}),
|
||||
@@ -206,8 +209,10 @@ export const customApiRouter = createTRPCRouter({
|
||||
const loadRequests = Object.entries(resolved.definition.requests).filter(
|
||||
([, request]) => request.kind === "query" && request.trigger === "load",
|
||||
);
|
||||
const entries = await Promise.all(
|
||||
loadRequests.map(async ([requestId, request]) => {
|
||||
const entries = await mapWithConcurrency(
|
||||
loadRequests,
|
||||
CUSTOM_WIDGET_USER_ITEM_CONCURRENCY_LIMIT,
|
||||
async ([requestId, request]) => {
|
||||
try {
|
||||
const response = await executeRequest(ctx, resolved, { id: requestId, ...request }, {});
|
||||
return [
|
||||
@@ -237,12 +242,13 @@ export const customApiRouter = createTRPCRouter({
|
||||
},
|
||||
] as const;
|
||||
}
|
||||
}),
|
||||
},
|
||||
);
|
||||
|
||||
return {
|
||||
type: "customJsx" as const,
|
||||
template: resolved.definition.template,
|
||||
canRefresh: ctx.session !== null,
|
||||
data: Object.fromEntries(entries.map(([id, result]) => [id, result.data])),
|
||||
status: Object.fromEntries(entries.map(([id, result]) => [id, result.status])),
|
||||
options: resolved.configuration,
|
||||
@@ -285,7 +291,7 @@ export const customApiRouter = createTRPCRouter({
|
||||
}
|
||||
const response = await executeRequest(ctx, resolved, request, input.params);
|
||||
if (response.ok && request.invalidates?.length) {
|
||||
invalidateCustomWidgetResponseCache(
|
||||
await invalidateCustomWidgetResponseCache(
|
||||
request.invalidates.flatMap((requestId) => [
|
||||
`custom-jsx:${resolved.item.id}:${getCustomWidgetCacheVersion(resolved.stored)}:${requestId}:`,
|
||||
`custom-widget:options:${resolved.stored.id}:${getCustomWidgetCacheVersion(resolved.stored)}:${requestId}:`,
|
||||
|
||||
@@ -13,7 +13,7 @@ import { ZodError } from "zod/v4";
|
||||
|
||||
import type { Session } from "@homarr/auth";
|
||||
import { extractBaseUrlFromHeaders, FlattenError } from "@homarr/common";
|
||||
import { userAgent } from "@homarr/common/server";
|
||||
import { ipAddressFromHeaders, userAgent } from "@homarr/common/server";
|
||||
import type { DeviceType } from "@homarr/common/server";
|
||||
import { createLogger } from "@homarr/core/infrastructure/logs";
|
||||
import { db } from "@homarr/db";
|
||||
@@ -43,6 +43,7 @@ interface ApiContext {
|
||||
deviceType: DeviceType;
|
||||
baseUrl?: `${string}://${string}`;
|
||||
onboardingClaimToken?: string;
|
||||
ipAddress?: string;
|
||||
db: typeof db;
|
||||
}
|
||||
|
||||
@@ -60,6 +61,7 @@ export const createTRPCContext = (opts: { headers: Headers; session: Session | n
|
||||
session,
|
||||
deviceType: userAgent(opts.headers).device.type,
|
||||
baseUrl: extractBaseUrlFromHeaders(opts.headers),
|
||||
ipAddress: ipAddressFromHeaders(opts.headers) ?? undefined,
|
||||
onboardingClaimToken: getOnboardingClaimTokenFromCookieHeader(opts.headers.get("cookie")),
|
||||
db,
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user