mirror of
https://github.com/dadosfera/maestro.git
synced 2026-08-31 19:58:21 +00:00
Marcos Rodrigues Silva
+2
-2
@@ -22,7 +22,7 @@ ENV PUPPETEER_SKIP_CHROMIUM_DOWNLOAD=true \
|
||||
# run aws cli without mounting secret, because CI already has AWS credentials
|
||||
FROM build_base AS ci_image
|
||||
RUN aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1
|
||||
RUN npm ci
|
||||
RUN npm ci --ignore-scripts
|
||||
COPY . .
|
||||
|
||||
|
||||
@@ -37,7 +37,7 @@ FROM build_base AS dev
|
||||
RUN --mount=type=secret,id=aws,target=/root/.aws/credentials \
|
||||
aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1
|
||||
# flag --build-from-source is required to force-build sqlite3
|
||||
RUN npm ci
|
||||
RUN npm ci --ignore-scripts
|
||||
COPY . .
|
||||
ENTRYPOINT npm run start:dev
|
||||
|
||||
|
||||
+1
-1
@@ -22,7 +22,7 @@ ENV PUPPETEER_SKIP_CHROMIUM_DOWNLOAD=true \
|
||||
FROM build_base AS build
|
||||
RUN --mount=type=secret,id=aws,target=/root/.aws/credentials \
|
||||
aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1
|
||||
RUN npm ci
|
||||
RUN npm ci --ignore-scripts
|
||||
COPY . .
|
||||
RUN npm run build
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
|
||||
# Maestro
|
||||
|
||||
|
||||
Maestro é a API principal da Dadosfera. É responsável pela comunicação do Frontend com nossos microsserviços.
|
||||
|
||||
```mermaid
|
||||
|
||||
+1359
-1321
File diff suppressed because it is too large
Load Diff
Generated
+8
-8
@@ -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.39.0",
|
||||
"@dadosfera/protospack-v2": "^3.40.0-beta.5",
|
||||
"@grpc/grpc-js": "^1.9.3",
|
||||
"@grpc/proto-loader": "^0.7.9",
|
||||
"@nestjs/cli": "^9.5.0",
|
||||
@@ -31,7 +31,7 @@
|
||||
"@nestjs/schematics": "^9.2.0",
|
||||
"@nestjs/swagger": "^6.3.0",
|
||||
"@nestjs/testing": "^9.4.3",
|
||||
"axios": "^0.30.2",
|
||||
"axios": "0.30.3",
|
||||
"cache-manager": "^5.1.4",
|
||||
"cache-manager-ioredis-yet": "^1.1.0",
|
||||
"class-transformer": "^0.5.1",
|
||||
@@ -1745,9 +1745,9 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@dadosfera/protospack-v2": {
|
||||
"version": "3.39.0",
|
||||
"resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.39.0.tgz",
|
||||
"integrity": "sha512-z6q+ZRpbDT6apfmly1+X818DgAFUCilFO0ATP7xR5+SH+GAUAJLY1dJRQDEkq5vFIfXnC2Nmm9aODZAs+9/5vw==",
|
||||
"version": "3.40.0-beta.5",
|
||||
"resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.5.tgz",
|
||||
"integrity": "sha512-iocKv/XXp2jKAasO5ONgm31cKfLgNsU4pEKZMr6YnR7nQaH11WcW7rnuagNxWoik++wLUqbYyf0bZWRDzMlCPA==",
|
||||
"dependencies": {
|
||||
"@grpc/grpc-js": "^1.9.3",
|
||||
"rxjs": "^7.5.5"
|
||||
@@ -5598,9 +5598,9 @@
|
||||
}
|
||||
},
|
||||
"node_modules/axios": {
|
||||
"version": "0.30.2",
|
||||
"resolved": "https://registry.npmjs.org/axios/-/axios-0.30.2.tgz",
|
||||
"integrity": "sha512-0pE4RQ4UQi1jKY6p7u6i1Tkzqmu+d+/tHS7Q7rKunWLB9WyilBTpHHpXzPNMDj5hTbK0B0PTLSz07yqMBiF6xg==",
|
||||
"version": "0.30.3",
|
||||
"resolved": "https://registry.npmjs.org/axios/-/axios-0.30.3.tgz",
|
||||
"integrity": "sha512-5/tmEb6TmE/ax3mdXBc/Mi6YdPGxQsv+0p5YlciXWt3PHIn0VamqCXhRMtScnwY3lbgSXLneOuXAKUhgmSRpwg==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"follow-redirects": "^1.15.4",
|
||||
|
||||
+7
-3
@@ -10,7 +10,7 @@
|
||||
},
|
||||
"scripts": {
|
||||
"co:login": "aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1",
|
||||
"proto-update": "npm i @dadosfera/protospack-v2@latest --save-exact",
|
||||
"proto-update": "npm i @dadosfera/protospack-v2@v3.40.0-beta.1 --save-exact",
|
||||
"prebuild": "rimraf dist",
|
||||
"build": "nest build",
|
||||
"format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
|
||||
@@ -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.39.0",
|
||||
"@dadosfera/protospack-v2": "^3.40.0-beta.5",
|
||||
"@grpc/grpc-js": "^1.9.3",
|
||||
"@grpc/proto-loader": "^0.7.9",
|
||||
"@nestjs/cli": "^9.5.0",
|
||||
@@ -49,7 +49,7 @@
|
||||
"@nestjs/schematics": "^9.2.0",
|
||||
"@nestjs/swagger": "^6.3.0",
|
||||
"@nestjs/testing": "^9.4.3",
|
||||
"axios": "^0.30.2",
|
||||
"axios": "0.30.3",
|
||||
"cache-manager": "^5.1.4",
|
||||
"cache-manager-ioredis-yet": "^1.1.0",
|
||||
"class-transformer": "^0.5.1",
|
||||
@@ -81,6 +81,7 @@
|
||||
"swagger-ui-express": "^4.6.3"
|
||||
},
|
||||
"overrides": {
|
||||
"axios": "0.30.3",
|
||||
"form-data": "^4.0.4",
|
||||
"body-parser": "^1.20.3",
|
||||
"cross-spawn": "^7.0.5",
|
||||
@@ -117,5 +118,8 @@
|
||||
"ts-node": "^10.9.1",
|
||||
"tsconfig-paths": "^3.14.2",
|
||||
"typescript": "^4.9.5"
|
||||
},
|
||||
"resolutions": {
|
||||
"axios": "0.30.3"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -111,3 +111,4 @@ function configureSwagger(app: INestApplication) {
|
||||
);
|
||||
}
|
||||
bootstrap();
|
||||
|
||||
|
||||
@@ -478,6 +478,7 @@ export class AuthController {
|
||||
@Get('me')
|
||||
async getMe(@Req() req: Request, @Res() res: Response) {
|
||||
this.logger.info('GET /auth/me ')
|
||||
this.logger.info(JSON.stringify(req.headers));
|
||||
|
||||
// Check for API key header first
|
||||
const apiKey = req.get('X-Api-key');
|
||||
@@ -488,6 +489,7 @@ export class AuthController {
|
||||
const userDto = {
|
||||
id: api_key.user_id,
|
||||
name: api_key.username,
|
||||
email: api_key.username,
|
||||
customer: {
|
||||
id: api_key.customer_id,
|
||||
name: api_key.customer_name,
|
||||
|
||||
@@ -437,7 +437,8 @@ export class AuthClientService implements OnModuleInit {
|
||||
|
||||
const userDto: UserDTO = {
|
||||
id: user.id,
|
||||
name: user.username,
|
||||
name: user.name,
|
||||
email: user.email,
|
||||
jobTitle: user?.jobTitle || null,
|
||||
department: user?.department || null,
|
||||
hierarchy: user?.hierarchy || null,
|
||||
|
||||
@@ -144,6 +144,7 @@ export interface BulkEditResponse {
|
||||
export type UserDTO = {
|
||||
id: string,
|
||||
name: string,
|
||||
email: string,
|
||||
jobTitle?: string,
|
||||
department?: string,
|
||||
hierarchy?: string,
|
||||
|
||||
@@ -734,6 +734,64 @@ class CatalogService implements OnModuleInit {
|
||||
}
|
||||
}
|
||||
|
||||
async renameTableOnNimbus(
|
||||
nimbusUrl: string,
|
||||
nimbusId: number,
|
||||
changes: { table_name?: string; table_schema?: string; display_name?: string },
|
||||
): Promise<void> {
|
||||
const endpoint = `${nimbusUrl}/api/catalog/table-metadata/${nimbusId}`;
|
||||
this.logger.info(`Renaming table-metadata ${nimbusId} on Nimbus`, { endpoint, changes });
|
||||
await axios.patch(endpoint, changes);
|
||||
}
|
||||
|
||||
async renameColumnMetadataOnNimbus(
|
||||
nimbusUrl: string,
|
||||
databaseName: string,
|
||||
oldTableName: string,
|
||||
oldTableSchema: string,
|
||||
newTableName: string,
|
||||
newTableSchema: string,
|
||||
): Promise<void> {
|
||||
const listEndpoint = `${nimbusUrl}/api/catalog/column-metadata/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
|
||||
this.logger.info(`Fetching column-metadata records to rename`, { listEndpoint });
|
||||
const { data: columns } = await axios.get(listEndpoint);
|
||||
|
||||
const filtered = Array.isArray(columns) ? columns : [];
|
||||
|
||||
for (const column of filtered) {
|
||||
const patchEndpoint = `${nimbusUrl}/api/catalog/column-metadata/${column.id}`;
|
||||
await axios.patch(patchEndpoint, {
|
||||
table_name: newTableName,
|
||||
table_schema: newTableSchema,
|
||||
});
|
||||
}
|
||||
this.logger.info(`Renamed ${filtered.length} column-metadata records on Nimbus`);
|
||||
}
|
||||
|
||||
async renameDataPreviewOnNimbus(
|
||||
nimbusUrl: string,
|
||||
databaseName: string,
|
||||
oldTableName: string,
|
||||
oldTableSchema: string,
|
||||
newTableName: string,
|
||||
newTableSchema: string,
|
||||
): Promise<void> {
|
||||
const listEndpoint = `${nimbusUrl}/api/catalog/data-preview/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
|
||||
this.logger.info(`Fetching data-preview records to rename`, { listEndpoint });
|
||||
const { data: previews } = await axios.get(listEndpoint);
|
||||
|
||||
const filtered = Array.isArray(previews) ? previews : [];
|
||||
|
||||
for (const preview of filtered) {
|
||||
const patchEndpoint = `${nimbusUrl}/api/catalog/data-preview/${preview.id}`;
|
||||
await axios.patch(patchEndpoint, {
|
||||
table_name: newTableName,
|
||||
table_schema: newTableSchema,
|
||||
});
|
||||
}
|
||||
this.logger.info(`Renamed ${filtered.length} data-preview records on Nimbus`);
|
||||
}
|
||||
|
||||
async catalogDatasetItem(table_metadata_id: number, metadata: Metadata) {
|
||||
const customer_name_raw = metadata.get('customer_name');
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -17,6 +17,8 @@ import {
|
||||
InputCreateGenericRequest,
|
||||
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';
|
||||
@@ -71,8 +73,8 @@ export class InputsService {
|
||||
objectCamelToSnake(createInputResponse);
|
||||
return createInputResponse;
|
||||
},
|
||||
update: async (updateInputDTO: UpdateInputRequest) => {
|
||||
this.logger.info('InputClientService - Update');
|
||||
update: async (updateInputDTO: UpdateInputRequest): Promise<InputUpdateResponse> => {
|
||||
this.logger.info('InputClientService - Update' + JSON.stringify(updateInputDTO));
|
||||
const updateInputResponse = await lastValueFrom(
|
||||
this.inputWriteService.InputUpdate(updateInputDTO),
|
||||
);
|
||||
@@ -205,23 +207,46 @@ export class InputsService {
|
||||
}
|
||||
|
||||
async update(id: string, data, info: Info) {
|
||||
this.validateCron({ ...data, info });
|
||||
// this.validateCron({ ...data, info });
|
||||
try {
|
||||
const updateInputResponse: any = await this.OLD_inputClient.update({
|
||||
const {
|
||||
tablesUpdate,
|
||||
dataAssetUpdate,
|
||||
input
|
||||
} = await this.OLD_inputClient.update({
|
||||
id,
|
||||
info,
|
||||
...data,
|
||||
info,
|
||||
});
|
||||
|
||||
updateInputResponse.input = this.adjustInputPayload(
|
||||
updateInputResponse?.input,
|
||||
const updateInputResponse = this.adjustInputPayload(
|
||||
input,
|
||||
);
|
||||
return updateInputResponse;
|
||||
return {
|
||||
input: updateInputResponse,
|
||||
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));
|
||||
}
|
||||
@@ -263,4 +288,12 @@ export class InputsService {
|
||||
};
|
||||
return formatedPayload;
|
||||
}
|
||||
|
||||
async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
|
||||
return lastValueFrom(this.inputWriteService.MarkTableDeleted(data));
|
||||
}
|
||||
|
||||
async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
|
||||
return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,15 @@
|
||||
import { ApiProperty, ApiPropertyOptional, OmitType } from '@nestjs/swagger';
|
||||
import { Info } from '@dadosfera/protospack/dist/lib/interfaces';
|
||||
|
||||
export class PipelineInputsDTO {
|
||||
@ApiProperty()
|
||||
tables: Array<{
|
||||
name: string,
|
||||
type: string,
|
||||
|
||||
}>
|
||||
}
|
||||
|
||||
export class IPipelineV2 {
|
||||
@ApiProperty()
|
||||
id: string;
|
||||
@@ -123,3 +132,30 @@ export class PipelineFindAllReq {
|
||||
@ApiPropertyOptional()
|
||||
type?: string | undefined;
|
||||
}
|
||||
|
||||
export interface UpdateTableDTO {
|
||||
name: string;
|
||||
type: string;
|
||||
columns: string[];
|
||||
destinations: {
|
||||
raw: {
|
||||
table_schema: string;
|
||||
table_name: string;
|
||||
};
|
||||
qualify: {
|
||||
table_schema: string;
|
||||
table_name: string;
|
||||
};
|
||||
};
|
||||
identifier_columns: string[];
|
||||
reference_column: {
|
||||
name: string;
|
||||
type: string;
|
||||
};
|
||||
memory: number;
|
||||
}
|
||||
|
||||
export interface UpdatePlatformInputRequest {
|
||||
cron: string;
|
||||
tables: Array<UpdateTableDTO>;
|
||||
}
|
||||
|
||||
@@ -43,11 +43,18 @@ import {
|
||||
IPipelineV2,
|
||||
IInitUploadCSVFile,
|
||||
PipelineFindAllReq,
|
||||
UpdatePlatformInputRequest,
|
||||
} from './interfaces';
|
||||
import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
|
||||
import { LanguageEnum } from 'src/utils/languages.enum';
|
||||
import { Language } from 'src/decorators/language.decorator';
|
||||
import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator';
|
||||
import { TableColumns } from '../inputs/dtos/input.model';
|
||||
import { UpdateInputRequest } from '../inputs/dtos/old_interfaces';
|
||||
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
|
||||
|
||||
type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any };
|
||||
type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] };
|
||||
|
||||
@ApiTags('PipelinesV2')
|
||||
@ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
|
||||
@@ -58,6 +65,7 @@ export class PipelinesController {
|
||||
constructor(
|
||||
@Inject(DadosferaLogger)
|
||||
dadosferaLogger: DadosferaLogger,
|
||||
|
||||
private pipelinesClientService: PipelinesService,
|
||||
private oldPipelinesService: OldPipelineService,
|
||||
) {
|
||||
@@ -218,28 +226,27 @@ export class PipelinesController {
|
||||
language,
|
||||
});
|
||||
|
||||
const result = await this.pipelinesClientService
|
||||
.findOne({ id }, metadata)
|
||||
.then((res) => {
|
||||
//{pipeline:{tables: {tables: [], input_id: ''}}}
|
||||
let tables = JSON.parse(res.pipeline.config.tables);
|
||||
if (tables?.tables) tables = tables.tables;
|
||||
Object.assign(res.pipeline, {
|
||||
transformations: res.pipeline.transformations
|
||||
? JSON.parse(res.pipeline.transformations)
|
||||
: [],
|
||||
config: {
|
||||
cron: res.pipeline.config.cron,
|
||||
tables,
|
||||
},
|
||||
properties: res.pipeline.properties
|
||||
? JSON.parse(res.pipeline.properties)
|
||||
: {},
|
||||
});
|
||||
return res;
|
||||
});
|
||||
const pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata);
|
||||
|
||||
return result;
|
||||
const parsed: PipelineTablesConfig = JSON.parse(pipelineRes.pipeline.config.tables);
|
||||
const input_id = parsed.input_id;
|
||||
const tables: PipelineTable[] = parsed.tables ?? [];
|
||||
|
||||
Object.assign(pipelineRes.pipeline, {
|
||||
transformations: pipelineRes.pipeline.transformations
|
||||
? JSON.parse(pipelineRes.pipeline.transformations)
|
||||
: [],
|
||||
config: {
|
||||
cron: pipelineRes.pipeline.config.cron,
|
||||
tables,
|
||||
input_id,
|
||||
},
|
||||
properties: pipelineRes.pipeline.properties
|
||||
? JSON.parse(pipelineRes.pipeline.properties)
|
||||
: {},
|
||||
});
|
||||
|
||||
return pipelineRes;
|
||||
}
|
||||
|
||||
@Patch('/:id')
|
||||
@@ -277,6 +284,46 @@ export class PipelinesController {
|
||||
return response;
|
||||
}
|
||||
|
||||
@Patch('/:pipelineId/inputs/:id')
|
||||
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
||||
async updatePipelineInput(
|
||||
@Language() language: LanguageEnum,
|
||||
@Body() pipelineInputDTO: UpdatePlatformInputRequest,
|
||||
@Param('id') inputId: string,
|
||||
@Param('pipelineId') pipelineId: string,
|
||||
@User() user: RequestUser,
|
||||
) {
|
||||
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,
|
||||
customer_id: user.customer_id,
|
||||
pipeline_id: pipelineId
|
||||
};
|
||||
|
||||
const metadata = PackTheMetadata({
|
||||
customer_id,
|
||||
customer_name,
|
||||
user_id,
|
||||
username,
|
||||
language,
|
||||
});
|
||||
|
||||
const response = await this.pipelinesClientService.updatePipelineInput(
|
||||
pipelineId,
|
||||
inputId,
|
||||
pipelineInputDTO,
|
||||
info,
|
||||
user,
|
||||
metadata,
|
||||
);
|
||||
|
||||
this.logger.info('PipelinesController - update: OK', { user });
|
||||
return response;
|
||||
}
|
||||
|
||||
@ApiInternalOnlyEndpoint()
|
||||
@Put('/:id')
|
||||
@ApiOperation({
|
||||
|
||||
@@ -11,6 +11,9 @@ import { PipelinesModule as OldPipelineModule } from 'src/modules/pipelines/pipe
|
||||
import { ConnectorModule } from '../connector/connector.module';
|
||||
import { InputsModule } from '../inputs/inputs.module';
|
||||
import { TransformationsModule } from '../transformations/transformations.module';
|
||||
import { PlatformApiModule } from '../platform-api/platform-api.module';
|
||||
import { NimbusServicesModule } from 'src/services/nimbus/nimbus.module';
|
||||
import { NimbusService } from 'src/services/nimbus/nimbus.service';
|
||||
|
||||
const client = new PipelinesClientConfiguration();
|
||||
|
||||
@@ -21,9 +24,11 @@ const client = new PipelinesClientConfiguration();
|
||||
ConnectorModule,
|
||||
InputsModule,
|
||||
TransformationsModule,
|
||||
PlatformApiModule,
|
||||
NimbusServicesModule
|
||||
],
|
||||
controllers: [PipelinesController],
|
||||
providers: [PipelinesService, DadosferaLogger],
|
||||
providers: [PipelinesService, DadosferaLogger, NimbusService],
|
||||
exports: [PipelinesService],
|
||||
})
|
||||
export class PipelinesV2Module {}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
/* eslint-disable no-async-promise-executor */
|
||||
import {
|
||||
BadRequestException,
|
||||
HttpException,
|
||||
@@ -16,7 +17,7 @@ import { lastValueFrom } from 'rxjs';
|
||||
|
||||
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
|
||||
import { PipelinesClientConfiguration } from './pipelines-client';
|
||||
import { ICreatePipelineV2Req } from './interfaces';
|
||||
import { ICreatePipelineV2Req, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces';
|
||||
import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
|
||||
import { Metadata } from '@grpc/grpc-js';
|
||||
import { ConnectorClientService } from '../connector/client.service';
|
||||
@@ -26,6 +27,13 @@ import { TransformationsService } from '../transformations/transformations.servi
|
||||
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 } 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;
|
||||
@@ -39,6 +47,8 @@ export class PipelinesService implements OnModuleInit {
|
||||
private readonly connectorService: ConnectorClientService,
|
||||
private readonly inputsService: InputsService,
|
||||
private readonly transformationsService: TransformationsService,
|
||||
private readonly platformAPI: PlatformApiService,
|
||||
private readonly nimbusService: NimbusService
|
||||
) {
|
||||
this.logger = dadosferaLogger.logger;
|
||||
}
|
||||
@@ -138,6 +148,7 @@ export class PipelinesService implements OnModuleInit {
|
||||
const findOnePipelineResponse = await lastValueFrom(
|
||||
this.pipelineReadService.PipelineV2FindOne(data, metadata),
|
||||
);
|
||||
console.log('pipeline find one response', findOnePipelineResponse);
|
||||
this.logger.info('Done');
|
||||
|
||||
return findOnePipelineResponse;
|
||||
@@ -339,4 +350,231 @@ export class PipelinesService implements OnModuleInit {
|
||||
|
||||
return res;
|
||||
}
|
||||
|
||||
async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) {
|
||||
this.logger.info('InputClientService - Update');
|
||||
|
||||
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('_');
|
||||
const rollback: RollbackPromise[] = [];
|
||||
|
||||
const updateInputResponse = await this.inputsService.update(
|
||||
inputId,
|
||||
updateInputDTO,
|
||||
info
|
||||
);
|
||||
|
||||
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,
|
||||
updateInputResponse.input.type,
|
||||
updateInputDTO,
|
||||
user
|
||||
);
|
||||
} catch (error) {
|
||||
this.logger.error(error);
|
||||
await this.executeRenameRollback(rollback)
|
||||
throw new Error("Error Platform API updating jobs");
|
||||
}
|
||||
|
||||
return updateInputResponse;
|
||||
}
|
||||
|
||||
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, {
|
||||
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);
|
||||
|
||||
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, pipelineType: string, updateInputDTO: UpdatePlatformInputRequest, user: RequestUser) {
|
||||
const jobsUpdated = [];
|
||||
|
||||
for (const [index, table] of updateInputDTO.tables.entries()) {
|
||||
const jobUpdate = {
|
||||
job_id: `${pipelineId}_${index}`,
|
||||
}
|
||||
|
||||
if (table.type !== "incremental_with_qualify") {
|
||||
delete table.destinations?.qualify;
|
||||
}
|
||||
|
||||
if (table.memory) {
|
||||
jobUpdate["memory"] = {
|
||||
amount: table.memory * 1000
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info('Updating input reference for table: ' + table.name);
|
||||
let hasUpdateSyncMode = false;
|
||||
|
||||
const jobSyncMode = {}
|
||||
|
||||
if (table.columns) {
|
||||
hasUpdateSyncMode = true;
|
||||
jobSyncMode['column_include_list'] = table.columns;
|
||||
}
|
||||
|
||||
if (table.reference_column) {
|
||||
hasUpdateSyncMode = true;
|
||||
jobSyncMode['incremental_column_name'] = table.reference_column.name;
|
||||
jobSyncMode['incremental_column_type'] = table.reference_column.type;
|
||||
}
|
||||
|
||||
if (table.identifier_columns) {
|
||||
hasUpdateSyncMode = true;
|
||||
jobSyncMode['primary_keys'] = table.identifier_columns;
|
||||
}
|
||||
|
||||
if (table.type) {
|
||||
hasUpdateSyncMode = true;
|
||||
|
||||
jobSyncMode['target_load_type'] = table.type;
|
||||
}
|
||||
|
||||
if(hasUpdateSyncMode) {
|
||||
jobUpdate["sync_mode"] = jobSyncMode;
|
||||
}
|
||||
|
||||
if (Object.keys(table.destinations).length > 1) {
|
||||
let hasChanges = false
|
||||
const jobRenameTables = {
|
||||
raw: {},
|
||||
qualify: {}
|
||||
}
|
||||
|
||||
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('Request body:' + JSON.stringify({
|
||||
jobs_updated: jobsUpdated
|
||||
}));
|
||||
|
||||
const response = await this.platformAPI.proxy(
|
||||
'PUT',
|
||||
`/pipeline/${pipelineId}/jobs`,
|
||||
user,
|
||||
{
|
||||
job_updates: jobsUpdated
|
||||
}
|
||||
)
|
||||
this.logger.info('Platform api response: ' + JSON.stringify(response));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -10,6 +10,8 @@ import {
|
||||
Query,
|
||||
Inject,
|
||||
BadRequestException,
|
||||
HttpException,
|
||||
NotFoundException,
|
||||
} from '@nestjs/common';
|
||||
import { ApiTags, ApiOperation } from '@nestjs/swagger';
|
||||
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
|
||||
@@ -25,6 +27,10 @@ import { ElasticsearchService } from '../../services/elasticsearch';
|
||||
import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
|
||||
import { CustomersService } from '../customers/customers.service';
|
||||
import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
|
||||
import { CatalogService } from '../catalog/catalog.service';
|
||||
import { PackTheMetadata } from '../../utils/PackTheMetadata';
|
||||
import { ValidationTableDTO } from './platform-api.dto';
|
||||
import { InputsService } from '../inputs/inputs.service';
|
||||
|
||||
|
||||
type ValidateTablesDTO = {
|
||||
@@ -34,6 +40,11 @@ type ValidateTablesDTO = {
|
||||
}>
|
||||
}
|
||||
|
||||
type RenameTablesBody = {
|
||||
raw?: { table_name: string; table_schema: string };
|
||||
qualify?: { table_name: string; table_schema: string };
|
||||
}
|
||||
|
||||
@ApiTags('Platform API')
|
||||
@Controller('platform')
|
||||
export class PlatformApiController {
|
||||
@@ -44,6 +55,8 @@ export class PlatformApiController {
|
||||
private readonly elasticsearchService: ElasticsearchService,
|
||||
private readonly dynamoDBService: DynamoDBService,
|
||||
private readonly customersService: CustomersService,
|
||||
private readonly catalogService: CatalogService,
|
||||
private readonly inputsService: InputsService,
|
||||
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
|
||||
) {
|
||||
this.logger = dadosferaLogger.logger;
|
||||
@@ -75,6 +88,23 @@ export class PlatformApiController {
|
||||
return jobId?.replace(/-/g, '_') || '';
|
||||
}
|
||||
|
||||
private async getJobByAnyConnectorType(normalizedJobId: string, user: RequestUser): Promise<any> {
|
||||
const connectorTypes = ['jdbc', 'singer', 's3'];
|
||||
for (const type of connectorTypes) {
|
||||
try {
|
||||
const job = await this.platformApiService.proxy(
|
||||
'GET',
|
||||
`/jobs/${type}/${normalizedJobId}`,
|
||||
user,
|
||||
);
|
||||
return job;
|
||||
} catch (error) {
|
||||
// Continue to next connector type
|
||||
}
|
||||
}
|
||||
throw new HttpException(`Job ${normalizedJobId} not found in any connector type (jdbc, singer, s3)`, 404);
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract the pipeline ID (base UUID) from a job ID.
|
||||
* Job IDs have format "uuid-suffix" where suffix is the job index (e.g., "0", "1").
|
||||
@@ -728,26 +758,10 @@ export class PlatformApiController {
|
||||
);
|
||||
}
|
||||
|
||||
// ==================== Catalog ROUTES ====================
|
||||
|
||||
@Get('pipelines/catalog/tables')
|
||||
@ApiOperation({ summary: 'Get all tables available' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
async getAvailableTables(
|
||||
@User() user: RequestUser,
|
||||
@Query() query: Record<string, string>,
|
||||
) {
|
||||
return this.platformApiService.proxy(
|
||||
'GET',
|
||||
'/catalog/tables',
|
||||
user,
|
||||
undefined,
|
||||
query,
|
||||
);
|
||||
}
|
||||
// ==================== PIPELINE VALIDATION ====================
|
||||
|
||||
@Get('pipelines/catalog/schemas')
|
||||
@ApiOperation({ summary: 'Get all schemas available' })
|
||||
@ApiOperation({ summary: 'Get available schemas' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
async getAvailableSchemas(
|
||||
@User() user: RequestUser,
|
||||
@@ -755,7 +769,7 @@ export class PlatformApiController {
|
||||
) {
|
||||
return this.platformApiService.proxy(
|
||||
'GET',
|
||||
'/catalog/schemas',
|
||||
`/catalog/schemas`,
|
||||
user,
|
||||
undefined,
|
||||
query,
|
||||
@@ -763,18 +777,18 @@ export class PlatformApiController {
|
||||
}
|
||||
|
||||
@Post('pipelines/catalog/tables/validate')
|
||||
@ApiOperation({ summary: 'Validate tables and schemas' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
||||
@ApiOperation({ summary: 'Validate Table and Schema' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
async validateTableAndSchema(
|
||||
@Body() payload: ValidationTableDTO,
|
||||
@User() user: RequestUser,
|
||||
@Query() query: Record<string, string>,
|
||||
@Body() validateTablesDto: ValidateTablesDTO[]
|
||||
) {
|
||||
return this.platformApiService.proxy(
|
||||
'POST',
|
||||
'/catalog/tables/validate',
|
||||
`/catalog/tables/validate`,
|
||||
user,
|
||||
validateTablesDto,
|
||||
payload,
|
||||
query,
|
||||
);
|
||||
}
|
||||
@@ -834,6 +848,24 @@ export class PlatformApiController {
|
||||
);
|
||||
}
|
||||
|
||||
@Post('pipeline/:pipelineId/pipeline_run/:runId/cancel')
|
||||
@ApiOperation({ summary: 'Cancel a running pipeline run' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
||||
async cancelPipelineRun(
|
||||
@Param('pipelineId') pipelineId: string,
|
||||
@Param('runId') runId: string,
|
||||
@User() user: RequestUser,
|
||||
) {
|
||||
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
|
||||
const normalizedRunId = this.normalizePipelineId(runId);
|
||||
|
||||
return this.platformApiService.proxy(
|
||||
'POST',
|
||||
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`,
|
||||
user,
|
||||
);
|
||||
}
|
||||
|
||||
// ==================== JOBS - COLUMN EDITING ROUTES ====================
|
||||
|
||||
@Put('jobs/:jobId/input')
|
||||
@@ -933,41 +965,53 @@ export class PlatformApiController {
|
||||
);
|
||||
}
|
||||
|
||||
// ==================== JOBS - JDBC SYNC MODE ROUTES ====================
|
||||
|
||||
@Get('jobs/jdbc/:jobId')
|
||||
@ApiOperation({ summary: 'Get JDBC job details' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
async getJdbcJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
|
||||
// Normalize job ID for Platform API (replace - with _)
|
||||
const normalizedJobId = this.normalizeJobId(jobId);
|
||||
return this.platformApiService.proxy('GET', `/jobs/jdbc/${normalizedJobId}`, user);
|
||||
}
|
||||
|
||||
@Post('jobs/jdbc/:jobId/sync-mode')
|
||||
@ApiOperation({ summary: 'Update JDBC job sync mode' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
||||
async updateJdbcSyncMode(
|
||||
@Param('jobId') jobId: string,
|
||||
@Body() body: any,
|
||||
@Delete('pipelines/:pipelineId/inputs/:inputId')
|
||||
@ApiOperation({ summary: 'Mark a table as deleted and delete its associated job via platform-api' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
|
||||
async deleteTable(
|
||||
@Param('pipelineId') pipelineId: string,
|
||||
@Param('inputId') inputId: string,
|
||||
@Body() body: { table_name: string },
|
||||
@User() user: RequestUser,
|
||||
) {
|
||||
// Normalize job ID for Platform API (replace - with _)
|
||||
const normalizedJobId = this.normalizeJobId(jobId);
|
||||
const tableName = body.table_name;
|
||||
const info = {
|
||||
customer_id: user.customer_id,
|
||||
customer: user.customer_name,
|
||||
user_id: user.user_id,
|
||||
};
|
||||
|
||||
const result = await this.platformApiService.proxy(
|
||||
'POST',
|
||||
`/jobs/jdbc/${normalizedJobId}/sync-mode`,
|
||||
user,
|
||||
body,
|
||||
);
|
||||
this.logger.info('deleteTable: marking table as deleted', { inputId, tableName });
|
||||
const updatedInput: any = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info });
|
||||
this.logger.info('deleteTable: table marked as deleted', { inputId, tableName });
|
||||
|
||||
// Sync to DynamoDB (pass raw jobId for pipeline extraction)
|
||||
await this.syncJdbcSyncModeToDynamoDB(jobId, body, user);
|
||||
try {
|
||||
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
|
||||
this.logger.info('deleteTable: fetching pipeline from platform-api', { pipelineId, normalizedPipelineId });
|
||||
const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
|
||||
this.logger.info('deleteTable: pipeline fetched', { jobCount: platformPipeline?.jobs?.length });
|
||||
|
||||
return result;
|
||||
const job = platformPipeline?.jobs?.find((j: any) => j.input?.table_name === tableName);
|
||||
if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`);
|
||||
|
||||
this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id });
|
||||
await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user);
|
||||
this.logger.info('deleteTable: job deleted', { jobId: job.job_id });
|
||||
|
||||
return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at };
|
||||
} catch (error) {
|
||||
this.logger.error('deleteTable: platform-api delete failed, attempting rollback', { tableName, error: error.message });
|
||||
try {
|
||||
await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: tableName, info });
|
||||
} catch (rollbackError) {
|
||||
this.logger.error('deleteTable: rollback failed', { tableName, error: rollbackError.message });
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== JOBS - JDBC SYNC MODE ROUTES ====================
|
||||
|
||||
@Get('jobs/jdbc/configs/allowed_datatypes')
|
||||
@ApiOperation({ summary: 'Get allowed datatypes for JDBC' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
@@ -979,52 +1023,6 @@ export class PlatformApiController {
|
||||
);
|
||||
}
|
||||
|
||||
// ==================== JOBS - SINGER REPLICATION ROUTES ====================
|
||||
|
||||
@Get('jobs/singer/:jobId')
|
||||
@ApiOperation({ summary: 'Get Singer job details' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
async getSingerJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
|
||||
// Normalize job ID for Platform API (replace - with _)
|
||||
const normalizedJobId = this.normalizeJobId(jobId);
|
||||
return this.platformApiService.proxy('GET', `/jobs/singer/${normalizedJobId}`, user);
|
||||
}
|
||||
|
||||
@Post('jobs/singer/:jobId/sync-mode')
|
||||
@ApiOperation({ summary: 'Update Singer job sync mode' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
||||
async updateSingerSyncMode(
|
||||
@Param('jobId') jobId: string,
|
||||
@Body() body: any,
|
||||
@User() user: RequestUser,
|
||||
) {
|
||||
// Normalize job ID for Platform API (replace - with _)
|
||||
const normalizedJobId = this.normalizeJobId(jobId);
|
||||
|
||||
const result = await this.platformApiService.proxy(
|
||||
'POST',
|
||||
`/jobs/singer/${normalizedJobId}/sync-mode`,
|
||||
user,
|
||||
body,
|
||||
);
|
||||
|
||||
// Sync to DynamoDB (pass raw jobId for pipeline extraction)
|
||||
await this.syncSingerSyncModeToDynamoDB(jobId, body, user);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
// ==================== JOBS - S3 ROUTES ====================
|
||||
|
||||
@Get('jobs/s3/:jobId')
|
||||
@ApiOperation({ summary: 'Get S3 job details' })
|
||||
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
||||
async getS3Job(@Param('jobId') jobId: string, @User() user: RequestUser) {
|
||||
// Normalize job ID for Platform API (replace - with _)
|
||||
const normalizedJobId = this.normalizeJobId(jobId);
|
||||
return this.platformApiService.proxy('GET', `/jobs/s3/${normalizedJobId}`, user);
|
||||
}
|
||||
|
||||
// ==================== HEALTH ROUTE ====================
|
||||
|
||||
@Get('health')
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
import { ApiProperty } from "@nestjs/swagger";
|
||||
|
||||
export class ValidationTableDTO {
|
||||
@ApiProperty()
|
||||
tables: Array<{
|
||||
table_name: string;
|
||||
table_schema: string;
|
||||
}>
|
||||
}
|
||||
@@ -7,9 +7,11 @@ import { PlatformApiService } from './platform-api.service';
|
||||
import { ElasticsearchModule } from '../../services/elasticsearch';
|
||||
import { DynamoDBModule } from '../../services/dynamodb';
|
||||
import { CustomersModule } from '../customers/customers.module';
|
||||
import { CatalogModule } from '../catalog/catalog.module';
|
||||
import { InputsModule } from '../inputs/inputs.module';
|
||||
|
||||
@Module({
|
||||
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule],
|
||||
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, InputsModule],
|
||||
controllers: [PlatformApiController],
|
||||
providers: [PlatformApiService, DadosferaLogger],
|
||||
exports: [PlatformApiService],
|
||||
|
||||
@@ -89,6 +89,12 @@ export class PlatformApiService {
|
||||
|
||||
// Propagate non-2xx responses as HttpExceptions
|
||||
if (response.status >= 400) {
|
||||
this.logger.error('Platform API upstream error' + JSON.stringify({
|
||||
status: response.status,
|
||||
data: response.data,
|
||||
path,
|
||||
method: method.toUpperCase(),
|
||||
}));
|
||||
throw new HttpException(response.data, response.status);
|
||||
}
|
||||
|
||||
@@ -101,6 +107,8 @@ export class PlatformApiService {
|
||||
method: method.toUpperCase(),
|
||||
});
|
||||
|
||||
this.logger.error(error)
|
||||
|
||||
if (error instanceof HttpException) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
@@ -125,9 +125,13 @@ export class DynamoDBService {
|
||||
},
|
||||
});
|
||||
|
||||
const { Item } = await this.documentClient.send(getCommand);
|
||||
|
||||
return Item as InputDocument | null;
|
||||
try {
|
||||
const { Item } = await this.documentClient.send(getCommand);
|
||||
return Item as InputDocument | null;
|
||||
} catch (error) {
|
||||
this.logger.error('DynamoDB: findInput failed', { inputId, clientId, error: error.message });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async deleteInput(clientId: string, inputId: string): Promise<void> {
|
||||
@@ -203,7 +207,6 @@ export class DynamoDBService {
|
||||
updatedTable.reference_column = changes.reference_column;
|
||||
}
|
||||
}
|
||||
|
||||
tables[tableIndex] = updatedTable;
|
||||
|
||||
// Save updated document
|
||||
@@ -231,4 +234,5 @@ export class DynamoDBService {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -358,6 +358,81 @@ export class ElasticsearchService {
|
||||
}
|
||||
}
|
||||
|
||||
private getDataAssetIndex(customerName: string): string {
|
||||
return `${customerName}_data_assets_catalog`;
|
||||
}
|
||||
|
||||
async findDataAssetByTable(
|
||||
customerName: string,
|
||||
tableName: string,
|
||||
tableSchema: string,
|
||||
): Promise<{ id: string; nimbus_id: number | null; [key: string]: any } | null> {
|
||||
const index = this.getDataAssetIndex(customerName);
|
||||
|
||||
this.logger.info('Elasticsearch: Searching data asset', {
|
||||
index,
|
||||
tableName,
|
||||
tableSchema,
|
||||
});
|
||||
|
||||
try {
|
||||
const response = await this.client.post(`/${index}/_search`, {
|
||||
query: {
|
||||
bool: {
|
||||
must: [
|
||||
{ term: { 'table_name.keyword': tableName.toUpperCase() } },
|
||||
{ term: { 'table_schema.keyword': tableSchema.toUpperCase() } },
|
||||
],
|
||||
},
|
||||
},
|
||||
size: 1,
|
||||
});
|
||||
|
||||
const hits = response.data.hits?.hits || [];
|
||||
if (hits.length === 0) {
|
||||
this.logger.warn('Elasticsearch: Data asset not found', { tableName, tableSchema, index });
|
||||
return null;
|
||||
}
|
||||
|
||||
return { ...hits[0]._source, _es_id: hits[0]._id };
|
||||
} catch (error) {
|
||||
this.handleError('findDataAssetByTable', error, { tableName, tableSchema, index });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async updateDataAsset(
|
||||
customerName: string,
|
||||
assetId: string,
|
||||
updates: Record<string, any>,
|
||||
): Promise<any> {
|
||||
const index = this.getDataAssetIndex(customerName);
|
||||
|
||||
this.logger.info('Elasticsearch: Updating data asset', {
|
||||
index,
|
||||
assetId,
|
||||
fields: Object.keys(updates),
|
||||
});
|
||||
|
||||
try {
|
||||
const response = await this.client.post(
|
||||
`/${index}/_update/${assetId}`,
|
||||
{ doc: updates },
|
||||
{ params: { refresh: 'wait_for' } },
|
||||
);
|
||||
|
||||
this.logger.info('Elasticsearch: Data asset updated', {
|
||||
assetId,
|
||||
result: response.data.result,
|
||||
});
|
||||
|
||||
return response.data;
|
||||
} catch (error) {
|
||||
this.handleError('updateDataAsset', error, { assetId, index });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private handleError(
|
||||
operation: string,
|
||||
error: any,
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { NimbusService } from "./nimbus.service";
|
||||
import DadosferaLogger from "@dadosfera/dadosfera-logs";
|
||||
|
||||
@Module({
|
||||
providers: [NimbusService, DadosferaLogger],
|
||||
exports: [NimbusService],
|
||||
})
|
||||
export class NimbusServicesModule {}
|
||||
@@ -0,0 +1,56 @@
|
||||
import DadosferaLogger from "@dadosfera/dadosfera-logs";
|
||||
import { Inject, Injectable } from "@nestjs/common";
|
||||
import axios from "axios";
|
||||
|
||||
type TableUpdate = {
|
||||
table_schema: string;
|
||||
table_name: string;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class NimbusService {
|
||||
private logger: DadosferaLogger;
|
||||
|
||||
constructor(
|
||||
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
|
||||
) {
|
||||
this.logger = dadosferaLogger.logger;
|
||||
}
|
||||
|
||||
private buildUrl(customerName: string) {
|
||||
if (process.env.ENV === 'prd') {
|
||||
return `https://nimbus-${customerName}.dadosfera.ai`;
|
||||
}
|
||||
|
||||
return `https://nimbus-${customerName}.${process.env.ENV.replace(
|
||||
'local',
|
||||
'stg',
|
||||
)}.dadosfera.ai`;
|
||||
}
|
||||
|
||||
async renameTable(customerName: string, database: string, old: TableUpdate, update: TableUpdate) {
|
||||
const nimbusUrl = this.buildUrl(customerName);
|
||||
|
||||
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("Payload: " + JSON.stringify(update));
|
||||
const { data } = await axios.patch(nimbusUrl + path, {
|
||||
table_name: update.table_name,
|
||||
table_schema: update.table_schema
|
||||
})
|
||||
|
||||
return data;
|
||||
} catch (error) {
|
||||
this.logger.error(error);
|
||||
return {
|
||||
message: error.message,
|
||||
database,
|
||||
old,
|
||||
update
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user