fix: revert unrelated changes (Dockerfile, Redis null-safety)
This commit is contained in:
4 files changed
+58
-75
No files matched your search
@@ -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
|
||||
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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 = <TData>(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 = <TData>(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<TData>(data) : null;
|
||||
},
|
||||
@@ -80,28 +69,28 @@ export const createListChannel = <TItem>(name: string) => {
|
||||
* @returns an array of all items
|
||||
*/
|
||||
getAllAsync: async () => {
|
||||
const items = await getSetClient?.lrange(listChannelName, 0, -1);
|
||||
return items?.map((item) => superjson.parse<TItem>(item)) ?? [];
|
||||
const items = await getSetClient.lrange(listChannelName, 0, -1);
|
||||
return items.map((item) => superjson.parse<TItem>(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 = <TData>(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<TData>(data) : null;
|
||||
},
|
||||
/**
|
||||
@@ -125,13 +114,13 @@ export const createGetSetChannel = <TData>(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 = <TData>(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 = <TData>(name: string, cacheDurationMs: number
|
||||
* @returns data or new data if not present or expired
|
||||
*/
|
||||
consumeAsync: async (callback: () => Promise<TData>) => {
|
||||
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 = <TData>(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 = <TData>(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 = <TData>(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 = <TData>(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 = <TData>(channelName: string, maxElement
|
||||
*/
|
||||
export const createChannelEventHistoryOld = <TData>(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 = <TData>(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 = <TData>(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 = <TData>(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> = TItem & { _id: string };
|
||||
export const createQueueChannel = <TItem>(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<WithId<TItem>[]>(data) : [];
|
||||
};
|
||||
const setDataAsync = async (data: WithId<TItem>[]) => {
|
||||
await queueClient?.set(queueChannelName, superjson.stringify(data));
|
||||
await queueClient.set(queueChannelName, superjson.stringify(data));
|
||||
};
|
||||
|
||||
return {
|
||||
@@ -435,5 +417,5 @@ export const createQueueChannel = <TItem>(name: string) => {
|
||||
};
|
||||
|
||||
export const handshakeAsync = async () => {
|
||||
await getSetClient?.hello();
|
||||
await getSetClient.hello();
|
||||
};
|
||||
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user