hf_dashboard.py

from __future__ import annotations

import copy
import hashlib
import io
import json
import math
import os
import re
from collections import Counter
from datetime import datetime, timedelta, timezone
from typing import Any, Mapping
from urllib import error as urllib_error
from urllib import parse as urllib_parse
from urllib import request as urllib_request

from PIL import Image

SUPABASE_URL = os.getenv("SUPABASE_URL", "").rstrip("/")
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")
GENERATIONS_BUCKET = os.getenv("SUPABASE_GENERATIONS_BUCKET", "qr-generations")

# Optional: dual-write to kvm9276 (qrcut.co) when these env vars are set.
# When unset, behavior is identical to today (Supabase only).
QRCUT_STORAGE_URL = os.getenv("QRCUT_STORAGE_URL", "").rstrip("/")
QRCUT_STORAGE_PUBLIC_URL = os.getenv("QRCUT_STORAGE_PUBLIC_URL", QRCUT_STORAGE_URL).rstrip("/")
QRCUT_WRITE_API_KEY = os.getenv("QRCUT_WRITE_API_KEY", "")


class DashboardError(RuntimeError):
    pass


_BUCKET_READY = False


def _require_supabase() -> None:
    if not SUPABASE_URL or not SUPABASE_SERVICE_ROLE_KEY:
        raise DashboardError(
            "Supabase is not configured. Set SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY."
        )


def _quote_filter(value: str) -> str:
    return urllib_parse.quote(str(value), safe="")


def _headers(*, json_body: bool = True, extra: Mapping[str, str] | None = None) -> dict[str, str]:
    headers = {
        "apikey": SUPABASE_SERVICE_ROLE_KEY,
        "Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}",
        "User-Agent": "hermes-space-server/1.0",
    }
    if json_body:
        headers["Content-Type"] = "application/json"
    if extra:
        headers.update(dict(extra))
    return headers


def _request_json(
    method: str,
    path: str,
    *,
    payload: Any | None = None,
    headers: Mapping[str, str] | None = None,
    timeout: float = 20,
) -> Any:
    _require_supabase()
    data = None
    body_headers = _headers(json_body=True, extra=headers)
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
    req = urllib_request.Request(
        f"{SUPABASE_URL}{path}",
        data=data,
        headers=body_headers,
        method=method,
    )
    try:
        with urllib_request.urlopen(req, timeout=timeout) as response:
            raw = response.read()
            if not raw:
                return None
            return json.loads(raw.decode("utf-8"))
    except urllib_error.HTTPError as exc:
        detail = exc.read().decode("utf-8", "ignore")
        raise DashboardError(f"Supabase request failed ({exc.code} {method} {path}): {detail}") from exc
    except urllib_error.URLError as exc:
        raise DashboardError(f"Supabase request failed ({method} {path}): {exc}") from exc


def _request_bytes(
    method: str,
    path: str,
    *,
    data: bytes,
    content_type: str,
    headers: Mapping[str, str] | None = None,
    timeout: float = 30,
) -> None:
    _require_supabase()
    body_headers = _headers(
        json_body=False,
        extra={"Content-Type": content_type, **(dict(headers or {}))},
    )
    req = urllib_request.Request(
        f"{SUPABASE_URL}{path}",
        data=data,
        headers=body_headers,
        method=method,
    )
    try:
        with urllib_request.urlopen(req, timeout=timeout):
            return None
    except urllib_error.HTTPError as exc:
        detail = exc.read().decode("utf-8", "ignore")
        raise DashboardError(f"Supabase upload failed ({exc.code} {method} {path}): {detail}") from exc
    except urllib_error.URLError as exc:
        raise DashboardError(f"Supabase upload failed ({method} {path}): {exc}") from exc


def _table_select(table: str, *, query: str) -> list[dict[str, Any]]:
    result = _request_json("GET", f"/rest/v1/{table}?{query}")
    if isinstance(result, list):
        return result
    return []


def _parse_timestamp(value: Any) -> datetime | None:
    text = str(value or "").strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    if "." in text:
        main, suffix = text.split(".", 1)
        tz_index_plus = suffix.find("+")
        tz_index_minus = suffix.find("-")
        tz_index_candidates = [idx for idx in (tz_index_plus, tz_index_minus) if idx != -1]
        tz_index = min(tz_index_candidates) if tz_index_candidates else -1
        if tz_index != -1:
            fraction = suffix[:tz_index]
            timezone_part = suffix[tz_index:]
        else:
            fraction = suffix
            timezone_part = ""
        if fraction:
            fraction = (fraction + "000000")[:6]
            text = f"{main}.{fraction}{timezone_part}"
    if len(text) >= 3 and text[-3] in {"+", "-"} and text[-2:].isdigit():
        text = text + ":00"
    try:
        return datetime.fromisoformat(text)
    except ValueError:
        return None


def _insert_row(table: str, row: Mapping[str, Any], *, upsert: bool = False) -> list[dict[str, Any]]:
    prefer = "return=representation"
    if upsert:
        prefer += ",resolution=merge-duplicates"
    result = _request_json(
        "POST",
        f"/rest/v1/{table}",
        payload=row,
        headers={"Prefer": prefer},
    )
    if isinstance(result, list):
        return result
    return []


def _patch_rows(table: str, *, filters: str, values: Mapping[str, Any]) -> list[dict[str, Any]]:
    result = _request_json(
        "PATCH",
        f"/rest/v1/{table}?{filters}",
        payload=values,
        headers={"Prefer": "return=representation"},
    )
    if isinstance(result, list):
        return result
    return []


def _ensure_bucket() -> None:
    global _BUCKET_READY
    if _BUCKET_READY:
        return
    try:
        _request_json(
            "POST",
            "/storage/v1/bucket",
            payload={
                "id": GENERATIONS_BUCKET,
                "name": GENERATIONS_BUCKET,
                "public": True,
            },
        )
    except DashboardError as exc:
        message = str(exc)
        if "Duplicate" not in message and "already exists" not in message:
            raise
    _BUCKET_READY = True


def _png_bytes(image: Image.Image, *, max_size: int | None = None) -> bytes:
    image = image.convert("RGB")
    if max_size is not None:
        image = image.copy()
        image.thumbnail((max_size, max_size))
    buf = io.BytesIO()
    image.save(buf, format="PNG", optimize=True)
    return buf.getvalue()


def _upload_public_png(*, image: Image.Image, object_path: str, max_size: int | None = None) -> str:
    png_bytes = _png_bytes(image, max_size=max_size)
    _ensure_bucket()
    _request_bytes(
        "POST",
        f"/storage/v1/object/{GENERATIONS_BUCKET}/{urllib_parse.quote(object_path, safe='/')}",
        data=png_bytes,
        content_type="image/png",
        headers={"x-upsert": "true"},
    )

    public_url = f"{SUPABASE_URL}/storage/v1/object/public/{GENERATIONS_BUCKET}/{object_path}"

    # Optional dual-write to kvm9276 (qrcut.co) when QRCUT_* env vars are set.
    # Failures here are logged but do NOT break the Supabase write -- Supabase
    # remains the source of truth until the user explicitly deletes the bucket.
    if QRCUT_STORAGE_URL and QRCUT_WRITE_API_KEY:
        print(
            f"[qrcut] dual-write enabled path={object_path} bytes={len(png_bytes)} target={QRCUT_STORAGE_URL}/storage-write/...",
            flush=True,
        )
        try:
            kvm_req = urllib_request.Request(
                f"{QRCUT_STORAGE_URL}/storage-write/{urllib_parse.quote(object_path, safe='/')}",
                data=png_bytes,
                headers={
                    "X-QRCUT-WRITE-KEY": QRCUT_WRITE_API_KEY,
                    "Content-Type": "image/png",
                    "X-Upsert": "true",
                    "User-Agent": "AI-QR-Code-Generator/1.0 (+https://huggingface.co/spaces/Oysiyl/AI-QR-code-generator)",
                    "Accept": "application/json, text/plain, */*",
                },
                method="PUT",
            )
            with urllib_request.urlopen(kvm_req, timeout=30) as kvm_resp:
                if kvm_resp.status == 201:
                    public_url = f"{QRCUT_STORAGE_PUBLIC_URL}/storage/{object_path}"
                    print(f"[qrcut] dual-write success HTTP 201: {object_path}", flush=True)
                else:
                    response_body = kvm_resp.read(400).decode("utf-8", errors="replace").strip()
                    print(
                        f"[qrcut] dual-write HTTP {kvm_resp.status}: {object_path} body={response_body}",
                        flush=True,
                    )
        except urllib_error.HTTPError as exc:
            response_body = exc.read(400).decode("utf-8", errors="replace").strip()
            print(
                f"[qrcut] dual-write failed HTTP {exc.code}: {exc.reason} body={response_body} -- {object_path}",
                flush=True,
            )
        except (urllib_error.URLError, TimeoutError, OSError) as exc:
            print(f"[qrcut] dual-write failed: {exc} -- {object_path}", flush=True)
    else:
        print(
            f"[qrcut] dual-write skipped missing env: storage_url={'yes' if QRCUT_STORAGE_URL else 'no'} write_key={'yes' if QRCUT_WRITE_API_KEY else 'no'} path={object_path}",
            flush=True,
        )

    return public_url


def _session_oauth_info(request: Any | None) -> dict[str, Any]:
    if request is None:
        return {}
    session = getattr(request, "session", None) or {}
    oauth_info = session.get("oauth_info") if isinstance(session, Mapping) else None
    if isinstance(oauth_info, Mapping):
        return dict(oauth_info)
    return {}


def _request_headers(request: Any | None) -> Mapping[str, Any]:
    if request is None:
        return {}
    headers = getattr(request, "headers", None)
    if isinstance(headers, Mapping):
        return headers
    return {}


def _build_anon_id(request: Any | None, source: str = "ui", fallback: str = "anonymous") -> str | None:
    if request is None:
        return None
    request_session = getattr(request, "session_hash", None)
    headers = _request_headers(request)
    raw = "|".join(
        [
            source or "unknown",
            str(request_session or ""),
            str(headers.get("x-forwarded-for", "")),
            str(headers.get("user-agent", "")),
            fallback,
        ]
    )
    if raw.count("|") == 4 and all(not part for part in raw.split("|")[:-1]):
        return None
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24]


def _build_browser_fingerprint(request: Any | None, source: str = "ui") -> str | None:
    if request is None:
        return None
    headers = _request_headers(request)
    raw = "|".join(
        [
            source or "unknown",
            str(headers.get("x-forwarded-for", "")),
            str(headers.get("user-agent", "")),
            str(headers.get("accept-language", "")),
        ]
    )
    if raw.count("|") == 3 and all(not part for part in raw.split("|")): 
        return None
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24]


def get_current_user_context(request: Any | None, username_hint: str | None = None) -> dict[str, Any] | None:
    oauth_info = _session_oauth_info(request)
    raw_userinfo = oauth_info.get("userinfo")
    userinfo: Mapping[str, Any] = raw_userinfo if isinstance(raw_userinfo, Mapping) else {}
    hinted_username = str(username_hint or "").strip() if oauth_info else ""
    username = str(userinfo.get("preferred_username") or getattr(request, "username", "") or hinted_username or "").strip()
    if not username:
        return None

    stable_provider_subject = str(userinfo.get("sub") or "").strip()
    provider_subject = stable_provider_subject or f"hf-username:{username.lower()}"
    display_name = str(userinfo.get("name") or username or "").strip()
    avatar_url = str(userinfo.get("picture") or "").strip()
    profile_url = str(userinfo.get("profile") or (f"https://huggingface.co/{username}" if username else "")).strip()
    email = str(userinfo.get("email") or "").strip() or None
    return {
        "provider": "huggingface",
        "provider_subject": provider_subject,
        "stable_provider_subject": stable_provider_subject or None,
        "provisional_provider_subject": f"hf-username:{username.lower()}",
        "stable_identity": bool(stable_provider_subject),
        "username": username,
        "display_name": display_name or username,
        "avatar_url": avatar_url or None,
        "profile_url": profile_url or None,
        "email": email,
        "raw_profile": {"userinfo": userinfo, "oauth_info": {k: v for k, v in oauth_info.items() if k != "access_token"}},
    }


def ensure_app_user_for_request(request: Any | None, username_hint: str | None = None) -> dict[str, Any] | None:
    user = get_current_user_context(request, username_hint=username_hint)
    if user is None:
        return None

    exact_filters = (
        f"provider=eq.{_quote_filter(user['provider'])}"
        f"&provider_subject=eq.{_quote_filter(user['provider_subject'])}"
        "&select=id,app_user_id,provider_subject"
        "&limit=1"
    )
    existing = _table_select("user_identities", query=exact_filters)
    existing_row = existing[0] if existing else None

    provisional_row = None
    if not existing_row and user.get("stable_identity"):
        provisional_filters = (
            f"provider=eq.{_quote_filter(user['provider'])}"
            f"&provider_subject=eq.{_quote_filter(user['provisional_provider_subject'])}"
            "&select=id,app_user_id,provider_subject"
            "&limit=1"
        )
        provisional = _table_select("user_identities", query=provisional_filters)
        provisional_row = provisional[0] if provisional else None

    identity_row = existing_row or provisional_row
    app_user_id = str(identity_row.get("app_user_id") or "").strip() if identity_row else ""

    now_iso = datetime.now(timezone.utc).isoformat()
    if not app_user_id:
        inserted_users = _insert_row(
            "app_users",
            {
                "display_name": user["display_name"],
                "avatar_url": user["avatar_url"],
                "created_at": now_iso,
                "updated_at": now_iso,
                "last_login_at": now_iso,
            },
        )
        if not inserted_users:
            raise DashboardError("Failed to create app user row.")
        app_user_id = str(inserted_users[0]["id"])
        _insert_row(
            "user_identities",
            {
                "app_user_id": app_user_id,
                "provider": user["provider"],
                "provider_subject": user["provider_subject"],
                "username": user["username"],
                "email": user["email"],
                "raw_profile": user["raw_profile"],
                "created_at": now_iso,
                "updated_at": now_iso,
            },
            upsert=True,
        )
    else:
        _patch_rows(
            "app_users",
            filters=f"id=eq.{_quote_filter(app_user_id)}",
            values={
                "display_name": user["display_name"],
                "avatar_url": user["avatar_url"],
                "updated_at": now_iso,
                "last_login_at": now_iso,
            },
        )
        identity_provider_subject = user["provider_subject"]
        if identity_row:
            identity_provider_subject = str(identity_row.get("provider_subject") or user["provider_subject"])
        identity_filters = (
            f"provider=eq.{_quote_filter(user['provider'])}"
            f"&provider_subject=eq.{_quote_filter(identity_provider_subject)}"
        )
        _patch_rows(
            "user_identities",
            filters=identity_filters,
            values={
                "provider_subject": user["provider_subject"],
                "username": user["username"],
                "email": user["email"],
                "raw_profile": user["raw_profile"],
                "updated_at": now_iso,
            },
        )

    _claim_anonymous_generations(request, app_user_id)

    return {
        **user,
        "app_user_id": app_user_id,
    }


def _claim_anonymous_generations(request: Any | None, app_user_id: str) -> int:
    if not app_user_id:
        return 0

    total = 0
    claimed_ids: set[str] = set()
    now_iso = datetime.now(timezone.utc).isoformat()

    anon_id = _build_anon_id(request)
    if anon_id:
        rows = _patch_rows(
            "generations",
            filters=(
                f"user_id=eq.{_quote_filter(anon_id)}"
                "&app_user_id=is.null"
            ),
            values={
                "app_user_id": app_user_id,
                "updated_at": now_iso,
            },
        )
        claimed_ids.update(str(row.get("id") or "") for row in rows)
        total += len(rows)

    session_hash = str(getattr(request, "session_hash", "") or "").strip() if request is not None else ""
    if session_hash:
        rows = _patch_rows(
            "generations",
            filters=(
                f"metadata->pending_claim->>session_hash=eq.{_quote_filter(session_hash)}"
                "&app_user_id=is.null"
            ),
            values={
                "app_user_id": app_user_id,
                "updated_at": now_iso,
            },
        )
        if anon_id:
            total += len([row for row in rows if str(row.get("id") or "") not in claimed_ids])
        else:
            total += len(rows)
        claimed_ids.update(str(row.get("id") or "") for row in rows)

    browser_fingerprint = _build_browser_fingerprint(request)
    if browser_fingerprint:
        rows = _patch_rows(
            "generations",
            filters=(
                f"metadata->pending_claim->>browser_fingerprint=eq.{_quote_filter(browser_fingerprint)}"
                "&app_user_id=is.null"
            ),
            values={
                "app_user_id": app_user_id,
                "updated_at": now_iso,
            },
        )
        total += len([row for row in rows if str(row.get("id") or "") not in claimed_ids])

    return total


def persist_generation(
    *,
    request: Any | None,
    generation_id: str,
    qr_mode: str,
    prompt: str,
    text_input: str,
    input_type: str,
    status: str,
    settings_dict: Mapping[str, Any],
    url_normalization: Mapping[str, Any] | None,
    shortener_result: Mapping[str, Any] | None,
    final_image: Image.Image,
) -> str | None:
    user = ensure_app_user_for_request(request)
    anon_id = _build_anon_id(request)
    if user is None and anon_id is None:
        return None

    timestamp = datetime.now(timezone.utc)
    owner_id = str(user["app_user_id"]) if user else str(anon_id)
    object_prefix = f"users/{owner_id}/{timestamp.strftime('%Y/%m/%d')}/{generation_id}"
    completed_image_url = _upload_public_png(
        image=final_image,
        object_path=f"{object_prefix}.png",
    )
    thumbnail_url = _upload_public_png(
        image=final_image,
        object_path=f"{object_prefix}_thumb.png",
        max_size=320,
    )

    normalized_destination_url = None
    if input_type == "URL" and url_normalization:
        normalized_destination_url = str(url_normalization.get("normalized_url") or "").strip() or None

    short_link_id = None
    short_code = None
    short_url = None
    shortener_expires_at = None
    shortener_applied = False
    if shortener_result:
        short_link_id = str(shortener_result.get("short_link_id") or "").strip() or None
        short_code = str(shortener_result.get("code") or "").strip() or None
        short_url = str(shortener_result.get("short_url") or "").strip() or None
        shortener_expires_at = str(shortener_result.get("expires_at") or "").strip() or None
        shortener_applied = bool(shortener_result.get("applied"))

    credit_cost = 4 if str(qr_mode).lower() == "artistic" else 1
    metadata = {
        "settings": dict(settings_dict),
    }
    if user:
        metadata["auth"] = {
            "provider": user["provider"],
            "provider_subject": user["provider_subject"],
            "username": user["username"],
            "display_name": user["display_name"],
            "profile_url": user["profile_url"],
        }
    elif anon_id:
        metadata["pending_claim"] = {
            "anon_id": anon_id,
            "session_hash": str(getattr(request, "session_hash", "") or "") or None,
            "browser_fingerprint": _build_browser_fingerprint(request),
        }

    row = {
        "id": generation_id,
        "user_id": owner_id,
        "app_user_id": user["app_user_id"] if user else None,
        "destination_url": str(text_input or ""),
        "qr_mode": str(qr_mode or "standard"),
        "preset_id": None,
        "prompt": prompt,
        "status": status,
        "credit_cost": credit_cost,
        "input_type": input_type,
        "destination_url_normalized": normalized_destination_url,
        "effective_qr_text": str(settings_dict.get("effective_qr_text") or text_input or ""),
        "shortener_applied": shortener_applied,
        "short_link_id": short_link_id,
        "short_code": short_code,
        "short_url": short_url,
        "shortener_expires_at": shortener_expires_at,
        "completed_image_url": completed_image_url,
        "thumbnail_url": thumbnail_url,
        "metadata": metadata,
        "updated_at": timestamp.isoformat(),
    }
    _insert_row("generations", row)
    if user:
        return f"Saved to My QR Codes for @{user['username']}."
    return "Saved this QR in the current browser session. Open My QR Codes while signed in to claim it."


def _ensure_user_for_request_compat(request: Any | None, username_hint: str | None = None) -> dict[str, Any] | None:
    if username_hint:
        return ensure_app_user_for_request(request, username_hint=username_hint)
    return ensure_app_user_for_request(request)


def list_my_generations(request: Any | None, username_hint: str | None = None) -> dict[str, Any]:
    user = _ensure_user_for_request_compat(request, username_hint=username_hint)
    if user is None:
        session_hash = str(getattr(request, "session_hash", "") or "").strip() if request is not None else ""
        browser_fingerprint = _build_browser_fingerprint(request)
        session_rows_by_id: dict[str, dict[str, Any]] = {}

        def _load_session_rows(filter_expr: str) -> None:
            query = (
                "select=id,app_user_id,created_at,updated_at,status,input_type,qr_mode,prompt,destination_url_original,destination_url_normalized,"
                "effective_qr_text,shortener_applied,short_link_id,short_code,short_url,shortener_expires_at,completed_image_url,thumbnail_url,metadata"
                f"&{filter_expr}"
                "&order=created_at.desc"
            )
            for row in _table_select("dashboard_generation_summaries", query=query):
                row_id = str(row.get("id") or "")
                if row_id and row_id not in session_rows_by_id:
                    session_rows_by_id[row_id] = row

        if session_hash:
            _load_session_rows(f"metadata->pending_claim->>session_hash=eq.{_quote_filter(session_hash)}")
        if browser_fingerprint:
            _load_session_rows(
                f"metadata->pending_claim->>browser_fingerprint=eq.{_quote_filter(browser_fingerprint)}"
            )
        pending_rows = sorted(
            session_rows_by_id.values(),
            key=lambda row: str(row.get("created_at") or ""),
            reverse=True,
        )
        pending_rows = _sync_record_destinations_from_short_links(pending_rows)
        gallery: list[tuple[str, str]] = []
        for row in pending_rows:
            image = str(row.get("thumbnail_url") or row.get("completed_image_url") or "").strip()
            created = str(row.get("created_at") or "")[:16].replace("T", " ")
            label = f"{str(row.get('qr_mode') or 'qr').title()} · {created}"
            gallery.append((image, label))
        if pending_rows:
            owned_count = sum(1 for row in pending_rows if row.get("app_user_id"))
            pending_count = len(pending_rows) - owned_count
            return {
                "authenticated": False,
                "status": (
                    f"Session view: found {len(pending_rows)} QR code(s) from this browser "
                    f"({owned_count} already attached to your account, {pending_count} still pending claim). "
                    "If Hugging Face auth looks broken, you can still browse these here and retry refresh after login."
                ),
                "gallery": gallery,
                "records": pending_rows,
                "user": None,
            }
        return {
            "authenticated": False,
            "status": "Sign in with Hugging Face to see your saved QR codes and analytics.",
            "gallery": [],
            "records": [],
            "user": None,
        }

    query = (
        "select=id,app_user_id,created_at,updated_at,status,input_type,qr_mode,prompt,destination_url_original,destination_url_normalized,"
        "effective_qr_text,shortener_applied,short_link_id,short_code,short_url,shortener_expires_at,completed_image_url,thumbnail_url,metadata"
        f"&app_user_id=eq.{_quote_filter(user['app_user_id'])}"
        "&order=created_at.desc"
    )
    rows = _table_select("dashboard_generation_summaries", query=query)

    session_rows_by_id: dict[str, dict[str, Any]] = {str(row.get("id") or ""): row for row in rows if str(row.get("id") or "")}
    session_hash = str(getattr(request, "session_hash", "") or "").strip() if request is not None else ""
    browser_fingerprint = _build_browser_fingerprint(request)

    def _merge_session_rows(filter_expr: str) -> None:
        session_query = (
            "select=id,app_user_id,created_at,updated_at,status,input_type,qr_mode,prompt,destination_url_original,destination_url_normalized,"
            "effective_qr_text,shortener_applied,short_link_id,short_code,short_url,shortener_expires_at,completed_image_url,thumbnail_url,metadata"
            f"&{filter_expr}"
            "&order=created_at.desc"
        )
        for row in _table_select("dashboard_generation_summaries", query=session_query):
            row_id = str(row.get("id") or "")
            if row_id and row_id not in session_rows_by_id:
                session_rows_by_id[row_id] = row

    if session_hash:
        _merge_session_rows(f"metadata->pending_claim->>session_hash=eq.{_quote_filter(session_hash)}")
    if browser_fingerprint:
        _merge_session_rows(
            f"metadata->pending_claim->>browser_fingerprint=eq.{_quote_filter(browser_fingerprint)}"
        )

    rows = sorted(
        session_rows_by_id.values(),
        key=lambda row: str(row.get("created_at") or ""),
        reverse=True,
    )
    rows = _sync_record_destinations_from_short_links(rows)
    gallery: list[tuple[str, str]] = []
    for row in rows:
        image = str(row.get("thumbnail_url") or row.get("completed_image_url") or "").strip()
        created = str(row.get("created_at") or "")[:16].replace("T", " ")
        label = f"{str(row.get('qr_mode') or 'qr').title()} · {created}"
        gallery.append((image, label))

    display_name = user.get("display_name") or user.get("username")
    return {
        "authenticated": True,
        "status": f"Signed in as {display_name} (@{user['username']}). Found {len(rows)} saved QR code(s).",
        "gallery": gallery,
        "records": rows,
        "user": user,
    }


def _extract_short_link_id(record: Mapping[str, Any]) -> str | None:
    short_link_id = str(record.get("short_link_id") or "").strip()
    if short_link_id:
        return short_link_id

    short_code = str(record.get("short_code") or "").strip()
    if not short_code:
        short_url = str(record.get("short_url") or "").strip()
        if short_url:
            short_code = short_url.rstrip("/").rsplit("/", 1)[-1]
    if not short_code:
        return None

    rows = _table_select(
        "short_links",
        query=f"select=id&code=eq.{_quote_filter(short_code)}&limit=1",
    )
    if not rows:
        return None
    return str(rows[0].get("id") or "").strip() or None


def _sync_record_destinations_from_short_links(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    hydrated_rows = copy.deepcopy(rows)
    short_link_ids = sorted(
        {
            short_link_id
            for short_link_id in (_extract_short_link_id(row) for row in hydrated_rows)
            if short_link_id
        }
    )
    if not short_link_ids:
        return hydrated_rows

    filters = "id=in.(%s)" % ",".join(_quote_filter(short_link_id) for short_link_id in short_link_ids)
    short_link_rows = _table_select(
        "short_links",
        query=(
            "select=id,original_url,normalized_url"
            f"&{filters}"
        ),
    )
    short_link_map = {
        str(row.get("id") or "").strip(): row
        for row in short_link_rows
        if str(row.get("id") or "").strip()
    }
    for row in hydrated_rows:
        short_link_id = _extract_short_link_id(row)
        if not short_link_id:
            continue
        short_link = short_link_map.get(short_link_id)
        if not short_link:
            continue
        original_url = str(short_link.get("original_url") or "").strip()
        normalized_url = str(short_link.get("normalized_url") or "").strip()
        if original_url:
            row["destination_url_original"] = original_url
        if normalized_url:
            row["destination_url_normalized"] = normalized_url
    return hydrated_rows


def _request_matches_record_session(request: Any | None, record: Mapping[str, Any]) -> bool:
    metadata = record.get("metadata")
    metadata_map: Mapping[str, Any] = metadata if isinstance(metadata, Mapping) else {}
    pending_claim = metadata_map.get("pending_claim")
    pending_claim_map: Mapping[str, Any] = pending_claim if isinstance(pending_claim, Mapping) else {}
    request_session_hash = str(getattr(request, "session_hash", "") or "").strip() if request is not None else ""
    record_session_hash = str(pending_claim_map.get("session_hash") or "").strip()
    if request_session_hash and record_session_hash and request_session_hash == record_session_hash:
        return True
    request_browser_fingerprint = str(_build_browser_fingerprint(request) or "").strip()
    record_browser_fingerprint = str(pending_claim_map.get("browser_fingerprint") or "").strip()
    if request_browser_fingerprint and record_browser_fingerprint and request_browser_fingerprint == record_browser_fingerprint:
        return True
    return False


def _authorize_record_access(
    request: Any | None,
    record: Mapping[str, Any],
    *,
    username_hint: str | None = None,
) -> dict[str, Any] | None:
    user = _ensure_user_for_request_compat(request, username_hint=username_hint)
    if user is None:
        if not _request_matches_record_session(request, record):
            raise DashboardError("Please sign in first.")
        return None
    if str(record.get("app_user_id") or "") != str(user.get("app_user_id") or "") and not _request_matches_record_session(request, record):
        raise DashboardError("That QR code does not belong to the current user.")
    return user


_URL_TRACKING_PARAM_PREFIXES = ("utm_", "mc_", "mkt_", "pk_")
_URL_TRACKING_PARAM_NAMES = {
    "fbclid",
    "gclid",
    "dclid",
    "gbraid",
    "wbraid",
    "msclkid",
    "igshid",
    "mibextid",
    "s_cid",
    "si",
    "vero_conv",
    "vero_id",
    "wickedid",
    "yclid",
}


def _should_strip_tracking_param(name: str) -> bool:
    lowered = str(name or "").strip().lower()
    if not lowered:
        return False
    if lowered in _URL_TRACKING_PARAM_NAMES:
        return True
    return lowered.startswith(_URL_TRACKING_PARAM_PREFIXES)


def _normalize_url_for_dashboard(text_input: str) -> dict[str, Any]:
    raw_input = str(text_input or "")
    stripped_input = raw_input.strip()
    result = {
        "original_input": raw_input,
        "normalized_url": stripped_input,
        "normalized_qr_text": stripped_input,
        "tracking_params_removed": 0,
        "chars_saved": max(0, len(raw_input) - len(stripped_input)),
        "changed": stripped_input != raw_input,
    }
    if not stripped_input:
        return result

    parse_target = stripped_input
    if not re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://", parse_target):
        parse_target = f"https://{parse_target.lstrip('/')}"

    parsed = urllib_parse.urlsplit(parse_target)
    if parsed.scheme.lower() not in {"http", "https"}:
        return result

    hostname = (parsed.hostname or "").lower()
    if not hostname:
        return result

    port = parsed.port
    netloc = hostname
    if parsed.username:
        auth = parsed.username
        if parsed.password:
            auth = f"{auth}:{parsed.password}"
        netloc = f"{auth}@{netloc}"
    if port and not (
        (parsed.scheme.lower() == "http" and port == 80)
        or (parsed.scheme.lower() == "https" and port == 443)
    ):
        netloc = f"{netloc}:{port}"

    filtered_query: list[tuple[str, str]] = []
    tracking_params_removed = 0
    for key, value in urllib_parse.parse_qsl(parsed.query, keep_blank_values=True):
        if _should_strip_tracking_param(key):
            tracking_params_removed += 1
            continue
        filtered_query.append((key, value))

    normalized_path = parsed.path or ""
    if normalized_path == "/":
        normalized_path = ""

    normalized_url = urllib_parse.urlunsplit(
        (
            "https",
            netloc,
            normalized_path,
            urllib_parse.urlencode(filtered_query, doseq=True),
            parsed.fragment,
        )
    )
    normalized_qr_text = normalized_url.replace("https://", "", 1)
    chars_saved = max(0, len(raw_input) - len(normalized_url))
    return {
        "original_input": raw_input,
        "normalized_url": normalized_url,
        "normalized_qr_text": normalized_qr_text,
        "tracking_params_removed": tracking_params_removed,
        "chars_saved": chars_saved,
        "changed": normalized_url != raw_input,
    }


def _normalized_hash(normalized_url: str) -> str:
    return hashlib.sha256(str(normalized_url or "").encode("utf-8")).hexdigest()[:32]


def _update_settings_with_new_destination(
    settings: Mapping[str, Any] | None,
    *,
    destination_url: str,
    normalized_url: str,
    normalized_qr_text: str,
    effective_qr_text: str,
    short_url: str | None,
    shortener_expires_at: str | None,
) -> dict[str, Any]:
    updated = copy.deepcopy(dict(settings or {}))
    updated["text_input"] = destination_url
    updated["input_type"] = "URL"
    updated["shortener_applied"] = bool(short_url)
    updated["short_url"] = short_url
    updated["shortener_expires_at"] = shortener_expires_at
    updated["effective_qr_text"] = effective_qr_text
    updated["normalized_qr_text"] = normalized_qr_text
    updated["url_normalization_applied"] = normalized_url != destination_url
    updated["url_tracking_params_removed"] = int(_normalize_url_for_dashboard(destination_url).get("tracking_params_removed") or 0)
    updated["url_chars_saved"] = int(_normalize_url_for_dashboard(destination_url).get("chars_saved") or 0)
    return updated


def update_generation_destination(
    request: Any | None,
    records: list[dict[str, Any]],
    index: int,
    new_destination: str,
    username_hint: str | None = None,
) -> dict[str, Any]:
    if index < 0 or index >= len(records):
        raise DashboardError("Selected QR code is out of range.")

    record = records[index]
    _authorize_record_access(request, record, username_hint=username_hint)

    if str(record.get("input_type") or "") != "URL":
        raise DashboardError("Only URL-based QR codes can be retargeted.")

    short_link_id = _extract_short_link_id(record)
    if not short_link_id:
        raise DashboardError("This QR does not use a temporary short link, so there is no redirect target to update.")

    normalization = _normalize_url_for_dashboard(new_destination)
    destination_url = str(normalization.get("original_input") or "").strip()
    normalized_url = str(normalization.get("normalized_url") or "").strip()
    normalized_qr_text = str(normalization.get("normalized_qr_text") or normalized_url or "").strip()
    if not destination_url or not normalized_url:
        raise DashboardError("Enter a valid destination URL.")

    existing_short_url = str(record.get("short_url") or "").strip() or None
    existing_effective_qr_text = str(record.get("effective_qr_text") or "").strip()
    effective_qr_text = existing_effective_qr_text or normalized_qr_text
    now = datetime.now(timezone.utc)
    expires_at_iso = (now + timedelta(days=7)).isoformat()

    short_link_rows = _patch_rows(
        "short_links",
        filters=f"id=eq.{_quote_filter(short_link_id)}",
        values={
            "original_url": destination_url,
            "normalized_url": normalized_url,
            "normalized_url_hash": _normalized_hash(normalized_url),
            "last_accessed_at": now.isoformat(),
            "expires_at": expires_at_iso,
            "is_active": True,
        },
    )
    if not short_link_rows:
        raise DashboardError("Short link not found or could not be updated.")

    generation_values = {
        "destination_url": destination_url,
        "destination_url_normalized": normalized_url,
        "updated_at": datetime.now(timezone.utc).isoformat(),
    }
    _patch_rows(
        "generations",
        filters=f"short_link_id=eq.{_quote_filter(short_link_id)}",
        values=generation_values,
    )

    generation_rows = _table_select(
        "generations",
        query=f"select=id,metadata&short_link_id=eq.{_quote_filter(short_link_id)}",
    )
    metadata_by_id = {str(row.get("id") or ""): row.get("metadata") for row in generation_rows}

    updated_records = copy.deepcopy(records)
    for idx, row in enumerate(updated_records):
        row_short_link_id = _extract_short_link_id(row)
        if row_short_link_id != short_link_id:
            continue
        row["destination_url_original"] = destination_url
        row["destination_url_normalized"] = normalized_url
        row["effective_qr_text"] = effective_qr_text
        if existing_short_url:
            row["short_url"] = existing_short_url
        if expires_at_iso:
            row["shortener_expires_at"] = expires_at_iso

        metadata = metadata_by_id.get(str(row.get("id") or ""), row.get("metadata"))
        metadata_map: dict[str, Any] = copy.deepcopy(dict(metadata or {})) if isinstance(metadata, Mapping) else {}
        settings = metadata_map.get("settings") if isinstance(metadata_map.get("settings"), Mapping) else {}
        metadata_map["settings"] = _update_settings_with_new_destination(
            settings,
            destination_url=destination_url,
            normalized_url=normalized_url,
            normalized_qr_text=normalized_qr_text,
            effective_qr_text=effective_qr_text,
            short_url=existing_short_url,
            shortener_expires_at=expires_at_iso,
        )
        row["metadata"] = metadata_map
        _patch_rows(
            "generations",
            filters=f"id=eq.{_quote_filter(str(row.get('id') or ''))}",
            values={"metadata": metadata_map},
        )

    detail = get_generation_detail(request, updated_records, index, username_hint=username_hint)
    return {
        "records": updated_records,
        "detail_markdown": detail["detail_markdown"],
        "image": detail["image"],
        "scan_breakdown_rows": detail["scan_breakdown_rows"],
        "settings_json": detail["settings_json"],
        "is_standard": detail["is_standard"],
        "is_artistic": detail["is_artistic"],
        "normalized_url": normalized_url,
        "short_url": existing_short_url,
    }


def build_generation_table_rows(
    records: list[dict[str, Any]],
    *,
    page: int = 1,
    limit: int = 10,
) -> tuple[list[list[Any]], list[int], str, int, int]:
    safe_limit = max(1, int(limit))
    total_records = len(records)
    total_pages = max(1, math.ceil(total_records / safe_limit))
    safe_page = min(max(1, int(page)), total_pages)
    start_idx = (safe_page - 1) * safe_limit
    end_idx = start_idx + safe_limit
    visible_records = list(records[start_idx:end_idx])
    row_map = list(range(start_idx, start_idx + len(visible_records)))
    table_rows: list[list[Any]] = []
    for position, record in enumerate(visible_records, start=start_idx + 1):
        created = str(record.get("created_at") or "")[:16].replace("T", " ")
        qr_mode = str(record.get("qr_mode") or "qr").title()
        destination = str(record.get("destination_url_original") or "—")
        short_url = str(record.get("short_url") or "—")
        status = str(record.get("status") or "unknown")
        table_rows.append([position, created, qr_mode, destination, short_url, status])
    if total_records:
        shown_start = start_idx + 1
        shown_end = start_idx + len(visible_records)
        range_label = f"Showing {shown_start}-{shown_end} of {total_records}"
    else:
        range_label = "Showing 0-0 of 0"
    summary = (
        f"{range_label} saved QR code(s). "
        f"Page {safe_page} of {total_pages}. Click a row to inspect it."
    )
    return table_rows, row_map, summary, safe_page, total_pages


def resolve_generation_index_from_table_event(
    row_map: list[int],
    event_index: Any,
) -> int | None:
    row_idx = event_index
    if isinstance(event_index, (tuple, list)):
        row_idx = event_index[0] if event_index else None
    if row_idx is None:
        return None
    try:
        row_number = int(row_idx)
    except (TypeError, ValueError):
        return None
    if row_number < 0 or row_number >= len(row_map):
        return None
    return int(row_map[row_number])


def get_generation_detail(
    request: Any | None,
    records: list[dict[str, Any]],
    index: int,
    username_hint: str | None = None,
) -> dict[str, Any]:
    if index < 0 or index >= len(records):
        raise DashboardError("Selected QR code is out of range.")

    record = records[index]
    _authorize_record_access(request, record, username_hint=username_hint)

    short_link_id = _extract_short_link_id(record)
    scan_rows: list[dict[str, Any]] = []
    if short_link_id:
        scan_rows = _table_select(
            "short_link_scan_events",
            query=(
                "select=scanned_at,country_code,visitor_hash,is_bot,is_prefetch"
                f"&short_link_id=eq.{_quote_filter(short_link_id)}"
                "&order=scanned_at.desc"
                "&limit=1000"
            ),
        )

    created_at_dt = _parse_timestamp(record.get("created_at"))
    filtered_scan_rows = [
        row
        for row in scan_rows
        if not bool(row.get("is_bot"))
        and not bool(row.get("is_prefetch"))
        and (
            created_at_dt is None
            or (_parse_timestamp(row.get("scanned_at")) or created_at_dt) >= created_at_dt
        )
    ]

    total_scans = len(filtered_scan_rows)
    last_scanned_at = (
        str(filtered_scan_rows[0].get("scanned_at") or "").replace("T", " ")[:19]
        if filtered_scan_rows
        else "Never"
    )
    unique_scans = len(
        {
            str(row.get("visitor_hash") or "")
            for row in filtered_scan_rows
            if str(row.get("visitor_hash") or "").strip()
        }
    )

    scan_breakdown = Counter()
    for row in filtered_scan_rows:
        scanned_at_raw = str(row.get("scanned_at") or "")
        scanned_at_dt = _parse_timestamp(scanned_at_raw) or created_at_dt or datetime.now(timezone.utc)
        country = str(row.get("country_code") or "Unknown")
        day = scanned_at_dt.strftime("%Y-%m-%d")
        hour = scanned_at_dt.strftime("%H:00")
        scan_breakdown[(country, day, hour)] += 1

    scan_breakdown_rows = [
        [country, day, hour, scan_breakdown[(country, day, hour)]]
        for country, day, hour in sorted(scan_breakdown.keys(), key=lambda item: (item[1], item[2], item[0]))
    ]

    metadata = record.get("metadata")
    metadata_map: Mapping[str, Any] = metadata if isinstance(metadata, Mapping) else {}
    raw_settings = metadata_map.get("settings")
    settings: Mapping[str, Any] = raw_settings if isinstance(raw_settings, Mapping) else {}
    settings_json = json.dumps(settings, indent=2, ensure_ascii=False) if settings else "{}"

    detail_markdown = "\n".join(
        [
            f"### {str(record.get('qr_mode') or 'QR').title()} QR",
            f"- Created: {str(record.get('created_at') or '').replace('T', ' ')[:19]}",
            f"- Status: {record.get('status') or 'unknown'}",
            f"- Input type: {record.get('input_type') or 'unknown'}",
            f"- Destination: {record.get('destination_url_original') or '—'}",
            f"- Short URL: {record.get('short_url') or '—'}",
            f"- Effective QR text: {record.get('effective_qr_text') or '—'}",
            f"- Total scans since this QR was created: {total_scans}",
            f"- Approx. unique scans since creation: {unique_scans}",
            f"- Last scanned at: {last_scanned_at}",
        ]
    )

    return {
        "record": record,
        "detail_markdown": detail_markdown,
        "image": record.get("completed_image_url") or record.get("thumbnail_url"),
        "scan_breakdown_rows": scan_breakdown_rows,
        "settings_json": settings_json,
        "is_standard": str(record.get("qr_mode") or "").lower() == "standard",
        "is_artistic": str(record.get("qr_mode") or "").lower() == "artistic",
    }