diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 133b239..d9006ef 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -23,6 +23,8 @@ import { PlatformApiService } from './platform-api.service'; import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum'; import { ElasticsearchService } from '../../services/elasticsearch'; import { DynamoDBService } from '../../services/dynamodb'; +import { CustomersService } from '../customers/customers.service'; +import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation'; @ApiTags('Platform API') @Controller('platform') @@ -33,6 +35,7 @@ export class PlatformApiController { private readonly platformApiService: PlatformApiService, private readonly elasticsearchService: ElasticsearchService, private readonly dynamoDBService: DynamoDBService, + private readonly customersService: CustomersService, @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, ) { this.logger = dadosferaLogger.logger; @@ -59,6 +62,22 @@ export class PlatformApiController { } } + /** + * 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 { + 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' @@ -201,6 +220,9 @@ export class PlatformApiController { 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 = { @@ -322,6 +344,11 @@ export class PlatformApiController { @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', @@ -331,16 +358,24 @@ export class PlatformApiController { ); // 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, - { - name: body.name, - description: body.description, - cron: body.cron, - status: body.status, - }, + esChanges, ); } catch (error) { this.logger.error('Failed to sync pipeline update to Elasticsearch', { diff --git a/src/modules/platform-api/platform-api.module.ts b/src/modules/platform-api/platform-api.module.ts index 427032a..c2b31e4 100644 --- a/src/modules/platform-api/platform-api.module.ts +++ b/src/modules/platform-api/platform-api.module.ts @@ -6,9 +6,10 @@ import { PlatformApiController } from './platform-api.controller'; import { PlatformApiService } from './platform-api.service'; import { ElasticsearchModule } from '../../services/elasticsearch'; import { DynamoDBModule } from '../../services/dynamodb'; +import { CustomersModule } from '../customers/customers.module'; @Module({ - imports: [ElasticsearchModule, DynamoDBModule], + imports: [ElasticsearchModule, DynamoDBModule, CustomersModule], controllers: [PlatformApiController], providers: [PlatformApiService, DadosferaLogger], exports: [PlatformApiService], diff --git a/src/services/elasticsearch/elasticsearch.service.ts b/src/services/elasticsearch/elasticsearch.service.ts index e03b127..281b5d6 100644 --- a/src/services/elasticsearch/elasticsearch.service.ts +++ b/src/services/elasticsearch/elasticsearch.service.ts @@ -234,70 +234,50 @@ export class ElasticsearchService { const index = this.getIndex(customerName); const now = new Date().toISOString(); - const formattedChanges: Record = { - updated_at: now, - }; - - if (changes.name !== undefined) { - formattedChanges.name = this.formatMultiLang(changes.name); - } - - if (changes.description !== undefined) { - formattedChanges.description = this.formatMultiLang(changes.description); - } - - if (changes.cron !== undefined) { - // Update cron inside config object using ES script - formattedChanges.config = { cron: changes.cron }; - } - - if (changes.status !== undefined) { - formattedChanges.status = changes.status; - formattedChanges.last_status_updated = now; - } - - if (changes.tags !== undefined) { - formattedChanges.tags = changes.tags; - } - this.logger.info('Elasticsearch: Updating pipeline', { index, pipelineId, - fields: Object.keys(formattedChanges), + fields: Object.keys(changes), }); try { - // Use script for partial config update to preserve other config fields - const updateBody = changes.cron !== undefined - ? { - script: { - source: ` - ctx._source.updated_at = params.updated_at; - if (params.name != null) ctx._source.name = params.name; - if (params.description != null) ctx._source.description = params.description; - if (params.cron != null) ctx._source.config.cron = params.cron; - if (params.status != null) { - ctx._source.status = params.status; - ctx._source.last_status_updated = params.last_status_updated; - } - if (params.tags != null) ctx._source.tags = params.tags; - `, - params: { - updated_at: now, - name: changes.name !== undefined ? this.formatMultiLang(changes.name) : null, - description: changes.description !== undefined ? this.formatMultiLang(changes.description) : null, - cron: changes.cron || null, - status: changes.status || null, - last_status_updated: changes.status !== undefined ? now : null, - tags: changes.tags || null, - }, - }, - } - : { doc: formattedChanges }; + // Fetch current document + const currentDoc = await this.client.get(`${index}/_doc/${pipelineId}`); + const current = currentDoc.data._source; + + // Build updated document, preserving existing values + const updated: Record = { + ...current, + updated_at: now, + }; + + if ('name' in changes) { + updated.name = this.formatMultiLang(changes.name); + } + + if ('description' in changes) { + updated.description = this.formatMultiLang(changes.description); + } + + if ('cron' in changes) { + updated.config = { + ...current.config, + cron: changes.cron, + }; + } + + if ('status' in changes) { + updated.status = changes.status; + updated.last_status_updated = now; + } + + if ('tags' in changes) { + updated.tags = changes.tags; + } const response = await this.client.post( - `${index}/_update/${pipelineId}`, - updateBody, + `${index}/_doc/${pipelineId}`, + updated, { params: { refresh: 'wait_for' } }, ); diff --git a/src/utils/cron-validation.ts b/src/utils/cron-validation.ts new file mode 100644 index 0000000..b46d0f6 --- /dev/null +++ b/src/utils/cron-validation.ts @@ -0,0 +1,90 @@ +import CronParser from 'cron-parser'; + +export enum ScheduleLimits { + MINUTE = 'minute', + HOUR = 'hour', + DAY = 'day', + UNLIMITED = 'unlimited', +} + +const SECONDS_IN_MINUTE = 60; +const SECONDS_IN_HOUR = 3600; +const SECONDS_IN_DAY = 86400; + +/** + * Airflow preset schedules mapped to cron expressions. + * @once is special - it means run only once (no recurring schedule). + */ +const AIRFLOW_PRESETS: Record = { + '@once': null, // No recurring schedule - always valid + '@hourly': '0 * * * *', // Every hour + '@daily': '0 0 * * *', // Every day at midnight + '@weekly': '0 0 * * 0', // Every week on Sunday + '@monthly': '0 0 1 * *', // First day of every month + '@yearly': '0 0 1 1 *', // First day of every year + '@annually': '0 0 1 1 *', // Same as @yearly +}; + +/** + * Convert Airflow preset to cron expression. + * Returns null for @once (no recurring schedule). + * Returns original string if not an Airflow preset. + */ +export function convertAirflowPresetToCron(schedule: string): string | null { + const preset = AIRFLOW_PRESETS[schedule.toLowerCase()]; + if (preset !== undefined) { + return preset; + } + return schedule; +} + +export function getMinimumIntervalSeconds(scheduleLimit: string): number { + switch (scheduleLimit) { + case ScheduleLimits.MINUTE: + return SECONDS_IN_MINUTE; + case ScheduleLimits.HOUR: + return SECONDS_IN_HOUR; + case ScheduleLimits.DAY: + return SECONDS_IN_DAY; + case ScheduleLimits.UNLIMITED: + default: + return 0; + } +} + +export function getCronIntervalSeconds(cron: string): number { + const interval = CronParser.parseExpression(cron); + const nextDate = interval.next().toDate(); + const afterNextDate = interval.next().toDate(); + return Math.floor((afterNextDate.getTime() - nextDate.getTime()) / 1000); +} + +export function validateCronAgainstScheduleLimit( + cron: string, + scheduleLimit: string, +): { valid: boolean; message?: string } { + if (!cron) return { valid: true }; + + // Convert Airflow presets to cron expressions + const cronExpression = convertAirflowPresetToCron(cron); + + // @once returns null - no recurring schedule, always valid + if (cronExpression === null) { + return { valid: true }; + } + + try { + const cronInterval = getCronIntervalSeconds(cronExpression); + const minInterval = getMinimumIntervalSeconds(scheduleLimit); + + if (cronInterval < minInterval) { + return { + valid: false, + message: `Schedule interval (${cronInterval}s) is below customer limit (${scheduleLimit}: ${minInterval}s minimum)`, + }; + } + return { valid: true }; + } catch (error) { + return { valid: false, message: `Invalid cron expression: ${error.message}` }; + } +}