Workarounds for AppSync Subscriptions triggers via Lambda functions

AWS AppSync is a serverless GraphQL offering by AWS, previously I authored a blog about AWS AppSync 101 which gets you up to speed with the capabilities of AppSync and how you can leverage them in your serverless applications.

One query that stands out is that -

AppSync subscriptions work amazing when client makes mutations but I have some downstreaming patterns in my application where the backend uses subscriptions to send data from back-end to front-end, how do I achieve this with AppSync?

Ask ChatGPT

Well, this does look simple so why not ask ChatGPT?

What would be the better way to handle AppSync Subscriptions which are triggered with Lambda functions?

ChatGPT's solution

Use the appropriate trigger:

To use the appropriate trigger, you can create a trigger on the event that is relevant to your use case. For example, if you want to trigger a subscription when a record is added to a database, you can use a database trigger to invoke your Lambda function.

Here's an example using AWS Lambda and Amazon DynamoDB:

• Create a DynamoDB table with a stream enabled.

• Create a Lambda function that is triggered by the DynamoDB stream.

• In the Lambda function, write code that parses the stream event and invokes the AppSync subscription.

And also, ChatGPT generated the code snippet for this.

// Sample code for a Lambda function that is triggered by a DynamoDB stream

exports.handler = async (event) => {
  // Parse the DynamoDB stream event
  const records = event.Records.map((record) => {
    return {
      eventName: record.eventName,
      record: record.dynamodb.NewImage,
    };
  });

  // Invoke the AppSync subscription
  const appSyncClient = new AWSAppSyncClient({
    url: 'https://my-appsync-endpoint.aws.com/graphql',
    region: 'us-east-1',
    auth: {
      type: 'AWS_IAM',
      credentials: AWS.config.credentials,
    },
  });

  const mutation = gql`
    subscription OnRecordAdded {
      onRecordAdded {
        id
        name
      }
    }
  `;

  const observable = appSyncClient.subscribe({ query: mutation });
  observable.subscribe((data) => console.log('Subscription data', data));
};

And the GraphQL schema for this was also generated by AppSync.

type Record {
  id: ID!
  name: String!
}

type Subscription {
  onRecordAdded: Record!
}

type Mutation {
  addRecord(name: String!): Record!
}

type Query {
  getRecord(id: ID!): Record
}

Let's validate ChatGPT 's response

Do you notice something from ChatGPT's response? Now let's put on the Serverless architect lens and see how this would be implemented.

GraphQL schema

The schema generated by ChatGPT is a generic one that looks good at first look but when from the application's perspective, all the operations - query, mutation and subscription are accessible based on the default authorization set on AppSync.

Ensure the subscription is defined for a specific mutation, this is to be noted when you have multiple mutations. Although this looks good in this single mutation schema however it's good to define them with @aws_subscribe directive.

type Subscription {
  onRecordAdded: Record
  @aws_subscribe(mutations: ["addRecord"])
}

Irrespective of what authorization is used in your application for the client (be it Cognito User Pool, API Key, Lambda authorizers or AWS IAM role), define the mutation which would be invoked from the Lambda function as AWS IAM role.

type Mutation {
  addRecord(name: String!): Record!
  @aws_iam
}

GraphQL resolvers

The conversation with ChatGPT didn't initially give me a resolver for the mutation addRecord, but a follow-up query did generate a VTL resolver (understand JS resolver is a very new, and fair guess that model doesn't know what JS resolver is at this point).

This VTL resolver defines the PutItem operation for the Mutation type in the AppSync schema. It uses the $ctx.args variable to extract the name argument from the mutation, and then generates a new id for the item using the $util.autoId() function. The resolver constructs a DynamoDB PutItem request using these values, and then returns the request in JSON format.

#set($name = $ctx.args.name)
#set($id = $util.autoId())
{
    "version": "2018-05-29",
    "operation": "PutItem",
    "tableName": "my-table",
    "item": {
        "id": { "S": "$id" },
        "name": { "S": "$name" }
    }
}

You may notice that the VTL resolver generated is doing a DynamoDB PutItem action even though the response ChatGPT gave didn't have the schema treating the Record type as a model or even give the steps to create a DynamoDB data source.

For the simplicity of things, let's create a simple VTL resolver that generates an ID and returns the name with the data source being a None type.

Request mapping template

{
    "version": "2017-02-28",
    "payload": {
        "name": "$context.arguments.name",
        "id": "$util.autoId()"
    }
}

Response mapping template

$util.toJson($context.result)

And the mutation addRecord is ready to be tested!

Lambda function to trigger subscriptions on client-end

The solution that a Serverless architect would have come up with is the one above, where any Lambda trigger could invoke the Lambda function, and the Lambda function in turn uses an AppSync mutation and on the successful execution of the mutation, the subscription is internally invoked by AppSync.

The Lambda function generated by ChatGPT uses AppSync Client SDK. And it subscribes to an AppSync subscription rather than a mutation. 🤔 Why! 😵‍💫

const observable = appSyncClient.subscribe({ query: mutation });
observable.subscribe((data) => console.log('Subscription data', data));

Firstly, create a Lambda layer with the npm dependencies -

npm install aws-sdk aws-appsync graphql-tag isomorphic-fetch axios

Create your Lambda function with the environment variable set with the AppSync API endpoint.

require('isomorphic-fetch');
const AWS = require('aws-sdk/global');
const AUTH_TYPE = require('aws-appsync').AUTH_TYPE;
const AWSAppSyncClient = require('aws-appsync').default;
const gql = require('graphql-tag');

const config = {
    url: process.env.APPSYNC_ENDPOINT,
    region: process.env.AWS_REGION,
    auth: {
        type: AUTH_TYPE.AWS_IAM,
        credentials: AWS.config.credentials,
    },
    disableOffline: true
};

const createTodo = gql`
    mutation MyMutation($name: String!) {
      addRecord(name: $name) {
        id
        name
      }
    }
`;

const client = new AWSAppSyncClient(config);

exports.handler = async function (event) {
    console.log("event ", event);
    try {
        const result = await client.mutate({
            mutation: createTodo,
            variables: {
                "name":event.name
            }
        });
        console.log("result ", result);
        return result
    } catch (error) {
        console.log("error ", error);
        return error
    }
};

credentials: AWS.config.credentials uses the IAM credentials from Lambda's runtime and for this to work, you would have to define

Update your Lambda's execution role with the policy -

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": "appsync:GraphQL",
            "Resource": "<Your AppSync ARN>"
        }
    ]
}

Back to your Lambda console, create an Event JSON.

{
  "name": "This is a record from Lambda fn"
}

When testing the subscription from the AppSync console. The API by default uses API Key as authorization.

Behind the scene, the event was tested from the Lambda function console and here the mutation is authorized by AWS IAM.

Wrap up

ChatGPT gave us a high-level solution in terms of what needs to be done but the how to execute and the minor details were missed out, this is exactly where Serverless architects would come into the picture to start building on top of what AI suggests.

The TL;DR of this is that you can start to get solutions from ChatGPT but the accuracy of how well it solves is something humans would have to intervene on. And building Lambda functions that invokes an AppSync mutation to trigger subscriptions would be helpful when you are relying on real-time sync from external sources other than the support data sources from AppSync.