fix(event-handler): fix decorated scope in appsync events

packages/event-handler/src/appsync-events/AppSyncEventsResolver.ts

@@ -5,6 +5,7 @@ import type {
   OnPublishHandlerAggregateFn,
   OnPublishHandlerFn,
   OnSubscribeHandler,
+  ResolveOptions,
 } from '../types/appsync-events.js';
 import { Router } from './Router.js';
 import { UnauthorizedException } from './errors.js';
@@ -67,7 +68,9 @@ class AppSyncEventsResolver extends Router {
    *   }
    *
    *   async handler(event, context) {
-   *     return app.resolve(event, context);
+   *     return app.resolve(event, context, {
+   *       scope: this, // bind decorated methods to the class instance
+   *     });
    *   }
    * }
    *
@@ -78,7 +81,11 @@ class AppSyncEventsResolver extends Router {
    * @param event - The incoming event from AppSync Events
    * @param context - The context object provided by AWS Lambda
    */
-  public async resolve(event: unknown, context: Context) {
+  public async resolve(
+    event: unknown,
+    context: Context,
+    options?: ResolveOptions
+  ) {
     if (!isAppSyncEventsEvent(event)) {
       this.logger.warn(
         'Received an event that is not compatible with this resolver'
@@ -87,11 +94,12 @@ class AppSyncEventsResolver extends Router {
     }
 
     if (isAppSyncEventsPublishEvent(event)) {
-      return await this.handleOnPublish(event, context);
+      return await this.handleOnPublish(event, context, options);
     }
     return await this.handleOnSubscribe(
       event as AppSyncEventsSubscribeEvent,
-      context
+      context,
+      options
     );
   }
 
@@ -100,10 +108,12 @@ class AppSyncEventsResolver extends Router {
    *
    * @param event - The incoming event from AppSync Events
    * @param context - The context object provided by AWS Lambda
+   * @param options - Optional resolve options
    */
   protected async handleOnPublish(
     event: AppSyncEventsPublishEvent,
-    context: Context
+    context: Context,
+    options?: ResolveOptions
   ) {
     const { path } = event.info.channel;
     const routeHandlerOptions = this.onPublishRegistry.resolve(path);
@@ -114,11 +124,10 @@ class AppSyncEventsResolver extends Router {
     if (aggregate) {
       try {
         return {
-          events: await (handler as OnPublishHandlerAggregateFn).apply(this, [
-            event.events,
-            event,
-            context,
-          ]),
+          events: await (handler as OnPublishHandlerAggregateFn).apply(
+            options?.scope ?? this,
+            [event.events, event, context]
+          ),
         };
       } catch (error) {
         this.logger.error(`An error occurred in handler ${path}`, error);
@@ -131,11 +140,10 @@ class AppSyncEventsResolver extends Router {
         event.events.map(async (message) => {
           const { id, payload } = message;
           try {
-            const result = await (handler as OnPublishHandlerFn).apply(this, [
-              payload,
-              event,
-              context,
-            ]);
+            const result = await (handler as OnPublishHandlerFn).apply(
+              options?.scope ?? this,
+              [payload, event, context]
+            );
             return {
               id,
               payload: result,
@@ -161,10 +169,12 @@ class AppSyncEventsResolver extends Router {
    *
    * @param event - The incoming event from AppSync Events
    * @param context - The context object provided by AWS Lambda
+   * @param options - Optional resolve options
    */
   protected async handleOnSubscribe(
     event: AppSyncEventsSubscribeEvent,
-    context: Context
+    context: Context,
+    options?: ResolveOptions
   ) {
     const { path } = event.info.channel;
     const routeHandlerOptions = this.onSubscribeRegistry.resolve(path);
@@ -173,7 +183,10 @@ class AppSyncEventsResolver extends Router {
     }
     const { handler } = routeHandlerOptions;
     try {
-      await (handler as OnSubscribeHandler).apply(this, [event, context]);
+      await (handler as OnSubscribeHandler).apply(options?.scope ?? this, [
+        event,
+        context,
+      ]);
     } catch (error) {
       this.logger.error(`An error occurred in handler ${path}`, error);
       if (error instanceof UnauthorizedException) throw error;

packages/event-handler/src/appsync-events/Router.ts

@@ -9,6 +9,9 @@ import type {
 } from '../types/appsync-events.js';
 import { RouteHandlerRegistry } from './RouteHandlerRegistry.js';
 
+// Simple global approach - store the last instance per router
+const routerInstanceMap = new WeakMap<Router, unknown>();
+
 /**
  * Class for registering routes for the `onPublish` and `onSubscribe` events in AWS AppSync Events APIs.
  */
@@ -194,11 +197,11 @@ class Router {
       return;
     }
 
-    return (_target, _propertyKey, descriptor: PropertyDescriptor) => {
+    return (target, _propertyKey, descriptor: PropertyDescriptor) => {
       const routeOptions = isRecord(handler) ? handler : options;
       this.onPublishRegistry.register({
         path,
-        handler: descriptor.value,
+        handler: descriptor?.value,
         aggregate: (routeOptions?.aggregate ?? false) as T,
       });
       return descriptor;
@@ -276,7 +279,7 @@ class Router {
     return (_target, _propertyKey, descriptor: PropertyDescriptor) => {
       this.onSubscribeRegistry.register({
         path,
-        handler: descriptor.value,
+        handler: descriptor?.value,
       });
       return descriptor;
     };

packages/event-handler/src/types/appsync-events.ts

@@ -1,8 +1,47 @@
 import type { Context } from 'aws-lambda';
+import type { AppSyncEventsResolver } from '../appsync-events/AppSyncEventsResolver.js';
 import type { RouteHandlerRegistry } from '../appsync-events/RouteHandlerRegistry.js';
 import type { Router } from '../appsync-events/Router.js';
 import type { Anything, GenericLogger } from './common.js';
 
+// #region resolve options
+
+/**
+ * Optional object to pass to the {@link AppSyncEventsResolver.resolve | `AppSyncEventsResolver.resolve()`} method.
+ */
+type ResolveOptions = {
+  /**
+   * Reference to `this` instance of the class that is calling the `resolve` method.
+   *
+   * This parameter should be used only when using {@link AppSyncEventsResolver.onPublish | `AppSyncEventsResolver.onPublish()`}
+   * and {@link AppSyncEventsResolver.onSubscribe | `AppSyncEventsResolver.onSubscribe()`} as class method decorators, and
+   * it's used to bind the decorated methods to your class instance.
+   *
+   * @example
+   * ```ts
+   * import { AppSyncEventsResolver } from '@aws-lambda-powertools/event-handler/appsync-events';
+   *
+   * const app = new AppSyncEventsResolver();
+   *
+   * class Lambda {
+   *   public scope = 'scoped';
+   *
+   *   ⁣@app.onPublish('/foo')
+   *   public async handleFoo(payload: string) {
+   *     return `${this.scope} ${payload}`;
+   *   }
+   *
+   *   public async handler(event: unknown, context: Context) {
+   *     return app.resolve(event, context, { scope: this });
+   *   }
+   * }
+   * const lambda = new Lambda();
+   * const handler = lambda.handler.bind(lambda);
+   * ```
+   */
+  scope?: unknown;
+};
+
 // #region OnPublish fn
 
 type OnPublishHandlerFn = (
@@ -17,11 +56,13 @@ type OnPublishHandlerSyncFn = (
   context: Context
 ) => unknown;
 
+type OnPublishAggregatePayload = Array<{
+  payload: Anything;
+  id: string;
+}>;
+
 type OnPublishHandlerAggregateFn = (
-  events: Array<{
-    payload: Anything;
-    id: string;
-  }>,
+  events: OnPublishAggregatePayload,
   event: AppSyncEventsPublishEvent,
   context: Context
 ) => Promise<unknown[]>;
@@ -294,8 +335,10 @@ export type {
   OnPublishHandlerSyncFn,
   OnPublishHandlerSyncAggregateFn,
   OnPublishHandlerAggregateFn,
+  OnPublishAggregatePayload,
   OnSubscribeHandler,
   OnPublishAggregateOutput,
   OnPublishEventPayload,
   OnPublishOutput,
+  ResolveOptions,
 };

packages/event-handler/src/types/index.ts

@@ -3,10 +3,12 @@ export type {
   AppSyncEventsPublishEvent,
   AppSyncEventsSubscribeEvent,
   OnPublishAggregateOutput,
+  OnPublishAggregatePayload,
   OnPublishEventPayload,
   OnPublishOutput,
   RouteOptions,
   RouterOptions,
+  ResolveOptions,
 } from './appsync-events.js';
 
 export type {

packages/event-handler/tests/unit/appsync-events/AppSyncEventsResolver.test.ts

@@ -1,9 +1,14 @@
 import context from '@aws-lambda-powertools/testing-utils/context';
+import type { Context } from 'aws-lambda';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import {
   AppSyncEventsResolver,
   UnauthorizedException,
 } from '../../../src/appsync-events/index.js';
+import type {
+  AppSyncEventsSubscribeEvent,
+  OnPublishAggregatePayload,
+} from '../../../src/types/appsync-events.js';
 import {
   onPublishEventFactory,
   onSubscribeEventFactory,
@@ -63,6 +68,112 @@ describe('Class: AppSyncEventsResolver', () => {
     });
   });
 
+  it.each([
+    { aggregate: true, channel: { path: '/foo', segments: ['foo'] } },
+    {
+      aggregate: false,
+      channel: {
+        path: '/bar',
+        segments: ['bar'],
+      },
+    },
+  ])(
+    'preserves the scope when decorating with onPublish aggregate=$aggregate',
+    async ({ aggregate, channel }) => {
+      // Prepare
+      const app = new AppSyncEventsResolver({ logger: console });
+
+      class Lambda {
+        public scope = 'scoped';
+
+        @app.onPublish('/foo', { aggregate })
+        public async handleFoo(payloads: OnPublishAggregatePayload) {
+          return payloads.map((payload) => {
+            return {
+              id: payload.id,
+              payload: `${this.scope} ${payload.payload}`,
+            };
+          });
+        }
+
+        @app.onPublish('/bar')
+        public async handleBar(payload: string) {
+          return `${this.scope} ${payload}`;
+        }
+
+        public async handler(event: unknown, context: Context) {
+          return this.stuff(event, context);
+        }
+
+        async stuff(event: unknown, context: Context) {
+          return app.resolve(event, context, { scope: this });
+        }
+      }
+      const lambda = new Lambda();
+      const handler = lambda.handler.bind(lambda);
+
+      // Act
+      const result = await handler(
+        onPublishEventFactory(
+          [
+            {
+              id: '1',
+              payload: 'foo',
+            },
+          ],
+          channel
+        ),
+        context
+      );
+
+      // Assess
+      expect(result).toEqual({
+        events: [
+          {
+            id: '1',
+            payload: 'scoped foo',
+          },
+        ],
+      });
+    }
+  );
+
+  it('preserves the scope when decorating with onSubscribe', async () => {
+    // Prepare
+    const app = new AppSyncEventsResolver({ logger: console });
+
+    class Lambda {
+      public scope = 'scoped';
+
+      @app.onSubscribe('/foo')
+      public async handleFoo(payload: AppSyncEventsSubscribeEvent) {
+        console.debug(`${this.scope} ${payload.info.channel.path}`);
+      }
+
+      public async handler(event: unknown, context: Context) {
+        return this.stuff(event, context);
+      }
+
+      async stuff(event: unknown, context: Context) {
+        return app.resolve(event, context, { scope: this });
+      }
+    }
+    const lambda = new Lambda();
+    const handler = lambda.handler.bind(lambda);
+
+    // Act
+    await handler(
+      onSubscribeEventFactory({
+        path: '/foo',
+        segments: ['foo'],
+      }),
+      context
+    );
+
+    // Assess
+    expect(console.debug).toHaveBeenCalledWith('scoped /foo');
+  });
+
   it('returns null if there are no onSubscribe handlers', async () => {
     // Prepare
     const app = new AppSyncEventsResolver({ logger: console });