Compare commits

..
Author SHA1 Message Date
Marcos Rodrigues 76485f929d FIX: user patch route 2025-12-26 17:53:02 -03:00
Marcos Rodrigues Silva 1f9d0c29ec Merge pull request #425 from dadosfera/beta
Beta
2025-12-18 17:42:10 -03:00
Marcos Rodrigues Silva 31f8c2c1a6 Merge pull request #424 from dadosfera/feature/user-form
FEAT: protected update route with admin seqid or if user request to e…
2025-12-18 12:33:29 -03:00
Marcos Rodrigues 8b93d4e97b FEAT: protected update route with admin seqid or if user request to equal a jwt user id claim 2025-12-18 12:32:53 -03:00
Rafael Santana adeb022818 Merge pull request #423 from dadosfera/fix/reference-column-object-format
FIX: inject customer_id in pipeline execute/pause/unpause routes
2025-12-18 10:53:31 -03:00
RafaelandClaude Opus 4.5 f61c241dde FIX: inject customer_id in pipeline execute/pause/unpause routes
Users no longer need to provide customer_id in the request body for
execute, pause, and unpause pipeline operations - it's now automatically
injected from the authenticated user's session.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-18 10:30:09 -03:00
Marcos Rodrigues Silva e326cab44d Merge pull request #422 from dadosfera/feature/user-form
FIX: send new user info
2025-12-18 10:29:09 -03:00
Marcos Rodrigues 61109f8ae9 FIX: send new user info 2025-12-18 10:24:29 -03:00
Marcos Rodrigues Silva 3f8dc5cabe Merge pull request #421 from dadosfera/feature/user-form
Feature/user form
2025-12-18 09:53:12 -03:00
Marcos Rodrigues 7f5d157739 FEAT: update protospack version 2025-12-18 09:52:11 -03:00
Marcos Rodrigues bce73fb11f FEAT: update user route 2025-12-17 17:24:03 -03:00
Rafael Santana 9473e65deb Merge pull request #419 from dadosfera/fix/reference-column-object-format
FIX: store reference_column as object with name and type
2025-12-17 13:48:38 -03:00
RafaelandClaude Opus 4.5 9c55c22230 FIX: store reference_column as object with name and type
The protobuf definition expects reference_column to be an object with
name and type fields, but it was being stored as just a string (column
name). This caused pipeline fetching to fail with the error:
".NewTable.reference_column: object expected"

Changes:
- Update ReferenceColumn interface in DynamoDB service
- Update extractTablesFromJobs to create reference_column object
- Update syncJobInputToDynamoDB to handle reference_column object

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-17 13:47:57 -03:00
Rafael Santana 77b9acd2d0 Merge pull request #418 from dadosfera/fix/reference-column-object-format
fix: store reference_column as object with name and type
2025-12-17 13:46:05 -03:00
RafaelandClaude Opus 4.5 e99306adba fix: store reference_column as object with name and type
The protobuf definition expects reference_column to be an object with
name and type fields, but it was being stored as just a string (column
name). This caused pipeline fetching to fail with the error:
".NewTable.reference_column: object expected"

Changes:
- Update ReferenceColumn interface in DynamoDB service
- Update extractTablesFromJobs to create reference_column object
- Update syncJobInputToDynamoDB to handle reference_column object

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-17 13:45:18 -03:00
Rafael Santana 96b947ebdc Merge pull request #417 from dadosfera/fix/redis-tls-configurable
FIX: pass type field to Elasticsearch createPipeline
2025-12-16 21:10:19 -03:00
RafaelandClaude Opus 4.5 fb521f53cd FIX: pass type field to Elasticsearch createPipeline
The type field was missing from the createPipeline call,
causing ES documents to not have the type field set.

Maps: jdbc->database, singer->application, s3->file

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 21:09:07 -03:00
Rafael Santana 6f7436f33f Merge pull request #416 from dadosfera/fix/redis-tls-configurable
fix: pass type field to Elasticsearch createPipeline
2025-12-16 21:05:55 -03:00
RafaelandClaude Opus 4.5 865140e681 fix: pass type field to Elasticsearch createPipeline
The type field was missing from the createPipeline call,
causing ES documents to not have the type field set.

Maps: jdbc->database, singer->application, s3->file

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 21:05:21 -03:00
Rafael Santana c4a664572a Merge pull request #415 from dadosfera/fix/redis-tls-configurable
UPDATE: force deployment
2025-12-16 19:17:08 -03:00
Rafael acb631e33d UPDATE: force deployment 2025-12-16 19:16:35 -03:00
Rafael Santana 7a10f88113 Merge pull request #414 from dadosfera/fix/redis-tls-configurable
fix: normalize IDs in pipeline run routes before calling Platform API
2025-12-16 19:11:09 -03:00
RafaelandClaude Opus 4.5 e616061c21 fix: normalize IDs in pipeline run routes before calling Platform API
- Add normalization for pipelineId and runId in getPipelineRuns, getPipelineRun, and getPipelineRunLogs
- Remove unused createPipelineRun route

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 19:10:23 -03:00
Rafael Santana 466f8fb8cc Merge pull request #413 from dadosfera/fix/redis-tls-configurable
UPDATE: force deployment
2025-12-16 18:23:44 -03:00
Rafael e03b9e7a14 UPDATE: force deployment 2025-12-16 18:23:19 -03:00
Rafael Santana 5989822263 Merge pull request #412 from dadosfera/fix/redis-tls-configurable
feat: add Elasticsearch access to Maestro deployment
2025-12-16 18:15:08 -03:00
RafaelandClaude Opus 4.5 2cc8f46418 feat: add Elasticsearch access to Maestro deployment
- Add ELASTICSEARCH_URL and ELASTICSEARCH_API_KEY env vars
- Pull credentials from {env}/microservices/elasticsearch secret

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 18:10:38 -03:00
vinicius gadea 64a3e2652e Merge pull request #410 from dadosfera/feature/catalog-update
Feature/catalog update
2025-12-16 18:01:47 -03:00
Rafael Moraes ea44a1cbb6 FEAT: Manual/Auto filter and UX Adjusments 2025-12-16 20:52:28 +00:00
Rafael Moraes fc9c0b0991 WIP: Filtro Source - backend 2025-12-16 20:13:20 +00:00
Rafael Moraes a4b5a44e44 Merge branch 'feature/catalog-update' of github.com:dadosfera/maestro into feature/catalog-update 2025-12-16 20:01:19 +00:00
Rafael Moraes d99a6aa322 Merge branch 'beta' of github.com:dadosfera/maestro into feature/catalog-update 2025-12-16 19:58:48 +00:00
Rafael Santana 3f910f851a Merge pull request #411 from dadosfera/fix/redis-tls-configurable
Fix/redis tls configurable
2025-12-16 16:58:41 -03:00
Rafael Moraes bd231382eb WIP: adiciona parâmetro asset_type no docsfera.json 2025-12-16 19:58:38 +00:00
Rafael 295f1f86ca Merge branch 'fix/redis-tls-configurable' of github.com:dadosfera/maestro into fix/redis-tls-configurable 2025-12-16 16:57:59 -03:00
Rafael 0a5e8001f9 UPDATE: adding platform_api_url 2025-12-16 16:57:25 -03:00
Rafael Santana 9c1979e17a Merge pull request #405 from dadosfera/fix/redis-tls-configurable
Fix/redis tls configurable
2025-12-16 16:41:04 -03:00
Rafael Santana b2700d4bb0 Merge branch 'beta' into fix/redis-tls-configurable 2025-12-16 16:39:11 -03:00
RafaelandClaude Opus 4.5 01c1087e07 FEAT: add schedule limit validation and improve ES update
- Add schedule limit validation against customer's scheduleLimit from DUC
- Improve ES update to only update provided fields
- Mark GET /platform/pipeline/:pipelineId/pipeline_run as READY

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 16:37:33 -03:00
viniciusgadea fcf7fb054e FIX: Update searchDataAssets to handle owner filtering and convert owner names to IDs 2025-12-16 09:55:24 -03:00
viniciusgadea 288aaeabc0 FIX: Enhance filter handling in BuildDataAssetsFilterTerms to support empty owner condition 2025-12-15 12:00:50 -03:00
viniciusgadea 918c3d7416 FIX: Enhance searchDataAssets to convert owner names to IDs and log the conversion 2025-12-15 09:08:08 -03:00
vinicius gadea 7dedb3bd33 Merge pull request #409 from dadosfera/feature/catalog-update
FIX: Simplify searchDataAssets by owner logging and remove log code
2025-12-12 15:51:17 -03:00
viniciusgadea 1ff5589a2e FIX: Simplify searchDataAssets by owner logging and remove log code 2025-12-12 15:03:43 -03:00
Rafael Moraes d24e9a1d80 Merge pull request #408 from dadosfera/feature/catalog-update
FIX: Calendar Fix
2025-12-11 22:26:25 +00:00
Rafael Moraes 7d3ef1ef92 FIX: Conflit resolution 2025-12-11 22:25:42 +00:00
Rafael Moraes df3f2489f9 FIX: Calendar Fix 2025-12-11 22:10:32 +00:00
Rafael Moraes 53245b0067 Merge pull request #407 from dadosfera/feature/catalog-update
FIX: correção no validate k8s
2025-12-11 16:50:11 +00:00
Rafael Moraes dd699614ae FIX: Maestro helm fix 2025-12-11 16:47:28 +00:00
Rafael Moraes 6919a2a8d0 FIX: helmfile diff plugin 2025-12-11 14:14:02 +00:00
Rafael Moraes 5c29e07450 FIX: helmfile 2025-12-11 14:07:04 +00:00
Rafael Moraes 00304262ba FIX: wget change in helmfile 2025-12-11 13:52:58 +00:00
vinicius gadea bc931c6dd8 Merge pull request #406 from dadosfera/feature/catalog-update
catalogo: alteracao nas pastas e acrescimo de filtros
2025-12-11 09:18:57 -03:00
Rafael Moraes c444d6e956 Merge pull request #404 from dadosfera/feat/add-docs-assets
Feat/add docs assets
2025-12-10 19:42:42 +00:00
RafaelandClaude Opus 4.5 f7efb757bf FEAT: add schedule limit validation and improve ES update
- Add cron-validation utility with Airflow presets support (@once, @daily, etc.)
- Validate cron expressions against customer scheduleLimit from DUC
- Fetch scheduleLimit via gRPC to ensure up-to-date configuration
- Improve ES updatePipeline to fetch current doc and merge changes
- Only update fields explicitly provided in PATCH request body
- Preserve config.tables when updating config.cron

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-10 13:56:49 -03:00
Rafael Moraes 2f140d213a catalogo: alteracao nas pastas e acrescimo de filtros 2025-12-10 16:51:28 +00:00
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
viniciusgadea 09dced9fcd REFACTOR: add asset_type query parameter to getDataDocs and separate data docs methods 2025-12-10 11:21:46 -03:00
viniciusgadea 165b533172 FIX: removing unnecessary type parameter from GetDatasetDoc call 2025-12-10 11:17:43 -03:00
viniciusgadea b1dc567394 DOCS: update @dadosfera/protospack-v2 to version 3.38.0-beta.16 2025-12-10 11:12:13 -03:00
viniciusgadea a0a1303515 DOCS: add automatic asset_type query parameter to API schema and packages dependencies 2025-12-09 14:56:51 -03:00
viniciusgadea c21d977f7c FEAT: enhance catalog service to support any asset type in data docs creation 2025-12-09 14:56:51 -03:00
viniciusgadea 399d3492d3 DOCS: update @dadosfera/protospack-v2 to version 3.38.0-beta.15 2025-12-09 14:56:51 -03:00
Marcos Rodrigues Silva b0b557246e Merge pull request #403 from dadosfera/feat/user-properties
FIX:  change origin to host
2025-12-03 15:26:10 -03:00
Marcos Rodrigues bb671a90d6 FIX: change origin to host 2025-12-03 15:20:41 -03:00
Marcos Rodrigues Silva f6ababbe7a Merge pull request #402 from dadosfera/feat/user-properties
FIX: using header host
2025-12-03 14:46:14 -03:00
Marcos Rodrigues 82290285d0 FIX: using header host 2025-12-03 14:41:51 -03:00
Marcos Rodrigues Silva c490814a98 Merge pull request #401 from dadosfera/feat/user-properties
FEAT: add resource origin validate in me endpoint
2025-12-03 13:53:33 -03:00
Marcos Rodrigues 7616b1e32c FEAT: add resource origin validate in me endpoint 2025-12-03 13:51:32 -03:00
Rafael Santana 3ba2c91893 Merge pull request #400 from dadosfera/fix/redis-tls-configurable
UPDATE: force deployment
2025-12-02 15:30:03 -03:00
Rafael e789076ed4 UPDATE: force deployment 2025-12-02 15:29:36 -03:00
Rafael Santana 3b6ddaaee6 Merge pull request #399 from dadosfera/fix/redis-tls-configurable
fix: make Redis TLS configurable via REDIS_TLS env var
2025-12-02 15:18:05 -03:00
RafaelandClaude 8073194604 fix: make Redis TLS configurable via REDIS_TLS env var
- Change TLS check from ENV !== 'local' to REDIS_TLS === 'true'
- Add REDIS_TLS env var to Helm deployment template
- Default to true for OCI Redis, can override for self-hosted K8s

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-12-02 15:10:31 -03:00
Marcos Rodrigues Silva e7f410831f Merge pull request #395 from dadosfera/beta
Beta
2025-12-01 17:47:24 -03:00
Marcos Rodrigues Silva cd2c53b5c5 Merge pull request #398 from dadosfera/feat/user-properties
FIX: user response
2025-12-01 17:36:56 -03:00
Marcos Rodrigues adf1b3b97e FIX: user response 2025-12-01 17:30:37 -03:00
Marcos Rodrigues Silva d84b5e184b Merge pull request #397 from dadosfera/feat/user-properties
FEAT: include more user info in /me
2025-12-01 12:36:16 -03:00
Marcos Rodrigues 6d608a0457 FEAT: include more user info in /me 2025-12-01 12:26:33 -03:00
Marcos Rodrigues Silva b8be2c7803 Merge pull request #396 from dadosfera/fix/api-key-catalog-roles
FIX: api key catalog roles
2025-12-01 09:34:21 -03:00
Marcos Rodrigues 24fce721e3 FIX: api key catalog roles 2025-12-01 09:32:26 -03:00
Marcos Rodrigues Silva 23a9a27db1 Merge pull request #394 from dadosfera/fix/private-front-cors
FIX: 404 in  get public key
2025-11-28 16:44:54 -03:00
Marcos Rodrigues 0f5ed50af9 FIX: 404 in get public key 2025-11-28 16:37:38 -03:00
Marcos Rodrigues Silva 5f8f6a64ab Merge pull request #393 from dadosfera/fix/private-front-cors
FIX: include private frontend url in cors origins list
2025-11-28 10:53:45 -03:00
Marcos Rodrigues 8800ac2736 FIX: include private frontend url in cors origins list 2025-11-28 10:50:31 -03:00
Marcos Rodrigues Silva bdb82c2ce4 Merge pull request #392 from dadosfera/fix/authenticated-track-events
Fix/authenticated track events
2025-11-27 09:45:46 -03:00
Marcos Rodrigues 6d9ecc3568 FIX: mixpanel events 2025-11-27 09:43:43 -03:00
Marcos Rodrigues Silva 7764447adc Merge pull request #391 from dadosfera/fix/private-ingress-annotations
FIX: ci to oracle github worker
2025-11-25 11:37:29 -03:00
Marcos Rodrigues 576fdecf89 FIX: ci to oracle github worker 2025-11-25 11:27:04 -03:00
Marcos Rodrigues Silva 3de1e90fa8 Merge pull request #390 from dadosfera/fix/private-ingress-annotations
Fix/private ingress annotations
2025-11-25 11:12:58 -03:00
Marcos Rodrigues 5e90950660 FIX: restore whitelist annotation 2025-11-25 10:59:06 -03:00
Marcos Rodrigues 3ed26e648f FIX: ci 2025-11-25 10:48:06 -03:00
Marcos Rodrigues Silva fff3523152 Merge pull request #389 from dadosfera/fix/private-ingress-annotations
FIX: private ingress annotations
2025-11-25 10:38:52 -03:00
Marcos Rodrigues 19e4daeea4 FIX: private ingress annotations 2025-11-25 10:35:08 -03:00
Marcos Rodrigues Silva 8624d3f016 Merge pull request #388 from dadosfera/fix/reset-theme
Fix/reset theme
2025-11-21 11:01:48 -03:00
Marcos Rodrigues 4f4da5bebe FIX: reset theme 2025-11-21 10:51:31 -03:00
Marcos Rodrigues f00d2bf41d FIX: reset cstumer theme 2025-11-21 09:55:39 -03:00
Marcos Rodrigues Silva dc1d1400d8 Merge pull request #387 from dadosfera/feat/lineage
FIX: change wget to curl
2025-11-13 18:19:20 -03:00
Marcos Rodrigues d4451153a3 FIX: change wget to curl 2025-11-13 18:14:47 -03:00
Marcos Rodrigues Silva 06ba759ea3 Merge pull request #386 from dadosfera/feat/lineage
Feat/lineage
2025-11-13 17:58:33 -03:00
Marcos Rodrigues dd0d08ad6e FEAT: add permissions to lineage module 2025-11-13 17:48:55 -03:00
Marcos Rodrigues ec81082877 FIX: resolve cors from localhost 2025-11-13 17:48:38 -03:00
Marcos Rodrigues Silva 7f58dc7090 Merge pull request #385 from dadosfera/hotfix/import-files
FIX: permissions
2025-10-22 18:35:35 -03:00
Marcos Rodrigues 43aa379c06 FIX: permissions 2025-10-22 18:29:43 -03:00
Marcos Rodrigues Silva c284f8753c Merge pull request #384 from dadosfera/hotfix/import-files
FIX: pipelines v2 permissions
2025-10-22 17:48:26 -03:00
Marcos Rodrigues 3a8f2495c4 FIX: pipelines v2 permissions 2025-10-20 18:13:54 -03:00
Marcos Rodrigues Silva af248716ef Merge pull request #383 from dadosfera/beta
Beta
2025-10-16 18:15:08 -03:00
Marcos Rodrigues Silva a57ad41ad4 Merge pull request #382 from dadosfera/features/roles
Features/roles
2025-10-16 17:21:37 -03:00
Marcos Rodrigues ffddceec3b FEAT: roles 2025-10-16 17:17:01 -03:00
Marcos Rodrigues Silva b04d5bb402 Merge pull request #381 from dadosfera/feature/tableau-dash
UPDATE: default roles and user
2025-09-23 18:10:35 -03:00
Marcos Rodrigues 53df3caf3f UPDATE: default roles and user 2025-09-23 18:09:25 -03:00
Marcos Rodrigues Silva ab2e35b54f Merge pull request #380 from dadosfera/beta
Beta
2025-09-17 18:11:05 -03:00
Marcos Rodrigues Silva bfa77d8f7b Merge pull request #379 from dadosfera/feature/auth-cookie
Feature/auth cookie
2025-09-16 17:40:10 -03:00
marcos-silva-rodrigues 5002d147ad UPDATE: add all dadosfera domains 2025-09-16 17:33:19 -03:00
Marcos Rodrigues Silva a2fbeb97cc Merge pull request #377 from dadosfera/feature/auth-cookie
UPDATE: test cors domain
2025-09-08 15:25:05 -03:00
marcos-silva-rodrigues 288a796f46 UPDATE: test cors domain 2025-09-08 15:22:26 -03:00
Marcos Rodrigues Silva b5a1e93770 Merge pull request #376 from dadosfera/feature/auth-cookie
UPDATE: cookie path
2025-09-08 10:32:32 -03:00
marcos-silva-rodrigues 1ee49ceab8 UPDATE: cookie path 2025-09-08 10:30:56 -03:00
Marcos Rodrigues Silva 6eaf9cf6d0 Merge pull request #375 from dadosfera/feature/auth-cookie
UDATE: test cookie domain dadosfera.ai
2025-09-08 09:32:21 -03:00
marcos-silva-rodrigues 075ca747df FIX: ghost commit 2025-09-08 09:26:47 -03:00
marcos-silva-rodrigues 9ead4588c1 UDATE: test cookie domain dadosfera.ai 2025-09-08 09:25:14 -03:00
Marcos Rodrigues Silva 92b64bb362 Merge pull request #374 from dadosfera/hotfix/oracle
Hotfix/oracle
2025-09-05 10:25:38 -03:00
Marcos Rodrigues Silva 6780167f2b Merge pull request #371 from dadosfera/hotfix/oracle
FIX: infinite loading when login is incorrect
2025-09-03 16:31:28 -03:00
Marcos Rodrigues Silva caeaf62a9d Merge pull request #370 from dadosfera/hotfix/oracle
Hotfix/oracle
2025-09-03 12:21:17 -03:00
51 changed files with 5656 additions and 1635 deletions
+3 -3
View File
@@ -56,9 +56,9 @@ jobs:
- name: Install Helmfile
run: |
wget https://github.com/helmfile/helmfile/releases/download/v0.148.0/helmfile_0.148.0_linux_amd64.tar.gz
curl -fsSLO https://github.com/helmfile/helmfile/releases/download/v0.148.0/helmfile_0.148.0_linux_amd64.tar.gz
tar -xzf helmfile_0.148.0_linux_amd64.tar.gz
mv helmfile /usr/local/bin/
sudo mv helmfile /usr/local/bin/
helmfile --version
- name: Install Helm Diff Plugin
@@ -105,7 +105,7 @@ jobs:
- name: Install Helmfile
run: |
wget https://github.com/helmfile/helmfile/releases/download/v0.148.0/helmfile_0.148.0_linux_amd64.tar.gz
curl -fsSLO https://github.com/helmfile/helmfile/releases/download/v0.148.0/helmfile_0.148.0_linux_amd64.tar.gz
tar -xzf helmfile_0.148.0_linux_amd64.tar.gz
sudo mv helmfile /usr/local/bin/
helmfile --version
+6 -3
View File
@@ -66,13 +66,16 @@ jobs:
- name: Install Helmfile
run: |
wget https://github.com/helmfile/helmfile/releases/download/v0.148.0/helmfile_0.148.0_linux_amd64.tar.gz
curl -fsSLO https://github.com/helmfile/helmfile/releases/download/v0.148.0/helmfile_0.148.0_linux_amd64.tar.gz
tar -xzf helmfile_0.148.0_linux_amd64.tar.gz
sudo mv helmfile /usr/local/bin/
helmfile --version
- name: Install Helm Diff Plugin
run: helm plugin install https://github.com/databus23/helm-diff || true
- name: Debug Helm env
run: |
helm env
echo "HOME=$HOME"
ls -R $HOME/.local/share/helm || true
- name: Authenticate with OKE cluster
env:
-1
View File
@@ -2,7 +2,6 @@
<image src="./assets/maestro.svg" style="width:10rem">
</p>
# Maestro
Maestro é a API principal da Dadosfera. É responsável pela comunicação do Frontend com nossos microsserviços.
Binary file not shown.
@@ -104,6 +104,10 @@ spec:
value: {{ .Values.maestro.redis_host }}
- name: REDIS_PORT
value: "{{ .Values.maestro.redis_port }}"
- name: REDIS_TLS
value: "{{ .Values.maestro.redis_tls }}"
- name: PLATFORM_API_URL
value: {{ .Values.maestro.platform_api_url }}
- name: JWT_PRIVATE_KEY
valueFrom:
secretKeyRef:
@@ -124,3 +128,14 @@ spec:
secretKeyRef:
name: prd-{{ .Values.app_name }}
key: AWS_DEFAULT_REGION
# Elasticsearch
- name: ELASTICSEARCH_URL
valueFrom:
secretKeyRef:
name: prd-{{ .Values.app_name }}
key: ELASTICSEARCH_URL
- name: ELASTICSEARCH_API_KEY
valueFrom:
secretKeyRef:
name: prd-{{ .Values.app_name }}
key: ELASTICSEARCH_API_KEY
@@ -4,9 +4,18 @@ metadata:
annotations:
nginx.ingress.kubernetes.io/whitelist-source-range: "69.49.241.121/32" # hostgator ip
nginx.ingress.kubernetes.io/proxy-body-size: "0"
nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
nginx.ingress.kubernetes.io/proxy-connect-timeout: "300"
nginx.ingress.kubernetes.io/proxy-send-timeout: "300"
nginx.ingress.kubernetes.io/server-snippet: |
underscores_in_headers on;
ignore_invalid_headers on;
nginx.ingress.kubernetes.io/proxy-buffer-size: "16k"
nginx.ingress.kubernetes.io/proxy-buffers-number: "8"
nginx.ingress.kubernetes.io/proxy-busy-buffers-size: "64k"
{{- if .Values.maestro.restricted_ip}}
nginx.ingress.kubernetes.io/whitelist-source-range: {{ .Values.maestro.restricted_ip }}
{{- end }}
generation: 1
labels:
+12
View File
@@ -38,3 +38,15 @@ spec:
version: "AWSCURRENT"
property: token
- secretKey: ELASTICSEARCH_URL
remoteRef:
key: {{ .Values.maestro.env }}/microservices/elasticsearch
version: "AWSCURRENT"
property: ELASTICSEARCH_URL
- secretKey: ELASTICSEARCH_API_KEY
remoteRef:
key: {{ .Values.maestro.env }}/microservices/elasticsearch
version: "AWSCURRENT"
property: ELASTICSEARCH_API_KEY
+1
View File
@@ -8,6 +8,7 @@ maestro:
open_group_id: e3f98a2f-7748-4981-8505-7695c8ca8218
cookie_secret: "ff7bc13823edb2ae50d248e5780bddc9d4b31c36"
redis_database: "1"
platform_api_url: https://xs2hkhq07k.execute-api.us-east-1.amazonaws.com
hostname: maestro.stg.dadosfera.ai
+2
View File
@@ -43,11 +43,13 @@ maestro:
upload_file_agent_connection: cbc2f881-58c4-4d60-8003-0979b0b5b911
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
dedicated_proxy: ""
restricted_ip: ""
redis_host: "aaapzppmlyamkocqwstpo7zvopczyyiyuy6xzm2g6c5k4mq3a66be4a-0.redis.sa-saopaulo-1.oci.oraclecloud.com"
redis_port: "6379"
redis_database: "0"
redis_tls: "true"
cookie_secret: "13cc5e136d3074bcc05bec8697092ec1f5f376bf"
autoscaling:
enabled: false
+984 -122
View File
File diff suppressed because it is too large Load Diff
+1 -8
View File
@@ -16,14 +16,7 @@ declare global {
OPEN_CUSTOMER_ID: string;
DEDICATED_PROXY: string;
COOKIE_SECRET: string;
// Autodrive Configuration
AUTODRIVE_USERNAME?: string;
AUTODRIVE_PASSWORD?: string;
AUTODRIVE_BASE_URL?: string;
AUTODRIVE_MODEL?: string;
AUTODRIVE_KEY?: string;
AUTO_DRIVE_KEY?: string;
REDIS_TLS?: string;
}
}
}
+1942 -328
View File
File diff suppressed because it is too large Load Diff
+7 -2
View File
@@ -27,10 +27,15 @@
"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/credential-provider-node": "^3.940.0",
"@aws-sdk/lib-dynamodb": "^3.414.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",
"@dadosfera/protospack-v2": "3.38.0-beta.18",
"@grpc/grpc-js": "^1.9.3",
"@grpc/proto-loader": "^0.7.9",
"@nestjs/cli": "^9.5.0",
@@ -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",
+2
View File
@@ -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,
],
+20
View File
@@ -0,0 +1,20 @@
import jwt, { JwtPayload } from 'jsonwebtoken';
export function extractUserFrom(aRawJwt: string) {
const decodedToken = jwt.decode(aRawJwt, {
complete: true,
});
const payload = decodedToken.payload as JwtPayload;
return {
user_id: payload.user_id,
username: payload.username,
permissions: payload.permissions,
customer_id: payload.customer_id,
customer_name: payload.customer_name,
customer_tier: payload.customer_tier,
customer_modules: payload.customer_modules,
access_token: aRawJwt,
}
}
+57
View File
@@ -116,6 +116,44 @@ export const PERMISSIONS_GROUPS = {
},
},
},
IMPORT_FILES: {
title: {
'pt-br': 'Coletar | Importar arquivos',
'en-us': 'Collect | Import files',
'es-es': 'Colecta | Importar archivos',
},
permissions: {
VIEW: {
seqid: 48,
claim: 'import-file:view',
usage: PermissionUsages.PUBLIC,
name: {
'pt-br': 'Importar arquivos',
'en-us': 'Import files',
'es-es': 'Importar archivos',
},
},
},
},
AI_CHAT: {
title: {
'pt-br': 'AutodriveDDF',
'en-us': 'AutodriveDDF',
'es-es': 'AutodriveDDF',
},
permissions: {
VIEW: {
seqid: 49,
claim: 'ai-chat:view',
usage: PermissionUsages.PUBLIC,
name: {
'pt-br': 'AutodriveDDF',
'en-us': 'AutodriveDDF',
'es-es': 'AutodriveDDF',
},
},
},
},
CONNECTION: {
title: {
'pt-br': 'Coletar | Fontes de dados',
@@ -352,6 +390,25 @@ export const PERMISSIONS_GROUPS = {
},
},
},
LINEAGE: {
title: {
'pt-br': 'Explorar | Linhagem',
'en-us': 'Explore | Lineage',
'es-es': 'Explorar | Linaje',
},
permissions: {
VIEW: {
seqid: 50,
claim: 'lineage:view',
usage: PermissionUsages.PUBLIC,
name: {
'pt-br': 'Acessar ao módulo de Linhagem',
'en-us': 'Access to Lineage module',
'es-es': 'Acceda al módulo de Linaje',
},
}
},
},
EMBED: {
title: {
'pt-br': 'Analisar | Incorporação',
+24 -4
View File
@@ -18,21 +18,41 @@ async function bootstrap() {
});
const logger = new DadosferaLogger();
const corsOrigins = [];
if (process.env.ENV === 'local') {
corsOrigins.push('http://localhost:4200');
} else {
corsOrigins.push(
'https://app.stg.dadosfera.ai',
'https://app.dadosfera.ai',
'https://private-frontend.stg.dadosfera.ai',
'https://unimed.dadosfera.ai',
'https://boston-scientific.dadosfera.ai',
'https://plataforma.dadosfera.ai'
);
}
const app = await NestFactory.create(AppModule, {
logger,
cors: {
origin: '*',
origin: corsOrigins,
methods: 'GET,HEAD,PUT,PATCH,POST,DELETE',
preflightContinue: false,
optionsSuccessStatus: 204,
credentials: true
credentials: true,
},
});
app.use(helmet());
app.use(cookieParser(process.env.COOKIE_SECRET));
if (process.env.ENV === 'prd') {
if (process.env.ENV !== 'local') {
app.use('/catalog/register-dataset', json({ limit: '10mb' }));
app.use('/catalog/register-dataset', urlencoded({ extended: true, limit: '10mb' }));
app.use(
'/catalog/register-dataset',
urlencoded({ extended: true, limit: '10mb' }),
);
}
configureSwagger(app);
+6 -2
View File
@@ -1,4 +1,4 @@
import { Controller, Post, Body, Put, Get} from '@nestjs/common';
import { Controller, Body, Put, Get, NotFoundException} from '@nestjs/common';
import { AssignService } from './assign.service';
import { CreateAssignDto } from './dto/create-assign.dto';
import { Authenticated, RequireModule, RequireSomePermission } from 'src/decorators/authentication.decorator';
@@ -28,6 +28,10 @@ export class AssignController {
@RequireModule(DADOSFERA_MODULES_KEYS.EMBED_ASSIGNED)
async get(@User() user: RequestUser) {
const metadata = PackTheMetadata(user);
return await this.assignService.get(metadata);
try {
return await this.assignService.get(metadata);
} catch (error) {
throw new NotFoundException(error.message)
}
}
}
+30 -127
View File
@@ -13,6 +13,7 @@ import {
Req,
Param,
Res,
UnauthorizedException,
} from '@nestjs/common';
import {
ApiHeaders,
@@ -54,7 +55,6 @@ 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 { Cookie } from 'express-session';
type CookiesValues = {
accessToken?: string;
@@ -103,14 +103,13 @@ export class AuthController {
const data = await this.authClient.signIn({ username, password, totp }, metadata);
if (data.tokens) {
this.addTokenInCookie(res, {
this.authClient.writeAuthSession(res, {
accessToken: data.tokens.accessToken,
refreshToken: data.tokens.refreshToken,
userId: data.user.id
});
}
return res.send(data);
} catch (error) {
this.logger.error('/auth - SignIn - ERROR', error);
@@ -127,27 +126,8 @@ export class AuthController {
) {
try {
this.logger.info('/auth - SignOut');
const exp = 1000 * 60 * 3;
res.cookie('ddf-auth', '', {
domain: 'dadosfera.local',
maxAge: Date.now() - exp,
expires: new Date(),
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
res.cookie('ddf-refresh-auth', '', {
domain: 'dadosfera.local',
maxAge: Date.now() - exp,
expires: new Date(),
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
this.logger.info('Clean cookie sessions');
this.authClient.cleanUpAuthSession(res);
return res.send();
} catch (error) {
@@ -176,8 +156,9 @@ export class AuthController {
const data = await this.authClient.refreshAccessToken({ refreshToken, userId }, metadata);
this.addTokenInCookie(res, {
this.authClient.writeAuthSession(res, {
accessToken: data.accessToken,
refreshToken: data.refreshToken,
userId
});
@@ -493,114 +474,36 @@ export class AuthController {
@Get('me')
async getMe(@Req() req: Request, @Res() res: Response) {
this.logger.info('GET /auth/me ')
// Lê cookies
// Get token and headers
const accessToken = req.cookies['ddf-auth'];
const userId = req.cookies['ddf-user-id'];
const refreshToken = req.cookies['ddf-refresh-auth'];
const userId = req.cookies['ddf-user-id'];
const resourceHost = req.headers["host"]
const hasUserSession = Boolean(accessToken) && Boolean(userId);
this.logger.info('Has User Session: ' + hasUserSession);
if (!hasUserSession) {
throw new UnauthorizedException()
}
this.logger.info('Has cookie: ' + Boolean(accessToken))
let payload: any;
let userInfo: any = {};
try {
// Decodifica e valida o JWT de acesso
const decoded: any = accessToken && jwt.decode(accessToken, { complete: true });
if (!decoded) throw new Error('Invalid token')
const { kid } = decoded.header;
// Busca a chave pública
const { keys } = await this.authClient.getPublicKeys();
const pemValue = keys.find((k) => k.kid === kid)?.pem;
if (!pemValue) throw new Error('Public key not found');
jwt.verify(accessToken, pemValue);
payload = decoded.payload;
userInfo = {
id: payload.user_id,
name: payload.username,
customer: {
id: payload.customer_id,
name: payload.customer_name,
tier: payload.customer_tier,
}
};
return res.status(200).json(userInfo);
} catch (err) {
this.logger.error(err.message);
const refreshToken = req.cookies['ddf-refresh-auth'];
const userDto = await this.authClient.validateUserSession(accessToken, resourceHost);
return res.status(200).json(userDto);
} catch (error) {
this.logger.info('Token is invalid')
this.logger.info('Has Refresh Token: '+ Boolean(refreshToken))
// Se access token inválido, tenta refresh
if (!refreshToken || !userId) {
if (!refreshToken) {
this.logger.error('Invalid refresh token or customer name');
return res.status(401).json({ error: 'Not authenticated' });
}
try {
// Chama refreshAccessToken
const metadata = PackTheMetadata({
});
this.logger.info('Call Refresh Token')
const data = await this.authClient.refreshAccessToken({ refreshToken, userId }, metadata);
this.logger.info('Finish Refresh Token')
// Retorna novo access token e dados mínimos
this.addTokenInCookie(res, {
accessToken: data.accessToken,
userId
});
// Decodifica novo token
const decoded: any = jwt.decode(data.accessToken, { complete: true });
const payload = decoded.payload;
userInfo = {
id: payload.user_id,
name: payload.username,
customer: {
id: payload.customer_id,
name: payload.customer_name,
tier: payload.customer_tier,
}
};
return res.status(200).json(userInfo);
} catch (refreshErr) {
this.logger.error(refreshErr)
return res.status(401).json({ error: 'Not authenticated' });
}
}
}
throw new UnauthorizedException("Invalid refresh token or customer name");
};
private addTokenInCookie(res: Response, data: CookiesValues) {
let exp = 1000 * 60 * 5; // 5 minutes
if (data.accessToken) {
const { exp: expiration } = jwt.decode(data.accessToken) as JwtPayload;
exp = (expiration - 30) * 1000; // exp em segundos, maxAge em ms
this.logger.info('Set Cookie ddf-auth')
res.cookie('ddf-auth', data.accessToken, {
domain: 'stg.dadosfera.ai',
maxAge: exp,
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
}
if (data.refreshToken) {
this.logger.info('Set Cookie ddf-refresh-auth')
res.cookie('ddf-refresh-auth', data.refreshToken, {
domain: 'stg.dadosfera.ai',
maxAge: exp,
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
}
if (data.userId) {
this.logger.info('Set Cookie ddf-refresh-auth')
res.cookie('ddf-user-id', data.userId, {
domain: 'stg.dadosfera.ai',
maxAge: exp,
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
const {
authSession,
user
} = await this.authClient.refreshUserSession(refreshToken, userId, resourceHost);
this.authClient.writeAuthSession(res, authSession);
return res.status(200).json(user);
}
}
}
+223 -17
View File
@@ -1,10 +1,20 @@
import { OnModuleInit, Inject, Injectable, ForbiddenException } from '@nestjs/common';
import {
OnModuleInit,
Inject,
Injectable,
ForbiddenException,
HttpException,
HttpStatus,
} from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import { lastValueFrom } from 'rxjs';
import { ProtoServices } from '@dadosfera/protospack-v2/dist/lib/Duc';
import { AuthProtoService as AuthServiceInterface, IdentityProviderProtoService } from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/write-service';
import {
AuthProtoService as AuthServiceInterface,
UsersProtoService,
} from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/write-service';
import {
AuthSnowflakeSignInRequest,
AuthSignInRequest,
@@ -21,18 +31,24 @@ import {
} from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/messages';
import { DucClient } from '../duc/client.config';
import { Metadata } from '@grpc/grpc-js';
import { BulkEditResponse } from './dtos/login';
import { BulkEditResponse, UserDTO } from './dtos/login';
import jwt, { JwtPayload } from 'jsonwebtoken';
import { PackTheMetadata } from 'src/utils/PackTheMetadata';
import { Request, Response } from 'express';
type AuthSession = {
accessToken?: string;
refreshToken?: string;
userId?: string;
};
@Injectable()
export class AuthClientService implements OnModuleInit {
logger: DadosferaLogger;
private authService: AuthServiceInterface;
private identityProviderService: IdentityProviderProtoService;
private userService: UsersProtoService;
constructor(
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
@@ -46,8 +62,8 @@ export class AuthClientService implements OnModuleInit {
ProtoServices.AuthProtoService,
);
this.identityProviderService = this.grpcClient.getService<IdentityProviderProtoService>(
ProtoServices.IdentityProviderProtoService,
this.userService = this.grpcClient.getService<UsersProtoService>(
ProtoServices.UsersProtoService,
);
}
@@ -63,11 +79,11 @@ export class AuthClientService implements OnModuleInit {
return lastValueFrom(this.authService.AuthSnowflakeSignIn(input));
}
checkDedicatedProxy({
customer
}: AuthSignInResponse) {
checkDedicatedProxy({ customer }: AuthSignInResponse) {
const DEDICATED_PROXY = process.env.DEDICATED_PROXY || '';
this.logger.info('SignIn - Setting customer ID for dedicated proxy: ' + DEDICATED_PROXY);
this.logger.info(
'SignIn - Setting customer ID for dedicated proxy: ' + DEDICATED_PROXY,
);
this.logger.info('Customer ID: ' + customer.id);
if (DEDICATED_PROXY !== '' && DEDICATED_PROXY !== customer.id) {
@@ -75,7 +91,9 @@ export class AuthClientService implements OnModuleInit {
}
// Bloquear o customer de acesso o maestro publico
this.logger.info('Check if customer have network policy: ' + customer.modules);
this.logger.info(
'Check if customer have network policy: ' + customer.modules,
);
const hasNetworkPolicyModule = customer.modules.includes('network-policy');
if (hasNetworkPolicyModule && DEDICATED_PROXY === '') {
throw new ForbiddenException();
@@ -94,7 +112,6 @@ export class AuthClientService implements OnModuleInit {
result = await lastValueFrom(
this.authService.AuthSignIn({ username, password, totp }, metadata),
);
} catch (error) {
this.logger.error('SignIn - Error during sign-in');
this.logger.error(error);
@@ -105,7 +122,7 @@ export class AuthClientService implements OnModuleInit {
this.checkDedicatedProxy(result);
}
return result
return result;
}
async refreshAccessToken(
@@ -115,7 +132,10 @@ export class AuthClientService implements OnModuleInit {
this.logger.info('RefreshAccessToken');
return lastValueFrom(
this.authService.AuthRefreshAccessToken({ refreshToken, userId }, metadata),
this.authService.AuthRefreshAccessToken(
{ refreshToken, userId },
metadata,
),
);
}
@@ -283,4 +303,190 @@ export class AuthClientService implements OnModuleInit {
throw error;
}
}
public async validateUserSession(accessToken: any, resourceHost: string) {
const payload = await this.validateJwtToken(accessToken);
const userDto = await this.getUserfromPayload(payload);
this.validateResourceAccess(resourceHost, userDto);
return userDto;
}
public async refreshUserSession(
refreshToken: string,
userId: string,
originHeader: string,
): Promise<{
user: UserDTO;
authSession: AuthSession;
}> {
const metadata = PackTheMetadata({});
this.logger.info('Call Refresh Token');
const refreshCredentials = await this.refreshAccessToken(
{ refreshToken, userId },
metadata,
);
this.logger.info('Finish Refresh Token');
const userDto = await this.validateUserSession(
refreshCredentials.accessToken,
originHeader,
);
return {
user: userDto,
authSession: {
accessToken: refreshCredentials.accessToken,
refreshToken: refreshCredentials.refreshToken,
userId,
},
};
}
public writeAuthSession(res: Response, data: AuthSession) {
let exp = 1000 * 60 * 5; // 5 minutes
if (data.accessToken) {
const { exp: expiration } = jwt.decode(data.accessToken) as JwtPayload;
exp = (expiration - 30) * 1000; // exp em segundos, maxAge em ms
this.logger.info('Set Cookie ddf-auth');
res.cookie('ddf-auth', data.accessToken, {
domain: '.dadosfera.ai',
maxAge: exp,
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
}
if (data.refreshToken) {
this.logger.info('Set Cookie ddf-refresh-auth');
res.cookie('ddf-refresh-auth', data.refreshToken, {
domain: '.dadosfera.ai',
maxAge: exp,
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
}
if (data.userId) {
this.logger.info('Set Cookie ddf-refresh-auth');
res.cookie('ddf-user-id', data.userId, {
domain: '.dadosfera.ai',
maxAge: exp,
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
}
}
public cleanUpAuthSession(res: Response) {
const exp = 1000 * 60 * 3;
res.cookie('ddf-auth', '', {
domain: 'dadosfera.ai',
maxAge: Date.now() - exp,
expires: new Date(),
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
res.cookie('ddf-refresh-auth', '', {
domain: 'dadosfera.ai',
maxAge: Date.now() - exp,
expires: new Date(),
httpOnly: true,
secure: true,
sameSite: 'none', // Necessário para cookies em requisições cross-site
});
this.logger.info('Clean cookie sessions');
}
private async validateJwtToken(token: string) {
const decoded: any = token && jwt.decode(token, { complete: true });
if (!decoded) throw new Error('Invalid token');
const { kid } = decoded.header;
// Busca a chave pública
const { keys } = await this.getPublicKeys();
const pemValue = keys.find((k) => k.kid === kid)?.pem;
if (!pemValue) throw new Error('Public key not found');
jwt.verify(token, pemValue);
return decoded.payload;
}
private async getUserfromPayload(payload: JwtPayload): Promise<UserDTO> {
this.logger.info('getUser');
const metadata = PackTheMetadata({
customer_id: payload.customer_id,
});
const { user } = await lastValueFrom(
this.userService.UserFindOneById({ id: payload.user_id }, metadata),
);
const userDto: UserDTO = {
id: user.id,
name: user.username,
jobTitle: user?.jobTitle || null,
department: user?.department || null,
hierarchy: user?.hierarchy || null,
customer: {
id: payload.customer_id,
name: payload.customer_name,
tier: payload.customer_tier,
},
};
return userDto;
}
private validateResourceAccess(host: string, user: UserDTO) {
this.logger.info(
"Validate whether the source URL is a resource belonging to the user's client",
);
this.logger.info('Host: ' + host);
this.logger.info('Customer: ' + user.customer.name);
const hostParts = host.split('.');
const domain = hostParts[0];
const isResouceStg = hostParts[1] === 'stg';
const notFoundCustomerInDomain = !domain.includes('-')
if (notFoundCustomerInDomain) {
this.logger.info(`Not found Customer Name in domain`);
return;
}
const domainParts = domain.split('-');
const customerInDomain = domainParts[domainParts.length - 1];
if (isResouceStg && process.env.ENV !== 'stg') {
this.logger.error(`Customer ${user.customer.name} cannot access ${host}`);
throw new HttpException(
`Customer ${user.customer.name} cannot access ${host}`,
HttpStatus.FORBIDDEN
);
}
if (customerInDomain != user.customer.name) {
this.logger.error(`Customer ${user.customer.name} cannot access ${host}`);
throw new HttpException(
`Customer ${user.customer.name} cannot access ${host}`,
HttpStatus.FORBIDDEN
);
}
return;
}
}
+13
View File
@@ -140,3 +140,16 @@ export interface BulkEditResponse {
successfulUsers: string[];
failedUsers: string[];
}
export type UserDTO = {
id: string,
name: string,
jobTitle?: string,
department?: string,
hierarchy?: string,
customer: {
id: string,
name: string,
tier: string,
}
}
+65 -112
View File
@@ -22,9 +22,6 @@ import {
ApiHeaders,
ApiOkResponse,
ApiTags,
ApiOperation,
ApiParam,
ApiResponse,
} from '@nestjs/swagger';
import {
Authenticated,
@@ -91,6 +88,10 @@ export class CatalogController {
@Query() query: ICatalogAllRequest,
): Promise<ICatalogAllResponse> {
const { user_id, customer_name, customer_id, username, permissions } = user;
this.logger.info(`/catalog - searchCatalog`, {
user_id,
customer_name,
});
const is_data_manager = permissions.includes(
PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER.seqid,
@@ -127,6 +128,10 @@ export class CatalogController {
@Res() res: Response
) {
const { user_id, customer_name, customer_id, username, permissions } = user;
this.logger.info(`/catalog/download - searchCatalog`, {
user_id,
customer_name,
});
const is_data_manager = permissions.includes(
PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER.seqid,
@@ -163,6 +168,10 @@ export class CatalogController {
async findByPipelineAndObject(@User() user: RequestUser, @Query() query) {
const { username, user_id, customer_id, customer_name, permissions } = user;
const { pipeline, object } = query;
this.logger.info(`/catalog - ON GET DATA ASSET BY PIPELINE AND OBJECT`, {
username,
customer_name,
});
if (!pipeline || !object) {
throw new BadRequestException('Query params not provided');
@@ -215,6 +224,10 @@ export class CatalogController {
PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
)
async findAllTags(@Body() body) {
this.logger.info(`/catalog - ON FIND ALL TAGS ROUTE`, {
user: body.info.user_id,
customer: body.info.customer,
});
const { user_id, customer, customer_id } = body.info;
const metadata = PackTheMetadata({
@@ -240,6 +253,10 @@ export class CatalogController {
) {
const { username, user_id, customer_id, customer_name, permissions } = user;
this.logger.info(`GET /data-asset/${id}`, {
username,
customer_name,
});
const is_data_manager = permissions.includes(
PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER.seqid,
@@ -292,6 +309,10 @@ export class CatalogController {
) {
const { username, user_id, customer_id, customer_name, permissions } = user;
this.logger.info(`/catalog - ON GET ONE DASHBOARD METABASE ROUTE`, {
username,
customer_name,
});
const is_data_manager = permissions.includes(
PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER.seqid,
@@ -343,6 +364,10 @@ export class CatalogController {
): Promise<IColumnsMetadataResponse> {
const { customer_name, customer_id, user_id, username } = user;
this.logger.info(`/catalog - columns-metadata`, {
user_id,
customer_name,
});
const metadata = PackTheMetadata({
customer_name,
@@ -370,6 +395,10 @@ export class CatalogController {
): Promise<IPreviewResponse> {
const { customer_name, customer_id, user_id, username, customer_modules } = user;
this.logger.info(`/catalog - ON GET DATA DOCS ROUTE`, {
user_id,
customer_name,
});
const metadata = PackTheMetadata({
customer_name,
@@ -394,32 +423,26 @@ export class CatalogController {
@User() user: RequestUser,
@Language() language: LanguageEnum,
@Param('id') id: string,
@Query('asset_type') asset_type: string,
): Promise<IDocsResponse> {
try {
const { customer_name, customer_id, user_id, username } = user;
const { customer_name, customer_id, user_id, username } = user;
this.logger.info(`/catalog - ON GET DATA DOCS ROUTE`, {
user_id,
customer_name,
id,
});
this.logger.info(`/catalog - ON GET DATA DOCS ROUTE`, {
user_id,
customer_name,
});
const metadata = PackTheMetadata({
customer_name,
customer_id,
user_id,
username,
language,
});
const metadata = PackTheMetadata({
customer_name,
customer_id,
user_id,
username,
language,
});
const docs = await this.catalogService.getDataDocs(id, metadata);
const docs = await this.catalogService.getDataDocs(id, asset_type, metadata);
return { docs };
} catch (error) {
this.logger.error(`Error in getDataAssetDocs for id ${id}: ${error.message}`);
this.logger.error(`Error details: ${JSON.stringify(error)}`);
throw error;
}
return { docs };
}
@Put('data-asset/:id')
@@ -465,22 +488,36 @@ export class CatalogController {
@Headers() headers,
@Param('id') table_id: string,
@Body('docs') docs: string,
@Query('asset_type') asset_type: string,
) {
const { user_id, customer_name } = user;
const { user_id, customer_name, customer_id, username } = user;
const metadata = PackTheMetadata({
customer_id,
customer_name,
user_id,
username,
});
this.logger.info(`/catalog - ON POST DATA DOCS ROUTE`, {
user_id,
customer_name,
});
const res = await this.catalogService.createDataDocs({
const body = {
table_id,
docs,
asset_type,
info: {
customer: customer_name,
},
});
}
const res = await this.catalogService.createDataDocs(body, metadata);
return res;
}
@ApiInternalOnlyEndpoint()
@Put('data-asset/:id/manage-permissions')
async manageDataAssetPermissions(
@@ -906,88 +943,4 @@ export class CatalogController {
this.logger.error(error.message);
}
}
@Post('/data-asset/:nimbus_id/docs/ai')
@ApiOperation({
summary: 'Save documentation for data asset',
description: 'Saves documentation content for a data asset',
})
@ApiParam({
name: 'nimbus_id',
description: 'Nimbus ID of the data asset',
type: 'string',
})
@ApiResponse({
status: 201,
description: 'Documentation saved successfully',
})
async saveDocumentation(
@Param('nimbus_id') nimbusId: string,
@Body() body: { docs: string },
@User() user: RequestUser,
) {
const metadata = PackTheMetadata(user);
try {
await this.catalogService.updateDataAssetDocumentation(nimbusId, body.docs, metadata);
return {
message: 'Documentation saved successfully',
};
} catch (error) {
this.logger.error(`Error saving documentation for ${nimbusId}: ${error.message}`);
throw error;
}
}
@Post('/data-asset/:nimbus_id/docs/generate-ai')
@ApiOperation({
summary: 'Generate AI documentation for data asset',
description: 'Generates comprehensive documentation for a data asset using AI (Autodrive)',
})
@ApiParam({
name: 'nimbus_id',
description: 'Nimbus ID of the data asset',
type: 'string',
})
@ApiResponse({
status: 201,
description: 'AI documentation generated successfully',
schema: {
type: 'object',
properties: {
message: { type: 'string' },
documentation: { type: 'string' },
},
},
})
@ApiResponse({
status: 400,
description: 'Bad request - invalid nimbus_id or missing data',
})
@ApiResponse({
status: 404,
description: 'Data asset not found',
})
@ApiResponse({
status: 500,
description: 'Internal server error during AI generation',
})
async generateAiDocumentation(
@Param('nimbus_id') dataAssetId: string,
@User() user: RequestUser,
) {
const metadata = PackTheMetadata(user);
try {
const result = await this.catalogService.generateAiDocumentation(dataAssetId, metadata, user);
return {
message: 'AI documentation generated successfully',
documentation: result,
};
} catch (error) {
this.logger.error(`Error generating AI documentation for ${dataAssetId}: ${error.message}`);
throw error;
}
}
}
File diff suppressed because it is too large Load Diff
@@ -1,42 +0,0 @@
/**
* Constantes relacionadas ao Autodrive
*
*/
export const AUTODRIVE_CONSTANTS = {
// URLs e endpoints (apenas do ENV)
BASE_URL: process.env.BASE_URL_AUTODRIVE || process.env.AUTODRIVE_BASE_URL,
// Credenciais (apenas do ENV, sem fallback para segurança)
USERNAME: process.env.AUTODRIVE_USERNAME,
PASSWORD: process.env.AUTODRIVE_PASSWORD,
// Modelo padrão
DEFAULT_MODEL: process.env.AUTODRIVE_MODEL || "gpt-4o",
// Timeouts (em milissegundos)
ASK_TIMEOUT: 120000,
ANSWER_TIMEOUT: 180000,
// Headers
HEADERS: {
'Content-Type': 'application/json',
},
} as const;
/**
* Chaves geográficas para detecção de dados de localização
*/
export const GEOGRAPHIC_KEYS = [
'country', 'countries', 'city', 'cities',
'region', 'regions', 'location', 'state',
'states', 'address'
] as const;
/**
* Países comuns para detecção automática
*/
export const COMMON_COUNTRIES = [
'brazil', 'brasil', 'usa', 'united states',
'canada', 'mexico', 'argentina', 'chile',
'colombia'
] as const;
+25
View File
@@ -147,6 +147,24 @@ export class ICatalogAllRequest {
description: 'Tipo de ordenação - `asc`: crescente; `desc`: decrescente ',
})
order?: OrderEnum;
@ApiPropertyOptional({
description: 'ID do usuário owner para filtrar data assets',
example: 'user-id-1,user-id-2',
})
owner?: string;
@ApiPropertyOptional({
description: 'Data inicial para filtro de catálogo (formato: YYYY-MM-DD)',
example: '2025-01-01',
})
catalog_date_from?: string;
@ApiPropertyOptional({
description: 'Data final para filtro de catálogo (formato: YYYY-MM-DD)',
example: '2025-12-31',
})
catalog_date_to?: string;
}
export class ICatalogAllResponse {
@@ -328,3 +346,10 @@ export type AssetReporter = {
created_at: string;
tags: string;
}
export type CreateDataDocsDTO = {
table_id: string;
docs: string;
asset_type: string;
}
@@ -1,88 +0,0 @@
/**
*/
export const AI_DOCUMENTATION_PROMPT = `crie uma documentação em Portugues, Ingles e Espanhol seguindo essas instruções
1. Persona: como profissional de governança e engenharia de dados
2. Tarefa: ao receber as informações da tabela criar uma documentação com o seguinte escopo
**A primeira linha do documento tem que conter a seguinte informação: ## Document languages: EN / BR / ES
**A segunda linha tem que obrigatoriamente conter a escrita Table: nome da tabela
**A terceira linha tem que obrigatoriamente conter a escrita Table Schema: nome do table schema
**DIRETRIZ CRUCIAL DE CONSISTÊNCIA E COMPLETUDE DE SCHEMA:**
**1. Fonte Exclusiva de Metadados:** O 'Table Schema' definido na linha acima é a ÚNICA fonte de verdade para o schema dos dados a serem documentados. TODAS as informações subsequentes, especialmente na seção 'Estrutura da Tabela' (incluindo a lista de colunas, seus nomes, tipos de dados, descrições e exemplos) DEVEM ser extraídas EXCLUSIVAMENTE de metadados que correspondem a ESTE 'Table Schema'. Se os dados de entrada que você recebeu contiverem informações para a mesma tabela ou colunas mas de schemas diferentes (ex: um schema 'bronze' e um 'silver'), você DEVE IGNORAR TOTALMENTE as informações dos schemas divergentes para esta tarefa de documentação e utilizar APENAS as do 'Table Schema' aqui especificado.
**2. Listagem Completa de Colunas:** Sua principal tarefa na seção 'Estrutura da Tabela' é identificar e listar TODAS as colunas que pertencem ao 'Table Schema' especificado. Verifique nos dados de entrada fornecidos se há uma indicação explícita do número total de colunas para esta tabela neste schema (por exemplo, um campo como 'Num_columns' ou similar nos metadados da tabela). Você deve se esforçar para listar exatamente essa quantidade de colunas. Se essa contagem não estiver disponível, liste todas as colunas que você puder identificar como pertencentes exclusivamente a este 'Table Schema'. A completude em relação ao schema especificado é essencial.
**Depois de "Estrutura da tablea", incluir a mensagem "Este documento foi gerado por IA", traduzida corretamente para cada idioma.**
**Obrigatoriamente:Após finalizar a versão em Inglês, começar a versão em Português** **Após finalizar a versão em Português, começar a versão em Espanhol** **Antes de começar cada versão, colocar um título como:** - \`## English Version\` (para inglês)
- \`## Versão em Português\` (para português)
- \`## Versión en Español\` (para espanhol)
- Descrição: fornece uma visão geral do ativo de dados,
destacando seu propósito e principal funcionalidade.
Esta sessão resume o conteúdo e o objetivo do ativo, ajudando os usuários a entender rapidamente o que o ativo representa
e como pode ser utilizado em suas análises e decisões.
- Sugestão de Domínio de Dados:
Analise cuidadosamente os dados da tabela e sugira o domínio mais apropriado. Inclua:
- Domínio Sugerido: [Nome do domínio]
- Motivo: [Explicação breve sobre porque a tabela pertence a este domínio]
- Observações: [Qualquer observação adicional relevante]
Exemplos de Domínios de Dados para referência:
- Financeiro: Dados sobre transações, receitas, despesas, etc.
- Recursos Humanos: Dados sobre funcionários, cargos, salários, etc.
- Produtos: Dados sobre produtos, categorias, preços, etc.
- Fornecedores: Dados sobre fornecedores, produtos fornecidos, localizações, etc.
- Marketing: Dados sobre campanhas, leads, conversões, etc.
- Vendas: Dados sobre vendas, clientes, produtos vendidos, etc.
- Operações: Dados sobre processos, logística, produção, etc.
- Clientes: Dados sobre clientes, interações, histórico, etc.
-Tags Sugeridas:
A IA deve gerar tags relevantes **com base nos dados da tabela**.
- **IMPORTANTE: Analise cuidadosamente os dados de preview da tabela (PREVIEW DATA) para encontrar países. Procure em todas as colunas por nomes de países, cidades ou regiões.**
- **Garanta que as tags estejam separadas por espaços vazios, todas na mesma linha, exemplo: #marketing #sales #australia #canada, limitar até 3 países que mais aparecem** - **Os países DEVEM ser extraídos dos dados de preview da tabela. Procure em colunas como City, Country, Region, Location, etc.** - Por que esta tabela é interessante:
Nesta sessão, é destacada a importância do ativo, explicando como ele pode ser útil para os usuários.
São abordadas as formas como o ativo pode melhorar a tomada de decisões, identificar padrões relevantes ou fornecer insights valiosos.
O objetivo é ressaltar a utilidade prática e o impacto positivo que o ativo pode ter em suas atividades.
- Análises potencialmente úteis feitas com esses dados:
Aqui são listadas algumas das análises que podem ser realizadas com o ativo de dados. Inclui sugestões de dashboards,
relatórios ou outros tipos de análises que aproveitam as informações fornecidas pelo ativo.
O objetivo é oferecer maneiras de utilizar os dados para obter insights valiosos e apoiar a tomada de decisões informadas.
- Links Úteis:
Os Links Úteis oferecem recursos adicionais relacionados ao ativo de dados, incluindo guias,
artigos ou outras fontes de informação que podem ajudar os usuários a compreender melhor o ativo e suas aplicações. Além disso,
inclui um link rápido dentro da Dadosfera para ativos relacionados diretamente com o ativo em questão, facilitando a navegação entre os ativos.
- Estrutura da Tabela:
A Estrutura da Tabela detalha TODAS as colunas e os dados disponíveis no ativo, conforme pertencentes ao 'Table Schema' principal definido no início deste documento.
**Instrução Detalhada para Estrutura da Tabela:**
Siga rigorosamente estes passos:
1. Identifique nos dados de entrada (metadados da tabela e das colunas) todas as colunas que pertencem EXCLUSIVAMENTE ao 'Table Schema' especificado no cabeçalho deste documento. Se houver uma contagem de colunas (ex: 'Num_columns') para este schema específico, assegure-se de listar essa quantidade.
2. Para CADA uma dessas colunas identificadas, formate a saída da seguinte maneira, **SEM utilizar NENHUM marcador de lista (como traços ou asteriscos) no início de cada entrada de coluna**. Cada coluna deve ser apresentada como um bloco de texto. Inclua uma linha em branco entre a documentação de cada coluna para separação visual.
- Apresente o NOME_DA_COLUNA em maiúsculas, seguido pelo (TIPO_DE_DADO_EXTRAÍDO_DOS_METADADOS_DO_SCHEMA_CORRETO) entre parênteses.
- O **NOME_DA_COLUNA (TIPO_DE_DADO_EXTRAÍDO_DOS_METADADOS_DO_SCHEMA_CORRETO)** deve estar na primeira linha do bloco da coluna e **inteiramente em negrito**.
- Na linha seguinte, a etiqueta "**Descrição:**" deve estar **em negrito**, seguida pelo texto da descrição da coluna.
- Na linha seguinte à descrição, a etiqueta "**Exemplo:**" deve estar **em negrito**, seguida pelo valor do exemplo. Se o exemplo for um valor literal ou código, formate-o entre crases (\`) se apropriado.
- Se houver informações adicionais relevantes (como "Valores Possíveis:", "Observações:", etc.), coloque a etiqueta correspondente **em negrito** em uma nova linha, seguida pelo seu texto.
Este documento foi gerado por IA.
NOME_COLUNA_1 (TIPO_DADO_SCHEMA_CORRETO_1):
Descrição: [Descrição da coluna 1, do schema correto]
Exemplo: \`[Exemplo de valor para coluna 1, do schema correto]\`
NOME_COLUNA_2 (TIPO_DADO_SCHEMA_CORRETO_2):
Descrição: [Descrição da coluna 2, do schema correto]
Exemplo: \`[Exemplo de valor para coluna 2, do schema correto]\`
(continue este formato com início de cada coluna para TODAS as colunas do 'Table Schema' especificado, garanta com que NUNCA tenha TRAÇO OU PONTO no inicio)
3. Contexto : O usuário ira cadastrar um ativo de dados na nossa plataforma e para ter um bom catalogo ele ira querer gerar a documentação padronizada mas explicativa e
automática
4. Restrições : A documentação deve seguir obrigatoriamente o mesmo padrão principalmente na parte de estrutura de dados
5. Objetivo: O principal objetivo é gerar uma documentação acessível, clara,
automática e padronizada para os usuários que desejem cadastrar um ativo de dados na plataforma`;
/**
* Configurações para a geração de documentação com IA
*/
export const AI_DOCUMENTATION_CONFIG = {
FETCH_K: 250,
K: 100,
} as const;
+3 -3
View File
@@ -160,7 +160,7 @@ export class ShareService implements OnModuleInit {
});
const { documentation } = await lastValueFrom(
this.catalogReadService.GetDatasetDoc({ id, type: undefined }, metadata),
this.catalogReadService.GetDatasetDoc({ id }, metadata),
);
console.log(documentation);
const docs = JSON.parse(documentation);
@@ -180,7 +180,7 @@ export class ShareService implements OnModuleInit {
return data_assets.map((data_asset) => {
const owner = customer_users.find(
(u) => u.id === data_asset.owner,
)?.username;
)?.email;
const roles = [];
const users = [];
@@ -190,7 +190,7 @@ export class ShareService implements OnModuleInit {
}
for (const user_id of data_asset.users) {
const user = customer_users.find((r) => r.id === user_id);
if (user) users.push({ id: user.id, username: user.username });
if (user) users.push({ id: user.id, email: user.email });
}
return {
...data_asset,
@@ -1,60 +0,0 @@
/**
* Tipos relacionados à geração de documentação com IA
*/
export interface AutodriveCredentials {
username: string;
password: string;
baseUrl: string;
model: string;
authHeader?: string;
}
export interface AutodriveAskPayload {
question: string;
fetch_k: number;
k: number;
model: string;
}
export interface AutodriveAskResponse {
answer?: string;
question_id?: string;
dataset_id?: string;
}
export interface AutodriveAnswerResponse {
status: 'started' | 'success' | 'failed';
answer?: string;
status_reason?: string;
}
export interface AutodriveUploadResponse {
dataset_id: string;
}
export interface DatasetStatusResponse {
status: 'processing' | 'success' | 'failed';
status_reason?: string;
}
export interface ColumnData {
name: string;
type: string;
description?: string;
nullable?: string;
}
export interface ColumnsMetadata {
columns: ColumnData[];
}
export interface DataPreview {
preview: any[];
}
export interface FormattedDataForAI {
dataAsset: any;
dataPreview: any[];
columnsData?: ColumnsMetadata;
}
+25 -8
View File
@@ -1,29 +1,46 @@
import { Body, Controller, Inject, Param, Post, Req } from '@nestjs/common';
import { init } from 'mixpanel';
import { Authenticated } from 'src/decorators/authentication.decorator';
import { ApiInternalOnlyController } from 'src/decorators/swagger.decorator';
import { RequestUser, User } from 'src/decorators/user.decorator';
import { RequestUser } from 'src/decorators/user.decorator';
import { MixpanelService } from './mixpanel.service';
import { extractUserFrom } from 'src/authentication/extract-user';
import DadosferaLogger from '@dadosfera/dadosfera-logs';
@ApiInternalOnlyController()
@Controller('trackEvent')
export class MixpanelController {
logger: DadosferaLogger;
constructor(
private mixpanelService: MixpanelService
) {}
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
private mixpanelService: MixpanelService,
) {
this.logger = dadosferaLogger.logger;
}
@Post(':id')
async trackEvent(
@Param('id') id,
@Body() body,
@User() user: RequestUser,
@Req() request
) {
this.logger.info(`POST Track Event: ${id}`)
delete body.info;
const anonymousUser = {
username: "anonymous",
customer_name: "anonymous"
} as RequestUser
const hasToken = request.headers['authorization'];
const user = hasToken ? extractUserFrom(hasToken) : anonymousUser;
this.logger.info(`Has user: ${typeof hasToken == "string"}`)
await this.mixpanelService.track(id, user, request, body)
this.logger.info(`Event successful`)
return { id, body, user: user.username };
}
}
+3 -20
View File
@@ -3,6 +3,7 @@ import { ApiOperation, ApiTags } from '@nestjs/swagger';
import {
AuthenticateCondition,
Authenticated,
RequireSomePermission,
} from 'src/decorators/authentication.decorator';
import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
import { PipelinesService } from './pipelines.service';
@@ -13,26 +14,6 @@ import { ApiInternalOnlyController } from 'src/decorators/swagger.decorator';
@ApiTags('Pipelines')
@Controller('pipelines')
@Authenticated()
@AuthenticateCondition((req, user) => {
let action;
switch (req.method) {
case 'POST':
action = 'CREATE';
break;
case 'PUT':
action = 'UPDATE';
break;
default:
action = req.method;
}
return user.permissions.includes(
PERMISSIONS_GROUPS.PIPELINE.permissions[action].seqid,
);
})
export class PipelinesController {
logger: DadosferaLogger;
constructor(
@@ -44,6 +25,7 @@ export class PipelinesController {
}
@Post('start/:id')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
@ApiOperation({
deprecated: true,
description:
@@ -71,6 +53,7 @@ export class PipelinesController {
description:
'This method is deprecated. Please use route /pipelinesV2/:id/status instead',
})
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineStatus(@Body() body, @Param('id') id: string) {
body.id = id;
+16 -27
View File
@@ -26,6 +26,7 @@ import {
import {
AuthenticateCondition,
RequireAllPermissions,
RequireSomePermission,
} from 'src/decorators/authentication.decorator';
import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
import { PipelinesService } from './pipelines.service';
@@ -52,30 +53,6 @@ import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator';
@ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
@UseFilters(new GrpcToHttpExceptionFilter())
@Controller('pipelinesV2')
@AuthenticateCondition((req, user) => {
let action;
switch (req.method) {
case 'POST':
action = 'CREATE';
break;
case 'PUT':
action = 'UPDATE';
break;
case 'PATCH':
action = 'UPDATE';
break;
default:
action = req.method;
}
return user.permissions.includes(
PERMISSIONS_GROUPS.PIPELINE.permissions[action].seqid,
);
})
export class PipelinesController {
logger: DadosferaLogger;
constructor(
@@ -88,6 +65,7 @@ export class PipelinesController {
}
@Get('monitoring-dashboard')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getMonitoringDashboard(@User() user: RequestUser) {
this.logger.info('PipelinesController - getMonitoringDashboard', { user });
@@ -100,6 +78,7 @@ export class PipelinesController {
}
@Post()
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
@ApiCreatedResponse({ type: IPipelineV2 })
async create(
@Language() language: LanguageEnum,
@@ -126,6 +105,7 @@ export class PipelinesController {
}
@Get()
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async findAll(
@User() user: RequestUser,
@Language() language: LanguageEnum,
@@ -150,6 +130,7 @@ export class PipelinesController {
}
@Get('/download-logs')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async downloadLogs(
@User() user: RequestUser,
@Language() language: LanguageEnum,
@@ -180,6 +161,7 @@ export class PipelinesController {
}
@Get(':id/config')
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW,PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineproperties(
@Language() language: LanguageEnum,
@User() user: RequestUser,
@@ -191,6 +173,7 @@ export class PipelinesController {
}
@Get(':id/objects')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineObjects(
@Language() language: LanguageEnum,
@User() user: RequestUser,
@@ -202,6 +185,7 @@ export class PipelinesController {
}
@Get(':id/status')
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineStatus(@Body() body, @Param('id') id: string) {
body.id = id;
@@ -216,6 +200,7 @@ export class PipelinesController {
}
@Get('/:id')
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async findOne(
@Language() language: LanguageEnum,
@User() user: RequestUser,
@@ -258,6 +243,7 @@ export class PipelinesController {
}
@Patch('/:id')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async update(
@Language() language: LanguageEnum,
@Body() updatePipelineDto,
@@ -297,6 +283,7 @@ export class PipelinesController {
deprecated: true,
description: 'This method is deprecated. Please use PATCH instead',
})
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updateDeprecated(
@Language() language: LanguageEnum,
@Body() updatePipelineDto,
@@ -312,6 +299,7 @@ export class PipelinesController {
@Delete(':id')
@ApiNoContentResponse()
@HttpCode(HttpStatus.NO_CONTENT)
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
async delete(@Param('id') id: string, @User() user: RequestUser) {
this.logger.info('PipelinesController - delete', { user });
const metadata = PackTheMetadata({
@@ -325,7 +313,7 @@ export class PipelinesController {
@ApiInternalOnlyEndpoint()
@Post('/init-upload')
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
@RequireAllPermissions(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW)
async initUploadFile(
@User() user: RequestUser,
@Body() body: IInitUploadCSVFile,
@@ -357,7 +345,7 @@ export class PipelinesController {
@ApiInternalOnlyEndpoint()
@Post('/complete-upload')
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
@RequireAllPermissions(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW)
async completeUploadFile(
@User() user: RequestUser,
@Body() body: ICompleteUploadCSVFile,
@@ -375,7 +363,7 @@ export class PipelinesController {
@ApiInternalOnlyEndpoint()
@Post('/file')
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
@RequireAllPermissions(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW)
async uploadedFile(
@User() user: RequestUser,
@Body() body: ICreatePipelineCSVFile,
@@ -425,6 +413,7 @@ export class PipelinesController {
@ApiInternalOnlyEndpoint()
@Post('start/:id')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
async activate(@Param('id') id: string, @Body() body) {
const { info } = body;
@@ -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),
};
@@ -0,0 +1,977 @@
import {
Controller,
Get,
Post,
Put,
Patch,
Delete,
Param,
Body,
Query,
Inject,
BadRequestException,
} from '@nestjs/common';
import { ApiTags, ApiOperation } from '@nestjs/swagger';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import {
Authenticated,
RequireAllPermissions,
} from '../../decorators/authentication.decorator';
import { User, RequestUser } from '../../decorators/user.decorator';
import { PlatformApiService } from './platform-api.service';
import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
import { ElasticsearchService } from '../../services/elasticsearch';
import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
import { CustomersService } from '../customers/customers.service';
import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
@ApiTags('Platform API')
@Controller('platform')
export class PlatformApiController {
private logger: any;
constructor(
private readonly platformApiService: PlatformApiService,
private readonly elasticsearchService: ElasticsearchService,
private readonly dynamoDBService: DynamoDBService,
private readonly customersService: CustomersService,
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
) {
this.logger = dadosferaLogger.logger;
}
/**
* Normalize pipeline ID to match Platform-API format.
* Platform-API replaces '-' with '_' in pipeline IDs.
*/
private normalizePipelineId(id: string): string {
return id?.replace(/-/g, '_') || '';
}
/**
* Denormalize ID back to UUID format (replace _ with -).
* Used when we receive a normalized ID but need the original UUID.
*/
private denormalizeId(id: string): string {
return id?.replace(/_/g, '-') || '';
}
/**
* Normalize job ID to match Platform-API format.
* Platform-API replaces '-' with '_' in job IDs.
*
* Example: "2ccf5481-59f5-4036-8a94-7d5f28f4f899-0" -> "2ccf5481_59f5_4036_8a94_7d5f28f4f899_0"
*/
private normalizeJobId(jobId: string): string {
return jobId?.replace(/-/g, '_') || '';
}
/**
* Extract the pipeline ID (base UUID) from a job ID.
* Job IDs have format "uuid-suffix" where suffix is the job index (e.g., "0", "1").
* Handles both hyphenated and underscored formats, always returns hyphenated UUID for ES.
*
* Examples:
* - "2ccf5481-59f5-4036-8a94-7d5f28f4f899-0" -> "2ccf5481-59f5-4036-8a94-7d5f28f4f899"
* - "2ccf5481_59f5_4036_8a94_7d5f28f4f899_0" -> "2ccf5481-59f5-4036-8a94-7d5f28f4f899"
*/
private extractPipelineIdFromJobId(jobId: string): string {
if (!jobId) return '';
// Determine the separator used in the jobId
const hasUnderscores = jobId.includes('_');
const separator = hasUnderscores ? '_' : '-';
const parts = jobId.split(separator);
// UUID has 5 parts (8-4-4-4-12), job suffix is the 6th part
if (parts.length >= 6) {
// Always return hyphenated format for Elasticsearch lookup
return parts.slice(0, 5).join('-');
}
// If no suffix found, return the ID in hyphenated format
return hasUnderscores ? jobId.replace(/_/g, '-') : jobId;
}
private readonly VALID_CONNECTORS = ['jdbc', 'singer', 's3'];
private readonly MAX_MEMORY_MB = 12000; // 12GB maximum memory per pipeline/job
/**
* Validate that connector is provided and is a valid type.
*/
private validateConnector(connector: string | undefined): void {
if (!connector || !this.VALID_CONNECTORS.includes(connector)) {
throw new BadRequestException(
`connector is required in job input and must be one of: ${this.VALID_CONNECTORS.join(', ')}`,
);
}
}
/**
* Validate memory allocation against maximum limit.
*/
private validateMemory(memoryMb: number): void {
if (memoryMb > this.MAX_MEMORY_MB) {
throw new BadRequestException(
`Memory limit exceeded. Maximum allowed: ${this.MAX_MEMORY_MB}MB (12GB)`,
);
}
}
/**
* Validate cron expression against customer's schedule limit.
* Fetches current scheduleLimit from DUC to ensure up-to-date configuration.
*/
private async validateScheduleLimit(cron: string, customerId: string): Promise<void> {
if (!cron) return;
const { customer } = await this.customersService.getCustomer(customerId);
const scheduleLimit = customer?.scheduleLimit || 'day';
const result = validateCronAgainstScheduleLimit(cron, scheduleLimit);
if (!result.valid) {
throw new BadRequestException(result.message);
}
}
/**
* Map connector type to DynamoDB type.
* jdbc -> 'database', singer -> 'application', s3 -> 'file'
*/
private mapConnectorToDynamoType(connector: string): string {
switch (connector) {
case 'jdbc':
return 'database';
case 'singer':
return 'application';
case 's3':
return 'file';
default:
return connector;
}
}
/**
* Extract and transform tables from jobs for DynamoDB input.
* Maps connector-specific fields to a common table format.
*
* - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name/type (reference_column object)
* - Singer: type maps replication_method (FULL_TABLE -> full_load, INCREMENTAL -> incremental), no columns
* - S3: same mapping as Singer, no columns
*/
private extractTablesFromJobs(jobs: any[], connector: string): Array<{
name: string;
type: string;
columns?: string[];
reference_column?: ReferenceColumn;
}> {
if (!jobs || jobs.length === 0) return [];
const tables: Array<{
name: string;
type: string;
columns?: string[];
reference_column?: ReferenceColumn;
}> = [];
for (const job of jobs) {
const input = job.input;
if (!input) continue;
if (connector === 'jdbc') {
// JDBC: table_name, load_type, column_include_list, incremental_column_name/type
const table: {
name: string;
type: string;
columns?: string[];
reference_column?: ReferenceColumn;
} = {
name: input.table_name || '',
type: input.load_type || 'full_load',
};
if (input.column_include_list && input.column_include_list.length > 0) {
table.columns = input.column_include_list;
}
if (input.incremental_column_name) {
// reference_column is stored as an object with name and type
table.reference_column = {
name: input.incremental_column_name,
type: input.incremental_column_type || 'unknown',
};
}
tables.push(table);
} else if (connector === 'singer' || connector === 's3') {
// Singer/S3: replication_method -> type mapping, no columns
let type = 'full_load';
if (input.replication_method === 'INCREMENTAL') {
type = 'incremental';
} else if (input.replication_method === 'FULL_TABLE') {
type = 'full_load';
}
tables.push({
name: input.table_name || '',
type,
});
}
}
return tables;
}
/**
* Build properties object for Elasticsearch based on connector type.
* Different connectors have different property structures.
*
* Note: In pi-factory flow, properties come pre-built from frontend.
* In Maestro proxy flow, we reconstruct from job input fields.
*/
private buildPipelineProperties(jobInput: any): Record<string, any> {
if (!jobInput) return {};
const connector = jobInput.connector;
const properties: Record<string, any> = {};
// Determine credentials_type
if (jobInput.auth_parameters?.credentials_type) {
properties.credentials_type = jobInput.auth_parameters.credentials_type;
} else {
// Default based on connector type
// S3 connector typically uses iam_user, others use basic_auth
properties.credentials_type = connector === 's3' ? 'iam_user' : 'basic_auth';
}
if (connector === 'jdbc') {
// JDBC connectors: schema comes from table_schema
if (jobInput.table_schema) {
properties.schema = jobInput.table_schema;
}
} else if (connector === 'singer') {
// Singer connectors: merge config fields (dates, selected_competitions, etc.)
if (jobInput.config) {
Object.assign(properties, jobInput.config);
}
} else if (connector === 's3') {
// S3 connector
if (jobInput.engine) properties.engine = jobInput.engine;
if (jobInput.source_bucket) properties.source_bucket = jobInput.source_bucket;
if (jobInput.source_prefix) properties.source_prefix = jobInput.source_prefix;
if (jobInput.file_format_params) properties.file_format_params = jobInput.file_format_params;
}
return properties;
}
/**
* Sync job input changes to DynamoDB for a specific connector type.
* Extracts pipeline ID from job ID, fetches ES document to find input ID,
* then updates the table entry in DynamoDB.
*
* Job ID transformations:
* - Raw format (from endpoint): "2ccf5481-59f5-4036-8a94-7d5f28f4f899-0"
* - Platform API format: "2ccf5481_59f5_4036_8a94_7d5f28f4f899_0" (underscores)
* - Elasticsearch pipeline ID: "2ccf5481-59f5-4036-8a94-7d5f28f4f899" (UUID only, hyphens)
*
* @param connectorType - The connector type ('jdbc', 'singer', 's3') for the Platform API endpoint
*/
private async syncJobInputToDynamoDB(
jobId: string,
body: any,
user: RequestUser,
connectorType: 'jdbc' | 'singer' | 's3',
): Promise<void> {
try {
// Normalize job ID for Platform API GET (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
// Get job details using connector-specific endpoint to find table_name
const jobResult = await this.platformApiService.proxy(
'GET',
`/jobs/${connectorType}/${normalizedJobId}`,
user,
);
// Extract the pipeline ID (base UUID) from the raw job ID for ES lookup
const esPipelineId = this.extractPipelineIdFromJobId(jobId);
const tableName = body.table_name || jobResult.source_config?.table_name;
if (!esPipelineId || !tableName) {
this.logger.warn('Cannot sync job input: missing pipeline_id or table_name', {
jobId,
esPipelineId,
tableName,
});
return;
}
// Get pipeline from ES to find input ID (stored in config.tables)
const pipeline = await this.elasticsearchService.getPipeline(
user.customer_name,
esPipelineId,
);
const inputId = pipeline?.config?.tables;
if (!inputId) {
this.logger.warn('Cannot sync job input: no input ID in ES', {
jobId,
esPipelineId,
});
return;
}
// Build changes for DynamoDB table entry
// reference_column is stored as an object with name and type
const changes: {
type?: string;
columns?: string[];
reference_column?: ReferenceColumn | null;
} = {};
if ('target_load_type' in body) {
changes.type = body.target_load_type;
}
if ('column_include_list' in body) {
changes.columns = body.column_include_list;
}
if ('incremental_column_name' in body) {
// reference_column is stored as an object with name and type
if (body.incremental_column_name) {
changes.reference_column = {
name: body.incremental_column_name,
type: body.incremental_column_type || 'unknown',
};
} else {
changes.reference_column = null;
}
}
// Update DynamoDB if there are changes
if (Object.keys(changes).length > 0) {
await this.dynamoDBService.updateInputTable(
user.customer_id,
inputId,
tableName,
changes,
);
}
} catch (error) {
this.logger.error('Failed to sync job input to DynamoDB', {
jobId,
connectorType,
error: error.message,
});
// Don't throw - Platform API update succeeded, just log the sync error
}
}
/**
* Sync sync-mode changes to DynamoDB for JDBC connectors.
* Always passes both target_load_type and incremental_column_name to ensure proper sync.
*/
private async syncJdbcSyncModeToDynamoDB(
jobId: string,
body: any,
user: RequestUser,
): Promise<void> {
// JDBC sync mode uses target_load_type field
const changes: any = {};
if ('target_load_type' in body) {
changes.target_load_type = body.target_load_type;
}
// Handle incremental_column_name:
// - If provided in body, use that value
// - If changing to full_load, explicitly clear it
if ('incremental_column_name' in body) {
changes.incremental_column_name = body.incremental_column_name;
changes.incremental_column_type = body.incremental_column_type;
} else if (body.target_load_type === 'full_load') {
// Changing to full_load without specifying incremental_column - clear it
changes.incremental_column_name = null;
}
await this.syncJobInputToDynamoDB(jobId, changes, user, 'jdbc');
}
/**
* Sync sync-mode changes to DynamoDB for Singer connectors.
*/
private async syncSingerSyncModeToDynamoDB(
jobId: string,
body: any,
user: RequestUser,
): Promise<void> {
// Singer sync mode uses replication_method field
// Map to DynamoDB type: FULL_TABLE -> full_load, INCREMENTAL -> incremental
if ('replication_method' in body) {
const type = body.replication_method === 'INCREMENTAL' ? 'incremental' : 'full_load';
await this.syncJobInputToDynamoDB(jobId, { load_type: type }, user, 'singer');
}
}
// ==================== PIPELINE ROUTES ====================
@Post('pipeline')
@ApiOperation({ summary: 'Create a new pipeline' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
async createPipeline(@Body() body: any, @User() user: RequestUser) {
// Validate that pipeline has jobs
if (!body.jobs || body.jobs.length === 0) {
throw new BadRequestException('Pipeline must have at least one job');
}
// Validate connector before proceeding
const firstJob = body.jobs[0]?.input;
this.validateConnector(firstJob?.connector);
// Validate cron against customer's schedule limit
await this.validateScheduleLimit(body.cron, user.customer_id);
// Inject customer_id (actually customer_name) and normalized ID into body for Platform-API
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
const enrichedBody = {
...body,
id: this.normalizePipelineId(body.id),
customer_id: user.customer_name,
};
const result = await this.platformApiService.proxy('POST', '/pipeline', user, enrichedBody);
// Sync to Elasticsearch and DynamoDB
try {
const plugin = firstJob?.plugin;
const connectorType = firstJob?.connector;
const connectionId = firstJob?.auth_parameters?.config_id;
// Build properties based on connector type
const properties = this.buildPipelineProperties(firstJob);
const connector = plugin
? await this.elasticsearchService.getConnectorByPlugin(plugin)
: null;
// Extract tables from jobs and create DynamoDB input
const tables = this.extractTablesFromJobs(body.jobs, connectorType);
let inputId: string | undefined;
if (tables.length > 0) {
const inputDoc = await this.dynamoDBService.createInput(
user.customer_id,
user.user_id,
{
name: body.name,
description: body.description,
plugin: plugin || '',
type: this.mapConnectorToDynamoType(connectorType),
tables,
},
);
inputId = inputDoc.id;
this.logger.info('Created DynamoDB input for tables config', {
inputId,
pipelineId: body.id,
tablesCount: tables.length,
});
}
const pipelineType = this.mapConnectorToDynamoType(connectorType);
this.logger.info('Syncing pipeline to Elasticsearch', {
customerName: user.customer_name,
pipelineId: body.id,
plugin,
connector: connectorType,
type: pipelineType,
properties,
inputId,
});
// Keep original UUID format for Elasticsearch (not normalized)
await this.elasticsearchService.createPipeline(
user.customer_name,
body.id,
{
name: body.name,
description: body.description,
user_id: user.user_id,
username: user.username,
customer_id: user.customer_id,
plugin,
connection_id: connectionId,
cron: body.cron,
tables: inputId,
properties,
type: pipelineType,
},
connector,
);
} catch (error) {
this.logger.error('Failed to sync pipeline creation to Elasticsearch/DynamoDB', {
pipelineId: body.id,
customerName: user.customer_name,
error: error.message,
errorName: error.name,
});
}
return result;
}
@Get('pipelines')
@ApiOperation({ summary: 'List all pipelines for customer' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelines(
@User() user: RequestUser,
@Query() query: Record<string, string>,
) {
return this.platformApiService.proxy(
'GET',
'/pipelines',
user,
undefined,
query,
);
}
@Get('pipeline/:pipelineId')
@ApiOperation({ summary: 'Get pipeline by ID' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipeline(
@Param('pipelineId') pipelineId: string,
@User() user: RequestUser,
) {
const normalizedId = this.normalizePipelineId(pipelineId);
return this.platformApiService.proxy('GET', `/pipeline/${normalizedId}`, user);
}
@Patch('pipeline/:pipelineId')
@ApiOperation({ summary: 'Update pipeline by ID' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updatePipeline(
@Param('pipelineId') pipelineId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Validate cron against customer's schedule limit if cron is being updated
if (body.cron) {
await this.validateScheduleLimit(body.cron, user.customer_id);
}
const normalizedId = this.normalizePipelineId(pipelineId);
const result = await this.platformApiService.proxy(
'PATCH',
`/pipeline/${normalizedId}`,
user,
body,
);
// Sync to Elasticsearch (use original UUID, not normalized)
// Only pass fields that are explicitly provided in the request body
try {
const esChanges: {
name?: string;
description?: string;
cron?: string;
status?: string;
} = {};
if ('name' in body) esChanges.name = body.name;
if ('description' in body) esChanges.description = body.description;
if ('cron' in body) esChanges.cron = body.cron;
if ('status' in body) esChanges.status = body.status;
await this.elasticsearchService.updatePipeline(
user.customer_name,
pipelineId,
esChanges,
);
} catch (error) {
this.logger.error('Failed to sync pipeline update to Elasticsearch', {
pipelineId,
error: error.message,
});
}
return result;
}
@Delete('pipeline/:pipelineId')
@ApiOperation({ summary: 'Delete pipeline by ID' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
async deletePipeline(
@Param('pipelineId') pipelineId: string,
@User() user: RequestUser,
) {
const normalizedId = this.normalizePipelineId(pipelineId);
const result = await this.platformApiService.proxy(
'DELETE',
`/pipeline/${normalizedId}`,
user,
);
// Sync to Elasticsearch (use original UUID, not normalized)
try {
await this.elasticsearchService.deletePipeline(
user.customer_name,
pipelineId,
);
} catch (error) {
this.logger.error('Failed to sync pipeline deletion to Elasticsearch', {
pipelineId,
error: error.message,
});
}
return result;
}
@Post('pipeline/execute')
@ApiOperation({ summary: 'Execute a pipeline' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async executePipeline(@Body() body: any, @User() user: RequestUser) {
// Inject customer_id (actually customer_name) into body for Platform-API
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
const enrichedBody = {
...body,
customer_id: user.customer_name,
};
return this.platformApiService.proxy('POST', '/pipeline/execute', user, enrichedBody);
}
@Post('pipeline/pause')
@ApiOperation({ summary: 'Pause a pipeline' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async pausePipeline(@Body() body: any, @User() user: RequestUser) {
// Inject customer_id (actually customer_name) into body for Platform-API
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
const enrichedBody = {
...body,
customer_id: user.customer_name,
};
return this.platformApiService.proxy('POST', '/pipeline/pause', user, enrichedBody);
}
@Post('pipeline/unpause')
@ApiOperation({ summary: 'Unpause a pipeline' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async unpausePipeline(@Body() body: any, @User() user: RequestUser) {
// Inject customer_id (actually customer_name) into body for Platform-API
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
const enrichedBody = {
...body,
customer_id: user.customer_name,
};
return this.platformApiService.proxy('POST', '/pipeline/unpause', user, enrichedBody);
}
@Put('pipeline/:pipelineId/memory')
@ApiOperation({ summary: 'Update pipeline memory configuration' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updatePipelineMemory(
@Param('pipelineId') pipelineId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Validate memory limit
if (body.amount) {
this.validateMemory(body.amount);
}
return this.platformApiService.proxy(
'PUT',
`/pipeline/${pipelineId}/memory`,
user,
body,
);
}
// ==================== PIPELINE METADATA ROUTES ====================
@Put('pipeline/:pipelineId/metadata')
@ApiOperation({ summary: 'Update pipeline metadata' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updatePipelineMetadata(
@Param('pipelineId') pipelineId: string,
@Body() body: any,
@User() user: RequestUser,
) {
return this.platformApiService.proxy(
'PUT',
`/pipeline/${pipelineId}/metadata`,
user,
body,
);
}
@Get('pipelines/metadata')
@ApiOperation({ summary: 'Get all pipelines metadata' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelinesMetadata(
@User() user: RequestUser,
@Query() query: Record<string, string>,
) {
return this.platformApiService.proxy(
'GET',
'/pipelines/metadata',
user,
undefined,
query,
);
}
// ==================== PIPELINE RUN ROUTES ====================
@Get('pipeline/:pipelineId/pipeline_run')
@ApiOperation({ summary: 'Get pipeline runs for a pipeline' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineRuns(
@Param('pipelineId') pipelineId: string,
@User() user: RequestUser,
@Query() query: Record<string, string>,
) {
const normalizedId = this.normalizePipelineId(pipelineId);
return this.platformApiService.proxy(
'GET',
`/pipeline/${normalizedId}/pipeline_run`,
user,
undefined,
query,
);
}
@Get('pipeline/:pipelineId/pipeline_run/:runId')
@ApiOperation({ summary: 'Get specific pipeline run' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineRun(
@Param('pipelineId') pipelineId: string,
@Param('runId') runId: string,
@User() user: RequestUser,
) {
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
const normalizedRunId = this.normalizePipelineId(runId);
return this.platformApiService.proxy(
'GET',
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`,
user,
);
}
@Get('pipeline/pipeline_run/:runId/logs')
@ApiOperation({ summary: 'Get pipeline run logs' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineRunLogs(
@Param('runId') runId: string,
@User() user: RequestUser,
@Query() query: Record<string, string>,
) {
const normalizedRunId = this.normalizePipelineId(runId);
return this.platformApiService.proxy(
'GET',
`/pipeline/pipeline_run/${normalizedRunId}/logs`,
user,
undefined,
query,
);
}
// ==================== JOBS - COLUMN EDITING ROUTES ====================
@Put('jobs/:jobId/input')
@ApiOperation({ summary: 'Update job input columns' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updateJobInput(
@Param('jobId') jobId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
const result = await this.platformApiService.proxy(
'PUT',
`/jobs/${normalizedJobId}/input`,
user,
body,
);
// Sync to DynamoDB if connector type is provided
const connectorType = body.connector as 'jdbc' | 'singer' | 's3' | undefined;
if (connectorType && this.VALID_CONNECTORS.includes(connectorType)) {
await this.syncJobInputToDynamoDB(jobId, body, user, connectorType);
}
return result;
}
@Patch('jobs/:jobId/input')
@ApiOperation({ summary: 'Partial update job input columns' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async patchJobInput(
@Param('jobId') jobId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
const result = await this.platformApiService.proxy(
'PATCH',
`/jobs/${normalizedJobId}/input`,
user,
body,
);
// Sync to DynamoDB if connector type is provided
const connectorType = body.connector as 'jdbc' | 'singer' | 's3' | undefined;
if (connectorType && this.VALID_CONNECTORS.includes(connectorType)) {
await this.syncJobInputToDynamoDB(jobId, body, user, connectorType);
}
return result;
}
@Put('jobs/:jobId/memory')
@ApiOperation({ summary: 'Update job memory configuration' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updateJobMemory(
@Param('jobId') jobId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Validate memory limit
if (body.amount) {
this.validateMemory(body.amount);
}
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
return this.platformApiService.proxy(
'PUT',
`/jobs/${normalizedJobId}/memory`,
user,
body,
);
}
@Post('jobs/:jobId/reset-state')
@ApiOperation({ summary: 'Reset job state' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async resetJobState(
@Param('jobId') jobId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
return this.platformApiService.proxy(
'POST',
`/jobs/${normalizedJobId}/reset-state`,
user,
body,
);
}
// ==================== JOBS - JDBC SYNC MODE ROUTES ====================
@Get('jobs/jdbc/:jobId')
@ApiOperation({ summary: 'Get JDBC job details' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getJdbcJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
return this.platformApiService.proxy('GET', `/jobs/jdbc/${normalizedJobId}`, user);
}
@Post('jobs/jdbc/:jobId/sync-mode')
@ApiOperation({ summary: 'Update JDBC job sync mode' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updateJdbcSyncMode(
@Param('jobId') jobId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
const result = await this.platformApiService.proxy(
'POST',
`/jobs/jdbc/${normalizedJobId}/sync-mode`,
user,
body,
);
// Sync to DynamoDB (pass raw jobId for pipeline extraction)
await this.syncJdbcSyncModeToDynamoDB(jobId, body, user);
return result;
}
@Get('jobs/jdbc/configs/allowed_datatypes')
@ApiOperation({ summary: 'Get allowed datatypes for JDBC' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getJdbcAllowedDatatypes(@User() user: RequestUser) {
return this.platformApiService.proxy(
'GET',
'/jobs/jdbc/configs/allowed_datatypes',
user,
);
}
// ==================== JOBS - SINGER REPLICATION ROUTES ====================
@Get('jobs/singer/:jobId')
@ApiOperation({ summary: 'Get Singer job details' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getSingerJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
return this.platformApiService.proxy('GET', `/jobs/singer/${normalizedJobId}`, user);
}
@Post('jobs/singer/:jobId/sync-mode')
@ApiOperation({ summary: 'Update Singer job sync mode' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
async updateSingerSyncMode(
@Param('jobId') jobId: string,
@Body() body: any,
@User() user: RequestUser,
) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
const result = await this.platformApiService.proxy(
'POST',
`/jobs/singer/${normalizedJobId}/sync-mode`,
user,
body,
);
// Sync to DynamoDB (pass raw jobId for pipeline extraction)
await this.syncSingerSyncModeToDynamoDB(jobId, body, user);
return result;
}
// ==================== JOBS - S3 ROUTES ====================
@Get('jobs/s3/:jobId')
@ApiOperation({ summary: 'Get S3 job details' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getS3Job(@Param('jobId') jobId: string, @User() user: RequestUser) {
// Normalize job ID for Platform API (replace - with _)
const normalizedJobId = this.normalizeJobId(jobId);
return this.platformApiService.proxy('GET', `/jobs/s3/${normalizedJobId}`, user);
}
// ==================== HEALTH ROUTE ====================
@Get('health')
@ApiOperation({ summary: 'Platform API health check' })
@Authenticated()
async healthCheck(@User() user: RequestUser) {
return this.platformApiService.proxy('GET', '/health', user);
}
}
@@ -0,0 +1,17 @@
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';
import { CustomersModule } from '../customers/customers.module';
@Module({
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule],
controllers: [PlatformApiController],
providers: [PlatformApiService, DadosferaLogger],
exports: [PlatformApiService],
})
export class PlatformApiModule {}
@@ -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);
}
}
}
+22 -16
View File
@@ -229,27 +229,33 @@ export class RolesService {
const [roleTreated] = this.getRolesPermissionsName([role.role]);
return { role: roleTreated };
}
getRolesPermissionsName(roles: GetRolesPermissionsName[]): RoleDto[] {
const newRoles: RoleDto[] = [];
for (const role of roles) {
const allPermissions = this.permissionsService.getAllPermissions(
this.language,
);
const newPermissions = role.permissions.map((p) => {
const permission = allPermissions.find((per) => per.seqid === p.seqid);
return {
...p,
name: permission.name,
id: p.seqid,
};
});
const newRole: RoleDto = {
...role,
permissions: newPermissions,
isPublic: role.isPublic,
};
const newRole: RoleDto = this.formatRole(role);
newRoles.push(newRole);
}
return newRoles;
}
private formatRole(role: GetRolesPermissionsName): RoleDto {
const allPermissions = this.permissionsService.getAllPermissions(
this.language
);
const newPermissions = role.permissions.map((p) => {
const permission = allPermissions.find((per) => per.seqid === p.seqid);
return {
...p,
name: permission.name,
id: p.seqid,
};
});
return {
...role,
permissions: newPermissions,
isPublic: role.isPublic,
};;
}
}
+20
View File
@@ -124,4 +124,24 @@ export class ThemeController {
}
}
@Post('/:id/theme/reset')
@ApiOkResponse({ type: CustomerThemeResponse })
async resetTheme(@Param('id') id: string) {
this.logger.info('getCustomerTheme with id' + id);
try {
await this.themeService.resetTheme(id);
return { theme: null };
}catch (err) {
if (err.details === ErrorCodes.CUSTOMER.NOT_FOUND) {
this.logger.error('Error - getCustomerTheme - Expect CUSTOMER.NOT_FOUND');
throw new HttpException(err.details, HttpStatus.NOT_FOUND);
} else {
this.logger.error('Error - getCustomerTheme Unknown Error:' + err?.message);
return { theme: null };
};
}
}
}
+12
View File
@@ -44,6 +44,18 @@ export class ThemeService implements OnModuleInit {
);
}
async resetTheme(id: string) {
const { theme } = await firstValueFrom(
this.themeService.ResetCustomerTheme({
id
}),
);
return {
theme
}
}
async createThemeByCustomer(id: string, theme: CustomerThemeRequest & Files) {
if (!id) {
this.logger.error('Error - saveCustomertheme - not found id:' + id);
+12
View File
@@ -38,6 +38,12 @@ export class User {
department?: string;
@ApiProperty()
hierarchy?: string;
@ApiProperty()
bio?: string;
@ApiProperty()
companyName?: string;
@ApiProperty()
companySite?: string;
@ApiPropertyOptional()
customer?: Customer;
@ApiProperty()
@@ -109,6 +115,12 @@ export class UpdateUserReq {
@ApiPropertyOptional()
hierarchy?: string;
@ApiPropertyOptional()
bio?: string;
@ApiPropertyOptional()
companySite?: string;
@ApiPropertyOptional()
companyName?: string;
@ApiPropertyOptional()
roleNames?: string[];
}
+12 -1
View File
@@ -266,7 +266,6 @@ export class UsersController {
}
@Patch(':id')
@RequireAllPermissions(PERMISSIONS_GROUPS.USERS.permissions.ADMIN)
@ApiOkResponse({ type: UpdateUserRes })
async updateUser(
@User() user: RequestUser,
@@ -274,6 +273,18 @@ export class UsersController {
@Param('id') id: string,
@Language() language: LanguageEnum,
) {
const isSameUser = user.user_id === id;
const isSuperAdmin = user.permissions.includes(PERMISSIONS_GROUPS.USERS.permissions.ADMIN.seqid)
if (!isSameUser && !isSuperAdmin) {
throw new ErrorBuilder(ErrorCodes.AUTH.FORBIDDEN);
}
if (isSameUser && !isSuperAdmin && body.roleNames) {
// Prevent users from updating their own roles
delete body.roleNames;
}
this.logger.info('updateUser', { user });
this.userService.setLanguage(language);
return await this.userService.updateUser(body, id, user.customer_id);
+10 -7
View File
@@ -148,20 +148,23 @@ export class UsersService implements OnModuleInit {
}
async updateUser(req: UpdateUserReq, id: string, customerId: string) {
const { department, hierarchy, jobTitle, name, roleNames, email } = req;
if (roleNames) {
const { roleNames, ...updateUserDTO } = req;
if (roleNames && roleNames.length > 0) {
await this.setRoles({ roleNames, userId: id }, customerId);
}
const { user } = await lastValueFrom(
this.usersClientService.UserUpdate({
name,
department: updateUserDTO.department,
email: updateUserDTO.email,
hierarchy: updateUserDTO.hierarchy,
jobTitle: updateUserDTO.jobTitle,
name: updateUserDTO.name,
bio: updateUserDTO.bio,
companyName: updateUserDTO.companyName,
companySite: updateUserDTO.companySite,
customerId,
id,
department,
hierarchy,
jobTitle,
email,
metabaseUserId: undefined,
}),
);
+4
View File
@@ -0,0 +1,4 @@
export const DYNAMODB_CONFIG = {
region: () => process.env.AWS_REGION || 'us-east-1',
inputsTable: () => process.env.INPUTS_DB || 'dadosfera-inputs-prd',
};
+9
View File
@@ -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 {}
+234
View File
@@ -0,0 +1,234 @@
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 ReferenceColumn {
name: string;
type: string;
}
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?: ReferenceColumn;
}>;
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?: ReferenceColumn;
}>;
},
): 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 });
}
/**
* Update a specific table entry in the input document.
* Fetches the current document, updates the matching table, and saves.
*/
async updateInputTable(
clientId: string,
inputId: string,
tableName: string,
changes: {
type?: string;
columns?: string[];
reference_column?: ReferenceColumn | null;
},
): Promise<void> {
const dynamoTableName = DYNAMODB_CONFIG.inputsTable();
this.logger.info('DynamoDB: Updating input table', {
inputId,
tableName,
changes: Object.keys(changes),
});
// Get current document
const current = await this.findInput(clientId, inputId);
if (!current) {
this.logger.warn('DynamoDB: Input not found for update', { inputId });
return;
}
// Find and update the matching table
const tables = current.tables || [];
const tableIndex = tables.findIndex((t) => t.name === tableName);
if (tableIndex === -1) {
this.logger.warn('DynamoDB: Table not found in input', {
inputId,
tableName,
});
return;
}
// Merge changes into the table entry
const updatedTable = { ...tables[tableIndex] };
if ('type' in changes) updatedTable.type = changes.type;
if ('columns' in changes) updatedTable.columns = changes.columns;
if ('reference_column' in changes) {
if (changes.reference_column === null) {
delete updatedTable.reference_column;
} else {
updatedTable.reference_column = changes.reference_column;
}
}
tables[tableIndex] = updatedTable;
// Save updated document
const putCommand = new PutCommand({
TableName: dynamoTableName,
Item: {
...current,
tables,
updated_at: new Date().toISOString(),
},
});
try {
await this.documentClient.send(putCommand);
this.logger.info('DynamoDB: Input table updated successfully', {
inputId,
tableName,
});
} catch (error) {
this.logger.error('DynamoDB: Failed to update input table', {
inputId,
tableName,
error: error.message,
});
throw error;
}
}
}
+3
View File
@@ -0,0 +1,3 @@
export * from './dynamodb.service';
export * from './dynamodb.module';
export * from './dynamodb.config';
@@ -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
};
@@ -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 {}
@@ -0,0 +1,382 @@
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();
this.logger.info('Elasticsearch: Updating pipeline', {
index,
pipelineId,
fields: Object.keys(changes),
});
try {
// Fetch current document
const currentDoc = await this.client.get(`${index}/_doc/${pipelineId}`);
const current = currentDoc.data._source;
// Build updated document, preserving existing values
const updated: Record<string, any> = {
...current,
updated_at: now,
};
if ('name' in changes) {
updated.name = this.formatMultiLang(changes.name);
}
if ('description' in changes) {
updated.description = this.formatMultiLang(changes.description);
}
if ('cron' in changes) {
updated.config = {
...current.config,
cron: changes.cron,
};
}
if ('status' in changes) {
updated.status = changes.status;
updated.last_status_updated = now;
}
if ('tags' in changes) {
updated.tags = changes.tags;
}
const response = await this.client.post(
`${index}/_doc/${pipelineId}`,
updated,
{ 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 getPipeline(
customerName: string,
pipelineId: string,
): Promise<PipelineDocument | null> {
const index = this.getIndex(customerName);
this.logger.info('Elasticsearch: Getting pipeline', {
index,
pipelineId,
});
try {
const response = await this.client.get(`${index}/_doc/${pipelineId}`);
return response.data._source as PipelineDocument;
} catch (error) {
if (error instanceof AxiosError && error.response?.status === 404) {
this.logger.warn('Elasticsearch: Pipeline not found', {
pipelineId,
index,
});
return null;
}
this.handleError('getPipeline', 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,
});
}
}
}
+3
View File
@@ -0,0 +1,3 @@
export * from './elasticsearch.module';
export * from './elasticsearch.service';
export * from './elasticsearch.config';
+1 -1
View File
@@ -14,7 +14,7 @@ import { redisStore } from 'cache-manager-ioredis-yet';
keyPrefix: 'maestro:sso',
}
if (process.env.ENV !== 'local') {
if (process.env.REDIS_TLS === 'true') {
baseRedisConfig['tls'] = {
servername: process.env.REDIS_HOST,
}
+90
View File
@@ -0,0 +1,90 @@
import CronParser from 'cron-parser';
export enum ScheduleLimits {
MINUTE = 'minute',
HOUR = 'hour',
DAY = 'day',
UNLIMITED = 'unlimited',
}
const SECONDS_IN_MINUTE = 60;
const SECONDS_IN_HOUR = 3600;
const SECONDS_IN_DAY = 86400;
/**
* Airflow preset schedules mapped to cron expressions.
* @once is special - it means run only once (no recurring schedule).
*/
const AIRFLOW_PRESETS: Record<string, string | null> = {
'@once': null, // No recurring schedule - always valid
'@hourly': '0 * * * *', // Every hour
'@daily': '0 0 * * *', // Every day at midnight
'@weekly': '0 0 * * 0', // Every week on Sunday
'@monthly': '0 0 1 * *', // First day of every month
'@yearly': '0 0 1 1 *', // First day of every year
'@annually': '0 0 1 1 *', // Same as @yearly
};
/**
* Convert Airflow preset to cron expression.
* Returns null for @once (no recurring schedule).
* Returns original string if not an Airflow preset.
*/
export function convertAirflowPresetToCron(schedule: string): string | null {
const preset = AIRFLOW_PRESETS[schedule.toLowerCase()];
if (preset !== undefined) {
return preset;
}
return schedule;
}
export function getMinimumIntervalSeconds(scheduleLimit: string): number {
switch (scheduleLimit) {
case ScheduleLimits.MINUTE:
return SECONDS_IN_MINUTE;
case ScheduleLimits.HOUR:
return SECONDS_IN_HOUR;
case ScheduleLimits.DAY:
return SECONDS_IN_DAY;
case ScheduleLimits.UNLIMITED:
default:
return 0;
}
}
export function getCronIntervalSeconds(cron: string): number {
const interval = CronParser.parseExpression(cron);
const nextDate = interval.next().toDate();
const afterNextDate = interval.next().toDate();
return Math.floor((afterNextDate.getTime() - nextDate.getTime()) / 1000);
}
export function validateCronAgainstScheduleLimit(
cron: string,
scheduleLimit: string,
): { valid: boolean; message?: string } {
if (!cron) return { valid: true };
// Convert Airflow presets to cron expressions
const cronExpression = convertAirflowPresetToCron(cron);
// @once returns null - no recurring schedule, always valid
if (cronExpression === null) {
return { valid: true };
}
try {
const cronInterval = getCronIntervalSeconds(cronExpression);
const minInterval = getMinimumIntervalSeconds(scheduleLimit);
if (cronInterval < minInterval) {
return {
valid: false,
message: `Schedule interval (${cronInterval}s) is below customer limit (${scheduleLimit}: ${minInterval}s minimum)`,
};
}
return { valid: true };
} catch (error) {
return { valid: false, message: `Invalid cron expression: ${error.message}` };
}
}