chore: reconfigured cors options and updated sse reconnection in sdk

Davidthecode · Davidthecode · commit 6bccd335e2f3 · 2025-12-29T19:10:30.000+01:00
diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts
@@ -1,6 +1,7 @@
 import { toNodeHandler } from "better-auth/node";
 import cors from "cors";
 import express, { type Express } from "express";
+import { corsOptions } from "@/config/cors";
 import { auth } from "@/lib/auth/auth";
 import { requireAuth } from "@/middleware/auth";
 import flagRoutes from "@/routes/flag/route";
@@ -12,12 +13,7 @@ import trackRoutes from "@/routes/track/route";
 const createApp = (): Express => {
   const app: Express = express();
 
-  app.use(
-    cors({
-      origin: "http://localhost:3000",
-      credentials: true,
-    })
-  );
+  app.use(cors(corsOptions));
 
   app.all("/api/auth/*splat", toNodeHandler(auth));
 
diff --git a/apps/api/src/config/cors.ts b/apps/api/src/config/cors.ts
@@ -0,0 +1,34 @@
+import type { CorsOptionsDelegate } from "cors";
+import type { Request } from "express";
+import { env } from "@/config/env";
+
+const SDK_PREFIXES = ["/api/flag-config", "/api/track", "/api/sse"];
+const ALLOWED_ADMIN_ORIGINS = [env.FRONTEND_URL, "http://localhost:3000"];
+
+export const corsOptions: CorsOptionsDelegate = (req, callback) => {
+  const expressReq = req as Request;
+
+  const origin = expressReq.header("Origin");
+  const isSdkRoute = SDK_PREFIXES.some((path) =>
+    expressReq.path.startsWith(path)
+  );
+
+  if (isSdkRoute) {
+    return callback(null, {
+      origin: origin || true,
+      credentials: true,
+      methods: ["GET", "POST", "OPTIONS"],
+      allowedHeaders: ["Content-Type", "X-Api-Key"],
+      maxAge: 86_400,
+    });
+  }
+
+  if (origin && ALLOWED_ADMIN_ORIGINS.includes(origin)) {
+    return callback(null, {
+      origin: true,
+      credentials: true,
+    });
+  }
+
+  callback(null, { origin: false });
+};
diff --git a/apps/api/src/lib/sse/service.ts b/apps/api/src/lib/sse/service.ts
@@ -41,6 +41,7 @@ export function sseHandler(req: Request, res: Response, environmentId: string) {
   res.setHeader("Content-Type", "text/event-stream");
   res.setHeader("Cache-Control", "no-cache");
   res.setHeader("Connection", "keep-alive");
+  res.setHeader("X-Accel-Buffering", "no"); // Disable nginx buffering if present
   res.flushHeaders();
 
   if (!clients.has(environmentId)) {
@@ -50,9 +51,47 @@ export function sseHandler(req: Request, res: Response, environmentId: string) {
 
   res.write("event: connected\ndata: {}\n\n");
 
+  const keepaliveInterval = setInterval(() => {
+    try {
+      if (!res.writable || res.destroyed) {
+        clearInterval(keepaliveInterval);
+        clients.get(environmentId)?.delete(res);
+        if (clients.get(environmentId)?.size === 0) {
+          clients.delete(environmentId);
+        }
+        return;
+      }
+
+      res.write(": keepalive\n\n");
+    } catch (error) {
+      console.error(
+        `[SSE] Error sending keepalive to environment ${environmentId}:`,
+        error
+      );
+      clearInterval(keepaliveInterval);
+      clients.get(environmentId)?.delete(res);
+      if (clients.get(environmentId)?.size === 0) {
+        clients.delete(environmentId);
+      }
+    }
+  }, 30_000);
+
   req.on("close", () => {
+    clearInterval(keepaliveInterval);
     console.log(`[SSE] Client disconnected from environment: ${environmentId}`);
+
     clients.get(environmentId)?.delete(res);
+
+    if (clients.get(environmentId)?.size === 0) {
+      clients.delete(environmentId);
+    }
+  });
+
+  req.on("error", (_error) => {
+    clearInterval(keepaliveInterval);
+
+    clients.get(environmentId)?.delete(res);
+
     if (clients.get(environmentId)?.size === 0) {
       clients.delete(environmentId);
     }
diff --git a/sdk/javascript/src/client.ts b/sdk/javascript/src/client.ts
@@ -30,6 +30,13 @@ export class FlagixClient {
   private isInitialized = false;
   private sseConnection: FlagStreamConnection | null = null;
   private readonly emitter: FlagixEventEmitter;
+  private reconnectAttempts = 0;
+  private reconnectTimeoutId: ReturnType<typeof setTimeout> | null = null;
+  private isReconnecting = false;
+  private hasEstablishedConnection = false;
+  private readonly maxReconnectAttempts = Number.POSITIVE_INFINITY;
+  private readonly baseReconnectDelay = 1000;
+  private readonly maxReconnectDelay = 30_000;
 
   constructor(options: FlagixClientOptions) {
     this.apiKey = options.apiKey;
@@ -161,23 +168,87 @@ export class FlagixClient {
   }
 
   private async setupSSEListener(): Promise<void> {
+    if (this.sseConnection) {
+      try {
+        this.sseConnection.close();
+      } catch (error) {
+        log(
+          "warn",
+          "[Flagix SDK] Error closing existing SSE connection",
+          error
+        );
+      }
+      this.sseConnection = null;
+    }
+
     const url = `${this.apiBaseUrl}/api/sse/stream`;
 
     const source = await createEventSource(url, this.apiKey);
     if (!source) {
+      log("warn", "[Flagix SDK] Failed to create EventSource. Retrying...");
+      this.scheduleReconnect();
       return;
     }
 
     this.sseConnection = source;
 
     source.onopen = () => {
+      this.reconnectAttempts = 0;
+      this.isReconnecting = false;
+      if (this.reconnectTimeoutId) {
+        clearTimeout(this.reconnectTimeoutId);
+        this.reconnectTimeoutId = null;
+      }
+
+      // If this is a reconnection and not the first connection, refresh the cache
+      // this ensures we have the latest flag values that may have changed while disconnected
+      if (this.hasEstablishedConnection && this.isInitialized) {
+        log(
+          "info",
+          "[Flagix SDK] SSE reconnected. Refreshing cache to sync with server..."
+        );
+        this.fetchInitialConfig().catch((error) => {
+          log(
+            "error",
+            "[Flagix SDK] Failed to refresh cache after reconnection",
+            error
+          );
+        });
+      } else {
+        this.hasEstablishedConnection = true;
+      }
+
       log("info", "[Flagix SDK] SSE connection established.");
     };
 
     source.onerror = (error) => {
-      log("error", "[Flagix SDK] SSE error", error);
+      const eventSource = error.target as EventSource;
+      const readyState = eventSource?.readyState;
+
+      // EventSource.readyState: 0 = CONNECTING, 1 = OPEN, 2 = CLOSED
+      if (readyState === 2) {
+        log(
+          "warn",
+          "[Flagix SDK] SSE connection closed. Attempting to reconnect..."
+        );
+        this.handleReconnect();
+      } else if (readyState === 0) {
+        log(
+          "warn",
+          "[Flagix SDK] SSE connection error (connecting state)",
+          error
+        );
+      } else {
+        log("error", "[Flagix SDK] SSE error", error);
+        this.handleReconnect();
+      }
     };
 
+    // Listen for the "connected" event from the server
+    source.addEventListener("connected", () => {
+      log("info", "[Flagix SDK] SSE connection confirmed by server.");
+    });
+
     source.addEventListener(EVENT_TO_LISTEN, (event) => {
       try {
         const data = JSON.parse(event.data);
@@ -195,6 +266,54 @@ export class FlagixClient {
     });
   }
 
+  private handleReconnect(): void {
+    if (this.isReconnecting || !this.isInitialized) {
+      return;
+    }
+
+    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
+      log(
+        "error",
+        "[Flagix SDK] Max reconnection attempts reached. Stopping reconnection."
+      );
+      return;
+    }
+
+    this.isReconnecting = true;
+    this.scheduleReconnect();
+  }
+
+  private scheduleReconnect(): void {
+    if (this.reconnectTimeoutId) {
+      clearTimeout(this.reconnectTimeoutId);
+    }
+
+    // Calculate exponential backoff delay with jitter
+    const delay = Math.min(
+      this.baseReconnectDelay * 2 ** this.reconnectAttempts,
+      this.maxReconnectDelay
+    );
+    // Add ±25% jitter to prevent thundering herd
+    const jitter = delay * 0.25 * (Math.random() * 2 - 1);
+    const finalDelay = Math.max(100, delay + jitter);
+
+    this.reconnectAttempts++;
+
+    log(
+      "info",
+      `[Flagix SDK] Scheduling SSE reconnection attempt ${this.reconnectAttempts} in ${Math.round(finalDelay)}ms...`
+    );
+
+    this.reconnectTimeoutId = setTimeout(() => {
+      this.isReconnecting = false;
+      this.reconnectTimeoutId = null;
+      this.setupSSEListener().catch((error) => {
+        log("error", "[Flagix SDK] Failed to reconnect SSE", error);
+        this.handleReconnect();
+      });
+    }, finalDelay);
+  }
+
   private async fetchSingleFlagConfig(
     flagKey: string,
     type: FlagUpdateType
@@ -281,8 +400,21 @@ export class FlagixClient {
    * Closes the Server-Sent Events (SSE) connection and cleans up resources.
    */
   close(): void {
+    if (this.reconnectTimeoutId) {
+      clearTimeout(this.reconnectTimeoutId);
+      this.reconnectTimeoutId = null;
+    }
+
+    this.isReconnecting = false;
+    this.reconnectAttempts = 0;
+    this.hasEstablishedConnection = false;
+
     if (this.sseConnection) {
-      this.sseConnection.close();
+      try {
+        this.sseConnection.close();
+      } catch (error) {
+        log("warn", "[Flagix SDK] Error closing SSE connection", error);
+      }
       this.sseConnection = null;
       log("info", "[Flagix SDK] SSE connection closed.");
     }
