おすすめ育児グッズ向け SNS を AWS 新人が作ってみた

"友人の友人" を効率的に取得するために、グラフデータベース Amazon Neptune でフォロー関係を管理します。

まず、"誰が誰をフォローしているか" を DynamoDB に書き込み、その DynamoDB Streams をトリガーに AWS Lambda の関数を実行し、Neptune を更新します。Neptune の起動や Lambda の設定は、Amazon Neptune でゲームの友人レコメンドエンジンを構築する を参考にしました。

[ソースコード] フォロー管理グラフDB更新 lambda

const gremlin = require('gremlin');
const async = require('async');

const traversal = gremlin.process.AnonymousTraversalSource.traversal;
const DriverRemoteConnection = gremlin.driver.DriverRemoteConnection;
const t = gremlin.process.t;
const __ = gremlin.process.statics;

let conn = null;
let client = null;

//関数：フォロー関係の追加
async function insertRelationship(followeeId,followerId) {

  //フォローするユーザーのノードを作成(未作成の場合)
  await client.V(followeeId)
    .fold()
    .coalesce(
      __.unfold(), 
      __.addV('User').property(t.id, followeeId).property('username', followeeId)
    )
    .id().next();
  
  //フォローされるユーザーのノードを作成(未作成の場合)  
  await client.V(followerId)
    .fold()
    .coalesce(
      __.unfold(), 
      __.addV('User').property(t.id, followerId).property('username', followerId)
    )
    .id().next();
   
  //フォローしている、フォローされていることを示すエッジ用のIdを定義
  const eId_follow = followeeId+followerId+'follow';
  const eId_follower = followerId+followeeId+'followee';
    
  //フォローしていることを示すエッジを作成
  await client.E(eId_follow)
    .fold()
    .coalesce(
        __.unfold(), 
        __.V(followerId).addE('follow').to(__.V(followeeId)).property(t.id, eId_follow).property('inV', followeeId)
    )
    .id().next();
  
  //フォローされていることを示すエッジを作成
  await client.E(eId_follower)
    .fold()
    .coalesce(
        __.unfold(), 
        __.V(followeeId).addE('followee').to(__.V(followerId)).property(t.id, eId_follower).property('inV', followerId)
    )
    .id().next();

}

//関数：フォロー関係の削除
async function removeRelationship(followeeId,followerId) {

　//フォローしている人とされている人の間にあるエッジを削除
  await client.V().has('User','username', followerId)
            .bothE().hasLabel('follow')
            .has('inV',followeeId)
            .drop().iterate();
            
  await client.V().has('User','username', followeeId)
            .bothE().hasLabel('followee')
            .has('inV',followerId)
            .drop().iterate();                        
              
}

//メイン：Neptuneに対してフォロー関係の更新
exports.handler = async (event, context) => {

  var eventname = '';
  var followeeId = '';
  var followerId = '';
  
  //Neptuneへ接続
  const getConnectionDetails = () => {
    const database_url = 'wss://' + process.env['NEPTUNE_ENDPOINT']  + '/gremlin';
    return database_url;
  };

  const createRemoteConnection = () => {
    const url = getConnectionDetails();
 
    const c = new DriverRemoteConnection(
      url, 
      { 
        mimeType: 'application/vnd.gremlin-v2.0+json'
      });  

     c._client._connection.on('close', (code, message) => {
         console.info(`close - ${code} ${message}`);
         if (code == 1006){
           console.error('Connection closed prematurely');
           throw new Error('Connection closed prematurely');
         }
       });  

     return c;     
  };

  const createGraphTraversalSource = (conn) => {
    return traversal().withRemote(conn);
  };

  if (conn == null){
    console.info("Initializing connection");
    conn = createRemoteConnection();
    client = createGraphTraversalSource(conn);
  }

  //DynamoDBストリームから取得したフォロー関係の処理
  for (let record of event.Records){

   //フォロー関係と、イベント（追加or削除）の取得
    if(record.dynamodb.Keys.followeeId != null){
       followeeId = record.dynamodb.Keys.followeeId.S;
       followerId = record.dynamodb.Keys.followerId.S;
       eventname = record.eventName;    
    }
  
   //イベント（追加or削除）に応じた関数を実行しNeptuneを更新
    try{
      if(eventname==='INSERT'){
        let insert = await insertRelationship(followeeId, followerId);
      }else if(eventname==='REMOVE'){
        let remove = await removeRelationship(followeeId, followerId);
      } 
    } catch (error) {
      console.error(JSON.stringify(error));
      return { error: error.message };
    }   
  }

};

“友人の友人まで“ の投稿を閲覧可能にするために、あるユーザーが新規投稿をすると関連するユーザーのタイムラインを更新します。(例 : 下記図で X さんが投稿すると A さん, B さんのタイムラインを更新。)

Lambdaは、AWS AppSync から投稿データを受け取った後、Neptune から対象ユーザーリストを取得し、タイムライン情報を更新します。投稿の元データは、投稿データのDBに格納しています。

AWS Amplify で作成した Lambda から、VPC 内の Neptune に Gremlin で接続 できるようにする方法は、こちらの記事 (Amazon Neptune と AWS Amplify を利用したグラフアプリケーション開発) を参考にしました。

ソースコード] 投稿内容反映 Lambda  ※AMPLIFY SNS WORKSHOP の差分個所

const gremlin = require('gremlin');
const traversal = gremlin.process.AnonymousTraversalSource.traversal;
const DriverRemoteConnection = gremlin.driver.DriverRemoteConnection;
const t = gremlin.process.t;
const __ = gremlin.process.statics;
const neq = gremlin.process.P.neq;
const without = gremlin.process.P.without;
const values = gremlin.process.column.values;
const P = gremlin.process.P;

let conn = null;
let client = null;

：※AMPLIFY SNS WORKSHOPに記載の部分は省略

    //投稿者の投稿を閲覧できるユーザーリストをNeptuneから取得
    //(投稿者本人、投稿者をフォローしている人、更にその人をフォローしている人)
       
        //Neptuneへ接続
        const getConnectionDetails = () => {
          const database_url = 'wss://' + process.env['NEPTUNE_ENDPOINT'] + ':' + process.env['NEPTUNE_PORT'] + '/gremlin';
          return database_url;
        };
    
        const createRemoteConnection = () => {
          const url = getConnectionDetails();
       
          const c = new DriverRemoteConnection(
            url, 
            { 
              mimeType: 'application/vnd.gremlin-v2.0+json'
            });  
      
           c._client._connection.on('close', (code, message) => {
               console.info(`close - ${code} ${message}`);
               if (code == 1006){
                 console.error('Connection closed prematurely');
                 throw new Error('Connection closed prematurely');
               }
             });  
      
           return c;     
        };
      
        const createGraphTraversalSource = (conn) => {
          return traversal().withRemote(conn);
        };
      
        if (conn == null){
          console.info("Initializing connection");
          conn = createRemoteConnection();
          client = createGraphTraversalSource(conn);
        }
        
        //投稿者のIDを取得
        const followeeId = event.identity.username;
    
        //投稿者をフォローしている人のリストをNeptuneから取得
        let friendlist = 
        await client.V().has('User','username', followeeId).as('user')
              .out('followee')
              .values('username')
              .toList();
        
        //"投稿者をフォローしている人"をフォローしている人のリストをNeptuneから取得
        let nextfriendlist = 
        await client.V().has('User', 'username', followeeId).as('user')
          .out('followee').aggregate('friends')
          .out('followee')
          .where(without('friends'))
          .where(neq('user'))
          .dedup()
          .values('username')
          .toList();
          
        //取得結果をマージし、投稿者の投稿を閲覧できるユーザーリストを作成
        //(投稿者本人、投稿者をフォローしている人、更にその人をフォローしている人)を    
        let followers = friendlist.concat(nextfriendlist);
        followers.push(followeeId);
        
        //投稿を閲覧できるユーザーのタイムラインに表示するように、タイムラインDBを更新
        const results = await Promise.all(followers.map((follower)=> createTimelineForAUser({follower: follower, post: post})));

    return post;

フォローしているユーザーリストと、「もしかして友人かも ?」というレコメンドを表示します。「もしかして友人かも ?」というレコメンドは、フォローしている人 (= 友人) の多くがフォローしている人は友人の可能性が高い、という考えに基づきます。フォローリストは DynamoDB から、レコメンドは Neptune から取得します。

[ソースコード] フォローレコメンド取得 Lambda

const gremlin = require('gremlin');
const async = require('async');

const traversal = gremlin.process.AnonymousTraversalSource.traversal;
const DriverRemoteConnection = gremlin.driver.DriverRemoteConnection;
const t = gremlin.process.t;
const __ = gremlin.process.statics;
const neq = gremlin.process.P.neq;
const without = gremlin.process.P.without;
const values = gremlin.process.column.values;
const P = gremlin.process.P;

const order = gremlin.process.order;
const local = gremlin.process.scope.local;
const desc = gremlin.process.order.desc;

let conn = null;
let client = null;

//フォローしている人の多くがフォローしている人(上位5名)を取得
exports.handler = async (event, context, callback) => {
 
        //Neptuneへ接続
        const getConnectionDetails = () => {
          const database_url = 'wss://' + process.env['NEPTUNE_ENDPOINT'] + ':' + process.env['NEPTUNE_PORT'] + '/gremlin';
          return database_url;
        };
    
        const createRemoteConnection = () => {
          const url = getConnectionDetails();
       
          const c = new DriverRemoteConnection(
            url, 
            { 
              mimeType: 'application/vnd.gremlin-v2.0+json'
            });  
      
           c._client._connection.on('close', (code, message) => {
               console.info(`close - ${code} ${message}`);
               if (code == 1006){
                 console.error('Connection closed prematurely');
                 throw new Error('Connection closed prematurely');
               }
             });  
      
           return c;     
        };
      
        const createGraphTraversalSource = (conn) => {
          return traversal().withRemote(conn);
        };
      
        if (conn == null){
          console.info("Initializing connection");
          conn = createRemoteConnection();
          client = createGraphTraversalSource(conn);
        }
        
        //レコメンドを取得したいユーザーのIDを取得
        const username = event.identity.username;
     
     　　//ユーザーがフォローしている人の多くがフォローしている人(上位5名)を取得
        let recomendlist = 
        await client.V().has('User', 'username', username).as('user')
          .out('follow').aggregate('friends')
          .out('follow')
          .where(without('friends'))
          .where(neq('user'))
          .values('username')
          .groupCount()
          .order(local)
          .by(values,desc)
          .limit(local,5)
          .toList();
          
  //レコメンドユーザーリスト(IDと何人のフォロワーがフォローしているか）を作成
  var keys = Object.keys(recomendlist[0]);
  var counts = Object.values(recomendlist[0]);
  var array = [];
  
  keys.forEach((elem,index)=> {
    array.push({'username':elem,'count':counts[index]});
  });

  return array;

};

画像も投稿できるようにします。Amplify では、CLI とライブラリを使用して Amazon S3  を利用するストレージサービスを簡単に作成可能です。作成は、こちらのハンズオン (AWS で React アプリケーションを構築する) を参考にしました。

せっかくなのでコメント投稿もできるようにします。投稿 ID に紐づけて管理をするために、コメントデータ用の DynamoDB を Amplify で作成します。

※実装したモデルでは、同じ時間に投稿が重複すると上書きされるため、本来は更に工夫すべきです。

[ソースコード] graphql スキーマ

type Comment
  @model (
    mutations: {create: "createComment", delete: "deleteComment", update: null}
    timestamps: null
    subscriptions: { level: public}
  )
  @auth(rules: [
    {allow: owner, ownerField:"owner", provider: userPools, operations:[read, create, delete]}
    {allow: private, provider: userPools, operations:[read, create, delete]}
  ])
  @key(fields: ["postId", "timestamp"])
{
  postId: ID!
  timestamp: Int!
  commentId: ID
  content: String!
  owner: String
}

本記事では詳細は割愛しますが、React + AppSync + Amplify で構築しています。

メール配信処理用にユーザー名とメールアドレスのみを保持するユーザー管理 DB を作成します。Amplify の認証機能で使用している Amazon Cognito で ユーザー確認を実施したことをトリガー に DynamoDB に書き込みます。

[ソースコード] ユーザー管理 DB 更新 Lambda

import os
import json
import datetime
import zoneinfo
import boto3


userdb = boto3.resource('dynamodb').Table(os.getenv['USER_TABLE_NAME'])

#ユーザー管理テーブルへ、ユーザーID・メールアドレス・タイムスタンプを書き込み
def lambda_handler(event, context):

    userdb.put_item(
        Item = {
            'user':event['userName'],
            'address':event['request']['userAttributes']['email'],
            'timestamp':datetime.datetime.now(zoneinfo.ZoneInfo('Asia/Tokyo')).strftime("%Y%m%d%H%M%S")
        }
    )
    return event;

投稿データ DB の DynamoDB Streams を Lambda でトリガーし、新規投稿があるたびに Amazon Comprehend で感情分析をします。結果は投稿 ID と共に評価データ DB に格納しておきます。Lambda と Comprehend の接続はこちらのハンズオン (AWS Hands-on for Beginners) を参考にしました。

[ソースコード] 投稿の感情分析 Lambda

import os
import json
import boto3
from decimal import Decimal

comprehend = boto3.client('comprehend')
ratingdb = boto3.resource('dynamodb').Table(os.environ['RATING_TABLE_NAME'])

#投稿本文をcomprehendで感情分析し、評価データDBを更新
def lambda_handler(event, context):

   #DynamoDBストリームから取得した投稿を1つずつ処理
    for record in event['Records']:
        
        #投稿本文を取得しcomprehendを実行
        content = record['dynamodb']['NewImage']['content']['S'] 
    
        response = comprehend.detect_sentiment(
            Text=content,
            LanguageCode='ja'
        )
       
        #投稿タイムスタンプを取得し、comprehend実行結果と共に評価データDBへ書き込み
        timestamp=int(record['dynamodb']['NewImage']['timestamp']['N'])
        
        ratingdb.put_item(
            Item = {              
                'postId':record['dynamodb']['NewImage']['id']['S'],
                'timestamp':timestamp,
                'type':'sentimentscore',
                'sentimentscore':{
                    'positive': Decimal(response['SentimentScore']['Positive']),
                    'negative': Decimal(response['SentimentScore']['Negative']),
                    'neutral': Decimal(response['SentimentScore']['Neutral']),
                    'mixed': Decimal(response['SentimentScore']['Mixed']),
                }
            }
        )
    return {
        'statusCode': 200,
    }

ユーザーのタイムライン内の投稿のうち、最も感情分析の Positive の値が高い = 熱量がある投稿内容、と判定しメールで配信します。

ユーザー数が増えた際に Lambda 関数がタイムアウトすることを防ぐため、SQS にユーザーリストをキューイングしておき、Lambda 関数にて一定数毎に配信するようにしました。

同一メールの重複配信を防ぐために FIFO キューを利用し、Lambda 関数が最大 5 並列になるように実装しています。FIFOキューでのLambda関数の並列実行（メッセージグループID設定）は、こちらの記事 (New for AWS Lambda – SQS FIFO as an event source) を参考にしました。

Amazon Simple Notification Service (Amazon SNS) でもメール配信は可能ですが、ユーザー毎にカスタマイズした文面を配信するために、Amazon Simple Email Service (Amazon SES) を利用します。

定期配信のために、EventBridge で設定したスケジュールに従って Lambda 関数を実行  しています。

[ソースコード] ユーザー情報取得 Lambda

import os
import json
import boto3
from boto3.dynamodb.conditions import Key

sqs = boto3.client('sqs')
queue_url = os.environ['QUEUE_URL']
queue_group = 'SQSキューグループ名'
userdb = boto3.resource('dynamodb').Table(os.environ['USER_TABLE_NAME'])

#ユーザーリストを取得し、ユーザー毎にIDとメールアドレスをSQSのキューへ登録
def lambda_handler(event, context):
    
    # ユーザーリスト取得
    response = userdb.scan()
    userlist = response['Items']
    
    while 'LastEvaluatedKey' in response:
        response = userdb.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
        userlist.extend(response['Items'])
        
    #FIFOキューで使用するgroupIdの定義
    groupId = 0 
    
    #ユーザー毎にユーザー情報をSQSに追加  
    for record in userlist:
    
        #MassageGroupIdを5つに分割するための処理(0～4になるようにリセット)
        if groupId = 5:　
            groupId = 0
        
        #SQSへユーザー情報を追加
        items = json.dumps(record)
        response = sqs.send_message(QueueUrl=queue_url, MessageBody=items, MessageGroupId=queue_group+groupId)

        groupId += 1 #ユーザー毎にMassageGroupIdのインクリメント
        
    return {
        'statusCode': 200
    }

[ソースコード] 配信メール作成 Lambda

import os
import json
import boto3
from boto3.dynamodb.conditions import Key
import pandas as pd

dynamodb = boto3.resource('dynamodb')
ratingdb = dynamodb.Table(os.environ['RATING_TABLE_NAME'])
timelinedb = dynamodb.Table(os.environ['TIMELINE_TABLE_NAME'])
postdb = dynamodb.Table(os.environ['POST_TABLE_NAME'])
sesclient = boto3.client('ses')

SOURCE_ADDRESS = '※配信元アドレス※'
START_TIMESTAMP = 1651330800     # 2022-05-01T0:10:00　※UNIX時間　
END_TIMESTAMP = 1656601200        # 2022-07-01T0:10:00　※UNIX時間
SUBJECT = '今週のおすすめ投稿'

#ユーザーへのおすすめ投稿を判定し、SESでメール配信
def lambda_handler(event, context):
     
    #期間内の投稿内容の評価結果(comprehend結果)を評価データDBから取得
    response = ratingdb.query(
        IndexName="type-timestamp-index",
        KeyConditionExpression=Key('type').eq('sentimentscore') & Key('timestamp').between(START_TIMESTAMP, END_TIMESTAMP)
    )
    
    #おすすめ投稿の判定のために、評価結果のリストを、pandasデータフレームに格納
    rating= response['Items']
    ratingdf = pd.json_normalize(rating)

    #ユーザー毎の配信作成。SQSのキューを1つずつ処理
    for record in event['Records']:     
    
        record=eval(record['body']) #body 内をstring -> dict
        user = record['user']
        
        #ユーザーのタイムライン情報(投稿ID)をタイムラインDBから取得
        response = timelinedb.query(
            KeyConditionExpression=Key('userId').eq(user) & Key('timestamp').between(START_TIMESTAMP, END_TIMESTAMP)
        )
        
        timelinelist=response['Items']
        
        while 'LastEvaluatedKey' in response:
            response = timelinedb.query(
                KeyConditionExpression=Key('userId').eq(user) & Key('timestamp').between(START_TIMESTAMP, END_TIMESTAMP),   
                ExclusiveStartKey=response['LastEvaluatedKey']
            )
            timelinelist.extend(response['Items'])
        
        #期間内のタイムラインに投稿が1件以上ある場合、処理を継続
        if response['Count'] != 0:
             
            #おすすめ投稿の判定のために、タイムラインの投稿リストを、pandasデータフレームに格納
            timelinedf = pd.json_normalize(timelinelist)
        
            #投稿リストから自分の投稿を削除
            timelinedf = timelinedf[~timelinedf['postUserId'].str.contains(user)]            

            #自分以外の投稿が1件以上ある場合、メール配信処理を実施
            if len(timelinedf) != 0:
            
                #評価結果データフレームとタイムラインデータフレームを投稿IDでマージ
                reportdf = pd.merge(ratingdf, timelinedf, on='postId',how='inner')
                
                #結果のscore部分をfloatに変更し、positiveの値がもっとも高い投稿IDを抽出
                reportdf = reportdf.astype({'sentimentscore.positive': float})
                reportpostindex = reportdf['sentimentscore.positive'].idxmax()              
                reportpostdf = reportdf.loc[reportpostindex,["postId"]]
                reportpostdf = reportpostdf.astype(str)
                               
                #投稿IDを持つ投稿の詳細を投稿データDBから取得
                response = postdb.get_item(
                    Key={
                        'id':reportpostdf['postId']
                    }
                )                           
                reportpost = response['Item']
                
                #投稿詳細とユーザーデータを使用してユーザーへemail送信
                response = sesclient.send_email(
                    Source=SOURCE_ADDRESS,
                    Destination={
                        'ToAddresses': [
                            record['address'],
                        ]
                    },
                    Message={
                        'Subject': {
                            'Data': SUBJECT,
                        },
                        'Body': {
                            'Text': {
                                'Data': 
                                    record['user']+"さんへのおすすめ投稿\n\n"+"商品名："+reportpost['title']+"\n"+"説明："+reportpost['content']+"\n"+"投稿者："+reportpost['owner']+"\n",
                            }
                        }
                    }
                )