adding the realtime notification handling logic.

Author: Samruddhi90

common/api-review/remote-config.api.md

@@ -42,11 +42,15 @@ export interface FetchResponse {
     config?: FirebaseRemoteConfigObject;
     eTag?: string;
     status: number;
+    templateVersionNumber?: number;
 }
 
 // @public
 export type FetchStatus = 'no-fetch-yet' | 'success' | 'failure' | 'throttle';
 
+// @public
+export type FetchType = 'BASE' | 'REALTIME';
+
 // @public
 export interface FirebaseRemoteConfigObject {
     // (undocumented)

packages/remote-config/src/client/realtime_handler.ts

@@ -17,17 +17,30 @@
 
 import { _FirebaseInstallationsInternal } from '@firebase/installations';
 import { Logger } from '@firebase/logger';
-import { ConfigUpdateObserver } from '../public_types';
+import {
+  ConfigUpdate,
+  ConfigUpdateObserver,
+  FetchResponse,
+  FirebaseRemoteConfigObject
+} from '../public_types';
 import { calculateBackoffMillis, FirebaseError } from '@firebase/util';
 import { ERROR_FACTORY, ErrorCode } from '../errors';
 import { Storage } from '../storage/storage';
 import { VisibilityMonitor } from './visibility_monitor';
+import { StorageCache } from '../storage/storage_cache';
+import {
+  FetchRequest,
+  RemoteConfigAbortSignal
+} from './remote_config_fetch_client';
+import { RestClient } from './rest_client';
 
 const API_KEY_HEADER = 'X-Goog-Api-Key';
 const INSTALLATIONS_AUTH_TOKEN_HEADER = 'X-Goog-Firebase-Installations-Auth';
 const ORIGINAL_RETRIES = 8;
+const MAXIMUM_FETCH_ATTEMPTS = 3;
 const NO_BACKOFF_TIME_IN_MILLIS = -1;
 const NO_FAILED_REALTIME_STREAMS = 0;
+const TEMPLATE_VERSION_KEY = 'latestTemplateVersionNumber';
 
 export class RealtimeHandler {
   constructor(
@@ -38,7 +51,9 @@ export class RealtimeHandler {
     private readonly projectId: string,
     private readonly apiKey: string,
     private readonly appId: string,
-    private readonly logger: Logger
+    private readonly logger: Logger,
+    private readonly storageCache: StorageCache,
+    private readonly restClient: RestClient
   ) {
     void this.setRetriesRemaining();
     void VisibilityMonitor.getInstance().on(
@@ -56,6 +71,7 @@ export class RealtimeHandler {
   private reader: ReadableStreamDefaultReader | undefined;
   private httpRetriesRemaining: number = ORIGINAL_RETRIES;
   private isInBackground: boolean = false;
+  private readonly decoder = new TextDecoder('utf-8');
 
   private async setRetriesRemaining(): Promise<void> {
     // Retrieve number of remaining retries from last session. The minimum retry count being one.
@@ -229,6 +245,239 @@ export class RealtimeHandler {
     return canMakeConnection;
   }
 
+  private fetchResponseIsUpToDate(
+    fetchResponse: FetchResponse,
+    lastKnownVersion: number
+  ): boolean {
+    if (fetchResponse.config != null && fetchResponse.templateVersionNumber) {
+      return fetchResponse.templateVersionNumber >= lastKnownVersion;
+    }
+    return false;
+  }
+
+  private parseAndValidateConfigUpdateMessage(message: string): string {
+    const left = message.indexOf('{');
+    const right = message.indexOf('}', left);
+
+    if (left < 0 || right < 0) {
+      return '';
+    }
+    return left >= right ? '' : message.substring(left, right + 1);
+  }
+
+  private isEventListenersEmpty(): boolean {
+    return this.observers.size === 0;
+  }
+
+  private getRandomInt(max: number): number {
+    return Math.floor(Math.random() * max);
+  }
+
+  private executeAllListenerCallbacks(configUpdate: ConfigUpdate): void {
+    this.observers.forEach(observer => observer.next(configUpdate));
+  }
+
+  private getChangedParams(
+    newConfig: FirebaseRemoteConfigObject,
+    oldConfig: FirebaseRemoteConfigObject
+  ): Set<string> {
+    const changed = new Set<string>();
+    const newKeys = new Set(Object.keys(newConfig || {}));
+    const oldKeys = new Set(Object.keys(oldConfig || {}));
+
+    for (const key of newKeys) {
+      if (!oldKeys.has(key)) {
+        changed.add(key);
+        continue;
+      }
+      if (
+        JSON.stringify((newConfig as any)[key]) !==
+        JSON.stringify((oldConfig as any)[key])
+      ) {
+        changed.add(key);
+        continue;
+      }
+    }
+
+    for (const key of oldKeys) {
+      if (!newKeys.has(key)) {
+        changed.add(key);
+      }
+    }
+    return changed;
+  }
+
+  private async fetchLatestConfig(
+    remainingAttempts: number,
+    targetVersion: number
+  ): Promise<void> {
+    const remainingAttemptsAfterFetch = remainingAttempts - 1;
+    const currentAttempt = MAXIMUM_FETCH_ATTEMPTS - remainingAttemptsAfterFetch;
+    const customSignals = this.storageCache.getCustomSignals();
+    if (customSignals) {
+      this.logger.debug(
+        `Fetching config with custom signals: ${JSON.stringify(customSignals)}`
+      );
+    }
+    try {
+      const fetchRequest: FetchRequest = {
+        cacheMaxAgeMillis: 0,
+        signal: new RemoteConfigAbortSignal(),
+        customSignals: customSignals,
+        fetchType: 'REALTIME',
+        fetchAttempt: currentAttempt
+      };
+
+      const fetchResponse: FetchResponse = await this.restClient.fetch(
+        fetchRequest
+      );
+      let activatedConfigs = await this.storage.getActiveConfig();
+
+      if (!this.fetchResponseIsUpToDate(fetchResponse, targetVersion)) {
+        this.logger.debug(
+          "Fetched template version is the same as SDK's current version." +
+            ' Retrying fetch.'
+        );
+        // Continue fetching until template version number is greater than current.
+        await this.autoFetch(remainingAttemptsAfterFetch, targetVersion);
+        return;
+      }
+
+      if (fetchResponse.config == null) {
+        this.logger.debug(
+          'The fetch succeeded, but the backend had no updates.'
+        );
+        return;
+      }
+
+      if (activatedConfigs == null) {
+        activatedConfigs = {};
+      }
+
+      const updatedKeys = this.getChangedParams(
+        fetchResponse.config,
+        activatedConfigs
+      );
+
+      if (updatedKeys.size === 0) {
+        this.logger.debug('Config was fetched, but no params changed.');
+        return;
+      }
+
+      const configUpdate: ConfigUpdate = {
+        getUpdatedKeys(): Set<string> {
+          return new Set(updatedKeys);
+        }
+      };
+      this.executeAllListenerCallbacks(configUpdate);
+    } catch (e: unknown) {
+      const errorMessage = e instanceof Error ? e.message : String(e);
+      const error = ERROR_FACTORY.create(ErrorCode.CONFIG_UPDATE_NOT_FETCHED, {
+        originalErrorMessage: `Failed to auto-fetch config update: ${errorMessage}`
+      });
+      this.propagateError(error);
+    }
+  }
+
+  private async autoFetch(
+    remainingAttempts: number,
+    targetVersion: number
+  ): Promise<void> {
+    if (remainingAttempts === 0) {
+      const error = ERROR_FACTORY.create(ErrorCode.CONFIG_UPDATE_NOT_FETCHED, {
+        originalErrorMessage:
+          'Unable to fetch the latest version of the template.'
+      });
+      this.propagateError(error);
+      return;
+    }
+
+    const timeTillFetch = this.getRandomInt(4);
+    setTimeout(async () => {
+      await this.fetchLatestConfig(remainingAttempts, targetVersion);
+    }, timeTillFetch);
+  }
+
+  private async handleNotifications(
+    reader: ReadableStreamDefaultReader
+  ): Promise<void> {
+    let partialConfigUpdateMessage: string;
+    let currentConfigUpdateMessage = '';
+
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) {
+        break;
+      }
+
+      partialConfigUpdateMessage = this.decoder.decode(value, { stream: true });
+      currentConfigUpdateMessage += partialConfigUpdateMessage;
+
+      if (partialConfigUpdateMessage.includes('}')) {
+        currentConfigUpdateMessage = this.parseAndValidateConfigUpdateMessage(
+          currentConfigUpdateMessage
+        );
+
+        if (currentConfigUpdateMessage.length === 0) {
+          continue;
+        }
+        try {
+          const jsonObject = JSON.parse(currentConfigUpdateMessage);
+
+          if (this.isEventListenersEmpty()) {
+            break;
+          }
+
+          if (TEMPLATE_VERSION_KEY in jsonObject) {
+            const oldTemplateVersion =
+              await this.storage.getLastKnownTemplateVersion();
+            let targetTemplateVersion = Number(
+              jsonObject[TEMPLATE_VERSION_KEY]
+            );
+
+            if (
+              oldTemplateVersion &&
+              targetTemplateVersion > oldTemplateVersion
+            ) {
+              await this.autoFetch(
+                MAXIMUM_FETCH_ATTEMPTS,
+                targetTemplateVersion
+              );
+            }
+          }
+        } catch (e: any) {
+          this.logger.error('Unable to parse latest config update message.', e);
+          this.propagateError(
+            ERROR_FACTORY.create(ErrorCode.CONFIG_UPDATE_MESSAGE_INVALID, {
+              originalErrorMessage: e
+            })
+          );
+        }
+        currentConfigUpdateMessage = '';
+      }
+    }
+  }
+
+  public async listenForNotifications(
+    reader: ReadableStreamDefaultReader
+  ): Promise<void> {
+    try {
+      await this.handleNotifications(reader);
+    } catch (e) {
+      if (!this.isInBackground) {
+        this.logger.debug(
+          'Real-time connection was closed due to an exception.',
+          e
+        );
+      }
+    } finally {
+      if (this.reader) {
+        this.reader.cancel();
+        this.reader = undefined;
+      }
+    }
+  }
+
   /**
    * Open the real-time connection, begin listening for updates, and auto-fetch when an update is
    * received.
@@ -263,8 +512,8 @@ export class RealtimeHandler {
       if (response.ok && response.body) {
         this.resetRetryCount();
         await this.resetRealtimeBackoff();
-        //const configAutoFetch = this.startAutoFetch(reader);
-        //await configAutoFetch.listenForNotifications();
+        const redaer = response.body.getReader();
+        this.listenForNotifications(redaer);
       }
     } catch (error) {
       if (this.isInBackground) {

packages/remote-config/src/client/remote_config_fetch_client.ts

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-import { CustomSignals, FetchResponse } from '../public_types';
+import { CustomSignals, FetchResponse, FetchType } from '../public_types';
 
 /**
  * Defines a client, as in https://en.wikipedia.org/wiki/Client%E2%80%93server_model, for the
@@ -100,4 +100,18 @@ export interface FetchRequest {
    * <p>Optional in case no custom signals are set for the instance.
    */
   customSignals?: CustomSignals;
+
+  /**
+   * The type of fetch to perform, such as a regular fetch or a real-time fetch.
+   *
+   * <p>Optional as not all fetch requests need to be distinguished.
+   */
+  fetchType?: FetchType;
+
+  /**
+   * The number of fetch attempts made so far for this request.
+   *
+   * <p>Optional as not all fetch requests are part of a retry series.
+   */
+  fetchAttempt?: number;
 }

packages/remote-config/src/client/rest_client.ts

@@ -82,12 +82,16 @@ export class RestClient implements RemoteConfigFetchClient {
 
     const url = `${urlBase}/v1/projects/${this.projectId}/namespaces/${this.namespace}:fetch?key=${this.apiKey}`;
 
+    const fetchType = request.fetchType || 'BASE';
+    const fetchAttempt = request.fetchAttempt || 0;
+
     const headers = {
       'Content-Type': 'application/json',
       'Content-Encoding': 'gzip',
       // Deviates from pure decorator by not passing max-age header since we don't currently have
       // service behavior using that header.
-      'If-None-Match': request.eTag || '*'
+      'If-None-Match': request.eTag || '*',
+      'X_FIREBASE_RC_FETCH_TYPE': `${fetchType}/${fetchAttempt}`
     };
 
     const requestBody: FetchRequestBody = {
@@ -140,6 +144,7 @@ export class RestClient implements RemoteConfigFetchClient {
 
     let config: FirebaseRemoteConfigObject | undefined;
     let state: string | undefined;
+    let templateVersionNumber: number | undefined;
 
     // JSON parsing throws SyntaxError if the response body isn't a JSON string.
     // Requesting application/json and checking for a 200 ensures there's JSON data.
@@ -154,6 +159,7 @@ export class RestClient implements RemoteConfigFetchClient {
       }
       config = responseBody['entries'];
       state = responseBody['state'];
+      templateVersionNumber = responseBody['templateVersionNumber'];
     }
 
     // Normalizes based on legacy state.
@@ -176,6 +182,6 @@ export class RestClient implements RemoteConfigFetchClient {
       });
     }
 
-    return { status, eTag: responseEtag, config };
+    return { status, eTag: responseEtag, config, templateVersionNumber };
   }
 }