maestro/src/modules/connection-test/connection-test.service.ts

import { HttpException, HttpStatus, Inject, Injectable } from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { ConnectionTest } from '@dadosfera/protospack-v2';
import { lastValueFrom } from 'rxjs';
import {
  ConnectionTestCredentialsReq,
  ConnectionTestCredentialsRes,
  ConnectionTestListSchemasReq,
  ConnectionTestListSchemasRes,
  ConnectionTestListTablesReq,
  ConnectionTestListTablesRes,
  ConnectionTestPingReq,
  ConnectionTestPingRes,
  GetTableMetadataReq,
  GetTableMetadataRes,
  ValidateCdcPrerequisitesReq,
  ValidateCdcPrerequisitesRes,
  RefreshCatalogReq,
  RefreshCatalogRes,
  RefreshCatalogStatusReq,
} from './dto/connection-test';
import { ConnectionClientService } from '../connection/client.service';
import {
  CreateConnectionDto,
  DatabaseConnectionPropertiesDto,
} from '../connection/dtos/connection';
import { RequestUser } from 'src/decorators/user.decorator';
import { PackTheMetadata } from 'src/utils/PackTheMetadata';
import { ConnectionsApiService } from '../connections-api/connections-api.service';
import { PlatformApiService } from '../platform-api/platform-api.service';

@Injectable()
export class ConnectionTestService {
  private connectionTestReadClient: ConnectionTest.ReadService.ConnectionTestReadServices;
  constructor(
    @Inject('ConnectionTestGrpcClient') private readonly grpcClient: ClientGrpc,
    private connectionsService: ConnectionClientService,
    private connectionsApiService: ConnectionsApiService,
    private platformApiService: PlatformApiService,
  ) {
    this.connectionTestReadClient =
      grpcClient.getService<ConnectionTest.ReadService.ConnectionTestReadServices>(
        ConnectionTest.ProtoServices.ConnectionTestReadServices,
      );
  }

  async connectionTestPing(
    body: ConnectionTestPingReq,
    user: RequestUser,
  ): Promise<ConnectionTestPingRes> {
    const {
      host: endpoint,
      plugin,
      port,
      network_config_id,
      connection_id,
    } = body;
    const { user_id, customer_id, customer_name } = user;
    const metadata = PackTheMetadata({
      customer_name,
      customer_id,
      user_id,
    });

    if (connection_id) {
      const con = await this.connectionsService.getConnectionDetails({
        body,
        metadata,
      });
      const conPayload = JSON.parse(con.connection.properties);
      const { plugin, network_config_id } = con.connection;
      const { host, port, endpoint } = conPayload;

      return lastValueFrom(
        this.connectionTestReadClient.ConnectionTestPing({
          customer_name,
          endpoint: endpoint || host,
          network_config_id,
          plugin,
          port: port ? port.toString() : undefined,
          connection_id,
        }),
      );
    } else {
      return lastValueFrom(
        this.connectionTestReadClient.ConnectionTestPing({
          customer_name,
          endpoint,
          network_config_id,
          plugin,
          port: port ? port.toString() : undefined,
          connection_id,
        }),
      );
    }
  }
  async connectionTestCredentials(
    body: ConnectionTestCredentialsReq,
    user: RequestUser,
  ): Promise<ConnectionTestCredentialsRes> {
    const { plugin, ...connectionBody } = body;
    let { connection_id } = body;
    const { customer_id, customer_name, user_id, username } = user;
    const metadata = PackTheMetadata({
      customer_id,
      customer_name,
      user_id,
      username,
    });

    if (body.network_config_id == '') {
      body.network_config_id = undefined;
    }

    /********* CREATE CONNECTION *********/
    if (!connection_id) {
      const properties: DatabaseConnectionPropertiesDto = {
        ...connectionBody.properties,
        plugin,
        credentials_type: 'basic_auth',
      };
      const createConnectionBody: CreateConnectionDto = {
        ...body,
        properties,
      };
      const { connection } = await this.connectionsService.createConnection(
        createConnectionBody,
        metadata,
      );
      connection_id = connection.id;
    }
    /************************************/
    const { operation_result } = await lastValueFrom(
      this.connectionTestReadClient.ConnectionTestCredentials({
        connection_id,
        customer_name,
        plugin,
      }),
    ).catch(async (e) => {
      if (!body.connection_id) {
        await this.connectionsService.deleteConnection({
          body: { id: connection_id },
          metadata,
        });
      }
      throw e;
    });
    if (!operation_result && !body.connection_id) {
      await this.connectionsService.deleteConnection({
        body: { id: connection_id },
        metadata,
      });
      connection_id = null;
    }
    return { operation_result, connection_id };
  }
  async connectionTestListSchemas(
    body: ConnectionTestListSchemasReq,
    user: RequestUser,
  ): Promise<ConnectionTestListSchemasRes> {
    const result = await this.connectionsApiService.proxy(
      'GET',
      `/connection_catalog/${encodeURIComponent(body.connection_id)}/schemas`,
      user,
    );
    return {
      operation_result: true,
      schema_list: result.schemas.map((schema) => schema.schema_name),
    };
  }

  async connectionTestListTables(
    body: ConnectionTestListTablesReq,
    user: RequestUser,
  ): Promise<ConnectionTestListTablesRes> {
    const result = await this.connectionsApiService.proxy(
      'GET',
      `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
        `/schemas/${encodeURIComponent(body.schema)}/tables`,
      user,
    );
    const table_names: string[] = result.tables.map((table) => table.table_name);
    // CDC create needs the primary keys per table (used to build the deduped
    // Iceberg identifier-fields). The catalog-cache list-tables endpoint returns
    // only names, so fetch each table's columns from the cache and keep the ones
    // flagged is_primary_key. Reads hit the stored catalog snapshot (populated by
    // refresh-catalog), never the live connection.
    const tables = await Promise.all(
      table_names.map(async (table_name) => {
        const columns = await this.connectionsApiService.proxy(
          'GET',
          `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
            `/schemas/${encodeURIComponent(body.schema)}` +
            `/tables/${encodeURIComponent(table_name)}/columns`,
          user,
        );
        return {
          table_name,
          primary_keys: columns.columns
            .filter((column) => column.is_primary_key)
            .map((column) => column.column_name),
        };
      }),
    );
    return {
      operation_result: true,
      table_list: table_names,
      tables,
    };
  }

  async getTableMetadata(
    body: GetTableMetadataReq,
    user: RequestUser,
  ): Promise<GetTableMetadataRes> {
    const tables_metadata = await Promise.all(
      body.table_list.map(async (table_name) => {
        const result = await this.connectionsApiService.proxy(
          'GET',
          `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
            `/schemas/${encodeURIComponent(body.schema)}` +
            `/tables/${encodeURIComponent(table_name)}/columns`,
          user,
        );
        return {
          table_name,
          columns: result.columns.map((column) => ({
            name: column.column_name,
            type: column.data_type,
            is_primary_key: column.is_primary_key,
          })),
          references: [],
        };
      }),
    );
    return { operation_result: true, tables_metadata };
  }

  async refreshCatalog(
    body: RefreshCatalogReq,
    user: RequestUser,
  ): Promise<RefreshCatalogRes> {
    const task = await this.platformApiService.proxy(
      'POST',
      '/connection_test',
      user,
      {
        customer_id: user.customer_name,
        plugin: body.plugin,
        task: {
          task_type: 'refresh_catalog',
          connection: {
            provider: 'connection_manager',
            config_id: body.connection_id,
          },
        },
      },
    );

    if (!task.session_id || !task.date) {
      throw new HttpException(
        'Platform API did not return a catalog refresh task identifier',
        HttpStatus.BAD_GATEWAY,
      );
    }

    return {
      operation_result: true,
      status: 'PENDING',
      session_id: task.session_id,
      date: task.date,
    };
  }

  async refreshCatalogStatus(
    body: RefreshCatalogStatusReq,
    user: RequestUser,
  ): Promise<RefreshCatalogRes> {
    const result = await this.platformApiService.proxy(
      'POST',
      '/connection_test/status',
      user,
      {
        session_id: body.session_id,
        date: body.date,
      },
    );

    if (result.status === 'DONE') {
      await this.connectionsApiService.proxy(
        'PUT',
        `/connection_config/${encodeURIComponent(
          body.connection_id,
        )}/catalog_metadata`,
        user,
        {
          last_catalog_refresh_status: 'SUCCESS',
          last_catalog_connection_test_date: body.date,
          last_catalog_connection_test_session_id: body.session_id,
        },
      );
    } else if (result.status === 'ERROR' || result.status === 'EXPIRED') {
      throw new HttpException(
        `Catalog refresh finished with status ${result.status}`,
        HttpStatus.BAD_GATEWAY,
      );
    }

    return {
      operation_result: result.status === 'DONE',
      status: result.status,
      session_id: body.session_id,
      date: body.date,
    };
  }

  async validateCdcPrerequisites(
    body: ValidateCdcPrerequisitesReq,
    customer_name: string,
  ): Promise<ValidateCdcPrerequisitesRes> {
    const { plugin, connection_id } = body;
    return lastValueFrom(
      this.connectionTestReadClient.ValidateCdcPrerequisites({
        connection_id,
        customer_name,
        plugin,
      }),
    );
  }
}