Implement asset model metadata propagation

lib/modules/asset/AssetService.ts

@@ -4,7 +4,14 @@ import { JSONObject, KDocument, KHit, SearchResult } from "kuzzle-sdk";
 
 import { MeasureContent } from "../measure/";
 import { AskDeviceUnlinkAsset } from "../device";
-import { EmbeddedMeasure, Metadata, lock, ask, flattenObject } from "../shared";
+import {
+  EmbeddedMeasure,
+  Metadata,
+  lock,
+  ask,
+  onAsk,
+  flattenObject,
+} from "../shared";
 import {
   DeviceManagerConfiguration,
   InternalCollection,
@@ -15,6 +22,7 @@ import { AskModelAssetGet } from "../model";
 import { AssetContent } from "./types/AssetContent";
 import { AssetSerializer } from "./model/AssetSerializer";
 import {
+  AskAssetRefreshModel,
   EventAssetUpdateAfter,
   EventAssetUpdateBefore,
 } from "./types/AssetEvents";
@@ -52,6 +60,43 @@ export class AssetService {
     this.context = plugin.context;
     this.config = plugin.config;
     this.assetHistoryService = assetHistoryService;
+
+    this.registerAskEvents();
+  }
+
+  registerAskEvents() {
+    onAsk<AskAssetRefreshModel>(
+      "ask:device-manager:asset:refresh-model",
+      async ({ assetModel, assetId, engineId }) => {
+        const asset = await this.get(engineId, assetId);
+
+        const modelMetadata = {};
+        for (const metadataName of Object.keys(
+          assetModel.asset.metadataMappings
+        )) {
+          modelMetadata[metadataName] = null;
+        }
+        for (const [metadataName, metadataValue] of Object.entries(
+          assetModel.asset.defaultMetadata
+        )) {
+          _.set(modelMetadata, metadataName, metadataValue);
+        }
+
+        const removedMetadata = _.difference(
+          Object.keys(asset._source.metadata),
+          Object.keys(modelMetadata)
+        );
+
+        const metadata = {
+          ...modelMetadata,
+          ..._.omit(asset._source.metadata, removedMetadata),
+        };
+
+        await this.replace(null, engineId, assetId, metadata, {
+          refresh: "wait_for",
+        });
+      }
+    );
   }
 
   /**
@@ -306,6 +351,68 @@ export class AssetService {
     return result;
   }
 
+  /**
+   * Replace an asset metadata
+   */
+  public async replace(
+    user: User,
+    engineId: string,
+    assetId: string,
+    metadata: Metadata,
+    { refresh }: { refresh: any }
+  ): Promise<KDocument<AssetContent>> {
+    return lock(`asset:${engineId}:${assetId}`, async () => {
+      const asset = await this.get(engineId, assetId);
+
+      const updatedPayload = await this.app.trigger<EventAssetUpdateBefore>(
+        "device-manager:asset:update:before",
+        { asset, metadata }
+      );
+
+      const updatedAsset = await this.impersonatedSdk(
+        user
+      ).document.replace<AssetContent>(
+        engineId,
+        InternalCollection.ASSETS,
+        assetId,
+        {
+          ...asset._source,
+          metadata: updatedPayload.metadata,
+        },
+        { refresh, source: true }
+      );
+
+      const removedMetadata = _.difference(
+        Object.keys(asset._source.metadata),
+        Object.keys(updatedPayload.metadata)
+      );
+
+      // @todo fix the metadata path for nested metadata
+      await this.assetHistoryService.add<AssetHistoryEventMetadata>(
+        engineId,
+        {
+          metadata: {
+            names: Object.keys(updatedPayload.metadata).concat(
+              removedMetadata.map((name) => `-${name}`)
+            ),
+          },
+          name: "metadata",
+        },
+        updatedAsset
+      );
+
+      await this.app.trigger<EventAssetUpdateAfter>(
+        "device-manager:asset:update:after",
+        {
+          asset: updatedAsset,
+          metadata: updatedPayload.metadata,
+        }
+      );
+
+      return updatedAsset;
+    });
+  }
+
   private async getEngine(engineId: string): Promise<JSONObject> {
     const engine = await this.sdk.document.get(
       this.config.adminIndex,

lib/modules/asset/types/AssetEvents.ts

@@ -1,4 +1,5 @@
 import { KDocument } from "kuzzle-sdk";
+import { AssetModelContent } from "lib/modules/model";
 
 import { Metadata } from "../../../modules/shared";
 
@@ -17,6 +18,18 @@ export type EventAssetUpdateAfter = {
   args: [{ asset: KDocument<AssetContent>; metadata: Metadata }];
 };
 
+export type AskAssetRefreshModel = {
+  name: "ask:device-manager:asset:refresh-model";
+
+  payload: {
+    assetModel: AssetModelContent;
+    assetId: string;
+    engineId: string;
+  };
+
+  result: void;
+};
+
 /**
  * @internal
  */

lib/modules/model/ModelService.ts

@@ -7,21 +7,27 @@ import {
 import { JSONObject, KDocument } from "kuzzle-sdk";
 
 import {
+  AskEngineList,
   AskEngineUpdateAll,
   DeviceManagerConfiguration,
   DeviceManagerPlugin,
   InternalCollection,
 } from "../plugin";
 import { ask, onAsk } from "../shared/utils/ask";
 
+import {
+  ApiAssetSearchRequest,
+  ApiAssetSearchResult,
+  AskAssetRefreshModel,
+} from "../asset";
+import { flattenObject } from "../shared/utils/flattenObject";
+import { ModelSerializer } from "./ModelSerializer";
 import {
   AssetModelContent,
   DeviceModelContent,
   MeasureModelContent,
 } from "./types/ModelContent";
-import { ModelSerializer } from "./ModelSerializer";
 import { AskModelAssetGet, AskModelDeviceGet } from "./types/ModelEvents";
-import { flattenObject } from "../shared/utils/flattenObject";
 
 export class ModelService {
   private config: DeviceManagerConfiguration;
@@ -90,7 +96,32 @@ export class ModelService {
     );
     await ask<AskEngineUpdateAll>("ask:device-manager:engine:updateAll");
 
-    // @todo update assets in every engine to add the new metadata with null value or default metadata
+    const engines = await ask<AskEngineList>("ask:device-manager:engine:list", {
+      group: engineGroup,
+    });
+
+    for (const engine of engines) {
+      const assets = await this.sdk.query<
+        ApiAssetSearchRequest,
+        ApiAssetSearchResult
+      >({
+        action: "search",
+        body: { query: { equals: { model } } },
+        controller: "device-manager/assets",
+        engineId: engine.index,
+        lang: "koncorde",
+      });
+
+      await Promise.all(
+        assets.result.hits.map((asset) =>
+          ask<AskAssetRefreshModel>("ask:device-manager:asset:refresh-model", {
+            assetId: asset._id,
+            assetModel: modelContent,
+            engineId: engine.index,
+          })
+        )
+      );
+    }
 
     return assetModel;
   }

lib/modules/plugin/DeviceManagerEngine.ts

@@ -2,6 +2,7 @@ import _ from "lodash";
 import { Backend, InternalError, Plugin } from "kuzzle";
 import { JSONObject } from "kuzzle-sdk";
 import { AbstractEngine, ConfigManager } from "kuzzle-plugin-commons";
+import { EngineContent } from "kuzzle-plugin-commons/lib/engine/EngineContent";
 
 import { assetsMappings, assetsHistoryMappings } from "../asset";
 import {
@@ -23,6 +24,16 @@ const digitalTwinMappings = {
   device: devicesMappings,
 } as const;
 
+export type AskEngineList = {
+  name: "ask:device-manager:engine:list";
+
+  payload: {
+    group: string | null;
+  };
+
+  result: EngineContent[];
+};
+
 export type AskEngineUpdateAll = {
   name: "ask:device-manager:engine:updateAll";
 
@@ -53,6 +64,10 @@ export class DeviceManagerEngine extends AbstractEngine<DeviceManagerPlugin> {
 
     this.context = plugin.context;
 
+    onAsk<AskEngineList>("ask:device-manager:engine:list", async ({ group }) =>
+      this.list(group)
+    );
+
     onAsk<AskEngineUpdateAll>(
       "ask:device-manager:engine:updateAll",
       async () => {