Merge pull request #461 from dadosfera/feat/qualify

FEAT: using batch route to update pipeline

src/modules/inputs/inputs.controller.ts

@@ -99,6 +99,7 @@ export class InputsController {
       customer: info.customer,
     });
 
+    this.logger.info(JSON.stringify(body))
     const response = await this.inputService.create({ body, info });
 
     return response;

src/modules/pipelinesV2/pipelines.service.ts

@@ -356,125 +356,92 @@ export class PipelinesService implements OnModuleInit {
       info
     )
 
-    const requests = [];
-
     this.logger.info('Dynamo Response', updateInputResponse);
+    const jobsUpdated = [];
 
     for (const [index, table] of updateInputDTO.tables.entries()) {
-      const id = `${pipelineIdFormat}_${index}`;
+      const jobUpdate = {
+        job_id: `${pipelineIdFormat}_${index}`,
+      }
+
+      if (table.memory) {
+        jobUpdate["memory"] = {
+          amount: table.memory * 1000
+        }
+      }
+      
       this.logger.info('Updating input reference for table', table.name);
-      const body = {}
+      let hasUpdateSyncMode = false;
+
+      const jobSyncMode = {}
+
       if (table.columns) {
-        body['column_include_list'] = table.columns;
+        hasUpdateSyncMode = true;
+        jobSyncMode['column_include_list'] = table.columns;
       }
 
       if (table.reference_column) {
-        body['incremental_column_name'] = table.reference_column.name;
+        hasUpdateSyncMode = true;
-        body['incremental_column_type'] = table.reference_column.type;
+        jobSyncMode['incremental_column_name'] = table.reference_column.name;
+        jobSyncMode['incremental_column_type'] = table.reference_column.type;
       }
 
       if (table.identifier_columns) {
-        body['primary_keys'] = table.identifier_columns;
+        hasUpdateSyncMode = true;
-      }
+        jobSyncMode['primary_keys'] = table.identifier_columns;
-
-      this.logger.info('Request body', body);
-      const updateCollumns = this.platformAPI.proxy(
-        'PATCH',
-        `/jobs/${id}/input`,
-        user,
-        body
-      )
-      requests.push(updateCollumns);
-
-      if (table.memory) {
-        this.logger.info('Updating memory allocation for table', table.name);
-        const updateMemory = this.platformAPI.proxy(
-          'PUT',
-          `/jobs/${id}/memory`,
-          user,
-          {
-            amount: table.memory
-          }
-        )
-        requests.push(updateMemory);
       }
 
       if (table.type) {
-          const updateSyncMode = this.updatePipelineSyncMode(table, id, user);
+        hasUpdateSyncMode = true;
-          requests.push(updateSyncMode);
+        jobSyncMode['target_load_type'] = table.type;
+
       }
-    }
 
-    this.logger.info('Create Platform Request for each JOB');
+      if(hasUpdateSyncMode) {
+        jobUpdate["sync_mode"] = jobSyncMode;
+      }
 
-    if (updateInputDTO.cron) {
+      if (Object.keys(table.destinations).length > 1) {
-      const crnUpdatedRequest = new Promise(async (resolve, reject) => {
+        let hasChanges = false
-        const response = await this.updatePipelineCron(updateInputDTO.cron, pipelineIdFormat, user);
+        const jobRenameTables = {
-
+            raw: {},
-        if (response.error) {
+            qualify: {}
-          this.logger.error('Error updating pipeline cron', response.error);
-          return reject(new ErrorBuilder(response.error));
         }
-        this.logger.error('Pipeline cron updated successfully', response);
+
-        return resolve(response);
+        if (Object.keys(table.destinations.raw).length > 1) {
-      });
+          hasChanges = true;
-      requests.push(crnUpdatedRequest);
+          jobRenameTables.raw = table.destinations.raw;
+        }
+
+        if (Object.keys(table.destinations.qualify).length > 1) {
+          hasChanges = true;
+          jobRenameTables.qualify = table.destinations.qualify;
+        }
+
+        if (hasChanges) {
+          jobUpdate['rename_tables'] = jobRenameTables;
+        }
+      }
+
+      jobsUpdated.push(jobUpdate);
     }
 
-    this.logger.info('Executing all request for the platform api');
+    this.logger.info('Request body:' + JSON.stringify({
+      jobs_updated: jobsUpdated
+    }));
 
-    const results = await Promise.allSettled(requests);
+    const response = await this.platformAPI.proxy(
-    this.logger.info('Platform api response', results);
+      'POST',
+        `/pipelines/${pipelineIdFormat}/batch-update`,
+      user,
+      {
+        job_updates: jobsUpdated
+      }
+    )
+    this.logger.info('Platform api response: ' + JSON.stringify(response));
 
     return updateInputResponse;
     
   }
 
-  private async updatePipelineSyncMode(table: UpdateTableDTO,  pipelineId: string, user: RequestUser) {
-    const body = {
-      target_load_type: table.type
-    }
-
-    if (table.type === 'incremental_with_qualify') {
-      body['incremental_column_name'] = table.reference_column.name;
-      body['incremental_column_type'] = table.reference_column.type;
-      body['primary_keys'] = table.identifier_columns;
-    }
-
-    if (table.type === 'incremental') {
-      body['incremental_column_name'] = table.reference_column.name;
-      body['incremental_column_type'] = table.reference_column.type;
-    }
-
-    this.logger.info('Updating pipeline sync mode', {
-      pipelineId,
-      body
-    });
-
-    return this.platformAPI.proxy(
-      "POST",
-      `/jobs/jdbc/${pipelineId}/sync-mode`,
-      user,
-      body
-    )
-  }
-
-  private async updatePipelineCron(cron: string, pipelineId: string, user: RequestUser) {
-    try {
-      const response = await this.platformAPI.proxy(
-        'PATCH',
-        `/pipeline/${pipelineId}`,
-        user,
-        {
-          cron
-        }
-      );
-
-      return response
-    } catch (error) {
-        return {
-          error: error.message
-        }
-    }
-  }
 }