CSC POC ontop of Parser

Author: sjpotter

packages/client/index.ts

@@ -34,3 +34,7 @@ export { GEO_REPLY_WITH, GeoReplyWith } from './lib/commands/GEOSEARCH_WITH';
 export { SetOptions } from './lib/commands/SET';
 
 export { REDIS_FLUSH_MODES } from './lib/commands/FLUSHALL';
+
+import { BasicClientSideCache, BasicPooledClientSideCache } from './lib/client/cache';
+export { BasicClientSideCache, BasicPooledClientSideCache };
+

packages/client/lib/client/commands-queue.ts

@@ -56,6 +56,8 @@ export default class RedisCommandsQueue {
     return this.#pubSub.isActive;
   }
 
+  #invalidateCallback?: (key: RedisArgument | null) => unknown;
+
   constructor(
     respVersion: RespVersions,
     maxLength: number | null | undefined,
@@ -109,13 +111,30 @@ export default class RedisCommandsQueue {
       onErrorReply: err => this.#onErrorReply(err),
       onPush: push => {
         if (!this.#onPush(push)) {
-
+          switch (push[0].toString()) {
+            case "invalidate": {
+              if (this.#invalidateCallback) {
+                if (push[1] !== null) {
+                  for (const key of push[1]) {
+                    this.#invalidateCallback(key);
+                  }
+                } else {
+                  this.#invalidateCallback(null);
+                }
+              }
+              break;
+            }
+          }
         }
       },
       getTypeMapping: () => this.#getTypeMapping()
     });
   }
 
+  setInvalidateCallback(callback?: (key: RedisArgument | null) => unknown) {
+    this.#invalidateCallback = callback;
+  }
+
   addCommand<T>(
     args: ReadonlyArray<RedisArgument>,
     options?: CommandOptions

packages/client/lib/client/index.ts

@@ -16,6 +16,7 @@ import { ScanOptions, ScanCommonOptions } from '../commands/SCAN';
 import { RedisLegacyClient, RedisLegacyClientType } from './legacy-mode';
 import { RedisPoolOptions, RedisClientPool } from './pool';
 import { RedisVariadicArgument, parseArgs, pushVariadicArguments } from '../commands/generic-transformers';
+import { BasicClientSideCache, ClientSideCacheConfig, ClientSideCacheProvider } from './cache';
 import { BasicCommandParser, CommandParser } from './parser';
 import SingleEntryCache from '../single-entry-cache';
 
@@ -81,6 +82,10 @@ export interface RedisClientOptions<
    * TODO
    */
   commandOptions?: CommandOptions<TYPE_MAPPING>;
+  /**
+   * TODO
+   */
+  clientSideCache?: ClientSideCacheProvider | ClientSideCacheConfig;
 }
 
 type WithCommands<
@@ -313,9 +318,8 @@ export default class RedisClient<
   // was in a watch transaction when 
   // a topology change occured
   #dirtyWatch?: string;
-  #epoch: number;
   #watchEpoch?: number; 
-
+  #clientSideCache?: ClientSideCacheProvider;
   #credentialsSubscription: Disposable | null = null;
 
   get options(): RedisClientOptions<M, F, S, RESP> | undefined {
@@ -334,6 +338,11 @@ export default class RedisClient<
     return this._self.#queue.isPubSubActive;
   }
 
+  get socketEpoch() {
+    return this._self.#socket.socketEpoch;
+  }
+
+
   get isWatching() {
     return this._self.#watchEpoch !== undefined;
   }
@@ -358,10 +367,20 @@ export default class RedisClient<
 
   constructor(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>) {
     super();
+
     this.#options = this.#initiateOptions(options);
     this.#queue = this.#initiateQueue();
     this.#socket = this.#initiateSocket();
-    this.#epoch = 0;
+
+    if (options?.clientSideCache) {
+      if (options.clientSideCache instanceof ClientSideCacheProvider) {
+        this.#clientSideCache = options.clientSideCache;
+      } else {
+        const cscConfig = options.clientSideCache;
+        this.#clientSideCache = new BasicClientSideCache(cscConfig);
+      }
+      this.#queue.setInvalidateCallback(this.#clientSideCache.invalidate.bind(this.#clientSideCache));
+    }
   }
 
   #initiateOptions(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>): RedisClientOptions<M, F, S, RESP, TYPE_MAPPING> | undefined {
@@ -522,6 +541,13 @@ export default class RedisClient<
       );
     }
 
+    if (this.#clientSideCache) {
+      const tracking = this.#clientSideCache.trackingOn();
+      if (tracking) {
+        commands.push(tracking);
+      }
+    }
+
     return commands;
   }
 
@@ -575,6 +601,7 @@ export default class RedisClient<
       })
       .on('error', err => {
         this.emit('error', err);
+        this.#clientSideCache?.onError();
         if (this.#socket.isOpen && !this.#options?.disableOfflineQueue) {
           this.#queue.flushWaitingForReply(err);
         } else {
@@ -583,7 +610,6 @@ export default class RedisClient<
       })
       .on('connect', () => this.emit('connect'))
       .on('ready', () => {
-        this.#epoch++;
         this.emit('ready');
         this.#setPingTimer();
         this.#maybeScheduleWrite();
@@ -711,14 +737,21 @@ export default class RedisClient<
     commandOptions: CommandOptions<TYPE_MAPPING> | undefined,
     transformReply: TransformReply | undefined,
   ) {
-    const reply = await this.sendCommand(parser.redisArgs, commandOptions);
+    const csc = this._self.#clientSideCache;
+    const defaultTypeMapping = this._self.#options?.commandOptions === commandOptions;
 
-    if (transformReply) {
-      const res = transformReply(reply, parser.preserve, commandOptions?.typeMapping);
-      return res
-    }
+    const fn = () => { return this.sendCommand(parser.redisArgs, commandOptions) };
 
-    return reply;
+    if (csc && command.CACHEABLE && defaultTypeMapping) {
+      return await csc.handleCache(this._self, parser as BasicCommandParser, fn, transformReply, commandOptions?.typeMapping);
+    } else {
+      const reply = await fn();
+
+      if (transformReply) {
+        return transformReply(reply, parser.preserve, commandOptions?.typeMapping);
+      }
+      return reply;
+    }
   }
 
   /**
@@ -883,7 +916,7 @@ export default class RedisClient<
     const reply = await this._self.sendCommand(
       pushVariadicArguments(['WATCH'], key)
     );
-    this._self.#watchEpoch ??= this._self.#epoch;
+    this._self.#watchEpoch ??= this._self.socketEpoch;
     return reply as unknown as ReplyWithTypeMapping<SimpleStringReply<'OK'>, TYPE_MAPPING>;
   }
 
@@ -950,7 +983,7 @@ export default class RedisClient<
     }
 
     const chainId = Symbol('Pipeline Chain'),
-      promise = Promise.all(
+      promise = Promise.allSettled(
         commands.map(({ args }) => this._self.#queue.addCommand(args, {
           chainId,
           typeMapping: this._commandOptions?.typeMapping
@@ -986,7 +1019,7 @@ export default class RedisClient<
       throw new WatchError(dirtyWatch);
     }
 
-    if (watchEpoch && watchEpoch !== this._self.#epoch) {
+    if (watchEpoch && watchEpoch !== this._self.socketEpoch) {
       throw new WatchError('Client reconnected after WATCH');
     }
 
@@ -1210,6 +1243,7 @@ export default class RedisClient<
     return new Promise<void>(resolve => {
       clearTimeout(this._self.#pingTimer);
       this._self.#socket.close();
+      this._self.#clientSideCache?.onClose();
 
       if (this._self.#queue.isEmpty()) {
         this._self.#socket.destroySocket();
@@ -1236,6 +1270,7 @@ export default class RedisClient<
     clearTimeout(this._self.#pingTimer);
     this._self.#queue.flushAll(new DisconnectsClientError());
     this._self.#socket.destroy();
+    this._self.#clientSideCache?.onClose();
     this._self.#credentialsSubscription?.dispose();
     this._self.#credentialsSubscription = null;
   }

packages/client/lib/client/linked-list.ts

@@ -114,6 +114,7 @@ export class DoublyLinkedList<T> {
 export interface SinglyLinkedNode<T> {
   value: T;
   next: SinglyLinkedNode<T> | undefined;
+  removed: boolean;
 }
 
 export class SinglyLinkedList<T> {
@@ -140,7 +141,8 @@ export class SinglyLinkedList<T> {
 
     const node = {
       value,
-      next: undefined
+      next: undefined,
+      removed: false
     };
 
     if (this.#head === undefined) {
@@ -151,6 +153,9 @@ export class SinglyLinkedList<T> {
   }
 
   remove(node: SinglyLinkedNode<T>, parent: SinglyLinkedNode<T> | undefined) {
+    if (node.removed) {
+      throw new Error("node already removed");
+    }
     --this.#length;
 
     if (this.#head === node) {
@@ -165,6 +170,8 @@ export class SinglyLinkedList<T> {
     } else {
       parent!.next = node.next;
     }
+
+    node.removed = true;
   }
 
   shift() {
@@ -177,6 +184,7 @@ export class SinglyLinkedList<T> {
       this.#head = node.next;
     }
 
+    node.removed = true;
     return node.value;
   }
 

packages/client/lib/client/parser.ts

@@ -33,6 +33,11 @@ export class BasicCommandParser implements CommandParser {
     return this.#keys[0];
   }
 
+  get cacheKey() {
+    let cacheKey = this.#redisArgs.map((arg) => arg.length).join('_');
+    return cacheKey + '_' + this.#redisArgs.join('_');
+  }
+
   push(...arg: Array<RedisArgument>) {
     this.#redisArgs.push(...arg);
   };