AWS CDK EventBridge -> AppSync Subscriptions Proxy

event-proxy-lambda.js

const AWS = require('aws-sdk')
const appsync = require('aws-appsync');
const gql = require('graphql-tag');
require('cross-fetch/polyfill');

exports.handler = async function(event) {
  const graphqlClient = new appsync.AWSAppSyncClient({
    url: process.env.APPSYNC_ENDPOINT_URL,
    region: process.env.AWS_REGION,
    auth: {
      type: 'AWS_IAM',
      credentials: {
        accessKeyId: process.env.AWS_ACCESS_KEY_ID,
        secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
        sessionToken: process.env.AWS_SESSION_TOKEN
      }
    },
    disableOffline: true
  });

  const mutation = gql`mutation publish($event: EventBridgeEventInput!) {
    publishEventBridgeEvent(event: $event) {
      id
      payload
    }
  }`;
  await graphqlClient.mutate({
    mutation,
    variables: {
      event: {
        id: event.id,
        payload: event
      }
    }
  });
}

eventBridgeAppSyncProxy.ts

import * as cdk from '@aws-cdk/core';
import * as lambda from '@aws-cdk/aws-lambda'
import { GraphQLApi, FieldLogLevel, LambdaDataSource, MappingTemplate, UserPoolDefaultAction, CfnGraphQLApi} from '@aws-cdk/aws-appsync'
import { Rule, EventPattern } from '@aws-cdk/aws-events';
import { LambdaFunction } from '@aws-cdk/aws-events-targets';
import * as iam from '@aws-cdk/aws-iam';

import path = require('path');

export class GraphQlStack extends cdk.Stack {
  constructor(scope: cdk.Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);
      const cfnApi = api.node.defaultChild as CfnGraphQLApi
      cfnApi.additionalAuthenticationProviders = [{
        authenticationType: 'AWS_IAM'
      }]

    const publishEvent = new lambda.Function(this, 'PublishEvent', {
      runtime: lambda.Runtime.NODEJS_10_X,
      handler: 'index.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, 'publish-event')),
    });

    const publishEventDataSource = new LambdaDataSource(this, 'PublishEventDataSource', {
      api,
      name: 'publishEvent',
      lambdaFunction: publishEvent
    })

    publishEventDataSource.createResolver({
      typeName: 'Mutation',
      fieldName: 'publishEventBridgeEvent',
      requestMappingTemplate: MappingTemplate.lambdaRequest(`
        {
          "field": "publishEventBridgeEvent",
          "arguments": $utils.toJson($context.arguments)
        }
      `),
      responseMappingTemplate: MappingTemplate.lambdaResult()
    })

    const eventProxy = new lambda.Function(this, 'EventProxy', {
      runtime: lambda.Runtime.NODEJS_10_X,
      handler: 'index.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, 'event-proxy')),
      environment: {
        APPSYNC_ENDPOINT_URL: api.graphQlUrl
      }
    });
    
    eventProxy.addToRolePolicy(new iam.PolicyStatement({
      effect: iam.Effect.ALLOW,
      resources: [
        new cdk.StringConcat().join(api.arn.toString(), "/types/Mutation/fields/publishEventBridgeEvent"),
      ],
      actions: [
        "appsync:GraphQL"
      ]
    }))

    new Rule(this, "EcsEventPattern", {
      ruleName: "ecs-event-proxy",
      eventPattern: {
        source: [
          "aws.ecs"
        ],
        detailType: [
          "ECS Task State Change"
        ]
      } as EventPattern,
      targets: [new LambdaFunction(eventProxy)]
    })
  }
}

publish-event-lambda.js

const AWS = require('aws-sdk')

exports.handler = async function(event) {  
  console.log(JSON.stringify(event))
  return {
    id: event.event.id, 
    payload: JSON.stringify(event.event.payload)
  }
}

schema.graphql

input EventBridgeEventInput {
  id: String!
  payload: String!
}

type EventBridgeEvent @aws_iam {
  id: String!
  payload: String!
}

type Query {
}

type Subscription {
  newEventBridgeEvent: EventBridgeEvent
    @aws_subscribe(mutations: ["publishEventBridgeEvent"])
}

type Mutation {
  publishEventBridgeEvent(event: EventBridgeEventInput!): EventBridgeEvent
    @aws_iam
}