Step FunctionsでRDS×ElastiCacheを統合してみた！

# はじめに ～RDS × ElastiCache って？～
#### 概要
Amazon ElastiCache は、フルマネージドのインメモリデータストアおよびキャッシュサービスです。
ざっくりいうと、RDSと比べ、複雑なデータ構造は持てないが、処理が超早いデータベースです。

主な用途はこんな感じですね。
- データのキャッシュ保存
- セッションや重要なデータの高速読み取りがあります

#### RDS × ElastiCache
以下の記事で紹介されているように、ElastiCacheにメインのDBのキャッシュを任せることで、処理速度の向上や、RDSのリードレプリカ数の削減によるコスト割引などのメリットを享受することができます。 

https://docs.aws.amazon.com/ja_jp/AmazonRDS/latest/UserGuide/creating-elasticache-cluster-with-RDS-settings.html

#### RDS × ElastiCacheの注意点
ここまで聞くと最高じゃん！となるのですが、ちょっとした注意点があります。
それは、**RDS × ElastiCache の統合は、AWSがサービスの機能として提供してくれるわけではない**ということです。
実現しようと思うと、自前での実装が必要になります。

トランザクションやエラー時の制御などを色々考えると、若干ハードルが高いわけですね。

#### 今回のソリューション
今回はこのハードルを少しでも下げられないか？ということで、 AWS Step Functions を使用して、キャッシュ読み取り機能を実装してみようと思います！

# つづいて ～AWS Step Functions って？～
#### 概要
ざっくりいうと、サービス間（または組み込み関数）を用いた複雑なフローを実行・制御してくれるサービスです！
AWS Batchをよりフレンドリーにした感じで、GUIで楽しく設計できるのが大きなウリです。

https://aws.amazon.com/jp/builders-flash/202208/awsgeek-stepfunctions/

https://dev.classmethod.jp/articles/stepfunctions-workflow-designer/

![cap0.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/5b2b0f4d-85f5-3588-0bc0-15e739d5825b.png)

こんな感じのGUIでフローを設定できます。
画像はS3オブジェクトをforeachしてDynamoDBに保存し、エラーがあればSNS通知するフローです。~~適当に置いたので動くのかわかりませんが~~
このように、各サービスの連携やloop、ifなどをグラフィカルに設定できるサービスです。
異常系の処理も簡単に設定できるのがすごいですよね。

# こんかいのアーキテクチャ図

![arch_a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/412959d0-7610-517a-ee33-d659133e49ef.png)

主な処理フローです
① キャッシュ取得lambdaを実行
② elasticacheへクエリ
③ 結果を返却
④ キャッシュが存在しない場合オリジンへクエリ
⑤ RDSへクエリ
⑥ 結果を返却
⑦ elasticacheへキャッシュをインサート
⑧ 結果を返却

※本来はキャッシュの更新は非同期でやるべしだと思うのですが、簡単アーキにしてます
※本来はRDS Proxyなどを使用するべしなのですが、簡単アーキにしています。あくまでsfnの検証がスコープ

今回はこのフローをAWS Step Functions で構築しようと思います。

# 下準備
主役以外のDB、lambda、AWS Step Functions用のサービスロールはTerraformでサクッと作ります。

<!-- open属性なし -->
<details><summary>PJTコード（長いのでおりたたみ）</summary>

```プロジェクト全体
SFN-MANAGE-CACHE/
│
├── lambda/
│   ├── operate-postgresql/
│   │   ├── node_modules/
│   │   ├── index.mjs
│   │   └── package.json             // postgresql操作用にpgライブラリを使用する
│   │
│   ├── operate-redis_get/           // キャッシュget用lambda
│   │   ├── node_modules/
│   │   ├── index.mjs
│   │   └── package.json             // redis操作用にioredisライブラリを使用する
│   │
│   ├── operate-redis_set/           // キャッシュset用lambda
│   │   ├── node_modules/
│   │   ├── index.mjs
│   │   └── package.json             // redis操作用にioredisライブラリを使用する
│   │
│   ├────── operate-postgresql.zip   // lambda用zip
│   ├────── operate-redis_get.zip    // lambda用zip
│   └────── operate-redis_set.zip    // lambda用zip
│
└── terraform/
    └── main.tf
```

zipの作成から、リソースの作成まで全てterraformに任せています。

```terraform/main.tf
############################################################################
## terraformブロック
############################################################################
terraform {
  # Terraformのバージョン指定
  # TODO:その日が来たらアプデする
  required_version = "~> 1.7.0"

  # Terraformのaws用ライブラリのバージョン指定
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.33.0"
    }
  }

  # 簡単な検証なので、ローカルにtfstateを保持する
  backend "local" {
    path = "local.tfstate"
  }
}

############################################################################
## providerブロック
############################################################################
provider "aws" {
  # リージョンを指定
  region = "ap-northeast-1"
}

############################################################################
## localsブロック
############################################################################
locals {
  project = "stepfunction-manage-cache"

  lambda_base_path  = "../lambda"

  lambda_redis_functions = ["get", "set"]
  lambda_redis      = "operate-redis"
  lambda_redis_path = "${local.lambda_base_path}/${local.lambda_redis}"

  lambda_postgresql      = "operate-postgresql"
  lambda_postgresql_path = "${local.lambda_base_path}/${local.lambda_postgresql}"

  db_count = 1

  db_username = "testuser"
  db_password = "password"
  db_name     = "test"
}

############################################################################
## VPC
## 特に理由もないので、公式modulesを使用する
## natも不要なので、intra subnetのみ作成する
############################################################################
module "intra_vpc" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "5.9.0" # TODO: providerと一緒に更新する

  name = "${local.project}-intra-vpc"
  cidr = "10.0.1.0/24"

  azs                = ["ap-northeast-1a", "ap-northeast-1c"]
  intra_subnets      = ["10.0.1.0/25", "10.0.1.128/25"]
  enable_nat_gateway = false
}

############################################################################
## Lambda 2つ作る
## 共通で参照するリソース群を定義
############################################################################
# lambda用AWSロール
data "aws_iam_policy" "vpc_access_execution" {
  arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole"
}

data "aws_iam_policy_document" "assume_lambda" {
  statement {
    effect = "Allow"

    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }

    actions = ["sts:AssumeRole"]
  }
}

resource "aws_iam_role" "lambda_vpc" {
  name               = "${local.project}-lambda-vpc-role"
  assume_role_policy = data.aws_iam_policy_document.assume_lambda.json
}

resource "aws_iam_role_policy_attachment" "lambda_vpc" {
  role       = aws_iam_role.lambda_vpc.name
  policy_arn = data.aws_iam_policy.vpc_access_execution.arn
}

# lambda用セキュリティグループ
# VPC内のみ通信できればよい
resource "aws_security_group" "lambda_vpc" {
  name   = "${local.project}-lambda-vpc-sg"
  vpc_id = module.intra_vpc.vpc_id

  tags = {
    Name = "${local.project}-lambda-vpc-sg"
  }
}

resource "aws_security_group_rule" "lambda_vpc" {
  security_group_id = aws_security_group.lambda_vpc.id
  type              = "egress"
  from_port         = 0
  to_port           = 0
  protocol          = "-1"
  cidr_blocks       = [module.intra_vpc.vpc_cidr_block]
}

############################################################################
## Lambda in VPC その１
## redisを操作するlambda関数
## GetとSet用に個別作成する
## 忘れてた。特に理由がないなら性能の高いARM64にすべき。デフォはx86_64
## https://qiita.com/hats_yaki/items/9d7e3522b1158f099645
## 検証後気づいたので、今後気を付ける
############################################################################
# ログ残す
resource "aws_cloudwatch_log_group" "redis" {
  for_each = toset(local.lambda_redis_functions)

  name              = "/aws/lambda/${local.lambda_redis}_${each.key}"
  retention_in_days = 14
}

# zipを作成
data "archive_file" "redis" {
  for_each = toset(local.lambda_redis_functions)

  type             = "zip"
  output_file_mode = "0666"
  source_dir       = "${local.lambda_redis_path}_${each.key}"
  output_path      = "${local.lambda_base_path}/${local.lambda_redis}_${each.key}.zip"
}

resource "aws_lambda_function" "redis" {
  for_each = toset(local.lambda_redis_functions)

  function_name = "${local.lambda_redis}_${each.key}"
  role          = aws_iam_role.lambda_vpc.arn

  runtime  = "nodejs18.x"
  filename = data.archive_file.redis[each.key].output_path
  handler  = "index.handler"

  source_code_hash = filebase64sha256(data.archive_file.redis[each.key].output_path)

  logging_config {
    log_format = "Text"
    log_group  = aws_cloudwatch_log_group.redis[each.key].name
  }

  vpc_config {
    subnet_ids         = module.intra_vpc.intra_subnets
    security_group_ids = [aws_security_group.lambda_vpc.id]
  }

  environment {
    variables = {
      # redisのaddressが評価できた場合、設定。countが0の場合はエラーになるので、tryでcatchしてnullを返却する。
      # coalesceはnullの場合第二引数をとる。冗長なcatchな気がする
      REDIS_HOST = coalesce(try(aws_elasticache_cluster.redis[0].cache_nodes[0].address, null), "DUMMY_REDIS_HOST")
    }
  }

  # zip内の変更は無視する
  lifecycle {
    ignore_changes = [filename, source_code_hash]
  }
}

############################################################################
## Lambda in VPC その２
## postgresqlを操作するlambda関数
############################################################################
# ログ残す
resource "aws_cloudwatch_log_group" "postgresql" {
  name              = "/aws/lambda/${local.lambda_postgresql}"
  retention_in_days = 14
}

# zipを作成
data "archive_file" "postgresql" {
  type             = "zip"
  output_file_mode = "0666"
  source_dir       = local.lambda_postgresql_path
  output_path      = "${local.lambda_base_path}/${local.lambda_postgresql}.zip"
}

resource "aws_lambda_function" "postgresql" {
  function_name = local.lambda_postgresql
  role          = aws_iam_role.lambda_vpc.arn

  runtime  = "nodejs18.x"
  filename = data.archive_file.postgresql.output_path
  handler  = "index.handler"

  source_code_hash = filebase64sha256(data.archive_file.postgresql.output_path)

  logging_config {
    log_format = "Text"
    log_group  = aws_cloudwatch_log_group.postgresql.name
  }

  vpc_config {
    subnet_ids         = module.intra_vpc.intra_subnets
    security_group_ids = [aws_security_group.lambda_vpc.id]
  }

  environment {
    variables = {
      # rdsのaddressが評価できた場合、設定。countが0の場合はエラーになるので、tryでcatchしてnullを返却する。
      # coalesceはnullの場合第二引数をとる。冗長なcatchな気がする
      DB_HOST     = coalesce(try(aws_db_instance.postgresql[0].address, null), "DUMMY_HOST") # port不要なのでaddressを取得する。endpointだとportもついてくる
      DB_USER     = coalesce(try(aws_db_instance.postgresql[0].username, null), "DUMMY_DB_USER")
      DB_PASSWORD = coalesce(try(aws_db_instance.postgresql[0].password, null), "DUMMY_DB_PASSWORD") # 簡単のため今回はこの方法で参照するが、tfstateに記載されるので、本来はsecret managerから取得するべき
      DB_DATABASE = coalesce(try(aws_db_instance.postgresql[0].db_name, null), "DUMMY_DB_DATABASE")
    }
  }

  # zip内の変更は無視する
  lifecycle {
    ignore_changes = [filename, source_code_hash]
  }
}

############################################################################
## ElastiCache SSO Redis
## valkeyはドキュメント少なそうだったので、とりあえずRedisを選択
## クラスタなどは不要なので最小構成で作成する
############################################################################
resource "aws_elasticache_subnet_group" "redis" {
  name       = "redis-cache-intra-subnet"
  subnet_ids = module.intra_vpc.intra_subnets
}

resource "aws_security_group" "redis" {
  name   = "${local.project}-redis-sg"
  vpc_id = module.intra_vpc.vpc_id

  tags = {
    Name = "${local.project}-redis-sg"
  }
}

resource "aws_security_group_rule" "redis" {
  security_group_id        = aws_security_group.redis.id
  type                     = "ingress"
  from_port                = 6379
  to_port                  = 6379
  protocol                 = "TCP"
  source_security_group_id = aws_security_group.lambda_vpc.id
}

resource "aws_elasticache_cluster" "redis" {
  # 節約のため不要時は削除
  count = local.db_count

  cluster_id           = "${local.project}-cluster-example"
  engine               = "redis"
  node_type            = "cache.t2.medium" # aws_elasticache_cluster.redis: Creation complete after 5m17s
  num_cache_nodes      = 1
  parameter_group_name = "default.redis7"
  engine_version       = "7.1"
  port                 = 6379
  subnet_group_name    = aws_elasticache_subnet_group.redis.name
  security_group_ids   = [aws_security_group.redis.id]
}

############################################################################
## RDS postgresql
## エンジンはなんでもいいので、なんとなくpostgresql
## クラスタなどは不要なので最小構成で作成する
############################################################################
resource "aws_db_subnet_group" "postgresql" {
  name       = "${local.project}-intra-subnet"
  subnet_ids = module.intra_vpc.intra_subnets

  tags = {
    Name = "${local.project}-intra-subnet"
  }
}

resource "aws_security_group" "postgresql" {
  name   = "${local.project}-postgresql-sg"
  vpc_id = module.intra_vpc.vpc_id

  tags = {
    Name = "${local.project}-postgresql-sg"
  }
}

resource "aws_security_group_rule" "postgresql" {
  security_group_id        = aws_security_group.postgresql.id
  type                     = "ingress"
  from_port                = 5432
  to_port                  = 5432
  protocol                 = "TCP"
  source_security_group_id = aws_security_group.lambda_vpc.id
}

resource "aws_db_instance" "postgresql" {
  # 節約のため不要時は削除
  count = local.db_count

  #################################################
  ## インスタンス基本設定
  #################################################
  identifier             = "${local.project}-rds-postgresql"
  engine                 = "postgres"
  engine_version         = "16.3"        # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/PostgreSQL.Concepts.General.DBVersions.html
  instance_class         = "db.t3.micro" # aws_db_instance.postgresql: Creation complete after 4m47s 
  vpc_security_group_ids = [aws_security_group.postgresql.id]

  #################################################
  ## DBアプリ設定
  #################################################
  db_name = local.db_name

  #################################################
  ## ストレージ設定
  #################################################
  storage_type      = "gp2"
  storage_encrypted = false
  allocated_storage = 10
  # ストレージ自動スケーリング上限（GB）
  max_allocated_storage = 30

  #################################################
  ## ログイン情報
  ## adminは基本的にアプリに使用しないが、簡単のため
  #################################################
  username = local.db_username
  password = local.db_password
  port     = 5432

  #################################################
  ## ネットワーク
  #################################################
  publicly_accessible  = false
  db_subnet_group_name = aws_db_subnet_group.postgresql.name
  multi_az             = false

  #################################################
  ## DBインスタンス管理
  #################################################
  backup_window = "09:10-09:40"
  # アップデートの実行を次のメンテナンスウィンドウまで待機
  apply_immediately          = false
  maintenance_window         = "mon:10:10-mon:10:40"
  auto_minor_version_upgrade = false

  #################################################
  ## 削除保護
  #################################################
  deletion_protection      = false
  skip_final_snapshot      = true
  delete_automated_backups = false
  backup_retention_period  = 0
}

############################################################################
## Step Functions
## 本体はGUIで作成するので、ロールだけ用意
## 作業証跡を残すため、sfn本体も一時的に定義
############################################################################
# Step Functions用のIAMロールを作成
data "aws_iam_policy_document" "assume_sfn" {
  statement {
    effect = "Allow"

    principals {
      type        = "Service"
      identifiers = ["states.amazonaws.com"]
    }

    actions = ["sts:AssumeRole"]
  }
}

resource "aws_iam_role" "sfn" {
  name               = "${local.project}-sfn-role"
  assume_role_policy = data.aws_iam_policy_document.assume_sfn.json
}

data "aws_iam_policy_document" "sfn" {
  statement {
    effect = "Allow"
    actions = [
      "logs:CreateLogGroup",
      "logs:CreateLogStream",
      "logs:PutLogEvents",
      "logs:DescribeLogGroups",
      "logs:DescribeLogStreams",
    ]
    resources = ["*"]
  }
  statement {
    effect = "Allow"
    actions = [
      "lambda:InvokeFunction",
    ]
    resources = concat(
      [for lambda in aws_lambda_function.redis : "${lambda.arn}:*"],
      ["${aws_lambda_function.postgresql.arn}:*"]
    )
  }
}

resource "aws_iam_policy" "sfn" {
  name        = "${local.project}-sfn-policy"
  description = "IAM policy for Step Functions"
  policy      = data.aws_iam_policy_document.sfn.json
}

resource "aws_iam_role_policy_attachment" "sfn" {
  role       = aws_iam_role.sfn.name
  policy_arn = aws_iam_policy.sfn.arn
}
```

```lambda/operate-postgresql/package.json
{
  "name": "operate-postgres",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "author": "",
  "license": "ISC",
  "dependencies": {
    "pg": "^8.13.0"
  }
}
```

```js
// lambda/operate-postgresql/index.mjs
export const handler = async (event, context) => {
    console.log("========================================");
    console.log(event);
    console.log("========================================");

    return {
        "statusCode": "200",
        "body": JSON.stringify({ "test": "value" })
    };
};
```

getもsetも同一の内容。あとでいろいろコードは変更します。

```lambda/operate-redis_get/package.json
{
  "name": "operate-redis",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "author": "",
  "license": "ISC",
  "dependencies": {
    "ioredis": "^5.4.1"
  }
}
```

```js
// lambda/operate-redis_get/index.mjs
export const handler = async (event, context) => {
    console.log("========================================");
    console.log(event);
    console.log("========================================");

    return {
        "statusCode": "200",
        "body": JSON.stringify({ "test": "value" })
    };
};
```
</details>

### リソースの確認
terraformを実行し、リソースを作成します

#### RDS

![cap4.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c2e941e2-a6cb-bf8c-22db-d0d9dc00ca47.png)

元気そうですね

#### Elasticache

![cap3.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/efd37bd8-ff81-9126-2130-e6e109260b78.png)

大丈夫そうですね

#### lambdaたち

![cap72.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c77f5e7f-230e-c830-b074-4e6f76f7821a.png)

しっかり作成されています。
コードを編集し、DBを操作してみます。

#### redisを操作する
index.mjsを以下のように編集します。
```js
import { Redis } from "ioredis"

// ElastiCacheの接続情報
const redisConfig = {
    host: process.env.REDIS_HOST,
    port: 6379,
};

const redisClient = new Redis(redisConfig);

export const handler = async (event, context) => {
    const res = await redisClient.set("test", "asfdaaaaa")
    console.log(res)
    console.log(111, await redisClient.get("test"));
    console.log(222, await redisClient.get("testaaaa")); // 存在しない場合、nullが返却される

    return {
        "statusCode": "200",
        "body": JSON.stringify({ "test": "value" })
    };
};
```

実行してみます。

![cap5.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/a954e5d2-97bc-3669-eac8-95796e577f14.png)

![cap6.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/8729b150-acda-8c12-24e0-5a999f9acab6.png)

ちゃんと疎通していることと、
データが存在しない場合、**null**が返却されることがわかりました。

動作確認が終わったので、以下のように編集し、準備完了です。

```js
// operate-redis_get/index.mjs
import { Redis } from "ioredis"

// ElastiCacheの接続情報
const redisConfig = {
    host: process.env.REDIS_HOST,
    port: 6379,
};

// ウォームスタート時にコネクションを再利用する
const redisClient = new Redis(redisConfig);

export const handler = async (event, context) => {
    const key = event.email
    console.log(`key is ${key}`)
    const result = await redisClient.get(key) // 存在しない場合、nullが返却される

    return {
        "cache": JSON.parse(result),
    };
};
```


```js
// operate-redis_set/index.mjs
import { Redis } from "ioredis"

// ElastiCacheの接続情報
const redisConfig = {
    host: process.env.REDIS_HOST,
    port: 6379,
};

// ウォームスタート時にコネクションを再利用する
const redisClient = new Redis(redisConfig);

export const handler = async (event, context) => {
    const user = event.user
    console.log(`user is`)
    console.log('%o', user)

    const result = await redisClient.set(user.email, JSON.stringify(user))

    return {
        user
    };
};
```

#### postgresqlを操作する

index.mjsを以下のように編集します。

```js
import pg from 'pg';
const { Client } = pg;

const client = new Client({
    host: process.env.DB_HOST,
    user: process.env.DB_USER,
    database: process.env.DB_DATABASE,
    password: process.env.DB_PASSWORD,
    port: 5432,
    ssl: {
        rejectUnauthorized: false // 注意: 本番環境では適切な証明書を使用すること
    }
});

const setup = async () => {
    await client.connect();

    // テーブルの作成 (既に存在する場合はスキップ)
    await client.query(`
        CREATE TABLE IF NOT EXISTS users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100) NOT NULL,
            email VARCHAR(100) UNIQUE NOT NULL
        )
    `);

    // 3件のデータを挿入
    const users = [
        ['John Doe', 'john@example.com'],
        ['Jane Smith', 'jane@example.com'],
        ['Bob Johnson', 'bob@example.com']
    ];

    for (const [name, email] of users) {
        const insertResult = await client.query(
            'INSERT INTO users (name, email) VALUES ($1, $2) ON CONFLICT (email) DO NOTHING RETURNING id',
            [name, email]
        );
        if (insertResult.rows.length > 0) {
            console.log('Inserted user ID:', insertResult.rows[0].id);
        } else {
            console.log('User already exists:', email);
        }
    }
}

export const handler = async (event, context) => {
    try {
        await setup();

        // データの選択
        const selectResult = await client.query('SELECT * FROM users');
        console.log('Current users in the table:', selectResult.rows);

        return {
            statusCode: 200,
            body: JSON.stringify({ 
                message: "Operation completed successfully",
                users: selectResult.rows
            })
        };
    } catch (error) {
        console.error('Error:', error);
        return {
            statusCode: 500,
            body: JSON.stringify({ error: "An error occurred during execution" })
        };
    } finally {
        await client.end();
    }
};
```

実行してみます。

![cap7.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/8feeb64b-93d3-acd3-0e82-cd90fe2233cd.png)

![cap8.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/513c0986-cc0a-1694-77d2-9f0c934a8ee6.png)

```
users テーブル:
+----+-------------+--------------------+
| id | name        | email              |
+----+-------------+--------------------+
|  1 | John Doe    | john@example.com   |
|  2 | Jane Smith  | jane@example.com   |
|  3 | Bob Johnson | bob@example.com    |
+----+-------------+--------------------+
(3 行)
```


ちゃんと疎通していることがわかりました。

動作確認が終わったので、以下のように編集し、準備完了です。

```js
import pg from 'pg';
const { Client } = pg;

const client = new Client({
    host: process.env.DB_HOST,
    user: process.env.DB_USER,
    database: process.env.DB_DATABASE,
    password: process.env.DB_PASSWORD,
    port: 5432,
    ssl: {
        rejectUnauthorized: false // 注意: 本番環境では適切な証明書を使用すること
    }
});

// ウォームスタート時にコネクションを再利用する
await client.connect()

export const handler = async (event, context) => {
    const email = event.email
    console.log(`email is ${email}`)
    
    // event.emailをキーにして1件のユーザー情報を取得
    const selectResult = await client.query('SELECT * FROM users WHERE email = $1', [email]);
    console.log(selectResult)

    return {
        user: selectResult.rows[0]
    };
};
```

今回、try-catch句はあえてjs側では実装しませんでした。
そこらへんは全てAWS Step Functionsに任せる方が、シンプルかなーと思いました。
意見あったらコメント欲しいです！

#### AWS Step Functions用のIamロール

ロールの存在を確認します。

![cap2.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/068b4c1e-1f04-adc1-65c4-56eb964f219f.png)

良い感じですね。
（出力しませんが）logsとlambda発火の権限を持っていることを確認します。


# AWS Step Functions
では、ここからが本番です。AWS Step Functionsを作成します。

## ステートマシンを作成する

![cap9.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/4f9784aa-3796-d6d7-f9ec-250d980ed49b.png)

コンソールから作成していきます。
今すぐ始める をクリックします。

![cap10.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/22ebfcd3-7410-8a44-c622-bc71d599ff70.png)

テンプレートを選べます。
今回は 自分で作成する をクリックします。

## ステートマシンをいったん保存する

![cap71.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/edc48106-08df-12a6-efc0-f2e74afb25bd.png)

作成されたステートマシンが表示されます。このままでは保存できないので、編集します。

![cap16.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/d6933633-2115-b3b1-db58-177686f7c724.png)

フロータブへ移動します。

![cap17.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/aa988487-a4fd-1fbe-65d5-89912edb23f7.png)

Success を追加します。これで保存可能ですが、ステートマシンの名前や、ロールなどを設定します。

![cap11.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/1fd6ed9f-47d5-1f36-3016-dc5eb1ee0c50.png)

設定タブを選択します。
ステートマシンの名前や、ロールなど、その他の設定を編集します。

![cap12.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/3676cecf-aed9-27d6-549f-edde53e9089b.png)

ステートマシン名を入力します。任意の名前で良いので、適当。

![cap13.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/5696c535-3f04-7971-5806-dee280b4797c.png)

ロールを選択します。Terraformで用意したものを選択します。

![cap14.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/7e2f77c1-d3ff-9c24-c853-f832da8c348b.png)

その他設定はいったんなしにしておきます。X-Rayと統合されてるの良いですね。

![cap15.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/fd97891f-df49-8ac0-0e5f-926a3f60c3b9.png)

作成 をクリックします。

![cap18.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/3c77a656-43bd-3016-2be6-72e2b19dbb63.png)

保存されました！！

## ステートマシンを実行してみる
実行してどんな感じか体験してみます。

![cap19.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/b3fd2558-ded7-c83f-2008-8c27a948b444.png)

右上の 実行を開始 をクリックします。

![cap20.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/1ea06e4e-7f21-0497-aa91-4a42bbc78f7c.png)

input Jsonが要求されます。何も設定せず、 実行を開始 をクリックします。

![cap22.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/886e60f7-eaf9-0a47-5b27-5c030e12630e.png)

フローが実行されました！
こんな感じで、フローをGUIで編集、Inputを指定して実行、を繰り返して開発していきます。

ではさっそくlambdaとつないでいきましょう！！

## キャッシュを取得する
まずはキャッシュ取得処理を追加します！

![cap23.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c315fbc9-c473-5fa8-4a3c-39226455b54f.png)

アクションタブから、 AWS Lambda Invoke アクションを選択します。
見ての通り、Lambdaを実行するアクションですね。
配置したら、アクションの設定を行っていきます。

![cap24.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/0d4b25c5-b7a9-703d-471f-7fd154440423.png)

ラベルと実行するlambda関数を設定します。
キャッシュを取得するので、「Query Cache」、関数は作成した redis_get を指定します。

![cap25.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c0b850f8-f9f9-1d8b-3f9f-2d5806720417.png)

保存してテストしましょう。

```js
export const handler = async (event, context) => {
    const key = event.email
    console.log(`key is ${key}`)
    const result = await redisClient.get(key) // 存在しない場合、nullが返却される

    return {
        "cache": JSON.parse(result),
    };
};
```

redisからキャッシュをgetするlambdaのコードの一部を再掲します。
event.emailをキーにしているので、Input Jsonにemailを含めてあげます。

![cap26.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/f70dcab7-d01e-3fb6-8e82-eb660ee21ffe.png)

こんな感じです！
先ほどlambdaとredisの疎通確認でtestをキーにvalueを保存していたので、取得されてほしいところです。
実行してみます。

![cap27.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/8fe3d94d-9c69-d709-cf83-cb28ebff058c.png)

lambdaの実行が成功しました。ロールは正しく設定されていそうです。
気になる出力を確認します。

![cap28.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/a55d08fe-10e8-b3f2-39b7-c659ec8ba511.png)

cacheに先ほどテスト用にsetした値が返ってきてますね！！
では、存在しないキーでも検索してもらいましょう。

![cap30.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/642b55c8-9aca-3615-2284-f52d83f23656.png)

Input Jsonでemailに存在しないキーを設定してみます。
実行を開始！！

![cap31.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/9f8b8078-3bb0-e6d2-bf7f-1b7d2fc59306.png)

lambdaが実行され、nullが返却されました。
いいですね！

では次は、lambdaの実行結果でif分岐するようにフローを編集します。

## Choiceフローを追加する
Step Functionsでは、Choiceという部品で分岐ロジックを実装できます。

![cap32.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/9c8d0f36-368e-9092-3292-2e0efa998a9e.png)

フローからChoice を選択し、lambda関数につなげます。

![cap33.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/a9ba02af-6c92-9ca7-35b8-04965881823a.png)

Choice Rulesを編集します。if文の評価式をいじるイメージです。

![cap34.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/1a917f8f-3afe-d2a8-234a-609d936d946f.png)

Add Conditionsをクリックします。

![cap35.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c45127f4-5817-fd5a-a2f6-b78437ee8ece.png)

$はinputを意味します。この場合はlambda関数のアウトプットですね。
画像の設定では、lambda関数のアウトプットのcacheが、Nullでなければ、という条件になります。
条件を保存します。

![cap40.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/ad0345ff-e560-b6db-96bc-41f3a2667115.png)

その後、next state を 新しい状態を追加 に設定します。

![cap38.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/9b6f74be-c96f-1ca6-00ab-bae1fb358bd0.png)

Choiceの下に、新しくスペースがうまれました。

![cap39.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/abd2577d-93f4-fdac-a754-c2d43a8d999e.png)

まずは分岐するかどうかを確認したいので、lambdaを設定せず、Passというデータを変換するフローを置いてみます。

![cap41.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/fe2ae979-8367-91cc-a4a5-428c2073a9ce.png)

Passの名前を変更して、どっちに分岐したかわかりやすくします。
Nullじゃなかった場合に通るフローなので、 Cache Hit と設定します。

Nullだった場合の分岐も作ります。
Choiceを再度編集します。

![cap36.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/e05c0488-b265-46c3-b4c0-130872cd859f.png)

Add new choice rule をクリックします。elseの役割をもつDefault ruleでも大丈夫です。

![cap37.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/a30f5a29-cdf6-8c15-2969-87db80b0cb3c.png)

2つ目のRuleを指定します。
cacheが存在しない場合（is present === false）または、Nullの場合、こちらに分岐させます。

![cap42.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/9f6e08e0-f337-e5a0-2ce1-4db8feaca04a.png)

同様に後続にPassを追加し、分岐ロジックを設定できました。

![cap43.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/24426a9d-9689-8c2d-a6f6-b58f0441ba51.png)

テストします！！

```input.json
{
	"email" : "test"
}
```
で実行しました。

![cap44.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/090f2551-afae-eaf6-a114-66282f79d2f0.png)

![cap45.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/9a8061e0-d4c7-8668-3ddf-7029f9a8e4bf.png)

いい感じです。

```input.json
{
	"email" : "sonzai shinai key"
}
```
で実行しました。

![cap46.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/92dc71ed-5247-069b-ce8d-72e9f351370a.png)

こっちもいい感じ！

if分岐が実装できたので、次は、キャッシュが存在しない場合、実際にRDSへクエリするようフローを編集していきます！

## RDSをクエリするlambdaを追加する

![cap47.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/a32b1f42-e84b-8300-4ee8-3b7554111221.png)

キャッシュが存在しなかった場合のPassの後続に、Invoke Lambdaを追加します。
わかりやすい名前と、関数の選択をします。

![cap48.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/6b645a94-e7e4-59ac-52d4-d52727c9fede.png)

lambdaのevent引数に渡す値を設定します。
Step Functionsはデフォルトでは一つ前のアクションのアウトプットが次のアクションのインプットになります。
このlambdaの場合はPassのアウトプットの ```{ "cache" : null}``` がインプットですね。
ただ、これをインプットにされても、クエリするべきemailが分かりません。
そこで、payloadを入力し、Passのアウトプットではない値をlambdaに渡してあげます。

![cap49.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c59a1144-29ee-bb62-3773-dcd759a8432c.png)

Input Jsonの中身は、```$$.Execution.Input``` で参照できます。
今回はemailを参照したいので、```$$.Execution.Input.email``` としました。
キーが「.$」で終わる必要があると記載のあるので、キーは ```user.$``` とします。

https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/input-output-contextobject.html

では実行してみましょう！！

![cap53.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/d3304f5d-92f7-43cf-6619-ac53c54380af.png)

RDSにレコードが存在するemailを指定します。

![cap50.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/455a37c0-8db5-1c02-e7c6-4809ba9cbd70.png)

きたーーーちゃんとクエリできてますね。
これでキャッシュミス時のオリジンRDSへのクエリ処理が実装できました。
次は、オリジンへのクエリ後、キャッシュを更新する処理を実装しましょう。
lambdaつなぐだけなので、Step Functionsなら簡単です！！！

## キャッシュを追加するlambdaを追加する

![cap52.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/c44cf3a5-f42e-2880-263c-5f175469e1a1.png)

Incoke Lambdaを追加します。
今回はひとつ前のlambdaのアウトプットをそのまま渡せばオッケーなので、他の操作は行いません。
テストします！！！！！

![cap53.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/fb29961d-41a4-c313-cc49-f1fcdf74ece0.png)

レコードの存在するemailを設定し、実行します。

![cap54.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/aea428d7-d24e-1f5c-849c-6c55bf9f7eeb.png)

RDSの検索結果が返却されました！！！
lambdaが落ちていないので、キャッシュは追加されたように見えます。

もう一度同じemailでステートマシンを実行します。

![cap55.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/76e9aa54-0e9d-958e-6eef-a964dfe4bd29.png)

きたーーーーーーーーキャッシュから取得し、RDSは触ってないですね！！
ただ。キー名がcache になってますね。オリジンからクエリしたときと同様の形式にしましょう。

![cap57.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/0124ab06-ac12-9fa2-557c-df26a0a30c4d.png)

Cache Hit時のPassを編集します。
Passは入力値を整形してアウトプットする使い方もできます。

入力の一部を使用して新しいJSONペイロードを構築、と書かれてますね。
Parametersを使用して入力を変換 にチェックをいれます。

![cap58.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/babad319-7543-fead-d0d6-8a89dd62f073.png)

```{ "user" : { ... }}``` の形式になるよう設定します。

再度同じemailでステートマシンを実行してみます。

![cap59.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/b0e5d436-a1d8-b8b3-49ce-66fe5f17bfb3.png)

おっけーですね！！！
RDSへクエリした時と同じ形式で返却されました。

![cap61.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/04e70997-c03c-6e65-bc12-91029a8bb392.png)

最後に、実行時間の差も見てみましょう。
set処理が同期的に実行されるため正しい比較ではないですが、結構違いますね。redisはやい！

以上で、今回作りたかったRDS × ElastiCache のキャッシュ処理フローは完成です！

# Step Functionsでフローを構築するメリットとデメリット
最後に、Step Functionsでフローを構築するメリットとデメリットをまとめてみます。

## メリット
- フローの流れを一望できる
Step FunctionsのGUIのおかげで、
各サービスがどのように処理されるか/エラーハンドリングされるか？はかなり分かりやすいと思います。
SNSやSQSで繋がってるフローであれば比較的楽に把握できるかもしれませんが、
eventbridgeが待機しているフローなどの全容の把握ってかなり難しいですよね。

![cap60.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654730/2da7d60e-d538-2508-fbbb-d4d672d508bd.png)

Step Functionsなら視覚的に把握できます！！

- フロー全体のエラーハンドリングを簡単に実装できる
記事引用で恐縮なのですが・・・。

https://kakakakakku.hatenablog.com/entry/2024/10/10/073421

上記ブログ記事の通り、step functions内のフロー全体をまるまるキャッチする仕組みを構築できます。
とりあえずSNSに送る！みたいな運用が可能で、どんなエラーが起こるかわからないローンチ初期などは、こういったキャッチ機構が存在すると嬉しいのではと思います。

## デメリット
- JSON pathの勉強が必要
理解してるような、してないような。自信がない・・・。

- Step Functionsに多くの権限が付与される
ゴッドクラス・ゴッドメソッドのような問題が発生するのでは、と思っています。
最小権限の法則とは若干喧嘩してしまうなーと感じます。

## 総論
デメリットはありつつ、いろんなサービスを統合するフローをGUIで作成できるのはかなり良いと感じます。
様々な使い方ができるサービスだと思うので、慣れていきたいですねー

## おまけ
今回の作業リポジトリ

https://github.com/suiwave/sfn-manage-cache