import DadosferaLogger from '@dadosfera/dadosfera-logs/dist'; import { WriteService, ReadService, ProtoServices, Messages, } from '@dadosfera/protospack-v2/dist/lib/Catalog'; import { Messages as PlatformInterfaceMessages, WriteService as PlatformInterfaceWriteService, ProtoServices as PlatformInterfacesProtoServices, } from '@dadosfera/protospack-v2/dist/lib/PlatformInterfaces'; import { BadRequestException, HttpException, HttpStatus, Inject, OnModuleInit, } from '@nestjs/common'; import { ClientGrpc } from '@nestjs/microservices'; import axios from 'axios'; import { lastValueFrom } from 'rxjs'; import { CatalogClientConfiguration } from './catalog-client'; import { UsersService } from '../users/users.service'; import { RolesService } from '../roles/roles.service'; import { Metadata } from '@grpc/grpc-js'; import { PackTheMetadata } from 'src/utils/PackTheMetadata'; import { AssetReporter, BatchRemoveRlsRulesRequest, CreateDataDocsDTO, IUpdateDataRequest, TriggerCatalogReq, } from './dtos'; import { AddRlsRuleRequest, GetNimbusDashboardsRequest, GetRlsRulesRequest, PiiMetadata, } from '@dadosfera/protospack-v2/dist/lib/Catalog/interfaces/messages'; import { TypeParser } from 'src/utils/FileParser/parser-types'; import { ParserBuilder } from 'src/utils/FileParser/parser.builder'; class CatalogService implements OnModuleInit { catalogReadService: ReadService.CatalogReadServices; catalogWriteService: WriteService.CatalogWriteServices; platformWriteService: PlatformInterfaceWriteService.PlatformInterfacesWriteServices; logger: any; constructor( @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, @Inject(CatalogClientConfiguration.name) private readonly grpcClient: ClientGrpc, private readonly userService: UsersService, private readonly roleService: RolesService, ) { this.logger = dadosferaLogger.logger; } onModuleInit() { this.catalogReadService = this.grpcClient.getService( ProtoServices.CatalogReadServices, ); this.catalogWriteService = this.grpcClient.getService( ProtoServices.CatalogWriteServices, ); this.platformWriteService = this.grpcClient.getService( PlatformInterfacesProtoServices.PlatformInterfacesWriteServices, ); } _getNimbusUrl(body) { this.logger.debug(`Body: ${JSON.stringify(body)}`); const customer = body.info.customer.toLowerCase(); if (process.env.ENV === 'prd') { return `https://nimbus-${customer}.dadosfera.ai`; } return `https://nimbus-${customer}.${process.env.ENV.replace( 'local', 'stg', )}.dadosfera.ai`; } async getPiiReporter(metadata: Metadata, type: TypeParser) { this.logger.info('getPiiReporter: ' + type); try { const { data } = await lastValueFrom( this.catalogWriteService.GetPiiReporter({}, metadata), ); this.logger.info('Finish grpc call'); const parser = ParserBuilder.build(type); this.logger.info('parser file to: ' + type); const file = await parser.parse(data); this.logger.info('finish parser'); const mimeTypes: Record = { csv: 'text/csv', html: 'text/html', pdf: 'application/pdf', }; const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); const filename = `relatorio-pii-${timestamp}.${type}`; return { file, filename: filename, type: mimeTypes[type], }; } catch (error) { this.logger.error(error.message); throw error; } } async createDataAsset(data: Messages.CreateDataAssetRequest, metadata) { this.logger.info('CatalogService - Manage Data assets permissions'); if (!data.embed) data.embed = undefined; return lastValueFrom( this.catalogWriteService.CreateDataAsset(data, metadata), ); } async managePermissions(data: Messages.ManagePermissionRequest, metadata) { this.logger.info('CatalogService - Manage Data assets permissions'); return lastValueFrom( this.catalogWriteService.ManagePermission(data, metadata), ).catch((err) => { throw new HttpException( err.details, err.code === 6 ? HttpStatus.CONFLICT : 404, ); }); } async revokePermissions(data: Messages.RevokePermissionRequest, metadata) { this.logger.info('CatalogService - Manage Data assets permissions'); return lastValueFrom( this.catalogWriteService.RevokePermission(data, metadata), ).catch((err) => { throw new HttpException( err.details, err.code === 6 ? HttpStatus.CONFLICT : 404, ); }); } async commentOnDataAsset(data: Messages.MakeACommentRequest, metadata) { this.logger.info('CatalogService - Manage Data assets permissions'); return lastValueFrom( this.catalogWriteService.MakeAComment(data, metadata), ).catch((err) => { throw new HttpException( err.details, err.code === 6 ? HttpStatus.CONFLICT : 404, ); }); } async deleteComment(data: Messages.UpdateACommentRequest, metadata) { this.logger.info('CatalogService - Manage Data assets permissions'); return lastValueFrom( this.catalogWriteService.UpdateAComment(data, metadata), ).catch((err) => { throw new HttpException( err.details, err.code === 6 ? HttpStatus.CONFLICT : 404, ); }); } async deleteDataAsset(data: Messages.DeleteDataAssetRequest, metadata) { this.logger.info('CatalogService - Manage Data assets permissions'); return lastValueFrom( this.catalogWriteService.DeleteDataAsset(data, metadata), ).catch((err) => { throw new HttpException( err.details, err.code === 6 ? HttpStatus.CONFLICT : 404, ); }); } async getUserRolesIds(userId: string) { const result = await this.userService.findOneById(userId).catch(() => null); if (result) { return result.user.roles.map((role) => role.id); } return []; } async searchDataAssets( query: Record, metadata: Metadata, customer_id: string, ) { this.logger.info('CatalogService - searchDataAssets', { query }); const { search, page, size, sort_by, order, ...filters } = query; this.logger.debug('Extracted filters:', { filters }); if ( filters.manually !== undefined && filters.manually !== null && filters.manually !== '' ) { filters.manually = Number(filters.manually); // 1 ou 0 } else { delete filters.manually; } if (filters.owner) { const { users: customer_users } = await this.userService.findAllUsersByCustomerId(customer_id); this.logger.info('Available users in database count:', { count: customer_users.length, }); this.logger.info('First 5 users:', { users: customer_users .slice(0, 5) .map((u) => ({ id: u.id, email: u.email, name: u.name })), }); const ownerValues = Array.isArray(filters.owner) ? filters.owner : typeof filters.owner === 'string' && filters.owner.includes(',') ? filters.owner.split(',').map((o: string) => o.trim()) : [filters.owner]; this.logger.info('Owner values to convert:', { ownerValues, ownerFiltersOriginal: filters.owner, }); const ownerIds = ownerValues .map((ownerValue: string) => { const normalizedOwner = ownerValue.replace(/\s/g, '+'); const user = customer_users.find((u) => { const isIdMatch = u.id === ownerValue; const isEmailMatch = u.email === ownerValue || u.email === normalizedOwner; const isNameMatch = u.name === ownerValue || u.name === normalizedOwner; this.logger.info('Comparing:', { userId: u.id, userEmail: u.email, userName: u.name, filterValue: ownerValue, normalizedFilter: normalizedOwner, idMatch: isIdMatch, emailMatch: isEmailMatch, nameMatch: isNameMatch, }); return isIdMatch || isEmailMatch || isNameMatch; }); this.logger.info('Looking for owner result:', { ownerValue, found: !!user, userId: user?.id, }); return user?.id || ownerValue; }) .filter((id: string) => id); if (ownerIds.length > 0) { filters.owner = ownerIds; } } const { data_assets, total } = await lastValueFrom( this.catalogReadService.GetAllDataAssets( { search, page, size, sort_by, order, filters: JSON.stringify(filters), }, metadata, ), ); console.log('MAESTRO RECEBEU RESPOSTA DO PI-FACTORY'); const result = JSON.parse(data_assets); const response = await this.getAssetsUsersAndRoles( result.data_assets, customer_id, ); return { data_assets: response, total }; } async downloadAssets( query: Record, metadata: Metadata, customer_id: string, ) { const data = await this.searchDataAssets(query, metadata, customer_id); const formatData = data.data_assets.map((asset) => ({ id: asset.id, display_name: asset.display_name, data_asset_type: asset.data_asset_type, created_at: asset.created_at, tags: '[' + asset.tags.join(', ') + ']', })); const parser = ParserBuilder.build('csv'); const file = await parser.parse(formatData); const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); const filename = `dadosfera_assets_${timestamp}.csv`; return { file, filename, }; } async getOneDataAsset(data: { id: string; customer_id: string; metadata: Metadata; }) { const { customer_id, id, metadata } = data; const { data_asset } = await lastValueFrom( this.catalogReadService.GetOneDataAsset( { id, type: undefined }, metadata, ), ); let asset = JSON.parse(data_asset); asset = { ...asset, p_roles: asset.roles, p_users: asset.users, }; asset = await this.getAssetsUsersAndRoles([asset], customer_id); return { data_asset: asset[0] }; } async getOneDataAssetByPipelineAndObject(data: { customer_id: string; pipeline: string; object: string; metadata: Metadata; }) { const { customer_id, metadata, pipeline, object } = data; const { data_asset } = await lastValueFrom( this.catalogReadService.GetOneDataAssetByPipelineAndObject( { pipeline, object }, metadata, ), ); let asset = JSON.parse(data_asset); asset = { ...asset, p_roles: asset.roles, p_users: asset.users, }; asset = await this.getAssetsUsersAndRoles([asset], customer_id); return { data_asset: asset[0] }; } async updateOneDataAsset(data: { data_asset_id: string; customer_id: string; body: IUpdateDataRequest; metadata: Metadata; }) { const { body, customer_id, data_asset_id, metadata } = data; const { data_asset } = await lastValueFrom( this.catalogWriteService.UpdateDataAsset( { id: data_asset_id, changes: JSON.stringify(body) }, metadata, ), ); let asset = JSON.parse(data_asset); asset = { ...asset, p_roles: asset.roles, p_users: asset.users, }; asset = await this.getAssetsUsersAndRoles([asset], customer_id); return { data_asset: asset[0] }; } async getDataDocs(id: string, assetType: string, metadata: Metadata) { const { documentation } = await lastValueFrom( this.catalogReadService.GetDatasetDoc({ id }, metadata), ); const docs = JSON.parse(documentation); return docs; } async getDatasetPreview(id: string, metadata: Metadata) { const { preview } = await lastValueFrom( this.catalogReadService.GetDatasetPreview( { id, type: undefined }, metadata, ), ); const result = JSON.parse(preview); return result; } async getDatasetColumnsMetadata(id: string, metadata: Metadata) { const { columns_metadata } = await lastValueFrom( this.catalogReadService.GetDatasetColumnsMetadata( { id, type: undefined }, metadata, ), ); const result = JSON.parse(columns_metadata); return result; } async createDataDocs(body: CreateDataDocsDTO, metadata: Metadata) { if (body.asset_type === 'table' || body.asset_type === 'view') { return this.createDataDocsViaNimbus(body); } return this.createDataDocsViaGrpc(body, metadata); } private async createDataDocsViaNimbus(body: CreateDataDocsDTO) { this.logger.info('Creating data docs via Nimbus for table/view'); const nimbusUrl = this._getNimbusUrl(body); const { data } = await axios.post( `${nimbusUrl}/api/catalog/data-docs/`, body, ); return data; } private async createDataDocsViaGrpc(body: CreateDataDocsDTO, metadata: Metadata) { this.logger.info('Creating data docs via gRPC for other asset types'); try { const response: any = await lastValueFrom( this.catalogWriteService.UpdateDataAssetDoc( { id: body.table_id, docs: body.docs, }, metadata, ), ); return response; } catch (error) { this.logger.error('Error creating data asset docs:', error); throw new HttpException( 'Failed to create data asset documentation', HttpStatus.INTERNAL_SERVER_ERROR, ); } } async findAllTags(data, metadata) { this.logger.info('CatalogService - findAllCustomerTags'); const response = await lastValueFrom( this.catalogReadService.GetCustomerTags(data, metadata), ) .then((res) => { this.logger.info('Done'); return res; }) .catch((err) => { this.logger.error(err.message); throw new Error(err); }); return response; } async findSchemas(metadata: Metadata) { this.logger.info('CatalogService - findSchemas'); try { const response = await lastValueFrom( this.catalogReadService.GetSchemas({}, metadata), ); return response; } catch (error) { this.logger.error('Error fetching schemas:', error); throw error; } } async getAssetsUsersAndRoles(data_assets: Array, customer_id: string) { const { users: customer_users } = await this.userService.findAllUsersByCustomerId(customer_id); const { roles: customer_roles } = await this.roleService.roleSearch( {}, { customer_id }, ); return data_assets.map((data_asset) => { const owner = customer_users.find( (u) => u.id === data_asset.owner, )?.email; const roles = []; const users = []; const data_asset_roles = data_asset?.roles || []; for (const role_id of data_asset_roles) { const role = customer_roles.find((r) => r.id === role_id); if (role) roles.push({ id: role.id, name: role.name }); } const data_asset_users = data_asset?.users || []; for (const user_id of data_asset_users) { const user = customer_users.find((r) => r.id === user_id); if (user) users.push({ id: user.id, email: user.email }); } return { ...data_asset, roles, users, owner, } as typeof data_asset; }); } async triggerCatalog(data: TriggerCatalogReq, metadata: Metadata) { const { session } = await lastValueFrom( this.catalogWriteService.TriggerDatasetCataloging(data, metadata), ); return session; } async getDatasetCatalogTask(session: string, metadata: Metadata) { const res = await lastValueFrom( this.catalogReadService.GetDatasetCatalogTask({ session }, metadata), ); return res; } async addRlsRule(data: AddRlsRuleRequest, metadata: Metadata) { const res = await lastValueFrom( this.catalogWriteService.AddRlsRule(data, metadata), ); return res; } async removeRlsRule(id: number, metadata: Metadata) { const res = await lastValueFrom( this.catalogWriteService.RemoveRlsRule({ id }, metadata), ); return res; } async batchRemoveRlsRule( query: BatchRemoveRlsRulesRequest, metadata: Metadata, ) { const { id_rls, nimbus_dashboard_id } = query; if (id_rls && nimbus_dashboard_id) { throw new BadRequestException( "You can't delete using both parameters. Choose either 'id_rls' or 'nimbus_dashboard_id'", ); } if (id_rls) { await lastValueFrom( this.catalogWriteService.RemoveRlsRulesByRlsId({ id_rls }, metadata), ); } else if (nimbus_dashboard_id) { await lastValueFrom( this.catalogWriteService.RemoveRlsRulesByDashboardId( { nimbus_dashboard_id: parseInt(nimbus_dashboard_id) }, metadata, ), ); } return 'OK'; } async getRlsRules(data: GetRlsRulesRequest, metadata: Metadata) { const res = await lastValueFrom( this.catalogReadService.GetRlsRules(data, metadata), ); return res.rls_rules; } async getOneRlsRule(id: number, metadata: Metadata) { const res = await lastValueFrom( this.catalogReadService.GetOneRlsRule({ id }, metadata), ); return res.rls_rule; } async getNimbusDashboards( data: GetNimbusDashboardsRequest, metadata: Metadata, ) { const res = await lastValueFrom( this.catalogReadService.GetNimbusDashboards(data, metadata), ); return res.dashboards; } async createTableMetadata(body: any): Promise { const nimbusUrl = this._getNimbusUrl(body); this.logger.info(`Nimbus URL: ${nimbusUrl}`, { ...body.logMetadata }); const endpoint = `${nimbusUrl}/api/catalog/table-metadata/`; this.logger.info( `Creating table metadata for table ${body.table_metadata.table_name}`, { ...body.logMetadata }, ); this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata }); this.logger.debug(`Payload: ${JSON.stringify(body.table_metadata)}`, { ...body.logMetadata, }); try { const { data, status } = await axios.post(endpoint, { ...body.table_metadata, }); this.logger.info( `Table metadata created successfully with status ${status} for table ${body.table_metadata.table_name}`, { ...body.logMetadata }, ); return data.id; } catch (error) { this.logger.error( `Failed to create table metadata for table ${body.table_metadata.table_name} failed with status ${ error.response?.status } because of ${JSON.stringify(error.response?.data) || error.message}`, { ...body.logMetadata }, ); throw new Error(error.response?.data?.message || error.message); } } async createColumnMetadata(body: any): Promise { const nimbusUrl = this._getNimbusUrl(body); this.logger.info(`Nimbus URL: ${nimbusUrl}`, body.logMetadata); const endpoint = `${nimbusUrl}/api/catalog/column-metadata/`; try { this.logger.info( `Creating column metadata for table ${body.column_metadata.table_name}`, { ...body.logMetadata }, ); this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata }); this.logger.debug( `Payload: ${JSON.stringify(body.column_metadata)}`, { ...body.logMetadata }, ); const { data, status } = await axios.post( endpoint, body.column_metadata, ); this.logger.info( `Column metadata created successfully with status ${status} for table ${body.column_metadata.table_name}`, { ...body.logMetadata }, ); return data.map((column) => column.id); } catch (error) { this.logger.error( `Failed to create column metadata failed with status for table ${body.column_metadata.table_name} ${ error.response?.status } because of ${error.response?.data || error.message}`, { ...body.logMetadata }, ); throw new Error(error.response?.data?.message || error.message); } } async createDataPreview(body: any): Promise { const nimbusUrl = this._getNimbusUrl(body); this.logger.info(`Nimbus URL: ${nimbusUrl}`, { ...body.logMetadata }); const endpoint = `${nimbusUrl}/api/catalog/data-preview/`; this.logger.info( `Creating data preview for table ${body.data_preview.table_name}`, { ...body.logMetadata }, ); this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata }); this.logger.debug( `Payload: ${JSON.stringify(body.data_preview)}`, { ...body.logMetadata }, ); try { const { data, status } = await axios.post(endpoint, body.data_preview); this.logger.info( `Data preview created successfully with status ${status} for table ${body.data_preview.table_name}`, { ...body.logMetadata }, ); return data.id; } catch (error) { this.logger.error( `Failed to create data preview for table ${body.data_preview.table_name} failed with status ${ error.response?.status } because of ${error.response?.data || error.message}`, { ...body.logMetadata }, ); throw new Error(error.response?.data?.message || error.message); } } async renameTableOnNimbus( nimbusUrl: string, nimbusId: number, changes: { table_name?: string; table_schema?: string; display_name?: string }, ): Promise { 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 { 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 { 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'); const customer_name = customer_name_raw?.[0]?.toString(); if (!customer_name) { throw new BadRequestException('Customer name not found in metadata'); } const res = await lastValueFrom( this.platformWriteService.CatalogDataAssets( { data_assets: [ { data_asset_id: table_metadata_id.toString(), customer_name: customer_name, data_asset_type: 'dataset', }, ], }, metadata, ), ); return res; } } export { CatalogService };