Claude/great hawking al4kaa

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@flagsync/js-sdk",
-  "version": "0.7.1",
+  "version": "0.7.2-alpha.0",
   "description": "FlagSync SDK for JavaScript",
   "author": "Mike Chabot",
   "license": "Apache-2.0",

src/logger/messages.ts

@@ -10,6 +10,7 @@ const STORAGE_MANAGER_FACTORY_MESSAGE = {
 
 const STORAGE_MANAGER_MESSAGE = {
   STORAGE_SET_FLAGS: 'storing flags',
+  STORAGE_REPLACE_FLAGS: 'replacing flags',
   STORAGE_GET_FLAGS: 'getting flags',
   STORAGE_PARSE_FAIL: 'failed to parse flags from storage',
 } as const;
@@ -22,10 +23,14 @@ const SERVICE_MANAGER_MESSAGE = {
 const STREAM_MANAGER_MESSAGE = {
   STREAM_CONNECTED: 'connection established',
   STREAM_MESSAGE_RECEIVED: 'message received',
+  STREAM_FULL_SET_RECEIVED: 'full flag set received',
   STREAM_CONN_OPEN: 'connection is open',
   STREAM_CONN_CLOSE: 'ungraceful connection close',
   STREAM_CONN_CLOSING: 'gracefully closing event stream',
   STREAM_RECONNECT: 'reestablishing connection',
+  STREAM_STALE: 'no events or heartbeats received, restarting connection',
+  STREAM_RESYNC_SUCCESS: 'flags resynced after reconnect',
+  STREAM_RESYNC_FAILED: 'flags resync failed',
   STREAM_MALFORMED_EVENT: 'malformed message event',
   STREAM_UNKNOWN_EVENT_STATE: 'unknown error state',
 } as const;

src/managers/event/types.ts

@@ -43,6 +43,7 @@ export type FsEventType = (typeof FsEvent)[keyof typeof FsEvent];
 
 export const FsIntervalEvent = {
   UPDATE_RECEIVED: 'state::update-received',
+  UPDATE_RECEIVED_FULL: 'state::update-received-full',
 } as const;
 
 export type FsIntervalEventType =
@@ -61,4 +62,5 @@ export interface FsEventTypePayload {
 
 export interface FsInternalEventTypePayload {
   [FsIntervalEvent.UPDATE_RECEIVED]: EventFlagSetPayload;
+  [FsIntervalEvent.UPDATE_RECEIVED_FULL]: EventFlagSetPayload;
 }

src/managers/storage/localstorage-manager.ts

@@ -28,6 +28,19 @@ export function localStorageManager(settings: FsSettings): IStoreManager {
     localStorage.setItem(buildKey(), JSON.stringify(flagSet));
   }
 
+  /**
+   * Replace the entire flag set, in memory and in localStorage. Unlike the
+   * merge in `set`, this drops flags absent from the incoming set — the only
+   * way deletes can take effect.
+   */
+  function replace(incoming: FsFlagSet) {
+    flagSet = {
+      ...incoming,
+    };
+    log.debug(formatter(MESSAGE.STORAGE_REPLACE_FLAGS));
+    localStorage.setItem(buildKey(), JSON.stringify(flagSet));
+  }
+
   function get(): FsFlagSet {
     log.debug(formatter(MESSAGE.STORAGE_GET_FLAGS));
     const cached = localStorage.getItem(buildKey());
@@ -55,6 +68,7 @@ export function localStorageManager(settings: FsSettings): IStoreManager {
 
   return {
     set,
+    replace,
     get,
   };
 }

src/managers/storage/memory-manager.ts

@@ -23,6 +23,17 @@ export function memoryManager(params: FsSettings): IStoreManager {
     };
   }
 
+  /**
+   * Replace the entire flag set. Unlike the merge in `set`, this drops flags
+   * absent from the incoming set — the only way deletes can take effect.
+   */
+  function replace(incoming: FsFlagSet) {
+    log.debug(formatter(MESSAGE.STORAGE_REPLACE_FLAGS));
+    flagSet = {
+      ...incoming,
+    };
+  }
+
   function get(): FsFlagSet {
     log.debug(formatter(MESSAGE.STORAGE_GET_FLAGS));
     return {
@@ -32,6 +43,7 @@ export function memoryManager(params: FsSettings): IStoreManager {
 
   return {
     set,
+    replace,
     get,
   };
 }

src/managers/storage/storage-manger-factory.ts

@@ -35,10 +35,10 @@ export function storageManagerFactory(
   }
 
   /**
-   * The sync managers emit an internal event when an update is received, either
-   * by stream or poll. Streaming updates only include the changed flags, while
-   * poll updates include the entire flag set. The storage manager spreads
-   * the update, partial or full.
+   * The sync managers emit internal events when updates are received.
+   * UPDATE_RECEIVED carries a partial set (legacy SSE updates) and is merged;
+   * UPDATE_RECEIVED_FULL carries the entire flag set (poll, WebSocket, and
+   * full-sync SSE) and replaces the store, so deleted flags drop out.
    */
   eventManager.internal.on(
     FsIntervalEvent.UPDATE_RECEIVED,
@@ -48,5 +48,13 @@ export function storageManagerFactory(
     },
   );
 
+  eventManager.internal.on(
+    FsIntervalEvent.UPDATE_RECEIVED_FULL,
+    (flagSet: FsFlagSet) => {
+      manager.replace(flagSet);
+      eventManager.emit(FsEvent.SDK_UPDATE);
+    },
+  );
+
   return manager;
 }

src/managers/storage/types.ts

@@ -2,5 +2,6 @@ import type { FsFlagSet } from '~config/types';
 
 export interface IStoreManager {
   set: (flagSet: FsFlagSet) => void;
+  replace: (flagSet: FsFlagSet) => void;
   get: () => FsFlagSet;
 }

src/managers/sync/poll-manager.ts

@@ -29,7 +29,7 @@ export function pollManager(
       });
       log.debug(formatter(MESSAGE.POLL_SUCCESS));
       eventManager.internal.emit(
-        FsIntervalEvent.UPDATE_RECEIVED,
+        FsIntervalEvent.UPDATE_RECEIVED_FULL,
         res?.flags ?? {},
       );
     } catch (e) {

src/managers/sync/stream-manager.ts

@@ -3,6 +3,8 @@ import { EventSource } from 'extended-eventsource';
 import type { FsFlagSet } from '~config/types';
 import type { FsSettings } from '~config/types.internal';
 
+import { apiClientFactory } from '~api/api-client-factory';
+
 import { FsIntervalEvent, IEventManager } from '~managers/event/types';
 import type { ISyncManager } from '~managers/sync/types';
 
@@ -11,19 +13,76 @@ import { formatMsg } from '~logger/utils';
 
 const formatter = formatMsg.bind(null, 'stream-manager');
 
+/**
+ * The server emits a heartbeat every 25s to keep intermediaries (Cloudflare)
+ * from closing the connection as idle. If nothing at all arrives for several
+ * heartbeat periods, the connection is presumed half-dead and restarted.
+ */
+const STALE_CONNECTION_TIMEOUT_MS = 80_000;
+
 export const streamManager = (
   settings: FsSettings,
   eventManager: IEventManager,
 ): ISyncManager => {
   const { urls, log, context } = settings;
+  const { sdk } = apiClientFactory(settings);
 
-  let es: EventSource;
+  let es: EventSource | undefined;
+  let watchdog: ReturnType<typeof setTimeout> | undefined;
+  let killed = false;
+  let hasConnected = false;
+
+  /**
+   * Servers that support full-set sync send the entire evaluated value map
+   * as named "flags" events. Once one is seen, legacy partial "message"
+   * events are ignored to avoid applying the same change twice.
+   */
+  let serverSupportsFullSync = false;
+
+  /**
+   * Restart the connection if no event, heartbeat, or open arrives within
+   * the stale window. Catches half-dead sockets that emit no error, which
+   * `EventSource` would otherwise never recover from.
+   */
+  function resetWatchdog() {
+    clearTimeout(watchdog);
+    watchdog = setTimeout(() => {
+      log.warn(formatter(MESSAGE.STREAM_STALE));
+      es?.close();
+      start();
+    }, STALE_CONNECTION_TIMEOUT_MS);
+  }
+
+  /**
+   * Updates emitted while the connection was down are lost — there is no
+   * replay on reconnect — so fetch the full value map to converge the store.
+   */
+  async function resyncOnReconnect() {
+    try {
+      const res = await sdk.sdkControllerGetFlags({
+        context,
+      });
+      eventManager.internal.emit(
+        FsIntervalEvent.UPDATE_RECEIVED_FULL,
+        res?.flags ?? {},
+      );
+      log.debug(formatter(MESSAGE.STREAM_RESYNC_SUCCESS));
+    } catch (error) {
+      log.error(formatter(MESSAGE.STREAM_RESYNC_FAILED), error?.toString());
+    }
+  }
 
   function start() {
+    if (killed) {
+      return;
+    }
+
     /**
      * Create a new EventSource instance and listen for incoming flag updates.
+     * Handlers close over `source` rather than the reassignable `es`, so a
+     * lingering handler from a restarted connection can't act on the new one.
      */
-    es = new EventSource(
+    const source = new EventSource(
       `${urls.sse}/sse/sdk-updates/client?timestamp=${new Date().getTime()}`,
       {
         withCredentials: true,
@@ -34,21 +93,57 @@ export const streamManager = (
         },
       },
     );
+    es = source;
 
-    /**
-     * For debug only
-     */
-    es.onopen = () => {
+    resetWatchdog();
+
+    source.onopen = () => {
       log.debug(formatter(MESSAGE.STREAM_CONNECTED));
+      resetWatchdog();
+      if (hasConnected) {
+        resyncOnReconnect();
+      }
+      hasConnected = true;
     };
 
     /**
-     * When a message is received, parse the JSON and emit an event
-     * to the event manager. This is only a partial update, that is,
-     * the flag that changed.
+     * Full-set sync: the server sends the entire evaluated value map, which
+     * replaces the store. This is how flag creates and deletes take effect.
+     */
+    source.addEventListener('flags', (event: MessageEvent<any>) => {
+      resetWatchdog();
+      try {
+        const flagSet = JSON.parse(event.data) as FsFlagSet;
+        log.debug(formatter(MESSAGE.STREAM_FULL_SET_RECEIVED));
+        serverSupportsFullSync = true;
+        eventManager.internal.emit(
+          FsIntervalEvent.UPDATE_RECEIVED_FULL,
+          flagSet,
+        );
+      } catch (error) {
+        log.error(formatter(MESSAGE.STREAM_MALFORMED_EVENT), error?.toString());
+      }
+    });
+
+    /**
+     * Heartbeats carry no data; they keep intermediaries from closing the
+     * connection as idle and feed the staleness watchdog.
+     */
+    source.addEventListener('heartbeat', () => {
+      resetWatchdog();
+    });
+
+    /**
+     * Legacy partial update, merged into the store. Servers that send
+     * full-set "flags" events also send these for older SDKs; ignore them
+     * once full-set support has been observed.
      * @param event
      */
-    es.onmessage = (event: MessageEvent<any>) => {
+    source.onmessage = (event: MessageEvent<any>) => {
+      resetWatchdog();
+      if (serverSupportsFullSync) {
+        return;
+      }
       try {
         const flagSet = JSON.parse(event.data) as FsFlagSet;
         log.debug(formatter(MESSAGE.STREAM_MESSAGE_RECEIVED));
@@ -58,27 +153,29 @@ export const streamManager = (
       }
     };
 
-    es.onerror = (event: Event) => {
-      switch (es.readyState) {
-        case es.CONNECTING:
+    source.onerror = (event: Event) => {
+      switch (source.readyState) {
+        case source.CONNECTING:
           log.debug(formatter(MESSAGE.STREAM_RECONNECT));
           break;
-        case es.OPEN:
+        case source.OPEN:
           log.debug(formatter(MESSAGE.STREAM_CONN_OPEN));
           break;
-        case es.CLOSED:
+        case source.CLOSED:
           log.debug(formatter(MESSAGE.STREAM_CONN_CLOSE));
           break;
         default:
           log.debug(
-            `${formatter(MESSAGE.STREAM_UNKNOWN_EVENT_STATE)}: "${es.readyState}"`,
+            `${formatter(MESSAGE.STREAM_UNKNOWN_EVENT_STATE)}: "${source.readyState}"`,
             event.toString(),
           );
       }
     };
   }
 
   function kill() {
+    killed = true;
+    clearTimeout(watchdog);
     if (es) {
       log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING));
       es.close();

src/managers/sync/ws-manager.ts

@@ -17,13 +17,17 @@ export const wsManager = (
 
   let ws: WebSocket;
   let reconnectTimeout: NodeJS.Timeout | null = null;
+  let killed = false;
   const RECONNECT_DELAY = 5000;
 
   const url = new URL(`${urls.ws.replace('https', 'wss')}/sdk/connect`);
   url.searchParams.append('x-ridgeline-key', settings.sdkKey);
   url.searchParams.append('x-ridgeline-user-ctx', JSON.stringify(context));
 
   function connect() {
+    if (killed) {
+      return;
+    }
     ws = new WebSocket(url.toString());
 
     /**
@@ -46,8 +50,13 @@ export const wsManager = (
         const data = JSON.parse(event.data.toString());
         log.debug(formatter(MESSAGE.STREAM_MESSAGE_RECEIVED));
         if (data.type === 'flagUpdate') {
+          // Sunrise always pushes the entire evaluated value map, so replace
+          // the store rather than merge — this is how deletes propagate.
           const ruleset = data.flags as FsFlagSet;
-          eventManager.internal.emit(FsIntervalEvent.UPDATE_RECEIVED, ruleset);
+          eventManager.internal.emit(
+            FsIntervalEvent.UPDATE_RECEIVED_FULL,
+            ruleset,
+          );
         }
       } catch (error) {
         log.error(formatter(MESSAGE.STREAM_MALFORMED_EVENT), error?.toString());
@@ -62,7 +71,7 @@ export const wsManager = (
         formatter(MESSAGE.STREAM_CONN_CLOSE),
         `Code: ${event.code}, Reason: ${event.reason}`,
       );
-      if (event.code !== 1000) {
+      if (!killed && event.code !== 1000) {
         log.debug(formatter(MESSAGE.STREAM_RECONNECT));
         reconnectTimeout = setTimeout(connect, RECONNECT_DELAY);
       }
@@ -83,8 +92,21 @@ export const wsManager = (
     connect();
   }
 
+  /**
+   * A kill must tear down whatever state the connection cycle is in: a
+   * pending reconnect timer or a CONNECTING socket would otherwise
+   * reconnect after shutdown.
+   */
   function kill() {
-    if (ws && ws.readyState === ws.OPEN) {
+    killed = true;
+    if (reconnectTimeout) {
+      clearTimeout(reconnectTimeout);
+      reconnectTimeout = null;
+    }
+    if (
+      ws &&
+      (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING)
+    ) {
       log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING));
       // Use code 1000 for normal closure
       ws.close(1000, 'SDK shutting down.');