[SigEvents] Only show rule-backed queries in stream details Significant Events tab (elastic#254953)

It closes elastic#254950

## Summary

This PR refines the stream details Significant Events tab to only
display rule-backed queries and changes the navigation in SigEvents
Discovery to land in the Significant Events tab when clicking on a
stream name.

### Changes

- **Rule-backed filter**: Adds a `ruleBacked` query parameter to the
`GET /internal/streams/_significant_events` API, threaded through to the
`QueryClient` methods (`getQueryLinks`, `findQueries`). The stream
details Significant Events tab passes `ruleBacked: true` to only show
queries that have been promoted to a backing Kibana rule. The Discovery
page continues to show all queries (backed and unbacked).
- **Discovery navigation**: Clicking a stream name in the SigEvents
Discovery streams table now navigates directly to the Significant Events
management tab (`/{key}/management/significantEvents`) instead of the
default stream overview.

Before:


https://github.com/user-attachments/assets/c88e0a23-5d38-4c1e-aa69-e092dc01fbe6

After:


https://github.com/user-attachments/assets/b05861f2-fee6-4b66-b0c7-f4812a35b7e2

Author: cesco-f

x-pack/platform/plugins/shared/streams/common/queries.ts

@@ -7,17 +7,14 @@
 
 import type { StreamQuery, StreamQueryInput } from '@kbn/streams-schema';
 
-// Legacy stored query links may not include rule_backed and should be treated as already backed.
-export const LEGACY_RULE_BACKED_FALLBACK = true;
-
 export interface QueryLink {
   'asset.uuid': string;
   'asset.type': 'query';
   'asset.id': string;
   query: StreamQuery;
   stream_name: string;
   /** Whether a Kibana rule exists for this query. */
-  rule_backed?: boolean;
+  rule_backed: boolean;
   /** The deterministic ID of the Kibana rule associated with this query. */
   rule_id: string;
 }

x-pack/platform/plugins/shared/streams/server/lib/significant_events/read_significant_events_from_alerts_indices.ts

@@ -13,24 +13,31 @@ import type { IScopedClusterClient } from '@kbn/core/server';
 import type { ChangePointType } from '@kbn/es-types/src';
 import type { StreamQuery, SignificantEventsGetResponse } from '@kbn/streams-schema';
 import { get, isArray, isEmpty, keyBy } from 'lodash';
-import { LEGACY_RULE_BACKED_FALLBACK, type QueryLink } from '../../../common/queries';
-import type { QueryClient } from '../streams/assets/query/query_client';
+import type { QueryLink } from '../../../common/queries';
+import type { QueryClient, QueryLinkFilters } from '../streams/assets/query/query_client';
 import { parseError } from '../streams/errors/parse_error';
 import { SecurityError } from '../streams/errors/security_error';
 
 export async function readSignificantEventsFromAlertsIndices(
-  params: { streamNames?: string[]; from: Date; to: Date; bucketSize: string; query?: string },
+  params: {
+    streamNames?: string[];
+    from: Date;
+    to: Date;
+    bucketSize: string;
+    query?: string;
+    filters?: QueryLinkFilters;
+  },
   dependencies: {
     queryClient: QueryClient;
     scopedClusterClient: IScopedClusterClient;
   }
 ): Promise<SignificantEventsGetResponse> {
   const { queryClient, scopedClusterClient } = dependencies;
-  const { streamNames = [], from, to, bucketSize, query } = params;
+  const { streamNames = [], from, to, bucketSize, query, filters } = params;
 
   const queryLinks = query
-    ? await queryClient.findQueries(streamNames, query)
-    : await queryClient.getQueryLinks(streamNames);
+    ? await queryClient.findQueries(streamNames, query, filters)
+    : await queryClient.getQueryLinks(streamNames, filters);
 
   if (isEmpty(queryLinks)) {
     return { significant_events: [], aggregated_occurrences: [] };
@@ -142,7 +149,7 @@ export async function readSignificantEventsFromAlertsIndices(
             stationary: { p_value: 0, change_point: 0 },
           },
         },
-        rule_backed: queryLink.rule_backed ?? LEGACY_RULE_BACKED_FALLBACK,
+        rule_backed: queryLink.rule_backed,
       })),
       aggregated_occurrences: [],
     };
@@ -168,7 +175,7 @@ export async function readSignificantEventsFromAlertsIndices(
             count: occurrence.doc_count,
           }))
         : [],
-      rule_backed: queryLink.rule_backed ?? LEGACY_RULE_BACKED_FALLBACK,
+      rule_backed: queryLink.rule_backed,
       change_points: changePoints,
     };
   });
@@ -185,7 +192,7 @@ export async function readSignificantEventsFromAlertsIndices(
           stationary: { p_value: 0, change_point: 0 },
         },
       },
-      rule_backed: queryLink.rule_backed ?? LEGACY_RULE_BACKED_FALLBACK,
+      rule_backed: queryLink.rule_backed,
     }));
 
   return {

x-pack/platform/plugins/shared/streams/server/lib/streams/assets/query/query_client.ts

@@ -16,7 +16,6 @@ import { isEqual } from 'lodash';
 import objectHash from 'object-hash';
 import pLimit from 'p-limit';
 import {
-  LEGACY_RULE_BACKED_FALLBACK,
   type Query,
   type QueryLink,
   type QueryLinkRequest,
@@ -45,6 +44,10 @@ import { computeRuleId } from './helpers/query';
 
 type TermQueryFieldValue = string | boolean | number | null;
 
+export interface QueryLinkFilters {
+  ruleBacked?: boolean;
+}
+
 interface TermQueryOpts {
   queryEmptyString: boolean;
 }
@@ -61,6 +64,30 @@ function termQuery<T extends string>(
   return [{ term: { [field]: value } }];
 }
 
+function ruleBackedFilter(value: boolean | undefined): QueryDslQueryContainer[] {
+  if (value === undefined) {
+    return [];
+  }
+
+  if (!value) {
+    return termQuery(RULE_BACKED, false);
+  }
+
+  // When filtering for rule-backed queries, also include legacy docs
+  // that predate the rule_backed field.
+  return [
+    {
+      bool: {
+        should: [
+          { term: { [RULE_BACKED]: true } },
+          { bool: { must_not: [{ exists: { field: RULE_BACKED } }] } },
+        ],
+        minimum_should_match: 1,
+      },
+    },
+  ];
+}
+
 function termsQuery<T extends string>(
   field: T,
   values: Array<TermQueryFieldValue | undefined> | null | undefined
@@ -114,7 +141,6 @@ type QueryLinkStorageFields = Omit<QueryLink, 'query' | 'stream_name'> & {
   [QUERY_KQL_BODY]: string;
   [QUERY_ESQL_QUERY]: string;
   [QUERY_SEVERITY_SCORE]?: number;
-  [RULE_BACKED]?: boolean;
 };
 
 export type StoredQueryLink = QueryLinkStorageFields & {
@@ -136,18 +162,16 @@ function fromStorage(link: StoredQueryLink): QueryLink {
     [QUERY_FEATURE_FILTER]: string;
     [QUERY_FEATURE_TYPE]: 'system';
     [QUERY_EVIDENCE]?: string[];
-    [RULE_BACKED]?: boolean;
   } = link as StoredQueryLink & {
     [QUERY_FEATURE_NAME]: string;
     [QUERY_FEATURE_FILTER]: string;
     [QUERY_FEATURE_TYPE]: 'system';
     [QUERY_EVIDENCE]?: string[];
-    [RULE_BACKED]?: boolean;
   };
   return {
     ...storageFields,
     stream_name: link[STREAM_NAME],
-    rule_backed: storageFields[RULE_BACKED] ?? LEGACY_RULE_BACKED_FALLBACK,
+    rule_backed: storageFields[RULE_BACKED],
     rule_id: storageFields[RULE_ID],
     query: {
       id: storageFields[ASSET_ID],
@@ -171,15 +195,9 @@ function fromStorage(link: StoredQueryLink): QueryLink {
   } satisfies QueryLink;
 }
 
-type QueryLinkRequestWithRuleBacked = QueryLinkRequest & { rule_backed?: boolean };
-
-function toStorage(
-  definition: Streams.all.Definition,
-  request: QueryLinkRequestWithRuleBacked
-): StoredQueryLink {
+function toStorage(definition: Streams.all.Definition, request: QueryLinkRequest): StoredQueryLink {
   const link = toQueryLink(definition, request);
   const { query, stream_name, ...rest } = link;
-  const ruleBacked = request.rule_backed ?? LEGACY_RULE_BACKED_FALLBACK;
   return {
     ...rest,
     [STREAM_NAME]: definition.name,
@@ -191,7 +209,7 @@ function toStorage(
     [QUERY_FEATURE_TYPE]: query.feature ? query.feature.type : '',
     [QUERY_SEVERITY_SCORE]: query.severity_score,
     [QUERY_EVIDENCE]: query.evidence,
-    [RULE_BACKED]: ruleBacked,
+    [RULE_BACKED]: request.rule_backed,
     [RULE_ID]: link.rule_id,
   } as StoredQueryLink;
 }
@@ -203,14 +221,23 @@ function hasBreakingChange(currentQuery: StreamQuery, nextQuery: StreamQuery): b
   );
 }
 
-function toQueryLinkFromQuery(query: StreamQuery, stream: string): QueryLink {
+function toQueryLinkFromQuery({
+  query,
+  stream,
+  ruleBacked = true,
+}: {
+  query: StreamQuery;
+  stream: string;
+  ruleBacked?: boolean;
+}): QueryLink {
   const assetUuid = getQueryLinkUuid(stream, { 'asset.type': 'query', 'asset.id': query.id });
   return {
     'asset.uuid': assetUuid,
     'asset.type': 'query',
     'asset.id': query.id,
     query,
     stream_name: stream,
+    rule_backed: ruleBacked,
     rule_id: computeRuleId(assetUuid, query.esql.query),
   };
 }
@@ -230,7 +257,7 @@ export class QueryClient {
 
   async syncQueryList(
     definition: Streams.all.Definition,
-    links: QueryLinkRequestWithRuleBacked[]
+    links: QueryLinkRequest[]
   ): Promise<{ deleted: QueryLink[]; indexed: QueryLink[] }> {
     const name = definition.name;
     const assetsResponse = await this.dependencies.storageClient.search({
@@ -313,8 +340,12 @@ export class QueryClient {
    * Returns all query links for given streams or
    * all query links if no stream names are provided.
    */
-  async getQueryLinks(streamNames: string[]): Promise<QueryLink[]> {
-    const filter = [...termsQuery(STREAM_NAME, streamNames), ...termQuery(ASSET_TYPE, 'query')];
+  async getQueryLinks(streamNames: string[], filters?: QueryLinkFilters): Promise<QueryLink[]> {
+    const filter = [
+      ...termsQuery(STREAM_NAME, streamNames),
+      ...termQuery(ASSET_TYPE, 'query'),
+      ...ruleBackedFilter(filters?.ruleBacked),
+    ];
 
     const queriesResponse = await this.dependencies.storageClient.search({
       size: 10_000,
@@ -334,23 +365,7 @@ export class QueryClient {
    * Used internally by promoteQueries.
    */
   private async getUnbackedQueries(streamName: string): Promise<QueryLink[]> {
-    const filter = [
-      ...termQuery(STREAM_NAME, streamName),
-      ...termQuery(ASSET_TYPE, 'query'),
-      ...termQuery(RULE_BACKED, false),
-    ];
-
-    const assetsResponse = await this.dependencies.storageClient.search({
-      size: 10_000,
-      track_total_hits: false,
-      query: {
-        bool: {
-          filter,
-        },
-      },
-    });
-
-    return assetsResponse.hits.hits.map((hit) => fromStorage(hit._source));
+    return this.getQueryLinks([streamName], { ruleBacked: false });
   }
 
   /**
@@ -377,19 +392,7 @@ export class QueryClient {
    * Returns all query links across streams that do not have a backing Kibana rule.
    */
   async getAllUnbackedQueries(): Promise<QueryLink[]> {
-    const filter = [...termQuery(ASSET_TYPE, 'query'), ...termQuery(RULE_BACKED, false)];
-
-    const assetsResponse = await this.dependencies.storageClient.search({
-      size: 10_000,
-      track_total_hits: false,
-      query: {
-        bool: {
-          filter,
-        },
-      },
-    });
-
-    return assetsResponse.hits.hits.map((hit) => fromStorage(hit._source));
+    return this.getQueryLinks([], { ruleBacked: false });
   }
 
   async bulkGetByIds(name: string, ids: string[]) {
@@ -413,8 +416,16 @@ export class QueryClient {
     return assetsResponse.hits.hits.map((hit) => fromStorage(hit._source));
   }
 
-  async findQueries(streamNames: string[], query: string): Promise<QueryLink[]> {
-    const filter = [...termsQuery(STREAM_NAME, streamNames), ...termQuery(ASSET_TYPE, 'query')];
+  async findQueries(
+    streamNames: string[],
+    query: string,
+    filters?: QueryLinkFilters
+  ): Promise<QueryLink[]> {
+    const filter = [
+      ...termsQuery(STREAM_NAME, streamNames),
+      ...termQuery(ASSET_TYPE, 'query'),
+      ...ruleBackedFilter(filters?.ruleBacked),
+    ];
 
     const assetsResponse = await this.dependencies.storageClient.search({
       size: 10_000,
@@ -446,7 +457,7 @@ export class QueryClient {
         if ('index' in operation) {
           const document = toStorage(
             definition,
-            Object.values(operation)[0].asset as QueryLinkRequestWithRuleBacked
+            Object.values(operation)[0].asset as QueryLinkRequest
           );
           return {
             index: {
@@ -482,6 +493,7 @@ export class QueryClient {
         query: link.query,
         title: link.query.title,
         stream_name: link.stream_name,
+        rule_backed: link.rule_backed,
         rule_id: link.rule_id,
       };
     });
@@ -520,11 +532,11 @@ export class QueryClient {
     for (const query of queries) {
       const currentLink = currentLinkByQueryId.get(query.id);
       if (!currentLink) {
-        const link = toQueryLinkFromQuery(query, stream);
+        const link = toQueryLinkFromQuery({ query, stream });
         nextQueriesToCreate.push(link);
         allNextQueryLinks.push(link);
       } else if (hasBreakingChange(currentLink.query, query)) {
-        const link = toQueryLinkFromQuery(query, stream);
+        const link = toQueryLinkFromQuery({ query, stream });
         nextQueriesUpdatedWithBreakingChange.push(link);
         allNextQueryLinks.push(link);
       } else {
@@ -638,28 +650,23 @@ export class QueryClient {
       ...operations
         .filter((operation) => operation.index && !currentIds.has(operation.index!.id))
         .map((operation) =>
-          toQueryLinkFromQuery(
-            {
+          toQueryLinkFromQuery({
+            query: {
               ...operation.index!,
               esql: {
                 query: buildEsqlQuery(indices, operation.index!),
               },
             },
-            stream
-          )
+            stream,
+            ruleBacked: options?.createRules !== false,
+          })
         ),
     ];
 
     if (options?.createRules === false) {
-      const nextQueriesWithRuleBacked = nextQueries.map((link) => ({
-        ...link,
-        rule_backed: currentIds.has(link.query.id)
-          ? link.rule_backed ?? LEGACY_RULE_BACKED_FALLBACK
-          : false,
-      }));
       await this.syncQueryList(
         definition,
-        nextQueriesWithRuleBacked.map((link) => ({
+        nextQueries.map((link) => ({
           [ASSET_ID]: link[ASSET_ID],
           [ASSET_TYPE]: link[ASSET_TYPE],
           query: link.query,
@@ -696,7 +703,7 @@ export class QueryClient {
     const idSet = new Set(queryIds);
     const toPromote = unbacked
       .filter((link) => idSet.has(link.query.id))
-      .map((link) => toQueryLinkFromQuery(link.query, streamName));
+      .map((link) => toQueryLinkFromQuery({ query: link.query, stream: streamName }));
 
     if (toPromote.length === 0) {
       return { promoted: 0 };

x-pack/platform/plugins/shared/streams/server/lib/streams/assets/query/query_service.ts

@@ -19,6 +19,7 @@ import {
   QUERY_FEATURE_NAME,
   STREAM_NAME,
   RULE_ID,
+  RULE_BACKED,
   ASSET_UUID,
 } from '../fields';
 import { queryStorageSettings, type QueryStorageSettings } from '../storage_settings';
@@ -95,6 +96,11 @@ export class QueryService {
             migrated = { ...migrated, [RULE_ID]: computeRuleId(uuid, kqlQuery) };
           }
 
+          // Pre-existing queries were all rule-backed; back-fill the flag so it is persisted.
+          if (!(RULE_BACKED in migrated)) {
+            migrated = { ...migrated, [RULE_BACKED]: true };
+          }
+
           return migrated as StoredQueryLink;
         },
       }