feat: add redis example (#974)

Author: n1ru4l

README.md

@@ -94,7 +94,7 @@ liveQueryStore.invalidate([
 
 Those invalidation calls could be done manually in the mutation resolvers or on more global reactive level e.g. as a listener on a database write log. The possibilities are infinite. 🤔
 
-For scaling horizontally the independent `InMemoryLiveQueryStore` instances can be wired together via a PubSub system such as Redis.
+For scaling horizontally the independent `InMemoryLiveQueryStore` instances [can be wired together via a PubSub system such as Redis](./packages/example-redis).
 
 ### How are the updates sent/applied to the client?
 

bob.config.js

@@ -9,6 +9,7 @@ module.exports = {
     "@n1ru4l/todo-example-server-ws",
     "@n1ru4l/todo-example-server-yoga",
     "@n1ru4l/end2end-tests",
+    "@n1ru4l/example-redis",
   ], // ignored packages
   base: "origin/main",
 };

packages/example-redis/README.md

@@ -0,0 +1,67 @@
+# Redis Live Query Example
+
+This example shows how you can distribute live query invalidation events across multiple live query store servers.
+
+## Instructions
+
+### Start Redis Container
+
+```
+docker run -p "6379:6379" redis:7.0.2
+```
+
+### Start GraphQL HTTP Server Instance 1
+
+```
+npx cross-env PORT=3000 yarn start
+```
+
+### Start GraphQL HTTP Server Instance 2
+
+```
+npx cross-env PORT=3001 yarn start
+```
+
+## Demo
+
+Open the following links and execute the live query:
+
+1. http://127.0.0.1:3001/graphql?query=query+%40live+%7B%0A++counter%0A%7D%0A
+
+2. http://127.0.0.1:3000/graphql?query=query+%40live+%7B%0A++counter%0A%7D%0A
+
+**Note:** Each of the live queries is executed on a different HTTP server (see the port)
+
+Open http://127.0.0.1:3001/graphql?query=mutation+%7B%0A++increment%0A%7D
+
+Execute the mutation operation.
+
+See how all the live query results are updated automatically
+
+**Bonus:**
+
+1. Get the container id
+
+```bash
+% docker ps
+CONTAINER ID   IMAGE         COMMAND                  CREATED          STATUS          PORTS                    NAMES
+06239a997e72   redis:7.0.2   "docker-entrypoint.s…"   12 minutes ago   Up 12 minutes   0.0.0.0:6379->6379/tcp   gallant_mirzakhani
+```
+
+2. Monitor the redis commands being executed
+
+```bash
+% docker exec -ti 06239a997e72 redis-cli MONITOR
+OK
+```
+
+3. Execute the mutation and observer the redis-cli output
+
+```
+% docker exec -ti 06239a997e72 redis-cli MONITOR
+OK
+1663236183.245626 [0 172.17.0.1:62562] "incr" "counter"
+1663236183.247329 [0 172.17.0.1:62562] "publish" "live-query-invalidations" "Query.counter"
+1663236183.249530 [0 172.17.0.1:62562] "get" "counter"
+1663236183.249542 [0 172.17.0.1:62560] "get" "counter"
+```

packages/example-redis/package.json

@@ -0,0 +1,22 @@
+{
+  "name": "@n1ru4l/example-redis",
+  "version": "0.1.2",
+  "private": true,
+  "dependencies": {
+    "@n1ru4l/in-memory-live-query-store": "0.10.0",
+    "@n1ru4l/graphql-live-query": "0.10.0",
+    "graphql": "16.0.0-experimental-stream-defer.5",
+    "@graphql-yoga/node": "2.13.13",
+    "ioredis": "5.2.3"
+  },
+  "devDependencies": {
+    "@types/node": "16.11.43",
+    "ts-node-dev": "2.0.0",
+    "typescript": "4.7.4"
+  },
+  "scripts": {
+    "build": "tsc",
+    "start": "ts-node-dev src/main.ts"
+  },
+  "bob": false
+}

packages/example-redis/src/main.ts

@@ -0,0 +1,106 @@
+import { InMemoryLiveQueryStore } from "@n1ru4l/in-memory-live-query-store";
+import { GraphQLLiveDirective } from "@n1ru4l/graphql-live-query";
+import { astFromDirective } from "@graphql-tools/utils";
+import { createServer, Plugin } from "@graphql-yoga/node";
+import Redis from "ioredis";
+import { execute as defaultExecute } from "graphql";
+
+const httpPort = parseInt(process.env.PORT ?? "3000", 10);
+const redisUri = process.env.REDIS_URI ?? "redis://localhost:6379";
+
+const inMemoryLiveQueryStore = new InMemoryLiveQueryStore();
+
+const client = new Redis(redisUri);
+const subClient = new Redis(redisUri);
+
+class RedisLiveQueryStore {
+  pub: Redis;
+  sub: Redis;
+  channel: string;
+  liveQueryStore: InMemoryLiveQueryStore;
+
+  constructor(
+    pub: Redis,
+    sub: Redis,
+    channel: string,
+    liveQueryStore: InMemoryLiveQueryStore
+  ) {
+    this.pub = pub;
+    this.sub = sub;
+    this.liveQueryStore = liveQueryStore;
+    this.channel = channel;
+
+    this.sub.subscribe(this.channel, (err) => {
+      if (err) throw err;
+    });
+
+    this.sub.on("message", (channel, resourceIdentifier) => {
+      if (channel === this.channel && resourceIdentifier)
+        this.liveQueryStore.invalidate(resourceIdentifier);
+    });
+  }
+
+  async invalidate(identifiers: Array<string> | string) {
+    if (typeof identifiers === "string") {
+      identifiers = [identifiers];
+    }
+    for (const identifier of identifiers) {
+      this.pub.publish(this.channel, identifier);
+    }
+  }
+
+  makeExecute(execute: typeof defaultExecute) {
+    return this.liveQueryStore.makeExecute(execute);
+  }
+}
+
+const liveQueryStore = new RedisLiveQueryStore(
+  client,
+  subClient,
+  "live-query-invalidations",
+  inMemoryLiveQueryStore
+);
+
+const liveQueryPlugin: Plugin = {
+  onExecute(params) {
+    params.setExecuteFn(liveQueryStore.makeExecute(params.executeFn));
+  },
+};
+
+const server = createServer({
+  context: () => ({
+    liveQueryStore,
+    redisClient: client,
+  }),
+  plugins: [liveQueryPlugin],
+  port: httpPort,
+  schema: {
+    typeDefs: [
+      /* GraphQL */ `
+        type Query {
+          counter: Int!
+        }
+        type Mutation {
+          increment: Boolean
+        }
+      `,
+      astFromDirective(GraphQLLiveDirective),
+    ],
+    resolvers: {
+      Query: {
+        async counter(_, __, context) {
+          const value = await context.redisClient.get("counter");
+          return value == null ? 0 : parseInt(value, 10);
+        },
+      },
+      Mutation: {
+        async increment(_, __, context) {
+          await context.redisClient.incr("counter");
+          await context.liveQueryStore.invalidate(["Query.counter"]);
+        },
+      },
+    },
+  },
+});
+
+server.start();

packages/example-redis/tsconfig.json

@@ -0,0 +1,12 @@
+{
+  "compilerOptions": {
+    "target": "ES2019",
+    "skipLibCheck": true,
+    "esModuleInterop": true,
+    "allowSyntheticDefaultImports": true,
+    "strict": true,
+    "module": "CommonJS",
+    "moduleResolution": "node",
+    "isolatedModules": true
+  }
+}

packages/in-memory-live-query-store/README.md

@@ -90,33 +90,41 @@ List of known and tested compatible transports/servers:
 ### Using with Redis
 
 You can use Redis to synchronize invalidations across multiple instances.
+[A full runnable example can be found here](../example-redis).
 
 ```ts
 import Redis from "ioredis";
-import {
-  InMemoryLiveQueryStore,
-  InMemoryLiveQueryStoreParameter,
-} from "@n1ru4l/in-memory-live-query-store";
-import { ExecutionArgs, execute as defaultExecute } from "graphql";
+import { InMemoryLiveQueryStore } from "@n1ru4l/in-memory-live-query-store";
+import { execute as defaultExecute } from "graphql";
 
-const CHANNEL = "LIVE_QUERY_INVALIDATIONS";
+const inMemoryLiveQueryStore = new InMemoryLiveQueryStore();
 
-export class RedisLiveQueryStore {
-  pub: Redis.Redis;
-  sub: Redis.Redis;
-  liveQueryStore: InMemoryLiveQueryStore;
+const client = new Redis(redisUri);
+const subClient = new Redis(redisUri);
 
-  constructor(redisUrl: string, parameter?: InMemoryLiveQueryStoreParameter) {
-    this.pub = new Redis(redisUrl);
-    this.sub = new Redis(redisUrl);
-    this.liveQueryStore = new InMemoryLiveQueryStore(parameter);
+class RedisLiveQueryStore {
+  pub: Redis;
+  sub: Redis;
+  channel: string;
+  liveQueryStore: InMemoryLiveQueryStore;
 
-    this.sub.subscribe(CHANNEL, (err) => {
+  constructor(
+    pub: Redis,
+    sub: Redis,
+    channel: string,
+    liveQueryStore: InMemoryLiveQueryStore
+  ) {
+    this.pub = pub;
+    this.sub = sub;
+    this.liveQueryStore = liveQueryStore;
+    this.channel = channel;
+
+    this.sub.subscribe(this.channel, (err) => {
       if (err) throw err;
     });
 
     this.sub.on("message", (channel, resourceIdentifier) => {
-      if (channel === CHANNEL && resourceIdentifier)
+      if (channel === this.channel && resourceIdentifier)
         this.liveQueryStore.invalidate(resourceIdentifier);
     });
   }
@@ -126,12 +134,19 @@ export class RedisLiveQueryStore {
       identifiers = [identifiers];
     }
     for (const identifier of identifiers) {
-      this.pub.publish(CHANNEL, identifier);
+      this.pub.publish(this.channel, identifier);
     }
   }
 
   makeExecute(execute: typeof defaultExecute) {
-    return this.liveQueryStore.makeExecute(args);
+    return this.liveQueryStore.makeExecute(execute);
   }
 }
+
+const liveQueryStore = new RedisLiveQueryStore(
+  client,
+  subClient,
+  "live-query-invalidations",
+  inMemoryLiveQueryStore
+);
 ```