dee/auth.py

"""Auth + anonymous-quota middleware for TuringDNA.

Two responsibilities, glued together because the endpoints that need them
overlap completely:

  1. Verify Supabase access tokens on /api/run (and any future protected
     endpoint). The token rides in the `Authorization: Bearer <jwt>`
     header. We validate signature + expiry against the Supabase JWT
     secret, then expose the resulting claims via flask.g.user.

  2. Enforce the one-free-anonymous-run quota. Anonymous users get exactly
     one /api/run before they need an account. We track this server-side
     via a signed cookie (`td_anon_runs`) so a sophisticated user clearing
     localStorage in the iframe wrapper still hits the wall here.

Both are best-effort: missing/invalid auth degrades to anonymous-mode
processing rather than 401, EXCEPT when require_auth_or_quota() rejects
the call because the user has used their free run AND has no valid JWT.

Env vars (set on the HF Space):
    SUPABASE_URL              e.g. https://abcdefg.supabase.co
    SUPABASE_JWT_SECRET       The "JWT Secret" from Supabase Settings → API
    SUPABASE_SERVICE_KEY      Service-role key (for inserting runs + aggregates)
    SUPABASE_ANON_KEY         Public anon/publishable key (the SAME one shipped
                              in the frontend). Used server-side so per-user
                              reads/writes run AS THE USER and Row-Level
                              Security enforces isolation in Postgres. If unset,
                              those calls fall back to the service-role key and
                              RLS is not enforced at the row level (a startup
                              warning fires). Strongly recommended in prod.
    TURINGDNA_ANON_SECRET     Random 32+ byte string for cookie signing
                              (any reasonably-strong value, doesn't need
                              to match Supabase). Generate with:
                                  python -c "import secrets; print(secrets.token_urlsafe(48))"

If env vars are missing, auth is OFF and every request is treated as
anonymous with unlimited runs — keeps local dev easy. The startup log
warns visibly so this isn't accidental in production.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import logging
import os
import threading
import time
from dataclasses import dataclass
from typing import Any, Dict, Optional

from flask import Flask, Response, g, jsonify, make_response, request

logger = logging.getLogger("dee.auth")

# ─── Config ────────────────────────────────────────────────────────────
SUPABASE_URL          = os.environ.get("SUPABASE_URL", "").rstrip("/")
SUPABASE_JWT_SECRET   = os.environ.get("SUPABASE_JWT_SECRET", "")
SUPABASE_SERVICE_KEY  = os.environ.get("SUPABASE_SERVICE_KEY", "")
# Public "anon" / publishable key. Safe to expose — it's already shipped in
# the frontend. We use it SERVER-SIDE too, paired with the requester's own
# JWT, so per-user reads/writes run *as the user* and Row-Level Security
# enforces isolation in the database (see _user_pgrst_headers). This means a
# forgotten `user_id=eq.` filter can no longer leak another user's rows —
# Postgres refuses. If unset, those calls fall back to the service-role key
# (current behaviour) and RLS is NOT enforced at the row level; a startup
# warning fires so this isn't silent.
SUPABASE_ANON_KEY     = os.environ.get("SUPABASE_ANON_KEY", "")
ANON_SECRET           = os.environ.get("TURINGDNA_ANON_SECRET", "")

# A Supabase auth.users.id is always a UUID. We validate the JWT `sub`
# against this before trusting it as a user_id — defence-in-depth so a
# malformed/crafted `sub` can never be interpolated into a PostgREST filter
# URL (the design/library ids are already regex-checked; this closes the
# same gap for user_id at the single chokepoint where it enters the system).
import re as _re_mod
_UUID_RE = _re_mod.compile(
    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
    r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
)


def _is_uuid(value: object) -> bool:
    return isinstance(value, str) and bool(_UUID_RE.match(value))

# Cookie name + lifetime for the anonymous-trial tracker. 365d is fine —
# the cookie records "this browser started a trial at timestamp X" and
# is replaced on sign-in.
ANON_COOKIE_NAME      = "td_anon_runs"
ANON_COOKIE_MAX_AGE_S = 365 * 24 * 3600

# Length of the anonymous trial window, in seconds. The user can run
# however many sequences they want during this window — when it expires,
# /api/run returns 402 and the frontend pops the sign-in modal.
#
# Lean-startup framing: 5 minutes is plenty to feel the magic but not
# enough to do real exploratory work. Catches users while they're still
# engaged. Was previously a count-based "1 free run" hard wall, swapped
# for time-based per user request 2026-05-26.
ANON_TRIAL_SECONDS    = int(os.environ.get("TURINGDNA_ANON_TRIAL_SECONDS", "300"))

# How long a generated library stays in Supabase Storage for free-tier
# users. Matches PRIVACY.md §7 ("Sequences: deleted ... 90 days after
# your last access"). Pro-tier users get no expiry (NULL in
# public.libraries.expires_at). Cleanup runs lazily on every new
# /api/run via cleanup_expired_libraries() — no cron required.
FREE_LIBRARY_TTL_DAYS = int(os.environ.get("TURINGDNA_FREE_LIBRARY_TTL_DAYS", "45"))

# Whether auth is configured. If False, every request is anonymous and
# unlimited — we log a warning at startup so production doesn't silently
# ship without auth.
AUTH_ENABLED = bool(SUPABASE_JWT_SECRET and ANON_SECRET)


# ─── User context exposed to handlers ─────────────────────────────────
@dataclass
class AuthContext:
    """What we know about the requester. Exactly one of (user_id, anon_id)
    is populated. anonymous=True is the catch-all for "no JWT, treat as
    anon" — and anon_runs counts how many free runs this anon has used."""
    user_id:        Optional[str] = None  # Supabase auth.users.id
    email:          Optional[str] = None
    anonymous:      bool = True
    # The raw, already-verified Supabase access token. Stored so per-user
    # PostgREST calls can be made AS THE USER (anon apikey + this Bearer
    # token) and have RLS enforce row isolation. Never logged, never sent
    # anywhere except back to Supabase's own REST endpoint.
    token:          Optional[str] = None
    # Unix timestamp when this anon visitor's trial started (i.e., when
    # they kicked off their first /api/run). 0 means "trial hasn't begun".
    anon_first_at:  int = 0
    # Count of anon runs in the trial, informational only — gate is
    # time-based, not count-based.
    anon_runs:      int = 0


def get_auth() -> AuthContext:
    """Convenience: returns the AuthContext attached to the current Flask
    request by load_auth_into_g(). Returns a fresh anonymous one if no
    middleware ran yet (defensive)."""
    return getattr(g, "auth", AuthContext())


# ─── JWT verification (no external deps) ──────────────────────────────
# We use PyJWT when available (clean signature/expiry handling), and fall
# back to a tiny in-house verifier so the auth module works even if PyJWT
# isn't installed yet. Supabase signs with HS256 (HMAC-SHA256) using the
# JWT Secret — straightforward to verify by hand.

def _b64url_decode(data: str) -> bytes:
    # JWT base64url: '-' '_' instead of '+' '/', no padding.
    pad = '=' * (-len(data) % 4)
    return _b64_translate(data + pad)


def _b64_translate(data: str) -> bytes:
    import base64
    return base64.urlsafe_b64decode(data.encode("ascii"))


def _verify_jwt_hs256(token: str, secret: str) -> Optional[Dict[str, Any]]:
    """Verify an HS256-signed JWT and return its claims, or None on failure.

    LEGACY path. Supabase used to sign everything with HS256 against the
    Legacy JWT Secret. After May 2026 Supabase migrated this project to
    asymmetric ES256 signing — those tokens are verified by the JWKS
    path below (_verify_jwt_jwks). We keep HS256 alive because:
      - The Supabase anon + service_role API keys are still HS256-signed
        against the Legacy Secret (per Supabase's own UI copy).
      - Projects that haven't migrated yet, or that revert, still need it.

    Validates: signature, exp (expiry), nbf (not-before if present), and
    iat (issued-at, sanity-checks not-in-future-by-more-than-60s). Does
    not strictly verify `aud` or `iss` — Supabase JWTs have iss=supabase
    and aud=authenticated, which we treat as informational only.
    """
    try:
        header_b64, payload_b64, sig_b64 = token.split(".")
    except ValueError:
        return None

    try:
        # Verify signature first — constant-time compare.
        signing_input = (header_b64 + "." + payload_b64).encode("ascii")
        expected = hmac.new(secret.encode("utf-8"), signing_input, hashlib.sha256).digest()
        actual = _b64url_decode(sig_b64)
        if not hmac.compare_digest(expected, actual):
            return None

        # Parse claims.
        claims = json.loads(_b64url_decode(payload_b64).decode("utf-8"))
        now = int(time.time())

        exp = claims.get("exp")
        if isinstance(exp, (int, float)) and now > int(exp):
            return None

        nbf = claims.get("nbf")
        if isinstance(nbf, (int, float)) and now < int(nbf):
            return None

        iat = claims.get("iat")
        # Allow some clock skew on iat — 5 minutes is generous.
        if isinstance(iat, (int, float)) and int(iat) > now + 300:
            return None

        return claims
    except Exception:
        return None


# ─── ES256 / RS256 verification via JWKS ───────────────────────────────
# 2026-05-28: Supabase migrated this project's access-token signing from
# HS256 (Legacy Secret) to ES256 (ECDSA P-256, asymmetric keys). The
# private key stays with Supabase; we fetch the matching public key from
# their JWKS endpoint and verify against it.
#
# Endpoint shape:
#   https://<project>.supabase.co/auth/v1/.well-known/jwks.json
#
# PyJWKClient handles HTTP fetch + caching + kid-based key lookup. We
# cache one client instance at module level — it caches keys in memory
# and refetches the JWKS only on kid miss (e.g., after Supabase rotates).
#
# Verification uses PyJWT with the cryptography backend (the [crypto]
# extra in requirements.txt). If PyJWT isn't installed, we degrade to
# returning None for asymmetric tokens — the legacy HS256 path still
# works for any users on projects that haven't migrated.

_jwks_client = None  # lazy-initialized PyJWKClient
_pyjwt_module = None  # lazy-imported jwt module


def _ensure_pyjwt():
    """Lazy-import PyJWT so the auth module loads even if PyJWT isn't
    installed (graceful local-dev fallback)."""
    global _pyjwt_module
    if _pyjwt_module is None:
        try:
            import jwt as _jwt  # PyJWT
            _pyjwt_module = _jwt
        except ImportError:
            logger.warning(
                "PyJWT not installed; ES256/RS256 Supabase tokens cannot be "
                "verified. Install with: pip install 'PyJWT[crypto]'"
            )
            _pyjwt_module = False  # sentinel: don't retry import
    return _pyjwt_module or None


def _get_jwks_client():
    """Lazy-init a PyJWKClient pointing at this project's JWKS endpoint.
    Returns None if Supabase URL isn't configured or PyJWT is missing."""
    global _jwks_client
    if _jwks_client is not None:
        return _jwks_client
    if not SUPABASE_URL:
        return None
    pyjwt = _ensure_pyjwt()
    if pyjwt is None:
        return None
    try:
        _jwks_client = pyjwt.PyJWKClient(
            f"{SUPABASE_URL}/auth/v1/.well-known/jwks.json",
            cache_keys=True,
            max_cached_keys=8,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to initialize JWKS client: %s", exc)
        _jwks_client = None
    return _jwks_client


def _verify_jwt_jwks(token: str, alg: str) -> Optional[Dict[str, Any]]:
    """Verify an ES256 / RS256 / ES384 / RS512 etc. JWT against the
    Supabase JWKS public keys. Returns claims or None.

    Same validation surface as _verify_jwt_hs256: signature, exp, nbf,
    iat (with 5-min clock-skew tolerance). audience NOT enforced
    (Supabase uses aud="authenticated" which is informational here)."""
    pyjwt = _ensure_pyjwt()
    if pyjwt is None:
        return None
    client = _get_jwks_client()
    if client is None:
        return None
    try:
        signing_key = client.get_signing_key_from_jwt(token)
        claims = pyjwt.decode(
            token,
            signing_key.key,
            algorithms=[alg],
            options={
                # iss/aud verification disabled — matches the HS256 path.
                # Supabase tokens carry aud="authenticated" which isn't
                # something we need to enforce at the API layer; the user_id
                # in `sub` is what we attribute work to.
                "verify_aud": False,
                "verify_iss": False,
            },
            # 5-min leeway covers minor clock skew between the Supabase
            # auth servers and the HF Space container.
            leeway=300,
        )
        return claims
    except pyjwt.exceptions.InvalidTokenError as exc:
        logger.info("JWKS verify rejected token: %s", exc)
        return None
    except Exception as exc:  # noqa: BLE001
        # Network errors fetching JWKS, malformed kid, etc.
        logger.warning("JWKS verify error: %s", exc)
        return None


def _verify_jwt_any(token: str) -> Optional[Dict[str, Any]]:
    """Dispatch JWT verification based on the algorithm in the header.

    Branches:
      - HS256 → legacy HMAC path using SUPABASE_JWT_SECRET
      - ES256/RS256/etc. → JWKS-based asymmetric verification

    Returns claims on success, None on failure. Always None if AUTH_ENABLED
    is False (caller already checks this, defensive double-guard)."""
    if not token:
        return None
    # Decode header WITHOUT verification to learn the algorithm. We do
    # this with PyJWT if available (cleaner) and fall back to manual
    # base64 + JSON parse otherwise.
    alg = ""
    pyjwt = _ensure_pyjwt()
    if pyjwt is not None:
        try:
            alg = pyjwt.get_unverified_header(token).get("alg", "")
        except Exception:
            alg = ""
    if not alg:
        try:
            header_b64 = token.split(".", 2)[0]
            alg = json.loads(_b64url_decode(header_b64).decode("utf-8")).get("alg", "")
        except Exception:
            return None

    if alg == "HS256":
        if not SUPABASE_JWT_SECRET:
            return None
        return _verify_jwt_hs256(token, SUPABASE_JWT_SECRET)

    if alg in ("ES256", "ES384", "ES512", "RS256", "RS384", "RS512"):
        return _verify_jwt_jwks(token, alg)

    # Unknown algorithm — refuse.
    logger.info("JWT uses unsupported alg=%r; rejecting.", alg)
    return None


# ─── Signed anon cookie ───────────────────────────────────────────────
# Format: "<first_at>.<runs>.<issued_at>.<hex_hmac>"
#
#   first_at  = unix seconds when this visitor's trial began (0 if first
#               request hasn't happened yet — gate not yet armed)
#   runs      = informational run count (gate is time-based, this is for
#               analytics convenience)
#   issued_at = cookie issuance time (for max-age sanity check)
#
# We sign with ANON_SECRET so the client can't reset first_at to 0 by
# editing the cookie. They CAN clear the cookie entirely — at which
# point they get a fresh trial — so cookie-clearing remains a free path
# around the wall. Tightening further (server-side IP fingerprint) is
# a deliberate later choice, not done here.

def _sign_anon_value(first_at: int, runs: int, issued: int) -> str:
    body = f"{first_at}.{runs}.{issued}"
    mac = hmac.new(
        ANON_SECRET.encode("utf-8"),
        body.encode("ascii"),
        hashlib.sha256,
    ).hexdigest()
    return f"{body}.{mac}"


def _parse_anon_cookie(raw: Optional[str]) -> tuple[int, int]:
    """Returns (first_at, runs). Both 0 if cookie missing/forged. Also
    rejects cookies issued more than ANON_COOKIE_MAX_AGE_S ago."""
    if not raw or not ANON_SECRET:
        return (0, 0)
    try:
        first_s, runs_s, issued_s, mac = raw.split(".")
        first_at = int(first_s)
        runs = int(runs_s)
        issued = int(issued_s)
    except (ValueError, AttributeError):
        return (0, 0)
    if runs < 0 or runs > 1_000_000:
        return (0, 0)
    if first_at < 0 or first_at > 4_000_000_000:
        return (0, 0)
    if int(time.time()) - issued > ANON_COOKIE_MAX_AGE_S:
        return (0, 0)
    expected = _sign_anon_value(first_at, runs, issued)
    if not hmac.compare_digest(expected, raw):
        return (0, 0)
    return (first_at, runs)


def _set_anon_cookie(response: Response, first_at: int, runs: int) -> None:
    if not ANON_SECRET:
        return
    issued = int(time.time())
    response.set_cookie(
        ANON_COOKIE_NAME,
        _sign_anon_value(first_at, runs, issued),
        max_age=ANON_COOKIE_MAX_AGE_S,
        httponly=True,
        secure=True,
        # Cross-site (iframe ≠ landing origin) requires SameSite=None +
        # Secure. Lax used to work because /app was same-origin in dev,
        # but in production the engine lives at *.hf.space inside an
        # iframe at turingdna.com, so the cookie has to cross sites.
        samesite="None",
    )


# ─── Flask integration ─────────────────────────────────────────────────
def load_auth_into_g() -> None:
    """before_request hook. Parses Authorization header + anon cookie and
    populates g.auth. Always populates something — handlers should read
    g.auth and decide what to do; this hook never returns a response."""
    auth = AuthContext()

    # 1) Try Authorization: Bearer <jwt>
    # _verify_jwt_any dispatches on the JWT's `alg`:
    #   HS256 → legacy SUPABASE_JWT_SECRET path
    #   ES256/RS256 → JWKS-based asymmetric verification
    # Either path returns the validated claims or None.
    auth_header = request.headers.get("Authorization", "")
    if AUTH_ENABLED and auth_header.lower().startswith("bearer "):
        token = auth_header[7:].strip()
        claims = _verify_jwt_any(token)
        if claims:
            sub = claims.get("sub")
            role = claims.get("role")
            # The signature is valid, but defence-in-depth on two fronts:
            #   (1) a Supabase user id is ALWAYS a uuid — if `sub` isn't one,
            #       refuse to treat it as a user_id (it would otherwise be
            #       interpolated into PostgREST filter URLs);
            #   (2) only a real user access token carries role="authenticated".
            #       Other token types Supabase signs with the same secret
            #       (anon/service_role API keys) have a different role and no
            #       uuid sub — requiring both closes any gap where such a token
            #       could be presented as a Bearer session.
            if _is_uuid(sub) and role == "authenticated":
                auth.user_id   = sub
                auth.email     = claims.get("email")
                auth.anonymous = False
                auth.token     = token
            else:
                logger.warning("JWT verified but sub/role not a user session (sub uuid=%s, role=%r); treating as anon.",
                               _is_uuid(sub), role)
        else:
            logger.info("Invalid or expired JWT received; treating as anon.")

    # 2) If still anonymous, read the signed trial cookie.
    if auth.anonymous:
        first_at, runs = _parse_anon_cookie(request.cookies.get(ANON_COOKIE_NAME))
        auth.anon_first_at = first_at
        auth.anon_runs = runs

    g.auth = auth

    # Engagement: mark signed-in users active (throttled to 1/hour/user,
    # fire-and-forget). Powers profiles.last_seen_at — previously never written.
    if not auth.anonymous and auth.user_id:
        try:
            touch_last_seen_async(auth.user_id)
        except Exception:  # noqa: BLE001 — telemetry must never block a request
            pass


def require_auth_or_quota() -> Optional[Response]:
    """Call from a route handler BEFORE doing any work. Returns a 402 JSON
    response (intercept) if the user is anonymous AND their trial window
    has expired. Returns None to let the handler proceed.

    The gate is **time-based**: from the moment of the first anon /api/run,
    the user has ANON_TRIAL_SECONDS of unlimited access. After that,
    further /api/run calls 402 until they sign in.

    402 instead of 401 because the user IS welcome to use this — they
    just need to sign up. The frontend interprets 402 specifically as
    "show the sign-up gate" rather than a hard auth error."""
    if not AUTH_ENABLED:
        return None  # local dev — never block

    auth = get_auth()
    if not auth.anonymous:
        return None  # signed in, proceed

    if not auth.anon_first_at:
        return None  # trial hasn't started yet — first run is always free

    elapsed = int(time.time()) - auth.anon_first_at
    if elapsed >= ANON_TRIAL_SECONDS:
        return jsonify({
            "error": (
                f"Your {ANON_TRIAL_SECONDS // 60}-minute free trial has "
                "ended. Sign in to keep going — no password, just an "
                "emailed link."
            ),
            "kind": "auth_required",
            # Default redirect target points to /signin since most repeat
            # visitors already have an account. /signup is a 1-click away
            # from /signin via the "Need an account?" link.
            "signup_url": "https://turingdna.com/signin/?from=app",
            "trial_seconds": ANON_TRIAL_SECONDS,
            "trial_started_at": auth.anon_first_at,
        }), 402

    return None


def increment_anon_runs_on_response(response: Response) -> Response:
    """Called from inside a /api/run handler that has decided to start a
    pipeline FOR an anonymous user. Stamps the first_at timestamp on the
    cookie if this is their first run; otherwise just bumps the
    informational run count without resetting the trial timer.

    Idempotent against being called twice on the same request — the
    response cookie just gets overwritten with the same data.
    """
    auth = get_auth()
    if not (auth.anonymous and AUTH_ENABLED):
        return response

    now = int(time.time())
    first_at = auth.anon_first_at or now
    runs = auth.anon_runs + 1
    _set_anon_cookie(response, first_at, runs)

    # Also surface the trial state in a custom response header so the
    # frontend can read it without parsing the cookie itself. The frontend
    # uses this to start its own client-side countdown to the in-session
    # "sign in to keep going" modal.
    elapsed = now - first_at
    remaining = max(0, ANON_TRIAL_SECONDS - elapsed)
    response.headers["X-TD-Trial-Remaining-S"] = str(remaining)
    response.headers["X-TD-Trial-Started-At"]  = str(first_at)
    return response


# ─── Run logging (best-effort) ────────────────────────────────────────
# After a /api/run starts we POST a row into Supabase public.runs via the
# REST API using the service-role key. Never blocks the user's request —
# fire-and-forget in a daemon thread, log on error.

def log_run_async(
    *,
    job_id: str,
    sequence_hash: str,
    sequence_length: int,
    model: Optional[str],
    host_organism: Optional[str],
) -> None:
    """Fire-and-forget insert into public.runs. Captures user_id from
    g.auth if signed in; otherwise stores anon_fingerprint = IP+UA hash."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return  # not configured; skip silently

    auth = get_auth()
    user_id          = auth.user_id
    anon_fingerprint = None
    if auth.anonymous:
        anon_fingerprint = _anon_fingerprint()

    # Data-minimization: persist only a /24-anonymized IP (host bits zeroed)
    # and a short UA. We never store a full client IP — it shrinks the blast
    # radius of any DB compromise and matches PRIVACY.md's minimization stance.
    raw_ip = request.headers.get("X-Forwarded-For", request.remote_addr or "").split(",")[0].strip()
    request_ip = _anonymize_ip(raw_ip)
    request_ua = request.headers.get("User-Agent", "")[:200]

    payload = {
        "user_id":          user_id,
        "anon_fingerprint": anon_fingerprint,
        "job_id":           job_id,
        "sequence_hash":    sequence_hash,
        "sequence_length":  sequence_length,
        "model":            model,
        "host_organism":    host_organism,
        "status":           "started",
        "request_ip":       request_ip or None,
        "request_ua":       request_ua or None,
    }

    import threading
    threading.Thread(
        target=_post_run_row, args=(payload,), daemon=True,
    ).start()


def _post_run_row(payload: Dict[str, Any]) -> None:
    try:
        import urllib.request
        import urllib.error
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/runs",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers={
                "Content-Type": "application/json",
                "apikey": SUPABASE_SERVICE_KEY,
                "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
                "Prefer": "return=minimal",
            },
        )
        with urllib.request.urlopen(req, timeout=4.0) as resp:
            if resp.status not in (200, 201, 204):
                logger.warning("Supabase runs insert returned %s", resp.status)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Could not log run to Supabase: %s", exc)


# ═══════════════════════════════════════════════════════════════════════
# ENGAGEMENT TELEMETRY (activation / retention / dwell)
# ───────────────────────────────────────────────────────────────────────
# All writers below are fire-and-forget and best-effort: they spawn a daemon
# thread, never raise into a request handler, and no-op when Supabase isn't
# configured. Privacy stance matches public.runs — no sequence content, only a
# /24-anonymized IP fingerprint for anonymous visitors. See 0011_analytics.sql.
# ═══════════════════════════════════════════════════════════════════════

def _iso_now() -> str:
    """Current UTC time as an ISO-8601 string PostgREST stores into timestamptz."""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()


def _post_supabase(table: str, payload: Dict[str, Any]) -> None:
    """Fire-and-forget POST to PostgREST (table insert or rpc/<fn>) with the
    service key. Best-effort; logs and swallows any failure."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    try:
        import urllib.request
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/{table}",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers=_supabase_headers({
                "Content-Type": "application/json",
                "Prefer": "return=minimal",
            }),
        )
        with urllib.request.urlopen(req, timeout=4.0) as resp:
            if resp.status not in (200, 201, 204):
                logger.warning("Supabase POST %s returned %s", table, resp.status)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Supabase POST %s failed: %s", table, exc)


def _patch_supabase(path_query: str, payload: Dict[str, Any]) -> None:
    """Fire-and-forget PATCH to PostgREST with the service key. Best-effort."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    try:
        import urllib.request
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/{path_query}",
            data=json.dumps(payload).encode("utf-8"),
            method="PATCH",
            headers=_supabase_headers({
                "Content-Type": "application/json",
                "Prefer": "return=minimal",
            }),
        )
        with urllib.request.urlopen(req, timeout=4.0) as resp:
            if resp.status not in (200, 204):
                logger.warning("Supabase PATCH %s returned %s", path_query, resp.status)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Supabase PATCH %s failed: %s", path_query, exc)


# Throttle: at most one profiles.last_seen_at UPDATE per user per hour, so a
# burst of API calls doesn't hammer PostgREST. In-memory; resets on process
# restart (worst case: one extra touch per user after a redeploy).
_LAST_SEEN_TOUCHED: Dict[str, float] = {}
_LAST_SEEN_LOCK = threading.Lock()
_LAST_SEEN_THROTTLE_S = 3600.0


def touch_last_seen_async(user_id: Optional[str]) -> None:
    """Mark a signed-in user active now (profiles.last_seen_at = now), throttled
    to once/hour/user. Called from the auth chokepoint on every authed request."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    now = time.time()
    with _LAST_SEEN_LOCK:
        if now - _LAST_SEEN_TOUCHED.get(user_id, 0.0) < _LAST_SEEN_THROTTLE_S:
            return
        _LAST_SEEN_TOUCHED[user_id] = now
        # Bound the cache so a churn of distinct users can't grow it forever.
        if len(_LAST_SEEN_TOUCHED) > 50000:
            cutoff = now - _LAST_SEEN_THROTTLE_S
            for k in [k for k, v in _LAST_SEEN_TOUCHED.items() if v < cutoff]:
                _LAST_SEEN_TOUCHED.pop(k, None)
    threading.Thread(
        target=_patch_supabase,
        args=(f"profiles?id=eq.{user_id}", {"last_seen_at": _iso_now()}),
        daemon=True,
    ).start()


def mark_run_finished_async(
    *, job_id: str, status: str,
    elapsed_seconds: Optional[float] = None,
    n_variants: Optional[int] = None,
) -> None:
    """PATCH the public.runs row for job_id with its terminal state — the row
    inserted by log_run_async only ever had status='started'. Best-effort."""
    if not job_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    payload: Dict[str, Any] = {
        "status": status if status in ("done", "error") else "done",
        "finished_at": _iso_now(),
    }
    if elapsed_seconds is not None:
        payload["elapsed_seconds"] = round(float(elapsed_seconds), 3)
    if n_variants is not None:
        payload["n_variants"] = int(n_variants)
    threading.Thread(
        target=_patch_supabase,
        args=(f"runs?job_id=eq.{job_id}", payload),
        daemon=True,
    ).start()


def log_event(kind: str, *, meta: Optional[Dict[str, Any]] = None) -> None:
    """Fire-and-forget insert into public.events — one row per meaningful tool
    action (plasmid map, CRISPR design, save, …) so engagement covers all tools,
    not just /api/run. Attributes to user_id, or an anon fingerprint. Must be
    called from a request context (reads g.auth + request headers)."""
    if not kind or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    try:
        auth = get_auth()
    except Exception:  # noqa: BLE001 — no request context
        return
    payload = {
        "user_id":          auth.user_id,
        "anon_fingerprint": _anon_fingerprint() if auth.anonymous else None,
        "kind":             kind[:64],
        "meta":             meta or {},
    }
    threading.Thread(target=_post_supabase, args=("events", payload), daemon=True).start()


def record_ping_async(*, session_id: str, visible_ms: int) -> None:
    """Upsert a dwell-session row via the record_ping() RPC (atomic insert-or-
    update keyed by session_id). Attributes to user_id or anon fingerprint.
    Best-effort; call from a request context."""
    if not session_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    try:
        auth = get_auth()
    except Exception:  # noqa: BLE001
        return
    body = {
        "p_session_id": session_id[:64],
        "p_user_id":    auth.user_id,
        "p_anon":       _anon_fingerprint() if auth.anonymous else None,
        "p_visible_ms": max(0, int(visible_ms)),
    }
    threading.Thread(target=_post_supabase, args=("rpc/record_ping", body), daemon=True).start()


# ─── Library save (signed-in users only) ─────────────────────────────
# After /api/run finishes, the worker thread calls save_library_async()
# with the per-job CSV path and a few metadata fields. We upload the CSV
# to the `libraries` Supabase Storage bucket and insert a metadata row
# into public.libraries pointing at it. Anonymous users are skipped —
# they don't have a user_id to attribute the library to.

def save_library_async(
    *,
    user_id: Optional[str],
    job_id: str,
    csv_path: str,
    name: str,
    wt_identifier: Optional[str],
    wt_protein: Optional[str],
    host_organism: Optional[str],
    n_variants: int,
    top_fitness: Optional[float],
    library_id: Optional[str] = None,
) -> None:
    """Fire-and-forget save of a generated library to Supabase Storage.

    No-op for anonymous users — they have no user_id to bind to. Also
    no-op if Supabase isn't configured. Spawns a daemon thread so the
    /api/run response isn't held up by the upload. ``library_id`` lets the
    caller pre-assign the id (so it can be returned to the client for the
    learning-flywheel outcome logging); a fresh hex id is minted if omitted.
    """
    if not user_id:
        return  # anonymous run; nothing to attribute
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return

    import threading
    threading.Thread(
        target=_do_save_library,
        kwargs={
            "user_id":       user_id,
            "job_id":        job_id,
            "csv_path":      csv_path,
            "name":          name,
            "wt_identifier": wt_identifier,
            "wt_protein":    wt_protein,
            "host_organism": host_organism,
            "n_variants":    n_variants,
            "top_fitness":   top_fitness,
            "library_id":    library_id,
        },
        daemon=True,
    ).start()


def _do_save_library(
    *,
    user_id: str,
    job_id: str,
    csv_path: str,
    name: str,
    wt_identifier: Optional[str],
    wt_protein: Optional[str],
    host_organism: Optional[str],
    n_variants: int,
    top_fitness: Optional[float],
    library_id: Optional[str] = None,
) -> None:
    """Worker for save_library_async. Two-step: upload CSV to Storage,
    then insert metadata row. Both steps log on failure but don't raise
    — analytics plumbing must never break user-facing runs."""
    import os as _os
    import uuid as _uuid
    import urllib.request
    import urllib.error

    if not _os.path.exists(csv_path):
        logger.warning("save_library: CSV not found at %s", csv_path)
        return

    try:
        with open(csv_path, "rb") as f:
            csv_bytes = f.read()
    except OSError as exc:
        logger.warning("save_library: couldn't read %s: %s", csv_path, exc)
        return

    csv_size = len(csv_bytes)
    library_id = library_id or _uuid.uuid4().hex
    # Path convention: <user_id>/<library_id>.csv
    # The user_id prefix lets the storage RLS policy do a simple
    # first-folder-segment check to gate access.
    storage_path = f"{user_id}/{library_id}.csv"

    # ── TTL computation per user plan ──
    # Free tier: 90-day retention (matches PRIVACY.md §7). Pro tier: no
    # expiry (storage cap is the only limit). The cleanup that actually
    # deletes expired rows runs in cleanup_expired_libraries() — fired
    # off as part of every new /api/run so users don't get charged for
    # data they're not using.
    expires_at_iso = None
    if not has_pro_plan(user_id):
        import datetime as _dt
        expiry = _dt.datetime.utcnow() + _dt.timedelta(days=FREE_LIBRARY_TTL_DAYS)
        # ISO 8601 with explicit Z for UTC — PostgREST/Postgres parse it
        # into timestamptz cleanly.
        expires_at_iso = expiry.strftime("%Y-%m-%dT%H:%M:%SZ")

    # ── 1. Upload CSV to libraries bucket ──
    # Supabase Storage REST endpoint:
    #   POST /storage/v1/object/<bucket>/<path>
    storage_url = f"{SUPABASE_URL}/storage/v1/object/libraries/{storage_path}"
    try:
        req = urllib.request.Request(
            storage_url,
            data=csv_bytes,
            method="POST",
            headers={
                "Content-Type":  "text/csv",
                "Cache-Control": "max-age=3600",
                "apikey":        SUPABASE_SERVICE_KEY,
                "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
                # 'x-upsert' allows re-running the same job_id to
                # overwrite without erroring.
                "x-upsert":      "true",
            },
        )
        with urllib.request.urlopen(req, timeout=15.0) as resp:
            if resp.status not in (200, 201):
                logger.warning("Storage upload returned %s for %s",
                               resp.status, storage_path)
                return
    except Exception as exc:  # noqa: BLE001
        logger.warning("Storage upload failed for %s: %s", storage_path, exc)
        return

    # ── 2. Insert metadata row into public.libraries ──
    payload = {
        "id":             library_id,
        "user_id":        user_id,
        "job_id":         job_id,
        "name":           name,
        "wt_identifier":  wt_identifier,
        "wt_protein":     wt_protein,
        "host_organism":  host_organism,
        "n_variants":     n_variants,
        "top_fitness":    top_fitness,
        "storage_path":   storage_path,
        "csv_size_bytes": csv_size,
        # NULL for Pro = never expires; ISO timestamp for Free = 90 days
        # from now per FREE_LIBRARY_TTL_DAYS.
        "expires_at":     expires_at_iso,
    }
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/libraries",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers={
                "Content-Type":  "application/json",
                "apikey":        SUPABASE_SERVICE_KEY,
                "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
                "Prefer":        "return=minimal",
            },
        )
        with urllib.request.urlopen(req, timeout=4.0) as resp:
            if resp.status not in (200, 201, 204):
                logger.warning("Libraries insert returned %s", resp.status)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Libraries insert failed: %s", exc)


# ─── Library expiry cleanup (lazy, runs on new /api/run) ─────────────
# When a user kicks off a new run, we opportunistically delete any of
# their OWN libraries that are past expires_at. Two-step: remove the
# CSV blob from Storage, then drop the metadata row. Best-effort:
# failures are logged but never block the user-facing run.
#
# Why lazy and not cron:
#   - Active users get cleanup essentially for free (piggybacked on
#     /api/run, which they're already waiting on).
#   - Dormant users' libraries linger forever in the DB, but they cost
#     nothing if no one's hitting the system. A small price for not
#     needing pg_cron (Pro tier feature) or an Edge Function.
#   - Founder can run a manual sweep any time:
#       delete from public.libraries where expires_at < now() returning id;

def cleanup_expired_libraries_async(user_id: Optional[str]) -> None:
    """Fire-and-forget delete of this user's expired libraries.
    No-op for anonymous users (no saved libraries to clean)."""
    if not user_id:
        return
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    import threading
    threading.Thread(
        target=_do_cleanup_expired_libraries,
        args=(user_id,),
        daemon=True,
    ).start()


def _do_cleanup_expired_libraries(user_id: str) -> None:
    """Worker for cleanup_expired_libraries_async. Best-effort.

    1. SELECT id, storage_path FROM libraries WHERE user_id = ? AND expires_at < now()
    2. DELETE the storage objects (batched in one HTTP call)
    3. DELETE the table rows
    """
    import urllib.request
    import urllib.error

    # ── 1. Find expired rows ──
    try:
        url = (
            f"{SUPABASE_URL}/rest/v1/libraries"
            f"?user_id=eq.{user_id}"
            f"&expires_at=lt.now()"
            f"&select=id,storage_path"
        )
        req = urllib.request.Request(
            url,
            method="GET",
            headers={
                "apikey":        SUPABASE_SERVICE_KEY,
                "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
            },
        )
        with urllib.request.urlopen(req, timeout=4.0) as resp:
            expired = json.loads(resp.read().decode("utf-8"))
    except Exception as exc:  # noqa: BLE001
        logger.warning("Expired-library lookup failed for %s: %s", user_id, exc)
        return

    if not expired:
        return

    # ── 2. Remove storage objects in one batch ──
    # Supabase Storage REST: DELETE /storage/v1/object/<bucket> with
    # JSON body {"prefixes": [...paths...]}
    paths = [r["storage_path"] for r in expired if r.get("storage_path")]
    if paths:
        try:
            req = urllib.request.Request(
                f"{SUPABASE_URL}/storage/v1/object/libraries",
                data=json.dumps({"prefixes": paths}).encode("utf-8"),
                method="DELETE",
                headers={
                    "Content-Type":  "application/json",
                    "apikey":        SUPABASE_SERVICE_KEY,
                    "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
                },
            )
            with urllib.request.urlopen(req, timeout=6.0) as resp:
                if resp.status not in (200, 204):
                    logger.warning(
                        "Storage cleanup returned %s for %s",
                        resp.status, paths,
                    )
        except Exception as exc:  # noqa: BLE001
            logger.warning("Storage cleanup failed for %s: %s", user_id, exc)
            # Continue — we'll still remove the DB rows so we don't
            # leave dangling pointers. Re-running this later will retry
            # the storage delete (no row → no future attempt; manual
            # sweep would be needed to truly forget those bytes).

    # ── 3. Drop the metadata rows ──
    ids = [r["id"] for r in expired]
    try:
        # PostgREST 'in.(uuid1,uuid2,...)' filter
        ids_filter = ",".join(ids)
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/libraries?id=in.({ids_filter})",
            method="DELETE",
            headers={
                "apikey":        SUPABASE_SERVICE_KEY,
                "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
                "Prefer":        "return=minimal",
            },
        )
        with urllib.request.urlopen(req, timeout=4.0) as resp:
            if resp.status not in (200, 204):
                logger.warning("Libraries delete returned %s", resp.status)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Libraries delete failed for %s: %s", user_id, exc)

    logger.info("Cleaned %d expired libraries for user %s", len(expired), user_id)


# ─── Plan helpers ────────────────────────────────────────────────────
# Pro-tier features call has_pro_plan() to gate themselves. Today the
# plan is read from profiles.plan via the service-role REST API on each
# call — cheap (single Postgres index lookup) and works without any
# client-side claim. A future optimization is to bake the plan into the
# Supabase JWT via a custom hook so we don't have to round-trip.

def has_pro_plan(user_id: Optional[str]) -> bool:
    """True if this user is on the 'pro' plan, False otherwise (including
    anonymous users). Never raises; on Supabase failure, returns False so
    paid features fail closed."""
    if not user_id:
        return False
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return False
    try:
        import urllib.request
        url = (
            f"{SUPABASE_URL}/rest/v1/profiles"
            f"?id=eq.{user_id}&select=plan&limit=1"
        )
        req = urllib.request.Request(
            url,
            method="GET",
            headers={
                "apikey":        SUPABASE_SERVICE_KEY,
                "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
                "Accept":        "application/json",
            },
        )
        with urllib.request.urlopen(req, timeout=2.0) as resp:
            if resp.status != 200:
                return False
            data = json.loads(resp.read().decode("utf-8"))
            if isinstance(data, list) and data:
                return (data[0] or {}).get("plan") == "pro"
            return False
    except Exception as exc:  # noqa: BLE001
        logger.warning("plan lookup failed for %s: %s", user_id, exc)
        return False


# ─── CRISPR design save / list / get (Phase 3, M4) ───────────────────
# Per-user saved guide sets in public.crispr_designs (migration 0004).
# Synchronous (the user waits on the Save modal); service-role REST with
# server-side user_id ownership filtering. Free users get a 90-day TTL;
# pro users keep designs forever (expires_at = NULL). Designs are small
# JSON, so — unlike libraries — there's no Storage upload, just a row.

_CRISPR_DESIGNS_MAX = 200          # max guides stored per saved design
_CRISPR_DESIGNS_LIST_LIMIT = 100   # max designs returned by the list call


def _supabase_headers(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
    h = {
        "apikey":        SUPABASE_SERVICE_KEY,
        "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
        "Accept":        "application/json",
    }
    if extra:
        h.update(extra)
    return h


def _user_pgrst_headers(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
    """Headers for a per-user PostgREST call made AS THE USER.

    Uses the public anon apikey + the requester's verified JWT as the Bearer
    token. PostgREST then evaluates Row-Level Security with ``auth.uid()`` set
    to the user's id, so the database itself guarantees a request can only ever
    touch that user's own rows — even if a handler forgets its ``user_id=eq.``
    filter. This is the structural backstop that turns RLS from dead weight
    (bypassed by the service key) into a real boundary.

    Graceful fallback: if the anon key isn't configured, or we're outside a
    request context, or the requester is anonymous / has no token, this returns
    the service-role headers (current behaviour). That keeps the per-user
    ``user_id=eq.`` filters — which we always also include — as the second line
    of defence, and makes rollout reversible (unset SUPABASE_ANON_KEY to
    revert). Only ever call this from a synchronous request-context handler;
    background threads have no ``g`` and must use ``_supabase_headers``.
    """
    h: Dict[str, str]
    try:
        auth = get_auth()
    except Exception:  # noqa: BLE001 — defensive: no request context
        auth = AuthContext()
    if SUPABASE_ANON_KEY and auth.token and not auth.anonymous:
        h = {
            "apikey":        SUPABASE_ANON_KEY,
            "Authorization": f"Bearer {auth.token}",
            "Accept":        "application/json",
        }
    else:
        h = {
            "apikey":        SUPABASE_SERVICE_KEY,
            "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
            "Accept":        "application/json",
        }
    if extra:
        h.update(extra)
    return h


def save_crispr_design(
    user_id: Optional[str], *,
    label: str, mode: str, enzyme: Optional[str], base_editor: Optional[str],
    organism: Optional[str], gene_symbol: Optional[str],
    input_length: Optional[int], guides: list,
) -> Dict[str, Any]:
    """Insert a saved CRISPR design. Returns {ok, id} or {ok:False, error}.
    Synchronous so the UI can confirm + show the new design immediately."""
    if not user_id:
        return {"ok": False, "error": "Sign in to save designs."}
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "Saving isn't configured on this server."}
    guides = (guides or [])[:_CRISPR_DESIGNS_MAX]
    from datetime import datetime, timezone, timedelta
    expires_at = (None if has_pro_plan(user_id) else
                  (datetime.now(timezone.utc) + timedelta(days=FREE_LIBRARY_TTL_DAYS)).isoformat())
    payload = {
        "user_id":      user_id,
        "label":        (label or "Untitled design")[:120],
        "mode":         mode or "knockout",
        "enzyme":       enzyme or None,
        "base_editor":  base_editor or None,
        "organism":     organism or None,
        "gene_symbol":  gene_symbol or None,
        "input_length": input_length,
        "n_guides":     len(guides),
        "guides":       guides,
        "expires_at":   expires_at,
    }
    import urllib.request
    import urllib.error
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/crispr_designs",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers=_user_pgrst_headers({"Content-Type": "application/json",
                                         "Prefer": "return=representation"}),
        )
        with urllib.request.urlopen(req, timeout=8.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
            new_id = (rows[0] or {}).get("id") if isinstance(rows, list) and rows else None
            return {"ok": True, "id": new_id}
    except urllib.error.HTTPError as exc:
        body = ""
        try:
            body = exc.read().decode("utf-8")[:200]
        except Exception:  # noqa: BLE001
            pass
        logger.warning("crispr_designs insert HTTP %s: %s", exc.code, body)
        # 404/PGRST means the table is missing — guide the operator.
        if exc.code in (404, 406) or "PGRST" in body:
            return {"ok": False, "error": "Saved designs aren't enabled yet (database migration pending)."}
        return {"ok": False, "error": "Couldn't save the design. Please try again."}
    except Exception as exc:  # noqa: BLE001
        logger.warning("crispr_designs insert failed: %s", exc)
        return {"ok": False, "error": "Couldn't save the design. Please try again."}


def list_crispr_designs(user_id: Optional[str]) -> list:
    """The user's saved designs (metadata only, newest first). [] on error."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/crispr_designs"
               f"?user_id=eq.{user_id}"
               f"&select=id,label,created_at,expires_at,mode,enzyme,base_editor,gene_symbol,organism,n_guides,input_length"
               f"&order=created_at.desc&limit={_CRISPR_DESIGNS_LIST_LIMIT}")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("crispr_designs list failed for %s: %s", user_id, exc)
        return []


def get_crispr_design(user_id: Optional[str], design_id: str) -> Optional[Dict[str, Any]]:
    """One full design incl. guides, enforcing ownership server-side.
    Returns None if not found / not owned / malformed id."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return None
    if not _re.match(r"^[0-9a-fA-F-]{36}$", design_id or ""):
        return None
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/crispr_designs"
               f"?id=eq.{design_id}&user_id=eq.{user_id}&select=*&limit=1")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
            return rows[0] if isinstance(rows, list) and rows else None
    except Exception as exc:  # noqa: BLE001
        logger.warning("crispr_designs get failed: %s", exc)
        return None


def cleanup_expired_crispr_designs_async(user_id: Optional[str]) -> None:
    """Fire-and-forget delete of this user's expired designs (lazy sweep,
    piggybacks the save call). Mirrors cleanup_expired_libraries_async."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    import threading
    threading.Thread(target=_do_cleanup_expired_crispr_designs,
                     args=(user_id,), daemon=True).start()


def _do_cleanup_expired_crispr_designs(user_id: str) -> None:
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/crispr_designs"
               f"?user_id=eq.{user_id}&expires_at=lt.now()")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_supabase_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=4.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("crispr_designs cleanup failed for %s: %s", user_id, exc)


_PLASMIDS_LIST_LIMIT = 100        # max plasmids returned by the list call
_PLASMID_MAX_LEN = 500_000        # max sequence length stored per plasmid
_PLASMID_FEATURES_MAX = 2000      # max features stored per plasmid


def save_plasmid(
    user_id: Optional[str], *,
    name: str, topology: str, length: Optional[int], gc: Optional[float],
    source: Optional[str], sequence: str, features: list,
) -> Dict[str, Any]:
    """Insert a saved plasmid. Returns {ok, id} or {ok:False, error}.
    Synchronous so the UI can confirm + show it in the library immediately."""
    if not user_id:
        return {"ok": False, "error": "Sign in to save plasmids."}
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "Saving isn't configured on this server."}
    sequence = (sequence or "")[:_PLASMID_MAX_LEN]
    if not sequence:
        return {"ok": False, "error": "Nothing to save — import a plasmid first."}
    features = (features or [])[:_PLASMID_FEATURES_MAX]
    from datetime import datetime, timezone, timedelta
    expires_at = (None if has_pro_plan(user_id) else
                  (datetime.now(timezone.utc) + timedelta(days=FREE_LIBRARY_TTL_DAYS)).isoformat())
    payload = {
        "user_id":  user_id,
        "name":     (name or "Untitled plasmid")[:120],
        "topology": "linear" if topology == "linear" else "circular",
        "length":   length if isinstance(length, int) else len(sequence),
        "gc":       gc,
        "source":   source or None,
        "sequence": sequence,
        "features": features,
        "expires_at": expires_at,
    }
    import urllib.request
    import urllib.error
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/plasmids",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers=_user_pgrst_headers({"Content-Type": "application/json",
                                         "Prefer": "return=representation"}),
        )
        with urllib.request.urlopen(req, timeout=8.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
            new_id = (rows[0] or {}).get("id") if isinstance(rows, list) and rows else None
            return {"ok": True, "id": new_id}
    except urllib.error.HTTPError as exc:
        body = ""
        try:
            body = exc.read().decode("utf-8")[:200]
        except Exception:  # noqa: BLE001
            pass
        logger.warning("plasmids insert HTTP %s: %s", exc.code, body)
        if exc.code in (404, 406) or "PGRST" in body:
            return {"ok": False, "error": "Saved plasmids aren't enabled yet (database migration pending)."}
        return {"ok": False, "error": "Couldn't save the plasmid. Please try again."}
    except Exception as exc:  # noqa: BLE001
        logger.warning("plasmids insert failed: %s", exc)
        return {"ok": False, "error": "Couldn't save the plasmid. Please try again."}


def list_plasmids(user_id: Optional[str]) -> list:
    """The user's saved plasmids (metadata only, newest first). [] on error."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/plasmids"
               f"?user_id=eq.{user_id}"
               f"&select=id,name,created_at,expires_at,topology,length,gc,source"
               f"&order=created_at.desc&limit={_PLASMIDS_LIST_LIMIT}")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("plasmids list failed for %s: %s", user_id, exc)
        return []


def get_plasmid(user_id: Optional[str], plasmid_id: str) -> Optional[Dict[str, Any]]:
    """One full plasmid incl. sequence + features, enforcing ownership.
    Returns None if not found / not owned / malformed id."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return None
    if not _re.match(r"^[0-9a-fA-F-]{36}$", plasmid_id or ""):
        return None
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/plasmids"
               f"?id=eq.{plasmid_id}&user_id=eq.{user_id}&select=*&limit=1")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
            return rows[0] if isinstance(rows, list) and rows else None
    except Exception as exc:  # noqa: BLE001
        logger.warning("plasmids get failed: %s", exc)
        return None


def delete_plasmid(user_id: Optional[str], plasmid_id: str) -> Dict[str, Any]:
    """Delete one plasmid, owner-enforced."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "Not configured."}
    if not _re.match(r"^[0-9a-fA-F-]{36}$", plasmid_id or ""):
        return {"ok": False, "error": "Bad id."}
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/plasmids"
               f"?id=eq.{plasmid_id}&user_id=eq.{user_id}")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_user_pgrst_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=6.0):
            return {"ok": True}
    except Exception as exc:  # noqa: BLE001
        logger.warning("plasmids delete failed: %s", exc)
        return {"ok": False, "error": "Couldn't delete."}


def cleanup_expired_plasmids_async(user_id: Optional[str]) -> None:
    """Fire-and-forget delete of this user's expired plasmids (lazy sweep)."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    import threading
    threading.Thread(target=_do_cleanup_expired_plasmids,
                     args=(user_id,), daemon=True).start()


def _do_cleanup_expired_plasmids(user_id: str) -> None:
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/plasmids"
               f"?user_id=eq.{user_id}&expires_at=lt.now()")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_supabase_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=4.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("plasmids cleanup failed for %s: %s", user_id, exc)


# ─── Primer Analysis: per-user saved sets (unified "My designs" hub) ──
# Mirrors the crispr_designs CRUD. Table: public.primer_analyses
# (migration 0006). The full analyze result is stored as JSONB so a saved
# set re-renders without re-running the in-silico PCR.

_PRIMER_ANALYSES_LIST_LIMIT = 100    # max sets returned by the list call
_PRIMER_RESULT_MAX_BYTES = 600_000   # cap the stored result JSON (~0.6 MB)


def save_primer_analysis(
    user_id: Optional[str], *,
    label: str, organism: Optional[str],
    template_len: Optional[int], n_primers: Optional[int],
    result: Dict[str, Any],
) -> Dict[str, Any]:
    """Insert a saved primer set. Returns {ok, id} or {ok:False, error}."""
    if not user_id:
        return {"ok": False, "error": "Sign in to save primer sets."}
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "Saving isn't configured on this server."}
    # Guard against an oversized result blob (a runaway off-target product
    # list): keep the ranked primers, drop the bulk.
    try:
        if len(json.dumps(result or {})) > _PRIMER_RESULT_MAX_BYTES:
            result = {"primers": (result or {}).get("primers", [])}
    except Exception:  # noqa: BLE001
        result = {}
    from datetime import datetime, timezone, timedelta
    expires_at = (None if has_pro_plan(user_id) else
                  (datetime.now(timezone.utc) + timedelta(days=FREE_LIBRARY_TTL_DAYS)).isoformat())
    payload = {
        "user_id":      user_id,
        "label":        (label or "Untitled primer set")[:120],
        "organism":     organism or None,
        "template_len": template_len,
        "n_primers":    n_primers,
        "result":       result or {},
        "expires_at":   expires_at,
    }
    import urllib.request
    import urllib.error
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/primer_analyses",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers=_user_pgrst_headers({"Content-Type": "application/json",
                                         "Prefer": "return=representation"}),
        )
        with urllib.request.urlopen(req, timeout=8.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
            new_id = (rows[0] or {}).get("id") if isinstance(rows, list) and rows else None
            return {"ok": True, "id": new_id}
    except urllib.error.HTTPError as exc:
        body = ""
        try:
            body = exc.read().decode("utf-8")[:200]
        except Exception:  # noqa: BLE001
            pass
        logger.warning("primer_analyses insert HTTP %s: %s", exc.code, body)
        if exc.code in (404, 406) or "PGRST" in body:
            return {"ok": False, "error": "Saved primer sets aren't enabled yet (database migration pending)."}
        return {"ok": False, "error": "Couldn't save the primer set. Please try again."}
    except Exception as exc:  # noqa: BLE001
        logger.warning("primer_analyses insert failed: %s", exc)
        return {"ok": False, "error": "Couldn't save the primer set. Please try again."}


def list_primer_analyses(user_id: Optional[str]) -> list:
    """The user's saved primer sets (metadata only, newest first). [] on error."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/primer_analyses"
               f"?user_id=eq.{user_id}"
               f"&select=id,label,created_at,expires_at,organism,template_len,n_primers"
               f"&order=created_at.desc&limit={_PRIMER_ANALYSES_LIST_LIMIT}")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("primer_analyses list failed for %s: %s", user_id, exc)
        return []


def get_primer_analysis(user_id: Optional[str], analysis_id: str) -> Optional[Dict[str, Any]]:
    """One full saved primer set incl. result, enforcing ownership server-side."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return None
    if not _re.match(r"^[0-9a-fA-F-]{36}$", analysis_id or ""):
        return None
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/primer_analyses"
               f"?id=eq.{analysis_id}&user_id=eq.{user_id}&select=*&limit=1")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
            return rows[0] if isinstance(rows, list) and rows else None
    except Exception as exc:  # noqa: BLE001
        logger.warning("primer_analyses get failed: %s", exc)
        return None


def delete_primer_analysis(user_id: Optional[str], analysis_id: str) -> Dict[str, Any]:
    """Delete one saved primer set, owner-enforced."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "Not configured."}
    if not _re.match(r"^[0-9a-fA-F-]{36}$", analysis_id or ""):
        return {"ok": False, "error": "Bad id."}
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/primer_analyses"
               f"?id=eq.{analysis_id}&user_id=eq.{user_id}")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_user_pgrst_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=6.0):
            return {"ok": True}
    except Exception as exc:  # noqa: BLE001
        logger.warning("primer_analyses delete failed: %s", exc)
        return {"ok": False, "error": "Couldn't delete."}


def cleanup_expired_primer_analyses_async(user_id: Optional[str]) -> None:
    """Fire-and-forget delete of this user's expired primer sets (lazy sweep)."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    import threading
    threading.Thread(target=_do_cleanup_expired_primer_analyses,
                     args=(user_id,), daemon=True).start()


def _do_cleanup_expired_primer_analyses(user_id: str) -> None:
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/primer_analyses"
               f"?user_id=eq.{user_id}&expires_at=lt.now()")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_supabase_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=4.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("primer_analyses cleanup failed for %s: %s", user_id, exc)


# ─── DE outcomes: per-library wet-lab results (the learning flywheel) ──
# One row per measured variant; the surrogate fits on these for round 2.
# Table: public.de_outcomes (migration 0007).

_DE_OUTCOMES_MAX = 2000              # max measured variants stored per library
_UUIDISH = r"^[0-9a-fA-F-]{32,36}$"  # accept 32-hex or dashed uuid (libraries.id form)


def save_de_outcomes(user_id: Optional[str], library_id: str, rows: list) -> Dict[str, Any]:
    """Replace this user's logged outcomes for a library with `rows`
    (each {mutations, measured_value, outcome}). Returns {ok, n} or error."""
    import re as _re
    if not user_id:
        return {"ok": False, "error": "Sign in to log results."}
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "Saving isn't configured on this server."}
    if not _re.match(_UUIDISH, library_id or ""):
        return {"ok": False, "error": "Bad library id."}
    rows = (rows or [])[:_DE_OUTCOMES_MAX]
    from datetime import datetime, timezone, timedelta
    expires_at = (None if has_pro_plan(user_id) else
                  (datetime.now(timezone.utc) + timedelta(days=FREE_LIBRARY_TTL_DAYS)).isoformat())
    payload = []
    for r in rows:
        if not isinstance(r, dict):
            continue
        outcome = str(r.get("outcome", "measured") or "measured")[:16]
        mv = r.get("measured_value")
        try:
            mv = float(mv) if mv is not None and mv != "" else None
        except (TypeError, ValueError):
            mv = None
        payload.append({
            "user_id":        user_id,
            "library_id":     library_id,
            "mutations":      str(r.get("mutations", "") or "")[:300],
            "measured_value": mv,
            "outcome":        outcome,
            "expires_at":     expires_at,
        })
    import urllib.request
    import urllib.error
    # Replace semantics: clear this library's prior outcomes, then insert.
    try:
        url = (f"{SUPABASE_URL}/rest/v1/de_outcomes"
               f"?user_id=eq.{user_id}&library_id=eq.{library_id}")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_user_pgrst_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=6.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("de_outcomes clear failed: %s", exc)
    if not payload:
        return {"ok": True, "n": 0}
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/de_outcomes",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers=_user_pgrst_headers({"Content-Type": "application/json",
                                         "Prefer": "return=minimal"}),
        )
        with urllib.request.urlopen(req, timeout=8.0):
            return {"ok": True, "n": len(payload)}
    except urllib.error.HTTPError as exc:
        body = ""
        try:
            body = exc.read().decode("utf-8")[:200]
        except Exception:  # noqa: BLE001
            pass
        logger.warning("de_outcomes insert HTTP %s: %s", exc.code, body)
        if exc.code in (404, 406) or "PGRST" in body:
            return {"ok": False, "error": "Result logging isn't enabled yet (database migration pending)."}
        return {"ok": False, "error": "Couldn't save your results. Please try again."}
    except Exception as exc:  # noqa: BLE001
        logger.warning("de_outcomes insert failed: %s", exc)
        return {"ok": False, "error": "Couldn't save your results. Please try again."}


def list_de_outcomes(user_id: Optional[str], library_id: str) -> list:
    """This user's logged outcomes for one library (newest first). [] on error."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    if not _re.match(_UUIDISH, library_id or ""):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/de_outcomes"
               f"?user_id=eq.{user_id}&library_id=eq.{library_id}"
               f"&select=id,mutations,measured_value,outcome,created_at"
               f"&order=created_at.desc&limit={_DE_OUTCOMES_MAX}")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("de_outcomes list failed: %s", exc)
        return []


def cleanup_expired_de_outcomes_async(user_id: Optional[str]) -> None:
    """Fire-and-forget delete of this user's expired outcome rows (lazy sweep)."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return
    import threading
    threading.Thread(target=_do_cleanup_expired_de_outcomes,
                     args=(user_id,), daemon=True).start()


def _do_cleanup_expired_de_outcomes(user_id: str) -> None:
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/de_outcomes"
               f"?user_id=eq.{user_id}&expires_at=lt.now()")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_supabase_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=4.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("de_outcomes cleanup failed for %s: %s", user_id, exc)


# ─── Cross-user AGGREGATE prior (the data moat) ──────────────────────
# Table: public.mutation_priors (migration 0008). De-identified, pooled
# substitution-effect statistics. The functions below use the service-role
# key (which bypasses RLS) — the unfiltered de_outcomes read is the ONLY
# place we read across users, and it pulls just the fields the aggregation
# needs (no IPs, no sequences). See dee/core/aggregate.py for the math + the
# 2026-07-08 execution gate.

_AGG_OUTCOME_SCAN_MAX = 100000      # hard ceiling on rows scanned per rebuild


def list_all_de_outcomes_grouped() -> list:
    """ALL users' outcomes, grouped per (user_id, library_id), for aggregation.

    Returns [(user_id, [(mutations, measured_value), ...]), ...]. Service-role
    read with NO user filter (this is the aggregation path). Only mutations +
    measured_value are pulled. [] on error/misconfig.
    """
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/de_outcomes"
               f"?select=user_id,library_id,mutations,measured_value"
               f"&measured_value=not.is.null"
               f"&order=user_id,library_id"
               f"&limit={_AGG_OUTCOME_SCAN_MAX}")
        req = urllib.request.Request(url, method="GET", headers=_supabase_headers())
        with urllib.request.urlopen(req, timeout=20.0) as resp:
            rows = json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("de_outcomes aggregate scan failed: %s", exc)
        return []

    grouped: Dict[tuple, list] = {}
    for r in rows:
        key = (r.get("user_id"), r.get("library_id"))
        if not key[0] or not key[1]:
            continue
        grouped.setdefault(key, []).append((r.get("mutations") or "", r.get("measured_value")))
    # collapse to (user_id, measurements) — aggregation keys k-anon by user_id,
    # so each (user, library) is one contributing unit.
    return [(uid, meas) for (uid, _lib), meas in grouped.items()]


def list_all_de_outcomes_grouped_with_lib() -> list:
    """Like list_all_de_outcomes_grouped but keeps library_id so the richer
    (ESM-binned) rebuild can fetch each library's WT. [(user_id, library_id,
    measurements), …]."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/de_outcomes"
               f"?select=user_id,library_id,mutations,measured_value"
               f"&measured_value=not.is.null&order=user_id,library_id"
               f"&limit={_AGG_OUTCOME_SCAN_MAX}")
        req = urllib.request.Request(url, method="GET", headers=_supabase_headers())
        with urllib.request.urlopen(req, timeout=20.0) as resp:
            rows = json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("de_outcomes lib-aware scan failed: %s", exc)
        return []
    grouped: Dict[tuple, list] = {}
    for r in rows:
        key = (r.get("user_id"), r.get("library_id"))
        if key[0] and key[1]:
            grouped.setdefault(key, []).append((r.get("mutations") or "", r.get("measured_value")))
    return [(uid, lib, meas) for (uid, lib), meas in grouped.items()]


def get_library_wt(library_id: str) -> Optional[str]:
    """The wild-type protein stored with a saved DE library (service-role)."""
    import re as _re
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY) or not _re.match(_UUIDISH, library_id or ""):
        return None
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/libraries"
               f"?id=eq.{library_id}&select=wt_protein&limit=1")
        req = urllib.request.Request(url, method="GET", headers=_supabase_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            data = json.loads(resp.read().decode("utf-8")) or []
        return (data[0].get("wt_protein") or None) if data else None
    except Exception as exc:  # noqa: BLE001
        logger.warning("library wt fetch failed: %s", exc)
        return None


def get_mutation_priors() -> list:
    """The stored aggregate rows (service-role read). [] if none/misconfig."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/mutation_priors"
               f"?select=substitution,n_users,n_obs,mean_effect&order=substitution")
        req = urllib.request.Request(url, method="GET", headers=_supabase_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("mutation_priors read failed: %s", exc)
        return []


def replace_mutation_priors(rows: list) -> Dict[str, Any]:
    """Replace the aggregate table with a fresh set of de-identified rows.
    Each row: {substitution, n_users, n_obs, mean_effect}. Service-role.
    """
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "supabase-not-configured"}
    import urllib.request
    # wipe then upsert (the aggregate is fully recomputed each run)
    try:
        del_url = f"{SUPABASE_URL}/rest/v1/mutation_priors?substitution=neq.__none__"
        req = urllib.request.Request(del_url, method="DELETE",
                                     headers=_supabase_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=10.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("mutation_priors clear failed: %s", exc)
    if not rows:
        return {"ok": True, "written": 0}
    try:
        payload = json.dumps(rows).encode("utf-8")
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/mutation_priors",
            data=payload, method="POST",
            headers=_supabase_headers({"Content-Type": "application/json",
                                       "Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=15.0):
            return {"ok": True, "written": len(rows)}
    except urllib.error.HTTPError as exc:  # noqa: BLE001
        body = exc.read().decode("utf-8", "replace")[:300]
        logger.warning("mutation_priors insert HTTP %s: %s", exc.code, body)
        return {"ok": False, "error": f"http-{exc.code}", "detail": body}
    except Exception as exc:  # noqa: BLE001
        logger.warning("mutation_priors insert failed: %s", exc)
        return {"ok": False, "error": str(exc)}


# ─── Unified outcomes framework (public.outcomes / public.outcome_priors) ─────
# Generic, tool-tagged version of the DE outcome CRUD above. Any tool's capture
# UI POSTs here; the per-tool aggregate (dee/core/outcomes.py) reads across users
# and writes outcome_priors. See migration 0009. DE keeps its dedicated tables.

_OUTCOMES_MAX = 2000
_TOOL_RE = __import__("re").compile(r"^[a-z][a-z0-9_]{1,23}$")  # valid tool slug


def save_outcomes(user_id: Optional[str], tool: str, design_id: str, rows: list) -> Dict[str, Any]:
    """Replace this user's outcomes for (tool, design_id). rows: [{subject,
    measured_value?, outcome?}]. Mirrors save_de_outcomes."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return {"ok": False, "error": "not-configured"}
    if not _TOOL_RE.match(tool or ""):
        return {"ok": False, "error": "bad-tool"}
    if not _re.match(_UUIDISH, design_id or ""):
        return {"ok": False, "error": "bad-design-id"}
    rows = (rows or [])[:_OUTCOMES_MAX]
    from datetime import datetime as _dtt, timezone as _tz, timedelta as _td
    expires_at = (None if has_pro_plan(user_id)
                  else (_dtt.now(_tz.utc) + _td(days=FREE_LIBRARY_TTL_DAYS)).isoformat())
    payload = []
    for r in rows:
        if not isinstance(r, dict):
            continue
        mv = r.get("measured_value")
        try:
            mv = float(mv) if mv is not None and mv != "" else None
        except (TypeError, ValueError):
            mv = None
        payload.append({
            "user_id": user_id, "tool": tool, "design_id": design_id,
            "subject": str(r.get("subject", ""))[:512],
            "measured_value": mv,
            "outcome": str(r.get("outcome", "measured"))[:32],
            "expires_at": expires_at,
        })
    import urllib.request
    try:  # replace-on-save: clear this user's rows for the design, then insert
        url = (f"{SUPABASE_URL}/rest/v1/outcomes"
               f"?user_id=eq.{user_id}&tool=eq.{tool}&design_id=eq.{design_id}")
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_user_pgrst_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=6.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcomes clear failed: %s", exc)
    if not payload:
        return {"ok": True, "saved": 0}
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/outcomes", data=json.dumps(payload).encode("utf-8"),
            method="POST", headers=_user_pgrst_headers({"Content-Type": "application/json",
                                                        "Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=10.0):
            return {"ok": True, "saved": len(payload)}
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcomes insert failed: %s", exc)
        return {"ok": False, "error": str(exc)}


def list_outcomes(user_id: Optional[str], tool: str, design_id: str) -> list:
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    if not _TOOL_RE.match(tool or "") or not _re.match(_UUIDISH, design_id or ""):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/outcomes"
               f"?user_id=eq.{user_id}&tool=eq.{tool}&design_id=eq.{design_id}"
               f"&select=id,subject,measured_value,outcome,created_at"
               f"&order=created_at.desc&limit={_OUTCOMES_MAX}")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcomes list failed: %s", exc)
        return []


def list_all_outcomes_grouped(tool: str) -> list:
    """ALL users' outcomes for one tool, grouped per (user, design) for the
    aggregate. [(user_id, [(subject, measured_value), …]), …]. Service-role."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY) or not _TOOL_RE.match(tool or ""):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/outcomes"
               f"?tool=eq.{tool}&measured_value=not.is.null"
               f"&select=user_id,design_id,subject,measured_value"
               f"&order=user_id,design_id&limit=100000")
        req = urllib.request.Request(url, method="GET", headers=_supabase_headers())
        with urllib.request.urlopen(req, timeout=20.0) as resp:
            data = json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcomes aggregate scan failed: %s", exc)
        return []
    grouped: Dict[tuple, list] = {}
    for r in data:
        key = (r.get("user_id"), r.get("design_id"))
        if key[0] and key[1]:
            grouped.setdefault(key, []).append((r.get("subject") or "", r.get("measured_value")))
    return [(uid, rows) for (uid, _d), rows in grouped.items()]


def get_outcome_priors(tool: str) -> list:
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY) or not _TOOL_RE.match(tool or ""):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/outcome_priors"
               f"?tool=eq.{tool}&select=tool,feature_key,n_users,n_obs,mean_effect&order=feature_key")
        req = urllib.request.Request(url, method="GET", headers=_supabase_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcome_priors read failed: %s", exc)
        return []


def replace_outcome_priors(tool: str, rows: list) -> Dict[str, Any]:
    """Replace one tool's aggregate rows. rows: [{tool, feature_key, n_users,
    n_obs, mean_effect}]. Service-role."""
    if not (SUPABASE_URL and SUPABASE_SERVICE_KEY) or not _TOOL_RE.match(tool or ""):
        return {"ok": False, "error": "not-configured-or-bad-tool"}
    import urllib.request
    try:
        url = f"{SUPABASE_URL}/rest/v1/outcome_priors?tool=eq.{tool}"
        req = urllib.request.Request(url, method="DELETE",
                                     headers=_supabase_headers({"Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=10.0):
            pass
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcome_priors clear failed: %s", exc)
    if not rows:
        return {"ok": True, "written": 0}
    try:
        req = urllib.request.Request(
            f"{SUPABASE_URL}/rest/v1/outcome_priors", data=json.dumps(rows).encode("utf-8"),
            method="POST", headers=_supabase_headers({"Content-Type": "application/json",
                                                      "Prefer": "return=minimal"}))
        with urllib.request.urlopen(req, timeout=15.0):
            return {"ok": True, "written": len(rows)}
    except urllib.error.HTTPError as exc:  # noqa: BLE001
        body = exc.read().decode("utf-8", "replace")[:300]
        logger.warning("outcome_priors insert HTTP %s: %s", exc.code, body)
        return {"ok": False, "error": f"http-{exc.code}", "detail": body}
    except Exception as exc:  # noqa: BLE001
        logger.warning("outcome_priors insert failed: %s", exc)
        return {"ok": False, "error": str(exc)}


# ─── Directed-Evolution libraries: per-user list + signed download ────
# DE runs already persist to public.libraries + the `libraries` Storage
# bucket (see _do_save_library). These surface them in the unified hub —
# list metadata + mint a short-lived signed URL for the CSV download.

_LIBRARIES_LIST_LIMIT = 100


def list_libraries(user_id: Optional[str]) -> list:
    """The user's saved DE libraries (metadata, newest first, non-expired)."""
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return []
    import urllib.request
    try:
        url = (f"{SUPABASE_URL}/rest/v1/libraries"
               f"?user_id=eq.{user_id}"
               f"&select=id,name,created_at,expires_at,n_variants,top_fitness,csv_size_bytes"
               f"&or=(expires_at.gte.now,expires_at.is.null)"
               f"&order=created_at.desc&limit={_LIBRARIES_LIST_LIMIT}")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            return json.loads(resp.read().decode("utf-8")) or []
    except Exception as exc:  # noqa: BLE001
        logger.warning("libraries list failed for %s: %s", user_id, exc)
        return []


def library_csv_url(user_id: Optional[str], library_id: str) -> Optional[str]:
    """Signed, time-limited download URL for a library's CSV, owner-enforced.
    Returns an absolute URL or None. Never exposes the service key."""
    import re as _re
    if not user_id or not (SUPABASE_URL and SUPABASE_SERVICE_KEY):
        return None
    if not _re.match(r"^[0-9a-fA-F-]{36}$", library_id or ""):
        return None
    import urllib.request
    # 1. Fetch the row to confirm ownership and get the storage_path.
    try:
        url = (f"{SUPABASE_URL}/rest/v1/libraries"
               f"?id=eq.{library_id}&user_id=eq.{user_id}&select=storage_path&limit=1")
        req = urllib.request.Request(url, method="GET", headers=_user_pgrst_headers())
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            rows = json.loads(resp.read().decode("utf-8"))
        storage_path = (rows[0] if isinstance(rows, list) and rows else {}).get("storage_path")
        if not storage_path:
            return None
    except Exception as exc:  # noqa: BLE001
        logger.warning("library_csv_url lookup failed: %s", exc)
        return None
    # 2. Mint a signed URL (valid 5 min) from Storage.
    try:
        sign_url = f"{SUPABASE_URL}/storage/v1/object/sign/libraries/{storage_path}"
        req = urllib.request.Request(
            sign_url,
            data=json.dumps({"expiresIn": 300}).encode("utf-8"),
            method="POST",
            headers=_supabase_headers({"Content-Type": "application/json"}),
        )
        with urllib.request.urlopen(req, timeout=6.0) as resp:
            data = json.loads(resp.read().decode("utf-8"))
        signed = (data or {}).get("signedURL") or (data or {}).get("signedUrl")
        if not signed:
            return None
        if signed.startswith("http"):
            return signed
        if not signed.startswith("/"):
            signed = "/" + signed
        # Storage returns "/object/sign/libraries/<path>?token=…" → prefix /storage/v1.
        return f"{SUPABASE_URL}{signed}" if signed.startswith("/storage/") else f"{SUPABASE_URL}/storage/v1{signed}"
    except Exception as exc:  # noqa: BLE001
        logger.warning("library sign failed: %s", exc)
        return None


def _anonymize_ip(ip: str) -> Optional[str]:
    """Drop the host-identifying bits of an IP so we keep coarse network
    provenance (useful for abuse triage) without persisting a full,
    person-linkable address. Standard GDPR IP-anonymization: zero the last
    IPv4 octet (/24) or collapse IPv6 to its first three hextets (~/48).

    Stays a *valid IP literal* so it's safe whether the runs.request_ip column
    is text or inet. Returns None for empty/garbage input.
    """
    ip = (ip or "").strip()
    if not ip:
        return None
    if ":" in ip:  # IPv6
        parts = ip.split(":")
        head = [p for p in parts[:3] if p != ""]
        return ":".join(head) + "::" if head else None
    if "." in ip:  # IPv4
        octets = ip.split(".")
        if len(octets) == 4 and all(o.isdigit() and 0 <= int(o) <= 255 for o in octets):
            return f"{octets[0]}.{octets[1]}.{octets[2]}.0"
    return None  # not a recognizable IP — don't store anything


def _anon_fingerprint() -> str:
    """Stable-ish hash of IP + UA, salted with ANON_SECRET. Used only to
    rough-deduplicate anonymous runs in analytics. Not identity-grade —
    same household + same browser → same fingerprint."""
    ip = request.headers.get("X-Forwarded-For", request.remote_addr or "").split(",")[0].strip()
    ua = request.headers.get("User-Agent", "")[:512]
    data = f"{ip}|{ua}".encode("utf-8")
    return hmac.new(
        (ANON_SECRET or "no-secret").encode("utf-8"),
        data,
        hashlib.sha256,
    ).hexdigest()[:32]


# ─── App registration ─────────────────────────────────────────────────
def init_app(app: Flask) -> None:
    """Wire up the before_request hook on the Flask app."""
    app.before_request(load_auth_into_g)

    if AUTH_ENABLED:
        logger.info(
            "Auth enabled: Supabase JWT + anon trial timer (%ds free window).",
            ANON_TRIAL_SECONDS,
        )
        # Surface whether per-user DB calls run AS THE USER (RLS-enforced) or
        # fall back to the service-role key. The anon key is public, so this
        # is just an operator nudge, not a secret.
        if SUPABASE_SERVICE_KEY and not SUPABASE_ANON_KEY:
            logger.warning(
                "SUPABASE_ANON_KEY is unset — per-user reads/writes will use the "
                "service-role key and Row-Level Security is NOT enforced at the "
                "row level (the user_id=eq. filters are the only guard). Set "
                "SUPABASE_ANON_KEY (the public anon key) to enable RLS-enforced "
                "per-user access."
            )
        elif SUPABASE_ANON_KEY:
            logger.info("Per-user DB access runs as the user — RLS enforced in Postgres.")
    else:
        logger.warning(
            "AUTH IS OFF — set SUPABASE_JWT_SECRET + TURINGDNA_ANON_SECRET "
            "as env vars to enable. All requests will be treated as "
            "anonymous with unlimited quota."
        )