fix(node-sdk): include userId when syncing company context

roncohen · roncohen · commit e5ffd815ca9d · 2026-03-27T15:36:10.000+01:00
diff --git a/packages/node-sdk/src/client.ts b/packages/node-sdk/src/client.ts
@@ -12,12 +12,17 @@ import {
   API_TIMEOUT_MS,
   FLAG_EVENT_RATE_LIMITER_WINDOW_SIZE_MS,
   FLAGS_REFETCH_MS,
+  PUBSUB_SSE_URL,
   loadConfig,
   REFLAG_LOG_PREFIX,
   SDK_VERSION,
   SDK_VERSION_HEADER_NAME,
 } from "./config";
 import fetchClient, { withRetry } from "./fetch-http-client";
+import {
+  FlagUpdatesSSESubscription,
+  openFlagUpdatesSSE,
+} from "./flag-updates-sse";
 import { isFlagsFallbackSnapshot } from "./flagsFallbackProvider";
 import { subscribe as triggerOnExit } from "./flusher";
 import inRequestCache from "./inRequestCache";
@@ -26,7 +31,7 @@ import { newRateLimiter } from "./rate-limiter";
 import type {
   BootstrappedFlags,
   CachedFlagDefinition,
-  CacheStrategy,
+  FlagsSyncMode,
   EvaluatedFlagsAPIResponse,
   FlagAPIResponse,
   FlagDefinition,
@@ -239,7 +244,8 @@ export class ReflagClient {
     configFile?: string;
     flagsFetchRetries: number;
     fetchTimeoutMs: number;
-    cacheStrategy: CacheStrategy;
+    flagsSyncMode: FlagsSyncMode;
+    flagsPushUrl: string;
   };
   httpClient: HttpClient;
 
@@ -257,10 +263,14 @@ export class ReflagClient {
 
   private initializationFinished = false;
   private canLoadFlagsFallbackProvider = true;
+  private pendingWaitForVersion: number | undefined;
+  private realtimeRefreshPromise: Promise<void> | undefined;
+  private flagsUpdatesSSESubscription: FlagUpdatesSSESubscription | undefined;
   private _initialize = once(async () => {
     const start = Date.now();
     if (!this._config.offline) {
       await this.flagsCache.refresh();
+      this.startFlagsUpdatesSSE();
     }
     this.logger.info(
       "Reflag initialized in " +
@@ -288,7 +298,7 @@ export class ReflagClient {
    * @param options.configFile - The path to the config file (optional).
    * @param options.flagsFetchRetries - Number of retries for fetching flags (optional, defaults to 3).
    * @param options.fetchTimeoutMs - Timeout for fetching flags (optional, defaults to 10000ms).
-   * @param options.cacheStrategy - The cache strategy to use for the client (optional, defaults to "periodically-update").
+   * @param options.flagsSyncMode - How flag definitions are synchronized (optional, defaults to "polling").
    *
    * @throws An error if the options are invalid.
    **/
@@ -351,6 +361,21 @@ export class ReflagClient {
       "fetchTimeoutMs must be a non-negative integer",
     );
 
+    ok(
+      options.flagsSyncMode === undefined ||
+        options.flagsSyncMode === "polling" ||
+        options.flagsSyncMode === "in-request" ||
+        options.flagsSyncMode === "push",
+      'flagsSyncMode must be one of "polling", "in-request", or "push"',
+    );
+
+    ok(
+      options.flagsPushUrl === undefined ||
+        (typeof options.flagsPushUrl === "string" &&
+          options.flagsPushUrl.length > 0),
+      "flagsPushUrl must be a non-empty string",
+    );
+
     if (!options.configFile) {
       options.configFile =
         (process.env.REFLAG_CONFIG_FILE ??
@@ -424,6 +449,10 @@ export class ReflagClient {
       logger: this.logger,
     });
 
+    const flagsSyncMode: FlagsSyncMode =
+      options.flagsSyncMode ??
+      (options.cacheStrategy === "in-request" ? "in-request" : "polling");
+
     this._config = {
       offline,
       apiBaseUrl: (config.apiBaseUrl ?? config.host) || API_BASE_URL,
@@ -441,7 +470,8 @@ export class ReflagClient {
       flagOverrides: baseFlagOverrides,
       flagsFetchRetries: options.flagsFetchRetries ?? 3,
       fetchTimeoutMs: options.fetchTimeoutMs ?? API_TIMEOUT_MS,
-      cacheStrategy: options.cacheStrategy ?? "periodically-update",
+      flagsSyncMode,
+      flagsPushUrl: options.flagsPushUrl ?? PUBSUB_SSE_URL,
     };
     this.baseFlagOverrides = baseFlagOverrides;
 
@@ -454,8 +484,16 @@ export class ReflagClient {
     }
 
     const fetchFlags = async () => {
+      const waitForVersion = this.pendingWaitForVersion;
+      this.pendingWaitForVersion = undefined;
+
+      const path =
+        waitForVersion === undefined
+          ? "features"
+          : `features?waitForVersion=${encodeURIComponent(String(waitForVersion))}`;
+
       const res = await this.get<FlagsAPIResponse>(
-        "features",
+        path,
         this._config.flagsFetchRetries,
       );
       if (!isObject(res) || !Array.isArray(res?.features)) {
@@ -467,21 +505,81 @@ export class ReflagClient {
       return compileFlagDefinitions(res.features);
     };
 
-    if (this._config.cacheStrategy === "periodically-update") {
-      this.flagsCache = periodicallyUpdatingCache<CachedFlagDefinition[]>(
+    if (this._config.flagsSyncMode === "push") {
+      this.flagsCache = inRequestCache<CachedFlagDefinition[]>(
+        Number.MAX_SAFE_INTEGER,
+        this.logger,
+        fetchFlags,
+      );
+    } else if (this._config.flagsSyncMode === "in-request") {
+      this.flagsCache = inRequestCache<CachedFlagDefinition[]>(
         this._config.refetchInterval,
         this.logger,
         fetchFlags,
       );
     } else {
-      this.flagsCache = inRequestCache<CachedFlagDefinition[]>(
+      this.flagsCache = periodicallyUpdatingCache<CachedFlagDefinition[]>(
         this._config.refetchInterval,
         this.logger,
         fetchFlags,
       );
     }
   }
 
+  private startFlagsUpdatesSSE() {
+    if (
+      this._config.offline ||
+      this._config.flagsSyncMode !== "push" ||
+      this.flagsUpdatesSSESubscription
+    ) {
+      return;
+    }
+
+    this.flagsUpdatesSSESubscription = openFlagUpdatesSSE({
+      url: this._config.flagsPushUrl,
+      headers: this._config.headers,
+      logger: this.logger,
+      onFlagStateVersion: (version) => {
+        this.pendingWaitForVersion =
+          this.pendingWaitForVersion === undefined
+            ? version
+            : Math.max(this.pendingWaitForVersion, version);
+        this.refreshFlagsToPendingVersion();
+      },
+    });
+  }
+
+  private refreshFlagsToPendingVersion() {
+    if (this.realtimeRefreshPromise) {
+      return;
+    }
+
+    this.realtimeRefreshPromise = (async () => {
+      while (true) {
+        const pendingAtStart = this.pendingWaitForVersion;
+        if (pendingAtStart === undefined) {
+          break;
+        }
+
+        await this.flagsCache.refresh();
+
+        const pendingNow = this.pendingWaitForVersion;
+        if (pendingNow === undefined || pendingNow <= pendingAtStart) {
+          break;
+        }
+      }
+    })()
+      .catch((error) => {
+        this.logger.warn("failed to refresh flags from SSE update", error);
+      })
+      .finally(() => {
+        this.realtimeRefreshPromise = undefined;
+        if (this.pendingWaitForVersion !== undefined) {
+          this.refreshFlagsToPendingVersion();
+        }
+      });
+  }
+
   private async loadFlagsFallbackDefinitions() {
     if (!this.canLoadFlagsFallbackProvider) {
       return undefined;
@@ -831,6 +929,7 @@ export class ReflagClient {
 
     await this.batchBuffer.flush();
     await this.flagsCache.waitRefresh();
+    await this.realtimeRefreshPromise;
   }
 
   /**
@@ -857,6 +956,9 @@ export class ReflagClient {
    * multiple background processes from running simultaneously.
    */
   public destroy() {
+    this.flagsUpdatesSSESubscription?.close();
+    this.flagsUpdatesSSESubscription = undefined;
+
     this.flagsCache.destroy();
     this.batchBuffer.destroy();
   }
@@ -1228,6 +1330,7 @@ export class ReflagClient {
       const { id: _, ...attributes } = options.company;
       promises.push(
         this.updateCompany(options.company.id, {
+          userId: options.user?.id,
           attributes,
           meta: options.meta,
         }),
diff --git a/packages/node-sdk/test/client.test.ts b/packages/node-sdk/test/client.test.ts
@@ -100,6 +100,17 @@ const expectedHeaders = {
   Authorization: `Bearer ${validOptions.secretKey}`,
 };
 
+function createSSEMessageStream(data: object) {
+  const encoder = new TextEncoder();
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(`event: message\ndata: ${JSON.stringify(data)}\n\n`),
+      );
+    },
+  });
+}
+
 const flagDefinitions: FlagsAPIResponse = {
   features: [
     {
@@ -989,6 +1000,75 @@ describe("ReflagClient", () => {
       );
     });
 
+    it("should refresh features with waitForVersion when receiving SSE updates", async () => {
+      const fetchMock = vi.spyOn(global, "fetch").mockResolvedValue(
+        new Response(
+          createSSEMessageStream({
+            id: "flag-state:test:0",
+            timestamp: 1774605839478,
+            name: "flag_state_updated",
+            clientId: "reflag-backend",
+            channel: "flag-state:test",
+            action: 0,
+            encoding: "json",
+            data: JSON.stringify({
+              envId: "endbMeqfmcgLyZ",
+              flagStateVersion: 22,
+            }),
+          }),
+          {
+            status: 200,
+            headers: {
+              "Content-Type": "text/event-stream",
+            },
+          },
+        ),
+      );
+
+      try {
+        httpClient.get.mockResolvedValue({
+          ok: true,
+          status: 200,
+          body: {
+            success: true,
+            ...flagDefinitions,
+          },
+        });
+
+        const client = new ReflagClient({
+          ...validOptions,
+          flagsSyncMode: "push",
+        });
+
+        await client.initialize();
+
+        await vi.waitFor(() => {
+          expect(httpClient.get).toHaveBeenCalledWith(
+            "https://api.example.com/features?waitForVersion=22",
+            expectedHeaders,
+            API_TIMEOUT_MS,
+          );
+        });
+
+        expect(fetchMock).toHaveBeenCalledWith(
+          "https://pubsub.reflag.com/sse",
+          {
+            method: "GET",
+            headers: {
+              ...expectedHeaders,
+              Accept: "text/event-stream",
+              "Cache-Control": "no-cache",
+            },
+            signal: expect.any(AbortSignal),
+          },
+        );
+
+        client.destroy();
+      } finally {
+        fetchMock.mockRestore();
+      }
+    });
+
     it("should load flag definitions from flagsFallbackProvider when live fetch fails", async () => {
       const savedAt = "2026-03-09T00:00:00.000Z";
       const flagsFallbackProvider: FlagsFallbackProvider = {
@@ -1083,7 +1163,7 @@ describe("ReflagClient", () => {
 
       const client = new ReflagClient({
         ...validOptions,
-        cacheStrategy: "in-request",
+        flagsSyncMode: "in-request",
         flagsFetchRetries: 1,
       });
 
@@ -1348,7 +1428,7 @@ describe("ReflagClient", () => {
               active: true,
             },
             type: "company",
-            userId: undefined,
+            userId: user.id,
           },
           {
             attributes: {
@@ -1546,7 +1626,7 @@ describe("ReflagClient", () => {
             employees: 100,
             name: "Acme Inc.",
           },
-          userId: undefined, // this is a bug, will fix in separate PR
+          userId: user.id,
           context: undefined,
         },
         {
@@ -1635,7 +1715,14 @@ describe("ReflagClient", () => {
       await client.initialize();
       client.getFlags({ user, company, other: otherContext });
 
-      expect(isAllowedSpy).toHaveBeenCalledWith("1GHpP+QfYperQ0AtD8bWPiRE4H0=");
+      expect(isAllowedSpy).toHaveBeenNthCalledWith(
+        1,
+        "5Zt35h50IPRNU8yXAj/YbPME/qE=",
+      );
+      expect(isAllowedSpy).toHaveBeenNthCalledWith(
+        2,
+        "J3G4oF56f2t+T6xYzES6inc78+c=",
+      );
     });
 
     it("should return evaluated flags when only user is defined", async () => {
@@ -1901,7 +1988,7 @@ describe("ReflagClient", () => {
             employees: 100,
             name: "Acme Inc.",
           },
-          userId: undefined, // this is a bug, will fix in separate PR
+          userId: user.id,
           context: undefined,
         },
         {
@@ -2559,7 +2646,14 @@ describe("ReflagClient", () => {
       await client.initialize();
       client.getFlagsForBootstrap({ user, company, other: otherContext });
 
-      expect(isAllowedSpy).toHaveBeenCalledWith("1GHpP+QfYperQ0AtD8bWPiRE4H0=");
+      expect(isAllowedSpy).toHaveBeenNthCalledWith(
+        1,
+        "5Zt35h50IPRNU8yXAj/YbPME/qE=",
+      );
+      expect(isAllowedSpy).toHaveBeenNthCalledWith(
+        2,
+        "J3G4oF56f2t+T6xYzES6inc78+c=",
+      );
     });
 
     it("should work in offline mode", async () => {
