diff --git a/packages/redis/src/index.ts b/packages/redis/src/index.ts index f8d8fd329..8e6e3db86 100644 --- a/packages/redis/src/index.ts +++ b/packages/redis/src/index.ts @@ -13,6 +13,8 @@ export { handshakeAsync, createSubPubChannel, createGetSetChannel, + createQueryCacheChannel, + getAllQueryCacheAsync, } from "./lib/channel"; export { createIntegrationHistoryChannel } from "./lib/channels/history-channel"; diff --git a/packages/redis/src/lib/channel.ts b/packages/redis/src/lib/channel.ts index 193f551e3..03025565e 100644 --- a/packages/redis/src/lib/channel.ts +++ b/packages/redis/src/lib/channel.ts @@ -125,6 +125,49 @@ export const createGetSetChannel = (name: string) => { }; }; +interface QueryCacheChannelOptions { + userId: string; + boardId: string; + key: string; + ttlMs: number; + maxValueBytes: number; +} + +const queryCacheHashKey = (userId: string, boardId: string) => `qc:${userId}:${boardId}`; + +export const createQueryCacheChannel = ({ userId, boardId, key, ttlMs, maxValueBytes }: QueryCacheChannelOptions) => { + const hkey = queryCacheHashKey(userId, boardId); + + return { + name: `${hkey}:${key}`, + getAsync: async () => { + return await getSetClient.hget(hkey, key); + }, + setAsync: async (value: string) => { + if (Buffer.byteLength(value, "utf8") > maxValueBytes) { + logger.warn("Query cache value exceeded maximum size", { + channel: hkey, + valueBytes: Buffer.byteLength(value, "utf8"), + maxValueBytes, + }); + return false; + } + + await getSetClient.hset(hkey, key, value); + await getSetClient.pexpire(hkey, ttlMs); + return true; + }, + removeAsync: async () => { + await getSetClient.hdel(hkey, key); + }, + }; +}; + +export const getAllQueryCacheAsync = async (userId: string, boardId: string): Promise> => { + const hkey = queryCacheHashKey(userId, boardId); + return await getSetClient.hgetall(hkey); +}; + /** * Creates a new cache channel. * @param name name of the channel