Merge pull request #317 from kuzzleio/feat/assetsMigrateTenant

feat/assets migrate tenant

doc/2/controllers/assets/migrate-tenant/index.md

@@ -0,0 +1,59 @@
+---
+code: true
+type: page
+title: migrateTenant
+description: Migrates a list of assets and their attached devices to another tenant
+---
+
+# update
+
+This action allow to migrates a list of assets and their attached devices to another tenant.
+
+
+## Query Syntax
+
+### HTTP
+
+```http
+URL: http://kuzzle:7512/_/device-manager/:engineId/assets/_migrateTenant
+Method: POST
+```
+
+### Other protocols
+
+```js
+{
+  "controller": "device-manager/assets",
+  "action": "migrateTenant",
+  "engineId": "<engineId>",
+  "body": {
+    "assetsList": ["<assetId>"],
+    "newEngineId": "<newEngineId>"
+  }
+}
+```
+
+---
+
+## Arguments
+
+- `engineId`: Engine ID
+
+## Body properties
+
+- `assetsList`: An array containing a list of asset ids to migrate
+- `newEngineId`: The id of the engine you want to migrate the assets to
+
+---
+
+## Response
+
+```js
+{
+  "status": 200,
+  "error": null,
+  "controller": "device-manager/assets",
+  "action": "migrateTenant",
+  "requestId": "<unique request identifier>",
+}
+```

lib/modules/asset/AssetService.ts

@@ -10,7 +10,12 @@ import {
 } from "kuzzle-sdk";
 import _ from "lodash";
 
-import { AskDeviceUnlinkAsset } from "../device";
+import {
+  AskDeviceAttachEngine,
+  AskDeviceDetachEngine,
+  AskDeviceLinkAsset,
+  AskDeviceUnlinkAsset,
+} from "../device";
 import { AskModelAssetGet, AssetModelContent } from "../model";
 import {
   AskEngineList,
@@ -39,6 +44,7 @@ import {
   AssetHistoryContent,
   AssetHistoryEventMetadata,
 } from "./types/AssetHistoryContent";
+import { RecoveryQueue } from "../shared/utils/recoveryQueue";
 
 export class AssetService {
   private context: PluginContext;
@@ -268,6 +274,246 @@ export class AssetService {
     return result;
   }
 
+  public async migrateTenant(
+    user: User,
+    assetsList: string[],
+    engineId: string,
+    newEngineId: string
+  ): Promise<void> {
+    return lock(`engine:${engineId}:${newEngineId}`, async () => {
+      const recovery = new RecoveryQueue();
+
+      if (!user.profileIds.includes("admin")) {
+        throw new BadRequestError(
+          `User ${user._id} is not authorized to migrate assets`
+        );
+      }
+
+      try {
+        // check if tenant destination of the the same group
+        const engine = await this.getEngine(engineId);
+        const newEngine = await this.getEngine(newEngineId);
+
+        if (engine.group !== newEngine.group) {
+          throw new BadRequestError(
+            `Engine ${newEngineId} is not in the same group as ${engineId}`
+          );
+        }
+
+        if (assetsList.length === 0) {
+          throw new BadRequestError("No assets to migrate");
+        }
+
+        const assets = await this.sdk.document.mGet<AssetContent>(
+          engineId,
+          InternalCollection.ASSETS,
+          assetsList
+        );
+
+        // check if the assets exists in the other engine
+        const existingAssets = await this.sdk.document.mGet<AssetContent>(
+          newEngineId,
+          InternalCollection.ASSETS,
+          assetsList
+        );
+
+        if (existingAssets.successes.length > 0) {
+          throw new BadRequestError(
+            `Assets ${existingAssets.successes
+              .map((asset) => asset._id)
+              .join(", ")} already exists in engine ${newEngineId}`
+          );
+        }
+        const assetsToMigrate = assets.successes.map((asset) => ({
+          _id: asset._id,
+          body: asset._source,
+        }));
+
+        const devices = await this.sdk.document.search<AssetContent>(
+          engineId,
+          InternalCollection.DEVICES,
+          {
+            query: {
+              bool: {
+                filter: {
+                  terms: {
+                    assetId: assetsList,
+                  },
+                },
+              },
+            },
+          }
+        );
+
+        // Map linked devices for assets.
+        const assetLinkedDevices = assets.successes
+          .filter((asset) => asset._source.linkedDevices.length > 0)
+          .map((asset) => ({
+            assetId: asset._id,
+            linkedDevices: asset._source.linkedDevices,
+          }));
+
+        // Extra recovery step to relink back assets to their devices in case of rollback
+        recovery.addRecovery(async () => {
+          // Link the devices to the new assets
+          const promises: Promise<void>[] = [];
+
+          for (const asset of assetLinkedDevices) {
+            const assetId = asset.assetId;
+            for (const device of asset.linkedDevices) {
+              const deviceId = device._id;
+              const measureNames = device.measureNames;
+              promises.push(
+                ask<AskDeviceLinkAsset>(
+                  "ask:device-manager:device:link-asset",
+                  {
+                    assetId,
+                    deviceId,
+                    engineId,
+                    measureNames: measureNames,
+                    user,
+                  }
+                )
+              );
+            }
+          }
+          await Promise.all(promises);
+        });
+
+        // detach from current tenant
+        await Promise.all(
+          devices.hits.map((device) => {
+            return ask<AskDeviceDetachEngine>(
+              "ask:device-manager:device:detach-engine",
+              { deviceId: device._id, user }
+            );
+          })
+        );
+
+        // Attach to new tenant
+        await Promise.all(
+          devices.hits.map((device) => {
+            return ask<AskDeviceAttachEngine>(
+              "ask:device-manager:device:attach-engine",
+              { deviceId: device._id, engineId: newEngineId, user }
+            );
+          })
+        );
+
+        // recovery function to reattach devices to the old tenant
+        recovery.addRecovery(async () => {
+          await Promise.all(
+            devices.hits.map((device) => {
+              return ask<AskDeviceDetachEngine>(
+                "ask:device-manager:device:detach-engine",
+                { deviceId: device._id, user }
+              );
+            })
+          );
+
+          await Promise.all(
+            devices.hits.map((device) => {
+              return ask<AskDeviceAttachEngine>(
+                "ask:device-manager:device:attach-engine",
+                { deviceId: device._id, engineId, user }
+              );
+            })
+          );
+        });
+
+        // Create the assets in the new tenant
+        await this.sdk.document.mCreate(
+          newEngineId,
+          InternalCollection.ASSETS,
+          assetsToMigrate
+        );
+
+        recovery.addRecovery(async () => {
+          await this.sdk.document.mDelete(
+            newEngineId,
+            InternalCollection.ASSETS,
+            assetsList
+          );
+        });
+
+        // Delete the assets in the old tenant
+        await this.sdk.document.mDelete(
+          engineId,
+          InternalCollection.ASSETS,
+          assetsList
+        );
+
+        recovery.addRecovery(async () => {
+          await this.sdk.document.mCreate(
+            engineId,
+            InternalCollection.ASSETS,
+            assetsToMigrate
+          );
+        });
+
+        // Link the devices to the new assets
+        const promises: Promise<void>[] = [];
+
+        for (const asset of assetLinkedDevices) {
+          const assetId = asset.assetId;
+          for (const device of asset.linkedDevices) {
+            const deviceId = device._id;
+            const measureNames = device.measureNames;
+            promises.push(
+              ask<AskDeviceLinkAsset>("ask:device-manager:device:link-asset", {
+                assetId,
+                deviceId,
+                engineId: newEngineId,
+                measureNames: measureNames,
+                user,
+              })
+            );
+          }
+        }
+
+        await Promise.all(promises);
+
+        recovery.addRecovery(async () => {
+          const promiseRecoveries: Promise<void>[] = [];
+
+          for (const asset of assetLinkedDevices) {
+            for (const device of asset.linkedDevices) {
+              const deviceId = device._id;
+              promiseRecoveries.push(
+                ask<AskDeviceUnlinkAsset>(
+                  "ask:device-manager:device:unlink-asset",
+                  {
+                    deviceId,
+                    user,
+                  }
+                )
+              );
+            }
+          }
+
+          await Promise.all(promiseRecoveries);
+        });
+
+        // clear the groups
+        await this.sdk.document.mUpdate<AssetContent>(
+          newEngineId,
+          InternalCollection.ASSETS,
+          assetsList.map((assetId) => ({
+            _id: assetId,
+            body: {
+              groups: [],
+            },
+          }))
+        );
+      } catch (error) {
+        await recovery.rollback();
+        throw new BadRequestError(
+          `An error occured while migrating assets: ${error}`
+        );
+      }
+    });
+  }
+
   /**
    * Replace an asset metadata
    */

lib/modules/asset/AssetsController.ts

@@ -96,6 +96,15 @@ export class AssetsController {
             },
           ],
         },
+        migrateTenant: {
+          handler: this.migrateTenant.bind(this),
+          http: [
+            {
+              path: "device-manager/:engineId/assets/_migrateTenant",
+              verb: "post",
+            },
+          ],
+        },
       },
     };
     /* eslint-enable sort-keys */
@@ -341,4 +350,16 @@ export class AssetsController {
 
     return { link };
   }
+
+  async migrateTenant(request: KuzzleRequest) {
+    const assetsList = request.getBodyArray("assetsList");
+    const engineId = request.getString("engineId");
+    const newEngineId = request.getBodyString("newEngineId");
+    await this.assetService.migrateTenant(
+      request.getUser(),
+      assetsList,
+      engineId,
+      newEngineId
+    );
+  }
 }

lib/modules/asset/types/AssetApi.ts

@@ -140,3 +140,14 @@ export interface ApiAssetExportRequest extends AssetsControllerRequest {
 export type ApiAssetExportResult = {
   link: string;
 };
+
+export interface ApiAssetMigrateTenantRequest extends AssetsControllerRequest {
+  action: "migrateTenant";
+  engineId: string;
+  body: {
+    assetsList: string[];
+    newEngineId: string;
+  };
+}
+
+export type ApiAssetMigrateTenantResult = void;