"""SQLite-backed persistence for accounts, subscriptions, orders, usage.

Uses the same env-driven get_db_path() as the rest of the system (memory.db in the
persistent bucket on the Space), so subscription state survives rebuilds exactly
like memory does — per huggingface-spaces-hosting.md's persistence rule.

Tables are additive (CREATE TABLE IF NOT EXISTS) so this never disturbs the
existing vault/memory tables in the same file.
"""
from __future__ import annotations

import sqlite3
import time
import uuid

from backend.services.usb_monitor import get_db_path


def _conn():
    c = sqlite3.connect(get_db_path())
    c.execute("PRAGMA journal_mode=WAL")
    c.row_factory = sqlite3.Row
    return c


def init_db() -> None:
    with _conn() as c:
        c.execute("""CREATE TABLE IF NOT EXISTS accounts (
            user_id TEXT PRIMARY KEY,
            email TEXT UNIQUE,
            created_at REAL
        )""")
        c.execute("""CREATE TABLE IF NOT EXISTS subscriptions (
            user_id TEXT PRIMARY KEY,
            plan_id TEXT NOT NULL,
            status TEXT NOT NULL,            -- active | provisioning | past_due | none
            activated_at REAL,
            provisioning_until REAL,         -- when disclosed delay ends (0 if none)
            updated_at REAL
        )""")
        c.execute("""CREATE TABLE IF NOT EXISTS orders (
            order_id TEXT PRIMARY KEY,
            user_id TEXT NOT NULL,
            plan_id TEXT NOT NULL,
            amount_cents INTEGER NOT NULL,
            currency TEXT NOT NULL,
            state TEXT NOT NULL,             -- pending | confirmed | failed
            processor TEXT,
            processor_ref TEXT,              -- processor's payment/session id
            created_at REAL,
            updated_at REAL
        )""")
        # Idempotency ledger: every processor event id we've already applied, so a
        # webhook arriving late / out of order / more than once never double-grants.
        c.execute("""CREATE TABLE IF NOT EXISTS payment_events (
            event_id TEXT PRIMARY KEY,
            order_id TEXT,
            received_at REAL
        )""")
        c.execute("""CREATE TABLE IF NOT EXISTS usage_counters (
            user_id TEXT NOT NULL,
            meter TEXT NOT NULL,             -- e.g. 'model_calls'
            window_start REAL NOT NULL,      -- start of the current reset window
            count INTEGER NOT NULL DEFAULT 0,
            PRIMARY KEY (user_id, meter)
        )""")
        # Abuse/rate-limit hit log: a sliding-window record of requests per bucket
        # (per-user or per-IP) so a route can't be hammered faster than the cap
        # accounting catches up (skill: abuse protection on billing/AI-heavy routes).
        c.execute("""CREATE TABLE IF NOT EXISTS rate_hits (
            bucket TEXT NOT NULL,            -- e.g. 'model_gen:user:alice' / ':ip:1.2.3.4'
            ts REAL NOT NULL
        )""")
        c.execute("CREATE INDEX IF NOT EXISTS ix_rate_hits ON rate_hits(bucket, ts)")
        # Audit log for privileged/operator actions (plan grants, flag flips, etc.).
        # Every privileged action leaves a queryable record, per the skill's
        # operator-audit-logging requirement — never a silent success.
        c.execute("""CREATE TABLE IF NOT EXISTS audit_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ts REAL NOT NULL,
            actor TEXT,                      -- who/what performed it
            action TEXT NOT NULL,            -- e.g. 'plan_grant', 'flag_set'
            target TEXT,                     -- affected user/order/flag
            detail TEXT                      -- JSON blob of specifics
        )""")
        # Operator kill-switch / feature flags. A risky surface (AR model-gen, the
        # billing webhook) can be disabled without a full rollback.
        c.execute("""CREATE TABLE IF NOT EXISTS feature_flags (
            name TEXT PRIMARY KEY,
            enabled INTEGER NOT NULL DEFAULT 1,
            updated_at REAL
        )""")
        c.commit()


# ─── Rate limiting (sliding window) ────────────────────────────────────────────

def rate_check(bucket: str, limit: int, window_seconds: float) -> dict:
    """Record this hit and report whether the bucket is over its limit in the
    trailing window. Atomic-enough for a single-process sidecar; the Space's real
    per-user isolation is the deployment story, this is the in-process guard."""
    now = time.time()
    with _conn() as c:
        c.execute("DELETE FROM rate_hits WHERE ts < ?", (now - window_seconds,))
        n = c.execute("SELECT COUNT(*) FROM rate_hits WHERE bucket=? AND ts >= ?",
                      (bucket, now - window_seconds)).fetchone()[0]
        if n >= limit:
            c.commit()
            return {"allowed": False, "count": n, "limit": limit,
                    "retry_after": window_seconds}
        c.execute("INSERT INTO rate_hits (bucket, ts) VALUES (?, ?)", (bucket, now))
        c.commit()
        return {"allowed": True, "count": n + 1, "limit": limit}


# ─── Audit log ─────────────────────────────────────────────────────────────────

def record_audit(action: str, actor: str | None = None, target: str | None = None,
                 detail: str | None = None) -> None:
    with _conn() as c:
        c.execute("INSERT INTO audit_log (ts, actor, action, target, detail) "
                  "VALUES (?,?,?,?,?)", (time.time(), actor, action, target, detail))
        c.commit()


def read_audit(limit: int = 100) -> list[dict]:
    with _conn() as c:
        rows = c.execute("SELECT * FROM audit_log ORDER BY id DESC LIMIT ?",
                         (limit,)).fetchall()
        return [dict(r) for r in rows]


# ─── Feature flags / kill-switch ───────────────────────────────────────────────

def flag_enabled(name: str, default: bool = True) -> bool:
    with _conn() as c:
        row = c.execute("SELECT enabled FROM feature_flags WHERE name=?",
                        (name,)).fetchone()
        return bool(row[0]) if row else default


def set_flag(name: str, enabled: bool) -> None:
    with _conn() as c:
        c.execute("INSERT INTO feature_flags (name, enabled, updated_at) VALUES (?,?,?) "
                  "ON CONFLICT(name) DO UPDATE SET enabled=excluded.enabled, "
                  "updated_at=excluded.updated_at",
                  (name, 1 if enabled else 0, time.time()))
        c.commit()


# ─── Accounts ──────────────────────────────────────────────────────────────────
def upsert_account(user_id: str, email: str | None = None) -> None:
    with _conn() as c:
        c.execute(
            "INSERT INTO accounts (user_id, email, created_at) VALUES (?,?,?) "
            "ON CONFLICT(user_id) DO UPDATE SET email=COALESCE(excluded.email, accounts.email)",
            (user_id, email, time.time()))
        c.commit()


# ─── Subscriptions ─────────────────────────────────────────────────────────────
def get_subscription(user_id: str) -> dict | None:
    with _conn() as c:
        r = c.execute("SELECT * FROM subscriptions WHERE user_id=?", (user_id,)).fetchone()
        return dict(r) if r else None


def set_subscription(user_id: str, plan_id: str, status: str,
                     provisioning_until: float = 0.0) -> None:
    now = time.time()
    with _conn() as c:
        c.execute(
            "INSERT INTO subscriptions (user_id, plan_id, status, activated_at, "
            "provisioning_until, updated_at) VALUES (?,?,?,?,?,?) "
            "ON CONFLICT(user_id) DO UPDATE SET plan_id=excluded.plan_id, "
            "status=excluded.status, activated_at=excluded.activated_at, "
            "provisioning_until=excluded.provisioning_until, updated_at=excluded.updated_at",
            (user_id, plan_id, status,
             now if status == "active" else None, provisioning_until, now))
        c.commit()


# ─── Orders ────────────────────────────────────────────────────────────────────
def create_order(user_id: str, plan_id: str, amount_cents: int, currency: str,
                 processor: str) -> str:
    order_id = "ord_" + uuid.uuid4().hex
    now = time.time()
    with _conn() as c:
        c.execute(
            "INSERT INTO orders (order_id, user_id, plan_id, amount_cents, currency, "
            "state, processor, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?)",
            (order_id, user_id, plan_id, amount_cents, currency, "pending", processor, now, now))
        c.commit()
    return order_id


def get_order(order_id: str) -> dict | None:
    with _conn() as c:
        r = c.execute("SELECT * FROM orders WHERE order_id=?", (order_id,)).fetchone()
        return dict(r) if r else None


def set_order_state(order_id: str, state: str, processor_ref: str | None = None) -> None:
    with _conn() as c:
        c.execute("UPDATE orders SET state=?, processor_ref=COALESCE(?, processor_ref), "
                  "updated_at=? WHERE order_id=?",
                  (state, processor_ref, time.time(), order_id))
        c.commit()


def find_orders(user_id: str) -> list[dict]:
    with _conn() as c:
        rows = c.execute("SELECT * FROM orders WHERE user_id=? ORDER BY created_at DESC",
                         (user_id,)).fetchall()
        return [dict(r) for r in rows]


# ─── Idempotency ───────────────────────────────────────────────────────────────
def event_already_applied(event_id: str) -> bool:
    with _conn() as c:
        return c.execute("SELECT 1 FROM payment_events WHERE event_id=?",
                         (event_id,)).fetchone() is not None


def mark_event_applied(event_id: str, order_id: str) -> None:
    with _conn() as c:
        c.execute("INSERT OR IGNORE INTO payment_events (event_id, order_id, received_at) "
                  "VALUES (?,?,?)", (event_id, order_id, time.time()))
        c.commit()


# ─── Usage counters ────────────────────────────────────────────────────────────
def get_usage(user_id: str, meter: str, window_seconds: float) -> tuple[int, float]:
    """Returns (count_in_current_window, window_start). Rolls the window over
    lazily when it has expired."""
    now = time.time()
    with _conn() as c:
        r = c.execute("SELECT window_start, count FROM usage_counters WHERE user_id=? AND meter=?",
                      (user_id, meter)).fetchone()
        if r is None:
            c.execute("INSERT INTO usage_counters (user_id, meter, window_start, count) VALUES (?,?,?,0)",
                      (user_id, meter, now)); c.commit()
            return 0, now
        if now - r["window_start"] >= window_seconds:
            c.execute("UPDATE usage_counters SET window_start=?, count=0 WHERE user_id=? AND meter=?",
                      (now, user_id, meter)); c.commit()
            return 0, now
        return r["count"], r["window_start"]


def increment_usage(user_id: str, meter: str, window_seconds: float, by: int = 1) -> int:
    count, _ = get_usage(user_id, meter, window_seconds)
    new = count + by
    with _conn() as c:
        c.execute("UPDATE usage_counters SET count=? WHERE user_id=? AND meter=?",
                  (new, user_id, meter)); c.commit()
    return new