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>

README.md

@@ -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.

docsfera.json

@@ -4483,14 +4483,6 @@
               "schema": {
                 "type": "string"
               }
-            },
-            {
-              "name": "asset_type",
-              "required": true,
-              "in": "query",
-              "schema": {
-                "type": "string"
-              }
             }
           ],
           "responses": {

src/modules/catalog/catalog.service.ts

@@ -41,6 +41,7 @@ import {
 import { TypeParser } from 'src/utils/FileParser/parser-types';
 import { ParserBuilder } from 'src/utils/FileParser/parser.builder';
 
+
 class CatalogService implements OnModuleInit {
   catalogReadService: ReadService.CatalogReadServices;
   catalogWriteService: WriteService.CatalogWriteServices;
@@ -57,6 +58,7 @@ class CatalogService implements OnModuleInit {
     this.logger = dadosferaLogger.logger;
   }
 
+
   onModuleInit() {
     this.catalogReadService =
       this.grpcClient.getService<ReadService.CatalogReadServices>(
@@ -72,65 +74,81 @@ class CatalogService implements OnModuleInit {
       );
   }
 
+
   _getNimbusUrl(body) {
     this.logger.debug(`Body: ${JSON.stringify(body)}`);
     const customer = body.info.customer.toLowerCase();
 
+
     if (process.env.ENV === 'prd') {
       return `https://nimbus-${customer}.dadosfera.ai`;
     }
 
+
     return `https://nimbus-${customer}.${process.env.ENV.replace(
       'local',
       'stg',
     )}.dadosfera.ai`;
   }
 
+
   async getPiiReporter(metadata: Metadata, type: TypeParser) {
-    this.logger.info('getPiiReporter: ' + type);
+    this.logger.info('getPiiReporter: ' + type)
     try {
-      const { data } = await lastValueFrom(
+      const {
-        this.catalogWriteService.GetPiiReporter({}, metadata),
+        data
-      );
+      } = await lastValueFrom(
-      this.logger.info('Finish grpc call');
+        this.catalogWriteService.GetPiiReporter({}, metadata)
+      )
+      this.logger.info("Finish grpc call")
+
 
       const parser = ParserBuilder.build<PiiMetadata>(type);
 
-      this.logger.info('parser file to: ' + type);
+
-      const file = await parser.parse(data);
+      this.logger.info('parser file to: ' + type)
-      this.logger.info('finish parser');
+      const file = await parser.parse(data)
+      this.logger.info('finish parser')
       const mimeTypes: Record<TypeParser, string> = {
-        csv: 'text/csv',
+        'csv': 'text/csv',
-        html: 'text/html',
+        'html': 'text/html',
-        pdf: 'application/pdf',
+        'pdf': 'application/pdf'
-      };
+      }
+
 
       const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
       const filename = `relatorio-pii-${timestamp}.${type}`;
 
+
       return {
         file,
         filename: filename,
-        type: mimeTypes[type],
+        type: mimeTypes[type]
-      };
+      }
     } catch (error) {
       this.logger.error(error.message);
       throw error;
     }
+
+
   }
 
+
   async createDataAsset(data: Messages.CreateDataAssetRequest, metadata) {
     this.logger.info('CatalogService - Manage Data assets permissions');
     if (!data.embed) data.embed = undefined;
 
+
     return lastValueFrom(
       this.catalogWriteService.CreateDataAsset(data, metadata),
     );
   }
 
+
   async managePermissions(data: Messages.ManagePermissionRequest, metadata) {
     this.logger.info('CatalogService - Manage Data assets permissions');
 
+
     return lastValueFrom(
       this.catalogWriteService.ManagePermission(data, metadata),
     ).catch((err) => {
@@ -141,9 +159,11 @@ class CatalogService implements OnModuleInit {
     });
   }
 
+
   async revokePermissions(data: Messages.RevokePermissionRequest, metadata) {
     this.logger.info('CatalogService - Manage Data assets permissions');
 
+
     return lastValueFrom(
       this.catalogWriteService.RevokePermission(data, metadata),
     ).catch((err) => {
@@ -154,9 +174,11 @@ class CatalogService implements OnModuleInit {
     });
   }
 
+
   async commentOnDataAsset(data: Messages.MakeACommentRequest, metadata) {
     this.logger.info('CatalogService - Manage Data assets permissions');
 
+
     return lastValueFrom(
       this.catalogWriteService.MakeAComment(data, metadata),
     ).catch((err) => {
@@ -167,9 +189,11 @@ class CatalogService implements OnModuleInit {
     });
   }
 
+
   async deleteComment(data: Messages.UpdateACommentRequest, metadata) {
     this.logger.info('CatalogService - Manage Data assets permissions');
 
+
     return lastValueFrom(
       this.catalogWriteService.UpdateAComment(data, metadata),
     ).catch((err) => {
@@ -180,9 +204,11 @@ class CatalogService implements OnModuleInit {
     });
   }
 
+
   async deleteDataAsset(data: Messages.DeleteDataAssetRequest, metadata) {
     this.logger.info('CatalogService - Manage Data assets permissions');
 
+
     return lastValueFrom(
       this.catalogWriteService.DeleteDataAsset(data, metadata),
     ).catch((err) => {
@@ -193,114 +219,30 @@ class CatalogService implements OnModuleInit {
     });
   }
 
+
   async getUserRolesIds(userId: string) {
     const result = await this.userService.findOneById(userId).catch(() => null);
 
+
     if (result) {
       return result.user.roles.map((role) => role.id);
     }
 
+
     return [];
   }
 
+
   async searchDataAssets(
     query: Record<string, any>,
     metadata: Metadata,
     customer_id: string,
   ) {
-    this.logger.info('CatalogService - searchDataAssets', { query });
+    this.logger.info('CatalogService - searchDataAssets');
+
 
     const { search, page, size, sort_by, order, ...filters } = query;
 
-    this.logger.debug('Extracted filters:', { filters });
-
-    console.log('MAESTRO VAI CHAMAR PI-FACTORY COM (ANTES AJUSTE):', {
-      search,
-      page,
-      size,
-      sort_by,
-      order,
-      filters,
-    });
-
-    if (
-      filters.manually !== undefined &&
-      filters.manually !== null &&
-      filters.manually !== ''
-    ) {
-      filters.manually = Number(filters.manually); // 1 ou 0
-    } else {
-      delete filters.manually;
-    }
-
-    console.log('MAESTRO VAI CHAMAR PI-FACTORY COM (DEPOIS AJUSTE):', {
-      search,
-      page,
-      size,
-      sort_by,
-      order,
-      filters,
-    });
-
-    if (filters.owner) {
-      const { users: customer_users } =
-        await this.userService.findAllUsersByCustomerId(customer_id);
-
-      this.logger.info('Available users in database count:', {
-        count: customer_users.length,
-      });
-      this.logger.info('First 5 users:', {
-        users: customer_users
-          .slice(0, 5)
-          .map((u) => ({ id: u.id, email: u.email, name: u.name })),
-      });
-
-      const ownerValues = Array.isArray(filters.owner)
-        ? filters.owner
-        : typeof filters.owner === 'string' && filters.owner.includes(',')
-        ? filters.owner.split(',').map((o: string) => o.trim())
-        : [filters.owner];
-
-      this.logger.info('Owner values to convert:', {
-        ownerValues,
-        ownerFiltersOriginal: filters.owner,
-      });
-
-      const ownerIds = ownerValues
-        .map((ownerValue: string) => {
-          const normalizedOwner = ownerValue.replace(/\s/g, '+');
-          const user = customer_users.find((u) => {
-            const isIdMatch = u.id === ownerValue;
-            const isEmailMatch =
-              u.email === ownerValue || u.email === normalizedOwner;
-            const isNameMatch =
-              u.name === ownerValue || u.name === normalizedOwner;
-            this.logger.info('Comparing:', {
-              userId: u.id,
-              userEmail: u.email,
-              userName: u.name,
-              filterValue: ownerValue,
-              normalizedFilter: normalizedOwner,
-              idMatch: isIdMatch,
-              emailMatch: isEmailMatch,
-              nameMatch: isNameMatch,
-            });
-            return isIdMatch || isEmailMatch || isNameMatch;
-          });
-          this.logger.info('Looking for owner result:', {
-            ownerValue,
-            found: !!user,
-            userId: user?.id,
-          });
-          return user?.id || ownerValue;
-        })
-        .filter((id: string) => id);
-
-      if (ownerIds.length > 0) {
-        filters.owner = ownerIds;
-      }
-    }
-
     const { data_assets, total } = await lastValueFrom(
       this.catalogReadService.GetAllDataAssets(
         {
@@ -315,18 +257,20 @@ class CatalogService implements OnModuleInit {
       ),
     );
 
-    console.log('MAESTRO RECEBEU RESPOSTA DO PI-FACTORY');
 
     const result = JSON.parse(data_assets);
 
+
     const response = await this.getAssetsUsersAndRoles(
       result.data_assets,
       customer_id,
     );
 
+
     return { data_assets: response, total };
   }
 
+
   async downloadAssets(
     query: Record<string, any>,
     metadata: Metadata,
@@ -334,27 +278,33 @@ class CatalogService implements OnModuleInit {
   ) {
     const data = await this.searchDataAssets(query, metadata, customer_id);
 
-    const formatData = data.data_assets.map((asset) => ({
+
+    const formatData = data.data_assets.map(asset => ({
       id: asset.id,
       display_name: asset.display_name,
       data_asset_type: asset.data_asset_type,
       created_at: asset.created_at,
-      tags: '[' + asset.tags.join(', ') + ']',
+      tags: '[' + asset.tags.join(', ') + ']'
-    }));
+    }))
+
 
     const parser = ParserBuilder.build<AssetReporter>('csv');
 
+
     const file = await parser.parse(formatData);
 
+
     const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
     const filename = `dadosfera_assets_${timestamp}.csv`;
 
+
     return {
       file,
-      filename,
+      filename
-    };
+    }
   }
 
+
   async getOneDataAsset(data: {
     id: string;
     customer_id: string;
@@ -375,9 +325,11 @@ class CatalogService implements OnModuleInit {
     };
     asset = await this.getAssetsUsersAndRoles([asset], customer_id);
 
+
     return { data_asset: asset[0] };
   }
 
+
   async getOneDataAssetByPipelineAndObject(data: {
     customer_id: string;
     pipeline: string;
@@ -399,9 +351,11 @@ class CatalogService implements OnModuleInit {
     };
     asset = await this.getAssetsUsersAndRoles([asset], customer_id);
 
+
     return { data_asset: asset[0] };
   }
 
+
   async updateOneDataAsset(data: {
     data_asset_id: string;
     customer_id: string;
@@ -410,6 +364,7 @@ class CatalogService implements OnModuleInit {
   }) {
     const { body, customer_id, data_asset_id, metadata } = data;
 
+
     const { data_asset } = await lastValueFrom(
       this.catalogWriteService.UpdateDataAsset(
         { id: data_asset_id, changes: JSON.stringify(body) },
@@ -424,6 +379,7 @@ class CatalogService implements OnModuleInit {
     };
     asset = await this.getAssetsUsersAndRoles([asset], customer_id);
 
+
     return { data_asset: asset[0] };
   }
 
@@ -431,11 +387,12 @@ class CatalogService implements OnModuleInit {
     const { documentation } = await lastValueFrom(
       this.catalogReadService.GetDatasetDoc({ id }, metadata),
     );
-
+ 
     const docs = JSON.parse(documentation);
     return docs;
   }
 
+
   async getDatasetPreview(id: string, metadata: Metadata) {
     const { preview } = await lastValueFrom(
       this.catalogReadService.GetDatasetPreview(
@@ -447,6 +404,7 @@ class CatalogService implements OnModuleInit {
     return result;
   }
 
+
   async getDatasetColumnsMetadata(id: string, metadata: Metadata) {
     const { columns_metadata } = await lastValueFrom(
       this.catalogReadService.GetDatasetColumnsMetadata(
@@ -462,7 +420,7 @@ class CatalogService implements OnModuleInit {
     if (body.asset_type === 'table' || body.asset_type === 'view') {
       return this.createDataDocsViaNimbus(body);
     }
-
+    
     return this.createDataDocsViaGrpc(body, metadata);
   }
 
@@ -476,6 +434,7 @@ class CatalogService implements OnModuleInit {
     return data;
   }
 
+
   private async createDataDocsViaGrpc(body: CreateDataDocsDTO, metadata: Metadata) {
     this.logger.info('Creating data docs via gRPC for other asset types');
     try {
@@ -488,8 +447,9 @@ class CatalogService implements OnModuleInit {
           metadata,
         ),
       );
-
+      
       return response;
+      
     } catch (error) {
       this.logger.error('Error creating data asset docs:', error);
       throw new HttpException(
@@ -502,6 +462,7 @@ class CatalogService implements OnModuleInit {
   async findAllTags(data, metadata) {
     this.logger.info('CatalogService - findAllCustomerTags');
 
+
     const response = await lastValueFrom(
       this.catalogReadService.GetCustomerTags(data, metadata),
     )
@@ -514,9 +475,9 @@ class CatalogService implements OnModuleInit {
         throw new Error(err);
       });
 
+
     return response;
   }
-
   async getAssetsUsersAndRoles(data_assets: Array<any>, customer_id: string) {
     const { users: customer_users } =
       await this.userService.findAllUsersByCustomerId(customer_id);
@@ -529,15 +490,17 @@ class CatalogService implements OnModuleInit {
         (u) => u.id === data_asset.owner,
       )?.email;
 
+
       const roles = [];
       const users = [];
-      const data_asset_roles = data_asset?.roles || [];
+      const data_asset_roles = data_asset?.roles || []
       for (const role_id of data_asset_roles) {
         const role = customer_roles.find((r) => r.id === role_id);
         if (role) roles.push({ id: role.id, name: role.name });
       }
 
-      const data_asset_users = data_asset?.users || [];
+
+      const data_asset_users = data_asset?.users || []
       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, email: user.email });
@@ -551,6 +514,7 @@ class CatalogService implements OnModuleInit {
     });
   }
 
+
   async triggerCatalog(data: TriggerCatalogReq, metadata: Metadata) {
     const { session } = await lastValueFrom(
       this.catalogWriteService.TriggerDatasetCataloging(data, metadata),
@@ -564,6 +528,7 @@ class CatalogService implements OnModuleInit {
     return res;
   }
 
+
   async addRlsRule(data: AddRlsRuleRequest, metadata: Metadata) {
     const res = await lastValueFrom(
       this.catalogWriteService.AddRlsRule(data, metadata),
@@ -571,6 +536,7 @@ class CatalogService implements OnModuleInit {
     return res;
   }
 
+
   async removeRlsRule(id: number, metadata: Metadata) {
     const res = await lastValueFrom(
       this.catalogWriteService.RemoveRlsRule({ id }, metadata),
@@ -578,12 +544,14 @@ class CatalogService implements OnModuleInit {
     return res;
   }
 
+
   async batchRemoveRlsRule(
     query: BatchRemoveRlsRulesRequest,
     metadata: Metadata,
   ) {
     const { id_rls, nimbus_dashboard_id } = query;
 
+
     if (id_rls && nimbus_dashboard_id) {
       throw new BadRequestException(
         "You can't delete using both parameters. Choose either 'id_rls' or 'nimbus_dashboard_id'",
@@ -604,6 +572,7 @@ class CatalogService implements OnModuleInit {
     return 'OK';
   }
 
+
   async getRlsRules(data: GetRlsRulesRequest, metadata: Metadata) {
     const res = await lastValueFrom(
       this.catalogReadService.GetRlsRules(data, metadata),
@@ -611,6 +580,7 @@ class CatalogService implements OnModuleInit {
     return res.rls_rules;
   }
 
+
   async getOneRlsRule(id: number, metadata: Metadata) {
     const res = await lastValueFrom(
       this.catalogReadService.GetOneRlsRule({ id }, metadata),
@@ -618,6 +588,7 @@ class CatalogService implements OnModuleInit {
     return res.rls_rule;
   }
 
+
   async getNimbusDashboards(
     data: GetNimbusDashboardsRequest,
     metadata: Metadata,
@@ -628,99 +599,88 @@ class CatalogService implements OnModuleInit {
     return res.dashboards;
   }
 
+
   async createTableMetadata(body: any): Promise<number> {
     const nimbusUrl = this._getNimbusUrl(body);
-    this.logger.info(`Nimbus URL: ${nimbusUrl}`, { ...body.logMetadata });
+    this.logger.info(`Nimbus URL: ${nimbusUrl}`, {...body.logMetadata});
+
 
     const endpoint = `${nimbusUrl}/api/catalog/table-metadata/`;
 
-    this.logger.info(
+
-      `Creating table metadata for table ${body.table_metadata.table_name}`,
+    this.logger.info(`Creating table metadata for table ${body.table_metadata.table_name}`, {...body.logMetadata});
-      { ...body.logMetadata },
+    this.logger.info(`Using endpoint: ${endpoint}`, {...body.logMetadata});
-    );
+    this.logger.debug(`Payload: ${JSON.stringify(body.table_metadata)}`, {...body.logMetadata});  
-    this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata });
+
-    this.logger.debug(`Payload: ${JSON.stringify(body.table_metadata)}`, {
-      ...body.logMetadata,
-    });
 
     try {
-      const { data, status } = await axios.post(endpoint, {
+      const { data, status } = await axios.post(endpoint, {...body.table_metadata});
-        ...body.table_metadata,
+
-      });
 
       this.logger.info(
         `Table metadata created successfully with status ${status} for table ${body.table_metadata.table_name}`,
-        { ...body.logMetadata },
+        {...body.logMetadata},
       );
       return data.id;
     } catch (error) {
       this.logger.error(
         `Failed to create table metadata for table ${body.table_metadata.table_name} failed with status ${
           error.response?.status
-        } because of ${JSON.stringify(error.response?.data) || error.message}`,
+        } because of ${JSON.stringify(error.response?.data) || error.message}`, {...body.logMetadata});
-        { ...body.logMetadata },
-      );
       throw new Error(error.response?.data?.message || error.message);
     }
   }
 
+
   async createColumnMetadata(body: any): Promise<number[]> {
     const nimbusUrl = this._getNimbusUrl(body);
     this.logger.info(`Nimbus URL: ${nimbusUrl}`, body.logMetadata);
     const endpoint = `${nimbusUrl}/api/catalog/column-metadata/`;
 
+
+ 
+
+
     try {
-      this.logger.info(
+      this.logger.info(`Creating column metadata for table ${body.column_metadata.table_name}`, {...body.logMetadata});
-        `Creating column metadata for table ${body.column_metadata.table_name}`,
+      this.logger.info(`Using endpoint: ${endpoint}`, {...body.logMetadata});
-        { ...body.logMetadata },
+      this.logger.debug(`Payload: ${JSON.stringify(body.column_metadata)}`, {...body.logMetadata});
-      );
+      const { data, status } = await axios.post(endpoint, body.column_metadata);
-      this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata });
+
-      this.logger.debug(
-        `Payload: ${JSON.stringify(body.column_metadata)}`,
-        { ...body.logMetadata },
-      );
-      const { data, status } = await axios.post(
-        endpoint,
-        body.column_metadata,
-      );
 
       this.logger.info(
         `Column metadata created successfully with status ${status} for table ${body.column_metadata.table_name}`,
-        { ...body.logMetadata },
+        {...body.logMetadata},
       );
       return data.map((column) => column.id);
     } catch (error) {
       this.logger.error(
         `Failed to create column metadata failed with status for table ${body.column_metadata.table_name} ${
           error.response?.status
-        } because of ${error.response?.data || error.message}`,
+        } because of ${error.response?.data || error.message}`, {...body.logMetadata});
-        { ...body.logMetadata },
-      );
       throw new Error(error.response?.data?.message || error.message);
     }
   }
 
+
   async createDataPreview(body: any): Promise<number> {
     const nimbusUrl = this._getNimbusUrl(body);
-    this.logger.info(`Nimbus URL: ${nimbusUrl}`, { ...body.logMetadata });
+    this.logger.info(`Nimbus URL: ${nimbusUrl}`, {...body.logMetadata});
     const endpoint = `${nimbusUrl}/api/catalog/data-preview/`;
 
-    this.logger.info(
+
-      `Creating data preview for table ${body.data_preview.table_name}`,
+    this.logger.info(`Creating data preview for table ${body.data_preview.table_name}`, {...body.logMetadata});
-      { ...body.logMetadata },
+    this.logger.info(`Using endpoint: ${endpoint}`, {...body.logMetadata});
-    );
+    this.logger.debug(`Payload: ${JSON.stringify(body.data_preview)}`, {...body.logMetadata});
-    this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata });
+
-    this.logger.debug(
-      `Payload: ${JSON.stringify(body.data_preview)}`,
-      { ...body.logMetadata },
-    );
 
     try {
       const { data, status } = await axios.post(endpoint, body.data_preview);
 
+
       this.logger.info(
         `Data preview created successfully with status ${status} for table ${body.data_preview.table_name}`,
-        { ...body.logMetadata },
+        {...body.logMetadata},
       );
       return data.id;
     } catch (error) {
@@ -728,15 +688,17 @@ class CatalogService implements OnModuleInit {
         `Failed to create data preview for table ${body.data_preview.table_name} failed with status ${
           error.response?.status
         } because of ${error.response?.data || error.message}`,
-        { ...body.logMetadata },
+        {...body.logMetadata},
       );
       throw new Error(error.response?.data?.message || error.message);
     }
   }
 
+
   async catalogDatasetItem(table_metadata_id: number, metadata: Metadata) {
     const customer_name_raw = metadata.get('customer_name');
 
+
     const customer_name = customer_name_raw?.[0]?.toString();
     if (!customer_name) {
       throw new BadRequestException('Customer name not found in metadata');
@@ -759,4 +721,5 @@ class CatalogService implements OnModuleInit {
   }
 }
 
+
 export { CatalogService };

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

@@ -22,7 +22,7 @@ import { User, RequestUser } from '../../decorators/user.decorator';
 import { PlatformApiService } from './platform-api.service';
 import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
 import { ElasticsearchService } from '../../services/elasticsearch';
-import { DynamoDBService } from '../../services/dynamodb';
+import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
 import { CustomersService } from '../customers/customers.service';
 import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
 
@@ -155,7 +155,7 @@ export class PlatformApiController {
    * Extract and transform tables from jobs for DynamoDB input.
    * Maps connector-specific fields to a common table format.
    *
-   * - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name (reference_column)
+   * - 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
    */
@@ -163,7 +163,7 @@ export class PlatformApiController {
     name: string;
     type: string;
     columns?: string[];
-    reference_column?: string;
+    reference_column?: ReferenceColumn;
   }> {
     if (!jobs || jobs.length === 0) return [];
 
@@ -171,7 +171,7 @@ export class PlatformApiController {
       name: string;
       type: string;
       columns?: string[];
-      reference_column?: string;
+      reference_column?: ReferenceColumn;
     }> = [];
 
     for (const job of jobs) {
@@ -179,12 +179,12 @@ export class PlatformApiController {
       if (!input) continue;
 
       if (connector === 'jdbc') {
-        // JDBC: table_name, load_type, column_include_list, incremental_column_name
+        // JDBC: table_name, load_type, column_include_list, incremental_column_name/type
         const table: {
           name: string;
           type: string;
           columns?: string[];
-          reference_column?: string;
+          reference_column?: ReferenceColumn;
         } = {
           name: input.table_name || '',
           type: input.load_type || 'full_load',
@@ -193,8 +193,11 @@ export class PlatformApiController {
           table.columns = input.column_include_list;
         }
         if (input.incremental_column_name) {
-          // reference_column is stored as a string (column name)
+          // reference_column is stored as an object with name and type
-          table.reference_column = input.incremental_column_name;
+          table.reference_column = {
+            name: input.incremental_column_name,
+            type: input.incremental_column_type || 'unknown',
+          };
         }
         tables.push(table);
       } else if (connector === 'singer' || connector === 's3') {
@@ -317,11 +320,11 @@ export class PlatformApiController {
       }
 
       // Build changes for DynamoDB table entry
-      // reference_column is stored as a string (column name), not an object
+      // reference_column is stored as an object with name and type
       const changes: {
         type?: string;
         columns?: string[];
-        reference_column?: string | null;
+        reference_column?: ReferenceColumn | null;
       } = {};
 
 
@@ -332,8 +335,15 @@ export class PlatformApiController {
         changes.columns = body.column_include_list;
       }
       if ('incremental_column_name' in body) {
-        // reference_column is just the column name as a string
+        // reference_column is stored as an object with name and type
-        changes.reference_column = body.incremental_column_name || null;
+        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
@@ -466,11 +476,13 @@ export class PlatformApiController {
         });
       }
 
+      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,
       });
@@ -490,6 +502,7 @@ export class PlatformApiController {
           cron: body.cron,
           tables: inputId,
           properties,
+          type: pipelineType,
         },
         connector,
       );
@@ -617,21 +630,39 @@ export class PlatformApiController {
   @ApiOperation({ summary: 'Execute a pipeline' })
   @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
   async executePipeline(@Body() body: any, @User() user: RequestUser) {
-    return this.platformApiService.proxy('POST', '/pipeline/execute', user, body);
+    // 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) {
-    return this.platformApiService.proxy('POST', '/pipeline/pause', user, body);
+    // 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) {
-    return this.platformApiService.proxy('POST', '/pipeline/unpause', user, body);
+    // 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')
@@ -691,18 +722,6 @@ export class PlatformApiController {
 
   // ==================== PIPELINE RUN ROUTES ====================
 
-  @Post('pipeline/pipeline_run')
-  @ApiOperation({ summary: 'Create a pipeline run' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
-  async createPipelineRun(@Body() body: any, @User() user: RequestUser) {
-    return this.platformApiService.proxy(
-      'POST',
-      '/pipeline/pipeline_run',
-      user,
-      body,
-    );
-  }
-
   @Get('pipeline/:pipelineId/pipeline_run')
   @ApiOperation({ summary: 'Get pipeline runs for a pipeline' })
   @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
@@ -711,9 +730,10 @@ export class PlatformApiController {
     @User() user: RequestUser,
     @Query() query: Record<string, string>,
   ) {
+    const normalizedId = this.normalizePipelineId(pipelineId);
     return this.platformApiService.proxy(
       'GET',
-      `/pipeline/${pipelineId}/pipeline_run`,
+      `/pipeline/${normalizedId}/pipeline_run`,
       user,
       undefined,
       query,
@@ -728,9 +748,11 @@ export class PlatformApiController {
     @Param('runId') runId: string,
     @User() user: RequestUser,
   ) {
+    const normalizedPipelineId = this.normalizePipelineId(pipelineId);
+    const normalizedRunId = this.normalizePipelineId(runId);
     return this.platformApiService.proxy(
       'GET',
-      `/pipeline/${pipelineId}/pipeline_run/${runId}`,
+      `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`,
       user,
     );
   }
@@ -743,9 +765,10 @@ export class PlatformApiController {
     @User() user: RequestUser,
     @Query() query: Record<string, string>,
   ) {
+    const normalizedRunId = this.normalizePipelineId(runId);
     return this.platformApiService.proxy(
       'GET',
-      `/pipeline/pipeline_run/${runId}/logs`,
+      `/pipeline/pipeline_run/${normalizedRunId}/logs`,
       user,
       undefined,
       query,

src/services/dynamodb/dynamodb.service.ts

@@ -11,6 +11,11 @@ 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;
@@ -24,7 +29,7 @@ export interface InputDocument {
     name: string;
     type: string;
     columns?: string[];
-    reference_column?: string;
+    reference_column?: ReferenceColumn;
   }>;
   credentials?: Record<string, any>;
 }
@@ -62,7 +67,7 @@ export class DynamoDBService {
         name: string;
         type: string;
         columns?: string[];
-        reference_column?: string;
+        reference_column?: ReferenceColumn;
       }>;
     },
   ): Promise<InputDocument> {
@@ -157,7 +162,7 @@ export class DynamoDBService {
     changes: {
       type?: string;
       columns?: string[];
-      reference_column?: string | null;
+      reference_column?: ReferenceColumn | null;
     },
   ): Promise<void> {
     const dynamoTableName = DYNAMODB_CONFIG.inputsTable();