mirror of
https://github.com/dadosfera/maestro.git
synced 2026-08-31 19:58:21 +00:00
1042 lines
34 KiB
TypeScript
1042 lines
34 KiB
TypeScript
import {
|
|
Controller,
|
|
Get,
|
|
Post,
|
|
Put,
|
|
Patch,
|
|
Delete,
|
|
Param,
|
|
Body,
|
|
Query,
|
|
Inject,
|
|
BadRequestException,
|
|
HttpException,
|
|
NotFoundException,
|
|
UseGuards,
|
|
} from '@nestjs/common';
|
|
import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger';
|
|
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
|
|
|
|
import {
|
|
Authenticated,
|
|
RequireAllPermissions,
|
|
RequireModule,
|
|
} from '../../decorators/authentication.decorator';
|
|
import { User, RequestUser } from '../../decorators/user.decorator';
|
|
import { PlatformApiService } from './platform-api.service';
|
|
import { DADOSFERA_MODULES_KEYS, PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
|
|
import { ElasticsearchService } from '../../services/elasticsearch';
|
|
import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
|
|
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 { InputsService } from '../inputs/inputs.service';
|
|
import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
|
|
|
|
|
|
type ValidateTablesDTO = {
|
|
tables: Array<{
|
|
table_schema: string,
|
|
table_name: string
|
|
}>
|
|
}
|
|
|
|
type RenameTablesBody = {
|
|
raw?: { table_name: string; table_schema: string };
|
|
qualify?: { table_name: string; table_schema: string };
|
|
}
|
|
|
|
@ApiTags('Platform API')
|
|
@Controller('platform')
|
|
@RequireModule(DADOSFERA_MODULES_KEYS.COLLECT)
|
|
export class PlatformApiController {
|
|
private logger: any;
|
|
|
|
constructor(
|
|
private readonly platformApiService: PlatformApiService,
|
|
private readonly elasticsearchService: ElasticsearchService,
|
|
private readonly dynamoDBService: DynamoDBService,
|
|
private readonly customersService: CustomersService,
|
|
private readonly catalogService: CatalogService,
|
|
private readonly inputsService: InputsService,
|
|
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
|
|
) {
|
|
this.logger = dadosferaLogger.logger;
|
|
}
|
|
|
|
/**
|
|
* Normalize pipeline ID to match Platform-API format.
|
|
* Platform-API replaces '-' with '_' in pipeline IDs.
|
|
*/
|
|
private normalizePipelineId(id: string): string {
|
|
return id?.replace(/-/g, '_') || '';
|
|
}
|
|
|
|
private decodePathParam(value: string): string {
|
|
return value ? decodeURIComponent(value) : '';
|
|
}
|
|
|
|
/**
|
|
* Denormalize ID back to UUID format (replace _ with -).
|
|
* Used when we receive a normalized ID but need the original UUID.
|
|
*/
|
|
private denormalizeId(id: string): string {
|
|
return id?.replace(/_/g, '-') || '';
|
|
}
|
|
|
|
/**
|
|
* Normalize job ID to match Platform-API format.
|
|
* Platform-API replaces '-' with '_' in job IDs.
|
|
*
|
|
* Example: "2ccf5481-59f5-4036-8a94-7d5f28f4f899-0" -> "2ccf5481_59f5_4036_8a94_7d5f28f4f899_0"
|
|
*/
|
|
private normalizeJobId(jobId: string): string {
|
|
return jobId?.replace(/-/g, '_') || '';
|
|
}
|
|
|
|
private async getJobByAnyConnectorType(normalizedJobId: string, user: RequestUser): Promise<any> {
|
|
const connectorTypes = ['jdbc', 'singer', 's3'];
|
|
for (const type of connectorTypes) {
|
|
try {
|
|
const job = await this.platformApiService.proxy(
|
|
'GET',
|
|
`/jobs/${type}/${normalizedJobId}`,
|
|
user,
|
|
);
|
|
return job;
|
|
} catch (error) {
|
|
// Continue to next connector type
|
|
}
|
|
}
|
|
throw new HttpException(`Job ${normalizedJobId} not found in any connector type (jdbc, singer, s3)`, 404);
|
|
}
|
|
|
|
/**
|
|
* Extract the pipeline ID (base UUID) from a job ID.
|
|
* Job IDs have format "uuid-suffix" where suffix is the job index (e.g., "0", "1").
|
|
* Handles both hyphenated and underscored formats, always returns hyphenated UUID for ES.
|
|
*
|
|
* Examples:
|
|
* - "2ccf5481-59f5-4036-8a94-7d5f28f4f899-0" -> "2ccf5481-59f5-4036-8a94-7d5f28f4f899"
|
|
* - "2ccf5481_59f5_4036_8a94_7d5f28f4f899_0" -> "2ccf5481-59f5-4036-8a94-7d5f28f4f899"
|
|
*/
|
|
private extractPipelineIdFromJobId(jobId: string): string {
|
|
if (!jobId) return '';
|
|
|
|
// Determine the separator used in the jobId
|
|
const hasUnderscores = jobId.includes('_');
|
|
const separator = hasUnderscores ? '_' : '-';
|
|
|
|
const parts = jobId.split(separator);
|
|
// UUID has 5 parts (8-4-4-4-12), job suffix is the 6th part
|
|
if (parts.length >= 6) {
|
|
// Always return hyphenated format for Elasticsearch lookup
|
|
return parts.slice(0, 5).join('-');
|
|
}
|
|
// If no suffix found, return the ID in hyphenated format
|
|
return hasUnderscores ? jobId.replace(/_/g, '-') : jobId;
|
|
}
|
|
|
|
private readonly VALID_CONNECTORS = ['jdbc', 'singer', 's3'];
|
|
private readonly MAX_MEMORY_MB = 12000; // 12GB maximum memory per pipelines/job
|
|
|
|
/**
|
|
* Validate that connector is provided and is a valid type.
|
|
*/
|
|
private validateConnector(connector: string | undefined): void {
|
|
if (!connector || !this.VALID_CONNECTORS.includes(connector)) {
|
|
throw new BadRequestException(
|
|
`connector is required in job input and must be one of: ${this.VALID_CONNECTORS.join(', ')}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Validate memory allocation against maximum limit.
|
|
*/
|
|
private validateMemory(memoryMb: number): void {
|
|
if (memoryMb > this.MAX_MEMORY_MB) {
|
|
throw new BadRequestException(
|
|
`Memory limit exceeded. Maximum allowed: ${this.MAX_MEMORY_MB}MB (12GB)`,
|
|
);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Validate cron expression against customer's schedule limit.
|
|
* Fetches current scheduleLimit from DUC to ensure up-to-date configuration.
|
|
*/
|
|
private async validateScheduleLimit(cron: string, customerId: string): Promise<void> {
|
|
if (!cron) return;
|
|
|
|
const { customer } = await this.customersService.getCustomer(customerId);
|
|
const scheduleLimit = customer?.scheduleLimit || 'day';
|
|
|
|
const result = validateCronAgainstScheduleLimit(cron, scheduleLimit);
|
|
if (!result.valid) {
|
|
throw new BadRequestException(result.message);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Map connector type to DynamoDB type.
|
|
* jdbc -> 'database', singer -> 'application', s3 -> 'file'
|
|
*/
|
|
private mapConnectorToDynamoType(connector: string): string {
|
|
switch (connector) {
|
|
case 'jdbc':
|
|
return 'database';
|
|
case 'singer':
|
|
return 'application';
|
|
case 's3':
|
|
return 'file';
|
|
default:
|
|
return connector;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Extract and transform tables from jobs for DynamoDB input.
|
|
* Maps connector-specific fields to a common table format.
|
|
*
|
|
* - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name/type (reference_column object)
|
|
* - Singer: type maps replication_method (FULL_TABLE -> full_load, INCREMENTAL -> incremental), no columns
|
|
* - S3: same mapping as Singer, no columns
|
|
*/
|
|
private extractTablesFromJobs(jobs: any[], connector: string): Array<{
|
|
name: string;
|
|
type: string;
|
|
columns?: string[];
|
|
reference_column?: ReferenceColumn;
|
|
}> {
|
|
if (!jobs || jobs.length === 0) return [];
|
|
|
|
const tables: Array<{
|
|
name: string;
|
|
type: string;
|
|
columns?: string[];
|
|
reference_column?: ReferenceColumn;
|
|
}> = [];
|
|
|
|
for (const job of jobs) {
|
|
const input = job.input;
|
|
if (!input) continue;
|
|
|
|
if (connector === 'jdbc') {
|
|
// JDBC: table_name, load_type, column_include_list, incremental_column_name/type
|
|
const table: {
|
|
name: string;
|
|
type: string;
|
|
columns?: string[];
|
|
reference_column?: ReferenceColumn;
|
|
} = {
|
|
name: input.table_name || '',
|
|
type: input.load_type || 'full_load',
|
|
};
|
|
if (input.column_include_list && input.column_include_list.length > 0) {
|
|
table.columns = input.column_include_list;
|
|
}
|
|
if (input.incremental_column_name) {
|
|
// reference_column is stored as an object with name and type
|
|
table.reference_column = {
|
|
name: input.incremental_column_name,
|
|
type: input.incremental_column_type || 'unknown',
|
|
};
|
|
}
|
|
tables.push(table);
|
|
} else if (connector === 'singer' || connector === 's3') {
|
|
// Singer/S3: replication_method -> type mapping, no columns
|
|
let type = 'full_load';
|
|
if (input.replication_method === 'INCREMENTAL') {
|
|
type = 'incremental';
|
|
} else if (input.replication_method === 'FULL_TABLE') {
|
|
type = 'full_load';
|
|
}
|
|
|
|
tables.push({
|
|
name: input.table_name || '',
|
|
type,
|
|
});
|
|
}
|
|
}
|
|
|
|
return tables;
|
|
}
|
|
|
|
/**
|
|
* Build properties object for Elasticsearch based on connector type.
|
|
* Different connectors have different property structures.
|
|
*
|
|
* Note: In pi-factory flow, properties come pre-built from frontend.
|
|
* In Maestro proxy flow, we reconstruct from job input fields.
|
|
*/
|
|
private buildPipelineProperties(jobInput: any): Record<string, any> {
|
|
if (!jobInput) return {};
|
|
|
|
const connector = jobInput.connector;
|
|
const properties: Record<string, any> = {};
|
|
|
|
// Determine credentials_type
|
|
if (jobInput.auth_parameters?.credentials_type) {
|
|
properties.credentials_type = jobInput.auth_parameters.credentials_type;
|
|
} else {
|
|
// Default based on connector type
|
|
// S3 connector typically uses iam_user, others use basic_auth
|
|
properties.credentials_type = connector === 's3' ? 'iam_user' : 'basic_auth';
|
|
}
|
|
|
|
if (connector === 'jdbc') {
|
|
// JDBC connectors: schema comes from table_schema
|
|
if (jobInput.table_schema) {
|
|
properties.schema = jobInput.table_schema;
|
|
}
|
|
} else if (connector === 'singer') {
|
|
// Singer connectors: merge config fields (dates, selected_competitions, etc.)
|
|
if (jobInput.config) {
|
|
Object.assign(properties, jobInput.config);
|
|
}
|
|
} else if (connector === 's3') {
|
|
// S3 connector
|
|
if (jobInput.engine) properties.engine = jobInput.engine;
|
|
if (jobInput.source_bucket) properties.source_bucket = jobInput.source_bucket;
|
|
if (jobInput.source_prefix) properties.source_prefix = jobInput.source_prefix;
|
|
if (jobInput.file_format_params) properties.file_format_params = jobInput.file_format_params;
|
|
}
|
|
|
|
return properties;
|
|
}
|
|
|
|
/**
|
|
* Sync job input changes to DynamoDB for a specific connector type.
|
|
* Extracts pipeline ID from job ID, fetches ES document to find input ID,
|
|
* then updates the table entry in DynamoDB.
|
|
*
|
|
* Job ID transformations:
|
|
* - Raw format (from endpoint): "2ccf5481-59f5-4036-8a94-7d5f28f4f899-0"
|
|
* - Platform API format: "2ccf5481_59f5_4036_8a94_7d5f28f4f899_0" (underscores)
|
|
* - Elasticsearch pipeline ID: "2ccf5481-59f5-4036-8a94-7d5f28f4f899" (UUID only, hyphens)
|
|
*
|
|
* @param connectorType - The connector type ('jdbc', 'singer', 's3') for the Platform API endpoint
|
|
*/
|
|
private async syncJobInputToDynamoDB(
|
|
jobId: string,
|
|
body: any,
|
|
user: RequestUser,
|
|
connectorType: 'jdbc' | 'singer' | 's3',
|
|
): Promise<void> {
|
|
try {
|
|
// Normalize job ID for Platform API GET (replace - with _)
|
|
const normalizedJobId = this.normalizeJobId(jobId);
|
|
|
|
// Get job details using connector-specific endpoint to find table_name
|
|
const jobResult = await this.platformApiService.proxy(
|
|
'GET',
|
|
`/jobs/${connectorType}/${normalizedJobId}`,
|
|
user,
|
|
);
|
|
|
|
// Extract the pipeline ID (base UUID) from the raw job ID for ES lookup
|
|
const esPipelineId = this.extractPipelineIdFromJobId(jobId);
|
|
const tableName = body.table_name || jobResult.source_config?.table_name;
|
|
|
|
if (!esPipelineId || !tableName) {
|
|
this.logger.warn('Cannot sync job input: missing pipeline_id or table_name', {
|
|
jobId,
|
|
esPipelineId,
|
|
tableName,
|
|
});
|
|
return;
|
|
}
|
|
|
|
// Get pipeline from ES to find input ID (stored in config.tables)
|
|
const pipeline = await this.elasticsearchService.getPipeline(
|
|
user.customer_name,
|
|
esPipelineId,
|
|
);
|
|
|
|
const inputId = pipeline?.config?.tables;
|
|
if (!inputId) {
|
|
this.logger.warn('Cannot sync job input: no input ID in ES', {
|
|
jobId,
|
|
esPipelineId,
|
|
});
|
|
return;
|
|
}
|
|
|
|
// Build changes for DynamoDB table entry
|
|
// reference_column is stored as an object with name and type
|
|
const changes: {
|
|
type?: string;
|
|
columns?: string[];
|
|
reference_column?: ReferenceColumn | null;
|
|
} = {};
|
|
|
|
|
|
if ('target_load_type' in body) {
|
|
changes.type = body.target_load_type;
|
|
}
|
|
if ('column_include_list' in body) {
|
|
changes.columns = body.column_include_list;
|
|
}
|
|
if ('incremental_column_name' in body) {
|
|
// reference_column is stored as an object with name and type
|
|
if (body.incremental_column_name) {
|
|
changes.reference_column = {
|
|
name: body.incremental_column_name,
|
|
type: body.incremental_column_type || 'unknown',
|
|
};
|
|
} else {
|
|
changes.reference_column = null;
|
|
}
|
|
}
|
|
|
|
// Update DynamoDB if there are changes
|
|
if (Object.keys(changes).length > 0) {
|
|
await this.dynamoDBService.updateInputTable(
|
|
user.customer_id,
|
|
inputId,
|
|
tableName,
|
|
changes,
|
|
);
|
|
}
|
|
} catch (error) {
|
|
this.logger.error('Failed to sync job input to DynamoDB', {
|
|
jobId,
|
|
connectorType,
|
|
error: error.message,
|
|
});
|
|
// Don't throw - Platform API update succeeded, just log the sync error
|
|
}
|
|
}
|
|
|
|
// ==================== PIPELINE ROUTES ====================
|
|
|
|
@Post('pipelines')
|
|
@ApiOperation({ summary: 'Create a new pipeline' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
|
|
async createPipeline(@Body() body: any, @User() user: RequestUser) {
|
|
// Validate that pipeline has jobs
|
|
if (!body.jobs || body.jobs.length === 0) {
|
|
throw new BadRequestException('Pipeline must have at least one job');
|
|
}
|
|
|
|
// Validate connector before proceeding
|
|
const firstJob = body.jobs[0]?.input;
|
|
this.validateConnector(firstJob?.connector);
|
|
|
|
// Validate cron against customer's schedule limit
|
|
await this.validateScheduleLimit(body.cron, user.customer_id);
|
|
|
|
// Inject customer_id (actually customer_name) and normalized ID into body for Platform-API
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
const enrichedBody = {
|
|
...body,
|
|
id: this.normalizePipelineId(body.id),
|
|
customer_id: user.customer_name,
|
|
};
|
|
|
|
const result = await this.platformApiService.proxy('POST', '/pipeline', user, enrichedBody);
|
|
|
|
// Sync to Elasticsearch and DynamoDB
|
|
try {
|
|
const plugin = firstJob?.plugin;
|
|
const connectorType = firstJob?.connector;
|
|
const connectionId = firstJob?.auth_parameters?.config_id;
|
|
|
|
// Build properties based on connector type
|
|
const properties = this.buildPipelineProperties(firstJob);
|
|
|
|
const connector = plugin
|
|
? await this.elasticsearchService.getConnectorByPlugin(plugin)
|
|
: null;
|
|
|
|
// Extract tables from jobs and create DynamoDB input
|
|
const tables = this.extractTablesFromJobs(body.jobs, connectorType);
|
|
let inputId: string | undefined;
|
|
|
|
if (tables.length > 0) {
|
|
const inputDoc = await this.dynamoDBService.createInput(
|
|
user.customer_id,
|
|
user.user_id,
|
|
{
|
|
name: body.name,
|
|
description: body.description,
|
|
plugin: plugin || '',
|
|
type: this.mapConnectorToDynamoType(connectorType),
|
|
tables,
|
|
},
|
|
);
|
|
inputId = inputDoc.id;
|
|
this.logger.info('Created DynamoDB input for tables config', {
|
|
inputId,
|
|
pipelineId: body.id,
|
|
tablesCount: tables.length,
|
|
});
|
|
}
|
|
|
|
const pipelineType = this.mapConnectorToDynamoType(connectorType);
|
|
this.logger.info('Syncing pipeline to Elasticsearch', {
|
|
customerName: user.customer_name,
|
|
pipelineId: body.id,
|
|
plugin,
|
|
connector: connectorType,
|
|
type: pipelineType,
|
|
properties,
|
|
inputId,
|
|
});
|
|
|
|
// Keep original UUID format for Elasticsearch (not normalized)
|
|
await this.elasticsearchService.createPipeline(
|
|
user.customer_name,
|
|
body.id,
|
|
{
|
|
name: body.name,
|
|
description: body.description,
|
|
user_id: user.user_id,
|
|
username: user.username,
|
|
customer_id: user.customer_id,
|
|
plugin,
|
|
connection_id: connectionId,
|
|
cron: body.cron,
|
|
tables: inputId,
|
|
properties,
|
|
type: pipelineType,
|
|
},
|
|
connector,
|
|
);
|
|
} catch (error) {
|
|
this.logger.error('Failed to sync pipeline creation to Elasticsearch/DynamoDB', {
|
|
pipelineId: body.id,
|
|
customerName: user.customer_name,
|
|
error: error.message,
|
|
errorName: error.name,
|
|
});
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
@Get('pipelines')
|
|
@ApiOperation({ summary: 'List all pipelines for customer' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipelines(
|
|
@User() user: RequestUser,
|
|
@Query() query: Record<string, string>,
|
|
) {
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
'/pipelines',
|
|
user,
|
|
undefined,
|
|
query,
|
|
);
|
|
}
|
|
|
|
@Get('pipelines/:pipelineId')
|
|
@ApiOperation({ summary: 'Get pipeline by ID' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipeline(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@User() user: RequestUser,
|
|
) {
|
|
const normalizedId = this.normalizePipelineId(pipelineId);
|
|
return this.platformApiService.proxy('GET', `/pipeline/${normalizedId}`, user);
|
|
}
|
|
|
|
@Patch('pipelines/:pipelineId')
|
|
@ApiOperation({ summary: 'Update pipeline by ID' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async updatePipeline(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
// Validate cron against customer's schedule limit if cron is being updated
|
|
if (body.cron) {
|
|
await this.validateScheduleLimit(body.cron, user.customer_id);
|
|
}
|
|
|
|
const normalizedId = this.normalizePipelineId(pipelineId);
|
|
const result = await this.platformApiService.proxy(
|
|
'PATCH',
|
|
`/pipeline/${normalizedId}`,
|
|
user,
|
|
body,
|
|
);
|
|
|
|
// Sync to Elasticsearch (use original UUID, not normalized)
|
|
// Only pass fields that are explicitly provided in the request body
|
|
try {
|
|
const esChanges: {
|
|
name?: string;
|
|
description?: string;
|
|
cron?: string;
|
|
status?: string;
|
|
} = {};
|
|
|
|
if ('name' in body) esChanges.name = body.name;
|
|
if ('description' in body) esChanges.description = body.description;
|
|
if ('cron' in body) esChanges.cron = body.cron;
|
|
if ('status' in body) esChanges.status = body.status;
|
|
|
|
await this.elasticsearchService.updatePipeline(
|
|
user.customer_name,
|
|
pipelineId,
|
|
esChanges,
|
|
);
|
|
} catch (error) {
|
|
this.logger.error('Failed to sync pipeline update to Elasticsearch', {
|
|
pipelineId,
|
|
error: error.message,
|
|
});
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
@Delete('pipelines/:pipelineId')
|
|
@ApiOperation({ summary: 'Delete pipeline by ID' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
|
|
async deletePipeline(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@User() user: RequestUser,
|
|
) {
|
|
const normalizedId = this.normalizePipelineId(pipelineId);
|
|
const result = await this.platformApiService.proxy(
|
|
'DELETE',
|
|
`/pipeline/${normalizedId}`,
|
|
user,
|
|
);
|
|
|
|
// Sync to Elasticsearch (use original UUID, not normalized)
|
|
try {
|
|
await this.elasticsearchService.deletePipeline(
|
|
user.customer_name,
|
|
pipelineId,
|
|
);
|
|
} catch (error) {
|
|
this.logger.error('Failed to sync pipeline deletion to Elasticsearch', {
|
|
pipelineId,
|
|
error: error.message,
|
|
});
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
@Post('pipelines/execute')
|
|
@ApiOperation({ summary: 'Execute a pipeline' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async executePipeline(@Body() body: any, @User() user: RequestUser) {
|
|
// Inject customer_id (actually customer_name) into body for Platform-API
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
const enrichedBody = {
|
|
...body,
|
|
customer_id: user.customer_name,
|
|
};
|
|
return this.platformApiService.proxy('POST', '/pipelines/execute', user, enrichedBody);
|
|
}
|
|
|
|
@Post('pipelines/pause')
|
|
@ApiOperation({ summary: 'Pause a pipeline' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async pausePipeline(@Body() body: any, @User() user: RequestUser) {
|
|
// Inject customer_id (actually customer_name) into body for Platform-API
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
const enrichedBody = {
|
|
...body,
|
|
customer_id: user.customer_name,
|
|
};
|
|
return this.platformApiService.proxy('POST', '/pipeline/pause', user, enrichedBody);
|
|
}
|
|
|
|
@Post('pipelines/unpause')
|
|
@ApiOperation({ summary: 'Unpause a pipeline' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async unpausePipeline(@Body() body: any, @User() user: RequestUser) {
|
|
// Inject customer_id (actually customer_name) into body for Platform-API
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
const enrichedBody = {
|
|
...body,
|
|
customer_id: user.customer_name,
|
|
};
|
|
return this.platformApiService.proxy('POST', '/pipeline/unpause', user, enrichedBody);
|
|
}
|
|
|
|
@Put('pipelines/:pipelineId/memory')
|
|
@ApiOperation({ summary: 'Update pipeline memory configuration' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async updatePipelineMemory(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
// Validate memory limit
|
|
if (body.amount) {
|
|
this.validateMemory(body.amount);
|
|
}
|
|
|
|
return this.platformApiService.proxy(
|
|
'PUT',
|
|
`/pipeline/${pipelineId}/memory`,
|
|
user,
|
|
body,
|
|
);
|
|
}
|
|
|
|
// ==================== PIPELINE METADATA ROUTES ====================
|
|
|
|
@Put('pipelines/:pipelineId/metadata')
|
|
@ApiOperation({ summary: 'Update pipeline metadata' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async updatePipelineMetadata(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
return this.platformApiService.proxy(
|
|
'PUT',
|
|
`/pipeline/${pipelineId}/metadata`,
|
|
user,
|
|
body,
|
|
);
|
|
}
|
|
|
|
@Get('pipelines/metadata')
|
|
@ApiOperation({ summary: 'Get all pipelines metadata' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipelinesMetadata(
|
|
@User() user: RequestUser,
|
|
@Query() query: Record<string, string>,
|
|
) {
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
'/pipelines/metadata',
|
|
user,
|
|
undefined,
|
|
query,
|
|
);
|
|
}
|
|
|
|
// ==================== PIPELINE VALIDATION ====================
|
|
|
|
@Get('pipelines/catalog/schemas')
|
|
@ApiOperation({ summary: 'Get available schemas' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getAvailableSchemas(
|
|
@User() user: RequestUser,
|
|
@Query() query: Record<string, string>,
|
|
) {
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
`/catalog/schemas`,
|
|
user,
|
|
undefined,
|
|
query,
|
|
);
|
|
}
|
|
|
|
@Post('pipelines/catalog/tables/validate')
|
|
@ApiOperation({ summary: 'Validate Table and Schema' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async validateTableAndSchema(
|
|
@Body() payload: ValidationTableDTO,
|
|
@User() user: RequestUser,
|
|
@Query() query: Record<string, string>,
|
|
) {
|
|
return this.platformApiService.proxy(
|
|
'POST',
|
|
`/catalog/tables/validate`,
|
|
user,
|
|
payload,
|
|
query,
|
|
);
|
|
}
|
|
|
|
// ==================== PIPELINE RUN ROUTES ====================
|
|
|
|
@Get('pipelines/:pipelineId/pipeline_run')
|
|
@ApiOperation({ summary: 'Get pipeline runs for a pipeline' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipelineRuns(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@User() user: RequestUser,
|
|
@Query() query: Record<string, string>,
|
|
) {
|
|
const normalizedId = this.normalizePipelineId(pipelineId);
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
`/pipeline/${normalizedId}/pipeline_run`,
|
|
user,
|
|
undefined,
|
|
query,
|
|
);
|
|
}
|
|
|
|
@Get('pipelines/:pipelineId/pipeline_run/:runId')
|
|
@ApiOperation({ summary: 'Get specific pipeline run' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipelineRun(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@Param('runId') runId: string,
|
|
@User() user: RequestUser,
|
|
) {
|
|
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
|
|
const normalizedRunId = this.normalizePipelineId(runId);
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`,
|
|
user,
|
|
);
|
|
}
|
|
|
|
@Get('pipelines/pipeline_run/:runId/logs')
|
|
@ApiOperation({ summary: 'Get pipeline run logs' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipelineRunLogs(
|
|
@Param('runId') runId: string,
|
|
@User() user: RequestUser,
|
|
@Query() query: Record<string, string>,
|
|
) {
|
|
const normalizedRunId = this.normalizePipelineId(runId);
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
`/pipeline/pipeline_run/${normalizedRunId}/logs`,
|
|
user,
|
|
undefined,
|
|
query,
|
|
);
|
|
}
|
|
|
|
@Post('pipelines/:pipelineId/pipeline_run/:runId/cancel')
|
|
@ApiOperation({ summary: 'Cancel a running pipeline run' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async cancelPipelineRun(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@Param('runId') runId: string,
|
|
@User() user: RequestUser,
|
|
) {
|
|
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
|
|
const normalizedRunId = this.normalizePipelineId(runId);
|
|
|
|
const status = await this.platformApiService.proxy(
|
|
'GET',
|
|
`/pipeline/${normalizedPipelineId}/pipeline_run`,
|
|
user,
|
|
);
|
|
|
|
if (status.length === 1) {
|
|
throw new BadRequestException('The first pipeline cannot be canceled');
|
|
}
|
|
|
|
return this.platformApiService.proxy(
|
|
'POST',
|
|
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`,
|
|
user,
|
|
);
|
|
}
|
|
|
|
@Get('pipelines/:pipelineId/pipeline_run/:runId/jobs')
|
|
@ApiOperation({
|
|
summary: 'Get pipeline run jobs',
|
|
description: 'Proxies platform-api DB-backed job runs and returns `{ jobs: [...] }`.',
|
|
})
|
|
@ApiOkResponse({
|
|
description: 'DB-backed job runs for the selected pipeline run.',
|
|
schema: {
|
|
type: 'object',
|
|
properties: {
|
|
jobs: {
|
|
type: 'array',
|
|
items: { type: 'object' },
|
|
},
|
|
},
|
|
required: ['jobs'],
|
|
},
|
|
})
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getPipelineRunJobs(
|
|
@Param('pipelineId') pipelineId: string,
|
|
@Param('runId') runId: string,
|
|
@User() user: RequestUser,
|
|
) {
|
|
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
|
|
const decodedRunId = this.decodePathParam(runId);
|
|
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
`/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/jobs`,
|
|
user,
|
|
);
|
|
}
|
|
|
|
// ==================== JOBS - COLUMN EDITING ROUTES ====================
|
|
|
|
@Put('jobs/:jobId/input')
|
|
@ApiOperation({ summary: 'Update job input columns' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async updateJobInput(
|
|
@Param('jobId') jobId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
// Normalize job ID for Platform API (replace - with _)
|
|
const normalizedJobId = this.normalizeJobId(jobId);
|
|
|
|
const result = await this.platformApiService.proxy(
|
|
'PUT',
|
|
`/jobs/${normalizedJobId}/input`,
|
|
user,
|
|
body,
|
|
);
|
|
|
|
// Sync to DynamoDB if connector type is provided
|
|
const connectorType = body.connector as 'jdbc' | 'singer' | 's3' | undefined;
|
|
if (connectorType && this.VALID_CONNECTORS.includes(connectorType)) {
|
|
await this.syncJobInputToDynamoDB(jobId, body, user, connectorType);
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
@Patch('jobs/:jobId/input')
|
|
@ApiOperation({ summary: 'Partial update job input columns' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async patchJobInput(
|
|
@Param('jobId') jobId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
// Normalize job ID for Platform API (replace - with _)
|
|
const normalizedJobId = this.normalizeJobId(jobId);
|
|
|
|
const result = await this.platformApiService.proxy(
|
|
'PATCH',
|
|
`/jobs/${normalizedJobId}/input`,
|
|
user,
|
|
body,
|
|
);
|
|
|
|
// Sync to DynamoDB if connector type is provided
|
|
const connectorType = body.connector as 'jdbc' | 'singer' | 's3' | undefined;
|
|
if (connectorType && this.VALID_CONNECTORS.includes(connectorType)) {
|
|
await this.syncJobInputToDynamoDB(jobId, body, user, connectorType);
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
@Put('jobs/:jobId/memory')
|
|
@ApiOperation({ summary: 'Update job memory configuration' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async updateJobMemory(
|
|
@Param('jobId') jobId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
// Validate memory limit
|
|
if (body.amount) {
|
|
this.validateMemory(body.amount);
|
|
}
|
|
|
|
// Normalize job ID for Platform API (replace - with _)
|
|
const normalizedJobId = this.normalizeJobId(jobId);
|
|
|
|
return this.platformApiService.proxy(
|
|
'PUT',
|
|
`/jobs/${normalizedJobId}/memory`,
|
|
user,
|
|
body,
|
|
);
|
|
}
|
|
|
|
@Post('jobs/:jobId/reset-state')
|
|
@ApiOperation({ summary: 'Reset job state' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
async resetJobState(
|
|
@Param('jobId') jobId: string,
|
|
@Body() body: any,
|
|
@User() user: RequestUser,
|
|
) {
|
|
// Normalize job ID for Platform API (replace - with _)
|
|
const normalizedJobId = this.normalizeJobId(jobId);
|
|
|
|
return this.platformApiService.proxy(
|
|
'POST',
|
|
`/jobs/${normalizedJobId}/reset-state`,
|
|
user,
|
|
body,
|
|
);
|
|
}
|
|
|
|
@Delete('pipelines/:pipelineId/inputs/:inputId')
|
|
@ApiOperation({ summary: 'Mark a table as deleted and delete its associated 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 },
|
|
@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 });
|
|
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;
|
|
}
|
|
}
|
|
|
|
// ==================== JOBS - JDBC SYNC MODE ROUTES ====================
|
|
|
|
@Get('jobs/jdbc/configs/allowed_datatypes')
|
|
@ApiOperation({ summary: 'Get allowed datatypes for JDBC' })
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
async getJdbcAllowedDatatypes(@User() user: RequestUser) {
|
|
return this.platformApiService.proxy(
|
|
'GET',
|
|
'/jobs/jdbc/configs/allowed_datatypes',
|
|
user,
|
|
);
|
|
}
|
|
|
|
// ==================== HEALTH ROUTE ====================
|
|
|
|
@Get('health')
|
|
@ApiOperation({ summary: 'Platform API health check' })
|
|
@Authenticated()
|
|
async healthCheck(@User() user: RequestUser) {
|
|
return this.platformApiService.proxy('GET', '/health', user);
|
|
}
|
|
}
|