[Streams 🌊] Stream enrichment processors management (elastic#204793)

## 📓 Summary

Part of #elastic/streams-program#32

This work implements a UI for basic stream enrichment, supporting grok
and dissect processor + detected fields mapping.

The main features implemented in this PR consist of:
- **Sortable processors list**
- **Add new processor - Grok, Dissect**
  - Ad-hoc forms for each processor
  - Simulated document outcome with extracted fields
  - Filter matching documents with parsed fields
  - Mapping detected fields (only available for wired streams)
- **Edit processor**
  - Change configuration only
  - Delete processor CTA

As a side quest, I added a small package for object utils as
@simianhacker suggested.
`@kbn/object-utils` exposes `calculateObjectDiff` and `flattenObject` to
detect the changed fields in a simulation.

## 🔜 Follow-up work

I'll work on minor updates on top of this MVP to make this available for
further testing from the team.
The next steps will be:
- **Tests** for features that consolidate on the functional pov.
- Better field mapping detection and UI feedback (infer the type of the
detected field, currently always unmapped)
- Add better form validation and feedback for processor configuration.

As discussed offline, state management is purely based on the built-in
react APIs + react-hook-form. It could be improved with different
approaches, including a more solid state management library to make it
easier to maintain and bulletproof to race conditions. No state syncs
with the URL currently.

## 🎥 Demo


https://github.com/user-attachments/assets/a48fade9-f5aa-4270-bb19-d91d1eed822b

---------

Co-authored-by: Marco Antonio Ghiani <marcoantonio.ghiani@elastic.co>
Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

.github/CODEOWNERS

@@ -497,6 +497,7 @@ src/platform/packages/shared/kbn-management/settings/field_definition @elastic/k
 src/platform/packages/shared/kbn-management/settings/types @elastic/kibana-management
 src/platform/packages/shared/kbn-management/settings/utilities @elastic/kibana-management
 src/platform/packages/shared/kbn-monaco @elastic/appex-sharedux
+src/platform/packages/shared/kbn-object-utils @elastic/kibana-core
 src/platform/packages/shared/kbn-object-versioning @elastic/appex-sharedux
 src/platform/packages/shared/kbn-object-versioning-utils @elastic/appex-sharedux
 src/platform/packages/shared/kbn-openapi-common @elastic/security-detection-rule-management

package.json

@@ -686,6 +686,7 @@
     "@kbn/newsfeed-test-plugin": "link:test/common/plugins/newsfeed",
     "@kbn/no-data-page-plugin": "link:src/platform/plugins/private/no_data_page",
     "@kbn/notifications-plugin": "link:x-pack/platform/plugins/shared/notifications",
+    "@kbn/object-utils": "link:src/platform/packages/shared/kbn-object-utils",
     "@kbn/object-versioning": "link:src/platform/packages/shared/kbn-object-versioning",
     "@kbn/object-versioning-utils": "link:src/platform/packages/shared/kbn-object-versioning-utils",
     "@kbn/observability-ai-assistant-app-plugin": "link:x-pack/solutions/observability/plugins/observability_ai_assistant_app",

src/platform/packages/shared/kbn-object-utils/README.md

@@ -0,0 +1,76 @@
+# @kbn/object-utils
+
+Utilities for objects manipulation and parsing.
+
+## Utilities
+
+### calculateObjectDiff
+
+This utils compares two JSON objects and calculates the added and removed properties, including nested properties.
+
+```ts
+const oldObject = {
+  alpha: 1,
+  beta: {
+    gamma: 2,
+    delta: {
+      sigma: 7,
+    },
+  },
+};
+
+const newObject = {
+  alpha: 1,
+  beta: {
+    gamma: 2,
+    eta: 4,
+  },
+};
+
+const diff = calculateObjectDiff(oldObject, newObject);
+
+/*
+Result:
+{
+  added: {
+    beta: {
+      eta: 4,
+    },
+  },
+  removed: {
+    beta: {
+      delta: {
+        sigma: 7,
+      },
+    },
+  },
+}
+*/
+```
+
+### flattenObject
+
+This utils returns a flattened version of the input object also accounting for nested properties.
+
+```ts
+const flattened = flattenObject({
+  alpha: {
+    gamma: {
+      sigma: 1,
+    },
+    delta: {
+      sigma: 2,
+    },
+  },
+  beta: 3,
+});
+
+/*
+Result:
+{
+  'alpha.gamma.sigma': 1,
+  'alpha.delta.sigma': 2,
+  beta: 3,
+}
+*/
+```

src/platform/packages/shared/kbn-object-utils/index.ts

@@ -0,0 +1,11 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+export * from './src/calculate_object_diff';
+export * from './src/flatten_object';

src/platform/packages/shared/kbn-object-utils/jest.config.js

@@ -0,0 +1,14 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+module.exports = {
+  preset: '@kbn/test',
+  rootDir: '../../../../..',
+  roots: ['<rootDir>/src/platform/packages/shared/kbn-object-utils'],
+};

src/platform/packages/shared/kbn-object-utils/kibana.jsonc

@@ -0,0 +1,10 @@
+{
+  "type": "shared-common",
+  "id": "@kbn/object-utils",
+  "owner": [
+    "@elastic/kibana-core"
+  ],
+  "group": "platform",
+  "visibility": "shared",
+  "devOnly": false
+}

src/platform/packages/shared/kbn-object-utils/package.json

@@ -0,0 +1,8 @@
+{
+  "description": "Object utils for Kibana",
+  "license": "Elastic License 2.0 OR AGPL-3.0-only OR SSPL-1.0",
+  "name": "@kbn/object-utils",
+  "private": true,
+  "version": "1.0.0",
+  "sideEffects": false
+}

src/platform/packages/shared/kbn-object-utils/src/calculate_object_diff.test.ts

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { calculateObjectDiff } from './calculate_object_diff';
+
+describe('calculateObjectDiff', () => {
+  it('should return the added and removed parts between 2 objects', () => {
+    const { added, removed } = calculateObjectDiff({ alpha: 1, beta: 2 }, { alpha: 1, gamma: 3 });
+    expect(added).toEqual({ gamma: 3 });
+    expect(removed).toEqual({ beta: 2 });
+  });
+
+  it('should work on nested objects', () => {
+    const { added, removed } = calculateObjectDiff(
+      { alpha: 1, beta: { gamma: 2, delta: { sigma: 7 } } },
+      { alpha: 1, beta: { gamma: 2, eta: 4 } }
+    );
+
+    expect(added).toEqual({ beta: { eta: 4 } });
+    expect(removed).toEqual({ beta: { delta: { sigma: 7 } } });
+  });
+
+  it('should return empty added/removed when the objects are the same', () => {
+    const { added, removed } = calculateObjectDiff({ alpha: 1, beta: 2 }, { alpha: 1, beta: 2 });
+    expect(added).toEqual({});
+    expect(removed).toEqual({});
+  });
+});

src/platform/packages/shared/kbn-object-utils/src/calculate_object_diff.ts

@@ -0,0 +1,75 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { isEmpty, isPlainObject } from 'lodash';
+
+interface Obj {
+  [key: PropertyKey]: Obj | unknown;
+}
+
+type DeepPartial<TInputObj> = {
+  [Prop in keyof TInputObj]?: TInputObj[Prop] extends Obj
+    ? DeepPartial<TInputObj[Prop]>
+    : TInputObj[Prop];
+};
+
+interface ObjectDiffResult<TBase, TCompare> {
+  added: DeepPartial<TCompare>;
+  removed: DeepPartial<TBase>;
+}
+
+/**
+ * Compares two JSON objects and calculates the added and removed properties, including nested properties.
+ * @param oldObj - The base object.
+ * @param newObj - The comparison object.
+ * @returns An object containing added and removed properties.
+ */
+export function calculateObjectDiff<TBase extends Obj, TCompare extends Obj>(
+  oldObj: TBase,
+  newObj?: TCompare
+): ObjectDiffResult<TBase, TCompare> {
+  const added: DeepPartial<TCompare> = {};
+  const removed: DeepPartial<TBase> = {};
+
+  if (!newObj) return { added, removed };
+
+  function diffRecursive(
+    base: Obj,
+    compare: Obj,
+    addedMap: DeepPartial<Obj>,
+    removedMap: DeepPartial<Obj>
+  ): void {
+    for (const key in compare) {
+      if (!(key in base)) {
+        addedMap[key] = compare[key];
+      } else if (isPlainObject(base[key]) && isPlainObject(compare[key])) {
+        addedMap[key] = {};
+        removedMap[key] = {};
+        diffRecursive(
+          base[key] as Obj,
+          compare[key] as Obj,
+          addedMap[key] as Obj,
+          removedMap[key] as Obj
+        );
+        if (isEmpty(addedMap[key])) delete addedMap[key];
+        if (isEmpty(removedMap[key])) delete removedMap[key];
+      }
+    }
+
+    for (const key in base) {
+      if (!(key in compare)) {
+        removedMap[key] = base[key];
+      }
+    }
+  }
+
+  diffRecursive(oldObj, newObj, added, removed);
+
+  return { added, removed };
+}

src/platform/packages/shared/kbn-object-utils/src/flatten_object.test.ts

@@ -0,0 +1,32 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { flattenObject } from './flatten_object';
+
+describe('flattenObject', () => {
+  it('should flat gamma object properties', () => {
+    const flattened = flattenObject({
+      alpha: {
+        gamma: {
+          sigma: 1,
+        },
+        delta: {
+          sigma: 2,
+        },
+      },
+      beta: 3,
+    });
+
+    expect(flattened).toEqual({
+      'alpha.gamma.sigma': 1,
+      'alpha.delta.sigma': 2,
+      beta: 3,
+    });
+  });
+});

src/platform/packages/shared/kbn-object-utils/src/flatten_object.ts

@@ -0,0 +1,33 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { isPlainObject } from 'lodash';
+
+/**
+ * Returns a flattened version of the input object also accounting for nested properties.
+ * @param obj - The input object.
+ * @param parentKey - The initial key used for recursive flattening.
+ * @returns An object containing all the flattened properties.
+ */
+export function flattenObject(obj: Record<PropertyKey, unknown>, parentKey: string = '') {
+  const result: Record<PropertyKey, unknown> = {};
+
+  for (const key in obj) {
+    if (Object.hasOwn(obj, key)) {
+      const value = obj[key];
+      const newKey = parentKey ? `${parentKey}.${key}` : key;
+      if (isPlainObject(value)) {
+        Object.assign(result, flattenObject(value as Record<PropertyKey, unknown>, newKey));
+      } else {
+        result[newKey] = value;
+      }
+    }
+  }
+  return result;
+}

src/platform/packages/shared/kbn-object-utils/tsconfig.json

@@ -0,0 +1,10 @@
+{
+  "compilerOptions": {
+    "outDir": "target/types",
+    "types": ["jest", "node"]
+  },
+  "exclude": ["target/**/*"],
+  "extends": "../../../../../tsconfig.base.json",
+  "include": ["**/*.ts"],
+  "kbn_references": []
+}

tsconfig.base.json

@@ -1308,6 +1308,8 @@
       "@kbn/no-data-page-plugin/*": ["src/platform/plugins/private/no_data_page/*"],
       "@kbn/notifications-plugin": ["x-pack/platform/plugins/shared/notifications"],
       "@kbn/notifications-plugin/*": ["x-pack/platform/plugins/shared/notifications/*"],
+      "@kbn/object-utils": ["src/platform/packages/shared/kbn-object-utils"],
+      "@kbn/object-utils/*": ["src/platform/packages/shared/kbn-object-utils/*"],
       "@kbn/object-versioning": ["src/platform/packages/shared/kbn-object-versioning"],
       "@kbn/object-versioning/*": ["src/platform/packages/shared/kbn-object-versioning/*"],
       "@kbn/object-versioning-utils": ["src/platform/packages/shared/kbn-object-versioning-utils"],

x-pack/packages/kbn-streams-schema/src/apis/list_streams_response.ts

@@ -5,10 +5,10 @@
  * 2.0.
  */
 import { z } from '@kbn/zod';
-import { streamDefintionSchema } from '../models';
+import { streamDefinitionSchema } from '../models';
 
 export const listStreamsResponseSchema = z.object({
-  streams: z.array(streamDefintionSchema),
+  streams: z.array(streamDefinitionSchema),
 });
 
 export type ListStreamsResponse = z.infer<typeof listStreamsResponseSchema>;

x-pack/packages/kbn-streams-schema/src/helpers/index.ts

@@ -5,4 +5,5 @@
  * 2.0.
  */
 
+export * from './processing';
 export * from './type_guards';

x-pack/packages/kbn-streams-schema/src/helpers/processing.ts

@@ -0,0 +1,38 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { Condition, ProcessingDefinition } from '../models';
+import {
+  isGrokProcessor,
+  isDissectProcessor,
+  isFilterCondition,
+  isAndCondition,
+  isOrCondition,
+} from './type_guards';
+
+export function getProcessorType(processor: ProcessingDefinition) {
+  if (isGrokProcessor(processor.config)) {
+    return 'grok';
+  }
+  if (isDissectProcessor(processor.config)) {
+    return 'dissect';
+  }
+  throw new Error('Unknown processor type');
+}
+
+export function isCompleteCondition(condition: Condition): boolean {
+  if (isFilterCondition(condition)) {
+    return condition.field !== undefined && condition.field !== '';
+  }
+  if (isAndCondition(condition)) {
+    return condition.and.every(isCompleteCondition);
+  }
+  if (isOrCondition(condition)) {
+    return condition.or.every(isCompleteCondition);
+  }
+  return false;
+}