MERGE: conflicts

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

@@ -834,6 +834,24 @@ export class PlatformApiController {
     );
   }
 
+  @Post('pipeline/:pipelineId/pipeline_run/:runId/cancel')
+  @ApiOperation({ summary: 'Cancel a running pipeline run' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async cancelPipelineRun(
+    @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(
+      'POST',
+      `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`,
+      user,
+    );
+  }
+
   // ==================== JOBS - COLUMN EDITING ROUTES ====================
 
   @Put('jobs/:jobId/input')

src/services/elasticsearch/elasticsearch.service.ts

@@ -50,6 +50,9 @@ interface PipelineDocument {
   type: string;
   in_use: number;
   keywords: MultiLangArray;
+  // Pipeline run tracking
+  last_run_status?: string;
+  last_run_canceled_at?: string;
 }
 
 @Injectable()
@@ -358,6 +361,119 @@ export class ElasticsearchService {
     }
   }
 
+  async updateLastRunStatus(
+    customerName: string,
+    pipelineId: string,
+    lastRunStatus: string,
+  ): Promise<any> {
+    const index = this.getIndex(customerName);
+    const now = new Date().toISOString();
+
+    this.logger.info('Elasticsearch: Updating last run status', {
+      index,
+      pipelineId,
+      lastRunStatus,
+    });
+
+    try {
+      const response = await this.client.post(
+        `/${index}/_update/${pipelineId}`,
+        {
+          doc: {
+            last_run_status: lastRunStatus,
+            ...(lastRunStatus === 'CANCELED' ? { last_run_canceled_at: now } : {}),
+          },
+        },
+        { params: { refresh: 'wait_for' } },
+      );
+
+      this.logger.info('Elasticsearch: Last run status updated', {
+        pipelineId,
+        result: response.data.result,
+      });
+
+      return response.data;
+    } catch (error) {
+      this.handleError('updateLastRunStatus', error, { pipelineId, index });
+      throw error;
+    }
+  }
+
+  private getDataAssetIndex(customerName: string): string {
+    return `${customerName}_data_assets_catalog`;
+  }
+
+  async findDataAssetByTable(
+    customerName: string,
+    tableName: string,
+    tableSchema: string,
+  ): Promise<{ id: string; nimbus_id: number | null; [key: string]: any } | null> {
+    const index = this.getDataAssetIndex(customerName);
+
+    this.logger.info('Elasticsearch: Searching data asset', {
+      index,
+      tableName,
+      tableSchema,
+    });
+
+    try {
+      const response = await this.client.post(`/${index}/_search`, {
+        query: {
+          bool: {
+            must: [
+              { term: { 'table_name.keyword': tableName.toUpperCase() } },
+              { term: { 'table_schema.keyword': tableSchema.toUpperCase() } },
+            ],
+          },
+        },
+        size: 1,
+      });
+
+      const hits = response.data.hits?.hits || [];
+      if (hits.length === 0) {
+        this.logger.warn('Elasticsearch: Data asset not found', { tableName, tableSchema, index });
+        return null;
+      }
+
+      return { ...hits[0]._source, _es_id: hits[0]._id };
+    } catch (error) {
+      this.handleError('findDataAssetByTable', error, { tableName, tableSchema, index });
+      throw error;
+    }
+  }
+
+  async updateDataAsset(
+    customerName: string,
+    assetId: string,
+    updates: Record<string, any>,
+  ): Promise<any> {
+    const index = this.getDataAssetIndex(customerName);
+
+    this.logger.info('Elasticsearch: Updating data asset', {
+      index,
+      assetId,
+      fields: Object.keys(updates),
+    });
+
+    try {
+      const response = await this.client.post(
+        `/${index}/_update/${assetId}`,
+        { doc: updates },
+        { params: { refresh: 'wait_for' } },
+      );
+
+      this.logger.info('Elasticsearch: Data asset updated', {
+        assetId,
+        result: response.data.result,
+      });
+
+      return response.data;
+    } catch (error) {
+      this.handleError('updateDataAsset', error, { assetId, index });
+      throw error;
+    }
+  }
+
   private handleError(
     operation: string,
     error: any,