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