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

import {
  OnModuleInit,
  Inject,
  HttpException,
  HttpStatus,
} 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';

export class ConnectionClientService implements OnModuleInit {
  private connectionServiceRead: ConnectionManager.ReadService.ConnectionManagerReadServices;
  private connectionServiceWrite: ConnectionManager.WriteService.ConnectionManagerWriteServices;
  constructor(
    @Inject(ConnectionClientConfiguration.name)
    private readonly grpcClient: ClientGrpc,
    private usersService: UsersService,
    private connectorService: ConnectorClientService,
  ) {}

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

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

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

    return lastValueFrom(
      this.connectionServiceWrite.DeleteConnection(
        {
          id,
        },
        metadata,
      ),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

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

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

    return lastValueFrom(
      this.connectionServiceRead.GetConnectionDetails({ id }, metadata),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

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

  async catalogExistingConnection(data: ConnectionToCatalogDto) {
    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.findaAllCustomers();
    const promises: Promise<{
      customer_name: string;
      customer_id: string;
      productConnections: Record<string, any>;
      platformConnections: ConnectionApiConnection[];
    }>[] = [];
    customers.forEach((c) => {
      const customer_name = c.name;
      const customer_id = c.id;
      promises.push(
        this.INTERNAL_getAllConnections({ customer_name }).then((res) => ({
          customer_name,
          customer_id,
          platformConnections: JSON.parse(res.platformConnections),
          productConnections: JSON.parse(res.productConnections),
        })),
      );
    });
    const customer_connections = await Promise.all(promises);

    const missingInProduct: ConnectionApiConnection[] = [];
    const migrations = customer_connections.map((cc) => {
      const {
        customer_name,
        platformConnections,
        productConnections,
        customer_id,
      } = cc;

      platformConnections.forEach((plc) => {
        const productConnection = productConnections.find(
          (prc) => prc.id === plc.config_id,
        );
        if (!productConnection) missingInProduct.push(plc);
      });

      const updatePromises: Promise<any>[] = [];
      for (const connection of missingInProduct) {
        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: 'auto',
          customer_name: customer_name,
        };
        updatePromises.push(
          this.catalogExistingConnection(connectionToCatalog),
        );
      }
      return { customer: { ...cc }, updatePromises, missingInProduct };
    });
    const customersStatus: any[] = [];
    for (const migration of migrations) {
      let status: any[] = [];
      if (apply) {
        status = await Promise.all(migration.updatePromises);
      }
      customersStatus.push({
        customer: migration.customer.customer_name,
        status,
        missingInProduct: migration.missingInProduct,
      });
    }
    return { customersStatus };
  }
}