前提

通知先デバイスのOSはAndroid
通知先デバイスは1台　※複数台に通知する場合はトピックを使用
Mobileアプリの環境構築、実装については本記事では言及しない

構成要素

Amazon SNSを使用してプッシュ通知を行う場合、以下の構成となる

準備

Amazon SNSからFCM経由でMolblieアプリにプッシュ通知を行うためには、
通知先のデバイスを特定するためのデバイストークンと、
Firebaseのプロジェクトを特定するためのサーバーキーが必要となる
そのため以下の準備を行う

Firebaseに任意のプロジェクトを作成
Firebase Cloud MessagingにMoblieアプリを登録
Firebaseの以下からサーバーキーを取得する

Mobileアプリからデバイストークンを取得

※取得方法については参考記事等を参照
※デバイストークンはMobileアプリのインストール毎に変化する

マネジメントコンソールでの環境構築

プッシュ通知から、
プラットフォームアプリケーションの作成をクリック

任意のアプリケーション名を入力
プッシュ通知プラットフォームはFCMを選択
APIキーに準備で取得したサーバーキーを入力
プラットフォームアプリケーションをクリック

アプリケーションエンドポイントの作成をクリック

デバイストークンに準備で取得したデバイストークンを入力
アプリケーションエンドポイントの作成をクリック

マネジメントコンソールからのプッシュ通知テストでは、
配信プロトコルごとにカスタムペイロードを選択

メッセージの内容は、
Mobileアプリがバックグラウンドでもプッシュ通知するためには以下のようなフォーマットとなる
※android_channel_idはMobileアプリの実装に合わせる

{
  "GCM": "{ \"notification\": { \"body\": \"message body\", \"title\": \"message title \", \"sound\":\"default\" ,\"click_action\": \"FLUTTER_NOTIFICATION_CLICK\", \"android_channel_id\":\"1\"} , \"data\" : {\"key1\" : \"value\", \"key2\" : \"value\" } }"
}

CloudFormationでの環境構築

Amazon Simple Notification Service リソースタイプのリファレンスからわかるように、
マネジメントコンソールで作成したプラットフォームアプリケーション、アプリケーションエンドポイントはCloudFormationに対応していない
今回は、カスタムリソースを使用してプラットフォームアプリケーションを作成し、
Lambdaにてアプリケーションエンドポイントを追加する

カスタムリソースにはcfn-responseモジュールが必要となる、cfn-responseモジュールについては以下を参照
cfn-responseモジュール
今回は、cfn-responseモジュールをLayer化して使用している

カスタムリソースによるプラットフォームアプリケーション作成

以下、プラットフォームアプリケーションを作成するためのテンプレート

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  child stack (SNS)

Globals:
  Function:
    Timeout: 3
    Runtime: python3.8
    Handler: function.lambda_handler

Parameters:
  StackName:
    Type: String
  CfnResponseArn: # cfn-responseモジュールLayer
    Type: String
  ApiKey: # プラットフォームアプリケーション生成に必要なAPIキー（サーバーキー）
    Type: String

Resources:
  CreatePlatformApplicationRole: # カスタムリソース生成時に実行されるLambdaに付与するロール
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Path: /
      Policies:
        - PolicyName: CreatePlatformApplicationPolicy # SNS操作用ポリシー
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: sns:*
                Resource: '*'

  CreatePlatformApplicationFunction: # カスタムリソース生成時に実行されるLambda
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub ${StackName}_CreatePlatformApplicationFunction
      Role: !GetAtt CreatePlatformApplicationRole.Arn
      CodeUri: create-platform-application/
      Layers:
        - !Ref CfnResponseArn

  # プラットフォームアプリケーションを生成するカスタムリソース
  CreatePlatformApplication:
    Type: Custom::CreatePlatformApplication
    Properties:
      ServiceToken: !GetAtt 'CreatePlatformApplicationFunction.Arn'　# リソースを生成するLambda Arn
      PlatformApplicationName: !Sub ${StackName}_sample # プラットフォームアプリケーション名
      PlatformApplicationArn: !Sub arn:aws:sns:${AWS::Region}:${AWS::AccountId}:app/GCM/${StackName}_sample # プラットフォームアプリケーションArn
      ApiKey: !Ref ApiKey # プラットフォームアプリケーション生成に必要なAPIキー（サーバーキー）

Outputs:
  PlatformApplicationArn: # 生成したプラットフォームアプリケーションを他テンプレートで使用できるようにOutput
    Value: !Sub arn:aws:sns:${AWS::Region}:${AWS::AccountId}:app/GCM/${StackName}_sample

以下、カスタムリソース生成時に実行されるLambda

import json
import boto3
from cfnresponse import CfnResponse

def lambda_handler(event, context):
  try:
    sns = boto3.client('sns')

    if event['RequestType'] == 'Create': # リソース生成時
      # プラットフォームアプリケーションの生成
      response = sns.create_platform_application(
        Name = event['ResourceProperties']['PlatformApplicationName'],
        Platform = 'GCM',
        Attributes = {
          'PlatformPrincipal': '',
          'PlatformCredential': event['ResourceProperties']['ApiKey']
        }
      )
      CfnResponse.send(event, context, CfnResponse.SUCCESS, {})

    elif event['RequestType'] == 'Delete': # リソース削除時
      # プラットフォームアプリケーションの削除
      response = sns.delete_platform_application(
        PlatformApplicationArn = event['ResourceProperties']['PlatformApplicationArn']
      )
      CfnResponse.send(event, context, CfnResponse.SUCCESS, {})

    else:
      CfnResponse.send(event, context, CfnResponse.SUCCESS, {})

  except Exception as e:
    CfnResponse.send(event, context, CfnResponse.FAILED, {})

Lambdaによるアプリケーションエンドポイントの追加

以下、アプリケーションエンドポイントを追加するLambdaとそのテンプレート

Parameters:
  StackName:
    Type: String
  PlatformApplicationArn:
    Type: String

Resources:
  SetEndpointFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub '${StackName}_SetEndpointFunction'
      CodeUri: set-endpoint/
      Environment:
        Variables:
          PlatformApplicationArn: !Ref PlatformApplicationArn # プラットフォームアプリケーションArn
      Policies:
        - Version: '2012-10-17'
          Statement:
          - Effect: Allow
            Action: sns:*
            Resource: '*'

import os
import json
import boto3

platformApplicationArn = os.getenv('PlatformApplicationArn')
sns = boto3.client('sns')

def add_endpoint(token):
  # アプリケーションエンドポイントの追加
  # ※既存のtokenの場合はendpoint（Arn）返却のみされる
  endpoint = sns.create_platform_endpoint(
    PlatformApplicationArn = platformApplicationArn,
    Token = token
  )

プッシュ通知を行うLambda

以下、プッシュ通知のLambda実装は以下のようになる

import os
import boto3
import json

sns = boto3.client('sns')
platformApplicationArn = os.getenv('PlatformApplicationArn')

# デバイストークンから通知先のendpoint（Arn）を取得する
def get_targetEndpoint(deviceToken):
  try:
    # ※既存のtokenの場合はendpoint（Arn）返却のみされる
    endpoint = sns.create_platform_endpoint(
      PlatformApplicationArn = platformApplicationArn,
      Token = deviceToken
    )
  except:
    raise ValueError(f'get_targetEndpoint error')
  return endpoint['EndpointArn']

def publish_notification(deviceToken):
  try:
    payload = {
  	    'notification': {
        'body': 'message body',
        'title': 'message title',
        'click_action': 'FLUTTER_NOTIFICATION_CLICK',
        'android_channel_id': '1'
      },
      'data': {
        'key1': value1,
        'key2': value2
      }
    }
    message = json.dumps({
      "GCM": json.dumps(payload)
    })
    response = sns.publish(
      TargetArn = get_targetEndpoint(deviceToken),
      Message = message,
      MessageStructure = 'json'
    )
  except:
    raise ValueError(f'publish_notification error')

参考