CDKを使ってAWS Lambda, Amazon RDS, SESでメール通知できるようにする

RDSにあるデータを読み取ってLambdaでメール通知をするシステムをAWS CDKで作ってみました。

SESを使うかSNSで使うか悩んだのですが、両方試してみてSESの方がカスタマイズ性に優れていて今回作りたかったものに適していると感じたのでSESにしました。

Amazon SNSとAmazon SESについて

Amazon SES・・・Amazonが提供するフルマネージド型のメール配信サービスです。メールサーバを構築せずともSESにリクエストを送ることでメールを送信します。
ユーザー自身の E メールアドレスとドメインを使用してEメールを送受信する簡単でコスト効率の高いメール機能に特化したサービスになります。

Amazon SNS・・・メッセージングやプッシュ通知向けのフルマネージドサービスになります。あるイベントをトピックとして作成し、メールアドレスなどをサブスクリプションとします。イベントが発生するとサブスクリプションしている通知先に通知を送ります。
SNSではイベントと通知相手（モバイルプッシュ通知、モバイルテキストメッセージ (SMS)、メールなど）を柔軟に選択できます。

作成した構成について

簡単な構成図は以下になります。
RDSからデータを読み取りLambdaを経由してSESでメールを送信します。Event bridgeを利用して1日1回メール通知をするようにしました。

RDSがプライベートサブネットにあるのでLambdaもプライベートサブネットに配置する必要があります。VPCエンドポイントを利用して繋げようとしたのですが、Lambdaから接続する場合はAPIエンドポイントを利用するためNat Gatewayを利用する必要がありました。VPCエンドポイントを利用して接続した場合、timeoutになってしまいます。

CDKのコード

以下がCDKのコードです。VPCやRDSは別スタックで作成したものを利用しています。
Lambda, SES, Eventbridgeの作成と紐付け、および権限付与をしています。
LambdaがRDSからデータを取得するためにはPrivate Subnetにある必要があるのでサブネットの指定もしています。

CDKのコード抜粋

export class Lambda extends Resource {
  constructor(
    scope: Construct,
    id: string,
    {vpc, cluster}: Props
  ) {
    super(scope, id);

    const rdsSecret = secretsmanager.Secret.fromSecretCompleteArn(this, 'RdsSecret', process.env.SECRET_COMPLETE_ARN!);

    // セキュリティグループ
    const lambdaSecurityGroup = new ec2.SecurityGroup(this, 'LambdaSecurityGroup', {
      vpc,
      allowAllOutbound: true,
    });

    // NatGateway
    const natGateway = new ec2.CfnNatGateway(this, 'NatGateway', {
      subnetId: vpc.publicSubnets[0].subnetId,
      allocationId: new ec2.CfnEIP(this, 'EIP', {}).attrAllocationId,
    });

    // ルートテーブル
    vpc.privateSubnets.forEach((subnet, index) => {
      new ec2.CfnRoute(this, `PrivateSubnetRoute${index}`, {
        routeTableId: subnet.routeTable.routeTableId,
        destinationCidrBlock: '0.0.0.0/0',
        natGatewayId: natGateway.attrNatGatewayId,
      });
    });

    // lambda
    const lambdaFunction = new aws_lambda_nodejs.NodejsFunction(this, "notificationLambda", {
      entry: join(__dirname, '../../lambda/index.ts'),
      runtime: lambda.Runtime.NODEJS_20_X,
      handler: 'handler',
      vpc,
      vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS },
      securityGroups: [lambdaSecurityGroup],
      environment: {
        SENDER_EMAIL: process.env.SENDER_EMAIL!,
        REGION: region,
        DB_HOST: process.env.POSTGRES_HOST!,
        DB_PORT: process.env.POSTGRES_PORT!,
        DB_USER: 'postgres',
        DB_NAME: 'postgres',
        DB_PASSWORD: process.env.POSTGRES_PASSWORD!,
        DB_SECRET_ARN: rdsSecret.secretArn,
      },
    });

    // EventBridge
    const rule = new events.Rule(this, 'Rule', {
      schedule: events.Schedule.cron({ minute: '30', hour: '9' }),
    });
    rule.addTarget(new targets.LambdaFunction(lambdaFunction));
    
    rdsSecret.grantRead(lambdaFunction)
    cluster.connections.allowDefaultPortFrom(lambdaFunction)
    lambdaFunction.addToRolePolicy(new iam.PolicyStatement({
      actions: ['ses:SendEmail', 'ses:SendRawEmail', 'ses:CreateEmailIdentity', 'ses:ListIdentities', 'ses:*'],
      resources: ['*'],
    }));
  }
}

SESで送信するLambdaのコード

Lambdaのコードが以下になります。
今回はメールアドレスごとに認証をしたのでexistingEmailsで認証されたメールアドレスを調べています。データベースから取得したメールアドレスが認証されていなかった場合は、認証メールを最初に送ります。
メールの本文は個別に変えて、メールアドレスごとに配信できています。

export const handler = async () => {
  try {
    const region = process.env.REGION
    const sesClient = await new SESv2Client({ region });
    
    const client = new Client({
      host: process.env.DB_HOST,
      user: process.env.DB_USER,
      password: process.env.DB_PASSWORD,
      database: process.env.DB_NAME,
      port: 5432,
    });

    const sql = "SELECT name, email FROM users"

    // rdsとの接続
    await client.connect();
    
    const res = await client.query(sql);
    const data = res.rows

    // 既存のSES Email Identitiesを取得
    const listEmailIdentitiesCommand = new ListEmailIdentitiesCommand({});
    const existingIdentities = await sesClient.send(listEmailIdentitiesCommand);
    const existingEmails = existingIdentities.EmailIdentities?.map(identity => {
      if (identity.IdentityType === 'EMAIL_ADDRESS' && identity.SendingEnabled) return identity.IdentityName
      });
  
    for (let i = 0; i<data.length;i++) {
      // Emailの認証
      if (existingEmails && existingEmails.length > 0 && !existingEmails.includes(data[i]['email'])) {
        // 未認証の場合
        const createEmailIdentityCommand = new CreateEmailIdentityCommand({
          EmailIdentity: data[i]['email'],
        });
        await sesClient.send(createEmailIdentityCommand);
      }　
        const emailParams = {
          FromEmailAddress: process.env.SENDER_EMAIL,
          Destination: {
            ToAddresses: [`${data[i]['email']}`],
          },
          Content: {
            Simple: {
              Subject: {
                Data: 'テストメール',
              },
              Body: {
                Text: {
                  Data: `お疲れ様です。テストメールです。`,
                }
              },
            },
          }
        };
        await sesClient.send(new SendEmailCommand(emailParams));
    }
    await client.end();

    return {
      statusCode: 200,
      body: JSON.stringify({ message: 'Email sent successfully' }),
    };
  } catch (error) {
    return {
      statusCode: 500,
      body: JSON.stringify({ message: 'Failed', error }),
    };
  }
};

SNSで送信するLambdaのコード

SNSでも同じことを試してみました。
トピックをサブスクライブしている人にしかメールが送れないので、サブスクライブしているか確認し、まだの人にはサブスクライブしてもらうメールを送信、サブスクライブ済みの人には送りたいメールを送信しています。

SNSだとダメだった点は個別にメールを送れないことです。

SNSはトピックごとなので同じトピックにをサブスクライブしている人には個別にメールを送れないです、、、
メッセージを個別に作成してループを回すので人数分のメールが全員に送信されてしまいました。

個別にトピックを作成して送信することも考えましたが、あまり得策ではない気がしてSESで作成しました。

export const handler = async () => {
  try {
    const region = process.env.REGION;

    const client = new Client({
      host: process.env.DB_HOST,
      user: process.env.DB_USER,
      password: process.env.DB_PASSWORD,
      database: process.env.DB_NAME,
      port: 5432,
    });

    const sql =
      "SELECT name, email FROM users";

    await client.connect();

    const res = await client.query(sql);
    const data = res.rows;

    for (let i = 0; i < data.length; i++) {
        const sns = new AWS.SNS();
        const subscriptions = await sns
          .listSubscriptionsByTopic({ TopicArn: process.env.SNS_TOPIC_ARN })
          .promise();
        const existingSubscription = subscriptions.Subscriptions.find(
          (sub: any) =>
            sub.Endpoint === data[i]["email"] &&
            sub.SubscriptionArn !== "PendingConfirmation"
        );
        // サブスクライブがまだの場合はサブスクライブの設定
        if (!existingSubscription) {
          const subscribeParams = {
            Protocol: "email",
            TopicArn: process.env.SNS_TOPIC_ARN,
            Endpoint: data[i]["email"],
          };
          await sns.subscribe(subscribeParams).promise();
        }
        // 認証済みの人にはメールを送る
        const publishParams = {
          Message: getMessage(data[i]["name"]),
          Subject: "お知らせ",
          TopicArn: process.env.SNS_TOPIC_ARN,
          MessageAttributes: {
            email: {
              DataType: "String",
              StringValue: data[i]["email"],
            },
          },
        };
        await sns.publish(publishParams).promise();
    }
    await client.end();

    return {
      statusCode: 200,
      body: JSON.stringify({ message: "Email sent successfully" }),
    };
  } catch (error) {
    return {
      statusCode: 500,
      body: JSON.stringify({ message: "Failed", error }),
    };
  }
};

const getMessage = (name: string) => {
  return `こんにちは ${name} さん,
\n
ご不明な点がございましたら、いつでもお問い合わせください。

よろしくお願いいたします。`;
};

所感

SNSとSESの違いやユースケースについてどちらを使うべきか迷うこともあると思います。SNSの方が使う機会は多い気はしますが、メールに特化した機能を作成したいときやメールで色々カスタマイズしたいときはSESで簡単な通知機能を作りたいときはSNSでいいのかなと他の記事でもありましたが、改めて感じました。

VPCエンドポイントでかなり詰まってしまっていたのでいい勉強になりました。