import { Body, Controller, Delete, Get, Inject, Param, Post, Put, Query, UseFilters, HttpCode, HttpStatus, Patch, HttpException, BadRequestException, CacheTTL, } from '@nestjs/common'; import { ApiCreatedResponse, ApiHeaders, ApiNoContentResponse, ApiOperation, ApiTags, } from '@nestjs/swagger'; import { AuthenticateCondition, RequireAllPermissions, } from 'src/decorators/authentication.decorator'; import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum'; import { PipelinesService } from './pipelines.service'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { Messages } from '@dadosfera/protospack-v2/dist/lib/PipelineV2'; import { RequestUser, User } from 'src/decorators/user.decorator'; import { PackTheMetadata } from 'src/utils/ PackTheMetadata'; import { PipelinesService as OldPipelineService } from 'src/modules/pipelines/pipelines.service'; import { ICompleteUploadCSVFile, ICreatePipelineCSVFile, ICreatePipelineV2Req, IPipelineV2, IInitUploadCSVFile, PipelineFindAllReq, } 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'; @ApiTags('PipelinesV2') @ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }]) @UseFilters(new GrpcToHttpExceptionFilter()) @Controller('pipelinesV2') @AuthenticateCondition((req, user) => { let action; switch (req.method) { case 'POST': action = 'CREATE'; break; case 'PUT': action = 'UPDATE'; break; case 'PATCH': action = 'UPDATE'; break; default: action = req.method; } return user.permissions.includes( PERMISSIONS_GROUPS.PIPELINE.permissions[action].seqid, ); }) export class PipelinesController { logger: DadosferaLogger; constructor( @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, private pipelinesClientService: PipelinesService, private oldPipelinesService: OldPipelineService, ) { this.logger = dadosferaLogger.logger; } @Get('monitoring-dashboard') async getMonitoringDashboard(@User() user: RequestUser) { this.logger.info('PipelinesController - getMonitoringDashboard', { user }); const metadata = PackTheMetadata(user); const response = await this.pipelinesClientService.getMonitoringDashboardUrl(metadata); return response; } @Post() @ApiCreatedResponse({ type: IPipelineV2 }) async create( @Language() language: LanguageEnum, @User() user: RequestUser, @Body() createPipelineDto: ICreatePipelineV2Req, ) { this.logger.info('PipelinesController - create', { user }); const { customer_id, customer_name, username, user_id } = user; const metadata = PackTheMetadata({ customer_id, customer_name, username, user_id, language, }); const response = await this.pipelinesClientService.create( createPipelineDto, metadata, ); return response; } @Get() async findAll( @User() user: RequestUser, @Language() language: LanguageEnum, @Query() data: PipelineFindAllReq, ) { this.logger.info('PipelinesController - findAll', { user }); const { customer_id, customer_name, user_id, username } = user; const metadata = PackTheMetadata({ customer_id, customer_name, user_id, username, language, }); const response = await this.pipelinesClientService.findAll(data, metadata); return { pipelines: [...response.pipelines], }; } @Get('/download-logs') async downloadLogs( @User() user: RequestUser, @Language() language: LanguageEnum, @Query('pipeline_run_id') pipeline_run_id: string, ) { this.logger.info('PipelinesController - downloadLogs', { user }); if (!pipeline_run_id) throw new BadRequestException('Missing pipeline_run_id'); const { customer_id, customer_name, user_id, username } = user; const metadata = PackTheMetadata({ customer_id, customer_name, user_id, username, language, }); const url = await this.pipelinesClientService.downloadLogs( pipeline_run_id, metadata, ); return { url, }; } @Get(':id/config') async getPipelineproperties( @Language() language: LanguageEnum, @User() user: RequestUser, @Param('id') id: string, ) { this.logger.info('PipelinesController - getPipelineproperties', { user }); const metadata = PackTheMetadata({ ...user, language: language }); return this.pipelinesClientService.findOneProperties(id, metadata); } @Get(':id/objects') async getPipelineObjects( @Language() language: LanguageEnum, @User() user: RequestUser, @Param('id') id: string, ) { this.logger.info('PipelinesController - getPipelineObjects', { user }); const metadata = PackTheMetadata({ ...user, language: language }); return this.pipelinesClientService.findOneObjects(id, metadata); } @Get(':id/status') async getPipelineStatus(@Body() body, @Param('id') id: string) { body.id = id; this.logger.info( process.env.DEV_URL + `/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`, { user: body.info.user_id, customer: body.info.customer, }, ); const response = await this.oldPipelinesService.getPipelineStatus(body); return response; } @Get('/:id') async findOne( @Language() language: LanguageEnum, @User() user: RequestUser, @Param('id') id: string, ): Promise { this.logger.info('PipelinesController - findOne', { user }); const { customer_id, customer_name, user_id, username } = user; const metadata = PackTheMetadata({ customer_id, customer_name, user_id, username, 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; }); return result; } @Patch('/:id') async update( @Language() language: LanguageEnum, @Body() updatePipelineDto, @Param('id') id: string, @User() user: RequestUser, ) { this.logger.info('PipelinesController - update', { user }); const { info } = updatePipelineDto; delete updatePipelineDto.info; const { customer_id, customer_name, user_id, username } = user; const metadata = PackTheMetadata({ customer_id, customer_name, user_id, username, language, }); const response = await this.pipelinesClientService.update( { ...updatePipelineDto, info, id, }, metadata, ); this.logger.info('PipelinesController - update: OK', { user }); return response; } @ApiInternalOnlyEndpoint() @Put('/:id') @ApiOperation({ deprecated: true, description: 'This method is deprecated. Please use PATCH instead', }) async updateDeprecated( @Language() language: LanguageEnum, @Body() updatePipelineDto, @Param('id') id: string, @User() user: RequestUser, ) { this.logger.info('PipelinesController - put', { user }); const response = await this.update(language, updatePipelineDto, id, user); this.logger.info('PipelinesController - put: OK', { user }); return response; } @Delete(':id') @ApiNoContentResponse() @HttpCode(HttpStatus.NO_CONTENT) async delete(@Param('id') id: string, @User() user: RequestUser) { this.logger.info('PipelinesController - delete', { user }); const metadata = PackTheMetadata({ customer_id: user.customer_id, customer_name: user.customer_name, user_id: user.user_id, }); await this.pipelinesClientService.remove({ id, metadata, user }); this.logger.info('PipelinesController - delete: OK'); } @ApiInternalOnlyEndpoint() @Post('/init-upload') @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE) async initUploadFile( @User() user: RequestUser, @Body() body: IInitUploadCSVFile, ) { this.logger.info('/upload - Init Upload File Route'); const metadata = PackTheMetadata({ ...user }); const { file_name, parts } = body; const { urls, upload_id } = await this.pipelinesClientService.initUploadFile( { name: file_name, parts: String(parts), }, metadata, ); if (!urls) { this.logger.info('pipeline/upload - Failed File pipeline'); throw new HttpException( 'Upload failed, try again in a few minutes, if the problem persists, contact support.', HttpStatus.INTERNAL_SERVER_ERROR, ); } return { urls: JSON.parse(urls), upload_id }; } @ApiInternalOnlyEndpoint() @Post('/complete-upload') @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE) async completeUploadFile( @User() user: RequestUser, @Body() body: ICompleteUploadCSVFile, ) { this.logger.info('/upload - Complete Upload File Route'); const metadata = PackTheMetadata({ ...user }); const { upload_id, parts, file_name } = body; return await this.pipelinesClientService.completeUploadFile( { upload_id, parts, file_name }, metadata, ); } @ApiInternalOnlyEndpoint() @Post('/file') @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE) async uploadedFile( @User() user: RequestUser, @Body() body: ICreatePipelineCSVFile, ) { this.logger.info('/file - Upload Connector Route'); const metadata = PackTheMetadata({ ...user }); const { name, sep, header, encoding, description, file_name, engine } = body; const file_format_params = { ...(sep && { sep }), ...(encoding && { encoding }), ...(header !== undefined ? { header } : {}), }; const source_prefix = `${user.customer_name}/${file_name}`; const upload_pipeline: ICreatePipelineV2Req = { connection_id: process.env.UPLOAD_FILE_AGENT_CONNECTION, connector_name: 'Amazon S3', connector_plugin: 'aws_s3', connector_version: '1.0.0', image_url: `https://assets.dadosfera.ai/images/connectors/${engine}.svg`, name, description, transformations_ids: [], tags: [], cron: '@once', type: 'upload', properties: { engine, source_bucket: process.env.BUCKET_CUSTOMER_CSV_ASSETS, source_prefix, file_format_params, }, input_id: undefined, }; const pipeline = await this.pipelinesClientService.create( upload_pipeline, metadata, ); return pipeline; } @ApiInternalOnlyEndpoint() @Post('start/:id') async activate(@Param('id') id: string, @Body() body) { const { info } = body; this.logger.info( process.env.DEV_URL + `/pipeline/start/${id} - ON START PIPELINE ROUTE`, { user: body.info.user_id, customer: body.info.customer, }, ); const response = await this.oldPipelinesService.runPipeline({ id, info }); return response; } }