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 { BatchRemoveRlsRulesRequest, IUpdateDataRequest, TriggerCatalogReq, } from './dtos'; import { AddRlsRuleRequest, GetNimbusDashboardsRequest, GetRlsRulesRequest, } from '@dadosfera/protospack-v2/dist/lib/Catalog/interfaces/messages'; 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 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); const roles_ids = result.user.roles.map((role) => role.id); return roles_ids; } async searchDataAssets( query: Record, metadata: Metadata, customer_id: string, ) { this.logger.info('CatalogService - searchDataAssets'); const { search, page, size, sort_by, order, ...filters } = query; const { data_assets, total } = await lastValueFrom( this.catalogReadService.GetAllDataAssets( { search, page, size, sort_by, order, filters: JSON.stringify(filters), }, metadata, ), ); const result = JSON.parse(data_assets); const response = await this.getAssetsUsersAndRoles( result.data_assets, customer_id, ); return { data_assets: response, total }; } 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, metadata: Metadata) { const { documentation } = await lastValueFrom( this.catalogReadService.GetDatasetDoc({ id, type: undefined }, metadata), ); console.log(documentation); 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) { const nimbusUrl = this._getNimbusUrl(body); const { data } = await axios.post( `${nimbusUrl}/api/catalog/data-docs/`, body, ); return data; } 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 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, )?.username; const roles = []; const users = []; 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 }); } 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, username: user.username }); } 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 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 };