maestro/src/modules/platform-api/platform-api.controller.ts

import {
  Controller,
  Get,
  Post,
  Put,
  Patch,
  Delete,
  Param,
  Body,
  Query,
  Inject,
  BadRequestException,
} from '@nestjs/common';
import { ApiTags, ApiOperation } from '@nestjs/swagger';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';

import {
  Authenticated,
  RequireAllPermissions,
} from '../../decorators/authentication.decorator';
import { User, RequestUser } from '../../decorators/user.decorator';
import { PlatformApiService } from './platform-api.service';
import { 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';

@ApiTags('Platform API')
@Controller('platform')
export class PlatformApiController {
  private logger: any;

  constructor(
    private readonly platformApiService: PlatformApiService,
    private readonly elasticsearchService: ElasticsearchService,
    private readonly dynamoDBService: DynamoDBService,
    private readonly customersService: CustomersService,
    @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, '_') || '';
  }

  /**
   * 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, '_') || '';
  }

  /**
   * 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 pipeline/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
    }
  }

  /**
   * Sync sync-mode changes to DynamoDB for JDBC connectors.
   * Always passes both target_load_type and incremental_column_name to ensure proper sync.
   */
  private async syncJdbcSyncModeToDynamoDB(
    jobId: string,
    body: any,
    user: RequestUser,
  ): Promise<void> {
    // JDBC sync mode uses target_load_type field
    const changes: any = {};

    if ('target_load_type' in body) {
      changes.target_load_type = body.target_load_type;
    }

    // Handle incremental_column_name:
    // - If provided in body, use that value
    // - If changing to full_load, explicitly clear it
    if ('incremental_column_name' in body) {
      changes.incremental_column_name = body.incremental_column_name;
      changes.incremental_column_type = body.incremental_column_type;
    } else if (body.target_load_type === 'full_load') {
      // Changing to full_load without specifying incremental_column - clear it
      changes.incremental_column_name = null;
    }

    await this.syncJobInputToDynamoDB(jobId, changes, user, 'jdbc');
  }

  /**
   * Sync sync-mode changes to DynamoDB for Singer connectors.
   */
  private async syncSingerSyncModeToDynamoDB(
    jobId: string,
    body: any,
    user: RequestUser,
  ): Promise<void> {
    // Singer sync mode uses replication_method field
    // Map to DynamoDB type: FULL_TABLE -> full_load, INCREMENTAL -> incremental
    if ('replication_method' in body) {
      const type = body.replication_method === 'INCREMENTAL' ? 'incremental' : 'full_load';
      await this.syncJobInputToDynamoDB(jobId, { load_type: type }, user, 'singer');
    }
  }

  // ==================== PIPELINE ROUTES ====================

  @Post('pipeline')
  @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('pipeline/: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('pipeline/: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('pipeline/: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('pipeline/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', '/pipeline/execute', user, enrichedBody);
  }

  @Post('pipeline/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('pipeline/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('pipeline/: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('pipeline/: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 RUN ROUTES ====================

  @Get('pipeline/: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('pipeline/: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('pipeline/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,
    );
  }

  // ==================== 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,
    );
  }

  // ==================== JOBS - JDBC SYNC MODE ROUTES ====================

  @Get('jobs/jdbc/:jobId')
  @ApiOperation({ summary: 'Get JDBC job details' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getJdbcJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
    // Normalize job ID for Platform API (replace - with _)
    const normalizedJobId = this.normalizeJobId(jobId);
    return this.platformApiService.proxy('GET', `/jobs/jdbc/${normalizedJobId}`, user);
  }

  @Post('jobs/jdbc/:jobId/sync-mode')
  @ApiOperation({ summary: 'Update JDBC job sync mode' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async updateJdbcSyncMode(
    @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(
      'POST',
      `/jobs/jdbc/${normalizedJobId}/sync-mode`,
      user,
      body,
    );

    // Sync to DynamoDB (pass raw jobId for pipeline extraction)
    await this.syncJdbcSyncModeToDynamoDB(jobId, body, user);

    return result;
  }

  @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,
    );
  }

  // ==================== JOBS - SINGER REPLICATION ROUTES ====================

  @Get('jobs/singer/:jobId')
  @ApiOperation({ summary: 'Get Singer job details' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getSingerJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
    // Normalize job ID for Platform API (replace - with _)
    const normalizedJobId = this.normalizeJobId(jobId);
    return this.platformApiService.proxy('GET', `/jobs/singer/${normalizedJobId}`, user);
  }

  @Post('jobs/singer/:jobId/sync-mode')
  @ApiOperation({ summary: 'Update Singer job sync mode' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async updateSingerSyncMode(
    @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(
      'POST',
      `/jobs/singer/${normalizedJobId}/sync-mode`,
      user,
      body,
    );

    // Sync to DynamoDB (pass raw jobId for pipeline extraction)
    await this.syncSingerSyncModeToDynamoDB(jobId, body, user);

    return result;
  }

  // ==================== JOBS - S3 ROUTES ====================

  @Get('jobs/s3/:jobId')
  @ApiOperation({ summary: 'Get S3 job details' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getS3Job(@Param('jobId') jobId: string, @User() user: RequestUser) {
    // Normalize job ID for Platform API (replace - with _)
    const normalizedJobId = this.normalizeJobId(jobId);
    return this.platformApiService.proxy('GET', `/jobs/s3/${normalizedJobId}`, user);
  }

  // ==================== HEALTH ROUTE ====================

  @Get('health')
  @ApiOperation({ summary: 'Platform API health check' })
  @Authenticated()
  async healthCheck(@User() user: RequestUser) {
    return this.platformApiService.proxy('GET', '/health', user);
  }
}