"""app.py — ani-web: FastAPI over upstream.py + HLS proxy + static UI.

Browser playback: hls.anidb.app sends no CORS headers, so the player is
pointed at /stream/{token}; the proxy fetches upstream (with the same
Cloudflare workaround) and rewrites playlist URIs to go through itself.
"""
import base64
import json
import os
import threading
import time
from collections import Counter, OrderedDict
from datetime import datetime
from pathlib import Path
from base64 import urlsafe_b64encode, urlsafe_b64decode
from urllib.parse import quote, urljoin, urlparse

from fastapi import Body, FastAPI, HTTPException, Query, Request
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response, StreamingResponse
from fastapi.staticfiles import StaticFiles
from itsdangerous import BadSignature, TimestampSigner

from . import db, upstream, users

ROOT = Path(__file__).parent.parent
app = FastAPI(title="ani-web")


def _cdn_probe():
    """Background probe: pull 64KB from the CDN every 3 min and log state
    flips, so server.log shows when a degraded window ends (CDN RECOVERED).
    Probe target: any valid episode (36019 = JJK S2 E25).
    ponytail: single fixed episode; if it ever disappears, the probe logs
    DEGRADED forever — swap the ID, don't build a picker."""
    def first(pl, base):
        l = next(x.strip() for x in pl.splitlines()
                 if x.strip() and not x.startswith("#"))
        return l if l.startswith("http") else base + l

    state = None
    while True:
        try:
            m = upstream.master_m3u8(36019, "jpn")
            if not m:
                raise RuntimeError("no master m3u8")
            base = m.rsplit("/", 1)[0] + "/"
            v = first(upstream._get(m, timeout=10).text, base)
            seg = first(upstream._get(v, timeout=10).text, base)
            t = time.time()
            r = upstream._session().get(seg, headers={"Range": "bytes=0-65535"},
                                        impersonate="chrome", timeout=10)
            kbps = len(r.content) / 1024 / max(time.time() - t, 0.01)
        except Exception:
            kbps = 0.0
        now = "OK" if kbps >= 40 else "DEGRADED"
        if now != state:
            print(f"[probe] CDN {now} ({kbps:.0f} KB/s)", flush=True)
        state = now
        time.sleep(180)


threading.Thread(target=_cdn_probe, daemon=True).start()

# ---------- auth: multi-user, signed cookies ----------
users.migrate_legacy()  # file-based legacy import; db.init() absorbs its output
db.init()
_SIGNER = TimestampSigner(users.get_secret())
_SESSION = "aniweb_session"
_SESSION_AGE = 30 * 86400
_PUBLIC = {("/login", "GET"), ("/login", "POST")}
_SECURE = os.environ.get("ANIWEB_SECURE") == "1"  # set when served over HTTPS

# ponytail: in-memory lockout per username, single process — 5 fails -> 15 min
_lockout: dict = {}


def _session_user(request) -> str | None:
    try:
        payload = _SIGNER.unsign(request.cookies.get(_SESSION, ""), max_age=_SESSION_AGE)
    except BadSignature:
        return None
    return payload.decode().split(":", 1)[1] if payload.startswith(b"user:") else None


def _lock(user: str, pw: str) -> str:
    lo = _lockout.setdefault(user, {"fails": 0, "until": 0.0})
    if time.time() < lo["until"]:
        return "locked"
    if users.verify(user, pw):
        lo.update(fails=0, until=0.0)
        return "ok"
    lo["fails"] += 1
    if lo["fails"] >= 5:
        lo.update(fails=0, until=time.time() + 900)
    time.sleep(0.5)  # slow down brute force
    return "bad"


@app.middleware("http")
async def csrf_header(request, call_next):
    # ponytail: a cross-site form POST (top-level navigation) does carry the
    # SameSite=Lax cookie, so <form action=.../account/delete> from a
    # malicious page would work; forms can't set a custom header, the SPA
    # always sends it. Runs *after* require_auth (registered earlier), so
    # unauthenticated requests still get 302/401 first.
    if request.method in ("POST", "DELETE") and request.url.path != "/login":
        if request.headers.get("x-aniweb") != "1":
            return JSONResponse({"detail": "forbidden"}, status_code=403)
    return await call_next(request)


@app.middleware("http")
async def require_auth(request, call_next):
    if (request.url.path, request.method) in _PUBLIC:
        return await call_next(request)
    user = _session_user(request)
    # stateless cookie -> re-verify the account still exists, so
    # deleted users' cookies die immediately
    if user is not None and user not in users.load():
        user = None
    if not user:
        if request.url.path.startswith("/api/"):
            return JSONResponse({"detail": "not authenticated"}, status_code=401)
        next_ = request.url.path + (f"?{request.url.query}" if request.url.query else "")
        return RedirectResponse(f"/login?next={quote(next_, safe='')}", status_code=302)
    request.state.user = user
    return await call_next(request)


def _session_cookie(resp, user: str):
    resp.set_cookie(_SESSION, _SIGNER.sign(b"user:" + user.encode()).decode(),
                    max_age=_SESSION_AGE, httponly=True, samesite="lax", secure=_SECURE)


@app.get("/login", include_in_schema=False)
def login_page():
    return FileResponse(ROOT / "static" / "login.html")


@app.post("/login")
async def login(payload: dict = Body(...)):
    u = payload.get("username", "").lower()
    if not u:
        return JSONResponse({"detail": "username required"}, status_code=400)
    state = _lock(u, payload.get("password", ""))
    if state == "ok":
        resp = JSONResponse({"ok": True, "user": u})
        _session_cookie(resp, u)
        return resp
    if state == "locked":
        return JSONResponse({"detail": "too many attempts, try later"}, status_code=429)
    return JSONResponse({"detail": "wrong username or password"}, status_code=401)


@app.get("/api/me", include_in_schema=False)
def api_me(request: Request):
    return {"user": request.state.user}


@app.post("/account/delete")
def delete_account(request: Request):
    users.remove(request.state.user)
    resp = JSONResponse({"ok": True})
    resp.delete_cookie(_SESSION, secure=_SECURE)
    return resp


@app.post("/logout")
def logout():
    resp = JSONResponse({"ok": True})
    resp.delete_cookie(_SESSION, secure=_SECURE)
    return resp


def _load_hist(user: str):
    c = db.connect()
    rows = c.execute("SELECT payload FROM history WHERE user=? ORDER BY t DESC",
                     (user,)).fetchall()
    c.close()
    return [json.loads(r[0]) for r in rows]


@app.get("/api/history")
def api_history(request: Request):
    # no-store: this JSON mutates while a tab is open (position saves);
    # a stale copy has caused "poster gone" confusion before.
    # ?all=1: full list for the /history page (no top-5 cap)
    entries = sorted(_load_hist(request.state.user), key=lambda e: e.get("t", 0), reverse=True)
    if request.query_params.get("all") != "1":
        entries = entries[:5]
    return JSONResponse(entries, headers={"Cache-Control": "no-store"})


@app.delete("/api/history")
def api_history_delete(request: Request, slug_id: str, ep_id: str):
    # ep_id is a stable INTEGER in the schema (ghost rows died in the import),
    # so a strict match suffices
    try:
        ep = int(ep_id)
    except ValueError:
        return {"removed": 0}
    c = db.connect()
    cur = c.execute("DELETE FROM history WHERE user=? AND slug_id=? AND ep_id=?",
                    (request.state.user, slug_id, ep))
    c.commit()
    c.close()
    return {"removed": cur.rowcount}


@app.post("/api/history")
def api_history_post(entry: dict = Body(...), request: Request = None):
    if "ep_id" not in entry:
        raise HTTPException(422, "ep_id required")
    # normalize: ghost rows were born when ep_id arrived as str against an
    # int row (or vice versa) — int() makes the stored type stable
    try:
        entry["ep_id"] = int(entry["ep_id"])
    except (TypeError, ValueError):
        raise HTTPException(422, "ep_id must be an integer")
    entry["t"] = time.time()
    u = request.state.user
    c = db.connect()
    # upsert: the (user, slug_id, ep_id) PK *is* the merge key, so no
    # read-modify-write (and no lost-update race between the read and write)
    c.execute("INSERT INTO history (user, slug_id, ep_id, t, payload) VALUES (?,?,?,?,?) "
              "ON CONFLICT(user, slug_id, ep_id) DO UPDATE SET "
              "t=excluded.t, payload=excluded.payload",
              (u, str(entry.get("slug_id")), entry["ep_id"], entry["t"], json.dumps(entry)))
    # cap: newest 50 per user
    c.execute("DELETE FROM history WHERE user=? AND rowid NOT IN "
              "(SELECT rowid FROM history WHERE user=? ORDER BY t DESC LIMIT 50)", (u, u))
    db.event(c, u, "position", entry)  # metrics: watch-time trajectory
    c.commit()
    c.close()
    return entry


def _load_searches(user: str):
    c = db.connect()
    rows = c.execute("SELECT q FROM searches WHERE user=? ORDER BY t DESC", (user,)).fetchall()
    c.close()
    return [r[0] for r in rows]


@app.get("/api/searches")
def api_searches(request: Request):
    return JSONResponse(_load_searches(request.state.user),
                         headers={"Cache-Control": "no-store"})


@app.post("/api/searches")
def api_searches_post(item: dict = Body(...), request: Request = None):
    q = str(item.get("q", "")).strip()[:60]
    if not q:
        raise HTTPException(422, "q required")
    u = request.state.user
    # NOCASE PK: re-posting a seen query (any casing) conflicts on the row,
    # so the upsert rewrites q to the newest casing, then caps at 8
    c = db.connect()
    c.execute("INSERT INTO searches (user, q, t) VALUES (?,?,?) "
              "ON CONFLICT(user, q) DO UPDATE SET q=excluded.q, t=excluded.t",
              (u, q, time.time()))
    c.execute("DELETE FROM searches WHERE user=? AND rowid NOT IN "
              "(SELECT rowid FROM searches WHERE user=? ORDER BY t DESC LIMIT 8)", (u, u))
    db.event(c, u, "search", {"q": q})
    c.commit()
    c.close()
    return {"ok": True}


@app.post("/api/events")
def api_events(evt: dict = Body(...), request: Request = None):
    # client-reported events; only 'error' is accepted (playback failures
    # happen in the browser, not in the request log)
    if evt.get("type") != "error":
        raise HTTPException(422, "unknown event type")
    c = db.connect()
    db.event(c, request.state.user, "error",
             {k: evt[k] for k in ("ep_id", "ep", "lang") if k in evt}
             | {"detail": str(evt.get("detail", ""))[:200]})
    c.commit()
    c.close()
    return {"ok": True}


@app.get("/api/stats")
def api_stats(request: Request, days: int = Query(30, ge=1, le=365),
              slug: str = Query(None)):
    u = request.state.user
    since = time.time() - days * 86400
    c = db.connect()
    if slug:
        positions = c.execute("SELECT ts, payload FROM events WHERE user=? AND type='position' "
                              "AND ts>=? AND json_extract(payload, '$.slug_id')=? ORDER BY ts",
                              (u, since, slug)).fetchall()
    else:
        positions = c.execute("SELECT ts, payload FROM events WHERE user=? AND type='position' "
                              "AND ts>=? ORDER BY ts", (u, since)).fetchall()
    plays = [r[0] for r in c.execute("SELECT payload FROM events WHERE user=? "
                                     "AND type='play' AND ts>=?", (u, since))]
    searches = [json.loads(r[0]).get("q") for r in c.execute(
        "SELECT payload FROM events WHERE user=? AND type='search' AND ts>=? "
        "ORDER BY ts", (u, since))]
    n_errors = c.execute("SELECT COUNT(*) FROM events WHERE user=? AND type='error' "
                         "AND ts>=?", (u, since)).fetchone()[0]
    day_set = [r[0] for r in c.execute(
        "SELECT DISTINCT CAST(ts/86400 AS INTEGER) FROM events WHERE user=? AND ts>=? "
        "ORDER BY 1", (u, since))]
    c.close()
    days_active = len(day_set)
    best = streak = 0  # max consecutive-day run (streak)
    for i, d in enumerate(day_set):
        streak = streak + 1 if i and d == day_set[i - 1] + 1 else 1
        best = max(best, streak)

    # ponytail: watch time = sum of positive pos deltas <= 90s per (anime, ep).
    # position events fire ~10s apart; a bigger jump is a seek/resume, not
    # viewing; a backward jump >60s within an ep counts as a rewatch
    eps = {}
    watch_sec = [0.0] * 7  # Mon..Sun
    by_day = {}  # local YYYY-MM-DD -> seconds
    for ts, payload in positions:
        e = json.loads(payload)
        k = (str(e.get("slug_id")), e.get("season_id"), e.get("ep_id"))
        s = eps.setdefault(k, {"title": e.get("title"), "poster": e.get("poster"),
                               "total": e.get("total"), "last_ts": 0.0, "last_ep": None,
                               "pos": None, "max": 0.0, "dur": 0.0, "lang": e.get("lang"),
                               "secs": 0.0, "rewinds": 0, "jpn": 0.0, "eng": 0.0,
                               "days": set()})
        p = e.get("pos") or 0
        s["days"].add(ts // 86400)
        s["last_ts"] = ts
        if e.get("ep") is not None:
            s["last_ep"] = e["ep"]
        if e.get("poster"):
            s["poster"] = e["poster"]
        if e.get("total"):
            s["total"] = e["total"]
        if s["pos"] is not None and 0 < p - s["pos"] <= 90:
            d = p - s["pos"]
            s["secs"] += d
            s["eng" if e.get("lang") == "eng" else "jpn"] += d
            dt = datetime.fromtimestamp(ts)
            watch_sec[dt.weekday()] += d
            k = dt.strftime("%Y-%m-%d")
            by_day[k] = by_day.get(k, 0) + d
        elif s["pos"] is not None and p < s["pos"] - 60:
            s["rewinds"] += 1
        s["pos"] = p
        s["max"] = max(s["max"], p)
        s["dur"] = e.get("dur") or s["dur"]
        if e.get("lang"):
            s["lang"] = e["lang"]

    per_anime, started, completed, on_hold = {}, 0, 0, 0
    for (slug, _season, _ep), s in eps.items():
        a = per_anime.setdefault(slug, {"title": s["title"], "poster": s["poster"],
                                        "total": s["total"], "last_seen": 0.0,
                                        "cur_ep": s["last_ep"], "minutes": 0.0,
                                        "started": 0, "completed": 0, "rewatches": 0,
                                        "sub": 0.0, "dub": 0.0, "days": set()})
        a["minutes"] += s["secs"] / 60
        a["started"] += 1
        a["rewatches"] += s["rewinds"]
        a["sub"] += s["jpn"]
        a["dub"] += s["eng"]
        a["days"] |= s["days"]
        a["poster"] = a["poster"] or s["poster"]
        a["total"] = a["total"] or s["total"]
        a["title"] = a["title"] or s["title"]
        started += 1
        if s["last_ts"] > a["last_seen"]:
            a["last_seen"] = s["last_ts"]
            a["cur_ep"] = s["last_ep"]
        if s["dur"] > 0 and s["max"] >= s["dur"] * 0.95:
            a["completed"] += 1
            completed += 1
    now = time.time()
    for a in per_anime.values():
        a["pace"] = round(a["started"] / len(a["days"]), 1)
        # ponytail: finished = every started ep completed and (total unknown,
        # or all eps done) — heuristic, fine for a v1 chip
        fin = a["completed"] == a["started"] and (a["total"] is None
                                                  or a["completed"] >= a["total"])
        a["status"] = ("finished" if fin else
                       "ongoing" if now - a["last_seen"] < 7 * 86400 else "hold")
        if a["status"] == "hold":
            on_hold += 1
        a["minutes"] = round(a["minutes"])
        a["sub"], a["dub"] = int(a["sub"]), int(a["dub"])
    top = sorted(per_anime.values(), key=lambda a: -a["minutes"])[:10]

    plays_by_lang = Counter(json.loads(p).get("lang", "jpn") for p in plays)
    if not plays_by_lang:  # pre-`play`-event data: fall back to position lang
        plays_by_lang = Counter(s["lang"] for s in eps.values() if s["lang"])

    return {
        "days": days,
        "watch_minutes": round(sum(s["secs"] for s in eps.values()) / 60),
        "episodes": {"started": started, "completed": completed},
        "days_active": days_active,
        "best_streak": best,
        "on_hold": on_hold,
        "plays": len(plays),
        "errors": n_errors,
        "by_weekday": [int(x) for x in watch_sec],  # seconds, Mon..Sun
        "by_day": {k: int(v) for k, v in by_day.items()},  # seconds per local day
        "top_anime": top,
        "top_searches": [q for q, _ in Counter(searches).most_common(5) if q],
        "sub_dub": {"jpn": plays_by_lang.get("jpn", 0), "eng": plays_by_lang.get("eng", 0)},
    }


def _b64(url: str) -> str:
    return base64.urlsafe_b64encode(url.encode()).decode()


def _url(token: str) -> str:
    return base64.urlsafe_b64decode(token.encode()).decode()


def _upstream(fn, *a, **kw):
    try:
        return fn(*a, **kw)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(502, f"upstream error: {e}")


@app.middleware("http")
async def no_cache(request, call_next):
    # ponytail: no-store on everything except /stream (immutable HLS data,
    # cached per upstream URL) and the vendored/static JS+CSS (re-downloaded
    # on every page load hurt on mobile)
    r = await call_next(request)
    p = request.url.path
    if p == "/hls.min.js":
        r.headers["Cache-Control"] = "public, max-age=2592000"
    elif p in ("/app.js", "/style.css"):
        r.headers["Cache-Control"] = "public, max-age=300"
    elif not p.startswith("/stream"):
        r.headers["Cache-Control"] = "no-store"
    return r


@app.get("/api/search")
def api_search(q: str = Query(min_length=1)):
    return _upstream(upstream.search, q)


@app.get("/api/anime/{slug_id}")
def api_anime(slug_id: str):
    return _upstream(upstream.details, slug_id)


@app.get("/api/intros/{mal_id}")
def api_intros(mal_id: int):
    # keyless Jikan/MAL intro timings; Jikan down or no data = no skip
    # button, never an error that could touch the player
    try:
        return upstream.intros(mal_id)
    except Exception:
        return []


@app.get("/api/anime/{anime_id}/episodes")
def api_episodes(anime_id: int):
    return _upstream(upstream.episodes, anime_id)


@app.get("/api/episode/{ep_id}/stream")
def api_stream(ep_id: int, lang: str = "jpn", request: Request = None):
    def go():
        master = _upstream(upstream.master_m3u8, ep_id, lang)
        if not master:
            raise HTTPException(404, f"no {lang} stream for episode {ep_id}")
        return {"token": _b64(master)}

    try:
        out = go()
    except HTTPException as e:
        # one-shot retry with a cold cache: recovers rotated/expired
        # playlist URLs from the TTL cache
        if e.status_code == 502:
            upstream.clear_cache()
            out = go()
        else:
            c = db.connect()
            db.event(c, request.state.user, "error",
                     {"ep_id": ep_id, "lang": lang, "detail": str(e.detail)})
            c.commit()
            c.close()
            raise
    c = db.connect()
    db.event(c, request.state.user, "play", {"ep_id": ep_id, "lang": lang})
    c.commit()
    c.close()
    return out


def _rewrite_m3u8(text: str, base: str) -> str:
    # ponytail: rewrites plain URIs only; URI-list attributes in
    # #EXT-X-* tags (e.g. SUBTITLES) are ignored — fine for video-only playlists
    return "\n".join(
        line if line.startswith("#") or not line
        else "/stream/" + _b64(urljoin(base, line))
        for line in text.splitlines()) + "\n"


_STREAM_CACHE = OrderedDict()  # ponytail: LRU of proxied HLS data (immutable);
_STREAM_MAX = 24                # segments ~1-2MB -> <= ~48MB
# ponytail: HLS VOD content is immutable and the token IS the exact upstream
# URL, so per-URL browser caching can't go stale — re-opens, re-seeks and
# auto-next come from the disk cache instead of proxy + CDN
_STREAM_CACHE_HDRS = {"Cache-Control": "public, max-age=86400"}


def _cache_put(url, ct, body):
    _STREAM_CACHE[url] = (ct, body)
    while len(_STREAM_CACHE) > _STREAM_MAX:
        _STREAM_CACHE.popitem(last=False)


def _seg_stream(r, url, ct):
    """Forward the CDN's bytes as they arrive; keep the LRU warm for
    completed downloads (a truncated one would poison the cache)."""
    import time
    t0, t_first, buf, complete = time.time(), None, [], False
    try:
        for c in r.iter_content(chunk_size=256 * 1024):
            if t_first is None:
                t_first = time.time()
            buf.append(c)
            yield c
        complete = True
    finally:
        r.close()
        if complete and buf:
            _cache_put(url, ct, b"".join(buf))
        # observability: a slow segment is the fragLoadTimeOut culprit. Log
        # only the slow ones so normal playback stays quiet. TTFB>3s or
        # total>8s means the CDN (or our upstream hop) stalled — that's what
        # the browser times out on. If these show ~0s but the user still
        # fails, the stall is in *their* path (HTTPS proxy / network), not ours.
        ttfb = (t_first - t0) if t_first is not None else None
        total = time.time() - t0
        if (ttfb is not None and ttfb > 3) or total > 8:
            name = url.rsplit("/", 1)[-1][:40]
            print(f"[stream-slow] {name} ttfb={ttfb:.1f}s total={total:.1f}s "
                  f"bytes={sum(len(b) for b in buf)} complete={complete}", flush=True)


def _stream_playlist(url):
    # buffered: the full body is needed to rewrite URIs through the proxy
    # ponytail: 12s < hls.js's fragLoadingTimeOut (20s) — on a hang the proxy
    # 502s first, so hls.js retries the frag (backoff) instead of freezing
    # on a fatal timeout race
    r = _upstream(upstream._get, url, timeout=12)
    body = r.content
    if "mpegurl" in r.headers.get("content-type", ""):
        body = _rewrite_m3u8(body.decode(), url).encode()
    ct = r.headers.get("content-type", "application/octet-stream")
    _cache_put(url, ct, body)
    return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS)


@app.get("/stream/{token}")
def stream(token: str):
    # ponytail: token is plain base64, so allowlist the host — otherwise any
    # logged-in user could use this as a GET proxy to arbitrary URLs (SSRF).
    # Verified 2026-08: master, variant and segment URLs all live on
    # hls.anidb.app; if the CDN moves, this 400s loudly and the list updates.
    url = _url(token)
    p = urlparse(url)
    if p.scheme != "https" or p.netloc != "hls.anidb.app":
        raise HTTPException(400, "bad stream token")
    # cache hit (a playlist, or a segment downloaded earlier): serve at once
    hit = _STREAM_CACHE.get(url)
    if hit:
        _STREAM_CACHE.move_to_end(url)
        return Response(hit[1], media_type=hit[0], headers=_STREAM_CACHE_HDRS)
    if url.endswith(".m3u8"):
        return _stream_playlist(url)
    # segment miss: try parallel byte-range lanes first (4 concurrent fetches
    # beat per-connection CDN throttling); ANY failure falls back to the
    # single streaming path, so this can never be worse than before
    t0 = time.time()
    try:
        body, ct = _upstream(upstream.fetch_parallel, url)
    except HTTPException:
        r = _upstream(upstream.get_stream, url, timeout=12)
        ct = r.headers.get("content-type", "application/octet-stream")
        if time.time() - t0 > 8:
            name = url.rsplit("/", 1)[-1][:40]
            print(f"[stream-slow] {name} parallel-fallback elapsed={time.time()-t0:.1f}s",
                  flush=True)
        return StreamingResponse(_seg_stream(r, url, ct), media_type=ct,
                                 headers=_STREAM_CACHE_HDRS)
    if time.time() - t0 > 8:
        name = url.rsplit("/", 1)[-1][:40]
        print(f"[stream-slow] {name} parallel total={time.time()-t0:.1f}s "
              f"bytes={len(body)}", flush=True)
    _cache_put(url, ct, body)
    return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS)


app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")