Merge pull request #86 from customerio/feature/INAPP-13259

BernardGatt · web-flow · commit 2e11e6ab7da5 · 2025-06-10T09:27:26.000+02:00
Adds realtime through SSE
diff --git a/src/gist.js b/src/gist.js
@@ -1,6 +1,6 @@
 import EventEmitter from "./utilities/event-emitter";
 import { log } from "./utilities/log";
-import { startQueueListener, checkMessageQueue } from "./managers/queue-manager";
+import { startQueueListener, checkMessageQueue, stopSSEListener } from "./managers/queue-manager";
 import { setUserToken, clearUserToken, useGuestSession } from "./managers/user-manager";
 import { showMessage, embedMessage, hideMessage, removePersistentMessage, fetchMessageByInstanceId, logBroadcastDismissedLocally } from "./managers/message-manager";
 import { setUserLocale } from "./managers/locale-manager";
@@ -63,6 +63,7 @@ export default class {
     if (this.config.useAnonymousSession) {
       useGuestSession();
     }
+    stopSSEListener();
     await startQueueListener();
   }
 
diff --git a/src/managers/message-manager.js b/src/managers/message-manager.js
@@ -98,8 +98,6 @@ function resetEmbedState(message) {
 }
 
 async function resetOverlayState(hideFirst, message) {
-  removeMessageByInstanceId(message.instanceId);
-  Gist.overlayInstanceId = null;
   if (hideFirst) {
     await hideOverlayComponent();
   } else {
@@ -110,6 +108,9 @@ async function resetOverlayState(hideFirst, message) {
     window.removeEventListener('message', handleGistEvents);
     window.removeEventListener('touchstart', handleTouchStartEvents);
   }
+
+  removeMessageByInstanceId(message.instanceId);
+  Gist.overlayInstanceId = null;
 }
 
 function loadMessageComponent(message, elementId = null) {
diff --git a/src/managers/queue-manager.js b/src/managers/queue-manager.js
@@ -1,16 +1,18 @@
 import Gist from '../gist';
 import { log } from "../utilities/log";
-import { getUserToken } from "./user-manager";
-import { getUserQueue, userQueueNextPullCheckLocalStoreName } from "../services/queue-service";
+import { getUserToken, isAnonymousUser } from "./user-manager";
+import { getUserQueue, getQueueSSEEndpoint, userQueueNextPullCheckLocalStoreName } from "../services/queue-service";
 import { showMessage, embedMessage } from "./message-manager";
 import { resolveMessageProperties } from "./gist-properties-manager";
 import { getKeyFromLocalStore } from '../utilities/local-storage';
 import { updateBroadcastsLocalStore, getEligibleBroadcasts, isShowAlwaysBroadcast } from './message-broadcast-manager';
 import { updateQueueLocalStore, getMessagesFromLocalStore, isMessageLoading, setMessageLoading } from './message-user-queue-manager';
+import { settings } from '../services/settings';
 
 var sleep = time => new Promise(resolve => setTimeout(resolve, time))
 var poll = (promiseFn, time) => promiseFn().then(sleep(time).then(() => poll(promiseFn, time)));
 var pollingSetup = false;
+let sseSource = null;
 
 export async function startQueueListener() {
   if (!pollingSetup) {
@@ -22,7 +24,7 @@ export async function startQueueListener() {
       log("User token not setup, queue not started.");
     }
   } else {
-    checkMessageQueue();
+    await checkMessageQueue();
   }
 }
 
@@ -42,7 +44,7 @@ export async function checkMessageQueue() {
 async function handleMessage(message) {
   var messageProperties = resolveMessageProperties(message);
   if (messageProperties.hasRouteRule) {
-    var currentUrl = Gist.currentRoute
+    var currentUrl = Gist.currentRoute;
     if (currentUrl == null) {
       currentUrl = new URL(window.location.href).pathname;
     }
@@ -75,6 +77,23 @@ async function handleMessage(message) {
 }
 
 export async function pullMessagesFromQueue() {
+  // If SSE connection is already active, just check the local queue
+  if (settings.hasActiveSSEConnection()) {
+    await checkMessageQueue();
+    return;
+  }
+
+  // If SSE is enabled and user is not anonymous, set up SSE listener
+  if (settings.useSSE() && !isAnonymousUser()) {
+    await setupSSEQueueListener();
+    return;
+  }
+
+  // Fall back to polling
+  await checkQueueThroughPolling();
+}
+
+async function checkQueueThroughPolling() {
   if (getUserToken()) {
     if (Gist.isDocumentVisible) {
       // We're using the TTL as a way to determine if we should check the queue, so if the key is not there, we check the queue.
@@ -87,8 +106,7 @@ export async function pullMessagesFromQueue() {
             responseData = response.data;
             updateQueueLocalStore(responseData);
             updateBroadcastsLocalStore(responseData);
-          }
-          else if (response.status === 304) {
+          } else if (response.status === 304) {
             log("304 response, using local store.");
           }
           await checkMessageQueue();
@@ -99,9 +117,66 @@ export async function pullMessagesFromQueue() {
         log(`Next queue pull scheduled for later.`);
       }
     } else {
-      log(`Document not visible, skipping queue check.`);  
+      log(`Document not visible, skipping queue check.`);
     }
   } else {
     log(`User token reset, skipping queue check.`);
   }
-}
+}
+
+async function setupSSEQueueListener() {
+  const sseURL = getQueueSSEEndpoint();
+  if (sseURL === null) {
+    log("SSE endpoint not available, falling back to polling.");
+    await checkQueueThroughPolling();
+    return;
+  }
+  log(`Starting SSE queue listener on ${sseURL}`);
+  sseSource = new EventSource(sseURL);
+  settings.setActiveSSEConnection();
+
+  sseSource.addEventListener("connected", async (event) => {
+    log("SSE connection received:", event);
+    settings.setActiveSSEConnection();
+    settings.setUseSSEFlag(true);
+  });
+
+  sseSource.addEventListener("messages", async (event) => {
+    try {
+      var messages = JSON.parse(event.data);
+      log("SSE message received:", messages);
+      await updateQueueLocalStore(messages);
+      await updateBroadcastsLocalStore(messages);
+      await checkMessageQueue();
+    } catch (e) {
+      log("Failed to parse SSE message", e);
+      stopSSEListener();
+    }
+  });
+
+  sseSource.addEventListener("error", async (event) => {
+    log("SSE error received:", event);
+    stopSSEListener();
+  });
+
+  sseSource.addEventListener("heartbeat", async (event) => {
+    log("SSE heartbeat received:", event);
+    settings.setActiveSSEConnection();
+    settings.setUseSSEFlag(true);
+  });
+}
+
+export function stopSSEListener() {
+  // No active SSE connection to stop
+  if (!sseSource) {
+    return;
+  }
+
+  // Close the connection and clean up
+  log("Stopping SSE queue listener...");
+  sseSource.close();
+  sseSource = null;
+  
+  // Update settings to reflect disconnected state
+  settings.setUseSSEFlag(false);
+}
diff --git a/src/managers/user-manager.js b/src/managers/user-manager.js
@@ -47,6 +47,10 @@ export function useGuestSession() {
   }
 }
 
+export function isAnonymousUser() {
+  return isUsingGuestUserToken();
+}
+
 export async function getHashedUserToken() {
   var userToken = getUserToken();
   if (userToken === null) {
@@ -55,6 +59,14 @@ export async function getHashedUserToken() {
   return await hashString(userToken);
 }
 
+export function getEncodedUserToken() {
+  var userToken = getUserToken();
+  if (userToken === null) {
+    return null;
+  }
+  return btoa(userToken);
+}
+
 export function clearUserToken() {
   clearKeyFromLocalStore(userTokenLocalStoreName);
   log(`Cleared user token`);
diff --git a/src/services/network.js b/src/services/network.js
@@ -1,6 +1,6 @@
 import Gist from '../gist';
 import { settings } from './settings';
-import { getUserToken } from "../managers/user-manager";
+import { getEncodedUserToken } from "../managers/user-manager";
 
 export function UserNetworkInstance() {
   const baseURL =
@@ -13,9 +13,9 @@ export function UserNetworkInstance() {
     'X-CIO-Client-Platform': 'web',
   };
 
-  const userToken = getUserToken();
+  const userToken = getEncodedUserToken();
   if (userToken != null) {
-    defaultHeaders['X-Gist-Encoded-User-Token'] = btoa(userToken);
+    defaultHeaders['X-Gist-Encoded-User-Token'] = userToken;
   }
 
   async function request(path, options = {}) {
diff --git a/src/services/queue-service.js b/src/services/queue-service.js
@@ -1,7 +1,8 @@
+import Gist from '../gist';
 import { UserNetworkInstance } from './network';
 import { getKeyFromLocalStore, setKeyToLocalStore } from '../utilities/local-storage';
 import { log } from "../utilities/log";
-import { isUsingGuestUserToken } from '../managers/user-manager';
+import { isUsingGuestUserToken, getEncodedUserToken } from '../managers/user-manager';
 import { getUserLocale } from '../managers/locale-manager';
 import { settings } from './settings';
 import { v4 as uuidv4 } from 'uuid';
@@ -39,6 +40,7 @@ export async function getUserQueue() {
     checkInProgress = false;
     scheduleNextQueuePull(response);
     setQueueAPIVersion(response);
+    setQueueUseSSE(response);
   }
 
   return response;
@@ -53,6 +55,11 @@ function setQueueAPIVersion(response) {
   }
 }
 
+function setQueueUseSSE(response) {
+  const useSSE = response?.headers?.["x-cio-use-sse"]?.toLowerCase() === "true";
+  settings.setUseSSEFlag(useSSE);
+}
+
 function getSessionId() {
   var sessionId = getKeyFromLocalStore(sessionIdLocalStoreName);
   if (!sessionId) {
@@ -73,3 +80,12 @@ function scheduleNextQueuePull(response) {
   var expiryDate = new Date(new Date().getTime() + currentPollingDelayInSeconds * 1000);
   setKeyToLocalStore(userQueueNextPullCheckLocalStoreName, currentPollingDelayInSeconds, expiryDate);
 }
+
+export function getQueueSSEEndpoint() {
+  var encodedUserToken = getEncodedUserToken();
+  if (encodedUserToken === null) {
+    log("No user token available for SSE endpoint.");
+    return null;
+  }
+  return settings.GIST_QUEUE_REALTIME_API_ENDPOINT[Gist.config.env] + `/api/v3/sse?userToken=${encodedUserToken}&siteId=${Gist.config.siteId}&sessionId=${getSessionId()}`;
+}
diff --git a/src/services/settings.js b/src/services/settings.js
@@ -1,6 +1,8 @@
 import { getKeyFromLocalStore, setKeyToLocalStore } from '../utilities/local-storage';
 import { log } from '../utilities/log';
 const userQueueVersionLocalStoreName = "gist.web.userQueueVersion";
+const userQueueUseSSELocalStoreName = "gist.web.userQueueUseSSE";
+const userQueueActiveSSEConnectionLocalStoreName = "gist.web.activeSSEConnection";
 
 export const settings = {
   RENDERER_HOST: "https://code.gist.build",
@@ -19,6 +21,11 @@ export const settings = {
     "dev": "https://consumer.cloud.dev.gist.build",
     "local": "http://api.local.gist.build:86"
   },
+  GIST_QUEUE_REALTIME_API_ENDPOINT: {
+    "prod": "https://realtime.cloud.gist.build",
+    "dev": "https://realtime.cloud.dev.gist.build",
+    "local": "http://api.local.gist.build:3000"
+  },
   GIST_VIEW_ENDPOINT: {
     "prod": "https://renderer.gist.build/3.0",
     "dev": "https://renderer.gist.build/3.0",
@@ -31,5 +38,18 @@ export const settings = {
     // The Queue API version TTL is renewed with every poll request and extended by 30 minutes.
     setKeyToLocalStore(userQueueVersionLocalStoreName, version, new Date(new Date().getTime() + 1800000));
     log(`Set user queue version to "${version}"`);
+  },
+  useSSE: function() {
+    return getKeyFromLocalStore(userQueueUseSSELocalStoreName) ?? false;
+  },
+  setUseSSEFlag: function(useSSE) {
+    setKeyToLocalStore(userQueueUseSSELocalStoreName, useSSE, new Date(new Date().getTime() + 60000));
+    log(`Set user uses SSE to "${useSSE}"`);
+  },
+  setActiveSSEConnection: function() {
+    setKeyToLocalStore(userQueueActiveSSEConnectionLocalStoreName, true, new Date(new Date().getTime() + 31000));
+  },
+  hasActiveSSEConnection: function() {
+    return getKeyFromLocalStore(userQueueActiveSSEConnectionLocalStoreName) ?? false;
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`import EventEmitter from "./utilities/event-emitter";`
`2`	`2`	`import { log } from "./utilities/log";`
`3`		`-import { startQueueListener, checkMessageQueue } from "./managers/queue-manager";`
	`3`	`+import { startQueueListener, checkMessageQueue, stopSSEListener } from "./managers/queue-manager";`
`4`	`4`	`import { setUserToken, clearUserToken, useGuestSession } from "./managers/user-manager";`
`5`	`5`	`import { showMessage, embedMessage, hideMessage, removePersistentMessage, fetchMessageByInstanceId, logBroadcastDismissedLocally } from "./managers/message-manager";`
`6`	`6`	`import { setUserLocale } from "./managers/locale-manager";`
`@@ -63,6 +63,7 @@ export default class {`
`63`	`63`	`if (this.config.useAnonymousSession) {`
`64`	`64`	`useGuestSession();`
`65`	`65`	`}`
	`66`	`+ stopSSEListener();`
`66`	`67`	`await startQueueListener();`
`67`	`68`	`}`
`68`	`69`
Original file line number	Diff line number	Diff line change
`@@ -98,8 +98,6 @@ function resetEmbedState(message) {`
`98`	`98`	`}`
`99`	`99`
`100`	`100`	`async function resetOverlayState(hideFirst, message) {`
`101`		`- removeMessageByInstanceId(message.instanceId);`
`102`		`- Gist.overlayInstanceId = null;`
`103`	`101`	`if (hideFirst) {`
`104`	`102`	`await hideOverlayComponent();`
`105`	`103`	`} else {`
`@@ -110,6 +108,9 @@ async function resetOverlayState(hideFirst, message) {`
`110`	`108`	`window.removeEventListener('message', handleGistEvents);`
`111`	`109`	`window.removeEventListener('touchstart', handleTouchStartEvents);`
`112`	`110`	`}`
	`111`	`+`
	`112`	`+ removeMessageByInstanceId(message.instanceId);`
	`113`	`+ Gist.overlayInstanceId = null;`
`113`	`114`	`}`
`114`	`115`
`115`	`116`	`function loadMessageComponent(message, elementId = null) {`