import { Body, Controller, Delete, Get, Inject, Param, Post, Put, Query, UseFilters, HttpCode, HttpStatus, Patch, HttpException, BadRequestException, UseGuards, Res, } from '@nestjs/common'; import { ApiCreatedResponse, ApiHeaders, ApiNoContentResponse, ApiOperation, ApiTags, } from '@nestjs/swagger'; import { RequireAllPermissions, RequireSomePermission, } 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 { ICompleteUploadCSVFile, ICreatePipelineCSVFile, ICreatePipelineV2Req, 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 { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; 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 }]) @UseFilters(new GrpcToHttpExceptionFilter()) @Controller('pipelinesV2') export class PipelinesController { logger: DadosferaLogger; constructor( @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, private pipelinesClientService: PipelinesService, ) { this.logger = dadosferaLogger.logger; } @Get('monitoring-dashboard') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) 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() @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE) @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() @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.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') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) 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') @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW,PERMISSIONS_GROUPS.PIPELINE.permissions.GET) 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') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) 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') @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getPipelineStatus(@Body() body, @Param('id') id: string) { body.id = id; this.logger.info(`/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`, { user: body.info.user_id, customer: body.info.customer, }); const response = await this.pipelinesClientService.getPipelineStatus(body); return response; } @Get('/:id') @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET) 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 pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata); 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; } @Get("/:id/data-assets") @RequireSomePermission( PERMISSIONS_GROUPS.PIPELINE.permissions.GET, PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER, PERMISSIONS_GROUPS.CATALOG.permissions.GET ) async findAllDataAssetByPipeline( @Language() language: LanguageEnum, @Param('id') id: string, @User() user: RequestUser, @Query('object') object: string ) { const payload = { pipeline: id, object: object, }; this.logger.info(`GET pipelinesV2/:id/data-assets` + JSON.stringify(payload)); const result = await this.pipelinesClientService.findAllDataAssetByPipeline(payload, user); return result; } @Patch('/:id') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) 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; } @Patch('/:pipelineId/inputs/:id') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) @UseGuards(PipelineExecutionGuard) 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({ deprecated: true, description: 'This method is deprecated. Please use PATCH instead', }) @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) 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; } @Patch('/:id/upgrade') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) @HttpCode(HttpStatus.NO_CONTENT) async upgradeConnector( @Language() language: LanguageEnum, @Param('id') id: string, @User() user: RequestUser ) { this.logger.info('PipelinesController - upgrade connector'); const metadata = PackTheMetadata(user); await this.pipelinesClientService.upgrade(id, metadata); } @Delete(':id') @ApiNoContentResponse() @HttpCode(HttpStatus.NO_CONTENT) @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE) 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.IMPORT_FILES.permissions.VIEW) 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.IMPORT_FILES.permissions.VIEW) 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.IMPORT_FILES.permissions.VIEW) 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') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE) 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.pipelinesClientService.runPipeline({ id, info }); return response; } }