mirror of
https://github.com/dadosfera/maestro.git
synced 2026-10-16 18:59:08 +00:00
FEAT: add schedule limit validation and improve ES update
- Add cron-validation utility with Airflow presets support (@once, @daily, etc.) - Validate cron expressions against customer scheduleLimit from DUC - Fetch scheduleLimit via gRPC to ensure up-to-date configuration - Improve ES updatePipeline to fetch current doc and merge changes - Only update fields explicitly provided in PATCH request body - Preserve config.tables when updating config.cron 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.5
parent
df674dd441
commit
f7efb757bf
@@ -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<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'
|
||||
@@ -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', {
|
||||
|
||||
Reference in New Issue
Block a user