auth.py

"""Accounts (email + password, PBKDF2, SQLite) and admin helpers."""
import hashlib, hmac, html, os, re, secrets, shutil, sqlite3, tempfile, threading, time
from pathlib import Path

import gradio as gr

import db

OUT_DIR = Path(tempfile.gettempdir()) / "cs_outputs"
OUT_DIR.mkdir(parents=True, exist_ok=True)

# ---------------------------------------------------------------------------
# Accounts (email + password, PBKDF2-hashed, SQLite)
# Stored in /data when persistent storage is enabled; otherwise next to app.py.
# ---------------------------------------------------------------------------
_PERSISTENT = Path("/data").is_dir() and os.access("/data", os.W_OK)
_DATA_DIR = Path("/data") if _PERSISTENT else Path(__file__).parent
# The Space's /data is a mounted storage bucket. SQLite must not be opened there: its locking and in-place page
# writes are not safe on such a mount and the file ends up "malformed". So the live database is on the container's
# own disk (USERS_DB) and a verified snapshot of it is kept in /data (STORE_DB): read at start, rewritten after
# every change. USERS_DB set by hand (tests, another host) is used as it is, with no snapshot.
STORE_DB = None
if os.getenv("USERS_DB"):
    USERS_DB = Path(os.environ["USERS_DB"])
elif _PERSISTENT:
    STORE_DB = _DATA_DIR / "users.db"
    USERS_DB = Path(os.getenv("USERS_LIVE_DIR", str(Path(tempfile.gettempdir()) / "cs_live"))) / "users.db"
else:
    USERS_DB = _DATA_DIR / "users.db"
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")

# Optional backup of the user database to a PRIVATE Hugging Face dataset repo, so accounts
# survive restarts/rebuilds even without Persistent storage.
#   USERS_REPO        e.g. "Rasheed/suite-users"   (created automatically, private)
#   USERS_REPO_TOKEN  write token (falls back to HF_TOKEN)
USERS_REPO = os.getenv("USERS_REPO", "").strip()
_REPO_TOKEN = os.getenv("USERS_REPO_TOKEN") or os.getenv("HF_TOKEN")
db._SQLITE_PATH = USERS_DB
if db.IS_MYSQL:                        # permanent MySQL: no database file to snapshot or back up
    STORE_DB = None
    USERS_REPO = ""
STORAGE_NOTE = ""
_sync_now = threading.Event()


def _restore_backup():
    global STORAGE_NOTE
    if db.IS_MYSQL:
        return
    if not USERS_REPO:
        STORAGE_NOTE = "" if _PERSISTENT else (
            "⚠️ Accounts are stored on temporary disk and are erased when the Space restarts. "
            "Enable Persistent storage or set the USERS_REPO secret.")
        return
    try:
        from huggingface_hub import HfApi, hf_hub_download
        HfApi(token=_REPO_TOKEN).create_repo(USERS_REPO, repo_type="dataset", private=True, exist_ok=True)
        if not USERS_DB.exists():
            p = hf_hub_download(USERS_REPO, "users.db", repo_type="dataset", token=_REPO_TOKEN)
            shutil.copy(p, USERS_DB)
            print(f"[auth] restored users.db from {USERS_REPO}")
    except Exception as err:          # first run: file not in repo yet
        print(f"[auth] backup restore skipped: {err}")


def _backup_loop():
    from huggingface_hub import HfApi
    api, last = HfApi(token=_REPO_TOKEN), None
    while True:
        _sync_now.wait(timeout=60)
        _sync_now.clear()
        try:
            if not USERS_DB.exists():
                continue
            m = USERS_DB.stat().st_mtime
            if m == last:
                continue
            snap = Path(tempfile.gettempdir()) / "users_backup.db"
            src, dst = sqlite3.connect(str(USERS_DB)), sqlite3.connect(str(snap))
            with dst:
                src.backup(dst)                     # consistent copy even during writes
            src.close(); dst.close()
            api.upload_file(path_or_fileobj=str(snap), path_in_repo="users.db", repo_id=USERS_REPO,
                            repo_type="dataset", commit_message="users backup")
            last = m
        except Exception as err:
            print(f"[auth] backup failed: {err}")
            time.sleep(30)


def _healthy(path):
    """True when SQLite reads the whole file without finding damage."""
    try:
        con = sqlite3.connect(str(path), timeout=10)
        try:
            return con.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
        finally:
            con.close()
    except sqlite3.Error:
        return False


def _salvage(bad, good):
    """Copy every row that can still be read from a damaged database into a new one. Returns {table: rows}."""
    src = sqlite3.connect(f"file:{Path(bad).as_posix()}?mode=ro", uri=True, timeout=10)
    dst = sqlite3.connect(str(good))
    saved = {}
    try:
        tables = src.execute("SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
                             ).fetchall()
        for name, sql in tables:
            try:
                dst.execute(sql)
                keyed = any(r[5] for r in src.execute(f'PRAGMA table_info("{name}")'))
            except sqlite3.Error as err:
                print(f"[auth] salvage: table {name} skipped: {err}")
                continue
            n = 0
            # from the start, then (when rows have a key, so none is copied twice) from the end: what lies on
            # either side of a damaged page is kept. NOT INDEXED: a damaged index must not hide good rows.
            for order in ("ASC", "DESC") if keyed else ("ASC",):
                try:
                    for row in src.execute(f'SELECT * FROM "{name}" NOT INDEXED ORDER BY rowid {order}'):
                        cur = dst.execute(f'INSERT OR IGNORE INTO "{name}" VALUES ({",".join("?" * len(row))})', row)
                        n += cur.rowcount
                    break
                except sqlite3.Error as err:
                    print(f"[auth] salvage: {name} read stopped ({order}): {err}")
            saved[name] = n
        dst.commit()
    finally:
        src.close()
        dst.close()
    return saved


def _load_store():
    """Start of the app: bring the snapshot in /data to the live database, repairing it when it is damaged."""
    global STORAGE_NOTE
    if not STORE_DB:
        return
    USERS_DB.parent.mkdir(parents=True, exist_ok=True)
    if USERS_DB.exists() or not STORE_DB.exists():
        return
    tmp = USERS_DB.with_name("incoming.db")
    try:
        for suffix in ("", "-journal", "-wal"):           # an unfinished write of the old in-place database
            p = Path(str(STORE_DB) + suffix)
            if p.exists():
                shutil.copy(p, str(tmp) + suffix)
        if _healthy(tmp):
            for suffix in ("-journal", "-wal", "-shm"):
                Path(str(tmp) + suffix).unlink(missing_ok=True)
            os.replace(tmp, USERS_DB)
            print(f"[auth] accounts loaded from {STORE_DB}")
            return
        keep = STORE_DB.with_name(f"users.damaged-{time.strftime('%Y%m%d-%H%M%S')}.db")
        shutil.copy(STORE_DB, keep)                       # the damaged file is kept, never thrown away
        saved = _salvage(tmp, USERS_DB)
        print(f"[auth] {STORE_DB} was damaged; rows recovered: {saved}; the damaged file is kept as {keep}")
    except Exception as err:
        print(f"[auth] could not load {STORE_DB}: {err}")
        STORAGE_NOTE = "⚠️ The saved accounts could not be read at start. Please contact support before signing up again."
    finally:
        for suffix in ("", "-journal", "-wal", "-shm"):
            Path(str(tmp) + suffix).unlink(missing_ok=True)


_store_lock, _stored = threading.Lock(), [None]


def store_now():
    """Write a consistent, checked snapshot of the live database to /data (whole file, then renamed into place)."""
    if not STORE_DB or not USERS_DB.exists():
        return False
    with _store_lock:
        st = USERS_DB.stat()
        stamp = (st.st_mtime_ns, st.st_size)
        if stamp == _stored[0]:
            return False
        snap = USERS_DB.with_name("snapshot.db")
        src, dst = sqlite3.connect(str(USERS_DB), timeout=10), sqlite3.connect(str(snap))
        try:
            with dst:
                src.backup(dst)                           # consistent copy even during writes
        finally:
            src.close()
            dst.close()
        if not _healthy(snap):
            print("[auth] snapshot failed its check; the stored copy is left as it was")
            return False
        new = STORE_DB.with_name("users.db.new")
        try:
            shutil.copyfile(snap, new)
            os.replace(new, STORE_DB)
        except OSError:                                   # a mount that cannot rename: write the file directly
            shutil.copyfile(snap, STORE_DB)
        for suffix in ("-journal", "-wal", "-shm"):       # left by the old in-place database
            Path(str(STORE_DB) + suffix).unlink(missing_ok=True)
        _stored[0] = stamp
        return True


def _store_loop():
    while True:
        _store_now.wait(timeout=15)
        _store_now.clear()
        try:
            store_now()
        except Exception as err:
            print(f"[auth] storing the accounts failed: {err}")
            time.sleep(15)


_store_now = threading.Event()


def request_sync():
    _sync_now.set()
    _store_now.set()


_load_store()
_restore_backup()
print("[db] backend:", db.ping(), flush=True)      # Space logs show at once whether MySQL is reachable
if STORE_DB:
    import atexit
    threading.Thread(target=_store_loop, daemon=True).start()
    atexit.register(lambda: store_now())
if USERS_REPO:
    threading.Thread(target=_backup_loop, daemon=True).start()


def _db():
    if db.IS_MYSQL:                        # tables are created once by db.py
        return db.connect()
    con = db.connect(USERS_DB)
    con.execute(
        "CREATE TABLE IF NOT EXISTS users (email TEXT PRIMARY KEY, name TEXT, salt TEXT, "
        "pw TEXT, created TEXT DEFAULT CURRENT_TIMESTAMP, last_login TEXT, runs INTEGER DEFAULT 0)"
    )
    con.execute("CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT)")
    con.execute("CREATE TABLE IF NOT EXISTS usage (who TEXT, day TEXT, kind TEXT, n INTEGER DEFAULT 0, "
                "PRIMARY KEY (who, day, kind))")
    cols = {r[1] for r in con.execute("PRAGMA table_info(users)")}
    for c, ddl in (("active", "INTEGER DEFAULT 1"), ("text_runs", "INTEGER DEFAULT 0"),
                   ("doc_limit", "INTEGER"), ("text_limit", "INTEGER")):   # limits: per-user overrides
        if c not in cols:
            try:
                con.execute(f"ALTER TABLE users ADD COLUMN {c} {ddl}")
            except sqlite3.OperationalError as err:       # another thread added it at the same moment
                if "duplicate column" not in str(err).lower():
                    raise
    return con


# ---------------------------------------------------------------------------
# "Stay signed in" tokens (kept in the visitor's browser, HMAC-signed, 30 days)
# ---------------------------------------------------------------------------
SESSION_DAYS = int(os.getenv("SESSION_DAYS", "30"))


def session_secret():
    env = os.getenv("SESSION_SECRET")
    if env:
        return env
    with _db() as con:
        row = con.execute("SELECT v FROM meta WHERE k='secret'").fetchone()
        if row:
            return row[0]
        s = secrets.token_hex(32)
        con.execute("INSERT INTO meta (k, v) VALUES ('secret', ?)", (s,))
    request_sync()
    return s


def make_token(user):
    exp = int(time.time()) + SESSION_DAYS * 86400
    body = f"{user['email']}|{exp}"
    sig = hmac.new(session_secret().encode(), body.encode(), hashlib.sha256).hexdigest()
    return f"{body}|{sig}"


def user_from_token(token):
    try:
        email, exp, sig = (token or "").rsplit("|", 2)
        body = f"{email}|{exp}"
        good = hmac.new(session_secret().encode(), body.encode(), hashlib.sha256).hexdigest()
        if not hmac.compare_digest(sig, good) or int(exp) < time.time():
            return None
        with _db() as con:
            row = con.execute("SELECT name, active FROM users WHERE email=?", (email,)).fetchone()
        if not row or row[1] == 0:
            return None
        return {"email": email, "name": row[0]}
    except Exception:
        return None


# Admins: comma-separated emails in the ADMIN_EMAILS secret (Space settings → Variables and secrets).
ADMIN_EMAILS = {e.strip().lower() for e in os.getenv("ADMIN_EMAILS", "").split(",") if e.strip()}


def is_admin(user):
    return bool(user and user.get("email", "").lower() in ADMIN_EMAILS)


def _hash(password, salt):
    return hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), 200_000).hex()


def create_user(name, email, password, confirm):
    name, email = (name or "").strip(), (email or "").strip().lower()
    if not name:
        return None, "Please enter your name."
    if not _EMAIL_RE.match(email):
        return None, "Please enter a valid email address."
    if len(password or "") < 8:
        return None, "Password must be at least 8 characters."
    if password != confirm:
        return None, "Passwords do not match."
    salt = secrets.token_hex(16)
    with _db() as con:
        if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone():
            return None, "An account with this email already exists. Please log in."
        con.execute("INSERT INTO users (email, name, salt, pw, last_login) VALUES (?,?,?,?,CURRENT_TIMESTAMP)",
                    (email, name, salt, _hash(password, salt)))
    request_sync()
    return {"email": email, "name": name}, None


# Wrong passwords per account: after LOGIN_TRIES within LOGIN_WINDOW seconds the account's log-in waits out the
# window, so a password cannot be guessed by trying thousands of them. (Kept in memory: a restart clears it.)
LOGIN_TRIES = int(os.getenv("LOGIN_TRIES", "8"))
LOGIN_WINDOW = int(os.getenv("LOGIN_WINDOW", "900"))
_FAILS, _fails_lock = {}, threading.Lock()


def _recent_fails(email, add=False, clear=False):
    now = time.time()
    with _fails_lock:
        if clear:
            _FAILS.pop(email, None)
            return 0
        times = [t for t in _FAILS.get(email, []) if now - t < LOGIN_WINDOW]
        if add:
            times.append(now)
        if times:
            _FAILS[email] = times
        else:
            _FAILS.pop(email, None)
        if len(_FAILS) > 20000:                           # never grows without bound
            _FAILS.clear()
        return len(times)


def check_user(email, password):
    email = (email or "").strip().lower()
    if _recent_fails(email) >= LOGIN_TRIES:
        return None, (f"Too many wrong passwords for this account. Please wait {LOGIN_WINDOW // 60} minutes, "
                      "or use 🔑 Forgot password.")
    with _db() as con:
        row = con.execute("SELECT name, salt, pw, active FROM users WHERE email=?", (email,)).fetchone()
        if not row:
            return None, "No account found with this email. Please check it or create an account under Sign up."
        if not hmac.compare_digest(row[2], _hash(password or "", row[1])):
            _recent_fails(email, add=True)
            return None, "Incorrect password. Please try again, or use 🔑 Forgot password."
        if row[3] == 0:
            return None, "This account has been disabled. Please contact CS Software Solutions."
        con.execute("UPDATE users SET last_login=CURRENT_TIMESTAMP WHERE email=?", (email,))
    _recent_fails(email, clear=True)
    return {"email": email, "name": row[0]}, None


def _count_run(user, document=True):
    """Increment the user's document or pasted-text run counter (no-op for guests)."""
    if not user or not user.get("email"):
        return
    col = "runs" if document else "text_runs"
    try:
        with _db() as con:
            con.execute(f"UPDATE users SET {col} = COALESCE({col}, 0) + 1 WHERE email=?", (user["email"],))
    except Exception:
        pass


class Upsell(gr.Error):
    """Not an error: a plan / sign-in / limit message. The app shows it as a friendly centred pop-up.
    kind: "upgrade" (paid plan unlocks it), "signin" (free account unlocks it), "limit" (try again later)."""

    def __init__(self, message, kind="upgrade", title=None):
        titles = {"upgrade": "Unlock more with a plan", "signin": "Create your free account",
                  "limit": "You've reached today's limit"}
        super().__init__(message, title=title or titles.get(kind, "Heads up"), print_exception=False)
        self.kind, self.headline, self.text = kind, title or titles.get(kind, "Heads up"), message


def _require(user):
    if not user or not user.get("email"):
        raise Upsell("Sign in or create a free account to upload Word and PDF documents — it takes 20 seconds. "
                     "Pasting text works without an account.", kind="signin", title="Upload documents for free")
    with _db() as con:
        row = con.execute("SELECT active FROM users WHERE email=?", (user["email"],)).fetchone()
    if not row or row[0] == 0:
        raise gr.Error("This account has been disabled.")


def _msg(text, ok=False):
    cls = "cs-ok" if ok else "cs-bad"
    return f'<div class="cs-authmsg {cls}">{text}</div>' if text else ""


# ---------------------------------------------------------------------------
# Admin panel
# ---------------------------------------------------------------------------
ADMIN_COLS = ["Name", "Email", "Joined", "Last login", "Doc runs", "Text runs", "Status", "Role"]


def _need_admin(user):
    if not is_admin(user):
        raise gr.Error("Admin access only.")


def admin_stats(user):
    _need_admin(user)
    with _db() as con:
        total = con.execute("SELECT COUNT(*) FROM users").fetchone()[0]
        active = con.execute("SELECT COUNT(*) FROM users WHERE active=1").fetchone()[0]
        runs = con.execute("SELECT COALESCE(SUM(runs),0) + COALESCE(SUM(text_runs),0) FROM users").fetchone()[0]
        new7 = con.execute("SELECT COUNT(*) FROM users WHERE created >= datetime('now','-7 days')").fetchone()[0]
        seen7 = con.execute("SELECT COUNT(*) FROM users WHERE last_login >= datetime('now','-7 days')").fetchone()[0]
    tiles = [("Total users", total), ("Active accounts", active), ("Total runs", runs),
             ("New (7 days)", new7), ("Logged in (7 days)", seen7)]
    cells = "".join(f'<div class="cs-stat"><div class="n">{v:,}</div><div class="l">{k}</div></div>' for k, v in tiles)
    return f'<div class="cs-stats">{cells}</div>'


def admin_table(user, query=""):
    _need_admin(user)
    q = f"%{(query or '').strip().lower()}%"
    with _db() as con:
        rows = con.execute(
            "SELECT name, email, created, last_login, runs, COALESCE(text_runs,0), active FROM users "
            "WHERE lower(name) LIKE ? OR lower(email) LIKE ? ORDER BY created DESC", (q, q)).fetchall()
    return [[n, e, (c or "")[:16], (l or "")[:16], r, t, "Active" if a else "Disabled",
             "Admin" if e in ADMIN_EMAILS else "User"] for n, e, c, l, r, t, a in rows]


def _admin_refresh(user, query, note, ok=True):
    request_sync()
    return admin_stats(user), admin_table(user, query), _msg(note, ok=ok)


def admin_search(user, query):
    _need_admin(user)
    return admin_table(user, query)


def admin_set_active(user, email, query, active):
    _need_admin(user)
    email = (email or "").strip().lower()
    if not email:
        return _admin_refresh(user, query, "Enter the user's email first.", ok=False)
    if email in ADMIN_EMAILS and not active:
        return _admin_refresh(user, query, "Admin accounts can't be disabled here.", ok=False)
    with _db() as con:
        n = con.execute("UPDATE users SET active=? WHERE email=?", (1 if active else 0, email)).rowcount
    if not n:
        return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
    return _admin_refresh(user, query, f"{email} is now {'active' if active else 'disabled'}.")


def admin_reset_pw(user, email, new_pw, query):
    _need_admin(user)
    email = (email or "").strip().lower()
    if len(new_pw or "") < 8:
        return _admin_refresh(user, query, "New password must be at least 8 characters.", ok=False)
    salt = secrets.token_hex(16)
    with _db() as con:
        n = con.execute("UPDATE users SET salt=?, pw=? WHERE email=?", (salt, _hash(new_pw, salt), email)).rowcount
    if n:
        with _pw_db() as con:
            con.execute("UPDATE reset_requests SET status='done' WHERE email=? AND status='open'", (email,))
        request_sync()
    if not n:
        return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
    return _admin_refresh(user, query, f"Password reset for {email}. Share it with them securely.")


def admin_delete(user, email, confirm, query):
    _need_admin(user)
    email = (email or "").strip().lower()
    if (confirm or "").strip().lower() != email or not email:
        return _admin_refresh(user, query, "Type the same email in the confirm box to delete.", ok=False)
    if email in ADMIN_EMAILS:
        return _admin_refresh(user, query, "Admin accounts can't be deleted here.", ok=False)
    with _db() as con:
        n = con.execute("DELETE FROM users WHERE email=?", (email,)).rowcount
    if not n:
        return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
    return _admin_refresh(user, query, f"Deleted {email}.")


def admin_export(user):
    _need_admin(user)
    import csv
    folder = Path(tempfile.mkdtemp(dir=OUT_DIR))
    path = folder / "users.csv"
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(ADMIN_COLS)
        w.writerows(admin_table(user))
    return str(path)


# ---------------------------------------------------------------------------
# Daily usage limits (0 = unlimited). Admins are never limited.
#   DAILY_DOC_LIMIT         document runs per signed-in user per day   (default 20)
#   DAILY_TEXT_LIMIT        pasted-text runs per signed-in user per day (default 100)
#   DAILY_GUEST_TEXT_LIMIT  pasted-text runs per guest (by IP) per day  (default 15)
# ---------------------------------------------------------------------------
LIMITS = {
    ("user", "doc"): int(os.getenv("DAILY_DOC_LIMIT", "20")),
    ("user", "text"): int(os.getenv("DAILY_TEXT_LIMIT", "100")),
    ("guest", "text"): int(os.getenv("DAILY_GUEST_TEXT_LIMIT", "15")),
}


def _who(user, request):
    if user and user.get("email"):
        return user["email"]
    ip = ""
    try:
        h = request.headers if request is not None else {}
        ip = (h.get("x-forwarded-for") or "").split(",")[0].strip() or getattr(request.client, "host", "")
    except Exception:
        pass
    return "ip:" + (ip or "unknown")


DAY_OFFSET_MIN = int(os.getenv("DAY_TZ_OFFSET_MIN", "330"))    # daily limits follow the Indian day (IST)


def _today():
    return time.strftime("%Y-%m-%d", time.gmtime(time.time() + DAY_OFFSET_MIN * 60))


def usage_today(who, kind):
    with _db() as con:
        row = con.execute("SELECT n FROM usage WHERE who=? AND day=? AND kind=?", (who, _today(), kind)).fetchone()
    return row[0] if row else 0


def user_limits(email):
    """Effective daily limits for a signed-in user: {'doc': n, 'text': n} (0 = unlimited)."""
    lim = {"doc": LIMITS[("user", "doc")], "text": LIMITS[("user", "text")]}
    try:
        with _db() as con:
            row = con.execute("SELECT doc_limit, text_limit FROM users WHERE email=?", (email,)).fetchone()
        if row:
            lim["doc"] = lim["doc"] if row[0] is None else row[0]
            lim["text"] = lim["text"] if row[1] is None else row[1]
    except Exception:
        pass
    return lim


def admin_set_limits(user, email, doc, text, query=""):
    """Blank = back to default, 0 = unlimited."""
    _need_admin(user)
    email = (email or "").strip().lower()
    conv = lambda v: None if v in (None, "") else int(float(v))
    try:
        d, t = conv(doc), conv(text)
    except ValueError:
        return _admin_refresh(user, query, "Limits must be whole numbers (blank = default, 0 = unlimited).", ok=False)
    with _db() as con:
        n = con.execute("UPDATE users SET doc_limit=?, text_limit=? WHERE email=?", (d, t, email)).rowcount
    if not n:
        return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
    show = lambda v: "default" if v is None else ("unlimited" if v == 0 else str(v))
    return _admin_refresh(user, query, f"Limits for {email}: documents {show(d)}, texts {show(t)} per day.")


def _override(email, col):
    try:
        with _db() as con:
            row = con.execute(f"SELECT {col} FROM users WHERE email=?", (email,)).fetchone()
        return row[0] if row else None
    except Exception:
        return None


def check_limit(user, request, document):
    """Raise a friendly error if today's limit is used up."""
    if is_admin(user):
        return
    import plans as P
    kind = "doc" if document else "text"
    role = "user" if user else "guest"
    if kind == "text":
        override = _override(user["email"], "text_limit") if user else None
        lim = override if override is not None else P.daily_paste_limit(user)
    else:                                   # monthly quota lives in plans.py; this is only an anti-abuse cap
        override = _override(user["email"], "doc_limit") if user else None
        st = P.state(user) if user else None
        if override is None and st and st["key"] in P.PAID:
            return                          # paid plans: the daily document allowance is checked in plans.check_doc
        lim = override if override is not None else LIMITS[("user", "doc")]
    if lim <= 0:
        return
    if usage_today(_who(user, request), kind) >= lim:
        what = "document runs" if document else "pasted-text runs"
        if not user:
            raise Upsell(f"Guests can do {lim} {what} a day. Create a free account for a higher limit — "
                         "or come back after midnight (IST).", kind="signin", title="You've reached today's free limit")
        paid = P.state(user)["key"] in P.PAID
        raise Upsell(f"You've used all {lim} {what} for today. It resets at midnight (IST)."
                     + ("" if paid else " Paid plans have much higher daily limits."),
                     kind="limit" if paid else "upgrade", title="You've reached today's limit")


def add_usage(user, request, document):
    kind = "doc" if document else "text"
    try:
        with _db() as con:
            con.execute("INSERT INTO usage (who, day, kind, n) VALUES (?,?,?,1) "
                        "ON CONFLICT(who, day, kind) DO UPDATE SET n = n + 1",
                        (_who(user, request), _today(), kind))
            con.execute("DELETE FROM usage WHERE day < date('now','-7 days')")
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Passwords without email:
#   * Recovery key — shown once at sign-up (and on demand in Account); stored only as a hash.
#     Forgot password = email + recovery key + new password.  A used key is replaced by a new one.
#   * Change password — signed-in users (current password + new password).
#   * No key? The user sends a reset request; the admin sees it in Admin -> Users and sets a
#     temporary password, shared by WhatsApp/phone.
# ---------------------------------------------------------------------------
KEY_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"      # no 0/O/1/I confusion
MAX_KEY_FAILS = 5                                      # wrong keys per email per hour


def _pw_db():
    con = _db()
    if db.IS_MYSQL:
        return con
    cols = {r[1] for r in con.execute("PRAGMA table_info(users)")}
    if "recovery_hash" not in cols:
        con.execute("ALTER TABLE users ADD COLUMN recovery_hash TEXT")
    con.execute("CREATE TABLE IF NOT EXISTS reset_fails (email TEXT, ts REAL)")
    con.execute("CREATE TABLE IF NOT EXISTS reset_requests (id INTEGER PRIMARY KEY AUTOINCREMENT, "
                "ts TEXT DEFAULT CURRENT_TIMESTAMP, email TEXT, contact TEXT, status TEXT DEFAULT 'open')")
    return con


def _key_hash(email, key):
    k = re.sub(r"[^A-Z0-9]", "", (key or "").upper())
    return hashlib.sha256(f"{email}:{k}".encode()).hexdigest()


def new_recovery_key(email):
    """Create (or replace) the recovery key for `email` and return it in XXXX-XXXX-XXXX-XXXX form."""
    email = (email or "").strip().lower()
    raw = "".join(secrets.choice(KEY_ALPHABET) for _ in range(16))
    key = "-".join(raw[i:i + 4] for i in range(0, 16, 4))
    with _pw_db() as con:
        con.execute("UPDATE users SET recovery_hash=? WHERE email=?", (_key_hash(email, key), email))
    request_sync()
    return key


def has_recovery_key(email):
    with _pw_db() as con:
        row = con.execute("SELECT recovery_hash FROM users WHERE email=?", ((email or "").lower(),)).fetchone()
    return bool(row and row[0])


def key_html(key, intro="Your recovery key"):
    return (f'<div class="cs-authmsg cs-ok" style="text-align:left"><b>🔑 {intro}:</b>'
            f'<div style="font:700 20px/1.6 monospace;letter-spacing:2px;margin:6px 0;user-select:all">{key}</div>'
            "Save it somewhere safe (screenshot or notes). It is the only way to reset your password yourself "
            "if you forget it. It is shown only once.</div>")


def reset_with_key(email, key, new_pw, confirm):
    """Returns (user, new_key, error)."""
    email = (email or "").strip().lower()
    if not _EMAIL_RE.match(email):
        return None, None, "Please enter your account email."
    if not (key or "").strip():
        return None, None, "Please enter your recovery key."
    if len(new_pw or "") < 8:
        return None, None, "New password must be at least 8 characters."
    if new_pw != confirm:
        return None, None, "Passwords do not match."
    now = time.time()
    with _pw_db() as con:
        con.execute("DELETE FROM reset_fails WHERE ts < ?", (now - 3600,))
        fails = con.execute("SELECT COUNT(*) FROM reset_fails WHERE email=?", (email,)).fetchone()[0]
        if fails >= MAX_KEY_FAILS:
            return None, None, "Too many wrong attempts. Please wait an hour, or ask the admin to reset it."
        row = con.execute("SELECT name, active, recovery_hash FROM users WHERE email=?", (email,)).fetchone()
        if not row or not row[2] or not hmac.compare_digest(row[2], _key_hash(email, key)):
            con.execute("INSERT INTO reset_fails (email, ts) VALUES (?,?)", (email, now))
            return None, None, "Email or recovery key is not correct."
        if row[1] == 0:
            return None, None, "This account has been disabled. Please contact CS Software Solutions."
        salt = secrets.token_hex(16)
        con.execute("UPDATE users SET salt=?, pw=?, last_login=CURRENT_TIMESTAMP WHERE email=?",
                    (salt, _hash(new_pw, salt), email))
        con.execute("DELETE FROM reset_fails WHERE email=?", (email,))
        con.execute("UPDATE reset_requests SET status='done' WHERE email=? AND status='open'", (email,))
    return {"email": email, "name": row[0]}, new_recovery_key(email), None


def change_password(user, old_pw, new_pw, confirm):
    """Signed-in user changes their password. Returns (ok, message)."""
    if not user:
        return False, "Please sign in first."
    _, err = check_user(user["email"], old_pw)
    if err:
        return False, "Current password is not correct."
    if len(new_pw or "") < 8:
        return False, "New password must be at least 8 characters."
    if new_pw != confirm:
        return False, "New passwords do not match."
    salt = secrets.token_hex(16)
    with _db() as con:
        con.execute("UPDATE users SET salt=?, pw=? WHERE email=?", (salt, _hash(new_pw, salt), user["email"]))
    request_sync()
    return True, "Password changed."


def request_admin_reset(email, contact):
    """User without a recovery key asks the admin for a reset. Neutral answer either way."""
    email, contact = (email or "").strip().lower(), (contact or "").strip()[:120]
    if not _EMAIL_RE.match(email):
        return False, "Please enter your account email."
    if not contact:
        return False, "Please add a WhatsApp / phone number so the admin can send you a temporary password."
    with _pw_db() as con:
        known = con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone()
        dup = con.execute("SELECT 1 FROM reset_requests WHERE email=? AND status='open'", (email,)).fetchone()
        if known and not dup:
            con.execute("INSERT INTO reset_requests (email, contact) VALUES (?,?)", (email, contact))
    request_sync()
    return True, ("Request sent. If an account exists for this email, the admin will contact you on "
                  f"{html.escape(contact)} with a temporary password (usually within a few hours).")


RESET_REQ_COLS = ["When (UTC)", "Email", "Contact", "Status"]


def admin_reset_requests(user):
    _need_admin(user)
    with _pw_db() as con:
        rows = con.execute("SELECT ts, email, contact, status FROM reset_requests "
                           "ORDER BY status='done', id DESC LIMIT 100").fetchall()
    return [[(r[0] or "")[:16], r[1], r[2], r[3]] for r in rows]


# ---------------------------------------------------------------------------
# Forgot password by email: 6-digit code (15 min, 5 tries, max 3 codes per hour). Works only when
# email is configured (mailer.ENABLED); the recovery key and admin request above always work.
# ---------------------------------------------------------------------------
RESET_MINUTES = 15


def _code_db():
    con = _pw_db()
    if db.IS_MYSQL:
        return con
    con.execute("CREATE TABLE IF NOT EXISTS reset_codes (email TEXT, code_hash TEXT, expires REAL, "
                "attempts INTEGER DEFAULT 0, created REAL)")
    return con


# Sign-up with a confirmed email: the account is created only after the 6-digit code sent to the address is entered
# (anyone could otherwise sign up with someone else's address). Off when email is not set up (sign-up then works as
# before) or with SIGNUP_VERIFY=0.
SIGNUP_VERIFY = os.getenv("SIGNUP_VERIFY", "1") != "0"


def _signup_db():
    con = _pw_db()
    if db.IS_MYSQL:
        return con
    con.execute("CREATE TABLE IF NOT EXISTS signup_codes (email TEXT, name TEXT, salt TEXT, pw TEXT, code_hash TEXT, "
                "expires REAL, attempts INTEGER DEFAULT 0, created REAL)")
    return con


def verify_signups():
    import mailer as M
    return SIGNUP_VERIFY and M.ENABLED


def start_signup(name, email, password, confirm):
    """Checks the form and emails a code. Returns (ok, message); nothing is created yet."""
    import mailer as M
    name, email = (name or "").strip(), (email or "").strip().lower()
    if not name:
        return False, "Please enter your name."
    if not _EMAIL_RE.match(email):
        return False, "Please enter a valid email address."
    if len(password or "") < 8:
        return False, "Password must be at least 8 characters."
    if password != confirm:
        return False, "Passwords do not match."
    now = time.time()
    with _signup_db() as con:
        if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone():
            return False, "An account with this email already exists. Please log in."
        if con.execute("SELECT COUNT(*) FROM signup_codes WHERE email=? AND created > ?",
                       (email, now - 3600)).fetchone()[0] >= 3:
            return False, "Too many sign-up codes for this address. Please wait an hour and try again."
        code = f"{secrets.randbelow(10 ** 6):06d}"
        salt = secrets.token_hex(16)
        con.execute("INSERT INTO signup_codes (email, name, salt, pw, code_hash, expires, created) VALUES "
                    "(?,?,?,?,?,?,?)", (email, name, salt, _hash(password, salt),
                                        hashlib.sha256(f"signup:{email}:{code}".encode()).hexdigest(),
                                        now + RESET_MINUTES * 60, now))
        con.execute("DELETE FROM signup_codes WHERE expires < ?", (now - 86400,))
    ok, _ = M.send_signup_code(email, name, code, RESET_MINUTES)
    if not ok:
        #return False, "We could not send the email right now. Please try again in a few minutes."
        return False, _
    return True, (f"We have sent a 6-digit code to <b>{html.escape(email)}</b>. Enter it below to create your "
                  f"account (it expires in {RESET_MINUTES} minutes; check your spam folder too).")


def confirm_signup(email, code):
    """Returns (user, error). Creates the account from the details given at start_signup."""
    email, code = (email or "").strip().lower(), (code or "").strip()
    now = time.time()
    with _signup_db() as con:
        row = con.execute("SELECT rowid, name, salt, pw, code_hash, attempts FROM signup_codes WHERE email=? AND "
                          "expires > ? ORDER BY created DESC LIMIT 1", (email, now)).fetchone()
        if not row:
            return None, "The code has expired or was not requested. Press Create account again for a new code."
        if row[5] >= 5:
            return None, "Too many wrong attempts. Press Create account again for a new code."
        if not hmac.compare_digest(row[4], hashlib.sha256(f"signup:{email}:{code}".encode()).hexdigest()):
            con.execute("UPDATE signup_codes SET attempts = attempts + 1 WHERE rowid=?", (row[0],))
            return None, "That code is not correct."
        if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone():
            return None, "An account with this email already exists. Please log in."
        con.execute("INSERT INTO users (email, name, salt, pw, last_login) VALUES (?,?,?,?,CURRENT_TIMESTAMP)",
                    (email, row[1], row[2], row[3]))
        con.execute("DELETE FROM signup_codes WHERE email=?", (email,))
    request_sync()
    return {"email": email, "name": row[1]}, None


def request_reset(email):
    """Sends a code. Same neutral message whether or not the email has an account."""
    import mailer as M
    email = (email or "").strip().lower()
    if not M.ENABLED:
        return False, "Email reset is not available right now — use your recovery key or ask the admin below."
    if not _EMAIL_RE.match(email):
        return False, "Please enter a valid email address."
    neutral = (f"If an account exists for {html.escape(email)}, a 6-digit code has been sent. It expires in "
               f"{RESET_MINUTES} minutes — check your spam folder too.")
    now = time.time()
    with _code_db() as con:
        row = con.execute("SELECT name, active FROM users WHERE email=?", (email,)).fetchone()
        recent = con.execute("SELECT COUNT(*) FROM reset_codes WHERE email=? AND created > ?",
                             (email, now - 3600)).fetchone()[0]
        if not row or row[1] == 0:
            return True, neutral
        if recent >= 3:
            return False, "Too many reset requests. Please wait an hour and try again."
        code = f"{secrets.randbelow(10 ** 6):06d}"
        con.execute("INSERT INTO reset_codes (email, code_hash, expires, created) VALUES (?,?,?,?)",
                    (email, hashlib.sha256(f"{email}:{code}".encode()).hexdigest(), now + RESET_MINUTES * 60, now))
        con.execute("DELETE FROM reset_codes WHERE expires < ?", (now - 86400,))
    ok, _ = M.send_reset_code(email, row[0], code, RESET_MINUTES)
    if not ok:
        return False, "We could not send the email right now — use your recovery key or ask the admin below."
    return True, neutral


def confirm_reset(email, code, new_pw, confirm):
    """Returns (user, error)."""
    email, code = (email or "").strip().lower(), (code or "").strip()
    if len(new_pw or "") < 8:
        return None, "New password must be at least 8 characters."
    if new_pw != confirm:
        return None, "Passwords do not match."
    now = time.time()
    with _code_db() as con:
        row = con.execute("SELECT rowid, code_hash, attempts FROM reset_codes WHERE email=? AND expires > ? "
                          "ORDER BY created DESC LIMIT 1", (email, now)).fetchone()
        if not row:
            return None, "The code has expired or was not requested. Please request a new code."
        if row[2] >= 5:
            return None, "Too many wrong attempts. Please request a new code."
        if not hmac.compare_digest(row[1], hashlib.sha256(f"{email}:{code}".encode()).hexdigest()):
            con.execute("UPDATE reset_codes SET attempts = attempts + 1 WHERE rowid=?", (row[0],))
            return None, "That code is not correct."
        salt = secrets.token_hex(16)
        con.execute("UPDATE users SET salt=?, pw=?, last_login=CURRENT_TIMESTAMP WHERE email=?",
                    (salt, _hash(new_pw, salt), email))
        con.execute("DELETE FROM reset_codes WHERE email=?", (email,))
        con.execute("UPDATE reset_requests SET status='done' WHERE email=? AND status='open'", (email,))
        u = con.execute("SELECT name, active FROM users WHERE email=?", (email,)).fetchone()
    request_sync()
    if not u or u[1] == 0:
        return None, "This account is disabled."
    return {"email": email, "name": u[0]}, None


def check_reset(email, code_or_key):
    """Forgot password, step 1: is this email code (6 digits) or recovery key right? It is not used up here (the
    new password step checks it again and uses it); a wrong one counts towards the same limits.
    Returns (ok, error)."""
    email, given = (email or "").strip().lower(), (code_or_key or "").strip()
    if not _EMAIL_RE.match(email):
        return False, "Please enter your account email."
    if not given:
        return False, "Please enter the 6-digit code from the email, or your recovery key."
    now = time.time()
    if re.fullmatch(r"\d{6}", given):
        with _code_db() as con:
            row = con.execute("SELECT rowid, code_hash, attempts FROM reset_codes WHERE email=? AND expires > ? "
                              "ORDER BY created DESC LIMIT 1", (email, now)).fetchone()
            if not row:
                return False, "The code has expired or was not requested. Please request a new code."
            if row[2] >= 5:
                return False, "Too many wrong attempts. Please request a new code."
            if not hmac.compare_digest(row[1], hashlib.sha256(f"{email}:{given}".encode()).hexdigest()):
                con.execute("UPDATE reset_codes SET attempts = attempts + 1 WHERE rowid=?", (row[0],))
                return False, "That code is not correct."
        return True, None
    with _pw_db() as con:
        con.execute("DELETE FROM reset_fails WHERE ts < ?", (now - 3600,))
        if con.execute("SELECT COUNT(*) FROM reset_fails WHERE email=?", (email,)).fetchone()[0] >= MAX_KEY_FAILS:
            return False, "Too many wrong attempts. Please wait an hour, or ask the admin to reset it."
        row = con.execute("SELECT active, recovery_hash FROM users WHERE email=?", (email,)).fetchone()
        if not row or not row[1] or not hmac.compare_digest(row[1], _key_hash(email, given)):
            con.execute("INSERT INTO reset_fails (email, ts) VALUES (?,?)", (email, now))
            return False, "Email or recovery key is not correct."
        if row[0] == 0:
            return False, "This account has been disabled. Please contact CS Software Solutions."
    return True, None


def reset_any(email, code_or_key, new_pw, confirm):
    """One form for both: a 6-digit number is an email code, anything else a recovery key.
    Returns (user, new_key_or_None, error)."""
    if re.fullmatch(r"\s*\d{6}\s*", code_or_key or ""):
        user, err = confirm_reset(email, code_or_key, new_pw, confirm)
        return user, None, err
    return reset_with_key(email, code_or_key, new_pw, confirm)