Merge pull request #368 from mike182uk/redis-cluster-support

Author: dahlia

CHANGES.md

@@ -64,6 +64,9 @@ To be released.
     hanging on slow or unresponsive servers.
     [[#258] by Hyunchae Kim]
 
+[#353]: https://github.com/fedify-dev/fedify/issues/353
+[#365]: https://github.com/fedify-dev/fedify/pull/365
+
 ### @fedify/next
 
  -  Created [Next.js] integration as the *@fedify/next* package.
@@ -72,6 +75,13 @@ To be released.
 [Next.js]: https://nextjs.org/
 [#313]: https://github.com/fedify-dev/fedify/issues/313
 
+### @fedify/redis
+
+ -  Added support for Redis Cluster to the *@fedify/redis* package.
+    [[#368] by Michael Barrett]
+
+[#368]: https://github.com/fedify-dev/fedify/pull/368
+
 
 Version 1.8.5
 -------------

packages/redis/README.md

@@ -16,12 +16,27 @@ implementations for Redis:
 ~~~~ typescript
 import { createFederation } from "@fedify/fedify";
 import { RedisKvStore, RedisMessageQueue } from "@fedify/redis";
-import { Redis } from "ioredis";
+import { Redis, Cluster } from "ioredis";
 
+// Using a standalone Redis instance:
 const federation = createFederation({
   kv: new RedisKvStore(new Redis()),
   queue: new RedisMessageQueue(() => new Redis()),
 });
+
+// Using a Redis Cluster:
+const federation = createFederation({
+  kv: new RedisKvStore(new Cluster([
+    { host: "127.0.0.1", port: 7000 },
+    { host: "127.0.0.1", port: 7001 },
+    { host: "127.0.0.1", port: 7002 },
+  ])),
+  queue: new RedisMessageQueue(() => new Cluster([
+    { host: "127.0.0.1", port: 7000 },
+    { host: "127.0.0.1", port: 7001 },
+    { host: "127.0.0.1", port: 7002 },
+  ])),
+});
 ~~~~
 
 [JSR]: https://jsr.io/@fedify/redis

packages/redis/src/kv.ts

@@ -1,5 +1,5 @@
 import type { KvKey, KvStore, KvStoreSetOptions } from "@fedify/fedify";
-import type { Redis, RedisKey } from "ioredis";
+import type { Cluster, Redis, RedisKey } from "ioredis";
 import { Buffer } from "node:buffer";
 import { type Codec, JsonCodec } from "./codec.ts";
 
@@ -27,26 +27,38 @@ export interface RedisKvStoreOptions {
  * ```ts ignore
  * import { createFederation } from "@fedify/fedify";
  * import { RedisKvStore } from "@fedify/redis";
- * import { Redis } from "ioredis";
+ * import { Redis, Cluster } from "ioredis";
  *
+ * // Using a standalone Redis instance:
  * const federation = createFederation({
  *   // ...
  *   kv: new RedisKvStore(new Redis()),
  * });
+ *
+ * // Using a Redis Cluster:
+ * const cluster = new Cluster([
+ *   { host: "127.0.0.1", port: 7000 },
+ *   { host: "127.0.0.1", port: 7001 },
+ *   { host: "127.0.0.1", port: 7002 },
+ * ]);
+ * const federation = createFederation({
+ *   // ...
+ *   kv: new RedisKvStore(cluster),
+ * });
  * ```
  */
 export class RedisKvStore implements KvStore {
-  #redis: Redis;
+  #redis: Redis | Cluster;
   #keyPrefix: RedisKey;
   #codec: Codec;
   #textEncoder = new TextEncoder();
 
   /**
    * Creates a new Redis key–value store.
-   * @param redis The Redis client to use.
+   * @param redis The Redis client (standalone or cluster) to use.
    * @param options The options for the key–value store.
    */
-  constructor(redis: Redis, options: RedisKvStoreOptions = {}) {
+  constructor(redis: Redis | Cluster, options: RedisKvStoreOptions = {}) {
     this.#redis = redis;
     this.#keyPrefix = options.keyPrefix ?? "fedify::";
     this.#codec = options.codec ?? new JsonCodec();

packages/redis/src/mq.ts

@@ -5,7 +5,7 @@ import type {
   MessageQueueListenOptions,
 } from "@fedify/fedify";
 import { getLogger } from "@logtape/logtape";
-import type { Redis, RedisKey } from "ioredis";
+import type { Cluster, Redis, RedisKey } from "ioredis";
 import { type Codec, JsonCodec } from "./codec.ts";
 
 const logger = getLogger(["fedify", "redis", "mq"]);
@@ -63,17 +63,28 @@ export interface RedisMessageQueueOptions {
  * ```ts ignore
  * import { createFederation } from "@fedify/fedify";
  * import { RedisMessageQueue } from "@fedify/redis";
- * import { Redis } from "ioredis";
+ * import { Redis, Cluster } from "ioredis";
  *
+ * // Using a standalone Redis instance:
  * const federation = createFederation({
  *   // ...
  *   queue: new RedisMessageQueue(() => new Redis()),
  * });
+ *
+ * // Using a Redis Cluster:
+ * const federation = createFederation({
+ *   // ...
+ *   queue: new RedisMessageQueue(() => new Cluster([
+ *     { host: "127.0.0.1", port: 7000 },
+ *     { host: "127.0.0.1", port: 7001 },
+ *     { host: "127.0.0.1", port: 7002 },
+ *   ])),
+ * });
  * ```
  */
 export class RedisMessageQueue implements MessageQueue, Disposable {
-  #redis: Redis;
-  #subRedis: Redis;
+  #redis: Redis | Cluster;
+  #subRedis: Redis | Cluster;
   #workerId: string;
   #channelKey: RedisKey;
   #queueKey: RedisKey;
@@ -87,7 +98,10 @@ export class RedisMessageQueue implements MessageQueue, Disposable {
    * @param redis The Redis client factory.
    * @param options The options for the message queue.
    */
-  constructor(redis: () => Redis, options: RedisMessageQueueOptions = {}) {
+  constructor(
+    redis: () => Redis | Cluster,
+    options: RedisMessageQueueOptions = {},
+  ) {
     this.#redis = redis();
     this.#subRedis = redis();
     this.#workerId = options.workerId ?? crypto.randomUUID();
@@ -196,9 +210,24 @@ export class RedisMessageQueue implements MessageQueue, Disposable {
       }
     };
     const promise = this.#subRedis.subscribe(this.#channelKey, () => {
-      this.#subRedis.on("message", poll);
+      /**
+       * Cast to Redis for event methods. Both Redis and Cluster extend EventEmitter
+       * and get the same methods via applyMixin at runtime, but their TypeScript
+       * interfaces are incompatible:
+       * - Redis declares specific overloads: on(event: "message", cb: (channel, message) => void)
+       * - Cluster only has generic: on(event: string | symbol, listener: Function)
+       *
+       * This makes the union type Redis | Cluster incompatible for these method calls.
+       * The cast is safe because both classes use applyMixin(Class, EventEmitter) which
+       * copies all EventEmitter prototype methods, giving them identical pub/sub functionality.
+       *
+       * @see https://github.com/redis/ioredis/blob/main/lib/Redis.ts#L863 (has specific overloads)
+       * @see https://github.com/redis/ioredis/blob/main/lib/cluster/index.ts#L1110 (empty interface)
+       */
+      const subRedis = this.#subRedis as Redis;
+      subRedis.on("message", poll);
       signal?.addEventListener("abort", () => {
-        this.#subRedis.off("message", poll);
+        subRedis.off("message", poll);
       });
     });
     signal?.addEventListener(