FEAT: add Platform API proxy with Elasticsearch and DynamoDB sync

- Add Platform API module to proxy requests to Platform API service
- Add Elasticsearch service for pipeline catalog sync (CRUD operations)
- Add DynamoDB service for storing pipeline tables/inputs configuration
- Sync pipeline creation/update/delete to Elasticsearch index
- Extract tables from jobs with connector-specific mappings:
  - JDBC: table_name, load_type, column_include_list, incremental_column_name
  - Singer: replication_method -> full_load/incremental
  - S3: same as Singer
- Map connector types to DynamoDB types (jdbc->database, singer->application, s3->file)
- Validate connector type is provided in job input
- Normalize pipeline IDs for Platform API (replace - with _), keep UUIDs for ES

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

package.json

@@ -27,7 +27,12 @@
     "test:e2e": "jest --config ./test/jest-e2e.json"
   },
   "dependencies": {
+    "@aws-crypto/sha256-js": "^5.2.0",
+    "@aws-sdk/client-dynamodb": "^3.414.0",
     "@aws-sdk/client-secrets-manager": "^3.414.0",
+    "@aws-sdk/lib-dynamodb": "^3.414.0",
+    "@aws-sdk/credential-provider-node": "^3.940.0",
+    "@aws-sdk/signature-v4": "^3.370.0",
     "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
     "@dadosfera/protospack": "2.5.3",
     "@dadosfera/protospack-v2": "3.38.0-beta.10",
@@ -78,8 +83,8 @@
     "multer": "1.4.5-lts.1"
   },
   "devDependencies": {
-    "@types/cookie-parser": "^1.4.9",
     "@types/cache-manager": "^4.0.6",
+    "@types/cookie-parser": "^1.4.9",
     "@types/express": "^4.17.17",
     "@types/express-session": "^1.18.1",
     "@types/jest": "27.0.2",

src/app.module.ts

@@ -33,6 +33,7 @@ import { NetworkPolicyModule } from './modules/network-policy/network-policy.mod
 import { AssignModule } from './modules/assign/assign.module';
 import { ShareMetadataModule } from './modules/share-metadata/share-metadata.module';
 import { ApiKeyModule } from './modules/api-key/api-key.module';
+import { PlatformApiModule } from './modules/platform-api/platform-api.module';
 
 @Module({
   providers: [
@@ -73,6 +74,7 @@ import { ApiKeyModule } from './modules/api-key/api-key.module';
     ApiKeyModule,
     IdentityProviderModule,
     NetworkPolicyModule,
+    PlatformApiModule,
     //Always leave HealthModule last, so it is on the bottom of swagger
     HealthModule,
   ],

src/modules/platform-api/platform-api.config.ts

@@ -0,0 +1,11 @@
+export const PLATFORM_API_CONFIG = {
+  getUrl: (): string => {
+    const url = process.env.PLATFORM_API_URL;
+    if (!url) {
+      throw new Error('PLATFORM_API_URL environment variable is not set');
+    }
+    return url;
+  },
+  region: process.env.AWS_REGION || 'us-east-1',
+  timeout: parseInt(process.env.PLATFORM_API_TIMEOUT || '30000', 10),
+};

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

@@ -0,0 +1,663 @@
+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 } from '../../services/dynamodb';
+
+@ApiTags('Platform API')
+@Controller('platform')
+export class PlatformApiController {
+  private logger: any;
+
+  constructor(
+    private readonly platformApiService: PlatformApiService,
+    private readonly elasticsearchService: ElasticsearchService,
+    private readonly dynamoDBService: DynamoDBService,
+    @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 readonly VALID_CONNECTORS = ['jdbc', 'singer', 's3'];
+
+  /**
+   * 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(', ')}`,
+      );
+    }
+  }
+
+  /**
+   * 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 (reference_column)
+   * - 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?: { name: string; type: string };
+  }> {
+    if (!jobs || jobs.length === 0) return [];
+
+    const tables: Array<{
+      name: string;
+      type: string;
+      columns?: string[];
+      reference_column?: { name: string; type: string };
+    }> = [];
+
+    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
+        const table: {
+          name: string;
+          type: string;
+          columns?: string[];
+          reference_column?: { name: string; type: string };
+        } = {
+          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) {
+          table.reference_column = {
+            name: input.incremental_column_name,
+            type: input.incremental_column_type || 'timestamp',
+          };
+        }
+        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;
+  }
+
+  // ==================== 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);
+
+    // 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,
+        });
+      }
+
+      this.logger.info('Syncing pipeline to Elasticsearch', {
+        customerName: user.customer_name,
+        pipelineId: body.id,
+        plugin,
+        connector: connectorType,
+        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,
+        },
+        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,
+  ) {
+    const normalizedId = this.normalizePipelineId(pipelineId);
+    const result = await this.platformApiService.proxy(
+      'PATCH',
+      `/pipeline/${normalizedId}`,
+      user,
+      body,
+    );
+
+    // Sync to Elasticsearch (use original UUID, not normalized)
+    try {
+      await this.elasticsearchService.updatePipeline(
+        user.customer_name,
+        pipelineId,
+        {
+          name: body.name,
+          description: body.description,
+          cron: body.cron,
+          status: body.status,
+        },
+      );
+    } 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) {
+    return this.platformApiService.proxy('POST', '/pipeline/execute', user, body);
+  }
+
+  @Post('pipeline/pause')
+  @ApiOperation({ summary: 'Pause a pipeline' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async pausePipeline(@Body() body: any, @User() user: RequestUser) {
+    return this.platformApiService.proxy('POST', '/pipeline/pause', user, body);
+  }
+
+  @Post('pipeline/unpause')
+  @ApiOperation({ summary: 'Unpause a pipeline' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async unpausePipeline(@Body() body: any, @User() user: RequestUser) {
+    return this.platformApiService.proxy('POST', '/pipeline/unpause', user, body);
+  }
+
+  @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,
+  ) {
+    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 ====================
+
+  @Post('pipeline/pipeline_run')
+  @ApiOperation({ summary: 'Create a pipeline run' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async createPipelineRun(@Body() body: any, @User() user: RequestUser) {
+    return this.platformApiService.proxy(
+      'POST',
+      '/pipeline/pipeline_run',
+      user,
+      body,
+    );
+  }
+
+  @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>,
+  ) {
+    return this.platformApiService.proxy(
+      'GET',
+      `/pipeline/${pipelineId}/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,
+  ) {
+    return this.platformApiService.proxy(
+      'GET',
+      `/pipeline/${pipelineId}/pipeline_run/${runId}`,
+      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>,
+  ) {
+    return this.platformApiService.proxy(
+      'GET',
+      `/pipeline/pipeline_run/${runId}/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,
+  ) {
+    return this.platformApiService.proxy(
+      'PUT',
+      `/jobs/${jobId}/input`,
+      user,
+      body,
+    );
+  }
+
+  @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,
+  ) {
+    return this.platformApiService.proxy(
+      'PATCH',
+      `/jobs/${jobId}/input`,
+      user,
+      body,
+    );
+  }
+
+  @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,
+  ) {
+    return this.platformApiService.proxy(
+      'PUT',
+      `/jobs/${jobId}/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,
+  ) {
+    return this.platformApiService.proxy(
+      'POST',
+      `/jobs/${jobId}/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) {
+    return this.platformApiService.proxy('GET', `/jobs/jdbc/${jobId}`, user);
+  }
+
+  @Put('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,
+  ) {
+    return this.platformApiService.proxy(
+      'PUT',
+      `/jobs/jdbc/${jobId}/sync-mode`,
+      user,
+      body,
+    );
+  }
+
+  @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) {
+    return this.platformApiService.proxy('GET', `/jobs/singer/${jobId}`, user);
+  }
+
+  @Put('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,
+  ) {
+    return this.platformApiService.proxy(
+      'PUT',
+      `/jobs/singer/${jobId}/sync-mode`,
+      user,
+      body,
+    );
+  }
+
+  // ==================== 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) {
+    return this.platformApiService.proxy('GET', `/jobs/s3/${jobId}`, user);
+  }
+
+  // ==================== HEALTH ROUTE ====================
+
+  @Get('health')
+  @ApiOperation({ summary: 'Platform API health check' })
+  @Authenticated()
+  async healthCheck(@User() user: RequestUser) {
+    return this.platformApiService.proxy('GET', '/health', user);
+  }
+}

src/modules/platform-api/platform-api.module.ts

@@ -0,0 +1,16 @@
+import { Module } from '@nestjs/common';
+
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+
+import { PlatformApiController } from './platform-api.controller';
+import { PlatformApiService } from './platform-api.service';
+import { ElasticsearchModule } from '../../services/elasticsearch';
+import { DynamoDBModule } from '../../services/dynamodb';
+
+@Module({
+  imports: [ElasticsearchModule, DynamoDBModule],
+  controllers: [PlatformApiController],
+  providers: [PlatformApiService, DadosferaLogger],
+  exports: [PlatformApiService],
+})
+export class PlatformApiModule {}

src/modules/platform-api/platform-api.service.ts

@@ -0,0 +1,123 @@
+import { Injectable, Inject, HttpException } from '@nestjs/common';
+import { SignatureV4 } from '@aws-sdk/signature-v4';
+import { Sha256 } from '@aws-crypto/sha256-js';
+import { defaultProvider } from '@aws-sdk/credential-provider-node';
+import axios, { AxiosResponse, Method } from 'axios';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+
+import { RequestUser } from '../../decorators/user.decorator';
+import { PLATFORM_API_CONFIG } from './platform-api.config';
+
+@Injectable()
+export class PlatformApiService {
+  private signer: SignatureV4;
+  private logger: any;
+
+  constructor(
+    @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
+  ) {
+    this.logger = dadosferaLogger.logger;
+    this.signer = new SignatureV4({
+      service: 'execute-api',
+      region: PLATFORM_API_CONFIG.region,
+      credentials: defaultProvider(),
+      sha256: Sha256,
+    });
+  }
+
+  async proxy(
+    method: string,
+    path: string,
+    user: RequestUser,
+    body?: any,
+    query?: Record<string, string>,
+  ): Promise<any> {
+    const baseUrl = PLATFORM_API_CONFIG.getUrl();
+    const url = new URL(`${baseUrl}${path}`);
+
+    // Add query params
+    if (query) {
+      Object.entries(query).forEach(([key, value]) => {
+        if (value !== undefined && value !== null) {
+          url.searchParams.set(key, String(value));
+        }
+      });
+    }
+
+    const headers: Record<string, string> = {
+      host: url.hostname,
+      'content-type': 'application/json',
+      // Forward user context headers
+      // Note: platform-api expects customer_name in the 'customer_id' header (contract inconsistency)
+      'customer_id': user.customer_name || '',
+      'customer_name': user.customer_name || '',
+      'x-user-id': user.user_id || '',
+      'x-username': user.username || '',
+      'x-customer-tier': user.customer_tier || '',
+      'x-customer-id': user.customer_id || '',
+    };
+
+    const requestToSign = {
+      method: method.toUpperCase(),
+      protocol: url.protocol,
+      hostname: url.hostname,
+      port: url.port ? parseInt(url.port, 10) : undefined,
+      path: url.pathname + url.search,
+      headers,
+      body: body ? JSON.stringify(body) : undefined,
+    };
+
+    this.logger.info('Proxying request to platform-api', {
+      method: method.toUpperCase(),
+      path,
+      customer_id: user.customer_id,
+      user_id: user.user_id,
+    });
+
+    try {
+      // Sign with IAM v4
+      const signedRequest = await this.signer.sign(requestToSign);
+
+      const response: AxiosResponse = await axios({
+        method: method as Method,
+        url: url.href,
+        headers: signedRequest.headers as Record<string, string>,
+        data: body,
+        timeout: PLATFORM_API_CONFIG.timeout,
+        validateStatus: () => true, // Don't throw on non-2xx
+      });
+
+      // Propagate non-2xx responses as HttpExceptions
+      if (response.status >= 400) {
+        throw new HttpException(response.data, response.status);
+      }
+
+      return response.data;
+    } catch (error) {
+      this.logger.error('Platform API proxy error', {
+        error: error.message,
+        status: error.response?.status,
+        path,
+        method: method.toUpperCase(),
+      });
+
+      if (error instanceof HttpException) {
+        throw error;
+      }
+
+      if (error.response) {
+        throw new HttpException(error.response.data, error.response.status);
+      }
+
+      if (error.code === 'ECONNREFUSED') {
+        throw new HttpException('Platform API service unavailable', 503);
+      }
+
+      if (error.code === 'ETIMEDOUT' || error.code === 'ECONNABORTED') {
+        throw new HttpException('Platform API request timeout', 504);
+      }
+
+      throw new HttpException('Internal server error', 500);
+    }
+  }
+}

src/services/dynamodb/dynamodb.config.ts

@@ -0,0 +1,4 @@
+export const DYNAMODB_CONFIG = {
+  region: () => process.env.AWS_REGION || 'us-east-1',
+  inputsTable: () => process.env.INPUTS_DB || 'dadosfera-inputs-prd',
+};

src/services/dynamodb/dynamodb.module.ts

@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+import { DynamoDBService } from './dynamodb.service';
+
+@Module({
+  providers: [DynamoDBService, DadosferaLogger],
+  exports: [DynamoDBService],
+})
+export class DynamoDBModule {}

src/services/dynamodb/dynamodb.service.ts

@@ -0,0 +1,148 @@
+import { Injectable, Inject } from '@nestjs/common';
+import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
+import {
+  DynamoDBDocumentClient,
+  GetCommand,
+  PutCommand,
+  DeleteCommand,
+  TranslateConfig,
+} from '@aws-sdk/lib-dynamodb';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+import { v4 as uuid } from 'uuid';
+import { DYNAMODB_CONFIG } from './dynamodb.config';
+
+export interface InputDocument {
+  id: string;
+  client_id: string;
+  user_id: string;
+  created_at: string;
+  name: string;
+  description?: string;
+  plugin: string;
+  type: string;
+  tables?: Array<{
+    name: string;
+    type: string;
+    columns?: string[];
+    reference_column?: { name: string; type: string };
+  }>;
+  credentials?: Record<string, any>;
+}
+
+@Injectable()
+export class DynamoDBService {
+  private documentClient: DynamoDBDocumentClient;
+  private logger: any;
+
+  constructor(
+    @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
+  ) {
+    this.logger = dadosferaLogger.logger;
+
+    const dynamoConfig = { region: DYNAMODB_CONFIG.region() };
+    const marshallOptions: TranslateConfig = {
+      marshallOptions: {
+        removeUndefinedValues: true,
+      },
+    };
+
+    const dynamoDb = new DynamoDBClient(dynamoConfig);
+    this.documentClient = DynamoDBDocumentClient.from(dynamoDb, marshallOptions);
+  }
+
+  async createInput(
+    clientId: string,
+    userId: string,
+    data: {
+      name: string;
+      description?: string;
+      plugin: string;
+      type: string;
+      tables?: Array<{
+        name: string;
+        type: string;
+        columns?: string[];
+        reference_column?: { name: string; type: string };
+      }>;
+    },
+  ): Promise<InputDocument> {
+    const tableName = DYNAMODB_CONFIG.inputsTable();
+    const id = uuid();
+    const created_at = new Date().toISOString();
+
+    const item: InputDocument = {
+      id,
+      client_id: clientId,
+      user_id: userId,
+      created_at,
+      name: data.name,
+      description: data.description,
+      plugin: data.plugin,
+      type: data.type,
+      tables: data.tables,
+    };
+
+    this.logger.info('DynamoDB: Creating input', {
+      tableName,
+      inputId: id,
+      plugin: data.plugin,
+    });
+
+    const putCommand = new PutCommand({
+      TableName: tableName,
+      Item: item,
+    });
+
+    try {
+      await this.documentClient.send(putCommand);
+      this.logger.info('DynamoDB: Input created successfully', { inputId: id });
+      return item;
+    } catch (error) {
+      this.logger.error('DynamoDB: Failed to create input', {
+        tableName,
+        inputId: id,
+        region: DYNAMODB_CONFIG.region(),
+        error: error.message,
+        errorName: error.name,
+      });
+      throw error;
+    }
+  }
+
+  async findInput(clientId: string, inputId: string): Promise<InputDocument | null> {
+    const tableName = DYNAMODB_CONFIG.inputsTable();
+
+    const getCommand = new GetCommand({
+      TableName: tableName,
+      Key: {
+        id: inputId,
+        client_id: clientId,
+      },
+    });
+
+    const { Item } = await this.documentClient.send(getCommand);
+
+    return Item as InputDocument | null;
+  }
+
+  async deleteInput(clientId: string, inputId: string): Promise<void> {
+    const tableName = DYNAMODB_CONFIG.inputsTable();
+
+    this.logger.info('DynamoDB: Deleting input', {
+      tableName,
+      inputId,
+    });
+
+    const deleteCommand = new DeleteCommand({
+      TableName: tableName,
+      Key: {
+        id: inputId,
+        client_id: clientId,
+      },
+    });
+
+    await this.documentClient.send(deleteCommand);
+
+    this.logger.info('DynamoDB: Input deleted successfully', { inputId });
+  }
+}

src/services/dynamodb/index.ts

@@ -0,0 +1,3 @@
+export * from './dynamodb.service';
+export * from './dynamodb.module';
+export * from './dynamodb.config';

src/services/elasticsearch/elasticsearch.config.ts

@@ -0,0 +1,5 @@
+export const ELASTICSEARCH_CONFIG = {
+  getUrl: () => process.env.ELASTICSEARCH_URL || 'http://localhost:9200',
+  getApiKey: () => process.env.ELASTICSEARCH_API_KEY || '',
+  timeout: 10000, // 10 seconds
+};

src/services/elasticsearch/elasticsearch.module.ts

@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+import { ElasticsearchService } from './elasticsearch.service';
+
+@Module({
+  providers: [ElasticsearchService, DadosferaLogger],
+  exports: [ElasticsearchService],
+})
+export class ElasticsearchModule {}

src/services/elasticsearch/elasticsearch.service.ts

@@ -0,0 +1,375 @@
+import { Injectable, Inject } from '@nestjs/common';
+import axios, { AxiosInstance, AxiosError } from 'axios';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+import { ELASTICSEARCH_CONFIG } from './elasticsearch.config';
+
+interface MultiLang {
+  'en-us': string;
+  'pt-br': string;
+  'es-es': string;
+}
+
+interface MultiLangArray {
+  'en-us': string[];
+  'pt-br': string[];
+  'es-es': string[];
+}
+
+interface ConnectorInfo {
+  plugin: string;
+  name: MultiLang;
+  image: string;
+  version: string;
+  tags: string[];
+}
+
+interface PipelineDocument {
+  id: string;
+  name: MultiLang;
+  description: MultiLang;
+  customer_id: string;
+  user_id: string;
+  username: string;
+  status: string;
+  created_at: string;
+  updated_at: string;
+  last_status_updated: string;
+  tags: string[];
+  // Connector metadata
+  connection_id: string;
+  connector_name: string;
+  connector_plugin: string;
+  connector_version: string;
+  image_url: string;
+  // Config
+  config: {
+    cron: string;
+    tables?: string;
+  };
+  properties: string;
+  type: string;
+  in_use: number;
+  keywords: MultiLangArray;
+}
+
+@Injectable()
+export class ElasticsearchService {
+  private client: AxiosInstance;
+  private logger: any;
+
+  constructor(
+    @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
+  ) {
+    this.logger = dadosferaLogger.logger;
+    this.client = axios.create({
+      baseURL: ELASTICSEARCH_CONFIG.getUrl(),
+      headers: {
+        Authorization: `ApiKey ${ELASTICSEARCH_CONFIG.getApiKey()}`,
+        'Content-Type': 'application/json',
+      },
+      timeout: ELASTICSEARCH_CONFIG.timeout,
+    });
+  }
+
+  private getIndex(customerName: string): string {
+    return `${customerName}_pipelines`;
+  }
+
+  private formatMultiLang(value: string): MultiLang {
+    return {
+      'en-us': value,
+      'pt-br': value,
+      'es-es': value,
+    };
+  }
+
+  private formatMultiLangArray(value: string[] = []): MultiLangArray {
+    return {
+      'en-us': value,
+      'pt-br': value,
+      'es-es': value,
+    };
+  }
+
+  buildPipelineDocument(
+    pipelineId: string,
+    data: {
+      name: string;
+      description?: string;
+      user_id: string;
+      username: string;
+      customer_id: string;
+      plugin: string;
+      connection_id: string;
+      cron?: string;
+      tables?: string;
+      properties?: Record<string, any>;
+      type?: string;
+      status?: string;
+      created_at?: string;
+      keywords?: string[];
+    },
+    connector: ConnectorInfo | null,
+  ): PipelineDocument {
+    const now = new Date().toISOString();
+
+    return {
+      id: pipelineId,
+      connection_id: data.connection_id,
+      connector_name: connector?.name?.['en-us'] || '',
+      connector_plugin: connector?.plugin || data.plugin,
+      connector_version: connector?.version || '1.0.0',
+      created_at: data.created_at || now,
+      updated_at: now,
+      customer_id: data.customer_id,
+      description: this.formatMultiLang(data.description || ''),
+      image_url: connector?.image || '',
+      keywords: this.formatMultiLangArray(data.keywords),
+      name: this.formatMultiLang(data.name || ''),
+      status: data.status || 'CREATED',
+      user_id: data.user_id,
+      username: data.username,
+      config: {
+        cron: data.cron,
+        tables: data.tables,
+      },
+      properties: data.properties ? JSON.stringify(data.properties) : '{}',
+      type: data.type,
+      in_use: 1,
+      last_status_updated: now,
+      tags: connector?.tags || [],
+    };
+  }
+
+  async getConnectorByPlugin(plugin: string): Promise<ConnectorInfo | null> {
+    this.logger.info('Elasticsearch: Looking up connector', { plugin });
+
+    try {
+      const response = await this.client.post('/connectors/_search', {
+        query: {
+          term: { plugin: plugin },
+        },
+        size: 1,
+      });
+
+      const hits = response.data.hits?.hits || [];
+      if (hits.length === 0) {
+        this.logger.warn('Elasticsearch: Connector not found', { plugin });
+        return null;
+      }
+
+      const source = hits[0]._source;
+      return {
+        plugin: source.plugin,
+        name: source.name,
+        image: source.image,
+        version: source.version,
+        tags: source.tags || [],
+      };
+    } catch (error) {
+      this.handleError('getConnectorByPlugin', error, { plugin });
+      return null;
+    }
+  }
+
+  async createPipeline(
+    customerName: string,
+    pipelineId: string,
+    data: {
+      name: string;
+      description?: string;
+      user_id: string;
+      username: string;
+      customer_id: string;
+      status?: string;
+      created_at?: string;
+      plugin: string;
+      connection_id: string;
+      cron?: string;
+      tables?: string;
+      properties?: Record<string, any>;
+      type?: string;
+    },
+    connector: ConnectorInfo | null,
+  ): Promise<any> {
+    const index = this.getIndex(customerName);
+    const document = this.buildPipelineDocument(pipelineId, data, connector);
+
+    this.logger.info('Elasticsearch: Creating pipeline', {
+      index,
+      pipelineId,
+      plugin: document.connector_plugin,
+    });
+
+    try {
+      const response = await this.client.post(
+        `${index}/_doc/${pipelineId}`,
+        document,
+        { params: { refresh: 'wait_for' } },
+      );
+
+      this.logger.info('Elasticsearch: Pipeline created successfully', {
+        pipelineId,
+        result: response.data.result,
+      });
+
+      return response.data;
+    } catch (error) {
+      this.handleError('createPipeline', error, { pipelineId, index });
+      throw error;
+    }
+  }
+
+  async updatePipeline(
+    customerName: string,
+    pipelineId: string,
+    changes: {
+      name?: string;
+      description?: string;
+      cron?: string;
+      status?: string;
+      tags?: string[];
+    },
+  ): Promise<any> {
+    const index = this.getIndex(customerName);
+    const now = new Date().toISOString();
+
+    const formattedChanges: Record<string, any> = {
+      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),
+    });
+
+    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 };
+
+      const response = await this.client.post(
+        `${index}/_update/${pipelineId}`,
+        updateBody,
+        { params: { refresh: 'wait_for' } },
+      );
+
+      this.logger.info('Elasticsearch: Pipeline updated successfully', {
+        pipelineId,
+        result: response.data.result,
+      });
+
+      return response.data;
+    } catch (error) {
+      this.handleError('updatePipeline', error, { pipelineId, index });
+      throw error;
+    }
+  }
+
+  async deletePipeline(
+    customerName: string,
+    pipelineId: string,
+  ): Promise<any> {
+    const index = this.getIndex(customerName);
+
+    this.logger.info('Elasticsearch: Deleting pipeline', {
+      index,
+      pipelineId,
+    });
+
+    try {
+      const response = await this.client.delete(
+        `${index}/_doc/${pipelineId}`,
+        { params: { refresh: 'wait_for' } },
+      );
+
+      this.logger.info('Elasticsearch: Pipeline deleted successfully', {
+        pipelineId,
+        result: response.data.result,
+      });
+
+      return response.data;
+    } catch (error) {
+      // If document not found, log warning but don't throw
+      if (error instanceof AxiosError && error.response?.status === 404) {
+        this.logger.warn('Elasticsearch: Pipeline not found for deletion', {
+          pipelineId,
+          index,
+        });
+        return { result: 'not_found' };
+      }
+
+      this.handleError('deletePipeline', error, { pipelineId, index });
+      throw error;
+    }
+  }
+
+  private handleError(
+    operation: string,
+    error: any,
+    context: Record<string, any>,
+  ): void {
+    if (error instanceof AxiosError) {
+      this.logger.error(`Elasticsearch: ${operation} failed`, {
+        ...context,
+        status: error.response?.status,
+        statusText: error.response?.statusText,
+        errorData: error.response?.data,
+        message: error.message,
+      });
+    } else {
+      this.logger.error(`Elasticsearch: ${operation} failed`, {
+        ...context,
+        message: error.message,
+        stack: error.stack,
+      });
+    }
+  }
+}

src/services/elasticsearch/index.ts

@@ -0,0 +1,3 @@
+export * from './elasticsearch.module';
+export * from './elasticsearch.service';
+export * from './elasticsearch.config';