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

import { Metadata } from '@grpc/grpc-js';
import {
  OnModuleInit,
  Inject,
  HttpException,
  HttpStatus,
} from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { ConnectorManager } from '@dadosfera/protospack-v2';
import { lastValueFrom } from 'rxjs';
import { ConnectorClientConfiguration } from './client.config';

export class ConnectorClientService implements OnModuleInit {
  private connectorServiceRead: ConnectorManager.ReadService.ConnectorManagerReadServices;
  private connectorServiceWrite: ConnectorManager.WriteService.ConnectorManagerWriteServices;
  constructor(
    @Inject(ConnectorClientConfiguration.name)
    private readonly grpcClient: ClientGrpc,
  ) {}

  onModuleInit() {
    this.connectorServiceWrite =
      this.grpcClient.getService<ConnectorManager.WriteService.ConnectorManagerWriteServices>(
        ConnectorManager.ProtoServices.ConnectorManagerWriteServices,
      );

    this.connectorServiceRead =
      this.grpcClient.getService<ConnectorManager.ReadService.ConnectorManagerReadServices>(
        ConnectorManager.ProtoServices.ConnectorManagerReadServices,
      );
  }

  async uploadConnector(uploadConnector) {
    const connector: ConnectorManager.Entities.ConnectorCreateRequest = {
      file: {
        buffer: uploadConnector.file.buffer,
        mimetypes: uploadConnector.file.mimetype,
      },
      connector: JSON.stringify(uploadConnector.connector),
    };
    const serviceBody: ConnectorManager.Messages.RegisterConnectorRequest = {
      connector,
    };

    return lastValueFrom(
      this.connectorServiceWrite.RegisterConnector(serviceBody),
    ).catch((err) => {
      console.log(err.details);
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async uploadFile(uploadFile) {
    const body: ConnectorManager.Messages.UploadFileRequest = {
      file: {
        buffer: uploadFile.file.buffer,
        mimetypes: uploadFile.file.mimetype,
      },
      name: uploadFile.name,
    };

    return lastValueFrom(this.connectorServiceWrite.UploadFile(body)).catch(
      (err) => {
        console.log(err.details);
        throw new HttpException(
          err.details,
          err.code === 6 ? HttpStatus.CONFLICT : 400,
        );
      },
    );
  }

  async uploadConnectorsFile(uploadFile) {
    const connectors = JSON.parse(uploadFile.file.buffer.toString('utf8'));

    const body = connectors.map((connector) => {
      return JSON.stringify(connector);
    });

    const request = {
      connectors: body,
    };

    return lastValueFrom(
      this.connectorServiceWrite.RegisterMultipleConnectorsWithoutImage(
        request,
      ),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async getAllConnectors(body) {
    const meta = new Metadata();
    meta.add('Language', body.language);
    return lastValueFrom(
      this.connectorServiceRead.GetAllConnectors(
        {
          search: body.search,
          filters: JSON.stringify(body.filters),
          language: body.language,
          page: body.page,
          size: body.size,
        },
        meta,
      ),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async getConnectorDetails(plugin: string, language: string) {
    const meta = new Metadata();
    meta.add('Language', language);
    return lastValueFrom(
      this.connectorServiceRead.GetConnectorDetails(
        {
          plugin,
        },
        meta,
      ),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async getConnector(plugin: string, language: string) {
    const meta = new Metadata();
    meta.add('Language', language);
    return lastValueFrom(
      this.connectorServiceRead.GetConnector(
        {
          plugin,
        },
        meta,
      ),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async getConnectorsTags() {
    return lastValueFrom(this.connectorServiceRead.GetConnectorsTags({})).catch(
      (err) => {
        throw new HttpException(
          err.details,
          err.code === 6 ? HttpStatus.CONFLICT : 400,
        );
      },
    );
  }

  async deleteConnector(plugin: string, version: string) {
    return lastValueFrom(
      this.connectorServiceWrite.DeleteConnector({
        version,
        plugin,
      }),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async updateConnector({ plugin, changes }) {
    return lastValueFrom(
      this.connectorServiceWrite.UpdateConnector({
        plugin,
        changes,
      }),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }
}