diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..3238dc0 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,56 @@ +# maestro — development policy + +Maestro is the NestJS BFF between the Angular frontend and the gRPC services +(pi-factory, in-factory) / platform-api. These rules come from code review and +apply to every change; the same rules live in pi-factory and in-factory. + +## Layering +- Controllers are thin: decorators, body validation, one call into a service, + response shape. Orchestration (multi-step calls, rollbacks, platform-api or + gRPC round-trips) lives in a `*.service.ts`. Example: + `platform-api/pipeline-tables.service.ts`. +- One platform-api route per operation. If the platform already dispatches by + pipeline type (batch vs CDC), do not branch on the type here — call the + route that dispatches (e.g. `DELETE /pipeline/{id}/jobs` for table removal, + never `DELETE /jobs/{id}` from maestro). + +## Types +- No `any` / `as any`. gRPC calls take the protospack request type exactly + (`AddCdcTableRequest`, `MarkTableDeletedRequest`, ...); DTOs and interfaces + are explicit classes/interfaces. Use `Record` or `unknown` (with + narrowing) when a shape is genuinely open — never `any`. +- Protospack payloads are built in one mapper per entity (e.g. + `inputs/cdc-table.mapper.ts`), so a field that mirrors another + (`CdcTable.name` == `table_name`) is derived in exactly one place. + +## Pipeline type +- Decide CDC vs batch with `src/utils/cdc.ts` (`isCdcPlugin`, `isCdcJob`, + `isCdcPipeline`), never with inline `plugin.endsWith('_cdc')`, `!!x`, or + the absence of some other data (e.g. "no run history ⇒ editable"). +- The platform-api stamps `job.input.connector === 'cdc'` on CDC jobs; that + is the authoritative discriminator once a pipeline exists. + +## Dependencies +- `@dadosfera/protospack-v2` is consumed from CodeArtifact, pinned to an exact + version (`npm i @dadosfera/protospack-v2@ --save-exact`). A `file:` + / tarball reference is for local development only and must never be + committed. Note: `^3.40.0-beta.N` resolves to the stable `3.40.0` — + prereleases must be pinned exactly. + +## Commits and releases +- Deploys are cut by semantic-release with the eslint preset: the commit + title MUST start with `FIX:` (patch), `UPDATE:` or `FEAT:` (minor). A + lowercase `feat(scope): ...` merges without producing a version, so the + code never reaches stg/prd. +- Pushing to `beta` deploys stg; `main` deploys prd. + +## Tests +- Every service method with a rollback path has a spec covering the + happy path, the platform failure (rollback fires) and a rollback failure + (does not mask the original error). Run `npx jest ` for a folder, + `npx tsc --noEmit -p tsconfig.json` for types. +- Several gRPC client configs read `process.env` at import time, so the full + suite needs the service URLs set (any `0.0.0.0:` value works): + `DUC_URL=0.0.0.0:50051 INFACTORY_URL=0.0.0.0:50052 PIFACTORY_URL=0.0.0.0:50053 npx jest`. + A `Cannot read properties of undefined (reading 'startsWith')` at import is + this, not a broken test. diff --git a/src/guards/pipeline-execution.guard.spec.ts b/src/guards/pipeline-execution.guard.spec.ts index 0e2dbf9..cd063d1 100644 --- a/src/guards/pipeline-execution.guard.spec.ts +++ b/src/guards/pipeline-execution.guard.spec.ts @@ -3,12 +3,23 @@ import { PipelineExecutionGuard } from './pipeline-execution.guard'; const logger = { info: jest.fn(), error: jest.fn() }; -function buildGuard(proxyImpl: jest.Mock) { - const platformApiService: any = { proxy: proxyImpl }; - return new PipelineExecutionGuard( - { logger } as any, - platformApiService, +type Routes = { + pipeline?: unknown | Error; + runs?: unknown | Error; +}; + +// The guard makes up to two platform-api reads: GET /pipeline/{id} (type) +// and, for batch only, GET /pipeline/{id}/pipeline_run (last run). +function buildGuard(routes: Routes) { + const proxy = jest.fn((method: string, path: string) => { + const answer = path.endsWith('/pipeline_run') ? routes.runs : routes.pipeline; + return answer instanceof Error ? Promise.reject(answer) : Promise.resolve(answer); + }); + const guard = new PipelineExecutionGuard( + { logger } as unknown as ConstructorParameters[0], + { proxy } as unknown as ConstructorParameters[1], ); + return { guard, proxy }; } function contextWith(pipelineId = 'abc-123') { @@ -16,56 +27,96 @@ function contextWith(pipelineId = 'abc-123') { switchToHttp: () => ({ getRequest: () => ({ params: { pipelineId }, user: {} }), }), - } as any; + } as unknown as Parameters[0]; } +const cdcPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 't' } }] }; +const batchPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 't' } }] }; + describe('PipelineExecutionGuard', () => { afterEach(() => jest.clearAllMocks()); - it('allows the edit when the pipeline has no run history (empty array)', async () => { - const guard = buildGuard(jest.fn().mockResolvedValue([])); - await expect(guard.canActivate(contextWith())).resolves.toBe(true); + describe('CDC pipelines (explicit connector type)', () => { + it('allows the edit without consulting run history', async () => { + const { guard, proxy } = buildGuard({ pipeline: cdcPipeline }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + expect(proxy).toHaveBeenCalledTimes(1); + expect(proxy).toHaveBeenCalledWith('GET', '/pipeline/abc_123', {}); + }); + + it('is CDC when any job is a CDC job', async () => { + const mixed = { jobs: [...batchPipeline.jobs, ...cdcPipeline.jobs] }; + const { guard, proxy } = buildGuard({ pipeline: mixed }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + expect(proxy).toHaveBeenCalledTimes(1); + }); }); - it('allows the edit when the last run has no last_status', async () => { - const guard = buildGuard(jest.fn().mockResolvedValue([{}])); - await expect(guard.canActivate(contextWith())).resolves.toBe(true); + describe('batch pipelines', () => { + it('allows the edit when the pipeline never ran (empty run history)', async () => { + const { guard } = buildGuard({ pipeline: batchPipeline, runs: [] }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('allows the edit when the last run has no last_status', async () => { + const { guard } = buildGuard({ pipeline: batchPipeline, runs: [{}] }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('allows the edit when the pipeline is not running', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: [{ last_status: 'SUCCEEDED' }], + }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('blocks with the is-running message when the pipeline is running', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: [{ last_status: 'RUNNING' }], + }); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Pipeline is running, cannot update input now', + ); + }); + + it('treats a pipeline with no jobs as batch and checks its runs', async () => { + const { guard, proxy } = buildGuard({ pipeline: { jobs: [] }, runs: [] }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + expect(proxy).toHaveBeenCalledTimes(2); + }); + + it('does not double-wrap the is-running BadRequestException', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: [{ last_status: 'running' }], + }); + await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf( + BadRequestException, + ); + await expect(guard.canActivate(contextWith())).rejects.not.toThrow( + /Error checking pipeline status/, + ); + }); }); - it('allows the edit when the pipeline is not running', async () => { - const guard = buildGuard( - jest.fn().mockResolvedValue([{ last_status: 'SUCCEEDED' }]), - ); - await expect(guard.canActivate(contextWith())).resolves.toBe(true); - }); + describe('failures', () => { + it('fails closed when the pipeline read fails', async () => { + const { guard } = buildGuard({ pipeline: new Error('platform down') }); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Error checking pipeline status: platform down', + ); + }); - it('blocks with the is-running message when the pipeline is running', async () => { - const guard = buildGuard( - jest.fn().mockResolvedValue([{ last_status: 'RUNNING' }]), - ); - await expect(guard.canActivate(contextWith())).rejects.toThrow( - 'Pipeline is running, cannot update input now', - ); - }); - - it('wraps a genuine status-check failure (fail closed)', async () => { - const guard = buildGuard( - jest.fn().mockRejectedValue(new Error('platform down')), - ); - await expect(guard.canActivate(contextWith())).rejects.toThrow( - 'Error checking pipeline status: platform down', - ); - }); - - it('does not double-wrap the is-running BadRequestException', async () => { - const guard = buildGuard( - jest.fn().mockResolvedValue([{ last_status: 'running' }]), - ); - await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf( - BadRequestException, - ); - await expect(guard.canActivate(contextWith())).rejects.not.toThrow( - /Error checking pipeline status/, - ); + it('fails closed when the run-history read fails', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: new Error('runs down'), + }); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Error checking pipeline status: runs down', + ); + }); }); }); diff --git a/src/guards/pipeline-execution.guard.ts b/src/guards/pipeline-execution.guard.ts index 1b6ebcd..077d87d 100644 --- a/src/guards/pipeline-execution.guard.ts +++ b/src/guards/pipeline-execution.guard.ts @@ -16,6 +16,7 @@ import { import { PipelinesClientConfiguration } from 'src/modules/pipelinesV2/pipelines-client'; import { PlatformApiService } from 'src/modules/platform-api/platform-api.service'; import DadosferaLogger from '@dadosfera/dadosfera-logs'; +import { isCdcPipeline } from 'src/utils/cdc'; @Injectable() export class PipelineExecutionGuard implements CanActivate { @@ -39,7 +40,22 @@ export class PipelineExecutionGuard implements CanActivate { const user = request.user; const idRegex = /[^0-9a-zA-Z_$]+/g; const convertedId = pipelineId.replace(idRegex, '_'); - + + // Decide by the pipeline's explicit type (platform job + // `input.connector`), never by the absence of run history. + const pipeline = await this.platformApiService.proxy( + 'GET', + `/pipeline/${convertedId}`, + user, + ); + + if (isCdcPipeline(pipeline)) { + // CDC pipelines have no batch runs; the platform-api gates connector + // edits itself (require_pipeline_running on add/remove tables). + this.logger.info('PipelineExecutionGuard: CDC pipeline, no run to block on'); + return true; + } + const status = await this.platformApiService.proxy( 'GET', `/pipeline/${convertedId}/pipeline_run`, @@ -50,8 +66,7 @@ export class PipelineExecutionGuard implements CanActivate { this.logger.info('Pipeline current status response:' + JSON.stringify(currentStatus)); - // No run history (e.g. CDC pipelines never record batch runs) means - // nothing is executing — allow the edit rather than crash on .last_status. + // Batch pipeline that never ran yet: nothing can be executing. if (!currentStatus?.last_status) { return true; } @@ -59,9 +74,9 @@ export class PipelineExecutionGuard implements CanActivate { if (currentStatus.last_status.toLowerCase() === 'running') { this.logger.error('Pipeline is running, cannot update input now'); throw new BadRequestException('Pipeline is running, cannot update input now'); - } else { - return true; } + + return true; } catch (error) { // Preserve the deliberate is-running rejection; only wrap genuine // status-check failures (fail closed on those for a destructive gate). @@ -71,6 +86,5 @@ export class PipelineExecutionGuard implements CanActivate { this.logger.error('Error in PipelineExecutionGuard: ' + error.message); throw new BadRequestException('Error checking pipeline status: ' + error.message); } - } } diff --git a/src/modules/inputs/cdc-table.mapper.ts b/src/modules/inputs/cdc-table.mapper.ts new file mode 100644 index 0000000..491c8b8 --- /dev/null +++ b/src/modules/inputs/cdc-table.mapper.ts @@ -0,0 +1,38 @@ +import { + CdcColumn, + CdcTable, +} from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; + +/** What a caller knows about a CDC table before it is stored. */ +export interface CdcTableInput { + // Optional on the create DTO (CdcTableReq); the platform validates it. + table_schema?: string; + table_name: string; + primary_keys?: string[]; + iceberg_table_name?: string; + iceberg_qualify_table_name?: string; + columns?: CdcColumn[]; + column_exclude_list?: string[]; +} + +/** + * The only place maestro builds a protospack `CdcTable`. + * + * `name` is the identity key shared with batch `NewTable` (in-factory keys + * update/soft-delete on it) and the proto keeps it required, so it mirrors + * `table_name` here and nowhere else. Follow-up (protospack): make `name` + * optional and let in-factory be its sole writer (it already backfills + * `name ?? table_name`). + */ +export function toCdcTable(table: CdcTableInput): CdcTable { + return { + table_schema: table.table_schema, + table_name: table.table_name, + name: table.table_name, + primary_keys: table.primary_keys ?? [], + iceberg_table_name: table.iceberg_table_name, + iceberg_qualify_table_name: table.iceberg_qualify_table_name, + columns: table.columns ?? [], + column_exclude_list: table.column_exclude_list ?? [], + }; +} diff --git a/src/modules/inputs/dtos/input.model.ts b/src/modules/inputs/dtos/input.model.ts index f7d3aa4..0b2490c 100644 --- a/src/modules/inputs/dtos/input.model.ts +++ b/src/modules/inputs/dtos/input.model.ts @@ -78,8 +78,10 @@ export class CdcColumnReq { export class CdcTableReq { @ApiProperty() name: string; - @ApiPropertyOptional() - table_schema?: string; + // Source database/schema. Required: Debezium addresses tables as + // `schema.table`, a CDC table without it cannot be replicated. + @ApiProperty() + table_schema: string; @ApiPropertyOptional({ type: [String] }) primary_keys?: string[]; // Per-table raw Iceberg table name override (iceberg destination only). diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 80eef5b..5e622e7 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -13,17 +13,21 @@ import { objectCamelToSnake } from 'src/utils/CaseConverter'; import { IIdRequest, UpdateInputRequest } from './dtos/old_interfaces'; import { Input } from '@dadosfera/protospack-v2'; import { + AddCdcTableRequest, GetAvailableEntitiesRequest, InputCreateGenericRequest, InputCreateCdcRequest, InputCreateS3Request, InputNewCreateRequest, InputUpdateResponse, + MarkTableDeletedRequest, + RemoveCdcTableRequest, RollbackInputRequest, TestConnectionRequest, } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; import { CreateCdcInputReq, CreateInputReq } from './dtos/input.model'; +import { toCdcTable } from './cdc-table.mapper'; import { Metadata } from '@grpc/grpc-js'; @Injectable() @@ -192,16 +196,9 @@ export class InputsService { name: body.name, plugin: body.plugin, read_only: body.read_only ?? true, - tables: body.tables.map((t) => ({ - table_schema: t.table_schema, - table_name: t.name, - name: t.name, // canonical identity == table_name (in-factory also backfills) - primary_keys: t.primary_keys ?? [], - iceberg_table_name: t.iceberg_table_name, - iceberg_qualify_table_name: t.iceberg_qualify_table_name, - columns: t.columns ?? [], - column_exclude_list: t.column_exclude_list ?? [], - })), + tables: body.tables.map((t) => + toCdcTable({ ...t, table_name: t.name }), + ), destination: body.destination, }, info, @@ -320,19 +317,19 @@ export class InputsService { return formatedPayload; } - async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) { + async markTableDeleted(data: MarkTableDeletedRequest) { return lastValueFrom(this.inputWriteService.MarkTableDeleted(data)); } - async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) { - return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data)); + async unmarkTableDeleted(data: MarkTableDeletedRequest) { + return lastValueFrom(this.inputWriteService.UnmarkTableDeleted(data)); } - async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) { - return lastValueFrom(this.inputWriteService.AddCdcTable(data as any)); + async addCdcTable(data: AddCdcTableRequest) { + return lastValueFrom(this.inputWriteService.AddCdcTable(data)); } - async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) { - return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data)); + async removeCdcTable(data: RemoveCdcTableRequest) { + return lastValueFrom(this.inputWriteService.RemoveCdcTable(data)); } } diff --git a/src/modules/pipelinesV2/interfaces.ts b/src/modules/pipelinesV2/interfaces.ts index c54e74a..90bf076 100644 --- a/src/modules/pipelinesV2/interfaces.ts +++ b/src/modules/pipelinesV2/interfaces.ts @@ -9,6 +9,33 @@ export class PipelineInputsDTO { }> } +export class PipelineTableDestination { + @ApiPropertyOptional() + table_schema?: string; + @ApiPropertyOptional() + table_name?: string; +} + +/** One entry of the create body's `config.tables`. Mirrors the frontend's + * EntityWithColumnsNames; only the fields maestro/pi-factory read are typed, + * the rest travels as-is. */ +export class PipelineTableConfig { + @ApiProperty() + name: string; + @ApiPropertyOptional({ type: () => PipelineTableDestination }) + destinations?: Partial>; + [extra: string]: unknown; +} + +/** `{ cron, tables }` as sent by the frontend on create; JSON-serialized onto + * the gRPC `config` string field. */ +export class PipelineConfig { + @ApiPropertyOptional() + cron?: string; + @ApiPropertyOptional({ type: [PipelineTableConfig] }) + tables?: PipelineTableConfig[]; +} + export class IPipelineV2 { @ApiProperty() id: string; @@ -31,8 +58,8 @@ export class IPipelineV2 { tags?: string[]; @ApiPropertyOptional() properties?: any; - @ApiPropertyOptional() - config?: any; + @ApiPropertyOptional({ type: () => PipelineConfig }) + config?: PipelineConfig; @ApiProperty() connector_name: string; diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index 77c8059..62287c4 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -1,3 +1,4 @@ +import { isCdcPlugin } from 'src/utils/cdc'; /* eslint-disable no-async-promise-executor */ import { BadRequestException, @@ -437,7 +438,7 @@ export class PipelinesService implements OnModuleInit { }); this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput)); - const isCdc = !!oldInput.plugin?.endsWith('_cdc'); + const isCdc = isCdcPlugin(oldInput.plugin); const pipelineIdFormat = pipelineId.split('-').join('_'); const rollback: RollbackPromise[] = []; diff --git a/src/modules/platform-api/pipeline-tables.service.spec.ts b/src/modules/platform-api/pipeline-tables.service.spec.ts new file mode 100644 index 0000000..f0cced5 --- /dev/null +++ b/src/modules/platform-api/pipeline-tables.service.spec.ts @@ -0,0 +1,243 @@ +// 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; + 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 = {}) => ({ + 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 }); + }); +}); diff --git a/src/modules/platform-api/pipeline-tables.service.ts b/src/modules/platform-api/pipeline-tables.service.ts new file mode 100644 index 0000000..1ac95a7 --- /dev/null +++ b/src/modules/platform-api/pipeline-tables.service.ts @@ -0,0 +1,157 @@ +import { Inject, Injectable, NotFoundException } from '@nestjs/common'; +import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; +import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; + +import { RequestUser } from '../../decorators/user.decorator'; +import { InputsService } from '../inputs/inputs.service'; +import { toCdcTable } from '../inputs/cdc-table.mapper'; +import { PipelinesService } from '../pipelinesV2/pipelines.service'; +import { PlatformApiService } from './platform-api.service'; +import { AddCdcTableBody } from './platform-api.dto'; + +interface PlatformJob { + job_id: string; + input?: { table_name?: string; connector?: string } | null; +} + +interface PlatformPipeline { + jobs?: PlatformJob[] | null; +} + +/** + * Add/remove tables of a pipeline: the DynamoDB input (soft-delete flags or + * a new CdcTable row) and the platform-api jobs move together, with the + * DynamoDB side rolled back when the platform side fails. + * + * Removal always goes through DELETE /pipeline/{id}/jobs — the one platform + * route that dispatches by pipeline type (Airflow refresh for batch, Kafka + * Connect reconfiguration for CDC). Landed destination data is kept + * (delete_snowflake_tables: false), matching the soft-delete in DynamoDB. + */ +@Injectable() +export class PipelineTablesService { + private logger: DadosferaLogger['logger']; + + constructor( + private readonly platformApiService: PlatformApiService, + private readonly inputsService: InputsService, + private readonly pipelinesClientService: PipelinesService, + @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, + ) { + this.logger = dadosferaLogger.logger; + } + + async removeTable(pipelineId: string, inputId: string, tableName: string, user: RequestUser) { + const info = this.infoOf(user); + + this.logger.info('removeTable: marking table as deleted', { inputId, tableName }); + const marked = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info }); + + try { + const jobIds = await this.resolveJobIds(pipelineId, [tableName], user); + await this.removeJobs(pipelineId, jobIds, user); + return { name: tableName, is_deleted: marked.is_deleted ?? true, deleted_at: marked.deleted_at }; + } catch (error) { + this.logger.error('removeTable: platform-api delete failed, rolling back the mark', { tableName, error: error.message }); + await this.unmarkAll(inputId, [tableName], info); + throw error; + } + } + + /** Removes N tables with ONE platform call, so CDC connectors are reconfigured once. */ + async removeTables(pipelineId: string, inputId: string, tableNames: string[], user: RequestUser) { + const info = this.infoOf(user); + + // Soft-delete each table, tracking which succeeded so a later failure + // only rolls back the marks made in THIS call. + const marked: string[] = []; + try { + for (const name of tableNames) { + await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info }); + marked.push(name); + } + + const jobIds = await this.resolveJobIds(pipelineId, tableNames, user); + await this.removeJobs(pipelineId, jobIds, user); + return { table_names: tableNames, deleted: true }; + } catch (error) { + this.logger.error("removeTables: failed, rolling back this call's marks", { tableNames, error: error.message }); + await this.unmarkAll(inputId, marked, info); + throw error; + } + } + + /** Appends a CDC table to the DynamoDB input, then dispatches its platform jobs. */ + async addTable(pipelineId: string, inputId: string, body: AddCdcTableBody, user: RequestUser) { + const info = this.infoOf(user); + const cdcTable = toCdcTable(body); + + this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name }); + await this.inputsService.addCdcTable({ client_id: user.customer_id, id: inputId, table: cdcTable, info }); + + try { + return await this.pipelinesClientService.addCdcJobs({ + pipeline_id: pipelineId, + input_id: inputId, + tables: [{ + table_schema: body.table_schema, + table_name: body.table_name, + primary_keys: body.primary_keys, + destinations: body.destinations, + }], + info, + }); + } catch (error) { + this.logger.error('addTable: platform AddJobs failed, rolling back the DynamoDB row', { tableName: body.table_name, error: error.message }); + try { + await this.inputsService.removeCdcTable({ client_id: user.customer_id, id: inputId, table_name: body.table_name, info }); + } catch (rollbackError) { + this.logger.error('addTable: rollback failed', { error: rollbackError.message }); + } + throw error; + } + } + + private infoOf(user: RequestUser): Info { + return { customer_id: user.customer_id, customer: user.customer_name, user_id: user.user_id }; + } + + /** Platform-API replaces '-' with '_' in pipeline IDs. */ + private normalizePipelineId(id: string): string { + return id.replace(/-/g, '_'); + } + + private async resolveJobIds(pipelineId: string, tableNames: string[], user: RequestUser): Promise { + const normalizedPipelineId = this.normalizePipelineId(pipelineId); + const pipeline: PlatformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user); + const jobs = pipeline?.jobs ?? []; + + return tableNames.map((name) => { + const job = jobs.find((j) => j.input?.table_name === name); + if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`); + return job.job_id; + }); + } + + private async removeJobs(pipelineId: string, jobIds: string[], user: RequestUser) { + const normalizedPipelineId = this.normalizePipelineId(pipelineId); + this.logger.info('removeJobs: deleting jobs from platform-api', { jobIds }); + await this.platformApiService.proxy( + 'DELETE', + `/pipeline/${normalizedPipelineId}/jobs`, + user, + { job_ids: jobIds, delete_snowflake_tables: false }, + ); + this.logger.info('removeJobs: jobs deleted', { jobIds }); + } + + private async unmarkAll(inputId: string, tableNames: string[], info: Info) { + for (const name of tableNames) { + try { + await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info }); + } catch (rollbackError) { + this.logger.error('rollback failed: table stays marked as deleted', { tableName: name, error: rollbackError.message }); + } + } + } +} diff --git a/src/modules/platform-api/platform-api.controller.spec.ts b/src/modules/platform-api/platform-api.controller.spec.ts index f3d58a3..fdf2519 100644 --- a/src/modules/platform-api/platform-api.controller.spec.ts +++ b/src/modules/platform-api/platform-api.controller.spec.ts @@ -5,53 +5,43 @@ 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 {} })); +jest.mock('./pipeline-tables.service', () => ({ PipelineTablesService: class {} })); +import { BadRequestException } from '@nestjs/common'; 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 { PipelineTablesService } from './pipeline-tables.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 logger = { info: jest.fn(), error: jest.fn() }; +const mockUser = { customer_id: 'c1', customer_name: 'cust', user_id: 'u1' } as never; -const mockUser: any = { - customer_id: 'c1', - customer_name: 'cust', - user_id: 'u1', -}; - -describe('PlatformApiController - deleteTable', () => { +// The table routes are thin: validate the body, delegate to +// PipelineTablesService (covered in pipeline-tables.service.spec.ts). +describe('PlatformApiController - table routes', () => { let controller: PlatformApiController; - let platformApiService: { proxy: jest.Mock }; - let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock }; + let tables: { removeTable: jest.Mock; removeTables: jest.Mock; addTable: jest.Mock }; beforeEach(async () => { - platformApiService = { proxy: jest.fn() }; - inputsService = { - markTableDeleted: jest.fn(), - unmarkTableDeleted: jest.fn(), - }; + tables = { removeTable: jest.fn(), removeTables: jest.fn(), addTable: jest.fn() }; const module: TestingModule = await Test.createTestingModule({ controllers: [PlatformApiController], providers: [ - { provide: PlatformApiService, useValue: platformApiService }, + { provide: PlatformApiService, useValue: {} }, { provide: ElasticsearchService, useValue: {} }, { provide: DynamoDBService, useValue: {} }, { provide: CustomersService, useValue: {} }, { provide: CatalogService, useValue: {} }, - { provide: InputsService, useValue: inputsService }, - { provide: PipelinesService, useValue: {} }, + { provide: InputsService, useValue: {} }, + { provide: PipelineTablesService, useValue: tables }, { provide: DadosferaLogger, useValue: { logger } }, ], }).compile(); @@ -59,314 +49,37 @@ describe('PlatformApiController - deleteTable', () => { controller = module.get(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('deleteTable delegates', async () => { + tables.removeTable.mockResolvedValue({ name: 'pedidos', is_deleted: true }); + await expect(controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser)) + .resolves.toEqual({ name: 'pedidos', is_deleted: true }); + expect(tables.removeTable).toHaveBeenCalledWith('pid', 'iid', 'pedidos', mockUser); }); - 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); + it('deleteTables delegates', async () => { + tables.removeTables.mockResolvedValue({ deleted: true }); + await controller.deleteTables('pid', 'iid', { table_names: ['a', 'b'] }, mockUser); + expect(tables.removeTables).toHaveBeenCalledWith('pid', 'iid', ['a', 'b'], mockUser); }); - 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('deleteTables rejects an empty table_names before touching anything', async () => { + await expect(controller.deleteTables('pid', 'iid', { table_names: [] }, mockUser)) + .rejects.toBeInstanceOf(BadRequestException); + expect(tables.removeTables).not.toHaveBeenCalled(); }); - 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', + it('addTable delegates', async () => { + const body = { 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({}), + table_schema: 'public', + primary_keys: ['id'], + destinations: { + raw: { table_schema: 'raw', table_name: 'orders' }, + qualify: { table_schema: 'qualify', table_name: 'orders' }, + }, }; - - 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); - }); - - 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(); + tables.addTable.mockResolvedValue({ job_ids: ['p_2'], skipped: [] }); + await expect(controller.addTable('pid', 'iid', body, mockUser)).resolves.toEqual({ job_ids: ['p_2'], skipped: [] }); + expect(tables.addTable).toHaveBeenCalledWith('pid', 'iid', body, mockUser); }); }); diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 44e1ac9..7943b9e 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -11,7 +11,6 @@ import { Inject, BadRequestException, HttpException, - NotFoundException, UseGuards, } from '@nestjs/common'; import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger'; @@ -31,9 +30,9 @@ import { CustomersService } from '../customers/customers.service'; import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation'; import { CatalogService } from '../catalog/catalog.service'; import { PackTheMetadata } from '../../utils/PackTheMetadata'; -import { ValidationTableDTO } from './platform-api.dto'; +import { AddCdcTableBody, DeleteTableBody, DeleteTablesBody, ValidationTableDTO } from './platform-api.dto'; import { InputsService } from '../inputs/inputs.service'; -import { PipelinesService } from '../pipelinesV2/pipelines.service'; +import { PipelineTablesService } from './pipeline-tables.service'; import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; @@ -62,7 +61,7 @@ export class PlatformApiController { private readonly customersService: CustomersService, private readonly catalogService: CatalogService, private readonly inputsService: InputsService, - private readonly pipelinesClientService: PipelinesService, + private readonly pipelineTablesService: PipelineTablesService, @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, ) { this.logger = dadosferaLogger.logger; @@ -988,60 +987,16 @@ export class PlatformApiController { } @Delete('pipelines/:pipelineId/inputs/:inputId') - @ApiOperation({ summary: 'Mark a table as deleted and delete its associated job via platform-api' }) + @ApiOperation({ summary: 'Mark a table as deleted and remove its job via platform-api' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE) @UseGuards(PipelineExecutionGuard) async deleteTable( @Param('pipelineId') pipelineId: string, @Param('inputId') inputId: string, - @Body() body: { table_name: string }, + @Body() body: DeleteTableBody, @User() user: RequestUser, ) { - const tableName = body.table_name; - const info = { - customer_id: user.customer_id, - customer: user.customer_name, - user_id: user.user_id, - }; - - this.logger.info('deleteTable: marking table as deleted', { inputId, tableName }); - const updatedInput: any = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info }); - this.logger.info('deleteTable: table marked as deleted', { inputId, tableName }); - - try { - const normalizedPipelineId = this.normalizePipelineId(pipelineId); - this.logger.info('deleteTable: fetching pipeline from platform-api', { pipelineId, normalizedPipelineId }); - const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user); - this.logger.info('deleteTable: pipeline fetched', { jobCount: platformPipeline?.jobs?.length }); - - const job = platformPipeline?.jobs?.find((j: any) => j.input?.table_name === tableName); - if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`); - - this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id }); - if (job.input?.connector === 'cdc') { - // CDC: reconfigure the Debezium/Kafka-Connect connector (stop - // replicating this table); keep the landed Snowflake data. - await this.platformApiService.proxy( - 'DELETE', - `/pipeline/${normalizedPipelineId}/jobs`, - user, - { job_ids: [job.job_id], delete_snowflake_tables: false }, - ); - } else { - await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user); - } - this.logger.info('deleteTable: job deleted', { jobId: job.job_id }); - - return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at }; - } catch (error) { - this.logger.error('deleteTable: platform-api delete failed, attempting rollback', { tableName, error: error.message }); - try { - await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: tableName, info }); - } catch (rollbackError) { - this.logger.error('deleteTable: rollback failed', { tableName, error: rollbackError.message }); - } - throw error; - } + return this.pipelineTablesService.removeTable(pipelineId, inputId, body.table_name, user); } @Delete('pipelines/:pipelineId/inputs/:inputId/tables') @@ -1051,66 +1006,14 @@ export class PlatformApiController { async deleteTables( @Param('pipelineId') pipelineId: string, @Param('inputId') inputId: string, - @Body() body: { table_names: string[] }, + @Body() body: DeleteTablesBody, @User() user: RequestUser, ) { const tableNames = body.table_names ?? []; if (tableNames.length === 0) { throw new BadRequestException('table_names must be a non-empty array'); } - const info = { - customer_id: user.customer_id, - customer: user.customer_name, - user_id: user.user_id, - }; - - // 1) Soft-delete each table in DynamoDB, tracking which succeeded so a later - // failure only rolls back the marks made in THIS call. - const marked: string[] = []; - const rollback = async () => { - for (const name of marked) { - try { - await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info }); - } catch (rollbackError) { - this.logger.error('deleteTables: rollback failed', { tableName: name, error: rollbackError.message }); - } - } - }; - - try { - for (const name of tableNames) { - await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info }); - marked.push(name); - } - - // 2) Resolve all table_names -> job_ids from the platform pipeline (one GET). - const normalizedPipelineId = this.normalizePipelineId(pipelineId); - const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user); - const jobs = platformPipeline?.jobs ?? []; - - const jobIds: string[] = []; - for (const name of tableNames) { - const job = jobs.find((j: any) => j.input?.table_name === name); - if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`); - jobIds.push(job.job_id); - } - - // 3) Remove them all in ONE platform call so the Debezium source + Snowflake - // sink connectors are reconfigured a single time, not once per table. - await this.platformApiService.proxy( - 'DELETE', - `/pipeline/${normalizedPipelineId}/jobs`, - user, - { job_ids: jobIds, delete_snowflake_tables: false }, - ); - this.logger.info('deleteTables: jobs deleted', { jobIds }); - - return { table_names: tableNames, deleted: true }; - } catch (error) { - this.logger.error('deleteTables: failed, rolling back this call\'s marks', { tableNames, error: error.message }); - await rollback(); - throw error; - } + return this.pipelineTablesService.removeTables(pipelineId, inputId, tableNames, user); } @Post('pipelines/:pipelineId/inputs/:inputId/tables') @@ -1120,77 +1023,10 @@ export class PlatformApiController { async addTable( @Param('pipelineId') pipelineId: string, @Param('inputId') inputId: string, - @Body() body: { - table_name: string; - table_schema: string; - primary_keys: string[]; - destinations: { - raw: { table_schema: string; table_name: string }; - qualify: { table_schema: string; table_name: string }; - }; - // Iceberg destination only (protospack CdcTable.iceberg_table_name); - // absent for snowflake, back-compat. - iceberg_table_name?: string; - // Per-table deduped (qualify) Iceberg table name (protospack - // CdcTable.iceberg_qualify_table_name); absent => same as the raw name. - iceberg_qualify_table_name?: string; - // Source column schema for iceberg deduped table pre-create (protospack CdcTable.columns). - columns?: { name: string; type: string; is_primary_key: boolean }[]; - // Columns the user chose to ignore -> Debezium column.exclude.list. - column_exclude_list?: string[]; - }, + @Body() body: AddCdcTableBody, @User() user: RequestUser, ) { - const info = { - customer_id: user.customer_id, - customer: user.customer_name, - user_id: user.user_id, - }; - - const cdcTable = { - table_schema: body.table_schema, - table_name: body.table_name, - primary_keys: body.primary_keys, - name: body.table_name, - iceberg_table_name: body.iceberg_table_name, - iceberg_qualify_table_name: body.iceberg_qualify_table_name, - columns: body.columns ?? [], - column_exclude_list: body.column_exclude_list ?? [], - }; - - this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name }); - await this.inputsService.addCdcTable({ id: inputId, table: cdcTable, info }); - this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name }); - - try { - const customInfo = { - customer_id: user.customer_id, - user_id: user.user_id, - customer: user.customer_name, - }; - - const res = await this.pipelinesClientService.addCdcJobs({ - pipeline_id: pipelineId, - input_id: inputId, - tables: [{ - table_schema: body.table_schema, - table_name: body.table_name, - primary_keys: body.primary_keys, - destinations: body.destinations, - }], - info: customInfo, - }); - - return res; - } catch (error) { - this.logger.error('addTable: platform AddJobs failed, rolling back the DynamoDB row', { tableName: body.table_name, error: error.message }); - try { - await this.inputsService.removeCdcTable({ id: inputId, table_name: body.table_name, info }); - } catch (rbErr) { - this.logger.error('addTable: rollback failed', { error: rbErr.message }); - } - throw error; - } + return this.pipelineTablesService.addTable(pipelineId, inputId, body, user); } // ==================== JOBS - JDBC SYNC MODE ROUTES ==================== diff --git a/src/modules/platform-api/platform-api.dto.ts b/src/modules/platform-api/platform-api.dto.ts index 3f6b89f..ab52b61 100644 --- a/src/modules/platform-api/platform-api.dto.ts +++ b/src/modules/platform-api/platform-api.dto.ts @@ -1,9 +1,66 @@ -import { ApiProperty } from "@nestjs/swagger"; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; export class ValidationTableDTO { - @ApiProperty() - tables: Array<{ - table_name: string; - table_schema: string; - }> -} \ No newline at end of file + @ApiProperty() + tables: Array<{ + table_name: string; + table_schema: string; + }>; +} + +export class DeleteTableBody { + @ApiProperty() + table_name: string; +} + +export class DeleteTablesBody { + @ApiProperty({ type: [String] }) + table_names: string[]; +} + +export class CdcTableDestination { + @ApiProperty() + table_schema: string; + @ApiProperty() + table_name: string; +} + +export class CdcTableDestinations { + @ApiProperty({ type: CdcTableDestination }) + raw: CdcTableDestination; + @ApiProperty({ type: CdcTableDestination }) + qualify: CdcTableDestination; +} + +export class CdcColumnBody { + @ApiProperty() + name: string; + @ApiProperty() + type: string; + @ApiProperty() + is_primary_key: boolean; +} + +export class AddCdcTableBody { + @ApiProperty() + table_name: string; + @ApiProperty() + table_schema: string; + @ApiProperty({ type: [String] }) + primary_keys: string[]; + @ApiProperty({ type: CdcTableDestinations }) + destinations: CdcTableDestinations; + // Iceberg destination only (protospack CdcTable.iceberg_table_name); + // absent for snowflake, back-compat. + @ApiPropertyOptional() + iceberg_table_name?: string; + // Per-table deduped (qualify) Iceberg table name; absent => same as raw. + @ApiPropertyOptional() + iceberg_qualify_table_name?: string; + // Source column schema for iceberg deduped table pre-create. + @ApiPropertyOptional({ type: [CdcColumnBody] }) + columns?: CdcColumnBody[]; + // Columns the user chose to ignore -> Debezium column.exclude.list. + @ApiPropertyOptional({ type: [String] }) + column_exclude_list?: string[]; +} diff --git a/src/modules/platform-api/platform-api.module.ts b/src/modules/platform-api/platform-api.module.ts index 726e743..d929c66 100644 --- a/src/modules/platform-api/platform-api.module.ts +++ b/src/modules/platform-api/platform-api.module.ts @@ -4,6 +4,7 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { PlatformApiController } from './platform-api.controller'; import { PlatformApiService } from './platform-api.service'; +import { PipelineTablesService } from './pipeline-tables.service'; import { ElasticsearchModule } from '../../services/elasticsearch'; import { DynamoDBModule } from '../../services/dynamodb'; import { CustomersModule } from '../customers/customers.module'; @@ -21,7 +22,7 @@ import { PipelinesV2Module } from '../pipelinesV2/pipelines.module'; forwardRef(() => PipelinesV2Module), ], controllers: [PlatformApiController], - providers: [PlatformApiService, DadosferaLogger], + providers: [PlatformApiService, PipelineTablesService, DadosferaLogger], exports: [PlatformApiService], }) export class PlatformApiModule {} diff --git a/src/utils/cdc.ts b/src/utils/cdc.ts new file mode 100644 index 0000000..e02f232 --- /dev/null +++ b/src/utils/cdc.ts @@ -0,0 +1,34 @@ +/** + * Single source of truth for "is this CDC?" across maestro. + * + * Two shapes carry the discriminator: + * - a connection/input `plugin` (`mysql_cdc`, ...), used before a platform + * pipeline exists (create flows, DynamoDB inputs); + * - a platform job `input.connector === 'cdc'`, the explicit type the + * platform-api stamps on every CDC job. + * + * Keep the plugin list in sync with pi-factory (`CDC_PLUGINS`) and in-factory + * (`cdcPlugins`). + */ +export const CDC_PLUGINS: readonly string[] = ['mysql_cdc', 'postgresql_cdc', 'oracle_cdc']; + +export function isCdcPlugin(plugin?: string | null): boolean { + return plugin != null && CDC_PLUGINS.includes(plugin); +} + +export interface PlatformJobLike { + input?: { connector?: string } | null; +} + +export interface PlatformPipelineLike { + jobs?: PlatformJobLike[] | null; +} + +export function isCdcJob(job?: PlatformJobLike | null): boolean { + return job?.input?.connector === 'cdc'; +} + +/** A pipeline is CDC when any of its platform jobs is a CDC job. */ +export function isCdcPipeline(pipeline?: PlatformPipelineLike | null): boolean { + return (pipeline?.jobs ?? []).some((job) => isCdcJob(job)); +}