Gemini の完了イベントは二度届きます — Webhook と照合ポーラーを「実質1回」にする冪等な受け口

import re
 
def idempotency_key(event: dict) -> str:
    """完了イベントから安定した冪等キーを取り出す。
 
    Webhook ペイロードと、照合ポーラーが list で得る operation の
    両方から、同じ完了なら必ず同じ文字列を返せる形にする。
    """
    # Webhook と REST list でフィールド名がぶれるため候補を順に当たる
    name = (
        event.get("name")
        or event.get("operation")
        or event.get("operationName")
        or event.get("resourceName")
    )
    if not name:
        raise ValueError(f"operation name が見つかりません: {list(event.keys())}")
 
    # 末尾スラッシュ・前後空白・大文字小文字のぶれを吸収する
    name = name.strip().rstrip("/")
    # projects/.../locations/... のプレフィックス差を畳んで末尾の安定部分を使う
    m = re.search(r"(operations|batches|batchJobs)/[A-Za-z0-9_\-]+$", name)
    canonical = m.group(0) if m else name
    return f"gemini:{canonical}"
 
 
# 同じ完了が Webhook 由来でも poller 由来でも同じキーになることを確認
webhook_event = {"name": "projects/p/locations/us/operations/abc-123/"}
poller_event = {"operation": "operations/abc-123"}
assert idempotency_key(webhook_event) == idempotency_key(poller_event)
print(idempotency_key(webhook_event))  # -> gemini:operations/abc-123

import sqlite3
import time
import contextlib
 
DB = "idempotency.db"
CLAIM_TTL_SEC = 600  # 副作用が確実に終わる想定時間より長く取る
 
def init_db():
    con = sqlite3.connect(DB)
    con.execute("""
        CREATE TABLE IF NOT EXISTS sink (
            key        TEXT PRIMARY KEY,   -- 冪等キー（UNIQUE）
            state      TEXT NOT NULL,      -- CLAIMED / DONE
            claimed_at REAL,
            done_at    REAL
        )
    """)
    con.commit()
    con.close()
 
 
@contextlib.contextmanager
def db():
    con = sqlite3.connect(DB, isolation_level=None)  # 明示的トランザクション
    try:
        yield con
    finally:
        con.close()
 
 
def try_claim(key: str) -> bool:
    """このプロセスが副作用を起こす権利を取れたら True。
 
    UNIQUE 制約により、同時に来た2つの呼び出しのうち
    INSERT に成功するのは必ず1つだけになる。
    """
    now = time.time()
    with db() as con:
        con.execute("BEGIN IMMEDIATE")  # 書き込みロックを先に取る
        row = con.execute(
            "SELECT state, claimed_at FROM sink WHERE key = ?", (key,)
        ).fetchone()
 
        if row is None:
            con.execute(
                "INSERT INTO sink(key, state, claimed_at) VALUES (?, 'CLAIMED', ?)",
                (key, now),
            )
            con.execute("COMMIT")
            return True
 
        state, claimed_at = row
        if state == "DONE":
            con.execute("COMMIT")
            return False  # 既に確定済み → 二度処理しない
 
        # CLAIMED のまま放置されている（処理中にプロセスが落ちた等）
        if now - (claimed_at or 0) > CLAIM_TTL_SEC:
            con.execute(
                "UPDATE sink SET claimed_at = ? WHERE key = ?", (now, key)
            )
            con.execute("COMMIT")
            return True  # 期限切れクレームを奪取して再試行
        con.execute("COMMIT")
        return False  # 他プロセスが処理中。今回は何もしない
 
 
def mark_done(key: str):
    with db() as con:
        con.execute(
            "UPDATE sink SET state = 'DONE', done_at = ? WHERE key = ?",
            (time.time(), key),
        )

def process_completion(event: dict, do_side_effect) -> str:
    """完了イベントを受け取り、副作用を実質1回だけ起こす。
 
    do_side_effect(key) は公開・課金・通知などの実処理。
    冪等キー単位で高々1回しか呼ばれないことを保証する。
    """
    key = idempotency_key(event)
 
    if not try_claim(key):
        return "skipped"  # DONE 済み or 他プロセスが処理中
 
    try:
        do_side_effect(key)   # ← ここが二度走ってはいけない処理
    except Exception:
        # 副作用が落ちたら DONE にしない。
        # クレームは TTL 後に奪取され、別の機会に再試行される。
        raise
 
    mark_done(key)
    return "processed"
 
 
# 使用例: Webhook も poller も同じ受け口を呼ぶ
def publish_article(key: str):
    print(f"[publish] {key} の記事を公開しました")
 
# Webhook 由来
print(process_completion({"name": "operations/abc-123"}, publish_article))
# -> [publish] gemini:operations/abc-123 の記事を公開しました
#    processed
 
# 照合ポーラーが同じ完了を後から拾った
print(process_completion({"operation": "operations/abc-123"}, publish_article))
# -> skipped （公開は二度走らない）