AmplifyでDynamoDB Streamを用いて変更履歴をS3に保存する

# はじめに

この記事は、「AWS AmplifyとAWS×フロントエンド Advent Calendar 2022」の12/17分です。  
だいぶ後半に予定を入れたつもりでしたが、あっという間に出番が回ってきました。
いよいよ年末ですね！！

テーマに沿っていない気がしますが、この記事では「バックエンド」の話しを書きます　^^;


# 想定されるユースケース

想定するユースケースを説明します。

あなたは、Webシステムのバックエンド担当で、APIから後ろをメインの担当をしています。
Amplifyを使うので、構成としてはオーソドックスに、API Gateway ＋ Lambda ＋ DynamoDBを使うことにしました。

### ある日のできごと（概ねフィクションです）

ユーザー　「データを変更した時に**変更履歴**をとっておきたい」
あなた　「変更履歴はどのように使うのでしょうか？」
ユーザー　「使い方は決まっていないが、**とにかく何かの時のために**とっておきたい」
あなた　「…(*´Д｀)」

あなた　「かくかくしかじか」
開発者Ｓ　「DynamoDBの属性追加するかテーブル追加するかして保存しますか？」
あなた　「うーん。微妙だな〜...とっておけばいいってことだし...」
開発者Ｓ　「S3に保存しておきますか？」
あなた　「そうっすね。DynamoDBのStreamからLambdaに変更前データを流して、S3にputするのがいいかな？」
開発者Ｓ・あなた　「それなら今まで開発した部分に影響もないから、それで行こう！！」

このようにして、めでたく変更履歴は耐久性に優れるS3に格納することになりました。

### システム構成図はこんな感じになりました

![all-component.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130262/e500e921-0f99-1def-e7cb-55e743ca06e3.png)


# 構築の流れ
前提として、Amplifyのプロジェクト内で以下のリソースは作成済みとします。
* API Gateway
* DynamoDB
* Lambda ファンクション（DynamoDBへのCRUD操作）
* 変更履歴を格納する S3
なお、これらのリソースはAmplifyを用いて作成しています。


従って、ここでは、以下の内容について触れていきます。
1. DynamoDBのStreamから変更前イメージを受け取れるようにする
2. Streamをトリガーとして起動するLambdaファンクションをAmplifyで追加する
3. Lambdaにトリガーを追加する

### 1. DynamoDBのStreamから変更前イメージを受け取れるようにする

Amplifyで作ったDynamoDBは自動的にStreamがEnableになっています。（便利！）
しかし、Streamには変更後イメージだけを流すようになっています。
![01_before_dynamo_stream.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130262/33de216e-8227-735e-5ea9-aba54aed2fd9.png)

ここでは、マネジメントコンソールを使って、一旦StreamをDisableして、再度Enableする際に、変更前後のイメージを流すようにします。

![04_enable_dynamo.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130262/c4f24da6-82c2-f161-8b2f-265d65eae817.png)



これで、変更前のイメージをLambdaで受け取ることが出来ます。

### 2. Streamをトリガーとして起動するLambdaファンクションをAmplifyで追加する

このパートが唯一Amplifyを使う部分です！
`amplify add function` を投げるだけですが...

```bash
$ amplify add function
? Select which capability you want to add: Lambda function (serverless function)
? Provide an AWS Lambda function name: basicInfoHistoryLambda
? Choose the runtime that you want to use: Python
Only one template found - using Hello World by default.

Available advanced settings:
- Resource access permissions
- Scheduled recurring invocation
- Lambda layers configuration
- Environment variables configuration
- Secret values configuration

? Do you want to configure advanced settings? Yes
? Do you want to access other resources in this project from your Lambda function? Yes
? Select the categories you want this function to have access to. storage
? Storage has 2 resources in this project. Select the one you would like your Lambda to access s3hogehogestorage, BasicInfoDynamo
? Select the operations you want to permit on s3hogehogestorage create, read
? Select the operations you want to permit on BasicInfoDynamo read

You can access the following resource attributes as environment variables from your Lambda function
        ENV
        REGION
        STORAGE_BASICINFODYNAMO_ARN
        STORAGE_BASICINFODYNAMO_NAME
        STORAGE_BASICINFODYNAMO_STREAMARN
        STORAGE_S3HOGEHOGESTORAGE_BUCKETNAME
? Do you want to invoke this function on a recurring schedule? No
? Do you want to enable Lambda layers for this function? No
? Do you want to configure environment variables for this function? No
? Do you want to configure secret values this function can access? No
? Do you want to edit the local lambda function now? No
? Press enter to continue 
Successfully added resource basicInfoHistoryLambda locally.
```


これで、ローカルの `amplify/backend/function/basicInfoHistoryLambda`にLambdaのひな形一式が出来ます。

Lambdaファンクションとカスタムポリシーを記載したら、`amplify push`します。

#### Lambdaファンクション
```python

import json
from aws_lambda_powertools import Logger
import boto3

bucket_name = "hogehoge-staging"
s3 = boto3.resource('s3')

# ロガー初期化
logger = Logger(level="INFO", service=__name__)


def handler(event, context):
  
  records = event['Records']
  for rec in records:
      logger.info(rec);
      if rec['eventName'] == 'MODIFY':
          oldImage = rec['dynamodb']['OldImage']

          id = rec['dynamodb']['Keys']['id']['S']
          approximateCreationDateTime = rec['dynamodb']['ApproximateCreationDateTime']
              
          # Dubug用のlog出力
          logger.info(id)
          logger.info(approximateCreationDateTime)
          logger.info(oldImage)
  
          # S3 に /history/<id>/<approximateCreationDateTime>.json というキーで保存
          json_key = "history/" + id + "/" + str(approximateCreationDateTime) + ".json"
          obj = s3.Object(bucket_name,json_key)

          r = obj.put(Body = json.dumps(oldImage))
  
  return {
      'statusCode': 200,
      'headers': {
          'Access-Control-Allow-Headers': '*',
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Methods': 'OPTIONS,POST,GET' },
      'body': json.dumps('Hello from your new Amplify Python lambda!')
  }
```

#### カスタムポリシー設定
LambdaがStreamを使う際にいくつか追加ポリシー設定が必要なので、`custom-policies.json`に記載します。
最初のブロックで指定するResourceには、対象のDynamoDB Streamのarnを指定します。


```json:custom-policies.json
[
  {
    "Action": [
      "dynamodb:GetShardIterator",
      "dynamodb:DescribeStream",
      "dynamodb:GetRecords"
    ],
    "Resource": ["arn:aws:dynamodb:ap-northeast-1:<YOUR-ACCOUNT>:table/<YOUR-DynamoDB>/stream/*"]
  },
  {
    "Action": ["dynamodb:ListStreams"],
    "Resource": ["*"]
  }
]
```

`amplify push` が無事に終わったら、次に、Lambdaにトリガーを設定します。

### 3. Lambdaにトリガーを追加する

この操作は、マネジメントコンソールで行いまいした。

＋Add triggerボタンを押して、対象のDynamoDBを選択します。
![01_lambda.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130262/61ed4bcd-8eca-5b28-15c4-799b1515abc7.png)


トリガーを追加した結果は以下の様になります。
![02_added_trigger.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130262/4d1f7714-9fbe-33dd-e1b4-46e5574e3fe5.png)


## まとめ

最後までお読みいただきありがとうございます。

スキル不足もあり、マネジメントコンソールでの操作が多くなってしまいました。
これは、本来はcustomやhookを活用することで、Amplifyの操作に一本化できるのだと思います。
今後も勉強して、手順が改良出来そうでしたら、記事をアップデートしていきたいと思います。


それでは、良いAmplifyライフを！！