[Streams 🌊] Ensure the members array is unique for GroupStreamDefinitions (#210089)

## Summary

This PR ensures that the `definition.group.members` is a unique array of
strings. I introduced a new private function to the StreamsClient called
`parseDefinition` that will parse the definition being upserted with the
runtime schemas to ensure they are properly formatted. This is also a
good extension point for doing any transformations we need.

x-pack/solutions/observability/plugins/streams/server/routes/streams/crud/route.ts

@@ -12,6 +12,7 @@ import {
   StreamGetResponse,
   isWiredStreamDefinition,
   streamUpsertRequestSchema,
+  isGroupStreamDefinitionBase,
 } from '@kbn/streams-schema';
 import { z } from '@kbn/zod';
 import { badData, badRequest } from '@hapi/boom';
@@ -163,8 +164,20 @@ export const editStreamRoute = createServerRoute({
       throw badRequest('Cannot create wired stream due to unsupported root stream');
     }
 
+    const body = isGroupStreamDefinitionBase(params.body.stream)
+      ? {
+          ...params.body,
+          stream: {
+            group: {
+              ...params.body.stream.group,
+              members: Array.from(new Set(params.body.stream.group.members)),
+            },
+          },
+        }
+      : params.body;
+
     return await streamsClient.upsertStream({
-      request: params.body,
+      request: body,
       name: params.path.name,
     });
   },

x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/group_streams.ts

@@ -40,7 +40,7 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
               body: {
                 stream: {
                   group: {
-                    members: ['logs', 'logs.test2'],
+                    members: ['logs', 'logs.test2', 'logs'],
                   },
                 },
                 dashboards: [],