Managed Agent の完了通知を取りこぼさない — サーバーレス受信とポーリング突き合わせの二重化設計

// worker: Managed Agent 完了 webhook の受信口
// KV binding: JOBS（ジョブレコード） / SEEN（配信重複の記録）
interface Env {
  JOBS: KVNamespace;
  SEEN: KVNamespace;
  WEBHOOK_SECRET: string;
}
 
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (req.method !== "POST") return new Response("nope", { status: 405 });
 
    const raw = await req.text();
    // 1) 署名検証（本文と共有秘密から HMAC を再計算して定数時間比較）
    const ok = await verifySignature(raw, req.headers.get("x-agent-signature"), env.WEBHOOK_SECRET);
    if (!ok) return new Response("bad signature", { status: 401 });
 
    const ev = JSON.parse(raw) as {
      deliveryId: string;   // 配信ごとに一意（重複検知に使う）
      jobId: string;        // ジョブの一意キー
      state: "running" | "succeeded" | "failed";
      seq: number;          // 送信側が振る単調増加の版番号
      finishedAt?: string;
    };
 
    // 2) 配信重複を弾く（同じ deliveryId を二度処理しない）
    if (await env.SEEN.get(ev.deliveryId)) {
      return new Response("duplicate", { status: 200 }); // 200 で返し再送を止める
    }
 
    // 3) 状態遷移を単調に適用（未知ジョブでもレコードを作って受け止める）
    await applyTransition(env, ev);
 
    // deliveryId は 3 日保持すれば十分（送信側のリトライ窓を超える長さ）
    await env.SEEN.put(ev.deliveryId, "1", { expirationTtl: 60 * 60 * 24 * 3 });
    return new Response("ok", { status: 200 });
  },
};

type JobState = "unknown" | "running" | "succeeded" | "failed";
const RANK: Record<JobState, number> = { unknown: 0, running: 1, succeeded: 2, failed: 2 };
 
interface JobRecord {
  jobId: string;
  state: JobState;
  seq: number;          // 採用済みイベントの最大 seq
  finishedAt?: string;
  claimedAt?: string;   // 副作用を実行した印（後述）
}
 
async function applyTransition(env: Env, ev: { jobId: string; state: JobState; seq: number; finishedAt?: string }) {
  const key = `job:${ev.jobId}`;
  const cur = (await env.JOBS.get(key, "json")) as JobRecord | null;
 
  // レコードが無ければ unknown から始める（未知ジョブの取りこぼし防止）
  const base: JobRecord = cur ?? { jobId: ev.jobId, state: "unknown", seq: -1 };
 
  // 単調性: より新しい seq、かつ後退しない状態遷移のみ採用する
  const forward = ev.seq > base.seq && RANK[ev.state] >= RANK[base.state];
  if (!forward) return; // 古い通知・巻き戻し通知は無視
 
  const next: JobRecord = {
    ...base,
    state: ev.state,
    seq: ev.seq,
    finishedAt: ev.finishedAt ?? base.finishedAt,
  };
  await env.JOBS.put(key, JSON.stringify(next));
 
  // 終端状態に初めて到達したときだけ、副作用を一度実行する
  if ((next.state === "succeeded" || next.state === "failed") && !base.claimedAt) {
    await runSideEffectOnce(env, next);
  }
}

// scheduled(): 一定間隔で未完了ジョブをクラウドと突き合わせる
export async function reconcile(env: Env) {
  const list = await env.JOBS.list({ prefix: "job:" });
  const now = Date.now();
 
  for (const k of list.keys) {
    const rec = (await env.JOBS.get(k.name, "json")) as JobRecord;
    if (rec.state === "succeeded" || rec.state === "failed") continue; // 終端は照合不要
 
    // クラウド側の権威的な状態を取得（Managed Agents の取得 API）
    const remote = await fetchJobStatus(env, rec.jobId);
    if (!remote) continue;
 
    // webhook と同じ applyTransition に通す（合流点は一つに保つ）
    await applyTransition(env, {
      jobId: rec.jobId,
      state: remote.state,
      seq: remote.seq,           // 取得 API も同じ seq を返す前提にする
      finishedAt: remote.finishedAt,
    });
 
    // 「起動したのに長時間 unknown/running のまま」を異常として拾う
    const ageMin = (now - Date.parse(rec.createdAt)) / 60000;
    if (rec.state !== "succeeded" && rec.state !== "failed" && ageMin > 30) {
      await alertStuckJob(env, rec.jobId, Math.round(ageMin));
    }
  }
}