Spaces:
Sleeping
Sleeping
| """ | |
| insert_guard.py — Atomic insert guard for all challenge generators. | |
| Guarantees that dedup check + DB insert are serialised per (table, team_role), | |
| eliminating the race condition where two concurrent inserts both pass the | |
| dedup check and create a duplicate. | |
| Usage: | |
| from app.services.insert_guard import atomic_insert | |
| async def refill_pool(...): | |
| ... | |
| ok = await atomic_insert("encryption_challenges", team_role, row, | |
| is_duplicate_func, [title, story]) | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| from collections import defaultdict | |
| from typing import Any, Callable, Optional | |
| _TABLE_LOCKS: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock) | |
| def _lock_key(table: str, team_role: str) -> str: | |
| return f"{table}::{team_role}" | |
| async def atomic_insert( | |
| table: str, | |
| team_role: str, | |
| row: dict, | |
| dedup_func: Optional[Callable[..., bool]] = None, | |
| dedup_args: Optional[list] = None, | |
| dedup_kwargs: Optional[dict] = None, | |
| ) -> bool: | |
| """Insert *row* into *table* after a serialised dedup check. | |
| Parameters | |
| ---------- | |
| table : str | |
| Supabase table name (e.g. ``"encryption_challenges"``). | |
| team_role : str | |
| ``"red"`` or ``"blue"`` — used to scope the lock. | |
| row : dict | |
| The full DB row to insert. | |
| dedup_func : callable, optional | |
| A synchronous function that returns ``True`` if a duplicate exists. | |
| Called **under the lock** so the check and insert are atomic. | |
| dedup_args : list, optional | |
| Positional args for *dedup_func*. | |
| dedup_kwargs : dict, optional | |
| Keyword args for *dedup_func*. | |
| Returns | |
| ------- | |
| bool | |
| ``True`` if the row was inserted, ``False`` if skipped (duplicate) | |
| or on error. | |
| """ | |
| key = _lock_key(table, team_role) | |
| lock = _TABLE_LOCKS[key] | |
| # Re-use the same httpx client across calls within the lock scope | |
| import httpx | |
| from app.services.supabase_service import supabase_headers | |
| from app.core.config import SUPABASE_URL, SUPABASE_ANON_KEY | |
| if not SUPABASE_ANON_KEY or not SUPABASE_URL: | |
| return False | |
| async with lock: | |
| # 1. Dedup check (under lock) | |
| if dedup_func is not None: | |
| try: | |
| is_dup = dedup_func( | |
| *(dedup_args or []), | |
| **(dedup_kwargs or {}), | |
| ) | |
| if is_dup: | |
| title = row.get("title", "?")[:60] | |
| print(f" [insert_guard] Dedup skipped: {title} in {table}") | |
| return False | |
| except Exception as e: | |
| print(f" [insert_guard] Dedup check error (non-fatal): {e}") | |
| # 2. Insert (under lock) | |
| url = f"{SUPABASE_URL}/rest/v1/{table}" | |
| headers = supabase_headers(content_type=True) | |
| headers["Prefer"] = "return=representation" | |
| try: | |
| async with httpx.AsyncClient(timeout=20) as client: | |
| resp = await client.post(url, json=row, headers=headers) | |
| if resp.status_code in (200, 201): | |
| return True | |
| print(f" [insert_guard] DB insert {resp.status_code}: {resp.text[:200]}") | |
| return False | |
| except Exception as e: | |
| print(f" [insert_guard] DB insert error: {e}") | |
| return False | |