maestro/src/modules/platform-api/platform-api.controller.spec.ts

// These service modules pull in gRPC client-config modules that read
// process.env at load time; mock them (hoisted before imports) so the spec
// needs no runtime env. Each mock severs an entire import subtree and still
// provides a class usable as a DI token.
jest.mock('../customers/customers.service', () => ({ CustomersService: class {} }));
jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} }));
jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));

import { Test, TestingModule } from '@nestjs/testing';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';

import { PlatformApiController } from './platform-api.controller';
import { PlatformApiService } from './platform-api.service';
import { ElasticsearchService } from '../../services/elasticsearch';
import { DynamoDBService } from '../../services/dynamodb';
import { CustomersService } from '../customers/customers.service';
import { CatalogService } from '../catalog/catalog.service';
import { InputsService } from '../inputs/inputs.service';
import { PipelinesService } from '../pipelinesV2/pipelines.service';

const logger = {
  info: (...args) => args,
  error: (...args) => args,
};

const mockUser: any = {
  customer_id: 'c1',
  customer_name: 'cust',
  user_id: 'u1',
};

describe('PlatformApiController - deleteTable', () => {
  let controller: PlatformApiController;
  let platformApiService: { proxy: jest.Mock };
  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };

  beforeEach(async () => {
    platformApiService = { proxy: jest.fn() };
    inputsService = {
      markTableDeleted: jest.fn(),
      unmarkTableDeleted: jest.fn(),
    };

    const module: TestingModule = await Test.createTestingModule({
      controllers: [PlatformApiController],
      providers: [
        { provide: PlatformApiService, useValue: platformApiService },
        { provide: ElasticsearchService, useValue: {} },
        { provide: DynamoDBService, useValue: {} },
        { provide: CustomersService, useValue: {} },
        { provide: CatalogService, useValue: {} },
        { provide: InputsService, useValue: inputsService },
        { provide: PipelinesService, useValue: {} },
        { provide: DadosferaLogger, useValue: { logger } },
      ],
    }).compile();

    controller = module.get<PlatformApiController>(PlatformApiController);
  });

  it('CDC removal reconfigures the connector', async () => {
    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
    platformApiService.proxy.mockImplementation((method: string, path: string) => {
      if (method === 'GET') {
        return Promise.resolve({
          jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } }],
        });
      }
      return Promise.resolve({});
    });

    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);

    expect(platformApiService.proxy).toHaveBeenCalledWith(
      'DELETE',
      '/pipeline/pid/jobs',
      mockUser,
      { job_ids: ['p_0'], delete_snowflake_tables: false },
    );
    const deletedViaJobsRoute = platformApiService.proxy.mock.calls.some(
      ([method, path]: any[]) => method === 'DELETE' && path === '/jobs/p_0',
    );
    expect(deletedViaJobsRoute).toBe(false);
  });

  it('batch removal unchanged', async () => {
    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
    platformApiService.proxy.mockImplementation((method: string, path: string) => {
      if (method === 'GET') {
        return Promise.resolve({
          jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 'pedidos' } }],
        });
      }
      return Promise.resolve({});
    });

    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);

    expect(platformApiService.proxy).toHaveBeenCalledWith(
      'DELETE',
      '/jobs/p_0',
      mockUser,
    );
    const reconfiguredConnector = platformApiService.proxy.mock.calls.some(
      ([method, path]: any[]) => method === 'DELETE' && path === '/pipeline/pid/jobs',
    );
    expect(reconfiguredConnector).toBe(false);
  });
});

describe('PlatformApiController - addTable', () => {
  let controller: PlatformApiController;
  let inputsService: { addCdcTable: jest.Mock; removeCdcTable: jest.Mock };
  let pipelinesClientService: { addCdcJobs: jest.Mock };

  const body = {
    table_name: 'orders',
    table_schema: 'public',
    primary_keys: ['id'],
    destinations: {
      raw: { table_schema: 'raw', table_name: 'orders' },
      qualify: { table_schema: 'qualify', table_name: 'orders' },
    },
  };

  beforeEach(async () => {
    inputsService = {
      addCdcTable: jest.fn(),
      removeCdcTable: jest.fn(),
    };
    pipelinesClientService = {
      addCdcJobs: jest.fn(),
    };

    const module: TestingModule = await Test.createTestingModule({
      controllers: [PlatformApiController],
      providers: [
        { provide: PlatformApiService, useValue: { proxy: jest.fn() } },
        { provide: ElasticsearchService, useValue: {} },
        { provide: DynamoDBService, useValue: {} },
        { provide: CustomersService, useValue: {} },
        { provide: CatalogService, useValue: {} },
        { provide: InputsService, useValue: inputsService },
        { provide: PipelinesService, useValue: pipelinesClientService },
        { provide: DadosferaLogger, useValue: { logger } },
      ],
    }).compile();

    controller = module.get<PlatformApiController>(PlatformApiController);
  });

  it('addTable: DynamoDB append then platform AddJobs, returns job_ids', async () => {
    inputsService.addCdcTable.mockResolvedValue({ input: {} });
    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });

    const result = await controller.addTable('pid', 'iid', body, mockUser);

    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
      id: 'iid',
      table: {
        table_schema: 'public',
        table_name: 'orders',
        primary_keys: ['id'],
        name: 'orders',
        iceberg_table_name: undefined,
        iceberg_qualify_table_name: undefined,
        columns: [],
        column_exclude_list: [],
      },
      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
    });
    expect(pipelinesClientService.addCdcJobs).toHaveBeenCalledWith({
      pipeline_id: 'pid',
      input_id: 'iid',
      tables: [{
        table_schema: 'public',
        table_name: 'orders',
        primary_keys: ['id'],
        destinations: body.destinations,
      }],
      info: { customer_id: 'c1', user_id: 'u1', customer: 'cust' },
    });
    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
    expect(inputsService.removeCdcTable).not.toHaveBeenCalled();

    const addCdcTableOrder = inputsService.addCdcTable.mock.invocationCallOrder[0];
    const addCdcJobsOrder = pipelinesClientService.addCdcJobs.mock.invocationCallOrder[0];
    expect(addCdcTableOrder).toBeLessThan(addCdcJobsOrder);
  });

  it('addTable: carries iceberg_table_name on the added table through to AddCdcTable', async () => {
    inputsService.addCdcTable.mockResolvedValue({ input: {} });
    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] });

    const icebergBody = { ...body, iceberg_table_name: 'cdc_raw.public__orders' };

    await controller.addTable('pid', 'iid', icebergBody, mockUser);

    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
      id: 'iid',
      table: {
        table_schema: 'public',
        table_name: 'orders',
        primary_keys: ['id'],
        name: 'orders',
        iceberg_table_name: 'cdc_raw.public__orders',
        iceberg_qualify_table_name: undefined,
        columns: [],
        column_exclude_list: [],
      },
      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
    });
  });

  it('addTable: carries columns on the added table through to AddCdcTable', async () => {
    inputsService.addCdcTable.mockResolvedValue({ input: {} });
    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_4'], skipped: [] });

    const columnsBody = {
      ...body,
      columns: [
        { name: 'id', type: 'int', is_primary_key: true },
        { name: 'descr', type: 'varchar(255)', is_primary_key: false },
      ],
    };

    await controller.addTable('pid', 'iid', columnsBody, mockUser);

    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
      id: 'iid',
      table: {
        table_schema: 'public',
        table_name: 'orders',
        primary_keys: ['id'],
        name: 'orders',
        iceberg_table_name: undefined,
        iceberg_qualify_table_name: undefined,
        columns: [
          { name: 'id', type: 'int', is_primary_key: true },
          { name: 'descr', type: 'varchar(255)', is_primary_key: false },
        ],
        column_exclude_list: [],
      },
      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
    });
  });

  it('addTable: rolls back the DynamoDB row when AddJobs fails', async () => {
    inputsService.addCdcTable.mockResolvedValue({ input: {} });
    pipelinesClientService.addCdcJobs.mockRejectedValue(new Error('platform down'));
    inputsService.removeCdcTable.mockResolvedValue({});

    await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down');

    expect(inputsService.removeCdcTable).toHaveBeenCalledWith({
      id: 'iid',
      table_name: 'orders',
      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
    });
  });
});


describe('PlatformApiController - deleteTables (batch)', () => {
  let controller: PlatformApiController;
  let platformApiService: { proxy: jest.Mock };
  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };

  beforeEach(async () => {
    platformApiService = { proxy: jest.fn() };
    inputsService = {
      markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }),
      unmarkTableDeleted: jest.fn().mockResolvedValue({}),
    };

    const module: TestingModule = await Test.createTestingModule({
      controllers: [PlatformApiController],
      providers: [
        { provide: PlatformApiService, useValue: platformApiService },
        { provide: ElasticsearchService, useValue: {} },
        { provide: DynamoDBService, useValue: {} },
        { provide: CustomersService, useValue: {} },
        { provide: CatalogService, useValue: {} },
        { provide: InputsService, useValue: inputsService },
        { provide: PipelinesService, useValue: {} },
        { provide: DadosferaLogger, useValue: { logger } },
      ],
    }).compile();

    controller = module.get<PlatformApiController>(PlatformApiController);
  });

  const pipelineWithJobs = () => ({
    jobs: [
      { job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } },
      { job_id: 'p_1', input: { connector: 'cdc', table_name: 'clientes' } },
      { job_id: 'p_2', input: { connector: 'cdc', table_name: 'produtos' } },
    ],
  });

  it('removes N tables in ONE platform call (connectors reconfigured once)', async () => {
    platformApiService.proxy.mockImplementation((method: string) =>
      Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}),
    );

    await controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'produtos'] }, mockUser);

    // one mark per table
    expect(inputsService.markTableDeleted).toHaveBeenCalledTimes(2);

    // exactly one DELETE to the batch endpoint, with BOTH job_ids
    const deleteCalls = platformApiService.proxy.mock.calls.filter(
      ([m, p]: any[]) => m === 'DELETE' && p === '/pipeline/pid/jobs',
    );
    expect(deleteCalls).toHaveLength(1);
    expect(deleteCalls[0][3]).toEqual({
      job_ids: ['p_0', 'p_2'],
      delete_snowflake_tables: false,
    });
    // never the per-job route
    const perJob = platformApiService.proxy.mock.calls.some(
      ([m, p]: any[]) => m === 'DELETE' && String(p).startsWith('/jobs/'),
    );
    expect(perJob).toBe(false);
  });

  it('rolls back only this call\'s marks when the platform delete fails', async () => {
    platformApiService.proxy.mockImplementation((method: string) => {
      if (method === 'GET') return Promise.resolve(pipelineWithJobs());
      return Promise.reject(new Error('platform boom'));
    });

    await expect(
      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser),
    ).rejects.toThrow();

    // both marks rolled back, nothing else
    expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(2);
    const unmarked = inputsService.unmarkTableDeleted.mock.calls.map((c: any[]) => c[0].table_name).sort();
    expect(unmarked).toEqual(['clientes', 'pedidos']);
  });

  it('rolls back the marks made so far if a later mark fails (atomic)', async () => {
    // second mark fails → first must be rolled back, no platform delete attempted
    inputsService.markTableDeleted
      .mockResolvedValueOnce({ is_deleted: true, deleted_at: 't' })
      .mockRejectedValueOnce(new Error('dynamo boom'));

    await expect(
      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser),
    ).rejects.toThrow();

    expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(1);
    expect(inputsService.unmarkTableDeleted.mock.calls[0][0].table_name).toBe('pedidos');
    // never reached the platform delete
    const attemptedDelete = platformApiService.proxy.mock.calls.some(([m]: any[]) => m === 'DELETE');
    expect(attemptedDelete).toBe(false);
  });

  it('404s when a requested table has no matching job', async () => {
    platformApiService.proxy.mockImplementation((method: string) =>
      Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}),
    );

    await expect(
      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'ghost'] }, mockUser),
    ).rejects.toThrow();
    // the successful mark (pedidos) must be rolled back
    expect(inputsService.unmarkTableDeleted).toHaveBeenCalled();
  });
});