システム間でデータを同期する

Event Streams を使用して、Auth0 のアイデンティティデータの変更に合わせて外部データベースやデータウェアハウスを同期します。

Event Streams を使用すると、Auth0 のアイデンティティデータのコピーを、リレーショナルデータベース、データウェアハウス、検索インデックスなどの外部システムに保持できます。Auth0 でユーザープロファイルが作成、更新、または削除されると、イベントがストリームの送信先に配信されるため、外部システムにも同じ変更を反映できます。

​

アイデンティティデータを同期する理由

アイデンティティデータのローカルコピーを保持しておくと、次のような場合に役立ちます。

• Management APIを呼び出さずに、分析、レポート作成、コンプライアンス対応のためのクエリを実行する。
• ユーザー属性を横断して低遅延で検索する必要がある検索機能を実現する。
• アイデンティティレコードを他の業務データと結合するデータパイプラインにデータを提供する。
• 災害復旧に備えて、ユーザープロファイルの状態をバックアップしておく。

​

仕組み

1. ユーザープロファイルが変更されるたびに、Auth0 はイベントを発行します。
2. Event Stream がそのイベントを送信先 (webhook、AWS EventBridge、または Auth0 Action) に配信します。
3. ハンドラーがイベントタイプを確認し、対応する書き込み処理を外部システムに適用します。

データ同期に関連するイベントタイプは次のとおりです。

イベントタイプ	発生するタイミング
user.created	Auth0 で新しいユーザープロファイルが作成されたとき。
user.updated	既存のユーザープロファイルが更新されたとき。
user.deleted	ユーザープロファイルが Auth0 から削除されたとき。

​

前提条件

始める前に、次のものを用意してください。

• Events が有効な Auth0 テナント。利用可能なプランについて詳しくは、Create an Event Stream を参照してください。
• user.created、user.updated、user.deleted をサブスクライブしている有効な Event Stream。詳しくは、Create an Event Stream を参照してください。
• ハンドラーから書き込み可能な外部データストア (例: PostgreSQL、MySQL、データウェアハウス) 。

​

Implement でデータ同期を行う

以下のセクションでは、各イベントタイプの処理方法を説明します。ハンドラー関数は、Event Stream の送信先にかかわらず共通です。イベントタイプごとにルーティングするセクションでは、webhook と Auth0 Action の両方の送信先に対してイベントを振り分ける方法を示します。

​

user.created を処理する

Auth0 が user.created イベントを発行したら、データベースに新しい行を追加します。

async function handleUserCreated(user, time) {
    const { user_id, email, name, nickname, created_at, updated_at } = user;

    const query = `
        INSERT INTO users (user_id, email, name, nickname, created_at, updated_at, last_event_processed)
        VALUES ($1, $2, $3, $4, $5, $6, $7)
    `;
    const values = [user_id, email, name, nickname, created_at, updated_at, time];

    await pool.query(query, values);
}

​

user.updated を処理する

Auth0 が user.updated イベントを発行したら、該当する行を更新します。古いデータで上書きしないよう、イベントのタイムスタンプを last_event_processed 列と比較してください。

async function handleUserUpdated(user, time) {
    const { user_id, email, name, nickname, updated_at } = user;

    const query = `
        UPDATE users
        SET email = $1, name = $2, nickname = $3, updated_at = $4, last_event_processed = $5
        WHERE user_id = $6 AND last_event_processed < $5
    `;
    const values = [email, name, nickname, updated_at, time, user_id];

    await pool.query(query, values);
}

イベントは順番どおりに届かないことがあります。古いデータで新しいレコードが上書きされるのを防ぐため、更新を適用する前に必ずタイムスタンプを比較してください。詳しくは、Events Best Practicesをご覧ください。

​

user.deleted を処理する

Auth0 が user.deleted イベントを発行したら、該当する行を削除するか、論理削除します。

async function handleUserDeleted(user) {
    const { user_id } = user;

    const query = `DELETE FROM users WHERE user_id = $1`;

    await pool.query(query, [user_id]);
}

​

タイプ別にイベントをルーティングする

トップレベルのルーターを使って、各イベントを適切なハンドラーに振り分けます。以下の例では、Webhook と Auth0 Action の送信先に対してイベントをルーティングする方法を示しています。

Webhook

app.post("/webhook", async (req, res) => {
    const { type, time, data } = req.body;
    const user = data.object;

    try {
        switch (type) {
            case "user.created":
                await handleUserCreated(user, time);
                break;
            case "user.updated":
                await handleUserUpdated(user, time);
                break;
            case "user.deleted":
                await handleUserDeleted(user);
                break;
            default:
                console.log(`Unhandled event type: ${type}`);
        }

        res.sendStatus(204);
    } catch (err) {
        console.error("Error processing event:", err);
        res.status(500).json({ error: "Internal server error" });
    }
});

HTTP 2XX レスポンスは、できるだけ早く返してください。ハンドラーで時間のかかる処理を行う必要がある場合は、イベントを内部キューに入れて非同期で処理してください。詳しくは、Events Best Practices を参照してください。

Auth0 Action

exports.onExecuteEventStream = async (event, api) => {
    const eventType = event.message.type;
    const user = event.message.data.object;
    const time = event.message.time;

    switch (eventType) {
        case "user.created":
            await handleUserCreated(user, time);
            break;
        case "user.updated":
            await handleUserUpdated(user, time);
            break;
        case "user.deleted":
            await handleUserDeleted(user);
            break;
        default:
            console.log(`Unhandled event type: ${eventType}`);
    }
};

​

重複や順序の問題に備える

Event Streams は少なくとも1回の配信を保証するため、ハンドラーが同じイベントを複数回受信することがあります。これに安全に対処するには、次のようにします。

• イベント ID を追跡する。 処理した各イベントの id を保存し、すでに処理済みのイベントはスキップします。
• タイムスタンプを比較する。 各イベントのペイロードには、data.object の created_at フィールドと updated_at フィールドが含まれます。これらのフィールドを使って、受信したイベントがシステムにすでに記録されているものより新しいかどうかを判断します。
• 冪等な書き込みを使う。 同じイベントを2回適用しても結果が変わらないように、データベース操作を設計します。たとえば、PostgreSQL では INSERT ... ON CONFLICT DO UPDATE を使用します。

INSERT INTO users (user_id, email, name, nickname, created_at, updated_at, last_event_processed)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (user_id) DO UPDATE
SET email = EXCLUDED.email,
    name = EXCLUDED.name,
    nickname = EXCLUDED.nickname,
    updated_at = EXCLUDED.updated_at,
    last_event_processed = EXCLUDED.last_event_processed
WHERE users.last_event_processed < EXCLUDED.last_event_processed;

​

同期を確認する

ハンドラーをデプロイしたら、Auth0 でテストユーザーを作成し、次の点を確認します。

1. 正しいプロフィールデータを持つ新しい行が外部データベースに作成されること。
2. Auth0 でユーザーの名前またはメールアドレスを更新し、その変更がデータベースの行に反映されることを確認します。
3. Auth0 でユーザーを削除し、データベース内の行も削除される (または削除済みとしてマークされる) ことを確認します。

Event Streams のテストについて詳しくは、Event Testing, Observability, and Failure Recovery を参照してください。

​

詳しくはこちら

• イベントカタログ
• Create an Event Stream
• Events Best Practices