Merge pull request #1189 from OneSignal/fix/externalIdHydration

[Fix] External Id hydration

__test__/unit/user/login.test.ts

@@ -22,24 +22,13 @@ jest.mock('../../../src/shared/libraries/Log');
 describe('Login tests', () => {
   beforeEach(() => {
     jest.useFakeTimers();
-    test.stub(
-      PropertiesExecutor.prototype,
-      'getOperationsFromCache',
-      Promise.resolve([]),
-    );
-    test.stub(
-      IdentityExecutor.prototype,
-      'getOperationsFromCache',
-      Promise.resolve([]),
-    );
-    test.stub(
-      SubscriptionExecutor.prototype,
-      'getOperationsFromCache',
-      Promise.resolve([]),
-    );
+    test.stub(PropertiesExecutor.prototype, 'getOperationsFromCache', []);
+    test.stub(IdentityExecutor.prototype, 'getOperationsFromCache', []);
+    test.stub(SubscriptionExecutor.prototype, 'getOperationsFromCache', []);
   });
 
   afterEach(() => {
+    jest.runOnlyPendingTimers();
     jest.resetAllMocks();
   });
 

src/core/CoreModule.ts

@@ -5,11 +5,13 @@ import { OSModelStoreFactory } from './modelRepo/OSModelStoreFactory';
 import Log from '../shared/libraries/Log';
 import { logMethodCall } from '../shared/utils/utils';
 import { SupportedModel } from './models/SupportedModels';
+import { NewRecordsState } from '../shared/models/NewRecordsState';
 
 export default class CoreModule {
   public modelRepo?: ModelRepo;
   public operationRepo?: OperationRepo;
   public initPromise: Promise<void>;
+  public newRecordsState?: NewRecordsState;
 
   private modelCache: ModelCache;
   private initResolver: () => void = () => null;
@@ -25,7 +27,11 @@ export default class CoreModule {
       .then((allCachedOSModels) => {
         const modelStores = OSModelStoreFactory.build(allCachedOSModels);
         this.modelRepo = new ModelRepo(this.modelCache, modelStores);
-        this.operationRepo = new OperationRepo(this.modelRepo);
+        this.newRecordsState = new NewRecordsState();
+        this.operationRepo = new OperationRepo(
+          this.modelRepo,
+          this.newRecordsState,
+        );
         this.initResolver();
       })
       .catch((e) => {

src/core/CoreModuleDirector.ts

@@ -63,14 +63,13 @@ export class CoreModuleDirector {
     await this.core.resetModelRepoAndCache();
   }
 
-  public hydrateUser(user: UserData): void {
-    logMethodCall('CoreModuleDirector.hydrateUser', { user });
+  public hydrateUser(user: UserData, externalId?: string): void {
+    logMethodCall('CoreModuleDirector.hydrateUser', { user, externalId });
     try {
       const identity = this.getIdentityModel();
       const properties = this.getPropertiesModel();
 
-      const { onesignal_id: onesignalId, external_id: externalId } =
-        user.identity;
+      const { onesignal_id: onesignalId } = user.identity;
 
       if (!onesignalId) {
         throw new OneSignalError('OneSignal ID is missing from user data');
@@ -83,6 +82,7 @@ export class CoreModuleDirector {
       if (externalId) {
         identity?.setExternalId(externalId);
         properties?.setExternalId(externalId);
+        user.identity.external_id = externalId;
       }
 
       // identity and properties models are always single, so we hydrate immediately (i.e. replace existing data)
@@ -109,6 +109,8 @@ export class CoreModuleDirector {
   ): void {
     logMethodCall('CoreModuleDirector._hydrateSubscriptions', {
       subscriptions,
+      onesignalId,
+      externalId,
     });
 
     if (!subscriptions) {
@@ -175,6 +177,9 @@ export class CoreModuleDirector {
   }
 
   /* G E T T E R S */
+  public getNewRecordsState(): NewRecordsState | undefined {
+    return this.core.newRecordsState;
+  }
 
   public getModelByTypeAndId(
     modelName: ModelName,

src/core/executors/ExecutorBase.ts

@@ -9,12 +9,14 @@ import { ExecutorResult } from './ExecutorResult';
 import Log from '../../shared/libraries/Log';
 import Database from '../../shared/services/Database';
 import LocalStorage from '../../shared/utils/LocalStorage';
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 
 const RETRY_AFTER = 5_000;
 
 export default abstract class ExecutorBase {
   protected _deltaQueue: CoreDelta<SupportedModel>[] = [];
   protected _operationQueue: Operation<SupportedModel>[] = [];
+  protected _newRecordsState: NewRecordsState;
 
   protected _executeAdd?: (
     operation: Operation<SupportedModel>,
@@ -31,7 +33,10 @@ export default abstract class ExecutorBase {
   static OPERATIONS_BATCH_PROCESSING_TIME = 5;
   static RETRY_COUNT = 5;
 
-  constructor(executorConfig: ExecutorConfig<SupportedModel>) {
+  constructor(
+    executorConfig: ExecutorConfig<SupportedModel>,
+    newRecordsState: NewRecordsState,
+  ) {
     setInterval(() => {
       Log.debug('OneSignal: checking for operations to process from cache');
       const cachedOperations = this.getOperationsFromCache();
@@ -48,6 +53,8 @@ export default abstract class ExecutorBase {
     this._executeAdd = executorConfig.add;
     this._executeUpdate = executorConfig.update;
     this._executeRemove = executorConfig.remove;
+
+    this._newRecordsState = newRecordsState;
   }
 
   abstract processDeltaQueue(): void;
@@ -124,7 +131,7 @@ export default abstract class ExecutorBase {
       if (operation) {
         OperationCache.enqueue(operation);
 
-        if (this.onlineStatus) {
+        if (this._canExecute(operation)) {
           this._processOperation(operation, ExecutorBase.RETRY_COUNT).catch(
             (err) => {
               Log.error(err);
@@ -187,4 +194,18 @@ export default abstract class ExecutorBase {
       this._processOperationQueue.call(this);
     }
   }
+
+  private _canExecute(operation: Operation<SupportedModel>): boolean {
+    if (!this.onlineStatus) {
+      return false;
+    }
+
+    if (operation.applyToRecordId) {
+      if (!this._newRecordsState.canAccess(operation.applyToRecordId)) {
+        return false;
+      }
+    }
+
+    return true;
+  }
 }

src/core/executors/ExecutorFactory.ts

@@ -1,3 +1,4 @@
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 import { Executor } from '../models/Executor';
 import { ExecutorConfig } from '../models/ExecutorConfig';
 import { ModelName, SupportedModel } from '../models/SupportedModels';
@@ -6,14 +7,17 @@ import { PropertiesExecutor } from './PropertiesExecutor';
 import { SubscriptionExecutor } from './SubscriptionExecutor';
 
 export class ExecutorFactory {
-  static build(executorConfig: ExecutorConfig<SupportedModel>): Executor {
+  static build(
+    executorConfig: ExecutorConfig<SupportedModel>,
+    newRecordsState: NewRecordsState,
+  ): Executor {
     switch (executorConfig.modelName) {
       case ModelName.Identity:
-        return new IdentityExecutor(executorConfig);
+        return new IdentityExecutor(executorConfig, newRecordsState);
       case ModelName.Properties:
-        return new PropertiesExecutor(executorConfig);
+        return new PropertiesExecutor(executorConfig, newRecordsState);
       case ModelName.Subscriptions:
-        return new SubscriptionExecutor(executorConfig);
+        return new SubscriptionExecutor(executorConfig, newRecordsState);
     }
   }
 }

src/core/executors/ExecutorStore.ts

@@ -1,3 +1,4 @@
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 import { ModelName } from '../models/SupportedModels';
 import OSExecutor from './ExecutorBase';
 import { EXECUTOR_CONFIG_MAP } from './ExecutorConfigMap';
@@ -10,10 +11,10 @@ type ExecutorStoreInterface = {
 export class ExecutorStore {
   store: ExecutorStoreInterface = {};
 
-  constructor() {
+  constructor(newRecordsState: NewRecordsState) {
     Object.values(ModelName).forEach((modelName) => {
       const config = EXECUTOR_CONFIG_MAP[modelName as ModelName];
-      this.store[modelName] = ExecutorFactory.build(config);
+      this.store[modelName] = ExecutorFactory.build(config, newRecordsState);
     });
   }
 

src/core/executors/IdentityExecutor.ts

@@ -1,3 +1,4 @@
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 import OperationCache from '../caching/OperationCache';
 import { CoreChangeType } from '../models/CoreChangeType';
 import { PropertyDelta } from '../models/CoreDeltas';
@@ -8,8 +9,11 @@ import { isPropertyDelta } from '../utils/typePredicates';
 import ExecutorBase from './ExecutorBase';
 
 export class IdentityExecutor extends ExecutorBase {
-  constructor(executorConfig: ExecutorConfig<SupportedModel>) {
-    super(executorConfig);
+  constructor(
+    executorConfig: ExecutorConfig<SupportedModel>,
+    newRecordsState: NewRecordsState,
+  ) {
+    super(executorConfig, newRecordsState);
   }
 
   processDeltaQueue(): void {

src/core/executors/PropertiesExecutor.ts

@@ -1,3 +1,4 @@
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 import ExecutorBase from './ExecutorBase';
 import { Operation } from '../operationRepo/Operation';
 import { CoreChangeType } from '../models/CoreChangeType';
@@ -6,8 +7,11 @@ import { ModelName, SupportedModel } from '../models/SupportedModels';
 import OperationCache from '../caching/OperationCache';
 
 export class PropertiesExecutor extends ExecutorBase {
-  constructor(executorConfig: ExecutorConfig<SupportedModel>) {
-    super(executorConfig);
+  constructor(
+    executorConfig: ExecutorConfig<SupportedModel>,
+    newRecordsState: NewRecordsState,
+  ) {
+    super(executorConfig, newRecordsState);
   }
 
   processDeltaQueue(): void {

src/core/executors/SubscriptionExecutor.ts

@@ -1,3 +1,4 @@
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 import OperationCache from '../caching/OperationCache';
 import { CoreChangeType } from '../models/CoreChangeType';
 import { CoreDelta } from '../models/CoreDeltas';
@@ -7,8 +8,11 @@ import { Operation } from '../operationRepo/Operation';
 import ExecutorBase from './ExecutorBase';
 
 export class SubscriptionExecutor extends ExecutorBase {
-  constructor(executorConfig: ExecutorConfig<SupportedModel>) {
-    super(executorConfig);
+  constructor(
+    executorConfig: ExecutorConfig<SupportedModel>,
+    newRecordsState: NewRecordsState,
+  ) {
+    super(executorConfig, newRecordsState);
   }
 
   processDeltaQueue(): void {

src/core/modelRepo/ModelRepo.ts

@@ -60,6 +60,7 @@ export class ModelRepo extends Subscribable<CoreDelta<SupportedModel>> {
     this.broadcast({
       model: payload,
       changeType: CoreChangeType.Add,
+      applyToRecordId: payload?.applyToRecordId,
     });
   }
 
@@ -77,6 +78,7 @@ export class ModelRepo extends Subscribable<CoreDelta<SupportedModel>> {
     this.broadcast({
       model: payload,
       changeType: CoreChangeType.Remove,
+      applyToRecordId: payload?.applyToRecordId,
     });
   }
 
@@ -103,6 +105,7 @@ export class ModelRepo extends Subscribable<CoreDelta<SupportedModel>> {
         property: payload.property,
         oldValue: payload.oldValue,
         newValue: payload.newValue,
+        applyToRecordId: payload.model?.applyToRecordId,
       };
       this.broadcast(delta);
     }

src/core/modelRepo/OSModel.ts

@@ -13,11 +13,11 @@ import { logMethodCall } from '../../shared/utils/utils';
 export class OSModel<Model> extends Subscribable<ModelStoreChange<Model>> {
   data: Model;
   modelId: string;
-
   onesignalId?: string;
   awaitOneSignalIdAvailable: Promise<string>;
   onesignalIdAvailableCallback?: (onesignalId: string) => void;
   externalId?: string;
+  applyToRecordId?: string;
 
   constructor(
     readonly modelName: ModelName,
@@ -55,6 +55,11 @@ export class OSModel<Model> extends Subscribable<ModelStoreChange<Model>> {
     this.externalId = externalId;
   }
 
+  public setApplyToRecordId(applyToRecordId: string): void {
+    logMethodCall('setapplyToRecordId', { applyToRecordId });
+    this.applyToRecordId = applyToRecordId;
+  }
+
   /**
    * We use this method to update the model data.
    * Results in a broadcasted update event.

src/core/models/CoreDeltas.ts

@@ -5,6 +5,7 @@ import { StringKeys } from './StringKeys';
 export type ModelDelta<Model> = {
   model: OSModel<Model>;
   changeType: CoreChangeType;
+  applyToRecordId?: string;
 };
 
 export interface PropertyDelta<Model> extends ModelDelta<Model> {

src/core/operationRepo/Operation.ts

@@ -11,6 +11,7 @@ export class Operation<Model> {
   timestamp: number;
   payload?: Partial<SupportedModel>;
   model?: OSModel<Model>;
+  applyToRecordId?: string;
   jwtTokenAvailable: Promise<void>;
   jwtToken?: string | null;
 
@@ -22,6 +23,7 @@ export class Operation<Model> {
     this.operationId = Math.random().toString(36).substring(2);
     this.payload = deltas ? this.getPayload(deltas) : undefined;
     this.model = deltas ? deltas[deltas.length - 1].model : undefined;
+    this.applyToRecordId = deltas?.[deltas.length - 1]?.applyToRecordId;
     this.timestamp = Date.now();
     // eslint-disable-next-line no-async-promise-executor
     this.jwtTokenAvailable = new Promise<void>(async (resolve) => {

src/core/operationRepo/OperationRepo.ts

@@ -3,15 +3,21 @@ import { ExecutorStore } from '../executors/ExecutorStore';
 import { CoreDelta } from '../models/CoreDeltas';
 import { SupportedModel } from '../models/SupportedModels';
 import { logMethodCall } from '../../shared/utils/utils';
+import { NewRecordsState } from '../../shared/models/NewRecordsState';
 
 export class OperationRepo {
   public executorStore: ExecutorStore;
+  public newRecordsState: NewRecordsState;
   private _unsubscribeFromModelRepo: () => void;
   private _deltaQueue: CoreDelta<SupportedModel>[] = [];
   static DELTAS_BATCH_PROCESSING_TIME = 1;
 
-  constructor(private modelRepo: ModelRepo) {
-    this.executorStore = new ExecutorStore();
+  constructor(
+    private modelRepo: ModelRepo,
+    newRecordsState: NewRecordsState,
+  ) {
+    this.newRecordsState = newRecordsState;
+    this.executorStore = new ExecutorStore(this.newRecordsState);
 
     this._unsubscribeFromModelRepo = this.modelRepo.subscribe(
       (delta: CoreDelta<SupportedModel>) => {

src/core/requestService/SubscriptionRequests.ts

@@ -25,13 +25,20 @@ export default class SubscriptionRequests {
     logMethodCall('SubscriptionRequests.addSubscription', operation);
 
     const appId = await MainHelper.getAppId();
-    const { subscription, aliasPair } = processSubscriptionOperation(operation);
+    const { subscription, aliasPair, subscriptionId } =
+      processSubscriptionOperation(operation);
 
     const response = await RequestService.createSubscription(
       { appId },
       aliasPair,
       { subscription },
     );
+
+    const status = response.status;
+    if (status >= 200 && status < 300) {
+      OneSignal.coreDirector.getNewRecordsState().add(subscriptionId);
+    }
+
     return SubscriptionRequests._processSubscriptionResponse(response);
   }