整理 AWS AppSync Events 数据源集成的使用方法

import * as cdk from 'aws-cdk-lib';
import { CfnOutput } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as appsync from 'aws-cdk-lib/aws-appsync';
import { AppSyncFieldLogLevel } from 'aws-cdk-lib/aws-appsync';
import * as logs from 'aws-cdk-lib/aws-logs';

export class AppSyncStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const apiKeyProvider: appsync.AppSyncAuthProvider = {
      authorizationType: appsync.AppSyncAuthorizationType.API_KEY
    };

    const api = new appsync.EventApi(this, 'EventApi', {
      apiName: 'AppSyncEventsSampleApi',
      authorizationConfig: {
        authProviders: [
          apiKeyProvider
        ],
        connectionAuthModeTypes: [
          appsync.AppSyncAuthorizationType.API_KEY
        ],
        defaultPublishAuthModeTypes: [
          appsync.AppSyncAuthorizationType.API_KEY
        ],
        defaultSubscribeAuthModeTypes: [
          appsync.AppSyncAuthorizationType.API_KEY
        ]
      },
      logConfig: {
        retention: logs.RetentionDays.ONE_DAY,
        fieldLogLevel: AppSyncFieldLogLevel.ALL
      }
    });
    // 无数据源集成的通道
    api.addChannelNamespace('sample');
  }
}

// (省略前文)

// 要进行数据源集成的 DynamoDB 表
const table = new dynamodb.Table(this, 'EventTable', {
  tableName: 'AppSyncEventsTable',
  partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING }
});
// 注册为 AppSync Events 的数据源
const dynamodbDataSource = api.addDynamoDbDataSource('EventDataSource', table);

// 用于 DynamoDB 的命名空间
api.addChannelNamespace('sample-dynamodb', {
  // 将 DynamoDB 设置为数据源
  publishHandlerConfig: {
    dataSource: dynamodbDataSource
  },
  // 自定义处理程序
  code: appsync.Code.fromInline(`
import * as ddb from '@aws-appsync/utils/dynamodb'

// 在事件发布及分发时执行的钩子
export const onPublish = {
  // 发布
  request(ctx) {
    const channel = ctx.info.channel.path
    return ddb.batchPut({
      tables: {
        '${table.tableName}': ctx.events.map(({ id, payload }) => ({ channel, id, ...payload })),
      }
    })
  },
  // 分发
  response(ctx) {
    // ctx.result 中存放了 request 的执行结果
    return ctx.result.data['${table.tableName}'].map(({ id, ...payload }) => ({ id, payload: 'DynamoDB:' + payload.message }))
  }
}
// 在频道订阅时执行的钩子
export const onSubscribe = (ctx) => {
  console.log('Joined:' + ctx.info.channel.path)
}`)
});

// 要进行数据源集成的 Lambda 函数
const fn = new lambda.Function(this, 'EventHandler', {
  functionName: 'SampleAppSyncDataSourceHandler',
  runtime: lambda.Runtime.NODEJS_22_X,
  handler: 'index.handler',
  code: lambda.Code.fromInline(`
exports.handler = async (event) => {
  if (event.info.operation === 'PUBLISH') {
    // onPublish 事件
    return {
      events: event.events.map(e => ({
        id: e.id,
        payload: 'Lambda:' + e.payload.message // 在此设置分发内容
      }))
      // 错误发生时可返回以下
      // error: 'error message'
    };
  } else {
    // onSubscribe 事件
    return null; // 订阅 OK
  }
}`)
});
// 注册为 AppSync Events 的数据源
const lambdaDataSource = api.addLambdaDataSource('EventHandler', fn);
// 用于 Lambda 的命名空间
api.addChannelNamespace('sample-lambda', {
  publishHandlerConfig: {
    direct: true, // 直接集成
    dataSource: lambdaDataSource
  },
  subscribeHandlerConfig: {
    direct: true, // 直接集成
    dataSource: lambdaDataSource
  }
});