Compare commits

..
Author SHA1 Message Date
RafaelandClaude Opus 4.6 051fb6e4dd FIX: uppercase table_name and table_schema in Nimbus rename calls
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-16 10:39:55 -03:00
RafaelandClaude Opus 4.6 e3099aa2b2 FIX: uppercase table_name and table_schema in ES lookup
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-16 10:17:56 -03:00
RafaelandClaude Opus 4.6 12c61d9b5d FIX: rename-tables proxy path and ES lookup
- Fix proxy path: /jobs/jdbc/:jobId/rename-tables → /jobs/:jobId/rename-tables
- Remove pipeline_id from ES data asset lookup, search by table_name + table_schema only

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-13 18:08:39 -03:00
Rafael 99c2a9ecf5 UPDATE: force deployment of maestro 2026-03-12 18:02:49 -03:00
Rafael Santana c0f75d241f Merge pull request #449 from dadosfera/feat/rename-tables-catalog-sync
Feat/rename tables catalog sync
2026-03-12 17:57:44 -03:00
Rafael Santana 1e0fb78dff Merge branch 'beta' into feat/rename-tables-catalog-sync 2026-03-12 17:57:37 -03:00
Marcos Rodrigues Silva b70d37423d Merge pull request #454 from dadosfera/fix/header-validation
FIX: types
2026-03-11 10:27:31 -03:00
marcos-silva-rodrigues 4ddd5edcfd FIX: types 2026-03-11 10:26:33 -03:00
Marcos Rodrigues Silva 3bcbba9581 Merge pull request #453 from dadosfera/fix/header-validation
Fix/header validation
2026-03-10 17:35:31 -03:00
marcos-silva-rodrigues 67d47a9642 FIX: correct header 2026-03-10 17:35:05 -03:00
Marcos Rodrigues Silva 6f9c967c96 Merge pull request #452 from dadosfera/feat/qualify
FIX: stringify headers
2026-03-10 17:17:48 -03:00
marcos-silva-rodrigues b8bdc5beea FIX: stringify headers 2026-03-10 16:00:39 -03:00
Marcos Rodrigues Silva 269f70b309 Merge pull request #451 from dadosfera/feat/qualify
FIX: ghost commit
2026-03-10 15:22:36 -03:00
marcos-silva-rodrigues 93f452ae05 FIX: ghost commit 2026-03-10 15:19:18 -03:00
Marcos Rodrigues Silva e39378229f Merge pull request #450 from dadosfera/feat/qualify
feat: list headers
2026-03-10 14:43:43 -03:00
marcos-silva-rodrigues 61a4f724ef feat: list headers 2026-03-10 14:43:11 -03:00
RafaelandClaude Opus 4.6 38a9e21f5f feat: add rename-tables endpoint with catalog sync and rollback
Add POST /platform/jobs/:jobId/rename-tables that renames Snowflake
tables via platform-api and syncs the rename to Elasticsearch and
Nimbus (table-metadata, column-metadata, data-preview). If catalog
sync fails, all completed catalog steps are rolled back in reverse
order and the Snowflake rename is reverted.

- Support any connector type (jdbc, singer, s3) via getJobByAnyConnectorType
- Resolve old table names from output_config (raw/qualify)
- Skip qualify sync when output_config.qualify has no table_name
- Add findDataAssetByPipelineAndTable and updateDataAsset to ElasticsearchService
- Add renameTableOnNimbus, renameColumnMetadataOnNimbus, renameDataPreviewOnNimbus to CatalogService
- Add upstream error logging to PlatformApiService

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-05 17:29:12 -03:00
Marcos Rodrigues Silva 1a2153d62f Merge pull request #448 from dadosfera/fix/proxy-urls
Fix/proxy urls
2026-02-20 16:03:05 -03:00
Marcos Rodrigues Silva d25bfd147c Merge pull request #447 from dadosfera/fix/proxy-urls
Fix/proxy urls
2026-02-20 14:32:50 -03:00
marcos-silva-rodrigues d4ba45fd03 FIX: storage url 2026-02-20 14:28:55 -03:00
marcos-silva-rodrigues 82f1035a5e FIX: update storage port 2026-02-20 14:28:36 -03:00
Marcos Rodrigues Silva 6c57bac235 Merge pull request #445 from dadosfera/feat/qualify
FIX: storage url
2026-02-18 15:43:13 -03:00
marcos-silva-rodrigues cf8eed35a3 FIX: storage url 2026-02-18 15:32:05 -03:00
Marcos Rodrigues Silva 55fc85c544 Merge pull request #444 from dadosfera/feat/qualify
FIX: update storage port
2026-02-16 18:12:27 -03:00
marcos-silva-rodrigues 72ed637640 FIX: update storage port 2026-02-16 18:11:49 -03:00
Marcos Rodrigues Silva 890364f597 Merge pull request #443 from dadosfera/feat/qualify
Feat/qualify
2026-02-13 11:27:41 -03:00
marcos-silva-rodrigues 30eec733b2 FEAT: update proto 2026-02-13 10:01:46 -03:00
marcos-silva-rodrigues 30a41ba144 FEAT: qualify route update 2026-02-12 17:48:05 -03:00
Marcos Rodrigues Silva 2dc032e7e7 Merge pull request #442 from dadosfera/feat/qualify
FIX: platform routes
2026-02-05 10:33:39 -03:00
marcos-silva-rodrigues 7f5981731f merge 2026-02-05 10:32:07 -03:00
marcos-silva-rodrigues 66309c7bbe FIX: platform routes 2026-02-05 10:29:26 -03:00
Marcos Rodrigues Silva 15048eaf8a Merge pull request #441 from dadosfera/feat/qualify
FEAT: qualify contract
2026-02-04 17:09:19 -03:00
marcos-silva-rodrigues 0e169a3cbc FIX: protospack version 2026-02-04 16:50:20 -03:00
marcos-silva-rodrigues b5d933eaf3 FEAT: qualify contract 2026-02-04 16:43:49 -03:00
Marcos Rodrigues Silva 6fa9bf861a Merge pull request #440 from dadosfera/fix/proxy-urls
FIX: storage service dns
2026-02-02 16:38:00 -03:00
marcos-silva-rodrigues 9a217dff57 FIX: storage service dns 2026-02-02 16:37:01 -03:00
Marcos Rodrigues Silva e08734c97f Merge pull request #439 from dadosfera/fix/proxy-urls
FIX: send token by storage
2026-02-02 15:39:00 -03:00
marcos-silva-rodrigues cd4382c1ff FIX: send token by storage 2026-02-02 15:37:13 -03:00
Marcos Rodrigues Silva 54b75ce11b Merge pull request #438 from dadosfera/fix/proxy-urls
FIX: storage api
2026-02-02 14:43:00 -03:00
marcos-silva-rodrigues 0e038d0b12 FIX: storage api 2026-02-02 14:41:53 -03:00
Marcos Rodrigues Silva 85234fe0dd Merge pull request #437 from dadosfera/fix/proxy-urls
Fix/proxy urls
2026-02-02 14:28:04 -03:00
marcos-silva-rodrigues c97a02cb17 FIX: remove / 2026-02-02 14:27:08 -03:00
marcos-silva-rodrigues 878ec977b8 FIX: maestro proxy urls 2026-02-02 14:25:13 -03:00
Marcos Rodrigues Silva 8006867bc2 Merge pull request #436 from dadosfera/beta
Beta
2026-01-26 18:03:42 -03:00
Marcos Rodrigues Silva 784b0ef090 Merge pull request #435 from dadosfera/fix/validation-pipe
FIX: Update ValidationPipe for class-validator 0.14.0+ compatibility
2026-01-21 14:46:57 -03:00
Rafael Santana 3d6328fb0b Merge pull request #434 from dadosfera/fix/class-validator-0.14-compatibility
UPDATE: updating /me to get api key
2026-01-20 17:53:09 -03:00
Rafael 4246c7495e UPDATE: updating /me to get api key 2026-01-20 17:52:34 -03:00
vinicius gadea 3d40746ccb Merge pull request #433 from dadosfera/feat/storage-explorer-api-integration
FEAT: integrate Storage Explorer API with new endpoints and permissions
2026-01-14 09:37:53 -03:00
viniciusgadea 3ab2f8f27d CHORE: remove linkFileToTable endpoint from Storage Explorer controller 2026-01-14 08:51:45 -03:00
viniciusgadea b44d23552c CHORE: remove unecessary header for request 2026-01-14 06:40:52 -03:00
viniciusgadea 8e63757738 FEAT: add STORAGE_EXPLORER_API_URL to deployment and values configuration 2026-01-14 06:26:21 -03:00
viniciusgadea 3b84409003 CHORE: add unique authentication decorator to Storage Explorer controller methods 2026-01-14 06:25:58 -03:00
viniciusgadea 0ce3822300 CHORE: update new endpoint link-table and remove /delete from storage-explorer api 2026-01-13 14:36:43 -03:00
viniciusgadea a2c7ce00db FEAT: integrate Storage Explorer API with new endpoints and permissions 2026-01-13 10:29:38 -03:00
Rafael Santana 121e30ce45 Merge pull request #432 from dadosfera/fix/class-validator-0.14-compatibility
FIX: Update ValidationPipe for class-validator 0.14.0+ compatibility
2026-01-12 16:30:41 -03:00
RafaelandClaude 85c8a4937d FIX: Update ValidationPipe for class-validator 0.14.0+ compatibility
This fix addresses a breaking change introduced in class-validator 0.14.0
where the default for forbidUnknownValues changed from false to true.

Issue:
- POST /connections was returning 400 "an unknown value was passed to the
  validate function" errors
- This occurred because CreateConnectionDto and UpdateConnectionDto have no
  validation decorators, causing class-validator 0.14.0+ to treat them as
  "unknown values"
- Extra fields (like connector_version) in request payloads would fail
  validation

Solution:
- Set forbidUnknownValues: false to allow DTOs without validation decorators
- Set whitelist: true to automatically strip extra properties not defined
  in DTOs
- This maintains backward compatibility while adding security by removing
  unexpected fields

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

Co-Authored-By: Claude <noreply@anthropic.com>
2026-01-12 16:22:25 -03:00
Rafael Santana 53920f2f3f Merge pull request #431 from dadosfera/feature/mgc-keycloak-migration
Feature/mgc keycloak migration
2026-01-08 18:21:10 -03:00
RafaelandClaude Opus 4.5 8c34914806 FEAT: add AUTH_PROVIDER config to Helm chart
- Add auth_provider to values.yaml (cognito/keycloak)
- Add AUTH_PROVIDER env var to deployment template
- Note: maestro only needs to know which provider is used, duc handles connection

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-08 12:22:47 -03:00
RafaelandClaude Opus 4.5 1881d07c4a FEAT: add dual auth provider support to Helm chart
- Add auth_provider config to values.yaml (cognito/keycloak)
- Add AUTH_PROVIDER env var to deployment template
- Add conditional Keycloak env vars (URL, realm, client ID, secret)
- Enables parallel deployments with different auth providers

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-08 12:10:17 -03:00
RafaelandClaude Opus 4.5 3b8310fdca CHORE: simplify Dockerfile.local to use npm ci only
Remove protospack tarball fallback logic

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-07 19:04:07 -03:00
RafaelandClaude Opus 4.5 52bda8ebe2 FEAT: add authProvider field to user response
- Add authProvider to user entity DTO
- Update user service to include authProvider
- Update auth controller response

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-07 18:55:36 -03:00
Rafael 275a53dbd1 Merge remote-tracking branch 'origin/beta' into feature/mgc-keycloak-migration 2026-01-06 12:44:05 -03:00
vinicius gadea 9cefdb226d Merge pull request #430 from dadosfera/feat/organization-form
Feat/organization form
2026-01-06 10:52:33 -03:00
viniciusgadea 8ac0a8a79f REFACTOR: rename 'companySite' to 'personalSite' in user DTOs and related services 2026-01-05 15:43:35 -03:00
viniciusgadea 285de97375 FIX: update organization info field from 'name' to 'companyName' in updateOrganizationInfo method 2026-01-05 12:01:04 -03:00
viniciusgadea e2a7d2b92b FIX: update organization info field from 'name' to 'companyName' 2026-01-05 11:47:20 -03:00
viniciusgadea da23ad76db DOCS: update docs to organization forms 2026-01-02 09:22:27 -03:00
viniciusgadea 55b0961b82 FEAT: add organization info endpoints and DTOs for update and retrieval new organization forms 2026-01-02 09:21:46 -03:00
RafaelandClaude Opus 4.5 8e0182aa50 FEAT: add TOTP support for change password and local build support
- Pass TOTP code to DUC for Keycloak users with MFA
- Add Dockerfile.local for local protospack builds
- Add .dockerignore to exclude node_modules from Docker context

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-24 19:03:50 -03:00
36 changed files with 3443 additions and 1134 deletions
+12
View File
@@ -0,0 +1,12 @@
node_modules
dist
.git
*.log
npm-debug.log*
.DS_Store
.env
.env.*
coverage
.nyc_output
*.tgz
!protospack.tgz
+1 -1
View File
@@ -1,4 +1,4 @@
FROM node:20-alpine AS base_image
FROM node:22-alpine AS base_image
RUN npm install -g npm@latest
FROM base_image AS build_base
+1
View File
@@ -4,6 +4,7 @@
# Maestro
Maestro é a API principal da Dadosfera. É responsável pela comunicação do Frontend com nossos microsserviços.
```mermaid
@@ -48,6 +48,9 @@ spec:
{{- toYaml .Values.resources | nindent 12 }}
{{- end }}
env:
# Auth Provider Configuration (cognito or keycloak)
- name: AUTH_PROVIDER
value: {{ .Values.maestro.auth_provider | default "cognito" | quote }}
- name: AWS_IDENTITY_POOL_ID
value: {{ .Values.maestro.aws_identity_pool_id }}
- name: AWS_REGION
@@ -108,6 +111,8 @@ spec:
value: "{{ .Values.maestro.redis_tls }}"
- name: PLATFORM_API_URL
value: {{ .Values.maestro.platform_api_url }}
- name: STORAGE_EXPLORER_API_URL
value: {{ .Values.maestro.storage_explorer_api_url | quote }}
- name: JWT_PRIVATE_KEY
valueFrom:
secretKeyRef:
+1
View File
@@ -9,6 +9,7 @@ maestro:
cookie_secret: "ff7bc13823edb2ae50d248e5780bddc9d4b31c36"
redis_database: "1"
platform_api_url: https://xs2hkhq07k.execute-api.us-east-1.amazonaws.com
storage_explorer_api_url: "http://storage-explorer-{customer}.data-apps.svc.cluster.local:8000/api"
hostname: maestro.stg.dadosfera.ai
+4
View File
@@ -27,6 +27,9 @@ resources:
cpu: 2000m
memory: 2Gi
maestro:
# Auth provider: "cognito" (default) or "keycloak"
# Note: maestro doesn't connect to Keycloak directly, only duc does
auth_provider: "cognito"
aws_identity_pool_id: "us-east-1_Mrezsw9Sn"
duc_url: duc.dadosfera.ai
in_factory_url: in-factory.dadosfera.ai
@@ -44,6 +47,7 @@ maestro:
open_customer_id: f239718a-a271-4ef9-ae7e-02a2f0f3aa6e
open_group_id: 401573bb-334f-44b2-b30e-88d4cea31ae9
platform_api_url: https://oz8v2zid1e.execute-api.us-east-1.amazonaws.com
storage_explorer_api_url: "https://storage-explorer-{customer}.dadosfera.ai/api"
dedicated_proxy: ""
restricted_ip: ""
redis_host: "aaapzppmlyamkocqwstpo7zvopczyyiyuy6xzm2g6c5k4mq3a66be4a-0.redis.sa-saopaulo-1.oci.oraclecloud.com"
+2011 -1115
View File
File diff suppressed because it is too large Load Diff
+4 -4
View File
@@ -17,7 +17,7 @@
"@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.19",
"@dadosfera/protospack-v2": "3.38.0-beta.28",
"@grpc/grpc-js": "^1.9.3",
"@grpc/proto-loader": "^0.7.9",
"@nestjs/cli": "^9.5.0",
@@ -1744,9 +1744,9 @@
}
},
"node_modules/@dadosfera/protospack-v2": {
"version": "3.38.0-beta.19",
"resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.19.tgz",
"integrity": "sha512-qevGunl1IKv4POa2hhXNaDpqlh/bvnq5Hxo6534OHYPZ//S+XcRZgTrxFnylK3+uoROM8epF3Mey5GwynxY7Tw==",
"version": "3.38.0-beta.28",
"resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.28.tgz",
"integrity": "sha512-w3Au0qschqZJ6OSHVDKaR2KeeCF2ncuJ4k7iZQceOLo1zxUU3TpDyLYyp/rp1DoSG763uIgqm3kJUXv7RlmfNw==",
"license": "ISC",
"dependencies": {
"@grpc/grpc-js": "^1.9.3",
+1 -1
View File
@@ -35,7 +35,7 @@
"@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.19",
"@dadosfera/protospack-v2": "3.38.0-beta.28",
"@grpc/grpc-js": "^1.9.3",
"@grpc/proto-loader": "^0.7.9",
"@nestjs/cli": "^9.5.0",
+2
View File
@@ -34,6 +34,7 @@ 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';
import { StorageExplorerModule } from './modules/storage-explorer/storage-explorer.module';
@Module({
providers: [
@@ -75,6 +76,7 @@ import { PlatformApiModule } from './modules/platform-api/platform-api.module';
IdentityProviderModule,
NetworkPolicyModule,
PlatformApiModule,
StorageExplorerModule,
//Always leave HealthModule last, so it is on the bottom of swagger
HealthModule,
],
+29
View File
@@ -668,6 +668,35 @@ export const PERMISSIONS_GROUPS = {
},
},
},
STORAGE_EXPLORER: {
title: {
'pt-br': 'Storage Explorer',
'en-us': 'Storage Explorer',
'es-es': 'Storage Explorer',
},
permissions: {
READ: {
seqid: 51,
claim: 'storage-explorer:read',
usage: PermissionUsages.PUBLIC,
name: {
'pt-br': 'Ler dados do Storage Explorer',
'en-us': 'Read Storage Explorer data',
'es-es': 'Leer datos del Storage Explorer',
},
},
WRITE: {
seqid: 52,
claim: 'storage-explorer:write',
usage: PermissionUsages.PUBLIC,
name: {
'pt-br': 'Escrever dados no Storage Explorer',
'en-us': 'Write Storage Explorer data',
'es-es': 'Escribir datos en Storage Explorer',
},
},
},
},
};
export interface DadosferaModule {
name: string;
+1
View File
@@ -111,3 +111,4 @@ function configureSwagger(app: INestApplication) {
);
}
bootstrap();
+29 -5
View File
@@ -55,6 +55,7 @@ import jwt, { JwtPayload } from 'jsonwebtoken';
import { LanguageEnum } from 'src/utils/languages.enum';
import { Language } from 'src/decorators/language.decorator';
import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator';
import { ApiKeyService } from 'src/modules/api-key/api-key.service';
type CookiesValues = {
accessToken?: string;
@@ -74,6 +75,7 @@ export class AuthController {
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
private authClient: AuthClientService,
private apiKeyService: ApiKeyService,
) {
this.logger = dadosferaLogger.logger;
@@ -190,13 +192,14 @@ export class AuthController {
) {
this.logger.info('/auth - change-password');
const { oldPassword, newPassword } = body;
const { oldPassword, newPassword, totpCode } = body;
const { authorization: accessToken } = headers;
return this.authClient.changePassword({
accessToken,
oldPassword,
newPassword,
totpCode,
});
}
@@ -214,7 +217,8 @@ export class AuthController {
const { username } = body;
return this.authClient.resetPassword({ username }, metadata);
await this.authClient.resetPassword({ username }, metadata);
return { authProvider: process.env.AUTH_PROVIDER || 'cognito' };
}
@ApiInternalOnlyEndpoint()
@@ -474,12 +478,32 @@ export class AuthController {
@Get('me')
async getMe(@Req() req: Request, @Res() res: Response) {
this.logger.info('GET /auth/me ')
this.logger.info(JSON.stringify(req.headers));
// Check for API key header first
const apiKey = req.get('X-Api-key');
if (apiKey) {
this.logger.info('Authenticating via X-Api-key header');
const { api_key } = await this.apiKeyService.get(apiKey);
const userDto = {
id: api_key.user_id,
name: api_key.username,
customer: {
id: api_key.customer_id,
name: api_key.customer_name,
tier: api_key.customer_tier,
}
};
return res.status(200).json(userDto);
}
// Get token and headers
const accessToken = req.cookies['ddf-auth'];
const refreshToken = req.cookies['ddf-refresh-auth'];
const userId = req.cookies['ddf-user-id'];
const resourceHost = req.headers["host"]
const refreshToken = req.cookies['ddf-refresh-auth'];
const userId = req.cookies['ddf-user-id'];
const resourceHost = req.headers["x-original-url"] as string || "" ;
const hasUserSession = Boolean(accessToken) && Boolean(userId);
this.logger.info('Has User Session: ' + hasUserSession);
+2 -1
View File
@@ -8,10 +8,11 @@ import { AuthClientService } from './auth.service';
import { DucClient } from '../duc/client.config';
import { GoogleLoginStrategy } from './passport-strategies/google-strategy';
import { getOauthSecrets } from 'src/utils/OauthSecrets';
import { ApiKeyModule } from '../api-key/api-key.module';
const client = new DucClient();
@Module({
imports: [ClientsModule.register([client.providerOptions])],
imports: [ClientsModule.register([client.providerOptions]), ApiKeyModule],
controllers: [AuthController],
providers: [
AuthClientService,
+2
View File
@@ -143,6 +143,7 @@ export class AuthClientService implements OnModuleInit {
accessToken,
oldPassword,
newPassword,
totpCode,
}: AuthChangePasswordRequest) {
this.logger.info('ChangePassword');
@@ -151,6 +152,7 @@ export class AuthClientService implements OnModuleInit {
accessToken,
oldPassword,
newPassword,
totpCode,
}),
);
}
+58
View File
@@ -734,6 +734,64 @@ class CatalogService implements OnModuleInit {
}
}
async renameTableOnNimbus(
nimbusUrl: string,
nimbusId: number,
changes: { table_name?: string; table_schema?: string; display_name?: string },
): Promise<void> {
const endpoint = `${nimbusUrl}/api/catalog/table-metadata/${nimbusId}`;
this.logger.info(`Renaming table-metadata ${nimbusId} on Nimbus`, { endpoint, changes });
await axios.patch(endpoint, changes);
}
async renameColumnMetadataOnNimbus(
nimbusUrl: string,
databaseName: string,
oldTableName: string,
oldTableSchema: string,
newTableName: string,
newTableSchema: string,
): Promise<void> {
const listEndpoint = `${nimbusUrl}/api/catalog/column-metadata/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
this.logger.info(`Fetching column-metadata records to rename`, { listEndpoint });
const { data: columns } = await axios.get(listEndpoint);
const filtered = Array.isArray(columns) ? columns : [];
for (const column of filtered) {
const patchEndpoint = `${nimbusUrl}/api/catalog/column-metadata/${column.id}`;
await axios.patch(patchEndpoint, {
table_name: newTableName,
table_schema: newTableSchema,
});
}
this.logger.info(`Renamed ${filtered.length} column-metadata records on Nimbus`);
}
async renameDataPreviewOnNimbus(
nimbusUrl: string,
databaseName: string,
oldTableName: string,
oldTableSchema: string,
newTableName: string,
newTableSchema: string,
): Promise<void> {
const listEndpoint = `${nimbusUrl}/api/catalog/data-preview/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
this.logger.info(`Fetching data-preview records to rename`, { listEndpoint });
const { data: previews } = await axios.get(listEndpoint);
const filtered = Array.isArray(previews) ? previews : [];
for (const preview of filtered) {
const patchEndpoint = `${nimbusUrl}/api/catalog/data-preview/${preview.id}`;
await axios.patch(patchEndpoint, {
table_name: newTableName,
table_schema: newTableSchema,
});
}
this.logger.info(`Renamed ${filtered.length} data-preview records on Nimbus`);
}
async catalogDatasetItem(table_metadata_id: number, metadata: Metadata) {
const customer_name_raw = metadata.get('customer_name');
@@ -136,4 +136,32 @@ export class CustomersController {
const result = await this.customersService.getAccessDashboardUrl(user.customer_name, metadata);
return result;
}
@Get(':id/organization-info')
@Authenticated()
@RequireAllPermissions(PERMISSIONS_GROUPS.USERS.permissions.ADMIN)
@ApiOkResponse({ description: 'Organization information' })
async getOrganizationInfo(@Param('id') id: string) {
this.logger.info('getOrganizationInfo', { id });
return this.customersService.getOrganizationInfo(id);
}
@Put(':id/organization-info')
@Authenticated()
@RequireAllPermissions(PERMISSIONS_GROUPS.USERS.permissions.ADMIN)
@HttpCode(HttpStatus.OK)
@ApiOkResponse({ description: 'Organization information updated' })
async updateOrganizationInfo(
@Param('id') id: string,
@Body() body: {
companyName: string;
companySite: string;
domain: string;
cnpj: string;
description: string;
},
) {
return this.customersService.updateOrganizationInfo(id, body);
}
}
+54 -1
View File
@@ -223,4 +223,57 @@ export class CustomersService implements OnModuleInit {
})
)
}
}
async updateOrganizationInfo(
customerId: string,
data: {
companyName: string;
companySite: string;
domain: string;
cnpj: string;
description: string;
},
) {
try {
const result = await lastValueFrom(
this.customerService.OrganizationUpdate({
customerId,
companyName: data.companyName || '',
companySite: data.companySite || '',
domain: data.domain || '',
cnpj: data.cnpj || '',
description: data.description || '',
}),
);
return result;
} catch (err) {
if (err.details === ErrorCodes.CUSTOMER.NOT_FOUND)
throw new HttpException(err.details, HttpStatus.NOT_FOUND);
else throw err;
}
}
async getOrganizationInfo(customerId: string) {
try {
const customerResponse = await lastValueFrom(
this.customerService.CustomerFindOneById({ id: customerId })
);
const customer = customerResponse.customer;
return {
companyName: customer.companyName || '',
companySite: customer.companySite || '',
domain: customer.domain || '',
cnpj: customer.cnpj || '',
description: customer.description || ''
};
} catch (err) {
if (err.details === ErrorCodes.CUSTOMER.NOT_FOUND)
throw new HttpException(err.details, HttpStatus.NOT_FOUND);
else throw err;
}
}
}
@@ -0,0 +1,27 @@
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
export class OrganizationUpdateRequest {
@ApiProperty()
name: string;
@ApiPropertyOptional()
companySite: string;
@ApiProperty()
domain: string;
@ApiPropertyOptional()
info: string;
@ApiPropertyOptional()
cnpj: string;
}
export class OrganizationResponse {
@ApiProperty()
name: string;
@ApiPropertyOptional()
companySite: string;
@ApiProperty()
domain: string;
@ApiPropertyOptional()
info: string;
@ApiPropertyOptional()
cnpj: string;
}
+7
View File
@@ -14,6 +14,13 @@ export class TableColumns {
@ApiProperty()
references: Column[];
@ApiProperty()
identifier_columns: string[];
@ApiProperty()
destination: Record<'raw' | 'qualify', {
table_name: string;
table_schema: string;
}> | null;
@ApiProperty()
type: string;
}
export class AvailableEntity {
+1 -1
View File
@@ -200,7 +200,7 @@ export class InputsService {
}
async update(id: string, data, info: Info) {
this.validateCron({ ...data, info });
// this.validateCron({ ...data, info });
try {
const updateInputResponse: any = await this.OLD_inputClient.update({
id,
+36
View File
@@ -1,6 +1,15 @@
import { ApiProperty, ApiPropertyOptional, OmitType } from '@nestjs/swagger';
import { Info } from '@dadosfera/protospack/dist/lib/interfaces';
export class PipelineInputsDTO {
@ApiProperty()
tables: Array<{
name: string,
type: string,
}>
}
export class IPipelineV2 {
@ApiProperty()
id: string;
@@ -123,3 +132,30 @@ export class PipelineFindAllReq {
@ApiPropertyOptional()
type?: string | undefined;
}
export interface UpdateTableDTO {
name: string;
type: string;
columns: string[];
destinations: {
raw: {
table_schema: string;
table_name: string;
};
qualify: {
table_schema: string;
table_name: string;
};
};
identifier_columns: string[];
reference_column: {
name: string;
type: string;
};
memory: number;
}
export interface UpdatePlatformInputRequest {
cron: string;
tables: Array<UpdateTableDTO>;
}
@@ -43,11 +43,15 @@ import {
IPipelineV2,
IInitUploadCSVFile,
PipelineFindAllReq,
UpdatePlatformInputRequest,
} from './interfaces';
import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
import { LanguageEnum } from 'src/utils/languages.enum';
import { Language } from 'src/decorators/language.decorator';
import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator';
import { TableColumns } from '../inputs/dtos/input.model';
import { UpdateInputRequest } from '../inputs/dtos/old_interfaces';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
@ApiTags('PipelinesV2')
@ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
@@ -223,6 +227,7 @@ export class PipelinesController {
.then((res) => {
//{pipeline:{tables: {tables: [], input_id: ''}}}
let tables = JSON.parse(res.pipeline.config.tables);
const input_id = tables?.input_id;
if (tables?.tables) tables = tables.tables;
Object.assign(res.pipeline, {
transformations: res.pipeline.transformations
@@ -231,6 +236,7 @@ export class PipelinesController {
config: {
cron: res.pipeline.config.cron,
tables,
input_id
},
properties: res.pipeline.properties
? JSON.parse(res.pipeline.properties)
@@ -277,6 +283,45 @@ export class PipelinesController {
return response;
}
@Patch('/:pipelineId/inputs/:id')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updatePipelineInput(
@Language() language: LanguageEnum,
@Body() pipelineInputDTO: UpdatePlatformInputRequest,
@Param('id') inputId: string,
@Param('pipelineId') pipelineId: string,
@User() user: RequestUser,
) {
this.logger.info('PipelinesController - update', { user });
const { customer_id, customer_name, user_id, username } = user;
const info: Info = {
user_id: user.user_id,
customer: user.customer_name,
customer_id: user.customer_id,
};
const metadata = PackTheMetadata({
customer_id,
customer_name,
user_id,
username,
language,
});
const response = await this.pipelinesClientService.updatePipelineInput(
pipelineId,
inputId,
pipelineInputDTO,
info,
user,
metadata,
);
this.logger.info('PipelinesController - update: OK', { user });
return response;
}
@ApiInternalOnlyEndpoint()
@Put('/:id')
@ApiOperation({
@@ -11,6 +11,7 @@ import { PipelinesModule as OldPipelineModule } from 'src/modules/pipelines/pipe
import { ConnectorModule } from '../connector/connector.module';
import { InputsModule } from '../inputs/inputs.module';
import { TransformationsModule } from '../transformations/transformations.module';
import { PlatformApiModule } from '../platform-api/platform-api.module';
const client = new PipelinesClientConfiguration();
@@ -21,6 +22,7 @@ const client = new PipelinesClientConfiguration();
ConnectorModule,
InputsModule,
TransformationsModule,
PlatformApiModule
],
controllers: [PipelinesController],
providers: [PipelinesService, DadosferaLogger],
+139 -1
View File
@@ -1,3 +1,4 @@
/* eslint-disable no-async-promise-executor */
import {
BadRequestException,
HttpException,
@@ -16,7 +17,7 @@ import { lastValueFrom } from 'rxjs';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import { PipelinesClientConfiguration } from './pipelines-client';
import { ICreatePipelineV2Req } from './interfaces';
import { ICreatePipelineV2Req, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces';
import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
import { Metadata } from '@grpc/grpc-js';
import { ConnectorClientService } from '../connector/client.service';
@@ -26,6 +27,8 @@ import { TransformationsService } from '../transformations/transformations.servi
import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath';
import ErrorCodes from 'src/utils/errorCodes';
import ErrorBuilder from 'src/utils/ErrorBuilder';
import { PlatformApiService } from '../platform-api/platform-api.service';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
export class PipelinesService implements OnModuleInit {
logger: DadosferaLogger;
@@ -39,6 +42,7 @@ export class PipelinesService implements OnModuleInit {
private readonly connectorService: ConnectorClientService,
private readonly inputsService: InputsService,
private readonly transformationsService: TransformationsService,
private readonly platformAPI: PlatformApiService
) {
this.logger = dadosferaLogger.logger;
}
@@ -138,6 +142,7 @@ export class PipelinesService implements OnModuleInit {
const findOnePipelineResponse = await lastValueFrom(
this.pipelineReadService.PipelineV2FindOne(data, metadata),
);
console.log('pipeline find one response', findOnePipelineResponse);
this.logger.info('Done');
return findOnePipelineResponse;
@@ -339,4 +344,137 @@ export class PipelinesService implements OnModuleInit {
return res;
}
async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) {
this.logger.info('InputClientService - Update');
this.logger.info('Update Dynamo Reference');
const pipelineIdFormat = pipelineId.split('-').join('_');
const updateInputResponse = await this.inputsService.update(
inputId,
updateInputDTO,
info
)
const requests = [];
this.logger.info('Dynamo Response', updateInputResponse);
for (const [index, table] of updateInputDTO.tables.entries()) {
const id = `${pipelineIdFormat}_${index}`;
this.logger.info('Updating input reference for table', table.name);
const body = {}
if (table.columns) {
body['column_include_list'] = table.columns;
}
if (table.reference_column) {
body['incremental_column_name'] = table.reference_column.name;
body['incremental_column_type'] = table.reference_column.type;
}
if (table.identifier_columns) {
body['primary_keys'] = table.identifier_columns;
}
this.logger.info('Request body', body);
const updateCollumns = this.platformAPI.proxy(
'PATCH',
`/jobs/${id}/input`,
user,
body
)
requests.push(updateCollumns);
if (table.memory) {
this.logger.info('Updating memory allocation for table', table.name);
const updateMemory = this.platformAPI.proxy(
'PUT',
`/jobs/${id}/memory`,
user,
{
amount: table.memory
}
)
requests.push(updateMemory);
}
if (table.type) {
const updateSyncMode = this.updatePipelineSyncMode(table, id, user);
requests.push(updateSyncMode);
}
}
this.logger.info('Create Platform Request for each JOB');
if (updateInputDTO.cron) {
const crnUpdatedRequest = new Promise(async (resolve, reject) => {
const response = await this.updatePipelineCron(updateInputDTO.cron, pipelineIdFormat, user);
if (response.error) {
this.logger.error('Error updating pipeline cron', response.error);
return reject(new ErrorBuilder(response.error));
}
this.logger.error('Pipeline cron updated successfully', response);
return resolve(response);
});
requests.push(crnUpdatedRequest);
}
this.logger.info('Executing all request for the platform api');
const results = await Promise.allSettled(requests);
this.logger.info('Platform api response', results);
return updateInputResponse;
}
private async updatePipelineSyncMode(table: UpdateTableDTO, pipelineId: string, user: RequestUser) {
const body = {
target_load_type: table.type
}
if (table.type === 'incremental_with_qualify') {
body['incremental_column_name'] = table.reference_column.name;
body['incremental_column_type'] = table.reference_column.type;
body['primary_keys'] = table.identifier_columns;
}
if (table.type === 'incremental') {
body['incremental_column_name'] = table.reference_column.name;
body['incremental_column_type'] = table.reference_column.type;
}
this.logger.info('Updating pipeline sync mode', {
pipelineId,
body
});
return this.platformAPI.proxy(
"POST",
`/jobs/jdbc/${pipelineId}/sync-mode`,
user,
body
)
}
private async updatePipelineCron(cron: string, pipelineId: string, user: RequestUser) {
try {
const response = await this.platformAPI.proxy(
'PATCH',
`/pipeline/${pipelineId}`,
user,
{
cron
}
);
return response
} catch (error) {
return {
error: error.message
}
}
}
}
@@ -10,6 +10,7 @@ import {
Query,
Inject,
BadRequestException,
HttpException,
} from '@nestjs/common';
import { ApiTags, ApiOperation } from '@nestjs/swagger';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
@@ -25,6 +26,22 @@ 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';
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')
@@ -36,6 +53,7 @@ export class PlatformApiController {
private readonly elasticsearchService: ElasticsearchService,
private readonly dynamoDBService: DynamoDBService,
private readonly customersService: CustomersService,
private readonly catalogService: CatalogService,
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
) {
this.logger = dadosferaLogger.logger;
@@ -67,6 +85,23 @@ export class PlatformApiController {
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").
@@ -720,6 +755,41 @@ export class PlatformApiController {
);
}
// ==================== 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('pipeline/:pipelineId/pipeline_run')
@@ -909,6 +979,192 @@ export class PlatformApiController {
return result;
}
@Post('jobs/:jobId/rename-tables')
@ApiOperation({ summary: 'Rename job output tables and sync to catalog' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async renameJobTables(
@Param('jobId') jobId: string,
@Body() body: RenameTablesBody,
@User() user: RequestUser,
) {
const normalizedJobId = this.normalizeJobId(jobId);
const currentJob = await this.getJobByAnyConnectorType(normalizedJobId, user);
const result = await this.platformApiService.proxy(
'POST',
`/jobs/${normalizedJobId}/rename-tables`,
user,
body,
);
try {
await this.syncTableRenameToCatalog(jobId, body, currentJob, user);
} catch (error) {
this.logger.error('Catalog sync failed, rolling back Snowflake rename', { jobId, error: error.message });
const reverseBody = this.buildSnowflakeRollbackBody(body, currentJob.output_config || {});
if (reverseBody) {
try {
await this.platformApiService.proxy('POST', `/jobs/${normalizedJobId}/rename-tables`, user, reverseBody);
this.logger.info('Snowflake rename rolled back', { jobId });
} catch (rollbackError) {
this.logger.error('Snowflake rollback failed', { jobId, error: rollbackError.message });
}
}
throw new HttpException('Table rename failed: catalog sync error, Snowflake reverted', 500);
}
return result;
}
private buildSnowflakeRollbackBody(
body: RenameTablesBody,
outputConfig: any,
): RenameTablesBody | null {
const reverse: RenameTablesBody = {};
if (body.raw) {
const nested = outputConfig.raw;
const oldTableName = nested?.table_name || outputConfig.table_name;
const oldTableSchema = nested?.table_schema || 'PUBLIC';
if (oldTableName) reverse.raw = { table_name: oldTableName, table_schema: oldTableSchema };
}
if (body.qualify) {
const nested = outputConfig.qualify;
if (nested?.table_name) reverse.qualify = { table_name: nested.table_name, table_schema: nested.table_schema || 'STAGED' };
}
return Object.keys(reverse).length > 0 ? reverse : null;
}
/**
* Sync table rename to Elasticsearch and Nimbus.
*
* For each target (raw, qualify):
* 1. Resolve old table name from output_config
* 2. Find the ES data asset by pipeline + table + schema
* 3. Update ES, Nimbus table-metadata, column-metadata, and data-preview
* 4. If any step fails, rollback all completed steps for that target
*/
private async syncTableRenameToCatalog(
jobId: string,
body: RenameTablesBody,
currentJob: any,
user: RequestUser,
): Promise<void> {
const pipelineId = this.extractPipelineIdFromJobId(jobId);
const outputConfig = currentJob.output_config || {};
const nimbusUrl = this.catalogService._getNimbusUrl({ info: { customer: user.customer_name } });
const databaseName = `DADOSFERA_PRD_${user.customer_name.toUpperCase()}`;
const targets = this.buildRenameTargets(body, outputConfig);
for (const { key, oldTableName, oldTableSchema, newValues } of targets) {
const rollbackSteps: Array<() => Promise<void>> = [];
try {
const dataAsset = await this.elasticsearchService.findDataAssetByTable(
user.customer_name, oldTableName, oldTableSchema,
);
if (!dataAsset) {
this.logger.warn(`No data asset found for ${key}`, { jobId, pipelineId, oldTableName, oldTableSchema });
continue;
}
const { _es_id: esAssetId, nimbus_id: nimbusId } = dataAsset;
const oldValues = { table_name: oldTableName, table_schema: oldTableSchema };
// ES update
const esFields = { name: newValues.table_name, table_name: newValues.table_name, table_schema: newValues.table_schema, display_name: newValues.table_name };
await this.elasticsearchService.updateDataAsset(user.customer_name, esAssetId, esFields);
rollbackSteps.push(() => this.elasticsearchService.updateDataAsset(
user.customer_name, esAssetId,
{ name: oldTableName, table_name: oldTableName, table_schema: oldTableSchema, display_name: oldTableName },
));
const newTableNameUpper = newValues.table_name.toUpperCase();
const newTableSchemaUpper = newValues.table_schema.toUpperCase();
const oldTableNameUpper = oldTableName.toUpperCase();
const oldTableSchemaUpper = oldTableSchema.toUpperCase();
// Nimbus table-metadata
if (nimbusId) {
await this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, { table_name: newTableNameUpper, table_schema: newTableSchemaUpper });
rollbackSteps.push(() => this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, { table_name: oldTableNameUpper, table_schema: oldTableSchemaUpper }));
}
// Nimbus column-metadata
await this.catalogService.renameColumnMetadataOnNimbus(
nimbusUrl, databaseName, oldTableNameUpper, oldTableSchemaUpper, newTableNameUpper, newTableSchemaUpper,
);
rollbackSteps.push(() => this.catalogService.renameColumnMetadataOnNimbus(
nimbusUrl, databaseName, newTableNameUpper, newTableSchemaUpper, oldTableNameUpper, oldTableSchemaUpper,
));
// Nimbus data-preview
await this.catalogService.renameDataPreviewOnNimbus(
nimbusUrl, databaseName, oldTableNameUpper, oldTableSchemaUpper, newTableNameUpper, newTableSchemaUpper,
);
rollbackSteps.push(() => this.catalogService.renameDataPreviewOnNimbus(
nimbusUrl, databaseName, newTableNameUpper, newTableSchemaUpper, oldTableNameUpper, oldTableSchemaUpper,
));
this.logger.info(`Synced catalog rename for ${key}`, { jobId, oldTableName, newTableName: newValues.table_name });
} catch (error) {
this.logger.error(`Catalog sync failed for ${key}, rolling back catalog`, { jobId, error: error.message });
await this.executeRollback(rollbackSteps, key, jobId);
throw error;
}
}
}
private buildRenameTargets(
body: RenameTablesBody,
outputConfig: any,
): Array<{ key: string; oldTableName: string; oldTableSchema: string; newValues: { table_name: string; table_schema: string } }> {
const DEFAULT_SCHEMAS = { raw: 'PUBLIC', qualify: 'STAGED' };
const targets: Array<{ key: string; oldTableName: string; oldTableSchema: string; newValues: { table_name: string; table_schema: string } }> = [];
for (const key of ['raw', 'qualify'] as const) {
if (!body[key]) continue;
const nested = outputConfig[key];
// qualify: only sync if output_config.qualify already exists
if (key === 'qualify' && !nested?.table_name) continue;
const oldTableName = nested?.table_name || outputConfig.table_name;
if (!oldTableName) continue;
targets.push({
key,
oldTableName,
oldTableSchema: nested?.table_schema || DEFAULT_SCHEMAS[key],
newValues: body[key],
});
}
return targets;
}
private async executeRollback(
steps: Array<() => Promise<void>>,
targetKey: string,
jobId: string,
): Promise<void> {
for (const rollback of steps.reverse()) {
try {
await rollback();
} catch (error) {
this.logger.error(`Rollback failed for ${targetKey}`, { jobId, error: error.message });
}
}
}
@Get('jobs/jdbc/configs/allowed_datatypes')
@ApiOperation({ summary: 'Get allowed datatypes for JDBC' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
@@ -0,0 +1,9 @@
import { ApiProperty } from "@nestjs/swagger";
export class ValidationTableDTO {
@ApiProperty()
tables: Array<{
table_name: string;
table_schema: string;
}>
}
@@ -7,9 +7,10 @@ import { PlatformApiService } from './platform-api.service';
import { ElasticsearchModule } from '../../services/elasticsearch';
import { DynamoDBModule } from '../../services/dynamodb';
import { CustomersModule } from '../customers/customers.module';
import { CatalogModule } from '../catalog/catalog.module';
@Module({
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule],
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule],
controllers: [PlatformApiController],
providers: [PlatformApiService, DadosferaLogger],
exports: [PlatformApiService],
@@ -89,6 +89,12 @@ export class PlatformApiService {
// Propagate non-2xx responses as HttpExceptions
if (response.status >= 400) {
this.logger.error('Platform API upstream error', {
status: response.status,
data: response.data,
path,
method: method.toUpperCase(),
});
throw new HttpException(response.data, response.status);
}
@@ -0,0 +1,13 @@
export const STORAGE_EXPLORER_CONFIG = {
getUrl: (customerName: string): string => {
const urlTemplate = process.env.STORAGE_EXPLORER_API_URL;
if (!urlTemplate) {
throw new Error('STORAGE_EXPLORER_API_URL environment variable is not set');
}
// Replace {customer_id} placeholder with actual customer ID
// For local: http://172.17.0.1:8000/api (no placeholder)
// For prod: https://storage-explorer-{customer_id}.dadosfera.ai/api
return urlTemplate.replace('{customer}', customerName);
},
timeout: parseInt(process.env.STORAGE_EXPLORER_TIMEOUT || '30000', 10),
};
@@ -0,0 +1,383 @@
import {
Controller,
Get,
Post,
Put,
Param,
Body,
Query,
Inject,
UseInterceptors,
UploadedFiles,
Headers,
} from '@nestjs/common';
import { ApiTags, ApiOperation, ApiConsumes } from '@nestjs/swagger';
import { FilesInterceptor } from '@nestjs/platform-express';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import FormData from 'form-data';
import {
Authenticated,
RequireAllPermissions,
} from '../../decorators/authentication.decorator';
import { User, RequestUser } from '../../decorators/user.decorator';
import { StorageExplorerService } from './storage-explorer.service';
import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
@ApiTags('Storage Explorer')
@Controller('storage-explorer')
@Authenticated()
export class StorageExplorerController {
private logger: any;
constructor(
private readonly storageExplorerService: StorageExplorerService,
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
) {
this.logger = dadosferaLogger.logger;
}
// ============================================
// TABLE OPERATIONS
// ============================================
@ApiOperation({ summary: 'Validate table name in PostgreSQL and Snowflake' })
@Post('tables/validate-name')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.WRITE)
async validateTableName(
@Body() body: any,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'POST',
'/tables/validate-name',
user,
body,
);
}
@ApiOperation({ summary: 'Create a new table' })
@Post('tables')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.WRITE)
async createTable(
@Body() body: any,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'POST',
'/tables/',
user,
body,
);
}
@ApiOperation({ summary: 'List all tables with pagination' })
@Get('tables')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async listTables(
@Query('page') page: number,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
'/tables/',
user,
undefined,
{ page },
);
}
@ApiOperation({ summary: 'Get table details by ID' })
@Get('tables/:tableId')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async getTable(
@Param('tableId') tableId: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
`/tables/${tableId}`,
user,
);
}
@ApiOperation({ summary: 'Link a dataset to a table' })
@Post('tables/:tableId/datasets/:datasetId')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.WRITE)
async linkDatasetToTable(
@Param('tableId') tableId: string,
@Param('datasetId') datasetId: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'POST',
`/tables/${tableId}/datasets/${datasetId}`,
user,
);
}
@ApiOperation({ summary: 'Get all datasets linked to a table' })
@Get('tables/:tableId/datasets')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async getTableDatasets(
@Param('tableId') tableId: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
`/tables/${tableId}/datasets`,
user,
);
}
@ApiOperation({ summary: 'Get table schema' })
@Get('tables/:tableId/schema')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async getTableSchema(
@Param('tableId') tableId: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
`/tables/${tableId}/schema`,
user,
);
}
@ApiOperation({ summary: 'Validate schema compatibility between table and dataset' })
@Post('tables/:tableId/validate-compatibility/:datasetId')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async validateSchemaCompatibility(
@Param('tableId') tableId: string,
@Param('datasetId') datasetId: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'POST',
`/tables/${tableId}/validate-compatibility/${datasetId}`,
user,
);
}
// ============================================
// DATASET OPERATIONS
// ============================================
@ApiOperation({ summary: 'Get dataset preview data' })
@Get('datasets/:datasetId/preview')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async getDatasetPreview(
@Param('datasetId') datasetId: string,
@Query('limit') limit: number,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
`/datasets/${datasetId}/preview`,
user,
undefined,
{ limit },
);
}
@ApiOperation({ summary: 'Get dataset schema information' })
@Get('datasets/:datasetId/schema')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async getDatasetSchema(
@Param('datasetId') datasetId: string,
@Query('force_refresh') forceRefresh: boolean,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
`/datasets/${datasetId}/schema`,
user,
undefined,
{ force_refresh: forceRefresh },
);
}
@ApiOperation({ summary: 'List all datasets for a specific upload' })
@Get('datasets/upload/:uploadId')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async listDatasetsByUpload(
@Param('uploadId') uploadId: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
`/datasets/upload/${uploadId}`,
user,
);
}
@ApiOperation({ summary: 'Refresh dataset schema with new parsing options (Excel)' })
@Put('datasets/:datasetId/refresh-schema')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.WRITE)
async refreshDatasetSchema(
@Param('datasetId') datasetId: string,
@Body() body: any,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'PUT',
`/datasets/${datasetId}/refresh-schema`,
user,
body,
);
}
// ============================================
// STORAGE OPERATIONS
// ============================================
@ApiOperation({ summary: 'List file explorer uploads with pagination' })
@Get('storage/uploads/history')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async listFileExplorerUploads(
@Query('page') page: number,
@Query('limit') limit: number,
@Query('folder_path') folderPath: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
'/storage/uploads/history',
user,
undefined,
{ page, limit, folder_path: folderPath },
);
}
@ApiOperation({ summary: 'Browse folders and files in storage' })
@Get('storage/browse')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async browseStorage(
@Query('path') path: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
'/storage/browse',
user,
undefined,
{ path },
);
}
@ApiOperation({ summary: 'Upload multiple files to storage' })
@Post('storage/upload/batch')
@ApiConsumes('multipart/form-data')
@UseInterceptors(FilesInterceptor('files'))
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.WRITE)
async batchUpload(
@UploadedFiles() files: Array<Express.Multer.File>,
@Body('folder_path') folderPath: string,
@User() user: RequestUser,
) {
// Create FormData to forward files to storage-explorer API
const formData = new FormData();
// Add files
if (files && files.length > 0) {
files.forEach((file) => {
formData.append('files', file.buffer, {
filename: file.originalname,
contentType: file.mimetype,
});
});
}
// Add folder_path
if (folderPath) {
formData.append('folder_path', folderPath);
}
return this.storageExplorerService.proxyFormData(
'POST',
'/storage/upload/batch',
user,
formData,
);
}
@ApiOperation({ summary: 'Create a new folder in storage' })
@Post('storage/folder/create')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.WRITE)
async createFolder(
@Body() body: any,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'POST',
'/storage/folder/create',
user,
body,
);
}
@ApiOperation({ summary: 'Download a file from storage' })
@Get('storage/download')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async downloadFile(
@Query('file_path') filePath: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
'/storage/download',
user,
undefined,
{ file_path: filePath },
);
}
@ApiOperation({ summary: 'Get detailed file metadata' })
@Get('storage/metadata')
@RequireAllPermissions(PERMISSIONS_GROUPS.STORAGE_EXPLORER.permissions.READ)
async getFileMetadata(
@Query('file_path') filePath: string,
@User() user: RequestUser,
) {
return this.storageExplorerService.proxy(
'GET',
'/storage/metadata',
user,
undefined,
{ file_path: filePath },
);
}
}
@@ -0,0 +1,13 @@
import { Module } from '@nestjs/common';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import { StorageExplorerController } from './storage-explorer.controller';
import { StorageExplorerService } from './storage-explorer.service';
@Module({
imports: [],
controllers: [StorageExplorerController],
providers: [StorageExplorerService, DadosferaLogger],
exports: [StorageExplorerService],
})
export class StorageExplorerModule {}
@@ -0,0 +1,178 @@
import { Injectable, Inject, HttpException } from '@nestjs/common';
import axios, { AxiosResponse, Method } from 'axios';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import { RequestUser } from '../../decorators/user.decorator';
import { STORAGE_EXPLORER_CONFIG } from './storage-explorer.config';
@Injectable()
export class StorageExplorerService {
private logger: any;
constructor(
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
) {
this.logger = dadosferaLogger.logger;
}
async proxy(
method: string,
path: string,
user: RequestUser,
body?: any,
query?: Record<string, any>
): Promise<any> {
// Validate customer_id is present for multi-tenant isolation
if (!user.customer_id) {
throw new HttpException('Customer ID is required for storage operations', 400);
}
// Get customer-specific storage-explorer URL
const baseUrl = STORAGE_EXPLORER_CONFIG.getUrl(user.customer_name);
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> = {
'content-type': 'application/json',
};
this.logger.info('Proxying request to storage-explorer', {
method: method.toUpperCase(),
path,
customer_id: user.customer_id,
storage_url: baseUrl,
user_id: user.user_id,
});
try {
const response: AxiosResponse = await axios({
method: method as Method,
url: url.href,
headers,
data: body,
timeout: STORAGE_EXPLORER_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('Storage Explorer API proxy error', {
error: error.message,
status: error.response?.status,
path,
storage_url: baseUrl,
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('Storage Explorer API service unavailable', 503);
}
if (error.code === 'ETIMEDOUT' || error.code === 'ECONNABORTED') {
throw new HttpException('Storage Explorer API request timeout', 504);
}
throw new HttpException('Internal server error', 500);
}
}
/**
* Proxy with file upload support (multipart/form-data)
*/
async proxyFormData(
method: string,
path: string,
user: RequestUser,
formData: any,
query?: Record<string, any>,
): Promise<any> {
// Validate customer_id is present for multi-tenant isolation
if (!user.customer_id) {
throw new HttpException('Customer ID is required for storage operations', 400);
}
// Get customer-specific storage-explorer URL
const baseUrl = STORAGE_EXPLORER_CONFIG.getUrl(user.customer_name);
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> = {
// Let axios set Content-Type for multipart/form-data with boundary
...formData.getHeaders?.(),
};
this.logger.info('Proxying form data request to storage-explorer', {
method: method.toUpperCase(),
path,
customer_id: user.customer_id,
storage_url: baseUrl,
user_id: user.user_id,
});
try {
const response: AxiosResponse = await axios({
method: method as Method,
url: url.href,
headers,
data: formData,
timeout: STORAGE_EXPLORER_CONFIG.timeout,
maxContentLength: Infinity,
maxBodyLength: Infinity,
validateStatus: () => true,
});
if (response.status >= 400) {
throw new HttpException(response.data, response.status);
}
return response.data;
} catch (error) {
this.logger.error('Storage Explorer API form data proxy error', {
error: error.message,
status: error.response?.status,
path,
storage_url: baseUrl,
method: method.toUpperCase(),
});
if (error instanceof HttpException) {
throw error;
}
if (error.response) {
throw new HttpException(error.response.data, error.response.status);
}
throw new HttpException('Internal server error', 500);
}
}
}
+4 -2
View File
@@ -43,7 +43,7 @@ export class User {
@ApiProperty()
companyName?: string;
@ApiProperty()
companySite?: string;
personalSite?: string;
@ApiPropertyOptional()
customer?: Customer;
@ApiProperty()
@@ -64,6 +64,8 @@ export class UserNoRolesAndCustomer extends OmitType(UserNoRoles, [
export class IUserByCustomer extends OmitType(User, ['customer']) {
@ApiPropertyOptional()
permissions?: string[];
@ApiPropertyOptional()
authProvider?: string;
}
export class CreateUserReq {
@@ -117,7 +119,7 @@ export class UpdateUserReq {
@ApiPropertyOptional()
bio?: string;
@ApiPropertyOptional()
companySite?: string;
personalSite?: string;
@ApiPropertyOptional()
companyName?: string;
@ApiPropertyOptional()
+2 -1
View File
@@ -125,6 +125,7 @@ export class UsersService implements OnModuleInit {
return { permissions };
});
res.user.permissions = permissions;
res.user.authProvider = process.env.AUTH_PROVIDER || 'cognito';
return res;
}
@@ -162,7 +163,7 @@ export class UsersService implements OnModuleInit {
name: updateUserDTO.name,
bio: updateUserDTO.bio,
companyName: updateUserDTO.companyName,
companySite: updateUserDTO.companySite,
personalSite: updateUserDTO.personalSite,
customerId,
id,
metabaseUserId: undefined,
@@ -358,6 +358,81 @@ export class ElasticsearchService {
}
}
private getDataAssetIndex(customerName: string): string {
return `${customerName}_data_assets_catalog`;
}
async findDataAssetByTable(
customerName: string,
tableName: string,
tableSchema: string,
): Promise<{ id: string; nimbus_id: number | null; [key: string]: any } | null> {
const index = this.getDataAssetIndex(customerName);
this.logger.info('Elasticsearch: Searching data asset', {
index,
tableName,
tableSchema,
});
try {
const response = await this.client.post(`/${index}/_search`, {
query: {
bool: {
must: [
{ term: { 'table_name.keyword': tableName.toUpperCase() } },
{ term: { 'table_schema.keyword': tableSchema.toUpperCase() } },
],
},
},
size: 1,
});
const hits = response.data.hits?.hits || [];
if (hits.length === 0) {
this.logger.warn('Elasticsearch: Data asset not found', { tableName, tableSchema, index });
return null;
}
return { ...hits[0]._source, _es_id: hits[0]._id };
} catch (error) {
this.handleError('findDataAssetByTable', error, { tableName, tableSchema, index });
throw error;
}
}
async updateDataAsset(
customerName: string,
assetId: string,
updates: Record<string, any>,
): Promise<any> {
const index = this.getDataAssetIndex(customerName);
this.logger.info('Elasticsearch: Updating data asset', {
index,
assetId,
fields: Object.keys(updates),
});
try {
const response = await this.client.post(
`/${index}/_update/${assetId}`,
{ doc: updates },
{ params: { refresh: 'wait_for' } },
);
this.logger.info('Elasticsearch: Data asset updated', {
assetId,
result: response.data.result,
});
return response.data;
} catch (error) {
this.handleError('updateDataAsset', error, { assetId, index });
throw error;
}
}
private handleError(
operation: string,
error: any,