realtime voting system

package.json

@@ -17,6 +17,7 @@
   },
   "dependencies": {
     "@fastify/cookie": "^9.3.1",
+    "@fastify/websocket": "^8.3.1",
     "@prisma/client": "^5.9.1",
     "fastify": "^4.26.0",
     "ioredis": "^5.3.2",

src/http/routes/vote-on-poll.ts

@@ -3,6 +3,7 @@ import z from "zod";
 import { randomUUID } from "node:crypto";
 import { prisma } from "../../lib/prisma";
 import { redis } from "../../lib/redis";
+import { voting } from "../../utils/voting-pub-sub";
 
 export async function voteOnPoll(app: FastifyInstance) {
   app.post(
@@ -41,7 +42,16 @@ export async function voteOnPoll(app: FastifyInstance) {
             },
           });
 
-          await redis.zincrby(pollId, -1, userPreviousVoteOnPoll.pollOptionId);
+          const votes = await redis.zincrby(
+            pollId,
+            -1,
+            userPreviousVoteOnPoll.pollOptionId
+          );
+
+          voting.publish(pollId, {
+            pollOptionId: userPreviousVoteOnPoll.pollOptionId,
+            votes: Number(votes),
+          });
         } else if (userPreviousVoteOnPoll) {
           return reply
             .status(400)
@@ -68,7 +78,12 @@ export async function voteOnPoll(app: FastifyInstance) {
         },
       });
 
-      await redis.zincrby(pollId, 1, pollOptionId);
+      const votes = await redis.zincrby(pollId, 1, pollOptionId);
+
+      voting.publish(pollId, {
+        pollOptionId,
+        votes: Number(votes),
+      });
 
       return reply.status(201);
     }

src/http/server.ts

@@ -1,8 +1,11 @@
 import fastify from "fastify";
 import cookie from "@fastify/cookie";
+import websocket from "@fastify/websocket";
+
 import { createPoll } from "./routes/create-poll";
 import { getPoll } from "./routes/get-poll";
 import { voteOnPoll } from "./routes/vote-on-poll";
+import { pollResults } from "./ws/poll-results";
 
 const app = fastify();
 
@@ -12,10 +15,14 @@ app.register(cookie, {
   parseOptions: {},
 });
 
+app.register(websocket);
+
 app.register(createPoll);
 app.register(getPoll);
 app.register(voteOnPoll);
 
+app.register(pollResults);
+
 app.listen({ port: 3333 }).then(() => {
   console.log("HTTP server running!");
 });

src/http/ws/poll-results.ts

@@ -0,0 +1,21 @@
+import { FastifyInstance } from "fastify";
+import { z } from "zod";
+import { voting } from "../../utils/voting-pub-sub";
+
+export async function pollResults(app: FastifyInstance) {
+  app.get(
+    "/polls/:pollId/results",
+    { websocket: true },
+    (connection, request) => {
+      const getPollParams = z.object({
+        pollId: z.string().uuid(),
+      });
+
+      const { pollId } = getPollParams.parse(request.params);
+
+      voting.subscribe(pollId, (message) => {
+        connection.socket.send(JSON.stringify(message));
+      });
+    }
+  );
+}

src/utils/voting-pub-sub.ts

@@ -0,0 +1,26 @@
+type Message = { pollOptionId: string; votes: number };
+type Subscriber = (message: Message) => void;
+
+class VotingPubSub {
+  private channels: Record<string, Subscriber[]> = {};
+
+  subscribe(pollId: string, subscriber: Subscriber) {
+    if (!this.channels[pollId]) {
+      this.channels[pollId] = [];
+    }
+
+    this.channels[pollId].push(subscriber);
+  }
+
+  publish(pollId: string, message: Message) {
+    if (!this.channels[pollId]) {
+      return;
+    }
+
+    for (const subscriber of this.channels[pollId]) {
+      subscriber(message);
+    }
+  }
+}
+
+export const voting = new VotingPubSub();