この記事を読む前に

この記事では、教室管理システムに実装した、LINEへ通知を配信するシステム紹介します。このシステムの概要についてはProjectの教室管理システムをご覧ください。

xe-portfolio xe-portfolio | 教室管理システム | 教室業務改善ツール | プロジェクト xe portfolio https://xenosaba.com/projects/classroom-ops-support/classroom-management

それと旧シフト通知システムについても少し触れているので、そちらもよかったらご覧ください。

xe-portfolio xe-portfolio | シフト通知システム（旧） | 教室業務改善ツール | Projects xe portfolio https://xenosaba.com/projects/classroom-ops-support/shift-notifier

はじめに

LINEを使ってシステムを作っている理由については、紹介ページを見てもらうとして以下の二つの機能にLINE APIを使用して実装しています。

• 月間シフト表を講師用LINEグループに送信する機能
• 勤務２時間前に、各講師宛に勤務前通知を送信する機能

で、このようなLINEに通知を送るだけならば、管理画面のAPIから直接Messaging APIを呼べば実装できると考えていました。

しかし現在の教室管理システムは、旧システムとは違いLINEに通知するシステム以外にも様々な機能を搭載したWEBアプリです。

そこでいくつか不安な点がありました。

例えば、LINE APIの応答が遅れた場合、管理画面のリクエストもその分待つことになります。またLINE側で一時的な障害が起きたり、429や5xxのエラーが返ってきたりした場合は、管理者に送信失敗として返すのでしょうか。

それに、リクエストがタイムアウトした時の処理はどうでしょうか。こっちでは失敗したように見えても、LINE側ではすでに通知が届いているかもしれません。その状態で単純に再送した場合、同じ通知が2回届いてしまう可能性があります。

（まぁシフト通知するだけだから2回あったとしても致命的ではないですが…）

どちらにせよ利用者のUXを高めるためになんとかしたいところでした。

そこでCloudflare Queuesというメッセージキューイングシステムを使ってみることにしました。

Cloudflare Queuesとは

Cloudflare Queuesは、アプリケーションで発生した処理をいったんメッセージとして預け、あとからコンシューマー側で非同期に処理するためのメッセージキューサービスです。

AWSで言うとAWS SQSのようなものですね。

下の図は、APIを直接呼び出す場合とCloudflare Queuesを比較したものになります。

通知を依頼する側をプロデューサー、Queueからメッセージを受け取って処理する側をコンシューマーと呼びます。

管理画面側では、Queueへメッセージを登録した時点で処理を終了できます。その後、コンシューマーとなるWorkerがQueueからメッセージを取得し、LINEへの通知などの処理を実行します。

これにより、外部APIの応答時間や一時的な障害が、管理画面の処理へ直接影響しにくくなります。 また、Queueは大量のメッセージを一時的に蓄えたり、複数のメッセージをまとめて処理したりできます。バッチサイズや待機時間などを設定することで、処理をまとめる単位やタイミングを調整でき、短時間に依頼が集中した場合の負荷を吸収する役割もあります。

ちなみにこれ、無料でも利用できます！

もちろん無料枠の範囲内ですが、今回のシステムは通知数が少ないため、現時点では無料枠の範囲内で利用できています♪

Cloudflare Docs Pricing Cloudflare Queues pricing for standard operations with included free usage. https://developers.cloudflare.com/queues/platform/pricing/ ここまではCloudflare Queuesについて説明しました。

次に、実際にこの教室管理システムでどのようにCloudflare Queuesを活用したかを紹介していきます。

教室管理システムでの処理の流れ

月間シフト表をLINEグループへ送信する機能

まずは、管理者が月間シフト表をLINEグループへ送信する場合を紹介します。

管理者が管理画面からシフト表送信のボタンを押すと、次のような処理が行われます。

1. 管理画面でシフト表の画像を作成する
2. Workerが画像や対象月を検証する
3. 画像をR2へ保存する
4. D1に配信依頼を保存する
5. Queueへ配信依頼のメッセージを登録する
6. Consumer WorkerがQueueからメッセージを取得する
7. LINE Messaging APIへ通知する

一見すると、Queueにメッセージを送って、Consumer WorkerでLINEへ通知しているだけのように見えますね。

しかし、Queueには画像データやLINEユーザーIDそのものを登録しているわけではありません。

実際に登録しているのは、配信の種類を示すkindと、配信依頼を識別するdeliveryIdだけです。

{
  kind: "shift_publication",
  deliveryId: "..."
}

Consumer Workerは、このdeliveryIdを使ってD1から配信依頼を取得します。

月間シフト表の場合は、その配信依頼に紐づいている対象月、閲覧用のdocumentId、初回公開か更新通知かといった情報をD1から取得します。

その後、documentIdからLIFFの閲覧URLを作り、閲覧ボタンを含むFlex MessageをLINEグループへ送信します。

シフト表の画像自体はQueueやConsumer Workerへ渡さず、非公開のR2に保存したままにしています。

次にProducerとConsumer側の処理を詳しく説明します。

Producer WorkerからQueueへ登録する

管理者がシフト表の送信ボタンを押すと、Workerは画像をR2へ保存し、画像に関する情報と配信依頼をD1へ保存します。

この時点で、D1上の配信依頼はpendingになっています。

D1への保存が完了したあと、Producer WorkerからQueueへメッセージを登録します。

await env.LINE_DELIVERY_QUEUE.send({
  kind: "shift_publication",
  deliveryId,
});

LINE_DELIVERY_QUEUEは、WorkerからCloudflare Queuesを利用するために設定したBindingです。 send()が完了すると、D1に保存している配信依頼の状態をqueuedへ変更します。

Consumer Workerでメッセージを処理する

Queueにメッセージが登録されると、CloudflareがConsumer側のqueueハンドラーを呼び出します。

今回の実装では、ProducerとConsumerで別々のWorkerを用意しているわけではなく、同じWorkerにHTTPリクエスト用のfetchと、Queue処理用のqueueを定義しています。

export default {
  fetch: app.fetch,
  queue: handleLineDeliveryQueue,
  scheduled: runScheduledTasks,
};

Consumer側では、次の順番で処理します。

1.deliveryIdを使ってD1から配信依頼を取得する
2.配信状態をprocessingへ変更する
3.配信依頼に紐づくシフト表の公開レコードを取得する
4.documentIdからLIFFの閲覧URLを作る
5.閲覧ボタンを含むFlex MessageをLINE Messaging APIへ送信する
6.送信に成功したら、D1上の配信状態をsentへ変更する
7.Queueのメッセージをack()する

ack()は、そのQueueメッセージの処理が完了したことをCloudflare Queuesへ伝えるための操作です。

ここまでが、問題なくLINEへ送信できた場合の処理です。

LINEへの送信に失敗した場合

では次に、LINE Messaging APIが429や5xxを返した場合はどうなるのでしょうか？

今回の実装では、すべてのエラーを同じように再試行するのではなく、エラーの種類によって処理を分けています。

LINEからの応答	処理
2xx	送信成功としてsentに変更し、ack()する
409	409で、x-line-accepted-request-idが返された場合は、同じリクエストがすでに受理されているものとして成功扱いにする。
5xx	Queueで再試行する
通信エラー・タイムアウト	Queueで再試行する
400・401・403・429など	failedに変更し、ack()する

LINE公式では、500番台のエラーやタイムアウトは再試行し、その他の400番台は再試行しないよう案内されています。そのため、このシステムでも5xxや通信エラーだけを再試行の対象にしています。 429も再試行しないエラーとして扱い、D1の配信状態をfailedへ変更して処理を終了します。

developers.line.biz 失敗したAPIリクエストを再試行する メッセージの送信処理は失敗する可能性があり、その場合は500番台のエラーが発生したり、リクエストがタイムアウトしたりします。ただし、このようなエラーが発生した場合でも、メッセージは送信されている可能性があります。つまり、エラーが起きたからといって同じリクエストを再送信すると、以下の図のように、ユーザーは同じメッセージを... https://developers.line.biz/ja/docs/messaging-api/retrying-api-request/

Queueを使って再試行する

5xxや通信エラーが発生した場合は、Queueメッセージをack()せず、retry()を呼び出します。

message.retry({
  delaySeconds,
});

これにより、指定した時間が経過したあと、同じメッセージがConsumer Workerへ再配信されます。

再試行までの時間は固定ではなく、失敗するたびに待ち時間を延ばす指数バックオフにしています。最初は2分後に再試行し、その後も失敗した場合は4分、8分と間隔を延ばします。

再試行を待っている間、D1上の配信状態はqueuedとして残します。再試行に成功した場合はsentへ変更し、Queueメッセージをack()します。

再試行回数を超えた場合

Consumer側では、再試行回数を次のように設定しています。

{
  "max_retries": 5,
  "dead_letter_queue": "tops-line-delivery-dlq"
}

初回の処理とは別に最大5回まで再試行し、それでも成功しなかった場合はD1の配信状態をfailedへ変更します。

Queueメッセージは通常のQueueからDead Letter Queue、通称DLQへ移されます。これにより、繰り返し失敗するメッセージが通常のQueueへ残り続けることを防いでいます。 なお、400番台など再試行しないエラーはack()して処理を終了するため、DLQには移動しません。失敗したことやエラーコードはD1に残します。

Cloudflare Docs Batching, Retries and Delays Configure message batching, retry behavior, and delivery delays for Cloudflare Queues. https://developers.cloudflare.com/queues/configuration/batching-retries/

再試行による重複送信を防ぐ

さっきのLINEのリファレンスにあった説明になるのですが、

通信エラーやタイムアウトが発生した場合、Consumer Workerからは失敗したように見えても、LINE側ではすでにリクエストを受理している可能性があります。 そこで、LINE Messaging APIへ送信するときは、配信ごとに発行したUUIDをX-Line-Retry-Keyとして指定しています。

headers: {
  Authorization: `Bearer ${channelAccessToken}`,
  "Content-Type": "application/json",
  "X-Line-Retry-Key": retryKey,
}

再試行するときも同じリトライキーを使用します。

最初のリクエストがすでにLINE側で受理されていた場合、再試行時には409が返されます。この場合は重複送信せず、すでに送信が受理されているものとしてD1の状態をsentへ変更します。

リトライキーの有効期間は最初の送信から24時間です。そのため、Consumer Workerでは24時間を過ぎたリトライキーを使って自動再試行しないようにしています。

失敗した月間シフト表の配信を管理者が手動で再試行する際、初回処理から24時間以上経過していれば、新しいリトライキーを発行します。24時間以内なら同じキーを再利用します。

Queueへの登録自体に失敗した場合

ここまでは、Queueに登録されたメッセージをConsumer Workerが処理するときの失敗について説明しました。

しかし、その前段階である Queue.send() 自体が失敗する可能性もあります。

このシステムでは、Queueへ登録する前に、画像をR2へ保存し、配信依頼をD1へ保存しています。そのため、Queueへの登録に失敗しても、配信依頼そのものはD1に残っています。

Queueへの登録に失敗した場合は、D1の配信状態をpendingのままにして、次の情報を記録します。

status: pending
last_error_code: QUEUE_ENQUEUE_FAILED
next_attempt_at: 15分後

ここでのpendingは、配信依頼はD1に保存されているものの、Queueへの登録完了を確定できていない状態を表します。

その後、15分ごとに実行しているCron Triggerが、再登録できる配信依頼をD1から取得します。

SELECT id, kind, attempt_count
FROM line_message_deliveries
WHERE status = 'pending'
  AND next_attempt_at <= ?

対象となる配信依頼が見つかった場合は、もう一度Queueへの登録を試します。

await enqueueLineDelivery(
  env,
  {
    kind: row.kind,
    deliveryId: row.id,
  },
  row.attempt_count,
);

再登録に成功すると、D1の配信状態をqueuedへ変更します。その後は通常どおり、Consumer Workerがメッセージを取得してLINEへの送信を行います。

再登録にも失敗した場合は、再びpendingのまま次回のCron実行を待ちます。

ここで重要なのは、Cronが自動で再登録するのは、D1上でpendingになっている配信依頼だけという点です。

LINE Messaging APIへの送信に失敗してfailedになった配信依頼は、Cronの再登録対象にはしていません。LINEへの一時的な送信失敗はQueue自身の再試行機能に任せることで、Cronによって同じ処理が何度も登録されることを防いでいます。

まとめると

Queueへの登録失敗
  → D1にpendingとして残す
  → CronからQueueへ再登録

LINEへの一時的な送信失敗
  → Queueのretry()で再試行
  → 上限を超えたらDLQへ移動

このように、Queueへ登録するまでの再試行はD1とCron、Queueへ登録したあとの再試行はCloudflare Queuesと役割を分けています。

勤務2時間前に通知する機能

次に、各講師へ勤務前通知を送信する場合を紹介します。

月間シフト表の配信は、管理者が送信ボタンを押したことをきっかけに始まります。一方、勤務前通知は、15分ごとに実行されるCron Triggerをきっかけに配信依頼を作成します。

処理の流れは次のとおりです。

1. Cron Triggerを15分ごとに実行する
2. D1から通知対象となる確定済みシフトを取得する
3. D1に配信依頼を保存する
4. Queueへ配信依頼のメッセージを登録する
5. Consumer WorkerがQueueからメッセージを取得する
6. D1からシフトと送信先を取得する
7. LINE Messaging APIへ勤務前通知を送信する

Cronでは、現在時刻から2時間15分以内に勤務が始まるシフトを取得しています。

15分間隔で対象を確認しているため、通常は勤務開始の2時間から2時間15分前に配信依頼が作られます。

勤務前通知の場合も、Queueへ登録する情報はkindとdeliveryIdだけです。

{
  kind: "shift_reminder",
  deliveryId: "..."
}

講師名、教室名、勤務時間、LINEユーザーIDなどはQueueへ登録しません。Consumer WorkerがdeliveryIdを使ってD1から必要な情報を取得し、勤務前通知のメッセージを作成します。

重複通知を防ぐ

Cronは15分ごとに実行されるため、同じシフトが複数回取得される可能性があります。

そこで、配信依頼にはシフトを識別するshiftEntryIdと、そのシフトの更新日時を表すshiftVersionを保存しています。

同じ種類、同じシフト、同じバージョンの配信依頼はD1へ重複して登録できないようにしているため、Cronが同じシフトを再び取得しても、新しい配信依頼は作成されないようにしています。

通知後にシフトが変更された場合

勤務前通知を送信したあとに、管理者が勤務時間や教室などを変更する場合もあります。

Cronで通知対象の確定シフトを取得した際、過去に送信済みのバージョンと現在のバージョンが異なれば、勤務予定の変更通知として配信依頼を作成します。

{
  kind: "shift_correction",
  deliveryId: "..."
}

送信直前にもシフトを確認する

Queueへ登録してからConsumer Workerが処理するまでの間に、シフトが変更されたり、確定が取り消されたりする可能性があります。

そのためConsumer Workerでは、LINEへ送信する直前にD1からシフトを取得し、次の状態を確認しています。

• シフトが確定済みであること
• Queueへ登録したときとshiftVersionが一致していること
• 勤務開始時刻を過ぎていないこと
• 講師のLINE通知が有効になっていること

条件を満たさない場合はLINEへ送信せず、D1の配信状態をcanceledへ変更してQueueメッセージをack()します。

これはCloudflare Queuesが自動的に判断しているわけではなく、Consumer Worker側で実装している処理です。

勤務前通知の送信で5xxや通信エラーが発生した場合は、月間シフト表の場合と同じようにQueueのretry()を使って再試行します。

実装して分かったこと

最初は、管理画面からLINE Messaging APIの呼び出しを切り離すためにCloudflare Queuesを使おうと考えていました。

実際に実装してみると、Queueへメッセージを登録するだけではなく、どの段階で失敗したのかによって再試行方法を分ける必要がありました。

D1への配信依頼保存後に起きたQueue登録処理の失敗
  → D1とCronで回収する

Queueへ登録したあとの一時的な失敗
  → Cloudflare Queuesで再試行する

LINE側で受理された可能性がある場合
  → X-Line-Retry-Keyで重複送信を防ぐ

送信前に配信内容が変わった場合
  → D1の現在の状態を確認して送信を中止する

また、Queueには最低限の情報だけを登録し、配信内容や処理状態をD1へ保存したことで、月間シフト表と勤務前通知を同じ配信処理で扱えるようになりました。

まとめ

今回は、管理画面やCronが作成した配信依頼をCloudflare Queuesに登録し、コンシューマー側でLINE Messaging APIを呼び出す仕組みを実装しました。

Queueを利用することで、管理画面の処理とLINEへの送信処理を切り離し、一時的なエラーが発生した場合もConsumer Worker側で再試行できるようになりました。

一方で、Queueへの登録失敗、LINEへの重複送信、古くなった配信内容などは、Queuesを導入するだけで自動的に解決するわけではありません。

D1による配信状態の管理、Cron Triggerによる未登録データの回収、X-Line-Retry-Keyによる重複送信対策を組み合わせることで、今回必要としていた配信処理を実装できました。