Upstash・Redis・QStashで実現するリアルタイム緊急対応システムの構築

本記事では、Upstashを活用して国の避難所マップに関する情報をRedisで安全に保存・アクセスする方法と、QStashを使ってデータベースをリアルタイムに更新する方法について解説します。

はじめに

今日の世界では、自然災害や軍事的な脅威がますます深刻化しており、それに伴い社会サービス分野におけるデジタル化のニーズも高まっています。

AMBERアラートのような緊急放送システムから、COVID-19の追跡アプリ、SOSシステムまで、国に影響を及ぼすあらゆる危機的状況に対して、幅広いテクノロジーが活用されてきました。

本記事では、Redisとサーバーレスワークロードがどのようにリアルタイム緊急対応システムの構築において重要な役割を果たせるのかを見ていきます。このシステムは、市民が最寄りの避難所(バンカー)を見つけるだけでなく、収容能力や設備(バリアフリー設備など)、利用可能な資源(水、電気、医薬品など)に関する情報も提供します。

Courtesy of Project Cloud4

なぜサーバーレスなのか?

いわゆる「サーバーレス」という概念は、ここ数年で開発者の間で広く普及しました。このアーキテクチャを採用するスタートアップが増えているだけでなく、大企業での採用事例も増加しています。

特に国家機関のような組織にとっては、迅速な開発環境と自動管理型サービスによって、緊急対応システム構築の複雑さと運用コストを大幅に削減できます。さらに、以下の2つの要素も考慮する必要があります。

1. 予算:もし国が新たな災害に見舞われていない場合、意思決定者は未使用のコンピューティングリソースへの支払いを承認できるでしょうか?おそらく難しいでしょう。
2. トラフィックの急増:警報が発報され、数百万人が最寄りの避難所を探すためにWebプラットフォームへアクセスする状況を想像してください。REST APIはこの突然のRPS(秒間リクエスト数)の急増にどう対応するのでしょうか?不可欠なリアルタイムコンポーネントはどうでしょうか?

こうした課題はすべて、UpstashやAblyのような真のサーバーレスプラットフォームを使用することで解決できます。これらは自動スケーリングを行い、実際に使用したリソース分だけ課金される仕組みです。

なぜRedisなのか?

本プロジェクトのデータ要件には、可能な限り高速に実行しなければならないさまざまなアルゴリズム処理が含まれます。避難所の情報を表現するための基本的なデータ構造から、特定の場所にある避難所をさまざまな基準で簡単にソートする方法まで、RedisはNoSQLデータベースの柔軟性を維持しながら、開発を簡素化・加速させる多彩な機能を提供してくれます。

なぜQStashなのか?

従来のメッセージキューはイベント駆動型システムを設計するための堅実な手段ですが、一つ注意点があります。その多くはAMQPのようなステートフルなプロトコルに完全に依存しているため、サーバーレス関数のような短命な実行環境では使用できません。

QStashは、HTTPというステートレスなプロトコル(こうした環境で一般的に使われるもの)によるリクエストを可能にすることで、この問題を解決します。これをWebhookと組み合わせることで、データベースとの間でデータを更新・取得するリアルタイムパイプラインを構築できます。

必要なもの

P.S.:サンプルコードだけ確認したい方は、「必要なもの」と「セットアップ」のセクションはスキップしても構いません。これらは使用技術とその使い方を示すためのものです。

このようなシステムを自力で構築したい場合は、以下が必要です。

• Upstashアカウント(Redisデータベースの作成と、QStashエンドポイントまたはトピックの利用のため)
• Ablyアカウントと設定済みのWebhook統合
• Next.jsプロジェクト
• ngrokのようなトンネリングサービス(ローカルのリクエストをインターネット上のURLにプロキシしたい場合)
• 必要なライブラリに対応した任意のパッケージマネージャー。本記事ではnpmを使用しますが、必須ではありません。

注意:本記事では、アプリケーションのアーキテクチャにおける各技術の使用方法を実証するための、基本的な概念実証(PoC)サンプルを提供します。提示するコードは本番環境対応ではなく、読みやすさのために簡略化・短縮されています。デプロイする場合は、必要な修正(例外処理、セキュリティ対策など)を必ず行ってください。

セットアップ

サーバーレス関数

Next.jsのEdge API Routesを使用して、ユーザーの所在地に近い場所でサーバーレスワークロードを実行します。

Next.jsプロジェクトを作成するには、親ディレクトリでnpx create-next-app@latest <your-app-name>を実行します。その後、npm run devで開発サーバーを起動し、npm run buildで本番バンドルを作成、npm run startで本番サーバーを起動できます。

Next.jsはAPI Routesに対してデフォルトでNode.js環境を使用します。これを変更する方法は2つあります。

1. ルートごとに変更する方法。対象のルートに以下のexportを追加します。

   export const config = {
    runtime: "experimental-edge",
   };

2. グローバルに変更する方法。edgeをデフォルトランタイムにするには、next.config.jsファイルに以下を追加します。

   const nextConfig = {
    // ...
    experimental: {
     runtime: "experimental-edge",
    },
    // ...
   };

Upstash

Upstashプロバイダーの作成方法については、公式ガイドを参照してください。

QStash受信用ルートのセットアップについては、こちらのガイドを参照してください。AblyはWebhookメッセージのエンコード方式としてJSONとMessagePackの2つを提供しています。前者(JSON)を使用する場合、暗号化の互換性問題に対応するためにJWTの代わりに認証ヘッダーを使用する必要があるかもしれません。ここではシンプルにJSONを採用します。

Ably

リアルタイム機能には、可能な限りWebSocketをデフォルトとするAblyのプロトコルを使用します。始めるには公式ガイドを参照してください。Next.jsを使用している場合は、@ably-labs/react-hooks npmパッケージも参考になります。

また、QStashのURLを指すようにWebhook統合を設定し、必要なヘッダーを追加してください。

本番環境に移行する場合は、認証とセキュリティも必ず確認してください。

オプション:ローカルトンネリング

ngrokの使い方については、公式ガイドを参照してください。

アーキテクチャ

フロー

アプリケーションのフローをより深く理解するために、次の図(excalidraw.comで作成)を見てみましょう。

ご覧のとおり、2種類のユーザーを定義しています。

1. 市民:避難所または地域の情報にアクセスでき、避難所へ向かう意向を通知できます。
2. 管理者:避難所の物流データを修正し、空き状況を更新できます(例:ある人/家族が現地に到着したことを確認)。

フロントエンドとバックエンドをできる限り疎結合に保ちながら、データをリアルタイムで提供したいと考えています。そこで、Ablyチャネルを使ってクライアントと通信し、空き状況が変化するたびにWebhookを発火させます。

また、ステートフルな接続は使用すべきではないため、バックエンドからのメッセージ公開にはAblyのREST APIを使用します。

データ

現在、保存すべき主なデータ型は2つあります。

• 避難所の情報:Redisハッシュとして保存でき、以下のプロパティを持ちます。

  • キー名:country-city-number形式(例:RO-CJ-01)
  • 空き状況(例:300)
  • 設備(例:["disabled persons special acces", "counseling"])
  • 資源:

    {
     resource: quantity,
     or
     resource: list
     ...
    }

    例:

    {
     "water": "200 liters",
     "medicine": ["insuline", ...]
    }

  • 電気:yes/no
  • 暖房:yes/no
  • 位置情報:[longitude, latitude]または

    {
     longitude: number,
     latitude: number
    }

  • _id:ランダムに生成します。

• 地域の情報

  指定された地域で利用可能な避難所を見つけるためにネストしたクエリを実行することも可能ですが、クエリを簡素化・最適化するために、別のデータ構造として保存する方が良いでしょう。そのためには、Redisセットを使用し、要素名を<geographical-unit>-<name>(例:country-ROやcity-CJ)とします。これにより重複も防止できます。

  例:

   city-CJ : ["RO-CJ-01", "RO-CJ-02", "RO-CJ-03"]

しかし、まだ1つ問題が残っています。市民が避難所へ向かう意向を表明したことと、管理者の確認によって引き起こされた実際の収容状況の更新を、どう区別すればよいのでしょうか?これには2つの方法があります。

1. 避難所の空き状況キーを常に更新する方法:
   ユーザーが来訪の意向を表明する(単独または家族など)→ xだけインクリメントする。
   管理者がその確認が無効であると判断する → xだけデクリメントする(または-xをインクリメントする)。

   この方法の問題点は、管理者が信頼できる情報源として機能し、市民の位置を追跡し続けなければならないことです。また、避難所のデータを即座に信頼することができません。

2. shelter-"availability"(例:RO-CJ-01-availability独立した文字列キーを作成し、リアルタイム更新された値を保存する方法。
こうすることで、リアルタイムに更新しつつ、ハッシュのavailabilityキーを信頼できる情報源として維持できます。

   クライアント側では、ハッシュと文字列キーの両方を取得し、それぞれを実際の空き状況と予測値として表示します。代替案としては、文字列のみを取得し、必要に応じて信頼できる情報源に戻す方法もあります。これはネットワーク負荷を軽減できますが、管理者が常時データを検証する必要があり(最初のシナリオと同様)、ネットワーク取得を本当に減らすには、避難所のavailabilityキーを別途保存する必要があります(他の情報も取得する必要があることを忘れないでください!)。

import { useEffect, useState } from "react";
 
import "../styles/globals.css";
 
import { configureAbly } from "@ably-labs/react-hooks";
 
export default function App({ Component, pageProps }) {
 const [loaded, setLoaded] = useState(false);
 useEffect(() => {
 configureAbly({
 // 本番システムでは認証を使用してください
 key: process.env.NEXT_PUBLIC_ABLY_API_KEY,
 });
 setLoaded(true);
 }, []);
 
 if (!loaded) return <div>loading...</div>;
 return <Component {...pageProps} />;
}

import { useChannel } from "@ably-labs/react-hooks";
 
export default function Test() {
 const [availability] = useChannel("availability", (msg) => {
 console.log(msg);
 });
 
 return <></>;
}

export default async (req) => {
 let data = await req.json();
 
 if (data.type === "shelter") {
 // 'type'キーは不要になったため削除
 delete data.type;
 const { name: shelter, availability } = data;
 
 let shelterData = data;
 // ランダムなIDを生成
 shelterData._id = Math.random()
 .toString(36)
 .replace(/[^a-z]+/g, "")
 .substring(0, 7);
 
 // 避難所の情報をハッシュとして保存
 await redis.hset(shelter, shelterData);
 
 // リアルタイム更新される空き状況を文字列として保存
 await redis.set(shelter + "-availability", availability);
 return new Response("ok");
 } else if (data.type === "location") {
 const { name: location, shelters } = data;
 
 await redis.sadd(location, ...shelters);
 
 return new Response("ok");
 }
};

availability.publish("update", {
 shelter: "RO-CJ-01",
 availability: 1, // ユーザーがキャンセルする場合は-x
});

await redis.incrby(shelter + "-availability", value);