From d2ce64252cbd794189d94a1cd9732c656be56597 Mon Sep 17 00:00:00 2001 From: Thomas Camlong Date: Mon, 25 May 2026 10:09:55 +0200 Subject: [PATCH] fix: revert unrelated changes (Dockerfile, Redis null-safety) --- Dockerfile | 2 + .../src/lib/channel-subscription-tracker.ts | 6 +- packages/redis/src/lib/channel.ts | 116 ++++++++---------- packages/redis/src/lib/connection.ts | 9 +- 4 files changed, 58 insertions(+), 75 deletions(-) diff --git a/Dockerfile b/Dockerfile index 1c1388ec5..51b562f7d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -15,6 +15,8 @@ RUN corepack enable pnpm && pnpm install --recursive --frozen-lockfile # Copy static data as it is not part of the build COPY static-data ./static-data ARG SKIP_ENV_VALIDATION='true' +ARG CI='true' +ARG DISABLE_REDIS_LOGS='true' RUN corepack enable pnpm && pnpm build diff --git a/packages/redis/src/lib/channel-subscription-tracker.ts b/packages/redis/src/lib/channel-subscription-tracker.ts index 3c76fd5eb..02adc6897 100644 --- a/packages/redis/src/lib/channel-subscription-tracker.ts +++ b/packages/redis/src/lib/channel-subscription-tracker.ts @@ -42,7 +42,7 @@ export class ChannelSubscriptionTracker { // If there are no subscriptions to the channel, subscribe to it if (channelSubscriptions.size === 0) { logger.debug("Subscribing to redis channel", { channel: channelName }); - void this.redis?.subscribe(channelName); + void this.redis.subscribe(channelName); } logger.debug("Adding redis channel callback", { channel: channelName, id }); @@ -65,7 +65,7 @@ export class ChannelSubscriptionTracker { } logger.debug("Unsubscribing from redis channel", { channel: channelName }); - void this.redis?.unsubscribe(channelName); + void this.redis.unsubscribe(channelName); this.subscriptions.delete(channelName); }; } @@ -75,7 +75,7 @@ export class ChannelSubscriptionTracker { */ private static activateListener() { logger.debug("Activating listener"); - this.redis?.on("message", (channel, message) => { + this.redis.on("message", (channel, message) => { const channelSubscriptions = this.subscriptions.get(channel); if (!channelSubscriptions) { logger.warn("Received message on unknown channel", { channel }); diff --git a/packages/redis/src/lib/channel.ts b/packages/redis/src/lib/channel.ts index d7c8fd4f6..193f551e3 100644 --- a/packages/redis/src/lib/channel.ts +++ b/packages/redis/src/lib/channel.ts @@ -12,8 +12,6 @@ const logger = createLogger({ module: "redisChannel" }); const publisher = createRedisConnection(); const lastDataClient = createRedisConnection(); -const noopSubscription = (): (() => void) => () => undefined; - /** * Creates a new pub/sub channel. * @param name name of the channel @@ -28,16 +26,13 @@ export const createSubPubChannel = (name: string, { persist }: { persist: * @param callback callback function to be called when new data is published */ subscribe: (callback: (data: TData) => void) => { - if (persist && lastDataClient) { + if (persist) { void lastDataClient.get(lastChannelName).then((data) => { if (data) { callback(superjson.parse(data)); } }); } - if (!publisher) { - return noopSubscription(); - } return ChannelSubscriptionTracker.subscribe(channelName, (message) => { callback(superjson.parse(message)); }); @@ -47,18 +42,12 @@ export const createSubPubChannel = (name: string, { persist }: { persist: * @param data data to be published */ publishAsync: async (data: TData) => { - if (!publisher || !lastDataClient) { - return; - } if (persist) { await lastDataClient.set(lastChannelName, superjson.stringify(data)); } await publisher.publish(channelName, superjson.stringify(data)); }, getLastDataAsync: async () => { - if (!lastDataClient) { - return null; - } const data = await lastDataClient.get(lastChannelName); return data ? superjson.parse(data) : null; }, @@ -80,28 +69,28 @@ export const createListChannel = (name: string) => { * @returns an array of all items */ getAllAsync: async () => { - const items = await getSetClient?.lrange(listChannelName, 0, -1); - return items?.map((item) => superjson.parse(item)) ?? []; + const items = await getSetClient.lrange(listChannelName, 0, -1); + return items.map((item) => superjson.parse(item)); }, /** * Remove an item from the channels list by item * @param item item to remove */ removeAsync: async (item: TItem) => { - await getSetClient?.lrem(listChannelName, 0, superjson.stringify(item)); + await getSetClient.lrem(listChannelName, 0, superjson.stringify(item)); }, /** * Clear all items from the channels list */ clearAsync: async () => { - await getSetClient?.del(listChannelName); + await getSetClient.del(listChannelName); }, /** * Add an item to the channels list * @param item item to add */ addAsync: async (item: TItem) => { - await getSetClient?.lpush(listChannelName, superjson.stringify(item)); + await getSetClient.lpush(listChannelName, superjson.stringify(item)); }, }; }; @@ -117,7 +106,7 @@ export const createGetSetChannel = (name: string) => { * @returns data or null if not found */ getAsync: async () => { - const data = await getSetClient?.get(name); + const data = await getSetClient.get(name); return data ? superjson.parse(data) : null; }, /** @@ -125,13 +114,13 @@ export const createGetSetChannel = (name: string) => { * @param data data to be stored in the channel */ setAsync: async (data: TData) => { - await getSetClient?.set(name, superjson.stringify(data)); + await getSetClient.set(name, superjson.stringify(data)); }, /** * Remove data from the channel */ removeAsync: async () => { - await getSetClient?.del(name); + await getSetClient.del(name); }, }; }; @@ -151,7 +140,7 @@ export const createCacheChannel = (name: string, cacheDurationMs: number * @returns data or null if not found or expired */ getAsync: async () => { - const data = await getSetClient?.get(cacheChannelName); + const data = await getSetClient.get(cacheChannelName); if (!data) return null; const parsedData = superjson.parse<{ data: TData; timestamp: Date }>(data); @@ -167,13 +156,13 @@ export const createCacheChannel = (name: string, cacheDurationMs: number * @returns data or new data if not present or expired */ consumeAsync: async (callback: () => Promise) => { - const data = await getSetClient?.get(cacheChannelName); + const data = await getSetClient.get(cacheChannelName); const getNewDataAsync = async () => { logger.debug(`Cache miss for channel '${cacheChannelName}'`); const newData = await callback(); const result = { data: newData, timestamp: new Date() }; - await getSetClient?.set(cacheChannelName, superjson.stringify(result)); + await getSetClient.set(cacheChannelName, superjson.stringify(result)); logger.debug(`Cache updated for channel '${cacheChannelName}'`); return result; }; @@ -198,9 +187,6 @@ export const createCacheChannel = (name: string, cacheDurationMs: number * Invalidate the cache channels data. */ invalidateAsync: async () => { - if (!getSetClient) { - return; - } await getSetClient.del(cacheChannelName); }, /** @@ -208,7 +194,7 @@ export const createCacheChannel = (name: string, cacheDurationMs: number * @param data data to be stored in the cache channel */ setAsync: async (data: TData) => { - await getSetClient?.set(cacheChannelName, superjson.stringify({ data, timestamp: new Date() })); + await getSetClient.set(cacheChannelName, superjson.stringify({ data, timestamp: new Date() })); }, }; }; @@ -250,12 +236,12 @@ export const createChannelEventHistory = (channelName: string, maxElement }); }, pushAsync: async (data: TData, options = { publish: false }) => { - if (options.publish) await publisher?.publish(channelName, superjson.stringify(data)); - await getSetClient?.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); - await getSetClient?.ltrim(channelName, 0, maxElements); + if (options.publish) await publisher.publish(channelName, superjson.stringify(data)); + await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); + await getSetClient.ltrim(channelName, 0, maxElements); }, clearAsync: async () => { - await getSetClient?.del(channelName); + await getSetClient.del(channelName); }, /** * Returns a slice of the available data in the channel. @@ -264,19 +250,17 @@ export const createChannelEventHistory = (channelName: string, maxElement * @param endIndex End index of the slice, negative values are counted from the end, defaults at end of range. */ getSliceAsync: async (startIndex = 0, endIndex = -1) => { - const range = await getSetClient?.lrange(channelName, startIndex, endIndex); - return range?.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)) ?? []; + const range = await getSetClient.lrange(channelName, startIndex, endIndex); + return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)); }, getSliceUntilTimeAsync: async (time: Date) => { - const itemsInCollection = await getSetClient?.lrange(channelName, 0, -1); - return ( - itemsInCollection - ?.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)) - .filter((item) => item.timestamp < time) ?? [] - ); + const itemsInCollection = await getSetClient.lrange(channelName, 0, -1); + return itemsInCollection + .map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)) + .filter((item) => item.timestamp < time); }, getLengthAsync: async () => { - return (await getSetClient?.llen(channelName)) ?? 0; + return await getSetClient.llen(channelName); }, name: channelName, }; @@ -287,11 +271,11 @@ export const createChannelEventHistory = (channelName: string, maxElement */ export const createChannelEventHistoryOld = (channelName: string, maxElements = 15) => { const popElementsOverMaxAsync = async () => { - const length = await getSetClient?.llen(channelName); - if (!length || length <= maxElements) { + const length = await getSetClient.llen(channelName); + if (length <= maxElements) { return; } - await getSetClient?.ltrim(channelName, 0, maxElements - 1); + await getSetClient.ltrim(channelName, 0, maxElements - 1); }; return { @@ -301,39 +285,37 @@ export const createChannelEventHistoryOld = (channelName: string, maxElem }); }, publishAndPushAsync: async (data: TData) => { - await publisher?.publish(channelName, superjson.stringify(data)); - await getSetClient?.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) => { - await getSetClient?.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); + await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() })); await popElementsOverMaxAsync(); }, clearAsync: async () => { - await getSetClient?.del(channelName); + await getSetClient.del(channelName); }, getLastAsync: async () => { - const length = await getSetClient?.llen(channelName); - const data = await getSetClient?.lrange(channelName, (length ?? 0) - 1, length ?? 0); - if (data?.length !== 1) return null; + 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) => { - const range = await getSetClient?.lrange(channelName, startIndex, endIndex); - return range?.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)) ?? []; + const range = await getSetClient.lrange(channelName, startIndex, endIndex); + return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item)); }, getSliceUntilTimeAsync: async (time: Date) => { - const length = await getSetClient?.llen(channelName); + const length = await getSetClient.llen(channelName); const items: TData[] = []; - const itemsInCollection = await getSetClient?.lrange(channelName, 0, (length ?? 0) - 1); + const itemsInCollection = await getSetClient.lrange(channelName, 0, length - 1); - if (!itemsInCollection) return items; - - for (const item of itemsInCollection) { - if (!item) continue; - const deserializedItem = superjson.parse<{ data: TData; timestamp: Date }>(item); + for (let i = 0; i < length - 1; i++) { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const deserializedItem = superjson.parse<{ data: TData; timestamp: Date }>(itemsInCollection[i]!); if (deserializedItem.timestamp < time) { continue; } @@ -342,7 +324,7 @@ export const createChannelEventHistoryOld = (channelName: string, maxElem return items; }, getLengthAsync: async () => { - return (await getSetClient?.llen(channelName)) ?? 0; + return await getSetClient.llen(channelName); }, name: channelName, }; @@ -356,14 +338,14 @@ export const createChannelWithLatestAndEvents = (channelName: string) => }); }, publishAndUpdateLastStateAsync: async (data: TData) => { - await publisher?.publish(channelName, superjson.stringify(data)); - await getSetClient?.set(channelName, superjson.stringify({ data, timestamp: new Date() })); + await publisher.publish(channelName, superjson.stringify(data)); + await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() })); }, setAsync: async (data: TData) => { - await getSetClient?.set(channelName, superjson.stringify({ data, timestamp: new Date() })); + await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() })); }, getAsync: async () => { - const data = await getSetClient?.get(channelName); + const data = await getSetClient.get(channelName); if (!data) return null; return superjson.parse<{ data: TData; timestamp: Date }>(data); @@ -384,11 +366,11 @@ type WithId = TItem & { _id: string }; export const createQueueChannel = (name: string) => { const queueChannelName = `queue:${name}`; const getDataAsync = async () => { - const data = await queueClient?.get(queueChannelName); + const data = await queueClient.get(queueChannelName); return data ? superjson.parse[]>(data) : []; }; const setDataAsync = async (data: WithId[]) => { - await queueClient?.set(queueChannelName, superjson.stringify(data)); + await queueClient.set(queueChannelName, superjson.stringify(data)); }; return { @@ -435,5 +417,5 @@ export const createQueueChannel = (name: string) => { }; export const handshakeAsync = async () => { - await getSetClient?.hello(); + await getSetClient.hello(); }; diff --git a/packages/redis/src/lib/connection.ts b/packages/redis/src/lib/connection.ts index 972b00bb8..ba945de33 100644 --- a/packages/redis/src/lib/connection.ts +++ b/packages/redis/src/lib/connection.ts @@ -5,11 +5,10 @@ import { createRedisClient } from "@homarr/core/infrastructure/redis"; * Creates a new Redis connection * @returns redis client */ -export const createRedisConnection = (): RedisClient | null => { - const disableRedisLogs = process.env.DISABLE_REDIS_LOGS === "true"; - - if (disableRedisLogs) { - return null; +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; } return createRedisClient();