Allow kafka:// URIs in webhook subscription TOML validation

packages/app/src/cli/models/app/loader.test.ts

@@ -2815,7 +2815,7 @@ describe('WebhooksSchema', () => {
     const errorObj = {
       code: zod.ZodIssueCode.custom,
       message:
-        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id} or Eventbridge ARN",
+        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id}, kafka://{topic-id} or Eventbridge ARN",
       path: ['webhooks', 'subscriptions', 0, 'uri'],
     }
 
@@ -2835,15 +2835,15 @@ describe('WebhooksSchema', () => {
     expect(result.parsedConfiguration.webhooks).toMatchObject(webhookConfig)
   })
 
-  test('throws an error if uri is not a valid https URL, pubsub URI, or Eventbridge ARN', async () => {
+  test('throws an error if uri is not a valid https URL, pubsub URI, kafka URI, or Eventbridge ARN', async () => {
     const webhookConfig: WebhooksConfig = {
       api_version: '2021-07',
       subscriptions: [{uri: 'my::URI-thing::Shopify::123', topics: ['products/create']}],
     }
     const errorObj = {
       code: zod.ZodIssueCode.custom,
       message:
-        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id} or Eventbridge ARN",
+        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id}, kafka://{topic-id} or Eventbridge ARN",
       path: ['webhooks', 'subscriptions', 0, 'uri'],
     }
 
@@ -2889,6 +2889,33 @@ describe('WebhooksSchema', () => {
     expect(result.parsedConfiguration.webhooks).toMatchObject(webhookConfig)
   })
 
+  test('accepts a kafka uri', async () => {
+    const webhookConfig: WebhooksConfig = {
+      api_version: '2021-07',
+      subscriptions: [{uri: 'kafka://my_topic.name-1', topics: ['products/create']}],
+    }
+
+    const result = await setupParsing({}, webhookConfig)
+    expect(result.threw).toBe(false)
+    expect(result.parsedConfiguration.webhooks).toMatchObject(webhookConfig)
+  })
+
+  test('throws an error if kafka topic contains invalid characters', async () => {
+    const webhookConfig: WebhooksConfig = {
+      api_version: '2021-07',
+      subscriptions: [{uri: 'kafka://invalid topic!', topics: ['products/create']}],
+    }
+    const errorObj = {
+      code: zod.ZodIssueCode.custom,
+      message:
+        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id}, kafka://{topic-id} or Eventbridge ARN",
+      path: ['webhooks', 'subscriptions', 0, 'uri'],
+    }
+
+    const result = await setupParsing(errorObj, webhookConfig)
+    expect(result.threw).toBe(true)
+  })
+
   test('accepts combination of uris', async () => {
     const webhookConfig: WebhooksConfig = {
       api_version: '2021-07',
@@ -2905,6 +2932,10 @@ describe('WebhooksSchema', () => {
           uri: 'pubsub://my-project-123:my-topic',
           topics: ['products/create', 'products/update'],
         },
+        {
+          uri: 'kafka://my-topic',
+          topics: ['products/create', 'products/update'],
+        },
       ],
     }
 
@@ -2927,6 +2958,14 @@ describe('WebhooksSchema', () => {
           uri: 'https://example.com',
           topics: ['products/update'],
         },
+        {
+          uri: 'kafka://my-topic',
+          topics: ['products/create'],
+        },
+        {
+          uri: 'kafka://my-topic',
+          topics: ['products/update'],
+        },
         {
           uri: 'pubsub://my-project-123:my-topic',
           topics: ['products/create'],
@@ -2997,7 +3036,7 @@ describe('WebhooksSchema', () => {
     expect(result.parsedConfiguration.webhooks).toMatchObject(webhookConfig)
   })
 
-  test('throws an error if uri is not an https uri', async () => {
+  test('throws an error for http:// uri in subscription (rejects insecure scheme)', async () => {
     const webhookConfig: WebhooksConfig = {
       api_version: '2021-07',
       subscriptions: [
@@ -3010,7 +3049,7 @@ describe('WebhooksSchema', () => {
     const errorObj = {
       code: zod.ZodIssueCode.custom,
       message:
-        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id} or Eventbridge ARN",
+        "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id}, kafka://{topic-id} or Eventbridge ARN",
       path: ['webhooks', 'subscriptions', 0, 'uri'],
     }
 

packages/app/src/cli/models/extensions/specifications/validation/common.ts

@@ -6,6 +6,8 @@ const pubSubRegex = /^pubsub:\/\/(?<gcp_project_id>[^:]+):(?<gcp_topic>.+)$/
 // example Eventbridge ARN - arn:aws:events:{region}::event-source/aws.partner/shopify.com/{app_id}/{path}
 const arnRegex =
   /^arn:aws:events:(?<aws_region>[a-z]{2}-[a-z]+-[0-9]+)::event-source\/aws\.partner\/shopify\.com(\.test)?\/(?<api_client_id>\d+)\/(?<event_source_name>.+)$/
+// example Kafka URI - kafka://{topic}. Restricted to internal apps; the platform enforces authorization.
+const kafkaRegex = /^kafka:\/\/(?<kafka_topic>[a-zA-Z0-9_.-]+)$/
 
 export function removeTrailingSlash(arg: string): string
 export function removeTrailingSlash(arg: unknown): unknown {
@@ -16,10 +18,10 @@ export const WebhookSubscriptionUriValidation = zod.string({invalid_type_error:
   (uri) => {
     if (uri.startsWith('/')) return true
 
-    return httpsRegex.test(uri) || pubSubRegex.test(uri) || arnRegex.test(uri)
+    return httpsRegex.test(uri) || pubSubRegex.test(uri) || arnRegex.test(uri) || kafkaRegex.test(uri)
   },
   {
     message:
-      "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id} or Eventbridge ARN",
+      "URI format isn't correct. Valid formats include: relative path starting with a slash, HTTPS URL, pubsub://{project-id}:{topic-id}, kafka://{topic-id} or Eventbridge ARN",
   },
 )