Blame for app/auth.py

"""Optional Supabase sign-in and math-class storage."""
from __future__ import annotations

import json
import logging
import os
import re
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
import time
import urllib.error
import urllib.parse
import urllib.request

log = logging.getLogger("mathly.auth")
TIMEOUT = 10
CACHE_TTL = 300
CLASSES_TTL = 60


class AuthUnavailable(RuntimeError):
    pass


class ClassConflict(ValueError):
    pass


class SupabaseAuth:
    def __init__(self, url: str, key: str, teachers: frozenset[str] = frozenset(), secret: str = "",
                 developers: frozenset[str] = frozenset()):
        self.url, self.key, self.secret = url.rstrip("/"), key, secret
        self.teachers, self.developers = teachers, developers
        self._users: dict[str, tuple[float, dict]] = {}
        self._classes: tuple[float, list[dict]] | None = None
        self._access: dict[str, tuple[float, dict | None]] = {}
        self._lock = threading.Lock()

    def user(self, token: str) -> dict | None:
        now = time.monotonic()
        with self._lock:
            cached = self._users.get(token)
        if cached and cached[0] > now:
            return cached[1]
        request = urllib.request.Request(self.url + "/auth/v1/user",
            headers={"Authorization": f"Bearer {token}", "apikey": self.key, "User-Agent": "mathly/1.0"})
        try:
            with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
                data = json.load(response)
        except urllib.error.HTTPError as exc:
            if exc.code in (400, 401, 403):
                return None
            raise AuthUnavailable(f"supabase answered {exc.code}") from exc
        except (urllib.error.URLError, TimeoutError, ValueError) as exc:
            raise AuthUnavailable(str(exc)) from exc
        if not data.get("id"):
            return None
        user = {"id": data["id"], "email": data.get("email", "")}
        with self._lock:
            self._users[token] = (now + CACHE_TTL, user)
        return user

    def _admin(self, path: str, method: str = "GET", body=None, prefer: str = "return=representation"):
        if not self.secret:
            raise AuthUnavailable("SUPABASE_SECRET_KEY is not set")
        request = urllib.request.Request(self.url + path, method=method,
            data=json.dumps(body).encode() if body is not None else None,
            headers={"Authorization": f"Bearer {self.secret}", "apikey": self.secret,
                     "Content-Type": "application/json", "Prefer": prefer, "User-Agent": "mathly/1.0"})
        try:
            with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
                return json.loads(response.read() or b"null")
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode(errors="replace")[:300]
            if exc.code == 409:
                raise ClassConflict(detail) from exc
            raise AuthUnavailable(f"supabase answered {exc.code}: {detail}") from exc
        except (urllib.error.URLError, TimeoutError, ValueError) as exc:
            raise AuthUnavailable(str(exc)) from exc

    def classes(self, fresh: bool = False) -> list[dict]:
        now = time.monotonic()
        with self._lock:
            cached = self._classes
        if cached and cached[0] > now and not fresh:
            return cached[1]
        rows: list[dict] = []
        if self.secret:
            try:
                rows = self._admin("/rest/v1/classes?select=*&order=created_at.desc") or []
            except AuthUnavailable as exc:
                if "answered 404" not in str(exc):
                    raise
                log.warning("the classes table is missing; every account is allowed")
        for row in rows:
            row["emails"] = [str(value).lower() for value in row.get("emails") or []]
            row["teachers"] = [str(value).lower() for value in row.get("teachers") or []]
            row["lessons"] = row.get("lessons") or []
            row["nicknames"] = row.get("nicknames") or {}
            row["exams"] = row.get("exams") or []
            row["exam_note"] = str(row.get("exam_note") or "")
            row["exam_title"] = str(row.get("exam_title") or "")
        with self._lock:
            self._classes = (now + CLASSES_TTL, rows)
        return rows

    def save_class(self, class_id: int | None, name: str, emails: list[str], nicknames: dict[str, str],
                   lessons: list[str], *, teachers: list[str] | None = None, **_ignored) -> None:
        body = {"name": name, "emails": emails, "nicknames": nicknames, "lessons": lessons,
                "teachers": list(teachers or [])}
        if class_id is None:
            self._admin("/rest/v1/classes", "POST", body)
        elif not self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "PATCH", body):
            raise KeyError(class_id)
        self._classes = None

    def save_class_exams(self, class_id: int, exams: list[dict]) -> None:
        """The class's exam board (each exam: subject, date, and the units and note it may name)."""
        if not self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "PATCH", {"exams": exams}):
            raise KeyError(class_id)
        self._classes = None

    def save_class_settings(self, class_id: int, fields: dict) -> None:
        """A class's AI settings (app/ai_batches.py): automatic practice checks, and its daily AI budget."""
        if not self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "PATCH", fields):
            raise KeyError(class_id)
        self._classes = None

    def delete_class(self, class_id: int) -> None:
        self._admin(f"/rest/v1/classes?id=eq.{int(class_id)}", "DELETE")
        self._classes = None

    def ping(self, timeout: float = 8) -> dict:
        """How Supabase answers right now, for Monitoring: its sign-in service and its database, each timed in ms.
        A part that fails holds {"ok": False, "error"} and does not hide the other."""
        def timed(path: str, key: str) -> dict:
            request = urllib.request.Request(self.url + path, headers={
                "apikey": key, "Authorization": f"Bearer {key}", "User-Agent": "mathly/1.0"})
            start = time.monotonic()
            try:
                with urllib.request.urlopen(request, timeout=timeout) as response:
                    response.read()
                return {"ok": True, "ms": round((time.monotonic() - start) * 1000)}
            except urllib.error.HTTPError as exc:
                return {"ok": False, "ms": round((time.monotonic() - start) * 1000), "error": f"HTTP {exc.code}"}
            except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc:
                return {"ok": False, "error": str(exc) or type(exc).__name__}
        with ThreadPoolExecutor(2) as pool:
            auth = pool.submit(timed, "/auth/v1/health", self.key)
            db = pool.submit(timed, "/rest/v1/classes?select=id&limit=1", self.secret) if self.secret else None
            return {"auth": auth.result(),
                    "db": db.result() if db else {"ok": False, "error": "SUPABASE_SECRET_KEY is not set"}}

    def answer_activity(self, since: datetime, limit: int = 10_000) -> dict | None:
        """Open-answer attempts and photo checks since a time: user, time and verdict only, never the answer.
        None when the answer tables are not installed."""
        stamp = urllib.parse.quote(since.isoformat())
        try:
            attempts = self._admin(f"/rest/v1/math_question_attempts?select=user_id,question_id,created_at"
                                   f"&created_at=gte.{stamp}&order=created_at.asc&limit={limit}") or []
        except AuthUnavailable as exc:
            if "answered 404" in str(exc):
                return None
            raise
        try:
            grades = self._admin(f"/rest/v1/math_answer_grades?select=user_id,state,created_at,verdict:result->>verdict"
                                 f"&created_at=gte.{stamp}&order=created_at.asc&limit={limit}") or []
        except AuthUnavailable as exc:
            if "answered 404" not in str(exc):
                raise
            grades = []
        return {"attempts": attempts, "grades": grades}

    def user_accounts(self, emails: list[str]) -> dict[str, dict]:
        """The id and Google name of each email that has signed in at least once (Supabase's admin user list), by email.
        The name is "" when the account has none."""
        wanted, found = {email.lower() for email in emails}, {}
        page = 1
        while wanted - found.keys():
            data = self._admin(f"/auth/v1/admin/users?page={page}&per_page=1000") or {}
            users = data.get("users") if isinstance(data, dict) else data
            for user in users or []:
                email = str(user.get("email") or "").lower()
                if email in wanted and user.get("id"):
                    meta = user.get("user_metadata") if isinstance(user.get("user_metadata"), dict) else {}
                    name = " ".join(str(meta.get("full_name") or meta.get("name") or "").split())
                    found[email] = {"id": user["id"], "name": name}
            if len(users or []) < 1000 or page >= 50:
                break
            page += 1
        return found

    def user_email(self, user_id: str) -> str:
        """The email of one account (Supabase's admin API), "" when it has none."""
        data = self._admin(f"/auth/v1/admin/users/{user_id}") or {}
        return str(data.get("email") or "").lower() if isinstance(data, dict) else ""

    def user_ids(self, emails: list[str]) -> dict[str, str]:
        """The account id of each email that has signed in at least once, by email."""
        return {email: account["id"] for email, account in self.user_accounts(emails).items()}

    def access_for(self, user_id: str, fresh: bool = False) -> dict | None:
        if not self.secret:
            return None
        now = time.monotonic()
        with self._lock:
            cached = self._access.get(user_id)
        if cached and cached[0] > now and not fresh:
            return cached[1]
        try:
            rows = self._admin("/rest/v1/user_access?" + urllib.parse.urlencode(
                {"user_id": f"eq.{user_id}", "select": "user_id,email,roles,updated_at"}))
            value = rows[0] if rows else None
        except AuthUnavailable as exc:
            if "answered 404" not in str(exc):
                raise
            value = None
        with self._lock:
            self._access[user_id] = (now + CLASSES_TTL, value)
        return value


def load_auth() -> SupabaseAuth | None:
    url = os.environ.get("SUPABASE_URL", "").strip()
    key = os.environ.get("SUPABASE_PUBLISHABLE_KEY", "").strip()
    if not url or not key:
        log.info("sign-in is disabled; the math path is public")
        return None
    emails = lambda name: frozenset(re.split(r"[\s,;]+", os.environ.get(name, "").strip().lower())) - {""}
    return SupabaseAuth(url, key, emails("TEACHER_EMAILS"), os.environ.get("SUPABASE_SECRET_KEY", "").strip(),
                        emails("DEVELOPER_EMAILS"))