# -*- coding: utf-8 -*-
"""The submission ledger, kept in a private Hugging Face dataset.

A Space's container filesystem is ephemeral. Every rebuild, restart or sleep wipes it,
and with it every submission and every score. That is survivable for a demo and not
survivable for a contest with a prize attached, so the ledger lives outside the container
in a dataset repo that the entrant's work outlives the service.

Layout - one file per record, never appended to:

    submissions/<id>.json   written by the web service when an entry is accepted
    results/<id>.json       written by the GPU worker when it finishes scoring
    leaderboard.json        rolled up by the worker so the page reads one file

One file per record is what makes concurrent writers safe. Two entrants submitting at the
same moment touch different paths, so neither commit can clobber the other - which an
append to a shared JSONL absolutely would.

The rollup exists because a page load must not fan out into one request per entry. The
worker already holds every score at the moment it writes one, so it is the natural place
to rebuild the table.
"""
import base64
import json
import os
import threading
import time
import urllib.error
import urllib.request

REPO = os.environ.get("ODC_DATASET", "FINAL-Bench/odc-submissions")
TOKEN = os.environ.get("HF_TOKEN", "")
API = "https://huggingface.co/api/datasets/%s" % REPO
RESOLVE = "https://huggingface.co/datasets/%s/resolve/main" % REPO


def _hdr(extra=None):
    h = {"User-Agent": "VIDRAFT-ODC/1.0"}
    if TOKEN:
        h["Authorization"] = "Bearer " + TOKEN
    if extra:
        h.update(extra)
    return h


def _get_json(url, timeout=30):
    req = urllib.request.Request(url, headers=_hdr())
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())


def read(path, default=None):
    """Fetch one record. Missing files are a normal state, not an error."""
    try:
        return _get_json("%s/%s" % (RESOLVE, path))
    except urllib.error.HTTPError as e:
        if e.code in (404, 401, 403):
            return default
        raise
    except Exception:
        return default


class Busy(Exception):
    """The Hub refused the commit and kept refusing it. Distinct from a bug: the caller
    can tell the entrant to try again, which is true, instead of returning a 500 whose
    body is not even JSON."""


# Retries, not attempts: 3 means up to four calls. Deliberately short - an entrant is
# waiting on this request, and a submit that takes a minute reads as broken too.
RETRIES = int(os.environ.get("ODC_WRITE_RETRIES", "3"))
BACKOFF = float(os.environ.get("ODC_WRITE_BACKOFF", "1.5"))
RETRY_ON = (429, 500, 502, 503, 504)


def write_many(files, summary=None):
    """Commit several records in ONE commit.

    Three commits per submission is three chances to hit the rate limit and three
    partially-written states to reason about. The endpoint takes any number of file
    entries in a single body, so one submission becomes one commit: it either all lands
    or none of it does, and nobody ends up queued after being told it failed.

    2026-09-01: not hypothetical. A burst before the daily-cap reset drew HTTP 429 from
    the Hub and 235 submissions died on an unretried write - 230 cleanly, 5 half-written.
    """
    lines = [json.dumps({"key": "header",
                         "value": {"summary": summary or "write %d file(s)" % len(files)}})]
    for path, obj in files:
        blob = base64.b64encode(json.dumps(obj, ensure_ascii=False).encode()).decode()
        lines.append(json.dumps({"key": "file",
                                 "value": {"path": path, "content": blob,
                                           "encoding": "base64"}}))
    body = ("\n".join(lines) + "\n").encode()

    last = None
    for attempt in range(RETRIES + 1):
        req = urllib.request.Request(API + "/commit/main", data=body,
                                     headers=_hdr({"Content-Type": "application/x-ndjson"}))
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                return json.loads(r.read().decode())
        except urllib.error.HTTPError as e:
            last = e
            if e.code not in RETRY_ON or attempt == RETRIES:
                break
            # The Hub says how long to wait when it knows. Guessing shorter than that is
            # how a retry storm keeps the limit tripped.
            try:
                wait = float((e.headers or {}).get("Retry-After") or 0)
            except (TypeError, ValueError):
                wait = 0
            time.sleep(min(max(wait, BACKOFF * (2 ** attempt)), 8.0))
        except Exception as e:            # timeout, connection reset
            last = e
            if attempt == RETRIES:
                break
            time.sleep(BACKOFF * (2 ** attempt))

    code = getattr(last, "code", None)
    if code in RETRY_ON or code is None:
        raise Busy(str(last))
    raise last


def write(path, obj, summary=None):
    """One record, same guarantees. NDJSON is the format this endpoint takes - a plain
    JSON body is accepted and then quietly does nothing, which is a long way to debug."""
    return write_many([(path, obj)], summary or ("write " + path))


MAX_PAGES = int(os.environ.get("ODC_MAX_PAGES", "200"))


def _tree_pages(prefix):
    """Every page of a tree listing, following the Link cursor.

    The Hub caps a page at 1,000 entries. Reading only the first page was silent while
    the dataset was small and started dropping records the moment it was not."""
    url = "%s/tree/main/%s" % (API, prefix)
    seen = 0
    for _ in range(MAX_PAGES):
        req = urllib.request.Request(url, headers=_hdr())
        with urllib.request.urlopen(req, timeout=60) as r:
            page = json.loads(r.read().decode())
            link = r.headers.get("Link") or ""
        yield page
        seen += len(page)
        nxt = ""
        for part in link.split(","):
            if 'rel="next"' in part and "<" in part:
                nxt = part[part.index("<") + 1:part.index(">")]
        if not nxt:
            return
        url = nxt
    raise RuntimeError("tree listing exceeded %d pages at %s (%d entries)"
                       % (MAX_PAGES, prefix, seen))


def listdir(prefix):
    """Record ids under a prefix. Absent directory means nothing has been written yet."""
    out = []
    try:
        for page in _tree_pages(prefix):
            for e in page:
                q = e.get("path", "")
                if q.endswith(".json") and not os.path.basename(q).startswith("_"):
                    out.append(os.path.basename(q)[:-5])
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return []
        raise
    except Exception:
        # a partial listing is worse than none: the caller would treat the missing ids as
        # unscored and the rollup would drop them, which is exactly the failure this fixes
        raise
    return out


class Cached:
    """A small TTL cache so a page refresh does not become a round trip to the Hub.

    Staleness is bounded and harmless here: the worst case is a leaderboard a few seconds
    behind, and the page polls anyway.

    **Single flight.** One refresher at a time; everyone else is served the value already
    held, so an expiry does not send every concurrent request to the Hub at once.

    **Stale beats blocking, and stale beats an error.** A leaderboard a minute old is a
    working page. A timeout is not.
    """

    def __init__(self, ttl=20):
        self.ttl = ttl
        self._v = {}
        self._locks = {}
        self._guard = threading.Lock()

    def _lock_for(self, key):
        with self._guard:
            lk = self._locks.get(key)
            if lk is None:
                lk = self._locks[key] = threading.Lock()
            return lk

    def get(self, key, produce):
        hit = self._v.get(key)
        if hit and time.time() - hit[0] < self.ttl:
            return hit[1]

        lk = self._lock_for(key)
        # Block only when there is nothing at all to serve. If a refresh is already in
        # flight and we hold a stale value, hand that back instead of joining the queue.
        if not lk.acquire(blocking=(hit is None)):
            return hit[1]
        try:
            hit = self._v.get(key)          # the winner may have filled it while we waited
            if hit and time.time() - hit[0] < self.ttl:
                return hit[1]
            try:
                val = produce()
            except Exception:
                if hit:
                    return hit[1]
                raise
            self._v[key] = (time.time(), val)
            return val
        finally:
            lk.release()

    def drop(self, key=None):
        if key is None:
            self._v.clear()
        else:
            self._v.pop(key, None)