maestro/src/modules/catalog/catalog.service.ts

import DadosferaLogger from '@dadosfera/dadosfera-logs/dist';
import {
  WriteService,
  ReadService,
  ProtoServices,
  Messages,
} from '@dadosfera/protospack-v2/dist/lib/Catalog';
import {
  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';

class CatalogService implements OnModuleInit {
  catalogReadService: ReadService.CatalogReadServices;
  catalogWriteService: WriteService.CatalogWriteServices;
  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<ReadService.CatalogReadServices>(
        ProtoServices.CatalogReadServices,
      );
    this.catalogWriteService =
      this.grpcClient.getService<WriteService.CatalogWriteServices>(
        ProtoServices.CatalogWriteServices,
      );
  }

  _getNimbusUrl(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');

    return lastValueFrom(
      this.catalogWriteService.CreateDataAsset(data, metadata),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 404,
      );
    });
  }

  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 getUsernamesAndRolesByCustomerId(customerId: string) {
    const resultUsers = await this.userService.findAllUsersByCustomerId(
      customerId,
    );
    const resultRoles = await this.roleService.roleSearch(
      {},
      { customer_id: customerId },
    );

    const customer_users = resultUsers.users.map((user) => {
      return {
        id: user.id,
        username: user.username,
      };
    });

    const customer_roles = resultRoles.roles.map((role) => {
      return {
        id: role.id,
        name: role.name,
      };
    });

    return [customer_users, customer_roles];
  }

  async getAllOpenSearchDataAssets(query, metadata, customer_id) {
    this.logger.info('CatalogService - getAllOpenSearchDashboard');

    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,
      ),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 404,
      );
    });

    const result = JSON.parse(data_assets);

    const [customer_users, customer_roles] =
      await this.getUsernamesAndRolesByCustomerId(customer_id);

    for (const asset of result.data_assets) {
      const condition =
        asset.owner && !asset.owner.includes('@') ? true : false;
      const result = condition
        ? await this.userService.findOneById(asset.owner).catch(() => null)
        : null;

      const owner = result !== null ? result.user.username : null;

      const roles = [];
      const users = [];

      for (const role of customer_roles) {
        for (const role_id of asset.roles) {
          if (role.id === role_id) roles.push(role);
        }
      }

      for (const user of customer_users) {
        for (const user_id of asset.users) {
          if (user.id === user_id) users.push(user);
        }
      }

      Object.assign(asset, {
        ...asset,
        owner,
        roles,
        users,
      });
    }

    return { data_assets: result.data_assets, total };
  }

  async getOneDataAsset({
    data_asset_id,
    data_asset_type,
    customer_id,
    metadata,
  }) {
    const { data_asset } = await lastValueFrom(
      this.catalogReadService.GetOneDataAsset(
        { id: data_asset_id, type: data_asset_type },
        metadata,
      ),
    );
    const opensearchTable = JSON.parse(data_asset);

    const result = opensearchTable.owner
      ? await this.userService
          .findOneById(opensearchTable.owner)
          .catch(() => null)
      : null;

    const owner = result !== null ? result.user.username : null;

    const [customer_users, customer_roles] =
      await this.getUsernamesAndRolesByCustomerId(customer_id);

    const users = [];
    const roles = [];

    for (const role of customer_roles) {
      for (const role_id of opensearchTable.roles) {
        if (role.id === role_id) roles.push(role);
      }
    }

    for (const user of customer_users) {
      for (const user_id of opensearchTable.users) {
        if (user.id === user_id) users.push(user);
      }
    }

    Object.assign(opensearchTable, {
      p_roles: opensearchTable.roles,
      p_users: opensearchTable.users,
      roles,
      users,
      owner,
    });

    return { data_asset: opensearchTable };
  }

  async updateOneDataAsset({ data_asset_id, customer_id, body, metadata }) {
    const { data_asset } = await lastValueFrom(
      this.catalogWriteService.UpdateDataAsset(
        { id: data_asset_id, changes: JSON.stringify(body) },
        metadata,
      ),
    );

    const opensearchTable = JSON.parse(data_asset);

    const result = opensearchTable.owner
      ? await this.userService
          .findOneById(opensearchTable.owner)
          .catch(() => null)
      : null;

    const owner = result !== null ? result.user.username : null;

    const [customer_users, customer_roles] =
      await this.getUsernamesAndRolesByCustomerId(customer_id);

    const users = [];
    const roles = [];

    for (const role of customer_roles) {
      for (const role_id of opensearchTable.roles) {
        if (role.id === role_id) roles.push(role);
      }
    }

    for (const user of customer_users) {
      for (const user_id of opensearchTable.users) {
        if (user.id === user_id) users.push(user);
      }
    }

    Object.assign(opensearchTable, {
      p_roles: opensearchTable.roles,
      p_users: opensearchTable.users,
      roles,
      users,
      owner,
    });

    return { data_asset: opensearchTable };
  }

  async getDataDocs(data_asset_id: string, metadata: Metadata) {
    const [data_asset_type, id] = data_asset_id.split('-');

    const { documentation } = await lastValueFrom(
      this.catalogReadService.GetDatasetDoc(
        { id, type: data_asset_type },
        metadata,
      ),
    );
    console.log(documentation);
    const docs = JSON.parse(documentation);
    return docs;
  }

  async getDatasetPreview(data_asset_id: string, metadata: Metadata) {
    const [data_asset_type, id] = data_asset_id.split('-');

    const { preview } = await lastValueFrom(
      this.catalogReadService.GetDatasetPreview(
        { id, type: data_asset_type },
        metadata,
      ),
    );
    const result = JSON.parse(preview);
    return result;
  }

  async getDatasetColumnsMetadata(data_asset_id: string, metadata: Metadata) {
    const [data_asset_type, id] = data_asset_id.split('-');

    const { columns_metadata } = await lastValueFrom(
      this.catalogReadService.GetDatasetColumnsMetadata(
        { id, type: data_asset_type },
        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;
  }
}

export { CatalogService };