Removed the Redis pubsub, just use the presence key

Author: matt-aitken

apps/webapp/app/presenters/v3/DevPresenceStream.server.ts

@@ -1,12 +1,7 @@
 const PRESENCE_KEY_PREFIX = "dev-presence:connection:";
-const PRESENCE_CHANNEL_PREFIX = "dev-presence:updates:";
 
 export class DevPresenceStream {
   static getPresenceKey(environmentId: string) {
     return `${PRESENCE_KEY_PREFIX}${environmentId}`;
   }
-
-  static getPresenceChannel(environmentId: string) {
-    return `${PRESENCE_CHANNEL_PREFIX}${environmentId}`;
-  }
 }

apps/webapp/app/routes/engine.v1.dev.presence.ts

@@ -29,50 +29,26 @@ export const loader = createSSELoader({
     const environmentId = authentication.environment.id;
 
     const presenceKey = DevPresenceStream.getPresenceKey(environmentId);
-    const presenceChannel = DevPresenceStream.getPresenceChannel(environmentId);
+
+    const ttl = (env.DEV_PRESENCE_POLL_INTERVAL_MS / 1000) * 2;
 
     return {
       beforeStream: async () => {
         logger.debug("Start dev presence SSE session", {
           environmentId,
           presenceKey,
-          presenceChannel,
         });
       },
       initStream: async ({ send }) => {
         // Set initial presence with more context
-        await redis.setex(presenceKey, env.DEV_PRESENCE_TTL_MS / 1000, new Date().toISOString());
-
-        // Publish presence update
-        await redis.publish(
-          presenceChannel,
-          JSON.stringify({
-            type: "connected",
-            environmentId,
-            timestamp: Date.now(),
-          })
-        );
-
+        await redis.setex(presenceKey, ttl, new Date().toISOString());
         send({ event: "start", data: `Started ${id}` });
       },
       iterator: async ({ send, date }) => {
-        await redis.setex(presenceKey, env.DEV_PRESENCE_TTL_MS / 1000, date.toISOString());
-
+        await redis.setex(presenceKey, ttl, date.toISOString());
         send({ event: "time", data: new Date().toISOString() });
       },
-      cleanup: async () => {
-        await redis.del(presenceKey);
-
-        // Publish disconnect event
-        await redis.publish(
-          presenceChannel,
-          JSON.stringify({
-            type: "disconnected",
-            environmentId,
-            timestamp: Date.now(),
-          })
-        );
-      },
+      cleanup: async () => {},
     };
   },
 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dev.presence.tsx

@@ -33,23 +33,15 @@ export const loader = createSSELoader({
     }
 
     const presenceKey = DevPresenceStream.getPresenceKey(environment.id);
-    const presenceChannel = DevPresenceStream.getPresenceChannel(environment.id);
 
-    // Create two Redis clients - one for subscribing and one for regular commands
-    const redisConfig = {
+    const cmdRedis = new Redis({
       port: env.RUN_ENGINE_DEV_PRESENCE_REDIS_PORT ?? undefined,
       host: env.RUN_ENGINE_DEV_PRESENCE_REDIS_HOST ?? undefined,
       username: env.RUN_ENGINE_DEV_PRESENCE_REDIS_USERNAME ?? undefined,
       password: env.RUN_ENGINE_DEV_PRESENCE_REDIS_PASSWORD ?? undefined,
       enableAutoPipelining: true,
       ...(env.RUN_ENGINE_DEV_PRESENCE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
-    };
-
-    // Subscriber client for pubsub
-    const subRedis = new Redis(redisConfig);
-
-    // Command client for regular Redis commands
-    const cmdRedis = new Redis(redisConfig);
+    });
 
     const checkAndSendPresence = async (send: SendFunction) => {
       try {
@@ -93,35 +85,19 @@ export const loader = createSSELoader({
       beforeStream: async () => {
         logger.debug("Start dev presence listening SSE session", {
           environmentId: environment.id,
-          presenceChannel,
+          presenceKey,
         });
       },
       initStream: async ({ send }) => {
         await checkAndSendPresence(send);
 
-        //start subscribing with the subscriber client
-        await subRedis.subscribe(presenceChannel);
-
-        subRedis.on("message", async (channel, message) => {
-          if (channel === presenceChannel) {
-            try {
-              await checkAndSendPresence(send);
-            } catch (error) {
-              logger.error("Failed to parse presence message", { error, message });
-            }
-          }
-        });
-
         send({ event: "time", data: new Date().toISOString() });
       },
       iterator: async ({ send, date }) => {
         await checkAndSendPresence(send);
       },
       cleanup: async ({ send }) => {
         await checkAndSendPresence(send);
-
-        await subRedis.unsubscribe(presenceChannel);
-        await subRedis.quit();
         await cmdRedis.quit();
       },
     };