Webhooks に移しても、デプロイの瞬間にイベントは落ちる — Gemini 長時間オペレーションを照合で二重化した運用メモ

// operation-ledger.ts
export type OpStatus = "pending" | "succeeded" | "failed";
 
export interface OpRecord {
  opName: string;        // Gemini が返すオペレーション名（resource name）
  jobKind: string;       // "batch-review-classify" など、用途の識別子
  status: OpStatus;
  submittedAt: number;   // epoch ms
  settledAt?: number;    // 終端確定時刻
  settledBy?: "webhook" | "reconcile";
  attempts: number;      // 照合で見に行った回数
  sideEffectDone: boolean; // 結果の書き戻しが完了したか
}
 
const TTL_SECONDS = 60 * 60 * 24 * 14; // 14日保持して監査に使う
 
export async function putOp(kv: KVNamespace, rec: OpRecord): Promise<void> {
  await kv.put(`op:${rec.opName}`, JSON.stringify(rec), {
    expirationTtl: TTL_SECONDS,
    metadata: { status: rec.status, submittedAt: rec.submittedAt },
  });
}
 
export async function getOp(kv: KVNamespace, opName: string): Promise<OpRecord | null> {
  const raw = await kv.get(`op:${opName}`);
  return raw ? (JSON.parse(raw) as OpRecord) : null;
}
 
export async function registerSubmission(
  kv: KVNamespace,
  opName: string,
  jobKind: string,
): Promise<void> {
  await putOp(kv, {
    opName,
    jobKind,
    status: "pending",
    submittedAt: Date.now(),
    attempts: 0,
    sideEffectDone: false,
  });
}

// webhook-handler.ts
import { getOp, putOp } from "./operation-ledger";
import { verifySignature } from "./verify-signature";
import { runSideEffect } from "./side-effect";
 
export async function handleWebhook(req: Request, env: Env): Promise<Response> {
  const raw = await req.text();
  if (!(await verifySignature(raw, req.headers, env.WEBHOOK_SECRET))) {
    return new Response("invalid signature", { status: 401 });
  }
 
  const event = JSON.parse(raw) as { opName: string; state: string };
  // 終端状態だけを扱う。中間イベントは無視してよい
  const terminal =
    event.state === "SUCCEEDED" ? "succeeded" :
    event.state === "FAILED" ? "failed" : null;
  if (!terminal) return new Response("ignored (non-terminal)", { status: 200 });
 
  await settle(env.OPS_KV, event.opName, terminal, "webhook");
  // 200 を返すのは副作用の確定後。早すぎる ACK は再送を止めて取りこぼす
  return new Response("ok", { status: 200 });
}
 
// 終端確定 + 副作用。webhook と reconcile の双方から呼ばれる共通路
export async function settle(
  kv: KVNamespace,
  opName: string,
  status: "succeeded" | "failed",
  by: "webhook" | "reconcile",
): Promise<void> {
  const rec = await getOp(kv, opName);
  if (!rec) return;                 // 台帳にない＝対象外
  if (rec.status !== "pending") return; // 既に閉じている＝冪等に無視
 
  if (status === "succeeded" && !rec.sideEffectDone) {
    await runSideEffect(opName);    // 結果の書き戻しは1回だけ
  }
  rec.status = status;
  rec.settledAt = Date.now();
  rec.settledBy = by;
  rec.sideEffectDone = status === "succeeded";
  await putOp(kv, rec);
}

// reconcile.ts
import { getOp, putOp } from "./operation-ledger";
import { settle } from "./webhook-handler";
import { fetchOperationState } from "./gemini-ops";
 
const GRACE_MS = 90_000;       // 送信から90秒は webhook を待つ
const STUCK_MS = 6 * 3600_000; // 6時間 pending なら異常として通報
 
export async function reconcile(env: Env): Promise<void> {
  const now = Date.now();
  let cursor: string | undefined;
 
  do {
    const page = await env.OPS_KV.list({ prefix: "op:", cursor, limit: 1000 });
    cursor = page.list_complete ? undefined : page.cursor;
 
    for (const key of page.keys) {
      const meta = key.metadata as { status?: string; submittedAt?: number } | undefined;
      // 本体を読まずに metadata で粗くふるう。pending 以外は読み込まない
      if (meta?.status !== "pending") continue;
      if (meta.submittedAt && now - meta.submittedAt < GRACE_MS) continue;
 
      const rec = await getOp(env.OPS_KV, key.name.slice(3));
      if (!rec || rec.status !== "pending") continue;
 
      const state = await fetchOperationState(rec.opName, env); // operations.get 相当
      if (state === "SUCCEEDED" || state === "FAILED") {
        await settle(env.OPS_KV, rec.opName, state === "SUCCEEDED" ? "succeeded" : "failed", "reconcile");
        continue;
      }
 
      // まだ走っている。試行回数を進め、長すぎるものは通報する
      rec.attempts += 1;
      await putOp(env.OPS_KV, rec);
      if (now - rec.submittedAt > STUCK_MS) {
        await alertStuck(rec); // Slack 等へ。自動では閉じない
      }
    }
  } while (cursor);
}

// 照合ループの中に追加する第2の走査
if (rec.status === "succeeded" && !rec.sideEffectDone) {
  await runSideEffect(rec.opName);
  rec.sideEffectDone = true;
  await putOp(env.OPS_KV, rec);
}