YouTuber になったから、活動を支えるサービスを React, API Gateway, Lambda, DynamoDB で作りかけた

2019年の7月、YouTube にゲーム実況チャンネルを開設しました。

本記事では、YouTube 活動に際して発生した問題点の紹介と、ツールをつくってそれを解決しようとした話をします。

どのような YouTube チャンネルか。収録環境の説明

開設したチャンネルの特徴は以下です。

• 配信する内容はゲーム実況
• 2人で実況する
• 1人がプレイヤー、もう1人が Skype の画面共有を通じてゲームプレイを見て喋る

収録環境について、図にするとこのようになります。

本記事では「YouTube 動画撮影の問題点と、その問題点を解決するためのサービスを作ろうとした話」を紹介しますが、問題点というのはこれから説明する収録環境によるものです。

チャットツールは Skype, Discord を使っています。Skype はゲーム映像・音声の共有に、Discord は2人の通話に使っています。

ゲーム映像・音声の収録はプレイヤーのマシンで OBS Studio を使って行います。

2人の会話は、それぞれのマシンで、Audacity や QuickTime Player を使って録音しています。当初は、片方のマシンで「マイク」「Discord の音声」をソースとして録音していました。

しかし、Discord を通じた音声よりも、直接マイクからの音声を収録した方が明らかに綺麗だったので、2人ともそれぞれのマシンで個別に実況音声を録音するようになりました。

収録で困っていたこと

さて、この収録環境のとき、困っていたことというのは、「それぞれのマシンで個別に実況音声を録音している」ことに起因します。

同時に収録開始したかったのですが、リモートにある双方で同時に録音ボタンを押すのは難しいです。

「せーの」（録音開始）というふうに声かけをしているのですが、実際には微妙なズレが発生します。

2人の実況音声にズレがあると、編集ソフトで解消する必要があるのですが、これがちょっとだけ手間だったりします。

下が音のズレを直す作業風景です。音声素材を微妙に動かしては聞き直し、2人の音声が同時に再生されるように調整しています。

というわけで、今回解決しようとした問題点というのは、「収録時に2人の音声がズレて録音されてしまう。編集時に合わせるのが手間」ということです。

以降では、これを解決するツールを作ろうとしたことについて紹介します。

おまけ：本当にツールを作らないといけない？？？？

本記事ではこのあと、ツール開発について紹介しますがその前に、先ほど紹介した「収録時に2人の音声がズレて録音されてしまう。編集時に合わせるのが手間」という問題は本当にツールを開発するぐらいでないと解決できない問題か、ということについて補足しておきます。

答えはおそらく No です。アナログ的な手法で解決できます。

今回の問題は「編集時に合わせるのが手間」ということですので、編集時に合わせやすいように収録すればよいです。

例えば、収録開始自体はそれぞれ適当にはじめてしまって、お互いに収録開始してから「せーの パン! 👏 」と声をかけあって手拍子するなど。

編集時に分かりやすいポイント（編集点）をつくってあげれば、編集ツールでの作業は比較的簡単です。

もし YouTube 活動で同じような問題を抱えている人は参考にしてください。

作ろうとしたものをデモ

作成途中でして、デモページはありません。上の Gif 画像をご覧ください。

2つ並んでいるブラウザウィンドウは、リモートで収録する2人がそれぞれブラウザを開いているという想定です。

どのような操作をしているか、手順は以下です。

• 「ルーム」と呼ばれる収録部屋を作成する
• 「ルーム」に対してユニークな URL が発行される
• URL をシェアする
• ルームの参加者が増える
• 収録の開始リクエストを送る
• 5秒後、収録が開始される
• 会話したのち、収録を停止する
• 音声ファイルがダウンロードされる
  • マシンを操作しているユーザーの声のみのファイルです

これによって同時に録音することを実現しています。

なにを使って作ったか

バックエンド、フロントエンド合わせて、おおよその技術スタックは以下です。

• TypeScript
• React
• AWS Lambda
• Amazon DyanamoDB
• Amazon API Gateway

詳しくはそれぞれのリポジトリをご覧ください。

バックエンド：https://github.com/karur4n/zenrecorder-server
フロントエンド：https://github.com/karur4n/zenrecorder-front

なぜ AWS か、なぜ GCP ではないのか

本ツールは AWS を中心に構成されています。PaaS として AWS と並んで有力である GCP ではなく、なぜ AWS を選んだかについて述べます。

構成する際に重視した点は2つです。

• できるだけインフラの面倒は見たくない
  • VPS 的に、OS ごと任されてプロセス管理などしたくない
• 本ツールはリアルタイム性が重視されるアプリケーションである

参加者の追加・削除や、収録開始・収録終了ではリアムタイム性が求められます。

GCP（Firebase） であれば、これらを満たすには Cloud Firestore を使うことになるかと思うのですが、Cloud Firestore ではひとつ機能が足りていませんでした。

切断時をトリガーとして処理を実行することができないのです。

本ツールでは「ブラウザを閉じたときにルームから参加者が削除される」という機能を実現する必要がありました。

絶対に切断時をトリガーに処理を実行することができない、というわけではなく技術的には可能なのですが、プレゼンス機能をサポートしている Firebase Realtime Database を併用する必要があります。

参考：Cloud Firestore でプレゼンスを構築する  |  Firebase

それに反して、AWS では API Gateway の WebSocket API で切断時に Lambda の実行をフックできます。

WebSocket 系を GCP で柔軟に使用するとなると、GCE (Google Compute Engine)を使うことになります。これは、OS 単位の管理をしたくない、という今回の要求にマッチしません。

よって、AWS を採用することに決めました。

こだわったところ

本ツールの開発でこだわった点として、ルーム情報のリアルタイムアップデートを実現しました。

リアルタイムアップデートとは、Cloud Firestore で使われている言葉で、Firestore, DynamoDB でいうところのドキュメントが更新されるたびに、取得するスナップショットオブジェクトの内容が更新されます。

参考：Cloud Firestore でリアルタイム アップデートを入手する  |  Firebase

下の図で雰囲気をつかんでもらえるかと思います。

Firestore ではリアルタイムアップデート機能はほぼメイン、主要機能としてサポートされていますが、今回採用した AWS Lambda, API Gateway, DynamoDB の組み合わせでは標準でサポートされていません。AWS 系でそういう機能があるのはおそらく AWS AppSync です。

本ツール開発のこだわった点は「AWS Lambda, API Gateway, DyamoDB を用いてリアルタイムアップデートを実現したこと」です。

次節では、このリアルタイムアップデートを中心に、本ツールがどのような実装になっているかを見ていきます。

どのような実装か

本ツールは、ルーム情報という状態オブジェクトを中心に実装されています。

ルーム情報は次のような構造をしています。

ルーム情報

{
  "roomId": "ad17e4ce-2dc2-4525-8497-10d16d0866e7",
  "status": "beforeRecording",
  "userIds": [
    "CyyhDfw1NjMCJug=",
    "BdjasfjasdfIIII-"
  ],
  "users": [
    {
      "id": "CyyhDfw1NjMCJug=",
      "name": "古河和樹"
    },
    {
      "id": "BdjasfjasdfIIII-",
      "name": "テスト1"
    }
  ]
}

例えば、ルーム情報の状態（statusプロパティ）が recording（収録中） に変わると、それを受け取ったクライアントが収録を開始します。ルーム情報のユーザー一覧（usersプロパティ）にユーザーオブジェクトが追加されると、変更を受け取ったクライアントは描画内容を更新します。

このように本ツールはサーバーサイドではルーム情報というアプリケーション状態を更新する、変更されたルーム情報を受け取ったクライアントサイドが更新内容に応じてプレゼンテーションするという実装になっています。

CQS（コマンドクエリ分離）を採用する

「ルーム情報を更新する」「更新されたルーム情報を取得する」ということについてCQS（コマンドクエリ分離）を採用しています。

CQS とは「コマンド（モデルの状態を変更する）」と「クエリ（モデルを参照する）」を分離することです。「ルーム情報を更新すること」はコマンド、「更新されたルーム情報を取得すること」はクエリに当たります。

※ CQS という言葉について個々人によって解釈に違いがあるかもしれません。本記事では下の記事を参考に CQS について述べます。

参考：CQSとCQRSの違いはメソッドの分離かモデルの分離かという観点 - Qiita

CQS の説明として、まず CQS でないコードを掲載します。以下の UserDataAccessor は CQS ではありません。コマンドである updateName が Promise<User> を戻り値にしていますので、updateName はクエリとしての役割も持っています。

CQSではない例

interface UserDataAccessor {
  find(userId: UserId): Promise<User>
  updateName(userId: UserId, name: string): Promise<User>
}

上の UserDataAccessor に CQS を適用すると次のようになります。

CQSである例

interface UserDataAccessor {
  find(userId: UserId): Promise<User>
  updateName(userId: UserId, name: string): Promise<void>
}

updateName は Promise<void> を返すようになり、コマンドとしての役割のみを担うことになりました。コマンドの成否は Promise が fulfilled, rejected になるかによって取得します。

---

この CQS を具体的には次のように適用しています。

『コマンドとして API Gateway にリクエストを送信すると、そのレスポンスについては無関心（そもそも WebSocket のメッセージングにレスポンスという概念はありませんが）。DynamoDB 上のルーム情報の収録ステータスが変更されると、WebSocket を通じて更新されたルーム情報が送信されてくるので更新情報を取得できる（クエリ）。』

クライアント側の実装

まずクライアントでの「ルームに参加する」処理と「更新されたルーム情報を受け取る」処理について見ていきます。

React のカスタムフックとして実装しています。joinRoom 関数が「ルームに参加する」処理を担っています。

src/useRecordingRoom.ts

// https://github.com/karur4n/zenrecorder-front/blob/master/src/useRecordingRoom.ts

import React from 'react'
import { Room } from './domain/Room'

type ConnectingStatus = 'pending' | 'connected' | 'failed' | 'closed'

export function useRecordingRoom(roomId: string) {
  const socketRef = React.useRef<WebSocket | undefined>(undefined)
  const [connectingStatus, setConnectingStatus] = React.useState<ConnectingStatus>('pending')
  const [userId, setUserId] = React.useState<string | undefined>(undefined)
  const [room, setRoom] = React.useState<Room | undefined>(undefined)

  // 略...

  //
  // Functions
  //
  function joinRoom(userName: string): void {
    setConnectingStatus('pending')

    const socket = new WebSocket('wss://websocket-api のエンドポイント')

    socketRef.current = socket

    socket.onopen = () => {
      setConnectingStatus('connected')

      const body = JSON.stringify({
        action: 'sendMessage',
        type: 'JOIN_ROOM',
        payload: {
          roomId: roomId,
          userName: userName,
        },
      })

      socket.send(body)
    }

    socket.onmessage = (event) => {
      const data = JSON.parse(event.data)
      setRoom({
        ...data,
        recordingStartedAt: new Date(data.recordingStartedAt),
      })
    }

    // 略...
  }

  function send(body: object): void {
    const socket = socketRef.current

    if (socket) {
      socket.send(
        JSON.stringify({
          action: 'sendMessage',
          ...body,
        })
      )
    }
  }

  return {
    room,
    joinRoom,
    connectingStatus,
    send,
  }
}

送られてきたルーム情報を state としてセットしているシンプルな実装です。送られているメッセージはいまのところ更新されたルーム情報だけのため、type プロパティなどを用いた振り分けもしてません。

socket.onmessage = (event) => {
  const data = JSON.parse(event.data)
  setRoom({
    ...data,
    recordingStartedAt: new Date(data.recordingStartedAt),
  })
}

このカスタムフックを使うコンポーネントは以下のようになっています。ファイル全体を掲載すると長くなるので、抜粋しております。詳しくはファイル全体（ https://github.com/karur4n/zenrecorder-front/blob/master/src/pages/RoomPage/RoomPage.tsx ）をご覧ください。room オブジェクトの更新情報を元にして、動作していることが分かるかと思います。

src/pages/RoomPage/RoomPage.tsx

// https://github.com/karur4n/zenrecorder-front/blob/master/src/pages/RoomPage/RoomPage.tsx

import React, { useState } from 'react'
import { useRecordingRoom } from '../../useRecordingRoom'
import { usePrevious } from '../../usePrevious'
// 略...

export const RoomPage: React.FC = () => {
  const { start, stop, durationSeconds, recordingState } = useAudioRecorder()
  const { connectingStatus, room, send, joinRoom } = useRecordingRoom(roomId)
  const prevRoom = usePrevious(room)
  const [recordingStartTimer, setRecordingStartTimer] = useState<number | undefined>(undefined)
  // 略...

  React.useEffect(() => {
    const 入室したら収録中だった = prevRoom == undefined && room && room.status === 'recording'
    const 入室してから収録中になった =
      prevRoom != undefined && room != undefined && prevRoom.status == 'beforeRecording' && room.status === 'recording'

    if (入室したら収録中だった || 入室してから収録中になった) {
      requestRecordingStart(room!.recordingStartedAt)
    }

    if (prevRoom != undefined && prevRoom.status !== 'completed' && room != undefined && room.status === 'completed') {
      stop()
        .then((audio) => {
          console.log('Audio', audio)
          setAudios((prev) => [...prev, audio])
        })
        .catch((e) => {
          console.log(e)
        })
    }
  }, [room])

  // 略...

  React.useEffect(() => {
    if (recordingStartTimer == undefined) {
      return
    }

    if (recordingStartTimer === 0) {
      start()
    }
  }, [recordingStartTimer])

  //
  // functions
  //
  function requestRecordingStart(startedAt: Date): void {
    let intervalId: number | undefined = undefined

    intervalId = setInterval(() => {
      const current = new Date()

      if (current > startedAt) {
        clear()
        return
      }

      const diffInSec = Math.abs(differenceInSeconds(current, startedAt))

      setRecordingStartTimer(diffInSec)
    }, 100)

    function clear() {
      if (intervalId != undefined) {
        clearInterval(intervalId)
      }
    }
  }

  // 略...
}

どのプロパティが更新されたのか知りたい場面があります。例えば、収録状態です。収録状態が「収録中」になったことを受けて、クライアントサイドは実際に MediaRecorder を通じて録音を開始するわけですが、録音開始は「開始前」から「収録中」に切り替わったときだけ実行したいです。

どのように更新されたかを知るためには前回の room オブジェクトとの比較が必要です。前回の値を管理するための usePrevious の実装はこちらを使いました。

usePrevious React Hook - useHooks

次は、サーバーからどのように、更新されたルーム情報をクライアントに送信しているかを見ます。

ルーム情報の更新をサーバーからクライアントへ通知する

これまで延々と本ツールの中心であると述べてきたルーム情報は DynamoDB にて永続化されています。

CQS を実現するにあたり DynamoDB の「トリガー」という機能に大いに助けられています。トリガー機能は「DynamoDB のテーブル内に更新があったとき、その更新内容を引数として Lambda を実行する」というものです。

この機能により以下のように「コマンド」と「クエリ」を分けています。

• クライアントからのリクエストによって更新されたルーム情報を DynamoDB に永続化する（コマンド）
• DynamoDB のトリガによって、新たに Lambda が起動し、更新されたルーム情報をクライアントに送信する（クエリ）

各サービス間はおおよそ以下のように動きます。

クエリ部分の実装について見ます。

AWS 管理画面よりトリガに対して Lambda を設定しています。

トリガ機能は DynamoDB Stream を通じて実現されます。Lambda と DynamoDB の連携について詳しくは以下の記事を参考にしてください。

DynamoDB StreamをトリガーにしてLambdaを実行する - Qiita

今回の実装では、DynamoDB の変更を受けた Lambda は OnDynamoDbChangeUseCase を実行します。

OnDynamoDbChangeUseCase はざっくりと以下のような実装になっています。

• 引数として受け取った DynamoDBStreamEvent から、更新されたルーム情報を取得する
• ルームの参加者一覧に対して、更新情報を送信する

src/application/usecases/OnDynamoDbChangeUseCase.ts

// https://github.com/karur4n/zenrecorder-server/blob/master/src/application/usecases/OnDynamoDbChangeUseCase.ts

import { inject, injectable } from 'inversify'
import { RoomRepository } from '../../domain/Room/RoomRepository'
import { RoomId } from '../../domain/Room/RoomId'
import { UserId } from '../../domain/Room/User/UserId'
import { UserName } from '../../domain/Room/User/UserName'
import { TYPES } from '../../di/types'
import { DynamoDBStreamEvent } from 'aws-lambda'
import { Room } from '../../domain/Room/Room'
import { User } from '../../domain/Room/User/User'
import { ClientMessenger, ClientMessengerStaleConnectionError } from '../ClientMessenger'
import { RecordingStartedAt } from '../../domain/Room/RecordingStartedAt'

@injectable()
export class OnDynamoDbChangeUseCase {
  public constructor(
    @inject(TYPES.ClientMessenger) private clientMessenger: ClientMessenger,
    @inject(TYPES.RoomRepository) private roomRepository: RoomRepository
  ) {}

  public async execute(event: DynamoDBStreamEvent): Promise<void> {
    const newRoomRecords = event.Records.filter((r) => r.eventName === 'MODIFY')

    const newRooms = newRoomRecords.map(
      (record): Room => {
        return mapNewImageToRoom(record.dynamodb!.NewImage)
      }
    )

    for (const room of newRooms) {
      for (const user of room.users) {
        try {
          await this.clientMessenger.post(user.id.asString(), JSON.stringify(room.toObject()))
        } catch (e) {
          if (e instanceof ClientMessengerStaleConnectionError) {
            room.exit(user)

            await this.roomRepository.store(room)

            continue
          }

          throw e
        }
      }
    }
  }
}

function mapNewImageToRoom(image: any): Room {
  const roomId = new RoomId(image.roomId.S)
  const recordingStatus = image.status.S

  const users = image.users.L.map(
    (userRecord: any): User => {
      const userId = new UserId(userRecord.M.id.S)
      const userName = new UserName(userRecord.M.name.S)

      return new User(userId, userName)
    }
  )

  const recordingStartedAt =
    image.recordingStartedAt && image.recordingStartedAt.S
      ? RecordingStartedAt.ofByString(image.recordingStartedAt.S)
      : undefined

  return new Room(roomId, users, recordingStartedAt, recordingStatus)
}

ClientMessanger の実装である ClientMessengerImpl にて、WebSocket を担ってくれる API Gateway にルーム情報を送り、API Gateway がクライアントへルーム情報を送信します。

src/infrastructure/ClientMessengerImpl.ts

import { ClientMessenger, ClientMessengerStaleConnectionError } from '../application/ClientMessenger'
import { ApiGatewayManagementApi } from 'aws-sdk'
import { injectable } from 'inversify'

@injectable()
export class ClientMessengerImpl implements ClientMessenger {
  private apiGatewayManagementApi: ApiGatewayManagementApi

  constructor() {
    this.apiGatewayManagementApi = new ApiGatewayManagementApi({
      apiVersion: '2018-11-29',
      endpoint: 'https://8mre732jfk.execute-api.ap-northeast-1.amazonaws.com/production',
    })
  }

  public async post(connectionId: string, body: string): Promise<any> {
    const params = {
      ConnectionId: connectionId,
      Data: body,
    }

    try {
      await this.apiGatewayManagementApi.postToConnection(params).promise()

      return
    } catch (e) {
      if (e.statusCode === 410) {
        throw new ClientMessengerStaleConnectionError()
      } else {
        throw e
      }
    }
  }
}

apiGatewayManagementApi.postToConnection で送信するのですが、接続が利用できなかった場合、ステータスコード 410 ( Gone ) が返ります。410 ( Gone ) ステータスは、指定したコネクションが利用できないことを表しますので、特別にエラーを投げて、ルームから削除するということを行っています。

参考：[発表]Amazon API GatewayでWebsocketが利用可能 | Amazon Web Services ブログ

その他のサーバーサイドの実装を紹介

ユーザーからのユースケースは以下です。

• ルームを作成する
• ルームに参加する
• 収録を開始する
• 収録を終了する
• ルームから退出する

ユーザーからのユースケースはすべて API Gateway で受けます。

ユースケースに対して、API Gateway 上で API とルートを以下のように対応させました。

REST API

ルート	対応するユースケース
POST /create-room	・ルームを作成する

WebSocket API

ルート	対応するユースケース
sendMessage	・ルームに参加する ・収録を開始する ・収録を終了する
$disconnect	・ルームから退出する

$ からはじまるルートは、プロトコルが WebSocket である API において、デフォルトで定義されるルートです。

$connect, $disconnect は文字通り、接続 / 切断されるときに呼び出される API です。

今回、$connect ルートをユースケースと対応させておりません。WebSocket に接続することはアプリケーション上のユースケースではありません。ルームに参加することをユースケースとしています。

そのため $connect に対応する Lambda の実装は、ハンドラーからユースケースを実行するようなことはしておらず、ただ 204 をレスポンスしています。

src/presentation/handlers/onConnect.ts

// https://github.com/karur4n/zenrecorder-server/blob/master/src/presentation/handlers/onConnect.ts

import { Context, Callback, APIGatewayEvent } from 'aws-lambda'

export const handler = (_: APIGatewayEvent, __: Context, callback: Callback) => {
  callback(null, {
    statusCode: 204,
  })
}

ルームに接続して以降は、ユースケースの呼び出しをすべて WebSocket API の sendMessage というルートを使っています。

sendMessage ハンドラーの実装は以下です。

クライアントから送信される Redux のアクションを模した Action オブジェクトから、ユースケースの呼び出しを振り分けています。

src/presentation/handlers/sendMessage/sendMessage.ts

// https://github.com/karur4n/zenrecorder-server/blob/master/src/presentation/handlers/sendMessage/sendMessage.ts

require('dotenv').config()

import { APIGatewayEvent, Context, Callback } from 'aws-lambda'
import { myContainer } from '../../../di/inversify.config'
import { TYPES } from '../../../di/types'
import { JoinRoomUseCase } from '../../../application/usecases/JoinRoomUseCase'
import { StartRecordingUseCase } from '../../../application/usecases/StartRecordingUseCase'
import { CompleteRecordingUseCase } from '../../../application/usecases/CompleteRecordingUseCase'

export const handler = async (event: APIGatewayEvent, _: Context, callback: Callback) => {
  if (event.body == undefined) {
    return callback(null, {
      statusCode: 400,
    })
  }

  const action = JSON.parse(event.body) as Action
  const requestUserId = event.requestContext.connectionId!

  switch (action.type) {
    case 'JOIN_ROOM':
      const joinRoomUseCase = myContainer.get<JoinRoomUseCase>(TYPES.JoinRoomUseCase)
      await joinRoomUseCase.execute(action.payload.roomId, requestUserId, action.payload.userName)
      return callback(null, { statusCode: 200 })
    case 'START_RECORDING':
      const startRecordingUseCase = myContainer.get<StartRecordingUseCase>(TYPES.StartRecordingUseCase)
      await startRecordingUseCase.execute(action.payload.roomId, requestUserId)
      return callback(null, { statusCode: 200 })
    case 'COMPLETE_RECORDING':
      const completeRecordingUseCase = myContainer.get<CompleteRecordingUseCase>(TYPES.CompleteRecordingUseCase)
      await completeRecordingUseCase.execute(action.payload.roomId, requestUserId)
      return callback(null, { statusCode: 200 })
  }
}

const JOIN_ROOM = 'JOIN_ROOM'

type JoinRoomAction = {
  type: typeof JOIN_ROOM
  payload: {
    roomId: string
    userName: string
  }
}

const START_RECORDING = 'START_RECORDING'

type StartRecordingAction = {
  type: typeof START_RECORDING
  payload: {
    roomId: string
  }
}

const COMPLETE_RECORDING = 'COMPLETE_RECORDING'

type CompleteRecordingAction = {
  type: typeof COMPLETE_RECORDING
  payload: {
    roomId: string
  }
}

type Action = JoinRoomAction | StartRecordingAction | CompleteRecordingAction

他の実装についてはコードを直接ご覧ください。

バックエンド：https://github.com/karur4n/zenrecorder-server
フロントエンド：https://github.com/karur4n/zenrecorder-front

まとめ

React, Amazon API Gateway, AWS Lambda, Amazon DynamoDB を使って YouTube に向けたゲーム実況収録支援ツールを作ろうとしました。

リアルタイムアップデートの実装部分について、Firestore に比べると自分で実装した部分が多いのですが、要件を満たすため、AWS 系サービスを組み合わせて実現しました。

DynamoDB のトリガーのように実現が簡単ではない機能を提供されると、やはり *aaS は便利だと感じます。

CQS は実装がうまくまわる仕組みを一度実装してしまえば、とてもシンプルになるので積極的に採用していきたいです。

オチ

YouTube 活動やめちゃったから、ツールを完成させる意欲が消えました