Revert "Optimise redis operations"

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

@@ -8,10 +8,17 @@ import {
   transformMinimalEvent,
 } from '@openpanel/db';
 import { setSuperJson } from '@openpanel/json';
-import { subscribeToPublishedEvent } from '@openpanel/redis';
+import {
+  psubscribeToPublishedEvent,
+  subscribeToPublishedEvent,
+} from '@openpanel/redis';
 import { getProjectAccess } from '@openpanel/trpc';
 import { getOrganizationAccess } from '@openpanel/trpc/src/access';
 
+export function getLiveEventInfo(key: string) {
+  return key.split(':').slice(2) as [string, string];
+}
+
 export function wsVisitors(
   socket: WebSocket,
   req: FastifyRequest<{
@@ -29,8 +36,21 @@ export function wsVisitors(
     }
   });
 
+  const punsubscribe = psubscribeToPublishedEvent(
+    '__keyevent@0__:expired',
+    (key) => {
+      const [projectId] = getLiveEventInfo(key);
+      if (projectId && projectId === params.projectId) {
+        eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
+          socket.send(String(count));
+        });
+      }
+    },
+  );
+
   socket.on('close', () => {
     unsubscribe();
+    punsubscribe();
   });
 }
 

packages/db/src/buffers/event-buffer.ts

@@ -278,6 +278,8 @@ return added
       if (!_multi) {
         await multi.exec();
       }
+
+      await publishEvent('events', 'received', transformEvent(event));
     } catch (error) {
       this.logger.error('Failed to add event to Redis buffer', { error });
     }