move stream length validity check to wired stream class

Author: couvq

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

@@ -11,15 +11,13 @@ import type {
   IndicesDataStream,
   IndicesGetDataStreamResponse,
   QueryDslQueryContainer,
-  Result,
 } from '@elastic/elasticsearch/lib/api/types';
-import type { IScopedClusterClient, Logger, KibanaRequest } from '@kbn/core/server';
+import type { IScopedClusterClient, KibanaRequest, Logger } from '@kbn/core/server';
 import { isNotFoundError } from '@kbn/es-errors';
-import type { RoutingStatus } from '@kbn/streams-schema';
-import { Streams, getAncestors, getParentId } from '@kbn/streams-schema';
 import type { LockManagerService } from '@kbn/lock-manager';
 import type { Condition } from '@kbn/streamlang';
-import { MAX_STREAM_NAME_LENGTH } from '../../../common/constants';
+import type { RoutingStatus } from '@kbn/streams-schema';
+import { Streams, getAncestors, getParentId } from '@kbn/streams-schema';
 import type { AssetClient } from './assets/asset_client';
 import { ASSET_ID, ASSET_TYPE } from './assets/fields';
 import type { QueryClient } from './assets/query/query_client';
@@ -29,18 +27,12 @@ import {
 } from './errors/definition_not_found_error';
 import { SecurityError } from './errors/security_error';
 import { StatusError } from './errors/status_error';
+import { StreamsStatusConflictError } from './errors/streams_status_conflict_error';
+import type { FeatureClient } from './feature/feature_client';
 import { LOGS_ROOT_STREAM_NAME, rootStreamDefinition } from './root_stream_definition';
-import type { StreamsStorageClient } from './storage/streams_storage_client';
 import { State } from './state_management/state';
+import type { StreamsStorageClient } from './storage/streams_storage_client';
 import { checkAccess, checkAccessBulk } from './stream_crud';
-import { StreamsStatusConflictError } from './errors/streams_status_conflict_error';
-import type { FeatureClient } from './feature/feature_client';
-interface AcknowledgeResponse<TResult extends Result> {
-  acknowledged: true;
-  result: TResult;
-}
-
-export type EnableStreamsResponse = AcknowledgeResponse<'noop' | 'created'>;
 export type DisableStreamsResponse = AcknowledgeResponse<'noop' | 'deleted'>;
 export type DeleteStreamResponse = AcknowledgeResponse<'noop' | 'deleted'>;
 export type SyncStreamResponse = AcknowledgeResponse<'updated' | 'created'>;
@@ -324,18 +316,6 @@ export class StreamsClient {
       throw new StatusError(`Child stream ${name} already exists`, 409);
     }
 
-    const prefix = parent + '.';
-    if (name.length <= prefix.length) {
-      throw new StatusError('Stream name must not be empty.', 400);
-    }
-
-    if (name.length > MAX_STREAM_NAME_LENGTH) {
-      throw new StatusError(
-        `Stream name cannot be longer than ${MAX_STREAM_NAME_LENGTH} characters.`,
-        400
-      );
-    }
-
     await State.attemptChanges(
       [
         {

x-pack/platform/plugins/shared/streams/server/lib/streams/state_management/streams/wired_stream.ts

@@ -25,6 +25,7 @@ import {
   isIlmLifecycle,
 } from '@kbn/streams-schema';
 import _, { cloneDeep } from 'lodash';
+import { MAX_STREAM_NAME_LENGTH } from '../../../../../common/constants';
 import { generateLayer } from '../../component_templates/generate_layer';
 import { getComponentTemplateName } from '../../component_templates/name';
 import { isDefinitionNotFoundError } from '../../errors/definition_not_found_error';
@@ -355,7 +356,22 @@ export class WiredStream extends StreamActiveRecord<Streams.WiredStream.Definiti
 
     // validate routing
     const children: Set<string> = new Set();
+    const prefix = this.definition.name + '.';
     for (const routing of this._definition.ingest.wired.routing) {
+      if (routing.destination.length <= prefix.length) {
+        return {
+          isValid: false,
+          errors: [new Error(`Stream name must not be empty.`)],
+        };
+      }
+      if (routing.destination.length > MAX_STREAM_NAME_LENGTH) {
+        return {
+          isValid: false,
+          errors: [
+            new Error(`Stream name cannot be longer than ${MAX_STREAM_NAME_LENGTH} characters.`),
+          ],
+        };
+      }
       if (children.has(routing.destination)) {
         return {
           isValid: false,