FEAT: add delete job endpoint and update related services; update protospack-v2 version to 3.40.0-beta.3

package.json

@@ -35,7 +35,7 @@
     "@aws-sdk/signature-v4": "^3.370.0",
     "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
     "@dadosfera/protospack": "2.5.3",
-    "@dadosfera/protospack-v2": "3.40.0-beta.2",
+    "@dadosfera/protospack-v2": "^3.40.0-beta.3",
     "@grpc/grpc-js": "^1.9.3",
     "@grpc/proto-loader": "^0.7.9",
     "@nestjs/cli": "^9.5.0",

src/modules/pipelinesV2/pipelines.controller.ts

@@ -224,11 +224,12 @@ export class PipelinesController {
 
     const result = await this.pipelinesClientService
       .findOne({ id }, metadata)
-      .then((res) => {
+      .then(async (res) => {
         //{pipeline:{tables: {tables: [], input_id: ''}}}
-        let tables = JSON.parse(res.pipeline.config.tables);
-        const input_id = tables?.input_id;
-        if (tables?.tables) tables = tables.tables;
+        const parsed = JSON.parse(res.pipeline.config.tables);
+        const input_id = parsed?.input_id;
+        const tables = parsed?.tables ?? parsed;
+
         Object.assign(res.pipeline, {
           transformations: res.pipeline.transformations
             ? JSON.parse(res.pipeline.transformations)

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

@@ -962,6 +962,80 @@ export class PlatformApiController {
     );
   }
 
+  @Delete('jobs/:jobId')
+  @ApiOperation({ summary: 'Delete a job and sync pipeline data' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
+  async deleteJob(
+    @Param('jobId') jobId: string,
+    @User() user: RequestUser,
+  ) {
+    // Get job details before deletion to extract table_name and connector type
+    const normalizedJobId = this.normalizeJobId(jobId);
+    let tableName: string | undefined;
+
+    this.logger.info('[deleteJob] Start', { jobId });
+
+    try {
+      const job = await this.platformApiService.proxy(
+        'GET',
+        `/jobs/${normalizedJobId}`,
+        user,
+      );
+      tableName = job?.source_config?.table_name;
+      this.logger.info('[deleteJob] Job fetched', { jobId, tableName });
+    } catch (err) {
+      this.logger.warn('[deleteJob] Could not fetch job before deletion', { jobId, error: err.message });
+    }
+
+    const result = await this.platformApiService.proxy(
+      'DELETE',
+      `/jobs/${normalizedJobId}`,
+      user,
+    );
+    this.logger.info('[deleteJob] Job deleted from platform-api', { jobId });
+
+    // Mark the table as deleted via in-factory (which has the correct DynamoDB credentials).
+    // config.tables in ES stores the DynamoDB input ID (legacy naming).
+    if (tableName) {
+      try {
+        const esPipelineId = this.extractPipelineIdFromJobId(jobId);
+        const esPipeline = await this.elasticsearchService.getPipeline(
+          user.customer_name,
+          esPipelineId,
+        );
+        // config.tables holds the DynamoDB input ID (see syncJobInputToDynamoDB for context)
+        const dynamoInputId = esPipeline?.config?.tables;
+
+        if (dynamoInputId) {
+          const info = { customer_id: user.customer_id, user_id: user.user_id, customer: user.customer_name };
+
+          // Read current input to get full tables array (adjustInputPayload flattens the response)
+          const { input } = await this.inputsService.findOne({ id: dynamoInputId, info });
+          const currentTables: any[] = input?.tables ?? [];
+
+          const updatedTables = currentTables.map((t: any) =>
+            t.name === tableName ? { ...t, status: 'deleted' } : t,
+          );
+
+          await this.inputsService.update(dynamoInputId, { tables: updatedTables }, info);
+          this.logger.info('[deleteJob] Marked table as deleted via in-factory', { jobId, tableName });
+        } else {
+          this.logger.warn('[deleteJob] No input ID in ES pipeline, skipping DynamoDB update', { jobId, tableName });
+        }
+      } catch (error) {
+        this.logger.error('[deleteJob] Failed to mark table as deleted via in-factory', {
+          jobId,
+          tableName,
+          error: error.message,
+        });
+      }
+    } else {
+      this.logger.warn('[deleteJob] tableName not resolved, skipping DynamoDB update', { jobId });
+    }
+
+    return result;
+  }
+
   // ==================== JOBS - JDBC SYNC MODE ROUTES ====================
 
   @Get('jobs/jdbc/:jobId')

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

@@ -8,9 +8,10 @@ import { ElasticsearchModule } from '../../services/elasticsearch';
 import { DynamoDBModule } from '../../services/dynamodb';
 import { CustomersModule } from '../customers/customers.module';
 import { CatalogModule } from '../catalog/catalog.module';
+import { PipelinesV2Module } from '../pipelinesV2/pipelines.module';
 
 @Module({
-  imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule],
+  imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, PipelinesV2Module],
   controllers: [PlatformApiController],
   providers: [PlatformApiService, DadosferaLogger],
   exports: [PlatformApiService],

src/services/dynamodb/dynamodb.service.ts

@@ -30,6 +30,7 @@ export interface InputDocument {
     type: string;
     columns?: string[];
     reference_column?: ReferenceColumn;
+    status?: string;
   }>;
   credentials?: Record<string, any>;
 }
@@ -125,9 +126,13 @@ export class DynamoDBService {
       },
     });
 
-    const { Item } = await this.documentClient.send(getCommand);
-
-    return Item as InputDocument | null;
+    try {
+      const { Item } = await this.documentClient.send(getCommand);
+      return Item as InputDocument | null;
+    } catch (error) {
+      this.logger.error('DynamoDB: findInput failed', { inputId, clientId, error: error.message });
+      throw error;
+    }
   }
 
   async deleteInput(clientId: string, inputId: string): Promise<void> {
@@ -163,6 +168,7 @@ export class DynamoDBService {
       type?: string;
       columns?: string[];
       reference_column?: ReferenceColumn | null;
+      status?: string;
     },
   ): Promise<void> {
     const dynamoTableName = DYNAMODB_CONFIG.inputsTable();
@@ -203,6 +209,7 @@ export class DynamoDBService {
         updatedTable.reference_column = changes.reference_column;
       }
     }
+    if ('status' in changes) updatedTable.status = changes.status;
 
     tables[tableIndex] = updatedTable;
 
@@ -231,4 +238,53 @@ export class DynamoDBService {
       throw error;
     }
   }
+
+  /**
+   * Remove a specific table entry from the input document.
+   * Fetches the current document, removes the matching table, and saves.
+   */
+  async removeInputTable(
+    clientId: string,
+    inputId: string,
+    tableName: string,
+  ): Promise<void> {
+    const dynamoTableName = DYNAMODB_CONFIG.inputsTable();
+
+    this.logger.info('DynamoDB: Removing input table', {
+      inputId,
+      tableName,
+    });
+
+    const current = await this.findInput(clientId, inputId);
+    if (!current) {
+      this.logger.warn('DynamoDB: Input not found for table removal', { inputId });
+      return;
+    }
+
+    const tables = (current.tables || []).filter((t) => t.name !== tableName);
+
+    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 removed successfully', {
+        inputId,
+        tableName,
+      });
+    } catch (error) {
+      this.logger.error('DynamoDB: Failed to remove input table', {
+        inputId,
+        tableName,
+        error: error.message,
+      });
+      throw error;
+    }
+  }
 }