[Entity Analytics][Watchlists] Fix source label override when adding entity via manual assign or CSV (elastic#264942)

## Summary

When an entity was added via the manual API or CSV upload after it had
already been sync'ed via index/integration, the used to query the
watchlist index to check if a doc already existed for that entity. If
that lookup missed, the write would replace the existing doc entirely
instead of merging into it, which would drop all previously written
source labels.

The fix removes the lookup and always uses a scripted upsert, which
merges source labels whether or not the doc already exists.

## How to test

1. Seed your ES instance with store data using the script and steps from
[here](elastic#263058 (comment)).

2. Configure a watchlist with an integration or index source and trigger
a sync so entities are populated in the watchlist index. Verify a doc
exists:
```
GET .entity_analytics.watchlists.default/_search
{
  "query": { "term": { "watchlist.id": "<your-watchlist-id>" } }
}
```
Note down any `entity.id` you find in the response

3. Manually assign that same entity to the same watchlist:
```
POST kbn:/api/entity_analytics/watchlists/<watchlist-id>/entities/assign
{
  "euids": ["<euid-from-step-2>"]
}
```

4. Re-fetch the doc and verify `labels.sources` contains **both** the
original sync source label and `"manual"` — not just `"manual"`.

5. Repeat step 3 using CSV upload and verify the same merge behaviour.

Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Jared Burgett <147995946+jaredburgettelastic@users.noreply.github.com>

Author: 3 people

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entities/utils.ts

@@ -13,7 +13,8 @@ export const ENTITY_ANALYTICS_WATCHLISTS_PREFIX = '.entity_analytics.watchlists'
 export const getIndexForWatchlist = (namespace: string) =>
   `${ENTITY_ANALYTICS_WATCHLISTS_PREFIX}.${namespace}`;
 
-/** Builds a composite document _id for the watchlist entity index. */
+// Design debt: this creates a per-(watchlist, entity) key instead of a per-entity key.
+// The intended design is one doc per entity across all watchlists; fixing this requires a migration.
 export const buildWatchlistDocId = (watchlistId: string, euid: string) => `${watchlistId}:${euid}`;
 
 /** Extracts the euid from a composite watchlist doc _id ({watchlistId}:{euid}). */

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/bulk/upsert.test.ts

@@ -20,93 +20,69 @@ describe('bulkUpsertOperationsFactory', () => {
     jest.clearAllMocks();
   });
 
-  it('generates index operations for new entities (no existingEntityId)', () => {
+  it('generates scripted upsert operations for new entities', () => {
     const entities: WatchlistBulkEntity[] = [
       { euid: 'user:alice', type: 'user', name: 'alice', sourceId: 'source-1' },
       { euid: 'user:bob', type: 'user', name: 'bob', sourceId: 'source-1' },
     ];
 
     const ops = buildOps({ entities, sourceLabel: 'index', targetIndex });
 
-    // Two entities → 4 elements (action + doc pairs)
+    // Two entities → 4 elements (action + body pairs)
     expect(ops).toHaveLength(4);
 
-    // First entity — index action
-    expect(ops[0]).toEqual({ index: { _index: targetIndex, _id: 'watchlist-1:user:alice' } });
-    // First entity — doc body
+    // First entity — update action with deterministic _id
+    expect(ops[0]).toEqual({ update: { _index: targetIndex, _id: 'watchlist-1:user:alice' } });
+    // First entity — scripted upsert body
     expect(ops[1]).toEqual(
       expect.objectContaining({
-        entity: { id: 'user:alice', name: 'alice', type: 'user' },
-        labels: { sources: ['index'], source_ids: ['source-1'] },
-        watchlist,
+        script: expect.objectContaining({
+          source: UPDATE_SCRIPT_SOURCE,
+          params: expect.objectContaining({ source_id: 'source-1', source_type: 'index' }),
+        }),
+        upsert: expect.objectContaining({
+          entity: { id: 'user:alice', name: 'alice', type: 'user' },
+          labels: { sources: ['index'], source_ids: ['source-1'] },
+          watchlist,
+        }),
       })
     );
 
-    // Second entity — index action
-    expect(ops[2]).toEqual({ index: { _index: targetIndex, _id: 'watchlist-1:user:bob' } });
+    // Second entity
+    expect(ops[2]).toEqual({ update: { _index: targetIndex, _id: 'watchlist-1:user:bob' } });
     expect(ops[3]).toEqual(
       expect.objectContaining({
-        entity: { id: 'user:bob', name: 'bob', type: 'user' },
-        labels: { sources: ['index'], source_ids: ['source-1'] },
-        watchlist,
+        script: expect.objectContaining({ source: UPDATE_SCRIPT_SOURCE }),
+        upsert: expect.objectContaining({
+          entity: { id: 'user:bob', name: 'bob', type: 'user' },
+        }),
       })
     );
   });
 
-  it('generates update operations for existing entities (with existingEntityId)', () => {
+  it('generates scripted upsert operations for existing entities, merging source labels', () => {
     const entities: WatchlistBulkEntity[] = [
-      {
-        euid: 'user:alice',
-        type: 'user',
-        name: 'alice',
-        existingEntityId: 'doc-123',
-        sourceId: 'source-2',
-      },
+      { euid: 'user:alice', type: 'user', name: 'alice', sourceId: 'source-2' },
     ];
 
     const ops = buildOps({ entities, sourceLabel: 'index', targetIndex });
 
     expect(ops).toHaveLength(2);
 
-    // Update action
-    expect(ops[0]).toEqual({ update: { _index: targetIndex, _id: 'doc-123' } });
-    // Script body
+    expect(ops[0]).toEqual({ update: { _index: targetIndex, _id: 'watchlist-1:user:alice' } });
     expect(ops[1]).toEqual(
       expect.objectContaining({
         script: expect.objectContaining({
           source: UPDATE_SCRIPT_SOURCE,
-          params: expect.objectContaining({
-            source_id: 'source-2',
-            source_type: 'index',
-          }),
+          params: expect.objectContaining({ source_id: 'source-2', source_type: 'index' }),
+        }),
+        upsert: expect.objectContaining({
+          labels: expect.objectContaining({ source_ids: ['source-2'] }),
         }),
       })
     );
   });
 
-  it('generates mixed operations for a mix of new and existing entities', () => {
-    const entities: WatchlistBulkEntity[] = [
-      { euid: 'user:new-user', type: 'user', name: 'new-user', sourceId: 'src-1' },
-      {
-        euid: 'user:existing-user',
-        type: 'user',
-        name: 'existing-user',
-        existingEntityId: 'doc-456',
-        sourceId: 'src-1',
-      },
-    ];
-
-    const ops = buildOps({ entities, sourceLabel: 'index', targetIndex });
-
-    expect(ops).toHaveLength(4);
-
-    // First is an index op
-    expect(ops[0]).toEqual({ index: { _index: targetIndex, _id: 'watchlist-1:user:new-user' } });
-
-    // Second pair is an update op
-    expect(ops[2]).toEqual({ update: { _index: targetIndex, _id: 'doc-456' } });
-  });
-
   it('returns an empty array when no entities are provided', () => {
     const ops = buildOps({ entities: [], sourceLabel: 'index', targetIndex });
     expect(ops).toHaveLength(0);

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/bulk/upsert.ts

@@ -69,26 +69,20 @@ export const bulkUpsertOperationsFactory =
     const ops: object[] = [];
     logger.debug(`[WatchlistSync] Building bulk operations for ${entities.length} entities`);
     for (const entity of entities) {
-      if (entity.existingEntityId) {
-        ops.push(
-          { update: { _index: targetIndex, _id: entity.existingEntityId } },
-          {
-            script: {
-              source: UPDATE_SCRIPT_SOURCE,
-              params: {
-                now: new Date().toISOString(),
-                source_id: entity.sourceId,
-                source_type: sourceLabel,
-              },
+      ops.push(
+        { update: { _index: targetIndex, _id: buildWatchlistDocId(watchlist.id, entity.euid) } },
+        {
+          script: {
+            source: UPDATE_SCRIPT_SOURCE,
+            params: {
+              now: new Date().toISOString(),
+              source_id: entity.sourceId,
+              source_type: sourceLabel,
             },
-          }
-        );
-      } else {
-        ops.push(
-          { index: { _index: targetIndex, _id: buildWatchlistDocId(watchlist.id, entity.euid) } },
-          buildCreateDoc(entity, sourceLabel, watchlist)
-        );
-      }
+          },
+          upsert: buildCreateDoc(entity, sourceLabel, watchlist),
+        }
+      );
     }
     return ops;
   };

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/csv/process_batch.ts

@@ -11,7 +11,7 @@ import pMap from 'p-map';
 import type { WatchlistCsvUploadResponseItem } from '../../../../../../common/api/entity_analytics/watchlists/csv_upload/csv_upload.gen';
 import { bulkUpsertOperationsFactory } from '../bulk/upsert';
 import { addWatchlistAttributeToStore } from '../sync/entity_store_sync';
-import { getExistingEntitiesMap, getErrorFromBulkResponse, errorsMsg } from '../sync/utils';
+import { getErrorFromBulkResponse, errorsMsg } from '../sync/utils';
 import { MANUAL_SOURCE_ID } from '../manual/constants';
 import type { MatchedEntity, Watchlist } from './types';
 import { lookupEntitiesForRow } from './lookup';
@@ -68,18 +68,11 @@ const upsertToWatchlistIndex = async (
     currentWatchlists: m.currentWatchlists,
   }));
 
-  const existingMap = await getExistingEntitiesMap(
-    esClient,
-    watchlist,
-    entities.map((e) => e.euid)
-  );
-  const enriched = entities.map((e) => ({ ...e, existingEntityId: existingMap.get(e.euid) }));
-
   const operations = bulkUpsertOperationsFactory(
     logger,
     watchlist
   )({
-    entities: enriched,
+    entities,
     sourceLabel: MANUAL_SOURCE_ID,
     targetIndex: watchlist.index,
   });

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/manual/service.ts

@@ -12,12 +12,7 @@ import { uniq } from 'lodash';
 
 import type { WatchlistEntityAssignResponseItem } from '../../../../../../common/api/entity_analytics/watchlists/entities/assign.gen';
 import type { WatchlistEntityUnassignResponseItem } from '../../../../../../common/api/entity_analytics/watchlists/entities/unassign.gen';
-import {
-  getExistingEntitiesMap,
-  getErrorFromBulkResponse,
-  errorsMsg,
-  partitionBulkResults,
-} from '../sync/utils';
+import { getErrorFromBulkResponse, errorsMsg, partitionBulkResults } from '../sync/utils';
 import { bulkUpsertOperationsFactory } from '../bulk/upsert';
 import { addWatchlistAttributeToStore } from '../sync/entity_store_sync';
 import { applyBulkRemoveSource } from '../bulk/soft_delete';
@@ -84,13 +79,11 @@ export const createManualEntityService = ({
     }
 
     try {
-      const existingMap = await getExistingEntitiesMap(esClient, watchlist, [...foundEuids]);
       const bulkEntities: WatchlistBulkEntity[] = foundEntities.map((e) => ({
         euid: e.euid,
         type: e.type,
         name: e.name,
         sourceId: MANUAL_SOURCE_ID,
-        existingEntityId: existingMap.get(e.euid),
         currentWatchlists: e.currentWatchlists,
       }));
 

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/sync/__mocks__/utils.ts

@@ -7,7 +7,6 @@
 
 const actual = jest.requireActual('../utils');
 
-export const getExistingEntitiesMap = jest.fn().mockResolvedValue(new Map());
 export const getErrorFromBulkResponse = jest.fn().mockReturnValue([]);
 export const errorsMsg = jest.fn().mockReturnValue('');
 export const isTimestampGreaterThan = jest.fn().mockReturnValue(false);

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/sync/update_detection/update_detection.ts

@@ -14,7 +14,7 @@ import type { EntityStoreEntityIdsByType, WatchlistsByEuid } from '../../../enti
 import type { WatchlistBulkEntity } from '../../types';
 import { createWatchlistSyncMarkersService } from '../sync_markers';
 import type { WatchlistEntitySourceClient } from '../../infra';
-import { isTimestampGreaterThan, getExistingEntitiesMap } from '../utils';
+import { isTimestampGreaterThan } from '../utils';
 import { buildEntitiesSearchBody, buildIndexSourceSearchBody } from './queries';
 import { applyBulkUpsert } from '../../bulk/upsert';
 import { getEntityNameFromDoc } from './entity_utils';
@@ -55,8 +55,6 @@ const pickLaterTimestamp = (
 };
 
 const paginatedDetection = async <B>(
-  esClient: ElasticsearchClient,
-  watchlist: { name: string; id: string; index: string },
   search: SearchPage<B>,
   mapBucket: MapBucket<B>
 ): Promise<{ entities: WatchlistBulkEntity[]; maxTimestamp?: string }> => {
@@ -77,11 +75,7 @@ const paginatedDetection = async <B>(
         return acc;
       }, []);
 
-      const batchEuids = mapped.map((m) => m.euid);
-      const existingMap = await getExistingEntitiesMap(esClient, watchlist, batchEuids);
-
-      for (const { euid, entity, timestamp } of mapped) {
-        entity.existingEntityId = existingMap.get(euid);
+      for (const { entity, timestamp } of mapped) {
         maxTimestamp = pickLaterTimestamp(maxTimestamp, timestamp);
         allEntities.push(entity);
       }
@@ -160,7 +154,7 @@ export const createUpdateDetectionService = ({
       return { euid, entity, timestamp: typeof ts === 'string' ? ts : undefined };
     };
 
-    return paginatedDetection(esClient, watchlist, search, mapBucket);
+    return paginatedDetection(search, mapBucket);
   };
 
   const detectForIndexSource = async (
@@ -207,7 +201,7 @@ export const createUpdateDetectionService = ({
       };
     };
 
-    return paginatedDetection(esClient, watchlist, search, mapBucket);
+    return paginatedDetection(search, mapBucket);
   };
 
   const detectForStoreSource = async (
@@ -222,21 +216,6 @@ export const createUpdateDetectionService = ({
       }
     }
 
-    if (allEntities.length === 0) {
-      return { entities: [] as WatchlistBulkEntity[] };
-    }
-
-    // Check which entities already exist in the target index
-    const pageSize = 100;
-    for (let start = 0; start < allEntities.length; start += pageSize) {
-      const batch = allEntities.slice(start, start + pageSize);
-      const batchEuids = batch.map((e) => e.euid);
-      const existingMap = await getExistingEntitiesMap(esClient, watchlist, batchEuids);
-      for (const entity of batch) {
-        entity.existingEntityId = existingMap.get(entity.euid);
-      }
-    }
-
     return { entities: allEntities };
   };
 

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/sync/utils.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { uniq } from 'lodash';
-import type { ElasticsearchClient } from '@kbn/core/server';
 import type { BulkResponse, ErrorCause } from '@elastic/elasticsearch/lib/api/types';
 import moment from 'moment';
 
@@ -50,33 +48,3 @@ export const isTimestampGreaterThan = (date1: string, date2: string) => {
   if (!m2.isValid()) return true;
   return m1.isAfter(m2);
 };
-
-/**
- * Looks up which EUIDs already have a document in the watchlist index,
- * returning a Map from EUID → ES document _id.
- */
-export const getExistingEntitiesMap = async (
-  esClient: ElasticsearchClient,
-  watchlist: { id: string; index: string },
-  euids: string[]
-): Promise<Map<string, string>> => {
-  if (euids.length === 0) return new Map();
-
-  const uniqueEuids = uniq(euids);
-  const response = await esClient.search<{ entity?: { id?: string } }>({
-    index: watchlist.index,
-    size: uniqueEuids.length,
-    query: {
-      bool: {
-        must: [{ terms: { 'entity.id': uniqueEuids } }, { term: { 'watchlist.id': watchlist.id } }],
-      },
-    },
-    _source: ['entity.id'],
-  });
-
-  return response.hits.hits.reduce((map, hit) => {
-    const euid = hit._source?.entity?.id;
-    if (euid && hit._id) map.set(euid, hit._id);
-    return map;
-  }, new Map<string, string>());
-};

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/watchlists/entity_sources/types.ts

@@ -13,7 +13,6 @@ export interface WatchlistBulkEntity {
   type: EntityType;
   name?: string;
   sourceId: string;
-  existingEntityId?: string;
   /** Current watchlist names from the entity store, used for store sync */
   currentWatchlists?: string[];
 }