From 202bfa41c3d101a2ecf061b4429ef93b1470ba29 Mon Sep 17 00:00:00 2001 From: Josue Zamudio Date: Mon, 24 Aug 2026 16:02:27 +0000 Subject: [PATCH] storage: JSON files -> SQLite (users/history/searches/events) - one-time import from the legacy JSON (left in place as snapshot); ghost rows (int vs str ep_id) deduped on import, unrepresentable after - history POST is now an upsert: the (user, slug_id, ep_id) PK is the merge key, so the unguarded read-modify-write (lost updates) and the corruption-swallow (JSONDecodeError -> [] -> fresh overwrite) are gone - searches: NOCASE PK upsert + cap 8 - events table + position/search logging (watch-time trajectory), 90-day retention pruned at boot - account delete cascades all four tables - 22/22 tests green; old JSON files untouched --- server/app.py | 104 +++++++++++++++++++++------------------------ server/db.py | 106 ++++++++++++++++++++++++++++++++++++++++++++++ server/users.py | 79 +++++++++++++++++----------------- tests/conftest.py | 15 ++++--- tests/test_api.py | 23 +++++++--- 5 files changed, 222 insertions(+), 105 deletions(-) create mode 100644 server/db.py diff --git a/server/app.py b/server/app.py index 5745ada..6539632 100644 --- a/server/app.py +++ b/server/app.py @@ -18,13 +18,14 @@ from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Resp from fastapi.staticfiles import StaticFiles from itsdangerous import BadSignature, TimestampSigner -from . import upstream, users +from . import db, upstream, users ROOT = Path(__file__).parent.parent app = FastAPI(title="ani-web") # ---------- auth: multi-user, signed cookies ---------- -users.migrate_legacy() +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 @@ -133,29 +134,12 @@ def logout(): return resp -HIST_DIR = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data")) / "history" - - def _load_hist(user: str): - try: - entries = json.loads((HIST_DIR / f"{user}.json").read_text()) - except (FileNotFoundError, json.JSONDecodeError): - return [] - # ponytail: dedup legacy ghost rows — the same episode was stored with ep_id - # as int AND string; keep the newest per (slug_id, str(ep_id)) - best = {} - for e in entries: - k = (str(e.get("slug_id")), str(e.get("ep_id"))) - if e.get("t", 0) >= best.get(k, {"t": 0})["t"]: - best[k] = e - return sorted(best.values(), key=lambda e: e.get("t", 0), reverse=True) - - -def _save_hist(user: str, entries): - HIST_DIR.mkdir(exist_ok=True) - # ponytail: unguarded read-modify-write per file, fine at this scale; - # atomic_write keeps a crash/race from corrupting the file - users.atomic_write(HIST_DIR / f"{user}.json", json.dumps(entries, indent=1)) + 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") @@ -171,13 +155,18 @@ def api_history(request: Request): @app.delete("/api/history") def api_history_delete(request: Request, slug_id: str, ep_id: str): - # loose str() match on purpose: legacy rows store the same ep_id as int - # AND as string — one delete clears both ghost rows - entries = _load_hist(request.state.user) - keep = [e for e in entries - if not (str(e.get("slug_id")) == slug_id and str(e.get("ep_id")) == ep_id)] - _save_hist(request.state.user, keep) - return {"removed": len(entries) - len(keep)} + # 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") @@ -192,31 +181,27 @@ def api_history_post(entry: dict = Body(...), request: Request = None): raise HTTPException(422, "ep_id must be an integer") entry["t"] = time.time() u = request.state.user - entries = _load_hist(u) - for i, e in enumerate(entries): - # loose str() match: merge onto a legacy ghost row of the other type - if e.get("slug_id") == entry.get("slug_id") and str(e.get("ep_id")) == str(entry["ep_id"]): - entries[i] = {**e, **entry} - break - else: - entries.append(entry) - _save_hist(u, sorted(entries, key=lambda e: e.get("t", 0), reverse=True)[:50]) + 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 -SEARCH_DIR = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data")) / "searches" - - def _load_searches(user: str): - try: - return json.loads((SEARCH_DIR / f"{user}.json").read_text()) - except (FileNotFoundError, json.JSONDecodeError): - return [] - - -def _save_searches(user: str, items): - SEARCH_DIR.mkdir(exist_ok=True) - users.atomic_write(SEARCH_DIR / f"{user}.json", json.dumps(items, indent=1)) + 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") @@ -231,10 +216,17 @@ def api_searches_post(item: dict = Body(...), request: Request = None): if not q: raise HTTPException(422, "q required") u = request.state.user - # dedupe case-insensitively, newest first, cap 8 - items = [x for x in _load_searches(u) if x.lower() != q.lower()] - items.insert(0, q) - _save_searches(u, items[:8]) + # 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} diff --git a/server/db.py b/server/db.py new file mode 100644 index 0000000..bfdbdfd --- /dev/null +++ b/server/db.py @@ -0,0 +1,106 @@ +"""db.py — SQLite storage for ani-web: users, history, searches, events. + +One file per data dir: /aniweb.db (+ -wal/-shm while running). +Fresh connection per call: sqlite3 connections can't cross threads without +a lock, the threadpool runs the sync endpoints, and at this scale the +reconnect cost is zero. WAL: SQLite serializes writers itself, so the +lost-update races of the JSON read-modify-write era are gone. + +One-time import: a table that is empty while its legacy JSON still exists +gets absorbed (users.json, history/*.json, searches/*.json). The JSON files +are left in place as a last-known-good snapshot; the empty-table guard makes +a second import impossible. Legacy ghost rows (same episode stored with int +AND str ep_id) are deduped on import: newest t per (user, slug, ep). +""" +import json +import os +import sqlite3 +import time +from pathlib import Path + +DATA = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data")) +DB = DATA / "aniweb.db" +EVENT_RETENTION = 90 * 86400 # ponytail: pruned at boot, not by a cron + + +def connect() -> sqlite3.Connection: + c = sqlite3.connect(DB, timeout=10) + c.execute("PRAGMA journal_mode=WAL") + return c + + +def init() -> None: + """Schema + one-time JSON import + event pruning. Idempotent.""" + DATA.mkdir(exist_ok=True) + c = connect() + c.executescript(""" + CREATE TABLE IF NOT EXISTS users ( + name TEXT PRIMARY KEY, salt TEXT NOT NULL, hash TEXT NOT NULL, + iter INTEGER, created REAL); + CREATE TABLE IF NOT EXISTS history ( + user TEXT NOT NULL, slug_id TEXT NOT NULL, ep_id INTEGER NOT NULL, + t REAL, payload TEXT NOT NULL, + PRIMARY KEY (user, slug_id, ep_id)); + CREATE TABLE IF NOT EXISTS searches ( + user TEXT NOT NULL, q TEXT NOT NULL, t REAL, + PRIMARY KEY (user, q COLLATE NOCASE)); + CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY, ts REAL, user TEXT, type TEXT, payload TEXT); + CREATE INDEX IF NOT EXISTS ix_events_user_ts ON events (user, ts); + """) + _import_json(c) + c.execute("DELETE FROM events WHERE ts < ?", (time.time() - EVENT_RETENTION,)) + c.commit() + c.close() + + +def _import_json(c: sqlite3.Connection) -> None: + def empty(table: str) -> bool: + return c.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0 + + if empty("users") and (DATA / "users.json").exists(): + for name, d in json.loads((DATA / "users.json").read_text()).items(): + c.execute("INSERT INTO users VALUES (?,?,?,?,?)", + (name, d["salt"], d["hash"], d.get("iter"), d.get("created", 0.0))) + + if empty("history") and (DATA / "history").is_dir(): + best = {} + for f in (DATA / "history").glob("*.json"): + try: + rows = json.loads(f.read_text()) + except (json.JSONDecodeError, OSError): + continue + for e in rows: + try: + k = (f.stem, str(e.get("slug_id")), int(e["ep_id"])) + except (KeyError, TypeError, ValueError): + continue + if e.get("t", 0) >= best.get(k, {"t": 0})["t"]: + best[k] = e + for (user, slug, ep), e in best.items(): + e = dict(e, ep_id=ep) # normalize ghost-row int/str + c.execute("INSERT INTO history VALUES (?,?,?,?,?)", + (user, slug, ep, e.get("t", 0.0), json.dumps(e))) + + if empty("searches") and (DATA / "searches").is_dir(): + # no per-item timestamps in the JSON: file mtime, oldest-last order + for f in (DATA / "searches").glob("*.json"): + try: + rows = json.loads(f.read_text()) + except (json.JSONDecodeError, OSError): + continue + t0, seen = f.stat().st_mtime, set() + for i, q in enumerate(rows): + q = str(q) + if q.lower() in seen: + continue + seen.add(q.lower()) + c.execute("INSERT OR REPLACE INTO searches VALUES (?,?,?)", + (f.stem, q, t0 - i)) + + +def event(c: sqlite3.Connection, user: str, type: str, payload: dict) -> None: + """Append one metric event (caller commits). Watch-time is derivable + from the 'position' stream: consecutive pos deltas per (user, ep).""" + c.execute("INSERT INTO events (ts, user, type, payload) VALUES (?,?,?,?)", + (time.time(), user, type, json.dumps(payload))) diff --git a/server/users.py b/server/users.py index 06a11ac..41bcdeb 100644 --- a/server/users.py +++ b/server/users.py @@ -1,4 +1,4 @@ -"""users.py — multi-user accounts for ani-web (PBKDF2, single json file). +"""users.py — multi-user accounts for ani-web (PBKDF2, SQLite via db.py). CLI: .venv/bin/python -m server.users add (password via prompt) .venv/bin/python -m server.users rm @@ -10,11 +10,12 @@ import json import os import re import secrets +import sqlite3 import time -from pathlib import Path -DATA = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data")) -USERS = DATA / "users.json" +from . import db + +DATA = db.DATA SECRET = DATA / "secret.key" ITER = 210_000 # OWASP 2023 minimum for PBKDF2-SHA256 NAME_RE = re.compile(r"^[a-z0-9_-]{3,32}$") @@ -32,9 +33,10 @@ def get_secret() -> bytes: def migrate_legacy(): - """one-time: single-password era -> users.json + per-user history files.""" + """one-time: single-password era -> users.json + per-user history files, + which db.init() (called after this) absorbs into the tables.""" cred = DATA / "credentials.json" - if not USERS.exists() and cred.exists(): + if not (DATA / "users.json").exists() and cred.exists(): d = json.loads(cred.read_text()) save({"admin": {"salt": d["salt"], "hash": d["hash"], "created": time.time()}}) cred.unlink() @@ -45,26 +47,16 @@ def migrate_legacy(): def load() -> dict: - try: - return json.loads(USERS.read_text()) - except (FileNotFoundError, json.JSONDecodeError): - return {} - - -def atomic_write(path, text): - """tmp + os.replace: a crash mid-write leaves the previous file intact - (in-place writes corrupted the file -> load() returned {} -> accounts - or history silently vanished).""" - tmp = path.parent / (path.name + ".tmp") - fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - with os.fdopen(fd, "w") as f: - f.write(text) - os.replace(tmp, path) + c = db.connect() + rows = c.execute("SELECT name, salt, hash, iter, created FROM users").fetchall() + c.close() + return {n: {"salt": s, "hash": h, "iter": i, "created": cr} for n, s, h, i, cr in rows} def save(users: dict): + """file-based save, only for migrate_legacy (runs before db.init).""" DATA.mkdir(exist_ok=True) - atomic_write(USERS, json.dumps(users, indent=1)) + (DATA / "users.json").write_text(json.dumps(users, indent=1)) def add(username: str, password: str): @@ -73,37 +65,48 @@ def add(username: str, password: str): raise ValueError("username: 3-32 chars, only a-z 0-9 _ -") if len(password) < 8: raise ValueError("password: min 8 chars") - users = load() - if u in users: - raise ValueError("username already taken") salt = secrets.token_bytes(16) - users[u] = {"salt": salt.hex(), - "hash": hashlib.pbkdf2_hmac("sha256", password.encode(), salt, ITER).hex(), - "created": time.time()} - save(users) + c = db.connect() + try: + c.execute("INSERT INTO users VALUES (?,?,?,?,?)", + (u, salt.hex(), + hashlib.pbkdf2_hmac("sha256", password.encode(), salt, ITER).hex(), + ITER, time.time())) + c.commit() + except sqlite3.IntegrityError: + raise ValueError("username already taken") + finally: + c.close() def verify(username: str, password: str) -> bool: - d = load().get(username.lower()) - if not d: + c = db.connect() + row = c.execute("SELECT salt, hash, iter FROM users WHERE name=?", + (username.lower(),)).fetchone() + c.close() + if not row: return False h = hashlib.pbkdf2_hmac("sha256", password.encode(), - bytes.fromhex(d["salt"]), d.get("iter", ITER)) - return secrets.compare_digest(h.hex(), d["hash"]) + bytes.fromhex(row[0]), row[2] or ITER) + return secrets.compare_digest(h.hex(), row[1]) def remove(username: str): u = username.lower() - users = load() - users.pop(u, None) - save(users) - (DATA / "history" / f"{u}.json").unlink(missing_ok=True) - (DATA / "searches" / f"{u}.json").unlink(missing_ok=True) + c = db.connect() + c.execute("DELETE FROM users WHERE name=?", (u,)) + c.execute("DELETE FROM history WHERE user=?", (u,)) + c.execute("DELETE FROM searches WHERE user=?", (u,)) + c.execute("DELETE FROM events WHERE user=?", (u,)) + c.commit() + c.close() if __name__ == "__main__": import sys cmd = sys.argv[1] if len(sys.argv) > 1 else "" + migrate_legacy() # file-based, runs before db.init absorbs its output + db.init() if cmd == "add" and len(sys.argv) > 2: name = sys.argv[2] pw = getpass.getpass(f"Password for {name} (min 8 chars): ") diff --git a/tests/conftest.py b/tests/conftest.py index be1137c..1ad3cb2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -54,13 +54,18 @@ def server(): data = Path(tempfile.mkdtemp(prefix="aniweb-test-")) os.environ["ANIWEB_DATA"] = str(data) sys.path.insert(0, str(ROOT)) - from server import users # noqa: E402 + from server import db, users # noqa: E402 + db.init() # schema before users.add / seeding for name, pw in USERS: users.add(name, pw) - (data / "history").mkdir() - (data / "history" / "admin.json").write_text(json.dumps([ - {"title": "Seed Anime", "slug_id": "seed-1", "season_id": 1, "ep_id": "1", - "lang": "eng", "ep": "1", "pos": 10.0, "t": time.time()}])) + c = db.connect() + seed_t = time.time() + c.execute("INSERT INTO history VALUES (?,?,?,?,?)", + ("admin", "seed-1", 1, seed_t, json.dumps( + {"title": "Seed Anime", "slug_id": "seed-1", "season_id": 1, + "ep_id": 1, "lang": "eng", "ep": "1", "pos": 10.0, "t": seed_t}))) + c.commit() + c.close() proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "server.app:app", diff --git a/tests/test_api.py b/tests/test_api.py index 725ef6c..d36ad60 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -96,7 +96,7 @@ def test_no_self_signup(base=BASE): jar = login("admin", "adminpw12345", base) assert req("POST", "/signup", {"username": "zed", "password": "whatever123"}, jar=jar, base=base) == 405 - assert "zed" not in json.load(open(_users_file())) + assert "zed" not in _users() def test_history_requires_ep_id(base=BASE): @@ -172,8 +172,8 @@ def test_self_delete(base=BASE): # give dave a searches file too: account delete must remove it assert req("POST", "/api/searches", {"q": "dave-search"}, jar=jar, base=base) == 200 assert req("POST", "/account/delete", jar=jar, base=base) == 200 - assert "dave" not in json.load(open(_users_file())) - assert not os.path.exists(os.path.join(os.environ["ANIWEB_DATA"], "searches", "dave.json")) + assert "dave" not in _users() + assert _search_count("dave") == 0 # account delete cascades the rows # old session cookie is dead immediately (middleware re-checks existence) assert req("GET", "/api/me", jar=jar, base=base) == 401 jar2 = CJ() @@ -230,6 +230,17 @@ def _json(method, path, jar=None, base=BASE): return json.loads(resp.read().decode()) -def _users_file(): - import os - return os.path.join(os.environ["ANIWEB_DATA"], "users.json") +def _users(): + import sqlite3 + c = sqlite3.connect(os.path.join(os.environ["ANIWEB_DATA"], "aniweb.db")) + names = [r[0] for r in c.execute("SELECT name FROM users")] + c.close() + return names + + +def _search_count(user): + import sqlite3 + c = sqlite3.connect(os.path.join(os.environ["ANIWEB_DATA"], "aniweb.db")) + n = c.execute("SELECT COUNT(*) FROM searches WHERE user=?", (user,)).fetchone()[0] + c.close() + return n