Merge pull request IDEMSInternational#2177 from IDEMSInternational/fi…

…x/dynamic-data-concurrent-queries

fix: dynamic data service multiple queries

packages/shared/src/utils/async-utils.ts

@@ -0,0 +1,8 @@
+/** helper function used for dev to wait a fixed amount of time */
+export function _wait(ms: number) {
+  return new Promise<void>((resolve) => {
+    setTimeout(() => {
+      resolve();
+    }, ms);
+  });
+}

packages/shared/src/utils/cli-utils.ts

@@ -39,11 +39,3 @@ export function pad(str: string | number, chars: number) {
   const padChars = Math.max(chars - str.length + 1, 0);
   return str + new Array(padChars).join(" ");
 }
-/** helper function used for dev to wait a fixed amount of time */
-export function _wait(ms: number) {
-  return new Promise<void>((resolve) => {
-    setTimeout(() => {
-      resolve();
-    }, ms);
-  });
-}

packages/shared/src/utils/index.ts

@@ -1,3 +1,4 @@
+export * from "./async-utils";
 export * from "./cli-utils";
 export * from "./delimiters";
 export * from "./file-utils";

packages/shared/src/utils/logging/console-logger.ts

@@ -2,7 +2,7 @@ import boxen from "boxen";
 import chalk from "chalk";
 import { Command } from "commander";
 
-import { _wait } from "../cli-utils";
+import { _wait } from "../async-utils";
 
 /**
  * HACK - export error within a Logger const to allow easier mocking in tests

packages/shared/src/utils/logging/file-logger.ts

@@ -1,7 +1,7 @@
 import winston from "winston";
 import path from "path";
 import { emptyDirSync, ensureDirSync, truncateSync } from "fs-extra";
-import { _wait } from "../cli-utils";
+import { _wait } from "../async-utils";
 import { Writable } from "stream";
 import { existsSync } from "fs";
 import { SCRIPTS_LOGS_DIR } from "../../paths";

src/app/shared/services/data/app-data.service.spec.ts

@@ -3,7 +3,7 @@ import { TestBed } from "@angular/core/testing";
 import { HttpClientTestingModule } from "@angular/common/http/testing";
 
 import { AppDataVariableService } from "./app-data-variable.service";
-import { AppDataService } from "./app-data.service";
+import { AppDataService, IAppDataCache } from "./app-data.service";
 import { FlowTypes } from "../../model";
 import { MockAppDataVariableService } from "./app-data-variable.service.spec";
 import { ErrorHandlerService } from "../error-handler/error-handler.service";
@@ -12,42 +12,57 @@ import { DbService } from "../db/db.service";
 import { MockDbService } from "../db/db.service.spec";
 import { Injectable } from "@angular/core";
 import { ISheetContents } from "src/app/data";
+import { _wait } from "packages/shared/src/utils/async-utils";
 
-const DATA_MOCK: { [flow_name: string]: FlowTypes.FlowTypeWithData } = {
-  flow_a: {
-    flow_name: "flow_a",
-    flow_type: "data_list",
-    rows: [
-      { id: "a_id1", number: 1, string: "a_hello", boolean: true },
-      { id: "a_id2", number: 2, string: "a_goodbye", boolean: false },
-    ],
-  },
-  flow_b: {
-    flow_name: "flow_b",
-    flow_type: "data_list",
-    rows: [
-      { id: "b_id1", number: 1, string: "b_hello", boolean: true },
-      { id: "b_id2", number: 2, string: "b_goodbye", boolean: false },
-    ],
-    _overrides: { flow_c: "true" },
-  },
-  flow_c: {
-    flow_name: "flow_c",
-    flow_type: "data_list",
-    rows: [
-      { id: "c_id1", number: 1, string: "c_hello", boolean: true },
-      { id: "c_id2", number: 2, string: "c_goodbye", boolean: false },
-    ],
-    _overrides: { flow_b: "true" },
-  },
-  flow_d: {
-    flow_name: "flow_d",
-    flow_type: "data_list",
-    rows: [
-      { id: "d_id1", number: 1, string: "d_hello", boolean: true },
-      { id: "d_id2", number: 2, string: "d_goodbye", boolean: false },
-    ],
-    _overrides: { flow_a: "true" },
+/** Base mock data for use with any services calling mock app-data handlers */
+const DATA_CACHE_CLEAN: IAppDataCache = {
+  asset_pack: {},
+  data_list: {},
+  data_pipe: {},
+  generator: {},
+  global: {},
+  template: {},
+  tour: {},
+};
+
+/** Mock data used specifically for the app-data service spec */
+const SPEC_MOCK_DATA: Partial<IAppDataCache> = {
+  data_list: {
+    flow_a: {
+      flow_name: "flow_a",
+      flow_type: "data_list",
+      rows: [
+        { id: "a_id1", number: 1, string: "a_hello", boolean: true },
+        { id: "a_id2", number: 2, string: "a_goodbye", boolean: false },
+      ],
+    },
+    flow_b: {
+      flow_name: "flow_b",
+      flow_type: "data_list",
+      rows: [
+        { id: "b_id1", number: 1, string: "b_hello", boolean: true },
+        { id: "b_id2", number: 2, string: "b_goodbye", boolean: false },
+      ],
+      _overrides: { flow_c: "true" },
+    },
+    flow_c: {
+      flow_name: "flow_c",
+      flow_type: "data_list",
+      rows: [
+        { id: "c_id1", number: 1, string: "c_hello", boolean: true },
+        { id: "c_id2", number: 2, string: "c_goodbye", boolean: false },
+      ],
+      _overrides: { flow_b: "true" },
+    },
+    flow_d: {
+      flow_name: "flow_d",
+      flow_type: "data_list",
+      rows: [
+        { id: "d_id1", number: 1, string: "d_hello", boolean: true },
+        { id: "d_id2", number: 2, string: "d_goodbye", boolean: false },
+      ],
+      _overrides: { flow_a: "true" },
+    },
   },
 };
 
@@ -83,12 +98,18 @@ const CONTENTS_MOCK: ISheetContents = {
 
 /** Mock calls for sheets from the appData service to return test data */
 export class MockAppDataService implements Partial<AppDataService> {
+  public appDataCache: IAppDataCache;
+
+  // allow additional specs implementing service to provide their own data if required
+  constructor(mockData: Partial<IAppDataCache> = {}) {
+    this.appDataCache = { ...DATA_CACHE_CLEAN, ...mockData };
+  }
   public async getSheet<T extends FlowTypes.FlowTypeWithData>(
     flow_type: FlowTypes.FlowType,
     flow_name: string
   ): Promise<T> {
-    const rows = DATA_MOCK[flow_name].rows || [];
-    return { flow_name, flow_type, rows } as any;
+    await _wait(50);
+    return this.appDataCache[flow_type]?.[flow_name] as T;
   }
 }
 
@@ -97,15 +118,7 @@ export class MockAppDataService implements Partial<AppDataService> {
 class AppDataServiceExtended extends AppDataService {
   protected sheetContents = CONTENTS_MOCK;
   protected translationContents = {};
-  public appDataCache = {
-    asset_pack: {},
-    data_list: { ...DATA_MOCK },
-    data_pipe: {},
-    generator: {},
-    global: {},
-    template: {},
-    tour: {},
-  };
+  public appDataCache = { ...DATA_CACHE_CLEAN, ...SPEC_MOCK_DATA };
 }
 
 /********************************************************************************

src/app/shared/services/data/app-data.service.ts

@@ -193,6 +193,6 @@ export class AppDataService extends SyncServiceBase {
   }
 }
 
-type IAppDataCache = {
+export type IAppDataCache = {
   [flow_type in FlowTypes.FlowType]: { [flow_name: string]: FlowTypes.FlowTypeWithData };
 };

src/app/shared/services/dynamic-data/dynamic-data.service.spec.ts

@@ -1,19 +1,20 @@
 import { TestBed } from "@angular/core/testing";
-
 import { HttpClientTestingModule } from "@angular/common/http/testing";
+import { firstValueFrom } from "rxjs";
 
 import { DynamicDataService } from "./dynamic-data.service";
-import { firstValueFrom } from "rxjs";
 import { AppDataService } from "../data/app-data.service";
 import { MockAppDataService } from "../data/app-data.service.spec";
 
-const DATA_MOCK = {
-  test_flow: [
-    { id: "id1", number: 1, string: "hello", boolean: true },
-    { id: "id2", number: 2, string: "goodbye", boolean: false },
-  ],
-};
+const TEST_DATA_ROWS = [
+  { id: "id1", number: 1, string: "hello", boolean: true },
+  { id: "id2", number: 2, string: "goodbye", boolean: false },
+];
 
+/**
+ * Call standalone tests via:
+ * yarn ng test --include src/app/shared/services/dynamic-data/dynamic-data.service.spec.ts
+ */
 describe("DynamicDataService", () => {
   let service: DynamicDataService;
 
@@ -22,7 +23,14 @@ describe("DynamicDataService", () => {
       imports: [HttpClientTestingModule],
       providers: [
         DynamicDataService,
-        { provide: AppDataService, useValue: new MockAppDataService() },
+        {
+          provide: AppDataService,
+          useValue: new MockAppDataService({
+            data_list: {
+              test_flow: { flow_name: "test_flow", flow_type: "data_list", rows: TEST_DATA_ROWS },
+            },
+          }),
+        },
       ],
     });
 
@@ -33,7 +41,7 @@ describe("DynamicDataService", () => {
     TestBed.inject(AppDataService);
 
     await service.ready();
-    service.resetFlow("data_list", "test_flow");
+    service.resetFlow("data_list", "test_flow", false);
   });
 
   it("populates initial flows from json", async () => {
@@ -46,7 +54,7 @@ describe("DynamicDataService", () => {
     await service.update("data_list", "test_flow", "id1", { number: 1.1 });
     const obs = await service.query$<any>("data_list", "test_flow");
     const data = await firstValueFrom(obs);
-    expect(data[0]).toEqual({ ...DATA_MOCK.test_flow[0], number: 1.1 });
+    expect(data[0]).toEqual({ ...TEST_DATA_ROWS[0], number: 1.1 });
   });
 
   it("populates cached data on load", async () => {
@@ -66,6 +74,15 @@ describe("DynamicDataService", () => {
     expect(queryResult.length).toEqual(2);
   });
 
+  it("Supports parallel requests without recreating collections", async () => {
+    const queries = new Array(20).fill(0).map(async () => {
+      const obs = await service.query$("data_list", "test_flow");
+      return firstValueFrom(obs);
+    });
+    const res = await Promise.all(queries);
+    expect(res.length).toEqual(20);
+  });
+
   // QA
   it("prevents query of non-existent data lists", async () => {
     let errMsg: string;
@@ -75,13 +92,6 @@ describe("DynamicDataService", () => {
     expect(errMsg).toEqual("No data exists for collection [fakeData], cannot initialise");
   });
 
-  it("prevents updates to non-existent rows", async () => {
-    let errMsg: string;
-    await service.update("data_list", "test_flow", "missing_row", { number: 1 }).catch((err) => {
-      errMsg = err.message;
-    });
-    expect(errMsg).toBe("cannot update row that does not exist: [test_flow]:[missing_row]");
-  });
   it("ignores cached data where initial data no longer exists", async () => {
     // TODO - add methods that ignore rows from cached data if row id deleted from source data_list
   });

src/app/shared/services/dynamic-data/dynamic-data.service.ts

@@ -1,6 +1,6 @@
 import { Injectable } from "@angular/core";
 import { addRxPlugin, MangoQuery, RxDocument } from "rxdb";
-import { firstValueFrom, map } from "rxjs";
+import { firstValueFrom, lastValueFrom, map, AsyncSubject } from "rxjs";
 
 import { FlowTypes } from "data-models";
 import { environment } from "src/environments/environment";
@@ -41,6 +41,9 @@ export class DynamicDataService extends AsyncServiceBase {
    */
   private writeCache: PersistedMemoryAdapter;
 
+  /** Hashmap to track pending collection creation and avoid duplicate requests */
+  private collectionCreators: Record<string, AsyncSubject<string>> = {};
+
   constructor(
     private appDataService: AppDataService,
     private templateActionRegistry: TemplateActionRegistry
@@ -126,30 +129,50 @@ export class DynamicDataService extends AsyncServiceBase {
   }
 
   /** Remove user writes on a flow to return it to its original state */
-  public async resetFlow(flow_type: FlowTypes.FlowType, flow_name: string) {
+  public async resetFlow(flow_type: FlowTypes.FlowType, flow_name: string, throwOnError = true) {
     await this.writeCache.delete(flow_type, flow_name);
     const collectionName = this.normaliseCollectionName(flow_type, flow_name);
     if (this.db.getCollection(collectionName)) {
       await this.db.removeCollection(collectionName);
       await this.ensureCollection(flow_type, flow_name);
     } else {
-      throw new Error(`Collection [${collectionName}] not found, cannot remove`);
+      if (throwOnError) {
+        throw new Error(`Collection [${collectionName}] not found, cannot remove`);
+      }
     }
   }
 
   /** Ensure a collection exists, creating if not and populating with corresponding list data */
   private async ensureCollection(flow_type: FlowTypes.FlowType, flow_name: string) {
     const collectionName = this.normaliseCollectionName(flow_type, flow_name);
     if (!this.db.getCollection(collectionName)) {
+      await this.createCollection(flow_type, flow_name);
+    }
+    return { collectionName };
+  }
+
+  private async createCollection(flow_type: FlowTypes.FlowType, flow_name: string) {
+    const collectionName = this.normaliseCollectionName(flow_type, flow_name);
+    // avoid duplicate creation requests by tracking create requests
+    if (this.collectionCreators[collectionName]) {
+      await lastValueFrom(this.collectionCreators[collectionName]);
+      return;
+    }
+    // create collection and insert initial data. Use AsyncSubject to notify only when complete
+    else {
+      this.collectionCreators[collectionName] = new AsyncSubject();
       const initialData = await this.getInitialData(flow_type, flow_name);
       if (initialData.length === 0) {
         throw new Error(`No data exists for collection [${flow_name}], cannot initialise`);
       }
       const schema = this.inferSchema(initialData[0]);
       await this.db.createCollection(collectionName, schema);
       await this.db.bulkInsert(collectionName, initialData);
+      // notify any observers that collection has been created
+      this.collectionCreators[collectionName].next(collectionName);
+      this.collectionCreators[collectionName].complete();
+      delete this.collectionCreators[collectionName];
     }
-    return { collectionName };
   }
 
   /** Retrive json sheet data and merge with any user writes */