[ResponseOps] Cases analytics index synchronization (#222820)

Closes [#221232](#221232)

**Merging into a feature branch**

## Summary

This PR adds:
- Synchronization task registration.
- Synchronization task scheduling.
- Testing

## How to test

A few things are needed.

1. Create a role that has index creation permission. Below is what works
for me; it might be overkill.
<img width="1030" alt="Screenshot 2025-04-25 at 11 30 45"
src="https://github.com/user-attachments/assets/cbb7b384-e438-43ec-bbca-a18e62243523"
/>

2. Create a user with the role created in step 1. Let's call the user
`index_creator`. **The password should be** `changeme`.
<img width="842" alt="Screenshot 2025-04-25 at 11 31 27"
src="https://github.com/user-attachments/assets/a9547c96-086e-43b4-a442-a18f558dcb96"
/>

6. Start Kibana with this user as the Elasticsearch user - `yarn start`
with the option `--elasticsearch.username=index_creator`.

7.  You can check the content of these indexes in the Dev Tools.
```
GET /.internal.cases/_search
GET /.internal.cases-comments/_search
GET /.internal.cases-attachments/_search
```

8. Create Cases, Comments, and Attachments.
9. After about 5 minutes, check again the content of the indexes using
the instructions in step 7.
10. Confirm the Cases, Comments, and Attachments created in step 8 can
be seen in the analytics indexes.

Author: adcoelho

x-pack/platform/plugins/shared/cases/server/cases_analytics/analytics_index.test.ts

@@ -187,7 +187,8 @@ describe('AnalyticsIndex', () => {
     expect(scheduleCAIBackfillTaskMock).toBeCalledTimes(0);
 
     expect(logger.debug).toBeCalledWith(
-      `[${indexName}] Mapping version is up to date. Skipping update.`
+      `[${indexName}] Mapping version is up to date. Skipping update.`,
+      { tags: ['cai-index-creation', `${indexName}`] }
     );
   });
 
@@ -204,7 +205,8 @@ describe('AnalyticsIndex', () => {
     expect(scheduleCAIBackfillTaskMock).toBeCalledTimes(0);
 
     expect(logger.debug).toBeCalledWith(
-      `[${indexName}] Mapping version is up to date. Skipping update.`
+      `[${indexName}] Mapping version is up to date. Skipping update.`,
+      { tags: ['cai-index-creation', `${indexName}`] }
     );
   });
 

x-pack/platform/plugins/shared/cases/server/cases_analytics/analytics_index.ts

@@ -122,10 +122,10 @@ export class AnalyticsIndex {
       const indexExists = await this.indexExists();
 
       if (!indexExists) {
-        this.logger.debug(`[${this.indexName}] Index does not exist. Creating.`);
+        this.logDebug(`Index does not exist. Creating.`);
         await this.createIndexMapping();
       } else {
-        this.logger.debug(`[${this.indexName}] Index exists. Updating mapping.`);
+        this.logDebug(`Index exists. Updating mapping.`);
         await this.updateIndexMapping();
       }
     } catch (error) {
@@ -140,7 +140,7 @@ export class AnalyticsIndex {
       if (shouldUpdateMapping) {
         await this.updateMapping();
       } else {
-        this.logger.debug(`[${this.indexName}] Mapping version is up to date. Skipping update.`);
+        this.logDebug(`Mapping version is up to date. Skipping update.`);
       }
     } catch (error) {
       this.handleError('Failed to update the index mapping.', error);
@@ -154,24 +154,24 @@ export class AnalyticsIndex {
   }
 
   private async updateMapping() {
-    this.logger.debug(`[${this.indexName}] Updating the painless script.`);
+    this.logDebug(`Updating the painless script.`);
     await this.putScript();
 
-    this.logger.debug(`[${this.indexName}] Updating index mapping.`);
+    this.logDebug(`Updating index mapping.`);
     await this.esClient.indices.putMapping({
       index: this.indexName,
       ...this.mappings,
     });
 
-    this.logger.debug(`[${this.indexName}] Scheduling the backfill task.`);
+    this.logDebug(`Scheduling the backfill task.`);
     await this.scheduleBackfillTask();
   }
 
   private async createIndexMapping() {
-    this.logger.debug(`[${this.indexName}] Creating painless script.`);
+    this.logDebug(`Creating painless script.`);
     await this.putScript();
 
-    this.logger.debug(`[${this.indexName}] Creating index.`);
+    this.logDebug(`Creating index.`);
     await this.esClient.indices.create({
       index: this.indexName,
       timeout: CAI_DEFAULT_TIMEOUT,
@@ -181,12 +181,12 @@ export class AnalyticsIndex {
       },
     });
 
-    this.logger.debug(`[${this.indexName}] Scheduling the backfill task.`);
+    this.logDebug(`Scheduling the backfill task.`);
     await this.scheduleBackfillTask();
   }
 
   private async indexExists(): Promise<boolean> {
-    this.logger.debug(`[${this.indexName}] Checking if index exists.`);
+    this.logDebug(`Checking if index exists.`);
     return this.esClient.indices.exists({
       index: this.indexName,
     });
@@ -197,12 +197,6 @@ export class AnalyticsIndex {
     return currentMapping[this.indexName].mappings._meta?.mapping_version < this.indexVersion;
   }
 
-  private handleError(message: string, error: EsErrors.ElasticsearchClientError) {
-    this.logger.error(`[${this.indexName}] ${message} Error message: ${error.message}`);
-
-    throw error;
-  }
-
   private async putScript() {
     await this.esClient.putScript({
       id: this.painlessScriptId,
@@ -217,8 +211,8 @@ export class AnalyticsIndex {
     indexVersion: number;
     painlessScriptId: string;
   }): MappingMeta {
-    this.logger.debug(
-      `[${this.indexName}] Construction mapping._meta. Index version: ${indexVersion}. Painless script: ${painlessScriptId}.`
+    this.logDebug(
+      `Construction mapping._meta. Index version: ${indexVersion}. Painless script: ${painlessScriptId}.`
     );
 
     return {
@@ -227,6 +221,17 @@ export class AnalyticsIndex {
     };
   }
 
+  public logDebug(message: string) {
+    this.logger.debug(`[${this.indexName}] ${message}`, {
+      tags: ['cai-index-creation', this.indexName],
+    });
+  }
+
+  private handleError(message: string, error: EsErrors.ElasticsearchClientError) {
+    this.logger.error(`[${this.indexName}] ${message} Error message: ${error.message}`);
+
+    throw error;
+  }
   private async scheduleBackfillTask() {
     await scheduleCAIBackfillTask({
       taskId: this.taskId,

x-pack/platform/plugins/shared/cases/server/cases_analytics/attachments_index/constants.ts

@@ -35,3 +35,48 @@ export const CAI_ATTACHMENTS_SOURCE_QUERY: QueryDslQueryContainer = {
 export const CAI_ATTACHMENTS_SOURCE_INDEX = ALERTING_CASES_SAVED_OBJECT_INDEX;
 
 export const CAI_ATTACHMENTS_BACKFILL_TASK_ID = 'cai_attachments_backfill_task';
+
+export const CAI_ATTACHMENTS_SYNCHRONIZATION_TASK_ID = 'cai_cases_attachments_synchronization_task';
+
+export const getAttachmentsSynchronizationSourceQuery = (
+  lastSyncAt: Date
+): QueryDslQueryContainer => ({
+  bool: {
+    must: [
+      {
+        term: {
+          type: 'cases-comments',
+        },
+      },
+      {
+        bool: {
+          must_not: {
+            term: {
+              'cases-comments.type': 'user',
+            },
+          },
+        },
+      },
+      {
+        bool: {
+          should: [
+            {
+              range: {
+                'cases-comments.created_at': {
+                  gte: lastSyncAt.toISOString(),
+                },
+              },
+            },
+            {
+              range: {
+                'cases-comments.updated_at': {
+                  gte: lastSyncAt.toISOString(),
+                },
+              },
+            },
+          ],
+        },
+      },
+    ],
+  },
+});

x-pack/platform/plugins/shared/cases/server/cases_analytics/attachments_index/index.ts

@@ -14,9 +14,11 @@ import {
   CAI_ATTACHMENTS_SOURCE_INDEX,
   CAI_ATTACHMENTS_SOURCE_QUERY,
   CAI_ATTACHMENTS_BACKFILL_TASK_ID,
+  CAI_ATTACHMENTS_SYNCHRONIZATION_TASK_ID,
 } from './constants';
 import { CAI_ATTACHMENTS_INDEX_MAPPINGS } from './mappings';
 import { CAI_ATTACHMENTS_INDEX_SCRIPT, CAI_ATTACHMENTS_INDEX_SCRIPT_ID } from './painless_scripts';
+import { scheduleCAISynchronizationTask } from '../tasks/synchronization_task';
 
 export const createAttachmentsAnalyticsIndex = ({
   esClient,
@@ -43,3 +45,23 @@ export const createAttachmentsAnalyticsIndex = ({
     sourceIndex: CAI_ATTACHMENTS_SOURCE_INDEX,
     sourceQuery: CAI_ATTACHMENTS_SOURCE_QUERY,
   });
+
+export const scheduleAttachmentsAnalyticsSyncTask = ({
+  taskManager,
+  logger,
+}: {
+  taskManager: TaskManagerStartContract;
+  logger: Logger;
+}) => {
+  scheduleCAISynchronizationTask({
+    taskId: CAI_ATTACHMENTS_SYNCHRONIZATION_TASK_ID,
+    sourceIndex: CAI_ATTACHMENTS_SOURCE_INDEX,
+    destIndex: CAI_ATTACHMENTS_INDEX_NAME,
+    taskManager,
+    logger,
+  }).catch((e) => {
+    logger.error(
+      `Error scheduling ${CAI_ATTACHMENTS_SYNCHRONIZATION_TASK_ID} task, received ${e.message}`
+    );
+  });
+};

x-pack/platform/plugins/shared/cases/server/cases_analytics/cases_index/constants.ts

@@ -22,3 +22,37 @@ export const CAI_CASES_SOURCE_QUERY: QueryDslQueryContainer = {
 export const CAI_CASES_SOURCE_INDEX = ALERTING_CASES_SAVED_OBJECT_INDEX;
 
 export const CAI_CASES_BACKFILL_TASK_ID = 'cai_cases_backfill_task';
+
+export const CAI_CASES_SYNCHRONIZATION_TASK_ID = 'cai_cases_synchronization_task';
+
+export const getCasesSynchronizationSourceQuery = (lastSyncAt: Date): QueryDslQueryContainer => ({
+  bool: {
+    must: [
+      {
+        term: {
+          type: 'cases',
+        },
+      },
+      {
+        bool: {
+          should: [
+            {
+              range: {
+                'cases.created_at': {
+                  gte: lastSyncAt.toISOString(),
+                },
+              },
+            },
+            {
+              range: {
+                'cases.updated_at': {
+                  gte: lastSyncAt.toISOString(),
+                },
+              },
+            },
+          ],
+        },
+      },
+    ],
+  },
+});

x-pack/platform/plugins/shared/cases/server/cases_analytics/cases_index/index.ts

@@ -14,9 +14,11 @@ import {
   CAI_CASES_SOURCE_INDEX,
   CAI_CASES_SOURCE_QUERY,
   CAI_CASES_BACKFILL_TASK_ID,
+  CAI_CASES_SYNCHRONIZATION_TASK_ID,
 } from './constants';
 import { CAI_CASES_INDEX_MAPPINGS } from './mappings';
 import { CAI_CASES_INDEX_SCRIPT_ID, CAI_CASES_INDEX_SCRIPT } from './painless_scripts';
+import { scheduleCAISynchronizationTask } from '../tasks/synchronization_task';
 
 export const createCasesAnalyticsIndex = ({
   esClient,
@@ -43,3 +45,23 @@ export const createCasesAnalyticsIndex = ({
     sourceIndex: CAI_CASES_SOURCE_INDEX,
     sourceQuery: CAI_CASES_SOURCE_QUERY,
   });
+
+export const scheduleCasesAnalyticsSyncTask = ({
+  taskManager,
+  logger,
+}: {
+  taskManager: TaskManagerStartContract;
+  logger: Logger;
+}) => {
+  scheduleCAISynchronizationTask({
+    taskId: CAI_CASES_SYNCHRONIZATION_TASK_ID,
+    sourceIndex: CAI_CASES_SOURCE_INDEX,
+    destIndex: CAI_CASES_INDEX_NAME,
+    taskManager,
+    logger,
+  }).catch((e) => {
+    logger.error(
+      `Error scheduling ${CAI_CASES_SYNCHRONIZATION_TASK_ID} task, received ${e.message}`
+    );
+  });
+};

x-pack/platform/plugins/shared/cases/server/cases_analytics/comments_index/constants.ts

@@ -33,3 +33,44 @@ export const CAI_COMMENTS_SOURCE_QUERY: QueryDslQueryContainer = {
 export const CAI_COMMENTS_SOURCE_INDEX = ALERTING_CASES_SAVED_OBJECT_INDEX;
 
 export const CAI_COMMENTS_BACKFILL_TASK_ID = 'cai_comments_backfill_task';
+
+export const CAI_COMMENTS_SYNCHRONIZATION_TASK_ID = 'cai_cases_comments_synchronization_task';
+
+export const getCommentsSynchronizationSourceQuery = (
+  lastSyncAt: Date
+): QueryDslQueryContainer => ({
+  bool: {
+    must: [
+      {
+        term: {
+          'cases-comments.type': 'user',
+        },
+      },
+      {
+        term: {
+          type: 'cases-comments',
+        },
+      },
+      {
+        bool: {
+          should: [
+            {
+              range: {
+                'cases-comments.created_at': {
+                  gte: lastSyncAt.toISOString(),
+                },
+              },
+            },
+            {
+              range: {
+                'cases-comments.updated_at': {
+                  gte: lastSyncAt.toISOString(),
+                },
+              },
+            },
+          ],
+        },
+      },
+    ],
+  },
+});

x-pack/platform/plugins/shared/cases/server/cases_analytics/comments_index/index.ts

@@ -14,9 +14,11 @@ import {
   CAI_COMMENTS_SOURCE_INDEX,
   CAI_COMMENTS_SOURCE_QUERY,
   CAI_COMMENTS_BACKFILL_TASK_ID,
+  CAI_COMMENTS_SYNCHRONIZATION_TASK_ID,
 } from './constants';
 import { CAI_COMMENTS_INDEX_MAPPINGS } from './mappings';
 import { CAI_COMMENTS_INDEX_SCRIPT, CAI_COMMENTS_INDEX_SCRIPT_ID } from './painless_scripts';
+import { scheduleCAISynchronizationTask } from '../tasks/synchronization_task';
 
 export const createCommentsAnalyticsIndex = ({
   esClient,
@@ -43,3 +45,23 @@ export const createCommentsAnalyticsIndex = ({
     sourceIndex: CAI_COMMENTS_SOURCE_INDEX,
     sourceQuery: CAI_COMMENTS_SOURCE_QUERY,
   });
+
+export const scheduleCommentsAnalyticsSyncTask = ({
+  taskManager,
+  logger,
+}: {
+  taskManager: TaskManagerStartContract;
+  logger: Logger;
+}) => {
+  scheduleCAISynchronizationTask({
+    taskId: CAI_COMMENTS_SYNCHRONIZATION_TASK_ID,
+    sourceIndex: CAI_COMMENTS_SOURCE_INDEX,
+    destIndex: CAI_COMMENTS_INDEX_NAME,
+    taskManager,
+    logger,
+  }).catch((e) => {
+    logger.error(
+      `Error scheduling ${CAI_COMMENTS_SYNCHRONIZATION_TASK_ID} task, received ${e.message}`
+    );
+  });
+};