[Streams 🌊] Introduce GroupStreams

api_docs/kbn_streams_schema.devdocs.json

@@ -1593,10 +1593,10 @@
       },
       {
         "parentPluginId": "@kbn/streams-schema",
-        "id": "def-common.isUnWiredStreamGetResponse",
+        "id": "def-common.isUnwiredStreamGetResponse",
         "type": "Function",
         "tags": [],
-        "label": "isUnWiredStreamGetResponse",
+        "label": "isUnwiredStreamGetResponse",
         "description": [],
         "signature": [
           "<TValue extends ",
@@ -1624,7 +1624,7 @@
         "children": [
           {
             "parentPluginId": "@kbn/streams-schema",
-            "id": "def-common.isUnWiredStreamGetResponse.$1",
+            "id": "def-common.isUnwiredStreamGetResponse.$1",
             "type": "Uncategorized",
             "tags": [],
             "label": "value",
@@ -5016,4 +5016,4 @@
       }
     ]
   }
-}
+}

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/api.ts

@@ -7,13 +7,55 @@
 
 import { z } from '@kbn/zod';
 import {
+  ingestStreamGetResponseSchema,
   ingestStreamUpsertRequestSchema,
+  unwiredStreamGetResponseSchema,
+  wiredStreamGetResponseSchema,
   type IngestStreamGetResponse,
   type IngestStreamUpsertRequest,
 } from './ingest';
+import {
+  GroupStreamGetResponse,
+  groupStreamGetResponseSchema,
+  GroupStreamUpsertRequest,
+  groupStreamUpsertRequestSchema,
+} from './group';
+import { createAsSchemaOrThrow, createIsNarrowSchema } from '../helpers';
+
+export const streamGetResponseSchema: z.Schema<StreamGetResponse> = z.union([
+  ingestStreamGetResponseSchema,
+  groupStreamGetResponseSchema,
+]);
+
+export const streamUpsertRequestSchema: z.Schema<StreamUpsertRequest> = z.union([
+  ingestStreamUpsertRequestSchema,
+  groupStreamUpsertRequestSchema,
+]);
+
+export const isWiredStreamGetResponse = createIsNarrowSchema(
+  streamGetResponseSchema,
+  wiredStreamGetResponseSchema
+);
+
+export const isUnwiredStreamGetResponse = createIsNarrowSchema(
+  streamGetResponseSchema,
+  unwiredStreamGetResponseSchema
+);
+
+export const asWiredStreamGetResponse = createAsSchemaOrThrow(
+  streamGetResponseSchema,
+  wiredStreamGetResponseSchema
+);
+
+export const asUnwiredStreamGetResponse = createAsSchemaOrThrow(
+  streamGetResponseSchema,
+  unwiredStreamGetResponseSchema
+);
 
-export const streamUpsertRequestSchema: z.Schema<StreamUpsertRequest> =
-  ingestStreamUpsertRequestSchema;
+export const asIngestStreamGetResponse = createAsSchemaOrThrow(
+  streamGetResponseSchema,
+  ingestStreamGetResponseSchema
+);
 
-export type StreamGetResponse = IngestStreamGetResponse;
-export type StreamUpsertRequest = IngestStreamUpsertRequest;
+export type StreamGetResponse = IngestStreamGetResponse | GroupStreamGetResponse;
+export type StreamUpsertRequest = IngestStreamUpsertRequest | GroupStreamUpsertRequest;

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/core.ts

@@ -8,9 +8,13 @@
 import { z } from '@kbn/zod';
 import { createIsNarrowSchema } from '../helpers';
 import { IngestStreamDefinition, ingestStreamDefinitionSchema } from './ingest';
+import { GroupStreamDefinition, groupStreamDefinitionSchema } from './group';
 
-export type StreamDefinition = IngestStreamDefinition;
+export type StreamDefinition = IngestStreamDefinition | GroupStreamDefinition;
 
-export const streamDefinitionSchema: z.Schema<StreamDefinition> = ingestStreamDefinitionSchema;
+export const streamDefinitionSchema: z.Schema<StreamDefinition> = z.union([
+  ingestStreamDefinitionSchema,
+  groupStreamDefinitionSchema,
+]);
 
 export const isStreamDefinition = createIsNarrowSchema(z.unknown(), streamDefinitionSchema);

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/group/api.ts

@@ -0,0 +1,49 @@
+/*
+ * 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 { z } from '@kbn/zod';
+import {
+  StreamGetResponseBase,
+  streamGetResponseSchemaBase,
+  StreamUpsertRequestBase,
+  streamUpsertRequestSchemaBase,
+} from '../base/api';
+import { GroupStreamDefinitionBase, groupStreamDefinitionBaseSchema } from './base';
+
+/**
+ * Group get response
+ */
+interface GroupStreamGetResponse extends StreamGetResponseBase {
+  stream: GroupStreamDefinitionBase;
+}
+
+const groupStreamGetResponseSchema: z.Schema<GroupStreamGetResponse> = z.intersection(
+  streamGetResponseSchemaBase,
+  z.object({
+    stream: groupStreamDefinitionBaseSchema,
+  })
+);
+
+/**
+ * Group upsert request
+ */
+interface GroupStreamUpsertRequest extends StreamUpsertRequestBase {
+  stream: GroupStreamDefinitionBase;
+}
+
+const groupStreamUpsertRequestSchema: z.Schema<GroupStreamUpsertRequest> = z.intersection(
+  streamUpsertRequestSchemaBase,
+  z.object({
+    stream: groupStreamDefinitionBaseSchema,
+  })
+);
+
+export {
+  type GroupStreamGetResponse,
+  type GroupStreamUpsertRequest,
+  groupStreamGetResponseSchema,
+  groupStreamUpsertRequestSchema,
+};

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/group/base.ts

@@ -0,0 +1,44 @@
+/*
+ * 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 { z } from '@kbn/zod';
+import { NonEmptyString } from '@kbn/zod-helpers';
+import { StreamDefinitionBase } from '../base';
+
+interface GroupBase {
+  description?: string;
+  members: string[];
+}
+
+const groupBaseSchema: z.Schema<GroupBase> = z.object({
+  description: z.optional(z.string()),
+  members: z.array(NonEmptyString),
+});
+
+interface GroupStreamDefinitionBase {
+  group: GroupBase;
+}
+
+const groupStreamDefinitionBaseSchema: z.Schema<GroupStreamDefinitionBase> = z.object({
+  group: groupBaseSchema,
+});
+
+type GroupStreamDefinition = StreamDefinitionBase & GroupStreamDefinitionBase;
+
+const groupStreamDefinitionSchema: z.Schema<GroupStreamDefinition> = z.intersection(
+  z.object({ name: NonEmptyString }),
+  groupStreamDefinitionBaseSchema
+);
+
+export {
+  type GroupBase,
+  type GroupStreamDefinitionBase,
+  type GroupStreamDefinition,
+  groupBaseSchema,
+  groupStreamDefinitionBaseSchema,
+  groupStreamDefinitionSchema,
+};

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/group/index.ts

@@ -0,0 +1,9 @@
+/*
+ * 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.
+ */
+
+export * from './base';
+export * from './api';

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/helpers.ts

@@ -4,9 +4,10 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-
-import { createIsNarrowSchema } from '../helpers';
+import { z } from '@kbn/zod';
+import { createAsSchemaOrThrow, createIsNarrowSchema } from '../helpers';
 import { streamDefinitionSchema } from './core';
+import { groupStreamDefinitionBaseSchema, groupStreamDefinitionSchema } from './group';
 import {
   ingestStreamDefinitionSchema,
   unwiredStreamDefinitionSchema,
@@ -23,11 +24,31 @@ export const isWiredStreamDefinition = createIsNarrowSchema(
   wiredStreamDefinitionSchema
 );
 
+export const asIngestStreamDefinition = createAsSchemaOrThrow(
+  streamDefinitionSchema,
+  ingestStreamDefinitionSchema
+);
+
+export const asWiredStreamDefinition = createAsSchemaOrThrow(
+  streamDefinitionSchema,
+  wiredStreamDefinitionSchema
+);
+
 export const isUnwiredStreamDefinition = createIsNarrowSchema(
   streamDefinitionSchema,
   unwiredStreamDefinitionSchema
 );
 
+export const isGroupStreamDefinition = createIsNarrowSchema(
+  streamDefinitionSchema,
+  groupStreamDefinitionSchema
+);
+
+export const isGroupStreamDefinitionBase = createIsNarrowSchema(
+  z.unknown(),
+  groupStreamDefinitionBaseSchema
+);
+
 export const isRootStreamDefinition = createIsNarrowSchema(
   streamDefinitionSchema,
   wiredStreamDefinitionSchema.refine((stream) => {

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/index.ts

@@ -11,3 +11,4 @@ export * from './legacy';
 export * from './api';
 export * from './core';
 export * from './helpers';
+export * from './group';

x-pack/solutions/observability/packages/kbn-streams-schema/src/models/ingest/api.ts

@@ -26,7 +26,6 @@ import {
   wiredStreamDefinitionSchemaBase,
 } from './base';
 import { ElasticsearchAsset, elasticsearchAssetSchema } from './common';
-import { createIsNarrowSchema, createAsSchemaOrThrow } from '../../helpers';
 import {
   UnwiredIngestStreamEffectiveLifecycle,
   WiredIngestStreamEffectiveLifecycle,
@@ -146,33 +145,12 @@ const ingestStreamGetResponseSchema: z.Schema<IngestStreamGetResponse> = z.union
   unwiredStreamGetResponseSchema,
 ]);
 
-const isWiredStreamGetResponse = createIsNarrowSchema(
-  ingestStreamGetResponseSchema,
-  wiredStreamGetResponseSchema
-);
-
-const isUnWiredStreamGetResponse = createIsNarrowSchema(
-  ingestStreamGetResponseSchema,
-  unwiredStreamGetResponseSchema
-);
-
-const asWiredStreamGetResponse = createAsSchemaOrThrow(
-  ingestStreamGetResponseSchema,
-  wiredStreamGetResponseSchema
-);
-
-const asUnwiredStreamGetResponse = createAsSchemaOrThrow(
-  ingestStreamGetResponseSchema,
-  unwiredStreamGetResponseSchema
-);
-
 export {
   ingestStreamUpsertRequestSchema,
   ingestUpsertRequestSchema,
-  isWiredStreamGetResponse,
-  isUnWiredStreamGetResponse,
-  asWiredStreamGetResponse,
-  asUnwiredStreamGetResponse,
+  ingestStreamGetResponseSchema,
+  wiredStreamGetResponseSchema,
+  unwiredStreamGetResponseSchema,
   type IngestGetResponse,
   type IngestStreamGetResponse,
   type IngestStreamUpsertRequest,