server side events and ui improvemnt

apps/sdk-api/src/controllers/live.controller.ts

@@ -0,0 +1,96 @@
+import { combine } from '@/sse/combine';
+import { redisMessageIterator } from '@/sse/redis-message-iterator';
+import type { FastifyReply, FastifyRequest } from 'fastify';
+
+import { getSafeJson } from '@mixan/common';
+import type { IServiceCreateEventPayload } from '@mixan/db';
+import { chQuery, getEvents } from '@mixan/db';
+import { redis, redisPub, redisSub } from '@mixan/redis';
+
+async function getLiveCount(projectId: string) {
+  const keys = await redis.keys(`live:event:${projectId}:*`);
+  return keys.length;
+}
+
+function getLiveEventInfo(key: string) {
+  return key.split(':').slice(2) as [string, string];
+}
+
+export async function test(request: FastifyRequest, reply: FastifyReply) {
+  const [event] = await getEvents(
+    `SELECT * FROM events LIMIT 1 OFFSET ${Math.floor(Math.random() * 1000)}`
+  );
+  if (!event) {
+    return reply.status(404).send('No event found');
+  }
+  redisPub.publish('event', JSON.stringify(event));
+  redis.set(`live:event:${event.projectId}:${event.profileId}`, '', 'EX', 10);
+  reply.status(202).send('OK');
+}
+
+export function events(
+  request: FastifyRequest<{
+    Params: { projectId: string };
+  }>,
+  reply: FastifyReply
+) {
+  const reqProjectId = request.params.projectId;
+
+  // Subscribe
+  redisSub.subscribe('event');
+  redisSub.psubscribe('__key*:*');
+  const listeners: ((...args: any[]) => void)[] = [];
+
+  const incomingEvents = redisMessageIterator({
+    listenOn: 'message',
+    async transformer(message) {
+      const event = getSafeJson<IServiceCreateEventPayload>(message);
+      if (event && event.projectId === reqProjectId) {
+        return {
+          visitors: await getLiveCount(event.projectId),
+          event,
+        };
+      }
+      return null;
+    },
+    registerListener(fn) {
+      listeners.push(fn);
+    },
+  });
+
+  const expiredEvents = redisMessageIterator({
+    listenOn: 'pmessage',
+    async transformer(message) {
+      // message = live:event:${projectId}:${profileId}
+      const [projectId] = getLiveEventInfo(message);
+      if (projectId && projectId === reqProjectId) {
+        return {
+          visitors: await getLiveCount(projectId),
+          event: null as null | IServiceCreateEventPayload,
+        };
+      }
+      return null;
+    },
+    registerListener(fn) {
+      listeners.push(fn);
+    },
+  });
+
+  async function* consumeMessages() {
+    for await (const result of combine([incomingEvents, expiredEvents])) {
+      if (result.data) {
+        yield {
+          data: JSON.stringify(result.data),
+        };
+      }
+    }
+  }
+
+  reply.sse(consumeMessages());
+
+  reply.raw.on('close', () => {
+    redisSub.unsubscribe('event');
+    redisSub.punsubscribe('__key*:expired');
+    listeners.forEach((listener) => redisSub.off('message', listener));
+  });
+}

apps/sdk-api/src/index.ts

@@ -1,9 +1,12 @@
 import cors from '@fastify/cors';
 import Fastify from 'fastify';
+import { FastifySSEPlugin } from 'fastify-sse-v2';
 import pino from 'pino';
 
+import { redisPub } from '@mixan/redis';
+
 import eventRouter from './routes/event.router';
-import { validateSdkRequest } from './utils/auth';
+import liveRouter from './routes/live.router';
 
 declare module 'fastify' {
   interface FastifyRequest {
@@ -23,22 +26,10 @@ const startServer = async () => {
       origin: '*',
     });
 
+    fastify.register(FastifySSEPlugin);
     fastify.decorateRequest('projectId', '');
-
-    fastify.addHook('preHandler', (req, reply, done) => {
-      validateSdkRequest(req.headers)
-        .then((projectId) => {
-          req.projectId = projectId;
-          done();
-        })
-        .catch((e) => {
-          console.log(e);
-
-          reply.status(401).send();
-        });
-    });
-
     fastify.register(eventRouter, { prefix: '/event' });
+    fastify.register(liveRouter, { prefix: '/live' });
     fastify.setErrorHandler((error, request, reply) => {
       fastify.log.error(error);
     });
@@ -65,6 +56,9 @@ const startServer = async () => {
     }
 
     await fastify.listen({ host: '0.0.0.0', port });
+
+    // Notify when keys expires
+    redisPub.config('SET', 'notify-keyspace-events', 'Ex');
   } catch (e) {
     console.error(e);
   }

apps/sdk-api/src/routes/event.router.ts

@@ -1,7 +1,21 @@
 import * as controller from '@/controllers/event.controller';
+import { validateSdkRequest } from '@/utils/auth';
 import type { FastifyPluginCallback } from 'fastify';
 
 const eventRouter: FastifyPluginCallback = (fastify, opts, done) => {
+  fastify.addHook('preHandler', (req, reply, done) => {
+    validateSdkRequest(req.headers)
+      .then((projectId) => {
+        req.projectId = projectId;
+        done();
+      })
+      .catch((e) => {
+        console.log(e);
+
+        reply.status(401).send();
+      });
+  });
+
   fastify.route({
     method: 'POST',
     url: '/',

apps/sdk-api/src/routes/live.router.ts

@@ -0,0 +1,19 @@
+import * as controller from '@/controllers/live.controller';
+import type { FastifyPluginCallback } from 'fastify';
+
+const liveRouter: FastifyPluginCallback = (fastify, opts, done) => {
+  fastify.route({
+    method: 'GET',
+    url: '/events/test',
+    handler: controller.test,
+  });
+
+  fastify.route({
+    method: 'GET',
+    url: '/events/:projectId',
+    handler: controller.events,
+  });
+  done();
+};
+
+export default liveRouter;

apps/sdk-api/src/sse/combine.ts

@@ -0,0 +1,33 @@
+// @ts-nocheck
+
+export async function* combine<T>(iterable: AsyncGenerator<T>[]): T[] {
+  const asyncIterators = Array.from(iterable, (o) => o[Symbol.asyncIterator]());
+  const results = [];
+  let count = asyncIterators.length;
+  const never = new Promise(() => {});
+  function getNext(asyncIterator: AsyncGenerator, index: number) {
+    return asyncIterator.next().then((result) => ({
+      index,
+      result,
+    }));
+  }
+  const nextPromises = asyncIterators.map(getNext);
+  try {
+    while (count) {
+      const { index, result } = await Promise.race(nextPromises);
+      if (result.done) {
+        nextPromises[index] = never;
+        results[index] = result.value;
+        count--;
+      } else {
+        nextPromises[index] = getNext(asyncIterators[index], index);
+        yield result.value;
+      }
+    }
+  } finally {
+    for (const [index, iterator] of asyncIterators.entries())
+      if (nextPromises[index] != never && iterator.return != null)
+        iterator.return();
+  }
+  return results;
+}

apps/sdk-api/src/sse/redis-message-iterator.ts

@@ -0,0 +1,49 @@
+import { redisSub } from '@mixan/redis';
+
+export async function* redisMessageIterator<T>(opts: {
+  transformer: (payload: string) => Promise<T>;
+  listenOn: string;
+  registerListener: (listener: (...args: any[]) => void) => void;
+}) {
+  // Subscribe to a channel
+  interface Payload {
+    data: T;
+  }
+  // Promise resolver to signal new messages
+  let messageNotifier: null | ((payload: Payload) => void) = null;
+
+  // Promise to wait for new messages
+  let waitForMessage: Promise<Payload> = new Promise<Payload>((resolve) => {
+    messageNotifier = resolve;
+  });
+
+  async function listener(pattern: string, channel: string, message: string) {
+    const data = await opts.transformer(
+      pattern && channel && message ? message : channel
+    );
+
+    // Resolve the waiting promise to notify the generator of new message arrival
+    if (typeof messageNotifier === 'function') {
+      messageNotifier({ data });
+    }
+    // Clear the notifier to avoid multiple resolutions for a single message
+    messageNotifier = null;
+  }
+
+  // Event listener for messages on the subscribed channel
+  redisSub.on(opts.listenOn, listener);
+  opts.registerListener(listener);
+
+  while (true) {
+    // Wait for a new message
+    const { data } = await waitForMessage;
+
+    // Reset the waiting promise for the next message
+    waitForMessage = new Promise((resolve) => {
+      messageNotifier = resolve;
+    });
+
+    // Yield the received message
+    yield { data };
+  }
+}