FEAT: update input

package-lock.json

@@ -17,7 +17,7 @@
         "@aws-sdk/signature-v4": "^3.370.0",
         "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
         "@dadosfera/protospack": "2.5.3",
-        "@dadosfera/protospack-v2": "3.38.0-beta.31",
+        "@dadosfera/protospack-v2": "3.40.0-beta.2",
         "@grpc/grpc-js": "^1.9.3",
         "@grpc/proto-loader": "^0.7.9",
         "@nestjs/cli": "^9.5.0",
@@ -1744,9 +1744,9 @@
       }
     },
     "node_modules/@dadosfera/protospack-v2": {
-      "version": "3.38.0-beta.31",
-      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.31.tgz",
-      "integrity": "sha512-IxnVEaVvQNumN/PQsnF+9CsFvmn6/0YyfXauoPQ1bccvqU8Lo+hKZv4qTBrR+urd05OWyg+dJBv/QtW45b04XQ==",
+      "version": "3.40.0-beta.2",
+      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.2.tgz",
+      "integrity": "sha512-WMoL9OhKJ05qDojd7cPcP7x45bHHnnQttfVoTWrc+lOkcVHygu0ob52ArfD8RRmpAXezBh9vDSbxz45n/lxbKg==",
       "license": "ISC",
       "dependencies": {
         "@grpc/grpc-js": "^1.9.3",

package.json

@@ -35,7 +35,7 @@
     "@aws-sdk/signature-v4": "^3.370.0",
     "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
     "@dadosfera/protospack": "2.5.3",
-    "@dadosfera/protospack-v2": "3.38.0-beta.31",
+    "@dadosfera/protospack-v2": "3.40.0-beta.2",
     "@grpc/grpc-js": "^1.9.3",
     "@grpc/proto-loader": "^0.7.9",
     "@nestjs/cli": "^9.5.0",

src/modules/inputs/inputs.service.ts

@@ -18,6 +18,7 @@ import {
   InputCreateS3Request,
   InputNewCreateRequest,
   InputUpdateResponse,
+  RollbackInputRequest,
   TestConnectionRequest,
 } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
 import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
@@ -209,7 +210,8 @@ export class InputsService {
     // this.validateCron({ ...data, info });
     try {
       const {
-        nimbusUpdates,
+        tablesUpdate,
+        dataAssetUpdate,
         input
       } = await this.OLD_inputClient.update({
         id,
@@ -222,13 +224,29 @@ export class InputsService {
       );
       return {
         input: updateInputResponse,
-        nimbusUpdates
+        tablesUpdate,
+        dataAssetUpdate
       };
     } catch (err) {
       throw new HttpException(err.message, HttpStatus.NOT_FOUND);
     }
   }
 
+  async rollbackUpdate(
+    data: RollbackInputRequest
+  ) {
+    this.logger.info('PipelinesClientService - rollbackUpdate');
+    this.logger.info('Rolling back input update with data: ' + JSON.stringify(data));
+    const updatePipelineResponse = await lastValueFrom(
+      this.inputWriteService.RollbackInputUpdate(
+        data
+      ),
+    );
+    this.logger.info('Done');
+
+    return updatePipelineResponse;
+  }
+
   async remove(idRequest: IIdRequest) {
     return lastValueFrom(this.inputWriteService.InputRemove(idRequest));
   }

src/modules/pipelinesV2/pipelines.service.ts

@@ -28,11 +28,13 @@ import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath';
 import ErrorCodes from 'src/utils/errorCodes';
 import ErrorBuilder from 'src/utils/ErrorBuilder';
 import { PlatformApiService } from '../platform-api/platform-api.service';
-import { Info, NewTable } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
-import { NimbusUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
-import axios from 'axios';
+import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
+import { TableUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
+import { AxiosError } from 'axios';
 import { NimbusService } from 'src/services/nimbus/nimbus.service';
 
+type RollbackPromise = () => Promise<any>;
+
 export class PipelinesService implements OnModuleInit {
   logger: DadosferaLogger;
   pipelineWriteService: WriteService.PipelineV2WriteService;
@@ -352,33 +354,116 @@ export class PipelinesService implements OnModuleInit {
   async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) {
     this.logger.info('InputClientService - Update');
 
-    this.logger.info('Update Dynamo Reference');
+    const {
+      input: oldInput
+    } = await this.inputsService.findOne({
+      id: inputId,
+      info: info
+    });
+
+    this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
     const pipelineIdFormat = pipelineId.split('-').join('_');
-    console.log("updatePipelineInput", info, updateInputDTO)
+    const rollback: RollbackPromise[] = [];
+
     const updateInputResponse = await this.inputsService.update(
       inputId,
       updateInputDTO,
       info
-    )
-
-    this.logger.info(JSON.stringify(updateInputResponse))
-    await this.updateNimbus(info.customer, updateInputResponse.nimbusUpdates);
-
-    await this.updatePlatformJobs(
-      pipelineIdFormat,
-      updateInputDTO,
-      user
     );
 
+    const inputRollback = () => {
+      this.logger.info("exec rollback to input: " + JSON.stringify(oldInput));
+      return this.inputsService.rollbackUpdate(
+        {
+          id: inputId,
+          dataAssetUpdate: updateInputResponse.dataAssetUpdate,
+          tables: oldInput.tables,
+          info
+        }
+      ) as Promise<any>;
+    }
+
+    rollback.push(inputRollback);
+
+    this.logger.info("Input Update Response: " + JSON.stringify(updateInputResponse))
+
+    const nimbusUpdates = updateInputResponse?.tablesUpdate || [];
+
+    nimbusUpdates.forEach(update => {
+      const nimbusRollback = () => {
+        return this.nimbusService.renameTable(
+          info.customer,
+          update.database,
+          {
+            table_name: update.table_name,
+            table_schema: update.table_schema
+          },
+          {
+            table_name: update.old_table_name,
+            table_schema: update.old_table_schema
+          }
+        );
+      }
+      rollback.push(nimbusRollback);
+    });
+
+    try {
+      await this.updateNimbus(info.customer, nimbusUpdates);
+    } catch (error) {
+      this.logger.error(error);
+      if (error instanceof AxiosError) {
+        this.logger.error(JSON.stringify(error.response.data));
+      }
+      await this.executeRenameRollback(rollback);
+
+      throw new Error("Error Nimbus updating tables");
+    }
+
+    try {
+      await this.updatePlatformJobs(
+        pipelineIdFormat,
+        updateInputDTO,
+        user
+      );
+    } catch (error) {
+      this.logger.error(error);
+      await this.executeRenameRollback(rollback)
+      throw new Error("Error Platform API updating jobs");
+    }
+
     return updateInputResponse;
   }
 
-  private async updateNimbus(customer: string, changes: NimbusUpdate[]) {
+  private async executeRenameRollback(request: RollbackPromise[]) {
+    this.logger.info('rollback steps: ' + request.length)
+    const result = await Promise.allSettled(request.map(func => func()));
+    result.forEach(promise => {
+      this.logger.info("Promise finish with status: " + promise.status)
+
+      if (promise.status === "rejected") {
+        this.logger.error("reject with: " + JSON.stringify(promise.reason || {}))
+      }
+
+      if (promise.status === "fulfilled") {
+        this.logger.info("success with: " + JSON.stringify(promise.value || {}))
+      }
+    });
+
+  }
+
+  private async updateNimbus(customer: string, changes: TableUpdate[]) {
+    // throw new Error("teste error nimbus");
     this.logger.info('Nimbus Changes: ' + JSON.stringify(changes));
     if(!changes || changes.length === 0) return;
 
     const requests = changes.map(change => {
-      return this.nimbusService.renameTable(customer, change.database, change.old, change.new);
+      return this.nimbusService.renameTable(customer, change.database, {
+        table_name: change.old_table_name,
+        table_schema: change.old_table_schema
+      }, {
+        table_name: change.table_name,
+        table_schema: change.table_schema
+      });
     })
 
     const values = await Promise.allSettled(requests);
@@ -386,6 +471,20 @@ export class PipelinesService implements OnModuleInit {
     const success = values.map(request => request.status === "fulfilled")
 
     this.logger.info("Updates with succes: " + success.length);
+
+    values.forEach(promise => {
+      this.logger.info("Promise finish with status: " + promise.status)
+
+      if (promise.status === "rejected") {
+        this.logger.error("reject with: " + JSON.stringify(promise.reason || {}));
+        throw new Error(promise.reason );
+      }
+
+      if (promise.status === "fulfilled") {
+        this.logger.info("success with: " + JSON.stringify(promise.value || {}));
+      }
+    });
+
   }
 
   async updatePlatformJobs(pipelineId: string, updateInputDTO: UpdatePlatformInputRequest, user: RequestUser) {
@@ -396,6 +495,10 @@ export class PipelinesService implements OnModuleInit {
         job_id: `${pipelineId}_${index}`,
       }
 
+      if (table.type !== "incremental_with_qualify") {
+        delete table.destinations?.qualify;
+      }
+
       if (table.memory) {
         jobUpdate["memory"] = {
           amount: table.memory * 1000

src/services/nimbus/nimbus.service.ts

@@ -34,7 +34,7 @@ export class NimbusService {
         const path = `/api/catalog/rename-tables/?database_name=${encodeURIComponent(database)}&table_name=${encodeURIComponent(old.table_name)}&table_schema=${encodeURIComponent(old.table_schema)}`;
 
         try {
-            this.logger.info("Request for PATCH" + nimbusUrl + path);
+            this.logger.info("Request for PATCH " + nimbusUrl + path);
             this.logger.info("Payload: " + JSON.stringify(update));
             const { data } = await axios.patch(nimbusUrl + path, {
                 table_name: update.table_name,