Test Event-Driven Architectures with EventBridge and AppSync Events

Testing Event-Driven applications is difficult. Last year, I wrote a piece on testing that your application(s) produce the events you expect, using Momento Topics. At the time, the setup was a lot simpler than what was possible with AppSync.

AWS introduced AppSync Events a couple of weeks back. When reading the announcement, I immediately thought of my old blog post. I hoped that AppSync Events could be used to build a solution that’s as simple as the previous one, using only AWS services.

In this post, I will show you how to set up the required resources to test your event-producing application(s) end-to-end.

AppSync Events

In the release post, AWS describes AppSync Events as follows:

Today, AWS AppSync announced AWS AppSync Events, a feature that lets developers easily broadcast real-time event data to a few or millions of subscribers using secure and performant serverless WebSocket APIs. With AWS AppSync Events, developers no longer have to worry about building WebSocket infrastructure, managing connection state, and implementing fan-out. Developers simply create their API, and publish events that are broadcast to clients subscribed over a WebSocket connection. AWS AppSync Event APIs are serverless, so you can get started quickly, your APIs automatically scale, and you only pay for what you use.

Compared to earlier alternatives, such as AppSync GraphQL subscriptions and API Gateway WebSocket API, AppSync Events are much simpler to set up. You only have to create an API, define configuration such as default authorization, and create a channel namespace. Then, you can start publishing and subscribing to channels in the namespace.

We will only use the basic features of AppSync Events in this post. We will create a single namespace and protect it using API_KEY authorization.

For the curious, there are more advanced features available, such as per-namespace authorization and custom handlers for onPublish and onSubscribe

Sample Application

The sample application is very similar to the application in the previous post. It’s a simple Order Service, consisting of an API Gateway with a single POST /event endpoint that triggers a Lambda function. This function generates a new order item and publishes an OrderCreated event to EventBridge.

To recap, the Lambda function looks like this:

1
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";
2
import type { APIGatewayProxyEventV2, APIGatewayProxyResultV2 } from "aws-lambda";
3
import { ulid } from "ulid";
4

5
const ebClient = new EventBridgeClient();
6
const EVENT_BUS_NAME = process.env.EVENT_BUS_NAME || "default";
7

8
export const handler = async (_event: APIGatewayProxyEventV2): Promise<APIGatewayProxyResultV2> => {
9
  const order = {
10
    id: ulid(),
11
    name: "test order",
12
  };
13

14
  await ebClient.send(
15
    new PutEventsCommand({
16
      Entries: [
17
        {
18
          EventBusName: EVENT_BUS_NAME,
19
          Source: "OrderService",
20
          DetailType: "OrderCreated",
21
          Detail: JSON.stringify(order),
22
        },
23
      ],
24
    }),
25
  );
26

27
  return {
28
    statusCode: 201,
29
    body: JSON.stringify(order),
30
  };
31
};

When a POST /event request is sent to the API Gateway, this handler is invoked, publishing an event with the following structure to EventBridge:

1
{
2
  "version": "0",
3
  "id": "11111111-2222-4444-5555-666666666666",
4
  "detail-type": "OrderCreated",
5
  "source": "OrderService",
6
  "account": "123456789012",
7
  "time": "2023-10-19T08:00:00Z",
8
  "region": "eu-west-1",
9
  "resources": [],
10
  "detail": {
11
    "id": "01F9ZQZJZJZJZJZJZJZJZJZJZJ",
12
    "name": "test order"
13
  }
14
}

The test setup will also be very similar to before. But let’s hold off on and start with the necessary test resources.

Test infrastructure

Last time, I used another SAM template to separate the application infrastructure from the test infrastructure. To add some variation, I’m using CDK this time. I decided to create a custom construct called TestResources. I also added a stack property to conditionally deploy the resources in the environments where it makes sense:

1
interface StackProps extends cdk.StackProps {
2
  includeTestStack?: boolean;
3
}
4

5
export class EventbridgeTestingWithAppsyncEventsStack extends cdk.Stack {
6
  constructor(scope: Construct, id: string, props?: StackProps) {
7
    super(scope, id, props);
8

9
    // Sample application resources
10
    const eventBus = ...
11
    const apiFn = ...
12
    const api = ...
13

14
    // Test resources
15
    if (props?.includeTestStack) {
16
      const testResources = new TestResources(this, "TestResources", {
17
        eventBus,
18
      });
19
    }
20
  }
21
}

So what test resources do we need? Let’s start with the obvious one, an AppSync Events API.

AppSync Events API

The core component of the test infrastructure is the AppSync Events API. For this demo, I will create an API with a namespace called default and use an API key for authorization. At the time of writing, there are still no L2 constructs for AppSync Events, so we’ll have to do with L1 constructs for now.

1
interface ConstructProps {
2
  eventBus: eventbridge.IEventBus;
3
}
4

5
export class TestResources extends Construct {
6
  public eventsApi: appsync.CfnApi;
7

8
  constructor(scope: Construct, id: string, props: ConstructProps) {
9
    super(scope, id);
10

11
  this.eventsApi = new appsync.CfnApi(this, "TestEventsApi", {
12
    name: "TestEventsApi",
13
    eventConfig: {
14
      authProviders: [{ authType: "API_KEY" }],
15
      connectionAuthModes: [{ authType: "API_KEY" }],
16
      defaultPublishAuthModes: [{ authType: "API_KEY" }],
17
      defaultSubscribeAuthModes: [{ authType: "API_KEY" }],
18
    },
19
  });
20

21
  const eventsApiKey = new appsync.CfnApiKey(this, "TestEventsApiKey", {
22
    apiId: this.eventsApi.attrApiId,
23
    expires: Math.floor(Date.now() / 1000) + 60 * 60 * 24 * 365, // 1 year from now
24
  });
25

26
  new appsync.CfnChannelNamespace(this, "TestEventsNamespace", {
27
    apiId: this.eventsApi.attrApiId,
28
    name: "default",
29
  });
30
}

The above creates an AppSync Events API with API_KEY authorization. I’m setting the API key’s expiration to a year from now during each deployment for demo purposes. In a real-world scenario, you’d want to manage the API key’s lifecycle more carefully.

Attempt #1: EventBridge API Destination

The release post included a section on how to integrate EventBridge and AppSync Events using API destinations. In the example, an event with the following detail is used:

1
{
2
  "channel": "/default/introductions",
3
  "events": ["{\"message\":\"Hello from EventBridge!\"}"]
4
}

In my case, I want to send arbitrary events (such as OrderCreated) to a specific channel. With Momento, this was as straightforward as setting the topic name as part of the URL in the API destination and sending the event payload in the body.

The request format in AppSync Events is a bit different. AppSync expects the payload to contain both the destination channel and up to five events. Each event must be stringified JSON, like {\"message\":\"Hello from EventBridge!\"} above. Sadly, this request schema makes it impossible to directly integrate EventBridge and AppSync Events without having something in between.

For my first attempt, I’d assume that I could use a template like this:

1
{
2
  "channel": "/namespace/channel",
3
  "events": [
4
    <detail>
5
  ]
6
}

However, this produces the following output:

1
{
2
  "channel": "/namespace/channel",
3
  "events": [
4
    {
5
      "id": "01JDSQKGFAHQMH866CARE8F9JC",
6
      "name": "test order"
7
    }
8
  ]
9
}

I need the event to be stringified, but there’s no way to do that with EventBridge input transformation. That AWS chose a request schema for AppSync that cannot be used this way with API destinations is a blunder, to be frank.

The integration only works if the EventBridge publisher is aware of the AppSync namespace and channel and can format the event accordingly. At that point, the publisher might as well publish to the AppSync Events API directly.

We require something between EventBridge and AppSync, but what?

Attempt #2: Lambda Function

After wrangling with API destinations for far too long, I took the easy way out. I wired an EventBridge rule to a basic Lambda function:

1
export const handler = async (event: EventBridgeEvent<string, unknown>): Promise<void> => {
2
  await fetch(`${EVENTS_API_URL}/event`, {
3
    method: "POST",
4
    headers: {
5
      "Content-Type": "application/json",
6
      "x-api-key": EVENTS_API_KEY,
7
    },
8
    body: JSON.stringify({
9
      channel: "/default/debug",
10
      events: [JSON.stringify(event)],
11
    }),
12
  });
13
};

This works and since the function is only used for testing purposes, cold starts aren’t a problem. But, it feels wrong to have to do this for such a simple transformation. Let’s try another way.

Attempt #3: Express Step Function

The only thing we need is to transform the input and send it to the AppSync Events API. This sounds like the perfect job for a Step Function. I will create an express state machine to transform the payload with JSONata and send it to AppSync using an HTTP task.

Let’s extend our TestResources construct. First, create a new EventBridge connection that the HTTP task can use:

1
const connection = new eventbridge.Connection(this, "TestEventsConnection", {
2
  authorization: eventbridge.Authorization.apiKey(
3
  "x-api-key",
4
  cdk.SecretValue.resourceAttribute(eventsApiKey.attrApiKey),
5
  ),
6
});

Then, add the state machine. In this case, there is only one step, so I’ll inline the state machine definition. I’d recommend keeping it in a separate .yml file for more complex state machines.

1
const stateMachine = new sfn.StateMachine(this, "TestEventsStateMachine", {
2
  stateMachineType: sfn.StateMachineType.EXPRESS,
3
  logs: {
4
  destination: new logs.LogGroup(this, "TestEventsStateMachineLogGroup", {
5
    logGroupName: "/aws/vendedlogs/states/TestEventsStateMachine",
6
    removalPolicy: cdk.RemovalPolicy.DESTROY,
7
  }),
8
  includeExecutionData: true,
9
  level: sfn.LogLevel.ALL,
10
  },
11
  definitionBody: sfn.DefinitionBody.fromString(
12
    JSON.stringify({
13
      QueryLanguage: "JSONata",
14
      StartAt: "PublishTestEvent",
15
      States: {
16
        PublishTestEvent: {
17
          Type: "Task",
18
          Resource: "arn:aws:states:::http:invoke",
19
          Arguments: {
20
            Method: "POST",
21
            Authentication: {
22
              ConnectionArn: "${ConnectionArn}",
23
              },
24
            ApiEndpoint: "${ApiEndpoint}",
25
            RequestBody: {
26
              channel: "/default/debug",
27
              events: ["{% $string($states.input) %}"],
28
            },
29
          },
30
          End: true,
31
        },
32
      },
33
    }),
34
  ),
35
  definitionSubstitutions: {
36
    ConnectionArn: connection.connectionArn,
37
    ApiEndpoint: `https://${this.eventsApi.attrDnsHttp}/event`,
38
  },
39
});

A lot is happening here. Let’s walk through it.

I’m setting the workflow type to express. Express workflows are ideal for short-running jobs such as this. I’m creating a log group and configuring logging on the state machine to be able to view executions in the console.

The state machine comprises a single HTTP task that uses the EventBridge connection created earlier to send requests to AppSync. The state input is transformed into the format AppSync expects with a little help from the JSONata $string function.

Next, I add a couple of permissions to the state machine:

1
stateMachine.role.attachInlinePolicy(
2
  new iam.Policy(this, "StateMachinePolicy", {
3
  statements: [
4
    new iam.PolicyStatement({
5
      sid: "AllowUseConnection",
6
      effect: iam.Effect.ALLOW,
7
      actions: ["events:RetrieveConnectionCredentials"],
8
      resources: [connection.connectionArn],
9
    }),
10
    new iam.PolicyStatement({
11
      sid: "AllowGetConnectionSecret",
12
      effect: iam.Effect.ALLOW,
13
      actions: ["secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret"],
14
      resources: [connection.connectionSecretArn],
15
    }),
16
    new iam.PolicyStatement({
17
      sid: "AllowInvokeHTTPEndpoint",
18
      effect: iam.Effect.ALLOW,
19
      actions: ["states:InvokeHTTPEndpoint"],
20
      resources: ["*"],
21
    }),
22
  ],
23
  }),
24
);

Finally, I glue everything together by creating an EventBridge rule that forwards events from the event bus to the state machine:

1
new eventbridge.Rule(this, "TestEventsRule", {
2
  eventBus: props.eventBus,
3
  eventPattern: {
4
  source: eventbridge.Match.prefix(""),
5
  },
6
  targets: [new eventbridgeTargets.SfnStateMachine(stateMachine)],
7
});

For demo purposes, I’m forwarding all events. If you want to use this method to run tests against production, you most likely only want test events to flow through the test infrastructure.

If you can distinguish test from production traffic in your event-producing services, you can format your event to allow filtering out only test events.

Verify the test infrastructure

Let’s verify that the test infrastructure is wired correctly before moving on. The AppSync console for Event APIs has a neat Pub/Sub Editor that lets you connect and subscribe to the API directly in the browser.

Connect to your API and subscribe to the debug channel in the default namespace. Then, go to the console for your event bus and send an event or two. The events you send should show up in the AppSync console.

Test the sample application end-to-end

With the test infrastructure in place, we can now test our application End-to-End and be confident that it emits the correct event(s). The full architecture looks like this:

The test code is very similar to the test code in my previous post. The major difference is that it subscribes to AppSync instead of Momento.

The test case itself is almost identical:

1
describe("When an order is created", async () => {
2
  const API_URL = process.env.API_URL || "";
3
  const subscription = await subscribe("default", "debug");
4

5
  afterAll(async () => {
6
    subscription.unsubscribe();
7
  });
8

9
  it("It should publish an OrderCreated event to EventBridge", async () => {
10
    const response = await fetch(`${API_URL}/event`, { method: "POST" });
11
    expect(response.status).toBe(201);
12

13
    const order = await response.json();
14

15
    const message = await subscription.waitForMessageMatching({
16
      source: "OrderService",
17
      "detail-type": "OrderCreated",
18
      detail: {
19
        id: order.id,
20
        name: order.name,
21
      },
22
    });
23

24
    expect(message).not.toBeNull();
25
  }, 5000);
26
});

First, we subscribe to the debug channel in the default namespace. The test then creates an order through the API and stores the returned order ID. Then, it waits a maximum of five seconds for a message matching the expected format to appear on the channel.

The major difference from the Momento post is how subscribe works:

1
import "dotenv/config";
2
import { Amplify } from "aws-amplify";
3
import { events } from "aws-amplify/data";
4
import * as _ from "lodash";
5
import { ReplaySubject, firstValueFrom } from "rxjs";
6
import { filter } from "rxjs/operators";
7
import { WebSocket } from "ws";
8

9
// @ts-expect-error type
10
globalThis.WebSocket = WebSocket;
11

12
Amplify.configure({
13
  API: {
14
    Events: {
15
      endpoint: process.env.EVENTS_API_URL || "missing",
16
      defaultAuthMode: "apiKey",
17
      apiKey: process.env.EVENTS_API_KEY || "missing",
18
    },
19
  },
20
});
21

22
export const subscribe = async (namespace: string, channelName: string) => {
23
  const messages = new ReplaySubject(100);
24

25
  const channel = await events.connect(`/${namespace}/${channelName}`);
26
  channel.subscribe({
27
    next: (data) => {
28
      messages.next(data.event);
29
    },
30
    error: (err) => console.error("error", err),
31
  });
32

33
  // Give the subscription some time to establish
34
  await new Promise((resolve) => setTimeout(resolve, 250));
35

36
  const unsubscribe = async () => {
37
    channel.close();
38
  };
39

40
  const waitForMessageMatching = async (expected: object) => {
41
    const predicate = (message: unknown) => {
42
      if (typeof message !== "object" || message === null) {
43
        return false;
44
      }
45

46
      return _.isMatch(message, expected);
47
    };
48

49
    const data = messages.pipe(filter((message) => predicate(message)));
50
    return firstValueFrom(data);
51
  };
52

53
  return {
54
    unsubscribe,
55
    waitForMessageMatching,
56
  };
57
};

I’m using the AWS Amplify client to handle the connection for me. The Amplify implementation requires a WebSocket client to be available. In node, this was previously available behind an experimental flag. Starting with Node.js 22, a browser-compatible implementation of WebSocket is enabled by default. For earlier versions, you can use the ws package and set globalThis.WebSocket = WebSocket.

In the subscribe function, I create an RxJS ReplaySubject and connect and subscribe to the AppSync Events API using the Amplify client. All incoming messages are stored in the ReplaySubject.

Individual tests can use the waitForMessageMatching to find events matching a specific pattern. In our test, we are looking for an OrderCreated event:

1
const message = await subscription.waitForMessageMatching({
2
  source: "OrderService",
3
  "detail-type": "OrderCreated",
4
  detail: {
5
    id: order.id,
6
    name: order.name,
7
  },
8
});
9

10
expect(message).not.toBeNull();

Conclusion

Serverless applications are inherently event-driven and can be challenging to test. Developers have gone to great lengths to try and emulate managed AWS services locally to facilitate testing. I’m a strong advocate for testing in the cloud, as it is the only way to truly test the behavior of your application(s) in a production-like environment.

With the introduction of AppSync Events, AWS has made it easier to set up pub/sub patterns using websockets with little to no effort. In this post, you have learned how to combine EventBridge, Step Functions, and AppSync Events to test that your application(s) emit the correct events.