Prototype Events identity with specification-owned fan-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

@@ -2198,6 +2198,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([
+        {
+          handle: 'order-notifier',
+          events: {
+            api_version: '2024-01',
+            subscription: {
+              topic: 'orders/create',
+              actions: ['create'],
+              uri: 'https://example.com/events/orders',
+            },
+          },
+        },
+        {
+          handle: 'product-sync',
+          events: {
+            api_version: '2024-01',
+            subscription: {
+              topic: 'products/update',
+              actions: ['update'],
+              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({
+      handle: 'order-notifier',
+      events: {
+        api_version: '2024-01',
+        subscription: {
+          topic: 'orders/create',
+          actions: ['create'],
+          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

@@ -1,5 +1,6 @@
 import {BaseConfigType, MAX_EXTENSION_HANDLE_LENGTH, MAX_UID_LENGTH} from './schemas.js'
 import {FunctionConfigType} from './specifications/function.js'
+import {eventSubscriptionHandle} from './specifications/transform/app_config_events.js'
 import {
   DevSessionUpdateContext,
   DevSessionWatchConfig,
@@ -522,7 +523,18 @@ export class ExtensionInstance<TConfiguration extends BaseConfigType = BaseConfi
     }
   }
 
+  private singleEventSubscriptionHandle(): string | undefined {
+    if (this.specification.identifier !== 'events') return undefined
+    const subscription = getPathValue<unknown>(this.configuration, 'events.subscription')
+    if (subscription == null || Array.isArray(subscription)) return undefined
+    const handle = typeof subscription === 'object' ? getPathValue<unknown>(subscription, 'handle') : undefined
+    return eventSubscriptionHandle(this.configuration.handle ?? handle)
+  }
+
   private buildHandle() {
+    const eventSubscriptionHandle = this.singleEventSubscriptionHandle()
+    if (eventSubscriptionHandle) return eventSubscriptionHandle
+
     switch (this.specification.uidStrategy) {
       case 'single':
         return this.specification.identifier
@@ -541,6 +553,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/models/extensions/specification.ts

@@ -30,6 +30,11 @@ export interface CustomTransformationConfig {
   reverse?: (obj: object, options?: {flags?: Flag[]}) => object
 }
 
+export interface ConfigurationModule {
+  readonly handle: string
+  readonly config: object
+}
+
 type ExtensionExperience = 'extension' | 'configuration'
 
 export function isAppConfigSpecification(spec: {experience: string}): boolean {
@@ -118,6 +123,13 @@ export interface ExtensionSpecification<TConfiguration extends BaseConfigType =
    */
   transformRemoteToLocal?: (remoteContent: object, options?: {flags?: Flag[]}) => object
 
+  /**
+   * Reconstruct a complete app-config section from this specification's modules, in input order.
+   * The specification must exclusively own the sections it emits. This replaces scalar readback;
+   * its result is merged at the first occurrence without reordering other specifications.
+   */
+  aggregateModuleConfigurations?: (modules: ReadonlyArray<ConfigurationModule>, options?: {flags?: Flag[]}) => object
+
   uidStrategy: UidStrategy
 
   /**
@@ -276,6 +288,7 @@ export function createConfigExtensionSpecification<TConfiguration extends BaseCo
   clientSteps?: ClientSteps
   appModuleFeatures?: (config?: TConfiguration) => ExtensionFeature[]
   transformConfig: TransformationConfig | CustomTransformationConfig
+  aggregateModuleConfigurations?: ExtensionSpecification<TConfiguration>['aggregateModuleConfigurations']
   uidStrategy?: UidStrategy
   getDevSessionUpdateMessages?: (config: TConfiguration, context: DevSessionUpdateContext) => Promise<string[]>
   patchWithAppDevURLs?: (config: TConfiguration, urls: ApplicationURLs) => void
@@ -288,7 +301,10 @@ export function createConfigExtensionSpecification<TConfiguration extends BaseCo
     schema: spec.schema,
     appModuleFeatures,
     transformLocalToRemote: resolveAppConfigTransform(spec.transformConfig),
-    transformRemoteToLocal: resolveReverseAppConfigTransform(spec.schema, spec.transformConfig),
+    transformRemoteToLocal: spec.aggregateModuleConfigurations
+      ? undefined
+      : resolveReverseAppConfigTransform(spec.schema, spec.transformConfig),
+    aggregateModuleConfigurations: spec.aggregateModuleConfigurations,
     experience: 'configuration',
     uidStrategy: spec.uidStrategy ?? 'single',
     clientSteps: spec.clientSteps,