Fan out one events module per subscription behind an organization flag or environment opt-in

.changeset/events-subscription-fanout.md

@@ -0,0 +1,5 @@
+---
+'@shopify/app': patch
+---
+
+Fan out one events module per subscription, gated by an organization flag or SHOPIFY_CLI_EVENTS_SUBSCRIPTION_FANOUT

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

@@ -2174,6 +2174,217 @@ describe('load', () => {
     ])
   })
 
+  test('fans out one events module per subscription when the fan-out is enabled', async () => {
+    // Given
+    const appConfigurationWithEvents = `
+    name = "for-testing-events"
+    client_id = "1234567890"
+    application_url = "https://example.com/lala"
+    embedded = true
+
+    [build]
+    include_config_on_deploy = true
+
+    [webhooks]
+    api_version = "2024-01"
+
+    [auth]
+    redirect_urls = [ "https://example.com/api/auth" ]
+
+    [events]
+    api_version = "2024-01"
+
+    [[events.subscription]]
+      topic = "orders/create"
+      actions = ["create"]
+      handle = "order-notifier"
+      uri = "https://example.com/events/orders"
+
+    [[events.subscription]]
+      topic = "products/update"
+      actions = ["update"]
+      handle = "product-sync"
+      uri = "https://example.com/events/products"
+    `
+    await writeConfig(appConfigurationWithEvents)
+    process.env.SHOPIFY_CLI_EVENTS_SUBSCRIPTION_FANOUT = '1'
+
+    try {
+      // When
+      const app = await loadTestingApp({remoteFlags: []})
+
+      // Then
+      const eventsExtensions = app.allExtensions.filter((ext) => ext.specification.identifier === 'events')
+      expect(eventsExtensions).toHaveLength(2)
+      expect(eventsExtensions.map((ext) => ext.configuration)).toEqual([
+        {
+          events: {
+            api_version: '2024-01',
+            subscription: {
+              topic: 'orders/create',
+              actions: ['create'],
+              handle: 'order-notifier',
+              uri: 'https://example.com/events/orders',
+            },
+          },
+        },
+        {
+          events: {
+            api_version: '2024-01',
+            subscription: {
+              topic: 'products/update',
+              actions: ['update'],
+              handle: 'product-sync',
+              uri: 'https://example.com/events/products',
+            },
+          },
+        },
+      ])
+      expect(eventsExtensions.map((ext) => ext.handle)).toEqual(['order-notifier', 'product-sync'])
+      expect(eventsExtensions.map((ext) => ext.uid)).toEqual(['order-notifier', 'product-sync'])
+    } finally {
+      delete process.env.SHOPIFY_CLI_EVENTS_SUBSCRIPTION_FANOUT
+    }
+  })
+
+  test('fans out events modules when the remote flag is enabled without the environment opt-in', async () => {
+    // Given
+    const appConfigurationWithEvents = `
+    name = "for-testing-events"
+    client_id = "1234567890"
+    application_url = "https://example.com/lala"
+    embedded = true
+
+    [webhooks]
+    api_version = "2024-01"
+
+    [auth]
+    redirect_urls = [ "https://example.com/api/auth" ]
+
+    [events]
+    api_version = "2024-01"
+
+    [[events.subscription]]
+      topic = "orders/create"
+      actions = ["create"]
+      handle = "order-notifier"
+      uri = "https://example.com/events/orders"
+    `
+    await writeConfig(appConfigurationWithEvents)
+
+    // When
+    const app = await loadTestingApp({remoteFlags: [Flag.SingleSubscriptionEventsModules]})
+
+    // Then
+    const eventsExtensions = app.allExtensions.filter((ext) => ext.specification.identifier === 'events')
+    expect(eventsExtensions).toHaveLength(1)
+    expect(eventsExtensions[0]!.configuration).toEqual({
+      events: {
+        api_version: '2024-01',
+        subscription: {
+          topic: 'orders/create',
+          actions: ['create'],
+          handle: 'order-notifier',
+          uri: 'https://example.com/events/orders',
+        },
+      },
+    })
+    expect(eventsExtensions[0]!.handle).toEqual('order-notifier')
+  })
+
+  test('loads a single events module with the subscription list when the fan-out is disabled', async () => {
+    // Given
+    const appConfigurationWithEvents = `
+    name = "for-testing-events"
+    client_id = "1234567890"
+    application_url = "https://example.com/lala"
+    embedded = true
+
+    [webhooks]
+    api_version = "2024-01"
+
+    [auth]
+    redirect_urls = [ "https://example.com/api/auth" ]
+
+    [events]
+    api_version = "2024-01"
+
+    [[events.subscription]]
+      topic = "orders/create"
+      actions = ["create"]
+      handle = "order-notifier"
+      uri = "https://example.com/events/orders"
+    `
+    await writeConfig(appConfigurationWithEvents)
+
+    // When
+    const app = await loadTestingApp({remoteFlags: []})
+
+    // Then
+    const eventsExtensions = app.allExtensions.filter((ext) => ext.specification.identifier === 'events')
+    expect(eventsExtensions).toHaveLength(1)
+    expect(eventsExtensions[0]!.configuration).toMatchObject({
+      events: {
+        api_version: '2024-01',
+        subscription: [
+          {
+            topic: 'orders/create',
+            actions: ['create'],
+            handle: 'order-notifier',
+            uri: 'https://example.com/events/orders',
+          },
+        ],
+      },
+    })
+  })
+
+  test('rejects duplicate event subscription handles when the fan-out is enabled', async () => {
+    // Given
+    const appConfigurationWithEvents = `
+    name = "for-testing-events"
+    client_id = "1234567890"
+    application_url = "https://example.com/lala"
+    embedded = true
+
+    [webhooks]
+    api_version = "2024-01"
+
+    [auth]
+    redirect_urls = [ "https://example.com/api/auth" ]
+
+    [events]
+    api_version = "2024-01"
+
+    [[events.subscription]]
+      topic = "orders/create"
+      actions = ["create"]
+      handle = "order-notifier"
+      uri = "https://example.com/events/orders"
+
+    [[events.subscription]]
+      topic = "products/update"
+      actions = ["update"]
+      handle = "order-notifier"
+      uri = "https://example.com/events/products"
+    `
+    await writeConfig(appConfigurationWithEvents)
+    process.env.SHOPIFY_CLI_EVENTS_SUBSCRIPTION_FANOUT = '1'
+
+    try {
+      // When
+      const app = await loadTestingApp({remoteFlags: []})
+
+      // Then
+      const errorMessages = app.errors
+        .getErrors()
+        .map((error) => error.message)
+        .join('\n')
+      expect(errorMessages).toContain('Duplicated handle "order-notifier"')
+    } finally {
+      delete process.env.SHOPIFY_CLI_EVENTS_SUBSCRIPTION_FANOUT
+    }
+  })
+
   test('loads the app with several functions that have valid configurations', async () => {
     // Given
     await writeConfig(appConfiguration)

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

@@ -26,6 +26,7 @@ import {ExtensionSpecification, isAppConfigSpecification} from '../extensions/sp
 import {CreateAppOptions, Flag} from '../../utilities/developer-platform-client.js'
 import {findConfigFiles} from '../../prompts/config.js'
 import {WebhookSubscriptionSpecIdentifier} from '../extensions/specifications/app_config_webhook_subscription.js'
+import {EventsSpecIdentifier} from '../extensions/specifications/app_config_events.js'
 import {WebhooksSchema} from '../extensions/specifications/app_config_webhook_schemas/webhooks_schema.js'
 import {ApplicationURLs, generateApplicationURLs} from '../../services/dev/urls.js'
 import {Project} from '../project/project.js'
@@ -49,6 +50,8 @@ import {AbortError} from '@shopify/cli-kit/node/error'
 import {outputContent, outputDebug, outputToken, stringifyMessage} from '@shopify/cli-kit/node/output'
 import {joinWithAnd} from '@shopify/cli-kit/common/string'
 import {getArrayRejectingUndefined} from '@shopify/cli-kit/common/array'
+import {getPathValue} from '@shopify/cli-kit/common/object'
+import {isTruthy} from '@shopify/cli-kit/node/context/utilities'
 import {showNotificationsIfNeeded} from '@shopify/cli-kit/node/notifications-system'
 import ignore from 'ignore'
 import type {ActiveConfig} from '../project/active-config.js'
@@ -787,11 +790,21 @@ class AppLoader<TConfig extends CurrentAppConfiguration, TModuleSpec extends Ext
           const specResult = parseConfigurationObjectAgainstSpecification(specification, configPath, appConfiguration)
           if (specResult.errors) {
             this.errors.addErrors(specResult.errors)
-            return [null, [] as string[]] as const
+            return [[], [] as string[]] as const
           }
           const specConfiguration = specResult.data
 
-          if (Object.keys(specConfiguration).length === 0) return [null, Object.keys(specConfiguration)] as const
+          if (Object.keys(specConfiguration).length === 0) return [[], Object.keys(specConfiguration)] as const
+
+          const eventSubscriptionInstances = await this.createEventSubscriptionInstances(
+            specification,
+            specConfiguration,
+            configPath,
+            directory,
+          )
+          if (eventSubscriptionInstances) {
+            return [eventSubscriptionInstances, Object.keys(specConfiguration)] as const
+          }
 
           const instance = await this.createExtensionInstance(
             specification.identifier,
@@ -805,7 +818,7 @@ class AppLoader<TConfig extends CurrentAppConfiguration, TModuleSpec extends Ext
               extensionInstance,
             ),
           )
-          return [instance, Object.keys(specConfiguration)] as const
+          return [[instance], Object.keys(specConfiguration)] as const
         }),
     )
 
@@ -824,9 +837,39 @@ class AppLoader<TConfig extends CurrentAppConfiguration, TModuleSpec extends Ext
         message: `Unsupported section(s) in app configuration: ${unusedKeys.sort().join(', ')}`,
       })
     }
-    return extensionInstancesWithKeys
-      .filter(([instance]) => instance)
-      .map(([instance]) => instance as ExtensionInstance)
+    return getArrayRejectingUndefined(extensionInstancesWithKeys.flatMap(([instances]) => instances))
+  }
+
+  private async createEventSubscriptionInstances(
+    specification: ExtensionSpecification,
+    specConfiguration: object,
+    configPath: string,
+    directory: string,
+  ): Promise<ExtensionInstance[] | undefined> {
+    if (specification.identifier !== EventsSpecIdentifier) return undefined
+    const fanoutEnabled =
+      this.remoteFlags.includes(Flag.SingleSubscriptionEventsModules) ||
+      isTruthy(process.env.SHOPIFY_CLI_EVENTS_SUBSCRIPTION_FANOUT)
+    if (!fanoutEnabled) return undefined
+
+    const events = getPathValue<{api_version?: string; subscription?: {[key: string]: unknown}[]}>(
+      specConfiguration,
+      'events',
+    )
+    const subscriptions = events?.subscription
+    if (!Array.isArray(subscriptions) || subscriptions.length === 0) return undefined
+
+    const instances = await Promise.all(
+      subscriptions.map(async (subscription) =>
+        this.createExtensionInstance(
+          specification.identifier,
+          {events: {api_version: events?.api_version, subscription}},
+          configPath,
+          directory,
+        ),
+      ),
+    )
+    return getArrayRejectingUndefined(instances)
   }
 
   private async validateConfigurationExtensionInstance(

packages/app/src/cli/models/extensions/extension-instance.ts

@@ -522,7 +522,17 @@ export class ExtensionInstance<TConfiguration extends BaseConfigType = BaseConfi
     }
   }
 
+  private singleEventSubscriptionHandle(): string | undefined {
+    if (this.specification.identifier !== 'events') return undefined
+    const subscription = getPathValue(this.configuration, 'events.subscription')
+    if (!subscription || Array.isArray(subscription)) return undefined
+    return getPathValue(subscription, 'handle')
+  }
+
   private buildHandle() {
+    const eventSubscriptionHandle = this.singleEventSubscriptionHandle()
+    if (eventSubscriptionHandle) return eventSubscriptionHandle
+
     switch (this.specification.uidStrategy) {
       case 'single':
         return this.specification.identifier
@@ -541,6 +551,9 @@ export class ExtensionInstance<TConfiguration extends BaseConfigType = BaseConfi
   }
 
   private buildUIDFromStrategy() {
+    const eventSubscriptionHandle = this.singleEventSubscriptionHandle()
+    if (eventSubscriptionHandle) return eventSubscriptionHandle
+
     switch (this.specification.uidStrategy) {
       case 'single':
         return this.specification.identifier

packages/app/src/cli/utilities/developer-platform-client.ts

@@ -141,7 +141,9 @@ export type AssetUrlSchema = WithUserErrors<{
   assetUrl?: string | null
 }>
 
-export enum Flag {}
+export enum Flag {
+  SingleSubscriptionEventsModules = 'single_subscription_events_modules',
+}
 
 const FlagMap: {[key: string]: Flag} = {}