a lot

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

@@ -1,96 +1,80 @@
-import { combine } from '@/sse/combine';
-import { redisMessageIterator } from '@/sse/redis-message-iterator';
 import type { FastifyReply, FastifyRequest } from 'fastify';
+import type * as WebSocket from 'ws';
 
 import { getSafeJson } from '@mixan/common';
 import type { IServiceCreateEventPayload } from '@mixan/db';
-import { chQuery, getEvents } from '@mixan/db';
+import { getEvents, getLiveVisitors } 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) {
+export function getLiveEventInfo(key: string) {
   return key.split(':').slice(2) as [string, string];
 }
 
-export async function test(request: FastifyRequest, reply: FastifyReply) {
+export async function test(
+  req: FastifyRequest<{
+    Params: {
+      projectId: string;
+    };
+  }>,
+  reply: FastifyReply
+) {
   const [event] = await getEvents(
-    `SELECT * FROM events LIMIT 1 OFFSET ${Math.floor(Math.random() * 1000)}`
+    `SELECT * FROM events WHERE project_id = '${req.params.projectId}' AND name = 'screen_view' LIMIT 1`
   );
   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');
+  redis.set(
+    `live:event:${event.projectId}:${Math.random() * 1000}`,
+    '',
+    'EX',
+    10
+  );
+  reply.status(202).send(event);
 }
 
-export function events(
-  request: FastifyRequest<{
-    Params: { projectId: string };
-  }>,
-  reply: FastifyReply
+export function wsVisitors(
+  connection: {
+    socket: WebSocket;
+  },
+  req: FastifyRequest<{
+    Params: {
+      projectId: string;
+    };
+  }>
 ) {
-  const reqProjectId = request.params.projectId;
+  const { params } = req;
 
-  // Subscribe
   redisSub.subscribe('event');
-  redisSub.psubscribe('__key*:*');
-  const listeners: ((...args: any[]) => void)[] = [];
+  redisSub.psubscribe('__key*:expired');
 
-  const incomingEvents = redisMessageIterator({
-    listenOn: 'message',
-    async transformer(message) {
+  const message = (channel: string, message: string) => {
+    if (channel === 'event') {
       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),
-        };
+      if (event?.projectId === params.projectId) {
+        getLiveVisitors(params.projectId).then((count) => {
+          connection.socket.send(String(count));
+        });
       }
     }
-  }
+  };
+  const pmessage = (pattern: string, channel: string, message: string) => {
+    const [projectId] = getLiveEventInfo(message);
+    if (projectId && projectId === params.projectId) {
+      getLiveVisitors(params.projectId).then((count) => {
+        connection.socket.send(String(count));
+      });
+    }
+  };
 
-  reply.sse(consumeMessages());
+  redisSub.on('message', message);
+  redisSub.on('pmessage', pmessage);
 
-  reply.raw.on('close', () => {
+  connection.socket.on('close', () => {
     redisSub.unsubscribe('event');
     redisSub.punsubscribe('__key*:expired');
-    listeners.forEach((listener) => redisSub.off('message', listener));
+    redisSub.off('message', message);
+    redisSub.off('pmessage', pmessage);
   });
 }

apps/sdk-api/src/index.ts

@@ -1,6 +1,5 @@
 import cors from '@fastify/cors';
 import Fastify from 'fastify';
-import { FastifySSEPlugin } from 'fastify-sse-v2';
 import pino from 'pino';
 
 import { redisPub } from '@mixan/redis';
@@ -29,7 +28,6 @@ const startServer = async () => {
       origin: '*',
     });
 
-    fastify.register(FastifySSEPlugin);
     fastify.decorateRequest('projectId', '');
     fastify.register(eventRouter, { prefix: '/event' });
     fastify.register(profileRouter, { prefix: '/profile' });

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

@@ -1,18 +1,25 @@
 import * as controller from '@/controllers/live.controller';
+import fastifyWS from '@fastify/websocket';
 import type { FastifyPluginCallback } from 'fastify';
 
 const liveRouter: FastifyPluginCallback = (fastify, opts, done) => {
   fastify.route({
     method: 'GET',
-    url: '/events/test',
+    url: '/events/test/:projectId',
     handler: controller.test,
   });
 
-  fastify.route({
-    method: 'GET',
-    url: '/events/:projectId',
-    handler: controller.events,
+  fastify.register(fastifyWS);
+
+  fastify.register((fastify, _, done) => {
+    fastify.get(
+      '/visitors/:projectId',
+      { websocket: true },
+      controller.wsVisitors
+    );
+    done();
   });
+
   done();
 };
 

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

@@ -1,33 +0,0 @@
-// @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

@@ -1,49 +0,0 @@
-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 };
-  }
-}

apps/sdk-api/src/utils/parseReferrer.ts

@@ -1,3 +1,5 @@
+import { stripTrailingSlash } from '@mixan/common';
+
 import referrers from '../referrers';
 
 function getHostname(url: string | undefined) {
@@ -18,7 +20,7 @@ export function parseReferrer(url: string | undefined) {
   return {
     name: match?.name ?? '',
     type: match?.type ?? 'unknown',
-    url: url ?? '',
+    url: stripTrailingSlash(url ?? ''),
   };
 }