maestro/src/modules/platform-api/pipeline-tables.service.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.
jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));

import { NotFoundException } from '@nestjs/common';
import { PipelineTablesService } from './pipeline-tables.service';

const logger = { info: jest.fn(), error: jest.fn() };
const user = { customer_id: 'c1', customer_name: 'cust', user_id: 'u1' } as never;
const info = { customer_id: 'c1', customer: 'cust', user_id: 'u1' };

type Mocks = {
  proxy: jest.Mock;
  markTableDeleted: jest.Mock;
  unmarkTableDeleted: jest.Mock;
  addCdcTable: jest.Mock;
  removeCdcTable: jest.Mock;
  addCdcJobs: jest.Mock;
};

function build(): { service: PipelineTablesService; m: Mocks } {
  const m: Mocks = {
    proxy: jest.fn(),
    markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }),
    unmarkTableDeleted: jest.fn().mockResolvedValue({}),
    addCdcTable: jest.fn().mockResolvedValue({ input: {} }),
    removeCdcTable: jest.fn().mockResolvedValue({ input: {} }),
    addCdcJobs: jest.fn(),
  };
  type Deps = ConstructorParameters<typeof PipelineTablesService>;
  const service = new PipelineTablesService(
    { proxy: m.proxy } as unknown as Deps[0],
    {
      markTableDeleted: m.markTableDeleted,
      unmarkTableDeleted: m.unmarkTableDeleted,
      addCdcTable: m.addCdcTable,
      removeCdcTable: m.removeCdcTable,
    } as unknown as Deps[1],
    { addCdcJobs: m.addCdcJobs } as unknown as Deps[2],
    { logger } as unknown as Deps[3],
  );
  return { service, m };
}

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

const deleteCalls = (proxy: jest.Mock) =>
  proxy.mock.calls.filter(([method]) => method === 'DELETE');

describe('PipelineTablesService.removeTable', () => {
  afterEach(() => jest.clearAllMocks());

  it.each(['cdc', 'jdbc'])(
    'removes a %s table through the single pipeline jobs route',
    async (connector) => {
      const { service, m } = build();
      m.proxy.mockImplementation((method: string) =>
        Promise.resolve(method === 'GET' ? pipelineWithJobs(connector) : {}),
      );

      const result = await service.removeTable('pi-d', 'iid', 'pedidos', user);

      expect(m.markTableDeleted).toHaveBeenCalledWith({ input_id: 'iid', table_name: 'pedidos', info });
      expect(m.proxy).toHaveBeenCalledWith('GET', '/pipeline/pi_d', user);
      expect(deleteCalls(m.proxy)).toEqual([
        ['DELETE', '/pipeline/pi_d/jobs', user, { job_ids: ['p_0'], delete_snowflake_tables: false }],
      ]);
      expect(result).toEqual({ name: 'pedidos', is_deleted: true, deleted_at: 't' });
      expect(m.unmarkTableDeleted).not.toHaveBeenCalled();
    },
  );

  it('rolls back the mark when the platform delete fails', async () => {
    const { service, m } = build();
    m.proxy.mockImplementation((method: string) =>
      method === 'GET' ? Promise.resolve(pipelineWithJobs('cdc')) : Promise.reject(new Error('platform boom')),
    );

    await expect(service.removeTable('pid', 'iid', 'pedidos', user)).rejects.toThrow('platform boom');
    expect(m.unmarkTableDeleted).toHaveBeenCalledWith({ input_id: 'iid', table_name: 'pedidos', info });
  });

  it('404s (and rolls back) when the table has no job', async () => {
    const { service, m } = build();
    m.proxy.mockResolvedValue(pipelineWithJobs('cdc'));

    await expect(service.removeTable('pid', 'iid', 'ghost', user)).rejects.toBeInstanceOf(NotFoundException);
    expect(deleteCalls(m.proxy)).toHaveLength(0);
    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(1);
  });
});

describe('PipelineTablesService.removeTables', () => {
  afterEach(() => jest.clearAllMocks());

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

    const result = await service.removeTables('pid', 'iid', ['pedidos', 'produtos'], user);

    expect(m.markTableDeleted).toHaveBeenCalledTimes(2);
    expect(deleteCalls(m.proxy)).toEqual([
      ['DELETE', '/pipeline/pid/jobs', user, { job_ids: ['p_0', 'p_2'], delete_snowflake_tables: false }],
    ]);
    expect(result).toEqual({ table_names: ['pedidos', 'produtos'], deleted: true });
  });

  it("rolls back only this call's marks when the platform delete fails", async () => {
    const { service, m } = build();
    m.proxy.mockImplementation((method: string) =>
      method === 'GET' ? Promise.resolve(pipelineWithJobs('cdc')) : Promise.reject(new Error('platform boom')),
    );

    await expect(service.removeTables('pid', 'iid', ['pedidos', 'clientes'], user)).rejects.toThrow();

    const unmarked = m.unmarkTableDeleted.mock.calls.map(([arg]) => arg.table_name).sort();
    expect(unmarked).toEqual(['clientes', 'pedidos']);
  });

  it('rolls back the marks made so far if a later mark fails (atomic)', async () => {
    const { service, m } = build();
    m.markTableDeleted
      .mockResolvedValueOnce({ is_deleted: true, deleted_at: 't' })
      .mockRejectedValueOnce(new Error('dynamo boom'));

    await expect(service.removeTables('pid', 'iid', ['pedidos', 'clientes'], user)).rejects.toThrow('dynamo boom');

    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(1);
    expect(m.unmarkTableDeleted.mock.calls[0][0].table_name).toBe('pedidos');
    expect(m.proxy).not.toHaveBeenCalled();
  });

  it('404s when a requested table has no matching job and rolls back the marks', async () => {
    const { service, m } = build();
    m.proxy.mockResolvedValue(pipelineWithJobs('cdc'));

    await expect(service.removeTables('pid', 'iid', ['pedidos', 'ghost'], user)).rejects.toBeInstanceOf(NotFoundException);
    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(2);
  });

  it('keeps going when a rollback step itself fails', async () => {
    const { service, m } = build();
    m.proxy.mockImplementation((method: string) =>
      method === 'GET' ? Promise.resolve(pipelineWithJobs('cdc')) : Promise.reject(new Error('platform boom')),
    );
    m.unmarkTableDeleted.mockRejectedValueOnce(new Error('unmark boom'));

    await expect(service.removeTables('pid', 'iid', ['pedidos', 'clientes'], user)).rejects.toThrow('platform boom');
    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(2);
  });
});

describe('PipelineTablesService.addTable', () => {
  afterEach(() => jest.clearAllMocks());

  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' },
    },
  };

  const storedTable = (extra: Record<string, unknown> = {}) => ({
    table_schema: 'public',
    table_name: 'orders',
    name: 'orders',
    primary_keys: ['id'],
    iceberg_table_name: undefined,
    iceberg_qualify_table_name: undefined,
    columns: [],
    column_exclude_list: [],
    ...extra,
  });

  it('appends to DynamoDB, then dispatches the platform jobs', async () => {
    const { service, m } = build();
    m.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });

    const result = await service.addTable('pid', 'iid', body, user);

    expect(m.addCdcTable).toHaveBeenCalledWith({ client_id: 'c1', id: 'iid', table: storedTable(), info });
    expect(m.addCdcJobs).toHaveBeenCalledWith({
      pipeline_id: 'pid',
      input_id: 'iid',
      tables: [{ table_schema: 'public', table_name: 'orders', primary_keys: ['id'], destinations: body.destinations }],
      info,
    });
    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
    expect(m.removeCdcTable).not.toHaveBeenCalled();
    expect(m.addCdcTable.mock.invocationCallOrder[0]).toBeLessThan(m.addCdcJobs.mock.invocationCallOrder[0]);
  });

  it('carries the iceberg names and columns through to the stored CdcTable', async () => {
    const { service, m } = build();
    m.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] });
    const columns = [
      { name: 'id', type: 'int', is_primary_key: true },
      { name: 'descr', type: 'varchar(255)', is_primary_key: false },
    ];

    await service.addTable('pid', 'iid', {
      ...body,
      iceberg_table_name: 'cdc_raw.public__orders',
      iceberg_qualify_table_name: 'orders_dedup',
      columns,
      column_exclude_list: ['descr'],
    }, user);

    expect(m.addCdcTable).toHaveBeenCalledWith({
      client_id: 'c1',
      id: 'iid',
      table: storedTable({
        iceberg_table_name: 'cdc_raw.public__orders',
        iceberg_qualify_table_name: 'orders_dedup',
        columns,
        column_exclude_list: ['descr'],
      }),
      info,
    });
  });

  it('rolls back the DynamoDB row when AddJobs fails', async () => {
    const { service, m } = build();
    m.addCdcJobs.mockRejectedValue(new Error('grpc boom'));

    await expect(service.addTable('pid', 'iid', body, user)).rejects.toThrow('grpc boom');
    expect(m.removeCdcTable).toHaveBeenCalledWith({ client_id: 'c1', id: 'iid', table_name: 'orders', info });
  });
});