feat: Thread FDv1 fallback TTL through polling and streaming data sources

This commit will add in fallback TTL reading support as well as allow RN
native event source read headers.

Author: joker23

packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts

@@ -38,7 +38,12 @@ describe('EventSource', () => {
       abort: jest.fn(),
     };
 
-    jest.spyOn(window, 'XMLHttpRequest').mockImplementation(() => mockXhr as XMLHttpRequest);
+    const xhrSpy = jest.spyOn(window, 'XMLHttpRequest').mockImplementation(() => mockXhr as XMLHttpRequest);
+    // Preserve static constants that EventSource reads from the constructor reference.
+    // @ts-ignore
+    xhrSpy.LOADING = 3;
+    // @ts-ignore
+    xhrSpy.DONE = 4;
 
     eventSource = new EventSource<EventName>(uri, { logger });
     eventSource.onclose = jest.fn();
@@ -135,4 +140,31 @@ describe('EventSource', () => {
 
     expect(mockXhr.open).toHaveBeenLastCalledWith('GET', `${uri}?basis=initial`, true);
   });
+
+  test('calls onopen with parsed response headers', () => {
+    const onopen = jest.fn();
+    eventSource.onopen = onopen;
+
+    mockXhr.getAllResponseHeaders = jest.fn(
+      () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream',
+    );
+    mockXhr.responseText = '';
+
+    jest.runAllTimers();
+
+    mockXhr.readyState = 4;
+    mockXhr.status = 200;
+    mockXhr.onreadystatechange();
+
+    expect(onopen).toHaveBeenCalledTimes(1);
+    expect(onopen).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: 'open',
+        headers: expect.objectContaining({
+          'x-ld-fd-fallback': 'true',
+          'x-ld-fd-fallback-ttl': '60',
+        }),
+      }),
+    );
+  });
 });

packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts

@@ -194,7 +194,10 @@ export default class EventSource<E extends string = never> {
           if (this._status === this.CONNECTING) {
             this._retryCount = 0;
             this._status = this.OPEN;
-            this.dispatch('open', { type: 'open' });
+            this.dispatch('open', {
+              type: 'open',
+              headers: this._parseResponseHeaders(this._xhr.getAllResponseHeaders()),
+            });
             this._logger?.debug('[EventSource][onreadystatechange][OPEN] Connection opened.');
           }
 
@@ -353,12 +356,28 @@ export default class EventSource<E extends string = never> {
     }
   }
 
+  private _parseResponseHeaders(raw: string | null): Record<string, string> {
+    const result: Record<string, string> = {};
+    if (!raw) {
+      return result;
+    }
+    raw.split('\r\n').forEach((line) => {
+      const separatorIndex = line.indexOf(': ');
+      if (separatorIndex > 0) {
+        const key = line.slice(0, separatorIndex).toLowerCase();
+        const value = line.slice(separatorIndex + 2);
+        result[key] = value;
+      }
+    });
+    return result;
+  }
+
   dispatch<T extends EventType<E>>(type: T, data: EventSourceEvent<T>) {
     this._eventHandlers[type]?.forEach((handler: EventSourceListener<E, T>) => handler(data));
 
     switch (type) {
       case 'open':
-        this.onopen();
+        this.onopen(data);
         break;
       case 'close':
         this.onclose();
@@ -389,7 +408,7 @@ export default class EventSource<E extends string = never> {
     return this._status;
   }
 
-  onopen() {}
+  onopen(_e: any) {}
   onclose() {}
   onerror(_err: any) {}
   onretrying(_e: any) {}

packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts

@@ -10,6 +10,7 @@ export interface MessageEvent {
 
 export interface OpenEvent {
   type: 'open';
+  headers?: Record<string, string>;
 }
 
 export interface CloseEvent {

packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts

@@ -303,6 +303,23 @@ describe('given a goodbye event in the response', () => {
       expect(result.reason).toBe('server-shutdown');
     }
   });
+
+  it('emits terminal_error when the response header signals fallback', async () => {
+    const body = makeFDv2Body([{ event: 'goodbye', data: { reason: 'bye' } }]);
+    const requestor = makeRequestor({
+      status: 200,
+      headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '45' }),
+      body,
+    });
+
+    const result = await poll(requestor, undefined, logger);
+
+    expect(result.type).toBe('status');
+    if (result.type !== 'status') return;
+    expect(result.state).toBe('terminal_error');
+    expect(result.fdv1Fallback).toBe(true);
+    expect((result as any).fdv1FallbackTtlMs).toBe(45000);
+  });
 });
 
 describe('given a server error event in the response', () => {
@@ -468,3 +485,61 @@ describe('given a delete-object event', () => {
     }
   });
 });
+
+describe('given x-ld-fd-fallback-ttl header', () => {
+  it('reads TTL in seconds and converts to ms on a changeSet', async () => {
+    const body = makeFullPayloadBody({ flagA: { value: true } });
+    const requestor = makeRequestor({
+      status: 200,
+      headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }),
+      body,
+    });
+
+    const result = await poll(requestor, undefined, logger);
+
+    expect(result.fdv1Fallback).toBe(true);
+    expect((result as any).fdv1FallbackTtlMs).toBe(60000);
+  });
+
+  it('treats TTL "0" as indefinite (0 ms)', async () => {
+    const body = makeFullPayloadBody({ flagA: { value: true } });
+    const requestor = makeRequestor({
+      status: 200,
+      headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '0' }),
+      body,
+    });
+
+    const result = await poll(requestor, undefined, logger);
+
+    expect(result.fdv1Fallback).toBe(true);
+    expect((result as any).fdv1FallbackTtlMs).toBe(0);
+  });
+
+  it('leaves TTL undefined when the ttl header is absent but fallback is true', async () => {
+    const body = makeFullPayloadBody({ flagA: { value: true } });
+    const requestor = makeRequestor({
+      status: 200,
+      headers: makeHeaders({ 'x-ld-fd-fallback': 'true' }),
+      body,
+    });
+
+    const result = await poll(requestor, undefined, logger);
+
+    expect(result.fdv1Fallback).toBe(true);
+    expect((result as any).fdv1FallbackTtlMs).toBeUndefined();
+  });
+
+  it('stamps TTL on a non-success error response', async () => {
+    const requestor = makeRequestor({
+      status: 503,
+      headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' }),
+      body: null,
+    });
+
+    const result = await poll(requestor, undefined, logger);
+
+    expect(result.fdv1Fallback).toBe(true);
+    expect((result as any).fdv1FallbackTtlMs).toBe(30000);
+  });
+});
+

packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts

@@ -514,3 +514,146 @@ it('calling start twice does not create a second EventSource', () => {
 
   base.close();
 });
+
+it('reads x-ld-fd-fallback-ttl on the error path', async () => {
+  const mockEventSource = createMockEventSource();
+  const mockRequests = createMockRequests(mockEventSource);
+  const base = createBase(mockRequests, logger);
+  base.start();
+
+  const willRetry = simulateErrorFilter(mockRequests, {
+    status: 200,
+    message: 'fallback',
+    headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' },
+  });
+  expect(willRetry).toBe(false);
+
+  const result = await base.takeResult();
+  expect(result.type).toBe('status');
+  if (result.type !== 'status') return;
+  expect(result.state).toBe('terminal_error');
+  expect(result.fdv1Fallback).toBe(true);
+  expect((result as any).fdv1FallbackTtlMs).toBe(60000);
+
+  base.close();
+});
+
+it('defers the open fallback directive to the next changeSet', async () => {
+  const mockEventSource = createMockEventSource();
+  const mockRequests = createMockRequests(mockEventSource);
+  const base = createBase(mockRequests, logger);
+  base.start();
+
+  mockEventSource.onopen({
+    type: 'open',
+    headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '90' },
+  });
+
+  sendFullTransfer(mockEventSource, [{ key: 'flag-a', version: 1, value: 'green' }]);
+
+  const result = await base.takeResult();
+  expect(result.type).toBe('changeSet');
+  if (result.type !== 'changeSet') return;
+  expect(result.fdv1Fallback).toBe(true);
+  expect((result as any).fdv1FallbackTtlMs).toBe(90000);
+
+  base.close();
+});
+
+it('emits terminal_error for a goodbye with protocolFallbackTTL', async () => {
+  const mockEventSource = createMockEventSource();
+  const mockRequests = createMockRequests(mockEventSource);
+  const base = createBase(mockRequests, logger);
+  base.start();
+
+  simulateEvent(mockEventSource, 'goodbye', {
+    reason: 'falling back',
+    protocolFallbackTTL: 60,
+  });
+
+  const result = await base.takeResult();
+  expect(result.type).toBe('status');
+  if (result.type !== 'status') return;
+  expect(result.state).toBe('terminal_error');
+  expect(result.fdv1Fallback).toBe(true);
+  expect((result as any).fdv1FallbackTtlMs).toBe(60000);
+
+  base.close();
+});
+
+it('emits terminal_error for a goodbye when a deferred open directive is pending', async () => {
+  // When onopen sets pendingFallback but the goodbye event has no protocolFallbackTTL,
+  // the pending directive triggers the fallback path with the deferred TTL.
+  const mockEventSource = createMockEventSource();
+  const mockRequests = createMockRequests(mockEventSource);
+  const base = createBase(mockRequests, logger);
+  base.start();
+
+  // Open with fallback headers (defers the directive)
+  mockEventSource.onopen({
+    type: 'open',
+    headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '45' },
+  });
+
+  // Goodbye fires before any payload — the pending directive triggers the fallback
+  simulateEvent(mockEventSource, 'goodbye', { reason: 'bye' });
+
+  const result = await base.takeResult();
+  expect(result.type).toBe('status');
+  if (result.type !== 'status') return;
+  expect(result.state).toBe('terminal_error');
+  expect(result.fdv1Fallback).toBe(true);
+  expect(result.fdv1FallbackTtlMs).toBe(45000);
+
+  base.close();
+});
+
+it('clears a pending fallback directive when onopen fires without the fallback header', async () => {
+  const mockEventSource = createMockEventSource();
+  const mockRequests = createMockRequests(mockEventSource);
+  const base = createBase(mockRequests, logger);
+  base.start();
+
+  // First open carries fallback headers — arms the deferred directive
+  mockEventSource.onopen({
+    type: 'open',
+    headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' },
+  });
+
+  // Reconnect without fallback header — must clear the stale directive
+  mockEventSource.onopen({ type: 'open', headers: {} });
+
+  // A payload arriving after the clean reconnect must NOT carry fdv1Fallback=true
+  sendFullTransfer(mockEventSource, [{ key: 'flag-a', version: 1, value: 'blue' }]);
+
+  const result = await base.takeResult();
+  expect(result.type).toBe('changeSet');
+  if (result.type !== 'changeSet') return;
+  expect(result.fdv1Fallback).toBe(false);
+  expect(result.fdv1FallbackTtlMs).toBeUndefined();
+
+  base.close();
+});
+
+it('close resets the pending fallback directive', async () => {
+  // Calling close() must clear pendingFallback/pendingFallbackTtlMs so a
+  // subsequently re-created source does not inherit stale state.
+  const mockEventSource = createMockEventSource();
+  const mockRequests = createMockRequests(mockEventSource);
+  const base = createBase(mockRequests, logger);
+  base.start();
+
+  // Arm the deferred directive
+  mockEventSource.onopen({
+    type: 'open',
+    headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' },
+  });
+
+  // Close before any payload arrives — the shutdown result must NOT carry TTL
+  base.close();
+  const result = await base.takeResult();
+  expect(result.type).toBe('status');
+  if (result.type !== 'status') return;
+  expect(result.state).toBe('shutdown');
+  expect(result.fdv1FallbackTtlMs).toBeUndefined();
+});