DBトリガーとキューテーブルで通知処理を疎結合にする

データが公開されたタイミングでメールやWebhookを送る処理は、更新処理の中から直接外部APIを呼ぶより、いったんDBのキューへ積んで別プロセスで処理したほうが扱いやすいことがあります。

今回は、更新対象テーブルの変化をDBトリガーで検知し、通知イベントと通知先をキューテーブルへ積む基本構成をまとめます。

記事公開やデータ更新のSQLから、そのままメール送信APIを呼ぶ構成にすると、外部サービスの一時障害が本来の更新処理まで巻き込んでしまいます。

データ更新
    ↓
通知イベントをDBへ記録
    ↓
更新処理は終了

別worker
    ↓
キューを読む
    ↓
メール・Webhook送信

このように分けると、外部APIが失敗しても再試行しやすくなります。

通知キューは1テーブルでも作れますが、イベントと宛先を分けると管理しやすくなります。

notify_event
  no
  kind
  object_key
  created_at
  completed_at

notify_recipient
  event_no
  user_id
  address
  status
  attempts
  last_error
  sent_at

イベント1件に複数の宛先がある場合でも、宛先ごとに送信結果や試行回数を持てます。

DBトリガーの中でメール送信まで行うのは避けます。トリガーは通知が必要かどうかを判定し、必要ならnotify_eventへ1行INSERTするだけにします。

CREATE FUNCTION enqueue_notify_event()
RETURNS trigger AS $$
BEGIN
    IF NEW.published = true
       AND (TG_OP = 'INSERT' OR OLD.published IS DISTINCT FROM NEW.published) THEN
        INSERT INTO notify_event(kind, object_key)
        VALUES ('publish', NEW.id)
        ON CONFLICT DO NOTHING;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

外部I/Oをトリガーから切り離すことで、データ更新トランザクションを短く保てます。

同じ対象へ何度も通知してはいけない場合、アプリ側のSELECT確認だけに頼らず、UNIQUE制約も使います。

UNIQUE(kind, object_key)

そしてINSERT側はON CONFLICT DO NOTHINGにしておくと、同時実行があっても重複イベントを作りにくくなります。

workerは未完了イベントを取得し、宛先を展開して送信します。1回のcronで無制限に処理するのではなく、1回あたりの件数やバッチ数に上限を設けておくと、異常時にも負荷が跳ね上がりにくくなります。

成功・失敗・再試行回数を宛先側へ記録しておけば、途中でプロセスが落ちても続きから処理できます。

通知キューを長期間の監査履歴としても使おうとすると、運用ルールが急に複雑になります。単純な用途なら、キューは処理のための一時データと割り切ったほうが分かりやすいです。

長期保存が必要なら、通知履歴は別テーブルやログへ分離したほうが、それぞれの役割が明確になります。

メール送信、Webhook、ファイル生成など、失敗や再試行があり得る処理は、本体の更新処理から切り離すだけで運用がかなり楽になります。

DBトリガーを使う場合も、トリガーには最小限のイベント生成だけを担当させるのがポイントです。