diff --git a/src/modules/inputs/inputs.controller.ts b/src/modules/inputs/inputs.controller.ts index 82ddd36..a8cc143 100644 --- a/src/modules/inputs/inputs.controller.ts +++ b/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; diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index e23ce23..c0a1ad6 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/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; - body['incremental_column_type'] = table.reference_column.type; + hasUpdateSyncMode = true; + 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; - } - - 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); + hasUpdateSyncMode = true; + jobSyncMode['primary_keys'] = table.identifier_columns; } if (table.type) { - const updateSyncMode = this.updatePipelineSyncMode(table, id, user); - requests.push(updateSyncMode); + hasUpdateSyncMode = true; + jobSyncMode['target_load_type'] = table.type; + } - } - this.logger.info('Create Platform Request for each JOB'); + if(hasUpdateSyncMode) { + jobUpdate["sync_mode"] = jobSyncMode; + } - if (updateInputDTO.cron) { - const crnUpdatedRequest = new Promise(async (resolve, reject) => { - const response = await this.updatePipelineCron(updateInputDTO.cron, pipelineIdFormat, user); - - if (response.error) { - this.logger.error('Error updating pipeline cron', response.error); - return reject(new ErrorBuilder(response.error)); + if (Object.keys(table.destinations).length > 1) { + let hasChanges = false + const jobRenameTables = { + raw: {}, + qualify: {} } - this.logger.error('Pipeline cron updated successfully', response); - return resolve(response); - }); - requests.push(crnUpdatedRequest); + + if (Object.keys(table.destinations.raw).length > 1) { + hasChanges = true; + 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); - this.logger.info('Platform api response', results); + const response = await this.platformAPI.proxy( + '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 - } - } - } }