Amplify Gen2 で DynamoDB Streams を使って DynamoDB テーブル間のデータ連携を行う

import { type ClientSchema, a, defineData } from '@aws-amplify/backend'; const schema = a.schema({ ScoreActivities: a .model({ team_id: a.string().required(), score_change: a.integer().required(), }) .authorization((allow) => [ allow.ownerDefinedIn("team_id").identityClaim("custom:team_id"), allow.publicApiKey() ]) }); : 

import { type ClientSchema, a, defineData } from '@aws-amplify/backend'; const schema = a.schema({ ScoreActivities: a .model({ team_id: a.string().required(), score_change: a.integer().required(), }) .authorization((allow) => [ allow.ownerDefinedIn("team_id").identityClaim("custom:team_id"), allow.publicApiKey() ]), TotalScore: a .model({ team_id: a.string().required(), total_score: a.integer().required(), }) .identifier(['team_id']) .authorization((allow) => [ allow.ownerDefinedIn("team_id").identityClaim("custom:team_id"), allow.publicApiKey() ]), }); : 

import { defineBackend } from '@aws-amplify/backend'; import { auth } from './auth/resource'; import { data } from './data/resource'; import { preTokenGenerationV2 } from './auth/pre-token-generation-v2/resource'; import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam'; import { EventSourceMapping, StartingPosition } from 'aws-cdk-lib/aws-lambda'; import { NodejsFunction} from 'aws-cdk-lib/aws-lambda-nodejs'; import * as url from 'node:url'; const backend = defineBackend({ auth, data, preTokenGenerationV2, }); const { cfnUserPool } = backend.auth.resources.cfnResources cfnUserPool.addPropertyOverride("LambdaConfig.PreTokenGenerationConfig",{ LambdaVersion: 'V2_0', LambdaArn: backend.preTokenGenerationV2.resources.lambda.functionArn, }); const updateTotalScoreStack = backend.createStack('UpdateTotalScoreStack'); const funcitonUpdateTotalScoreOnActivity = new NodejsFunction( updateTotalScoreStack, 'update-totalscore-on-activity', { entry: url.fileURLToPath(new URL('./functions/update-totalscore-on-activity/handler.ts', import.meta.url)), environment: { APPSYNC_ENDPOINT: backend.data.resources.cfnResources.cfnGraphqlApi.attrGraphQlUrl, APPSYNC_API_KEY: backend.data.resources.cfnResources.cfnApiKey?.attrApiKey || '', }, initialPolicy: [ new PolicyStatement({ effect: Effect.ALLOW, actions: [ "dynamodb:DescribeStream", "dynamodb:GetRecords", "dynamodb:GetShardIterator", "dynamodb:ListStreams", ], resources: ["*"], }), ], } ) new EventSourceMapping( updateTotalScoreStack, "TotalScoreFunctionEventSourceMapping", { eventSourceArn: backend.data.resources.tables["ScoreActivities"].tableStreamArn, target: funcitonUpdateTotalScoreOnActivity, batchSize: 100, startingPosition: StartingPosition.LATEST, } ); 

import axios, { AxiosError } from "axios"; import type { DynamoDBStreamHandler } from "aws-lambda"; import { Logger } from "@aws-lambda-powertools/logger"; type TotalScoreInput = { team_id: string; total_score: number; }; type GraphQLError = { message: string; path: string[]; errorType: string; }; const logger = new Logger({ logLevel: "INFO", serviceName: "dynamodb-stream-handler", }); // GraphQL操作の定義 const CREATE_TOTAL_SCORE = ` mutation CreateTotalScore($input: CreateTotalScoreInput!) { createTotalScore(input: $input) { team_id total_score } } `; const UPDATE_TOTAL_SCORE = ` mutation UpdateTotalScore($input: UpdateTotalScoreInput!) { updateTotalScore(input: $input) { team_id total_score } } `; const GET_TOTAL_SCORE = ` query GetTotalScore($team_id: String!) { getTotalScore(team_id: $team_id) { team_id total_score } } `; async function executeGraphQLRequest(query: string, variables: any) { if (!process.env.APPSYNC_ENDPOINT || !process.env.APPSYNC_API_KEY) { throw new Error("Missing required environment variables: APPSYNC_ENDPOINT or APPSYNC_API_KEY"); } const response = await axios.post( process.env.APPSYNC_ENDPOINT, { query, variables, }, { headers: { 'Content-Type': 'application/json', 'x-api-key': process.env.APPSYNC_API_KEY, }, } ); if (response.data.errors) { throw new Error(`GraphQL Error: ${JSON.stringify(response.data.errors)}`); } return response.data; } async function updateOrCreateTotalScore(teamId: string, newTotalScore: number) { try { const updateResponse = await executeGraphQLRequest( UPDATE_TOTAL_SCORE, { input: { team_id: teamId, total_score: newTotalScore } } ); logger.info('Successfully updated total score', { teamId, newTotalScore }); return updateResponse; } catch (error) { if (error instanceof Error) { if (error.message.includes('ConditionalCheckFailedException')) { logger.info('Record not found, creating new total score', { teamId, newTotalScore }); const createResponse = await executeGraphQLRequest( CREATE_TOTAL_SCORE, { input: { team_id: teamId, total_score: newTotalScore } } ); return createResponse; } } throw error; } } export const handler: DynamoDBStreamHandler = async (event) => { const batchItemFailures: Array<{itemIdentifier: string}> = []; logger.info('Starting to process DynamoDB Stream records', { recordCount: event.Records.length }); for (const record of event.Records) { try { if (record.eventName !== 'INSERT') { logger.info(`Skipping non-INSERT event: ${record.eventName}`); continue; } const newImage = record.dynamodb?.NewImage; if (!newImage) { logger.warn('No NewImage in record, skipping'); continue; } const teamId = newImage.team_id?.S; const scoreChange = parseInt(newImage.score_change?.N ?? '0'); if (!teamId || scoreChange === 0) { logger.warn('Invalid record data', { teamId, scoreChange }); continue; } const getTotalScoreResponse = await executeGraphQLRequest( GET_TOTAL_SCORE, { team_id: teamId } ); const currentScore = getTotalScoreResponse.data.getTotalScore?.total_score ?? 0; const newTotalScore = currentScore + scoreChange; const result = await updateOrCreateTotalScore(teamId, newTotalScore); logger.info('Successfully processed score change', { teamId, scoreChange, currentScore, newTotalScore, operation: result.data.createTotalScore ? 'create' : 'update' }); } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); const errorStack = error instanceof Error ? error.stack : undefined; logger.error('Error processing record', { error: errorMessage, stack: errorStack, recordId: record.eventID }); if (record.eventID) { batchItemFailures.push({ itemIdentifier: record.eventID }); } } } return { batchItemFailures }; };