Files
maestro/src/modules/platform-api/platform-api.service.ts
T
RafaelandClaude Opus 4.5 df674dd441 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>
2025-12-10 12:42:18 -03:00

124 lines
3.7 KiB
TypeScript

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