はじめに

Firebaseの「Cloud Tasks」にて遅延実行を試してみることになった！

背景

私達のチームでは配送パートナーの運行支援をサポートする部署が使用する社内ツール開発を担当しています
運行の問い合わせ方法には 「チャット」 と 「電話」 があります
「チャット」では問い合わせ内容をある程度整理した上でチケットを作成できるようにワークフローが存在します

「解決済みですか？」を投げたい

ユーザーの要望にもあったが、一定時間問いに対して回答がない場合に「解決済みですか？(選択肢'はい', 'いいえ')」を再度投げたい
- 例えば、　3分経過して応答がない場合「解決済みですか？」を投げる
じゃあ、遅延実行をサポートしている「Cloud Tasks」を試してみるか！ということになったので試す

そもそもCloud Tasksとは？

Cloud Tasks は、大量の分散タスクの実行、ディスパッチ、配信を管理できるフルマネージドサービスです
ユーザーからのリクエストを非同期で実行してくれます

「キュー」と「タスク」

Cloud Tasksがユーザーのリクエストによって代わりに実行する独立的な作業の事を 「タスク」 といいます
そのタスクを作成したあと 「キュー」 という入れ物に詰めてあげると作成したタスクの設定に基づいて実行してくれます

上の図で言う Target はタスクに対して レスポンス「2xx」を返す必要があり、返さないと再試行されます

大まかな処理の流れ

1. タスクを処理するためのワーカーを作成します
2. キューを定義します
3. プログラム側でタスクを作成して、キューに詰めます
4. キューに詰めたタスクによってワーカーが呼ばれます
5. 呼ばれたワーカーがタスクを処理します
6. ワーカーは処理が完了したら、レスポンスでステータス200を返すと、Cloud Tasksは完了になります

実装方法

動かしてみて、実際に遅延実行を実現できたコードを紹介します
ほとんど公式のサンプルや他で試している人のを参考にすればできるかと思います

タスクキューを定義する

まずは、タスクを詰めるキューを定義してあげます
maxAttempts：タスクが再試行される最大実行回数指定する
minBackoffSeconds：各タスクで再試行される際の各試行の時間間隔を指定する
maxConcurrentDispatches：同時にディスパッチされるタスクの最大個数を指定できます

キューを定義する：

exports.publishDelayedMessage = onTaskDispatched({
  retryConfig: {
    maxAttempts: 5, 
    minBackoffSeconds: 60
  },
  rateLimits: {
    maxConcurrentDispatches: 6
  }
}, async (request: any) => {
  const currentDate = request.data.date.current
  // 何かしらの処理
  try {
      const response = await fetch('...')
  } catch(err) {
      console.error(err)
      throw new HttpsError('internal', 'Failed to request')
  }
})

タスクをキューに詰める

作成したタスクを先ほど定義したキューに詰めます
詰める際に「遅延させたい時間」を scheduleDelaySeconds に指定してあげえることで遅延実行を実現できます
scheduleDelaySeconds：ここで遅延させてい時間を指定します(今回は2分遅延させる)
dispatchDeadlineSeconds：タスクが完了するまでにCloud Tasksが待機する最長時間

タスクをキューに詰める箇所：

exports.processPublishMessage = onRequest(
  async (_request: any, response: any) => {
    const queue = getFunctions().taskQueue('publishDelayedMessage')
    const taregetUrl = await getFunctionUrl('publishDelayedMessage')
    // ※ getFunctionUrl()については後述

    const enqueues = []

    // タスクに渡すパラメーターを定義する
    const date = {
      current: new Date()
    }

    enqueues.push(
      queue.enqueue({ date }, {
        scheduleDelaySeconds: 60 * 2, // ここで指定した時間遅延させる
        dispatchDeadlineSeconds: 60,
        uri: taregetUrl
      })
    )
    await Promise.all(enqueues)
    response.sendStatus(200)
  }
)

ターゲットURLを取得する

タスクを作成する際にワーカーのURIを指定する必要があります
公式のサンプルコードにもありますが、以下のgetFunctionUrl()メソッドが使用できます

let auth: GoogleAuth<JSONClient>

const getFunctionUrl = async (name: string, location = 'us-central1') => {
  if (!auth) {
    auth = new GoogleAuth({
      scopes: 'https://www.googleapis.com/auth/cloud-platform'
    })
  }
  const projectId = await auth.getProjectId()
  const url = 'https://cloudfunctions.googleapis.com/v2/' +
    `projects/${projectId}/locations/${location}/functions/${name}`

  const client = await auth.getClient()
  const res = await client.request({ url }) as any
  const uri = res.data?.serviceConfig?.uri
  if (!uri) {
    throw new Error(`Unable to retreive uri for function at ${url}`)
  }
  return uri
}

サービスアカウントも必要

Cloud Functions実行に使用するサービスアカウントへroleを付与する必要があります
渡しの場合は、Compute Engineのデフォルトサービスアカウントを使用していました
以下の記事がわかりやすかった

サービスアカウントに必要なroleを付与

実装結果

上記で紹介したサンプルコードにて遅延実行を実現することができました〜！

タスクをキューに詰める箇所：

キューが定義されている箇所：

あとは、Sendbird APIを叩いてあげたり、その他必要な処理を施してあげれば
タイムアウト機能が出来上がるはず...

実装する上で参考にした記事

宣伝

自分が所属している会社についても紹介させてください 🙇‍♂️
弊社では物流課題解決や荷物を配送するドライバーの価値を向上させるために
「ピックゴー」というサービスを運用しています

あと、那覇Baseが本店になりました

私が在籍している那覇Baseが本店に移転しました🎉
エンジニアも募集中です！
気になる方は会社説明会も実施してますので気軽にお話聞きに来てくれたら嬉しいです😎