maestro/src/modules/connection/client.service.ts

import { OnModuleInit, Inject } from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { ConnectionManager } from '@dadosfera/protospack-v2';
import { lastValueFrom } from 'rxjs';
import {
  ConnectionApiConnection,
  ConnectionRes,
  ConnectionsRes,
  ConnectionToCatalogDto,
  CreateConnectionDto,
  UpdateConnectionDto,
} from './dtos/connection';
import { ConnectionClientConfiguration } from './client.config';
import {
  GetConnectionDetailsResponse,
  INTERNAL_GetAllConnectionsResponse,
} from '@dadosfera/protospack-v2/dist/lib/ConnectionManager/interfaces/messages';
import { UsersService } from '../users/users.service';
import { ConnectorClientService } from '../connector/client.service';
import {
  ConnectionsApiNetworkConfig,
  NetworkConfigDto,
} from '../network-config/dto/network-config';
import DadosferaLogger from '@dadosfera/dadosfera-logs/dist';

export class ConnectionClientService implements OnModuleInit {
  private connectionServiceRead: ConnectionManager.ReadService.ConnectionManagerReadServices;
  private connectionServiceWrite: ConnectionManager.WriteService.ConnectionManagerWriteServices;
  logger: DadosferaLogger;

  constructor(
    @Inject(ConnectionClientConfiguration.name)
    private readonly grpcClient: ClientGrpc,
    private usersService: UsersService,
    private connectorService: ConnectorClientService,
    private dadosferaLogger: DadosferaLogger,
  ) {
    this.logger = dadosferaLogger.logger;
  }

  onModuleInit() {
    this.connectionServiceWrite =
      this.grpcClient.getService<ConnectionManager.WriteService.ConnectionManagerWriteServices>(
        ConnectionManager.ProtoServices.ConnectionManagerWriteServices,
      );

    this.connectionServiceRead =
      this.grpcClient.getService<ConnectionManager.ReadService.ConnectionManagerReadServices>(
        ConnectionManager.ProtoServices.ConnectionManagerReadServices,
      );
  }

  async createConnection(
    body: CreateConnectionDto,
    metadata,
  ): Promise<ConnectionRes> {
    const {
      connector_id,
      description,
      image_url,
      name,
      network_config_id,
      properties,
      type,
      connector_name,
      plugin,
    } = body;

    return lastValueFrom(
      this.connectionServiceWrite.CreateConnection(
        {
          connector_id,
          connector_name,
          description,
          image_url,
          name,
          network_config_id,
          properties: JSON.stringify(properties),
          type,
          plugin,
        },
        metadata,
      ),
    );
  }

  async updateConnection(
    id: string,
    body: UpdateConnectionDto,
    metadata,
  ): Promise<ConnectionRes> {
    const { description, name, network_config_id, properties } = body;

    return lastValueFrom(
      this.connectionServiceWrite.UpdateConnection(
        {
          id,
          description,
          name,
          network_config_id,
          properties: JSON.stringify(properties),
        },
        metadata,
      ),
    );
  }

  async deleteConnection({ body, metadata }): Promise<ConnectionRes> {
    const { id } = body;

    return lastValueFrom(
      this.connectionServiceWrite.DeleteConnection(
        {
          id,
        },
        metadata,
      ),
    );
  }

  async getAllConnections({ body, metadata }): Promise<ConnectionsRes> {
    const { search, filters, size, page } = body;

    return lastValueFrom(
      this.connectionServiceRead.GetAllConnection(
        { search, filters: JSON.stringify(filters), size, page },
        metadata,
      ),
    );
  }

  async getConnectionDetails({
    body,
    metadata,
  }): Promise<GetConnectionDetailsResponse> {
    const id = body.connection_id || body.id;

    return lastValueFrom(
      this.connectionServiceRead.GetConnectionDetails({ id }, metadata),
    );
  }

  async INTERNAL_getAllConnections({
    customer_name,
  }: {
    customer_name: string;
  }): Promise<INTERNAL_GetAllConnectionsResponse> {
    return lastValueFrom(
      this.connectionServiceRead.INTERNAL_GetAllConnections({ customer_name }),
    );
  }

  async catalogExistingConnection(data: {
    connection?: ConnectionToCatalogDto;
    networkConfig?: NetworkConfigDto;
  }) {
    return lastValueFrom(
      this.connectionServiceRead.INTERNAL_CatalogExistingConnection({
        data_str: JSON.stringify(data),
      }),
    );
  }

  async synchronizeConnections(apply: boolean) {
    const connectors = await this.connectorService
      .getAllConnectors({
        search: undefined,
        filters: undefined,
        language: 'pt-br',
        page: 1,
        size: 500,
      })
      .then(({ connectors: connectorsStr }) =>
        JSON.parse(connectorsStr).map((c) => c),
      );
    const { customers } = await this.usersService.findAllCustomers();

    const customer_connections: {
      customer_name: string;
      customer_id: string;
      productConnections: Record<string, any>[];
      platformConnections: ConnectionApiConnection[];
      productNetworkConfigs: Record<string, any>[];
      platformNetworkConfigs: ConnectionsApiNetworkConfig[];
    }[] = [];
    for (const customer of customers) {
      const { name: customer_name, id: customer_id } = customer;

      const customer_connection = await this.INTERNAL_getAllConnections({
        customer_name,
      }).then((res) => ({
        customer_name,
        customer_id,
        platformConnections: res.platformConnections
          ? JSON.parse(res.platformConnections)
          : [],
        productConnections: res.productConnections
          ? JSON.parse(res.productConnections)
          : [],
        productNetworkConfigs: res.productNetworkConfigs
          ? JSON.parse(res.productNetworkConfigs)
          : [],
        platformNetworkConfigs: res.platformNetworkConfigs
          ? JSON.parse(res.platformNetworkConfigs)
          : [],
      }));
      customer_connections.push(customer_connection);
    }
    this.logger.info(
      'DEBUG_ConnectionsMigrations - ' + JSON.stringify(customer_connections),
    );

    const migrations: {
      customer_name: string;
      status: any[];
      productConnectionsMissing: ConnectionApiConnection[];
      productNetworkConfigsMissing: ConnectionsApiNetworkConfig[];
    }[] = [];
    for (const customer_connection of customer_connections) {
      const productConnectionsMissing: ConnectionApiConnection[] = [];
      const productNetworkConfigsMissing: ConnectionsApiNetworkConfig[] = [];
      const status: any[] = [];
      let updatePromises: Promise<any>[] = [];

      this.logger.info(
        'DEBUG_ConnectionsMigrations - customer_connections ' +
          JSON.stringify(customer_connection),
      );
      const {
        customer_name,
        productConnections,
        platformConnections,
        productNetworkConfigs,
        platformNetworkConfigs,
        customer_id,
      } = customer_connection;

      platformConnections?.forEach((plc) => {
        const productConnection = productConnections.find(
          (prc) => prc.id === plc.config_id,
        );
        if (!productConnection) productConnectionsMissing.push(plc);
      });
      platformNetworkConfigs?.forEach((platNetConfig) => {
        const productNetworkConfig = productNetworkConfigs.find(
          (prodNetConfig) => prodNetConfig.id === platNetConfig.config_id,
        );
        if (!productNetworkConfig)
          productNetworkConfigsMissing.push(platNetConfig);
      });
      if (apply) {
        for (const network_config of productNetworkConfigsMissing) {
          const networkConfigToCatalog = {
            id: network_config.config_id,
            created_at: network_config.created_at,
            updated_at: network_config.updated_at,
            customer_name: customer_name,
            customer_id: customer_id,
            user_id: '',
            network_type_id: '',
            network_type_name: network_config.type,
            type: network_config.type,
            plugin: network_config.plugin,
            name: network_config.name,
            description: network_config.description,
            properties: undefined,
            keywords: undefined,
            tags: undefined,
          };

          updatePromises.push(
            this.catalogExistingConnection({
              networkConfig: networkConfigToCatalog,
            }),
          );
        }

        this.logger.info(
          'DEBUG_ConnectionsMigrations - Creating missing network configs for customer' +
            customer_name +
            ' - ' +
            JSON.stringify(customer_connection),
        );
        const updateStatuses = await Promise.all(updatePromises);
        status.push(...updateStatuses);
        updatePromises = [];
        for (const connection of productConnectionsMissing) {
          const connector = connectors.find(
            (c) => c._source.plugin === connection.plugin,
          );
          const connectionToCatalog = {
            id: connection.config_id,
            created_at: connection.created_at,
            updated_at: new Date().toISOString(),
            customer_id: customer_id,
            user_id: '',
            plugin: connection.plugin,
            name: connection.name,
            description: connection.description,
            image_url: connector._source.image,
            network_config_id: connection.network_config_id,
            type: connection.type,
            connector_id: connector._id,
            connector_name: connector._source.name,
            in_use: 1,
            username: '',
            customer_name,
          };

          updatePromises.push(
            this.catalogExistingConnection({ connection: connectionToCatalog }),
          );
        }
        this.logger.info(
          'DEBUG_ConnectionsMigrations - Creating missing connections for customer' +
            ' - ' +
            customer_name +
            JSON.stringify(customer_connection),
        );
        const updateStatuses_2 = await Promise.all(updatePromises);
        status.push(...updateStatuses_2);
      }

      this.logger.info(
        'DEBUG_ConnectionsMigrations - migrations for customer ' +
          customer_name +
          ' - ' +
          JSON.stringify(customer_connection),
      );
      migrations.push({
        customer_name,
        status,
        productConnectionsMissing,
        productNetworkConfigsMissing,
      });
    }

    return { migrations };
  }
}