Feature request: AppSync Event schema/envelope

[Jimmy89]

Use case

I'm starting with Appsync events, but the lambda that is called as data source (on publish/subscribe) also receives input from other sources.

Therefore, I want to filter within the lambda event handler from which service the event came from to check which message processor is required. The current new AppSync Events event headers examples contain only the app.resolve(event, context) but I would like to place a filter in advance like "if event === appsyncevent". I do not see Validation schema/envelope for these appsync events. I could make one myself, but I think it would be a great addition to this package.

Solution/User Experience

A validator that can verify if an incoming message is a AppSync Events event, for use cases where multiple sources/services call the same lambda.

Alternative solutions

Acknowledgment

• This feature request meets Powertools for AWS Lambda (TypeScript) Tenets
• Should this be considered in other Powertools for AWS Lambda languages? i.e. Python, Java, and .NET

Future readers

Please react with 👍 and your use case to help us understand customer demand.

[dreamorosi]

Hi @Jimmy89, thank you for opening this issue.

We actually do some rudimentary shape validation (see implementation) so that if your function gets a payload with a different it shouldn't get to the route handler.

With that said, I agree that we should have more robust validation and as a matter of fact that was part of our initial design - so thank you again for asking for it.

The experience that we had in mind would be something like this:

import { AppSyncEventsPublishEvent } from '@aws-lambda-powertools/parser/schemas/appsync-events';
import { z } from 'zod';

const app = new AppSyncEventsResolver();

const eventSchema = z.object({
  foo: z.string(),
  bar: z.number(),
});

const eventSchema = AppSyncEventsPublishEvent.extend({
  events: z.array(eventSchema)
})

app.onPublish(
  '/default/channel1',
  async (payload) => {
    // your logic here
  },
  {
    validation: {
      input: eventSchema,
    },
  }
);

export const handler = async (event, context) =>
  app.resolve(event, context);

As a question to you, what would you expect to happen when the validation fails? Should we:

• throw an error, not call the route handler, and consequently drop the messages (aka block publishing)
• log an error, not call the route handler, but let the messages pass through

[Jimmy89]

@dreamorosi thank you for your quick response. Good to know there is rudimentary shape validation.

First, starting with the current rudimentary shape validation that is executed: how do I know if app.resolve(event) returns undefined because the event was not from appsync events OR whether it is a successful subscribe action? If I'm not mistaken, both return the same value.

Currently, my lambda checks for a S3 event like this:

import { S3Schema } from '@aws-lambda-powertools/parser/schemas';
/**
 * Check if an incoming lambda event is a S3 event
 * @param {object} event
 * @return {Promise<boolean>} True if S3 event
 */
async function isS3Event(event) {
  const parsedEvent = await S3Schema.safeParseAsync(event);
  const { success } = parsedEvent;
  simpleObjectCache.set(event, parsedEvent);
  return success;
}

/**
 * Extract the important details from a S3 event.
 * @param event
 * @return {Promise<{}[]>}
 */
async function extractS3EventDetails(event) {
  const parsedEvent = await S3Schema.safeParseAsync(event);
  const parseCreateEvent = (obj) => ({}); // Specific object

  return parsedEvent.success ? parsedEvent.data.Records.map((obj) => {
    return parseCreateEvent(obj);
  }) : [];
}

Where is the lambda handler I check if isS3Event and then use parse logic. I want to do the same with appsync events, like ifIsAppSyncEvent. I could use your current check for that, but as it is not officially documented, I must assume that with a version upgrade my lambda could break (which I like to avoid ;)). So any proper solution for this would be great.

Secondly, back to your question. In your proposal, you write:

app.onPublish(
  '/default/channel1',
  async (payload) => {
    // your logic here
  },
  {
    validation: {
      input: eventSchema,
    },
  }
);

I would assume the following:

1. If multiple messages are sent, and one is invalidated, the other ones are still processed.
2. I can use the validation together with aggregate: true
3. I can be in control of how the error message is handled.

An option that could be considered is:

app.onPublish(
  '/default/channel1',
  async (payload, event, context, validation) => {
  {
   aggregate: true | false,
    validation: {
      input: eventSchema,
    },
  }
);

Where the validation object could contain information like:

{
  success: boolean,
  failed: [index numbers]  //in case aggregate: true, or an array of objects with validation message and index number that failed
} 

This would ensure that the developer is able to steer the invalidation (error/ignore/warning). If the developer likes to throw an error, that's up to him/here.

[dreamorosi]

First, starting with the current rudimentary shape validation that is executed: how do I know if app.resolve(event) returns undefined because the event was not from appsync events OR whether it is a successful subscribe action? If I'm not mistaken, both return the same value.

You're correct - there's no way to distinguish besides from looking at the logs. The overall stance we assumed in this type of situations is that we leave the event as-is when the resolver is unable to handle it. That validation is there only to prevent a runtime error down the line, since the required fields would not be present.

---

Based on your code, I think I now better understand the request and at least for now we should just add a AppSyncEventsPublishEvent and AppSyncEventsSubscribeEvent schemas to the Parser utility - so that you can implement your own logic.

The deeper integration with the Event Handler utility needs maybe a bit more thought, I didn't think about what should happen when aggregate is true - but perhaps we should just support this when aggregate is false. Either way it needs more thought and I don't want to slow anyone down.

I'll move forward with implementing the two Zod schemas, so that they are included in the next release.

---

Besides this, I have to ask: why not just have two separate Lambda functions that each process one type of events? You'd have less logic to handle.

[Jimmy89]

I'll move forward with implementing the two Zod schemas, so that they are included in the next release.

Thank you, that's great.

Besides this, I have to ask: why not just have two separate Lambda functions that each process one type of events? You'd have less logic to handle.

Some of our lambda's contain quite a lot of business logic, where the events trigger small updates on the same entities (storing items within dynamodb according to a schema, sending mail etc). If we separate them, we end up with one big library and small lambda's entries, all doing almost the same thing, with permissions split up. This would add complexity in our codebase to maintain, while we're a small team of 5 developers. Therefore, we have a more monolith lambda that extracts the goal of the input events and then passes this to our general business logic layer. Sure, It does have it tradeoffs (e.g. not being able to tweak a lambda to specific memory for a specific task), but so far we managed to handle SQS, SNS, S3, Dynamodb streams and AppSync events/calls, as well as direct invocation.

[github-actions]

⚠️ COMMENT VISIBILITY WARNING ⚠️

This issue is now closed. Please be mindful that future comments are hard for our team to see.

If you need more assistance, please either tag a team member or open a new issue that references this one.

If you wish to keep having a conversation with other community members under this issue feel free to do so.