mirror of
https://github.com/dadosfera/maestro.git
synced 2026-08-31 19:58:21 +00:00
FIX: skip nimbus update when customer haven't catalog module
This commit is contained in:
@@ -4570,50 +4570,6 @@
|
||||
]
|
||||
}
|
||||
},
|
||||
"/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": {
|
||||
"get": {
|
||||
"operationId": "PlatformApiController_getPipelineRunJobs",
|
||||
"summary": "Get pipeline run jobs",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "pipelineId",
|
||||
"required": true,
|
||||
"in": "path",
|
||||
"schema": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "runId",
|
||||
"required": true,
|
||||
"in": "path",
|
||||
"schema": {
|
||||
"type": "string"
|
||||
}
|
||||
}
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"tags": [
|
||||
"Platform API"
|
||||
],
|
||||
"security": [
|
||||
{
|
||||
"access-token": []
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"/platform/jobs/{jobId}/input": {
|
||||
"put": {
|
||||
"operationId": "PlatformApiController_updateJobInput",
|
||||
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
} from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
|
||||
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
|
||||
import { CreateInputReq } from './dtos/input.model';
|
||||
import { Metadata } from '@grpc/grpc-js';
|
||||
|
||||
@Injectable()
|
||||
|
||||
@@ -73,10 +74,10 @@ export class InputsService {
|
||||
objectCamelToSnake(createInputResponse);
|
||||
return createInputResponse;
|
||||
},
|
||||
update: async (updateInputDTO: UpdateInputRequest): Promise<InputUpdateResponse> => {
|
||||
update: async (updateInputDTO: UpdateInputRequest, metadata: Metadata): Promise<InputUpdateResponse> => {
|
||||
this.logger.info('InputClientService - Update' + JSON.stringify(updateInputDTO));
|
||||
const updateInputResponse = await lastValueFrom(
|
||||
this.inputWriteService.InputUpdate(updateInputDTO),
|
||||
this.inputWriteService.InputUpdate(updateInputDTO, metadata),
|
||||
);
|
||||
|
||||
return updateInputResponse;
|
||||
@@ -206,7 +207,7 @@ export class InputsService {
|
||||
return findOneInputResponse;
|
||||
}
|
||||
|
||||
async update(id: string, data, info: Info) {
|
||||
async update(id: string, data, info: Info, metadata?: Metadata) {
|
||||
// this.validateCron({ ...data, info });
|
||||
try {
|
||||
const {
|
||||
@@ -217,7 +218,7 @@ export class InputsService {
|
||||
id,
|
||||
...data,
|
||||
info,
|
||||
});
|
||||
}, metadata);
|
||||
|
||||
const updateInputResponse = this.adjustInputPayload(
|
||||
input,
|
||||
|
||||
@@ -317,7 +317,6 @@ export class PipelinesController {
|
||||
) {
|
||||
this.logger.info('PipelinesController - update', { user });
|
||||
|
||||
const { customer_id, customer_name, user_id, username } = user;
|
||||
const info: Info = {
|
||||
user_id: user.user_id,
|
||||
customer: user.customer_name,
|
||||
@@ -325,13 +324,7 @@ export class PipelinesController {
|
||||
pipeline_id: pipelineId
|
||||
};
|
||||
|
||||
const metadata = PackTheMetadata({
|
||||
customer_id,
|
||||
customer_name,
|
||||
user_id,
|
||||
username,
|
||||
language,
|
||||
});
|
||||
const metadata = PackTheMetadata(user);
|
||||
|
||||
const response = await this.pipelinesClientService.updatePipelineInput(
|
||||
pipelineId,
|
||||
|
||||
@@ -383,7 +383,8 @@ export class PipelinesService implements OnModuleInit {
|
||||
const updateInputResponse = await this.inputsService.update(
|
||||
inputId,
|
||||
updateInputDTO,
|
||||
info
|
||||
info,
|
||||
metadata
|
||||
);
|
||||
|
||||
const inputRollback = () => {
|
||||
@@ -404,34 +405,36 @@ export class PipelinesService implements OnModuleInit {
|
||||
|
||||
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
|
||||
}
|
||||
);
|
||||
if (user.customer_modules.includes('catalog')) {
|
||||
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");
|
||||
}
|
||||
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 {
|
||||
|
||||
@@ -9,6 +9,7 @@ interface IMetadata {
|
||||
details?: string;
|
||||
sensitive?: string;
|
||||
roles?: string[];
|
||||
customer_modules?: string[];
|
||||
is_data_manager?: boolean;
|
||||
access_token?: string;
|
||||
host?: string;
|
||||
|
||||
Reference in New Issue
Block a user