store.py

"""Durable per-user storage for FrogQuest — one JSON file per user.

Why JSON files, not SQLite: HF retired the classic Persistent Storage disk; the in-Space option is
now a mounted **Storage Bucket** (S3-like object storage). Buckets do whole-file writes only — no
byte-range writes or POSIX locks — which breaks SQLite but is a perfect match for our access
pattern: load/save ONE state blob per user. So each user gets `<DATA_DIR>/users/<uid>.json`
holding the entire app state (photo, theme, adventure, quests, campaigns, generated images,
selected id) — the same dict gr.BrowserState holds.

Identity: no login. The uid is generated client-side (app.py boot), lives in BrowserState, and is
shown to the user as their "Hero Code" so they can reclaim their data on a new/wiped browser.

DATA_DIR resolution (first hit wins):
  1. FROGQUEST_DATA_DIR env var (explicit override)
  2. /data            — mount your Storage Bucket here (Space Settings -> Storage Buckets ->
                        Mount a bucket -> path /data, read-write)
  3. ./.frogquest     — local dev fallback (gitignored)
"""
from __future__ import annotations

import json
import os
import re
import threading

_lock = threading.Lock()
_data_dir: str | None = None


def _resolve_data_dir() -> str:
    override = os.environ.get("FROGQUEST_DATA_DIR")
    if override:
        return override
    if os.path.isdir("/data") and os.access("/data", os.W_OK):  # mounted Storage Bucket
        return "/data"
    return os.path.join(os.path.dirname(os.path.abspath(__file__)), ".frogquest")  # local dev


def _users_dir() -> str:
    global _data_dir
    if _data_dir is None:
        _data_dir = _resolve_data_dir()
    d = os.path.join(_data_dir, "users")
    os.makedirs(d, exist_ok=True)
    return d


def _safe_uid(uid: str) -> str | None:
    """The uid comes from the client (BrowserState / pasted Hero Code) — sanitize before it
    touches a filesystem path. Returns None if nothing safe remains."""
    cleaned = re.sub(r"[^a-zA-Z0-9_-]", "", str(uid or ""))[:64]
    return cleaned or None


def _path_for(uid: str) -> str | None:
    safe = _safe_uid(uid)
    return os.path.join(_users_dir(), f"{safe}.json") if safe else None


def load(uid: str | None) -> dict | None:
    """Return the saved state dict for this user, or None (no uid / no file / unreadable)."""
    if not uid:
        return None
    try:
        path = _path_for(uid)
        if not path or not os.path.isfile(path):
            return None
        with _lock, open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else None
    except (OSError, json.JSONDecodeError):
        return None


def save(uid: str | None, state: dict) -> None:
    """Write the full state dict for this user. Atomic (tmp + os.replace) so a crash mid-write
    never leaves a half-written file. No-op without a uid; never raises into the request path."""
    if not uid or not isinstance(state, dict):
        return
    try:
        path = _path_for(uid)
        if not path:
            return
        blob = json.dumps(state, ensure_ascii=False)
        tmp = path + ".tmp"
        with _lock:
            with open(tmp, "w", encoding="utf-8") as f:
                f.write(blob)
            os.replace(tmp, path)
    except (OSError, TypeError, ValueError) as e:
        print(f"[store] save failed for uid={uid!r}: {e}")


if __name__ == "__main__":  # quick self-test (no model, no network)
    import tempfile
    with tempfile.TemporaryDirectory() as td:
        os.environ["FROGQUEST_DATA_DIR"] = td
        _data_dir = None  # re-resolve with the override
        assert load(None) is None
        assert load("nobody") is None
        sample = {"theme": "fantasy", "quests": [{"id": "q1"}], "images": {"q1": {"initial": "data:..."}},
                  "campaigns": []}
        save("hero-123", sample)
        assert load("hero-123") == sample
        sample["quests"].append({"id": "q2"})
        save("hero-123", sample)
        assert len(load("hero-123")["quests"]) == 2
        save("../../evil", sample)  # path traversal attempt -> sanitized, stays inside users/
        assert os.path.isfile(os.path.join(td, "users", "evil.json"))
        assert load("../../evil") == sample
        print("store.py self-test OK ->", _users_dir())