"""
localVaultMirror.py — keeps the cloud copy of a device vault up to date.
=======================================================================
docs/plans/obsidian-vault-connector.md §6.4 (S4); used by localVault.py
(``mirror_sync`` / ``mirror_drop``) when the vault's policy is
``storage: mirror``. Bulk data goes over authenticated HTTPS to the Vault API
(through the server's /api/v1/vault-api/mirror/<vault_id>/ proxy); the socket
only coordinates (src/vaultMirror.ts).

What changed is NOT detected here: the vault index (index.sqlite, built by the
engine from fs.watch batches plus the 10-minute stat-only reconciliation, which
hashes only stat-changed files) already holds ``(path, size, mtime, sha256)``
for every visible file, under the same path policy as S1. This module diffs
that manifest against its own ledger of what the copy will hold once every
queued op is published:

  ~/.melaya-runner/vaults/<vault_id>/mirror.sqlite
    files    path → sha256, size: the copy's state once the outbox is published
    outbox   ordered ops (seq, put | delete, path, sha256, size, mtime_ms): the
             durable queue; a delete is a tombstone with its own sequence
             number, so a retried upload can never resurrect the note
    cfg      stat cache of the allow-listed settings files (config folder)
    meta     next_seq, replica_id, journal_seq

One sync round: plan → begin → for each batch of ops: upload the contents the
copy lacks (resumable chunks with per-chunk and whole-file checksums), send the
ops → publish (with the device change journal since the last round). The
outbox is trimmed only after a publish is acknowledged, so an interrupted
round resumes where the copy's staging area stopped.

Never read into memory as a whole: files are streamed chunk by chunk. Never
logged: note content, paths, the capability.
"""
from __future__ import annotations

import hashlib
import json
import os
import sqlite3
import stat as _stat
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any, Callable, Iterable


def _runner_user_agent() -> str:
    # Cloudflare's browser integrity check refuses urllib's default
    # "Python-urllib/x.y" agent (403, error 1010) before the request reaches
    # Melaya, so every upload must name the runner.
    try:
        pkg = json.loads((Path(__file__).resolve().parent / "package.json").read_text(encoding="utf-8"))
        return f"melaya-runner/{pkg.get('version') or 'dev'} (vault-mirror)"
    except Exception:
        return "melaya-runner (vault-mirror)"


_USER_AGENT = _runner_user_agent()

CONFIG_FILES = (
    "app.json", "daily-notes.json", "templates.json", "zk-prefixer.json", "bookmarks.json",
    "types.json", "workspace.json", "workspaces.json", "core-plugins.json",
    "community-plugins.json", "appearance.json",
)
MAX_FILE_BYTES = 512 * 1024 * 1024
OPS_BATCH = 500
JOURNAL_BATCH = 2000
FRESH_JOURNAL_BACKLOG = 2000
MAX_BUSY_RETRIES = 6
DEFAULT_CHUNK = 4 * 1024 * 1024


class MirrorError(Exception):
    """``code``: auth | quota | retry | error (fixed, short)."""

    def __init__(self, code: str, message: str = ""):
        super().__init__(code)
        self.code = code
        self.message = message or code


class FileChanged(Exception):
    """The file no longer matches the content queued for it: rescan, then retry."""

    def __init__(self, path: str):
        super().__init__(path)
        self.path = path


# ── transport ─────────────────────────────────────────────────────────────

Response = tuple[int, dict, dict]


class HttpTransport:
    """HTTPS to ``base`` (…/api/v1/vault-api/mirror/<vault_id>), ``Authorization:
    Mirror <capability>``. Paths are relative: begin, ops, publish,
    blobs/check, blobs/<sha256>."""

    def __init__(self, base: str, token: str, timeout: float = 120.0):
        self.base = base.rstrip("/")
        self._token = token
        self.timeout = timeout

    def request(self, method: str, rel: str, *, json_body: Any = None, data: bytes | None = None,
                params: dict | None = None, headers: dict | None = None) -> Response:
        url = f"{self.base}/{rel}"
        if params:
            url += "?" + urllib.parse.urlencode(params)
        h = {"Authorization": f"Mirror {self._token}", "User-Agent": _USER_AGENT, **(headers or {})}
        body = data
        if json_body is not None:
            body = json.dumps(json_body).encode("utf-8")
            h["Content-Type"] = "application/json"
        elif data is not None:
            h["Content-Type"] = "application/octet-stream"
        req = urllib.request.Request(url, data=body, method=method, headers=h)
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                raw = r.read()
                return r.status, _json(raw), {k.lower(): v for k, v in r.headers.items()}
        except urllib.error.HTTPError as e:
            raw = e.read() if hasattr(e, "read") else b""
            return e.code, _json(raw), {k.lower(): v for k, v in (e.headers.items() if e.headers else [])}
        except (urllib.error.URLError, OSError, TimeoutError) as e:
            raise MirrorError("retry", f"network ({type(e).__name__})") from None


def _json(raw: bytes) -> dict:
    try:
        v = json.loads(raw.decode("utf-8") or "{}")
        return v if isinstance(v, dict) else {}
    except Exception:
        return {}


# ── ledger ────────────────────────────────────────────────────────────────

_SCHEMA = """
CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value TEXT);
CREATE TABLE IF NOT EXISTS files(path TEXT PRIMARY KEY, sha256 TEXT NOT NULL, size INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS outbox(seq INTEGER PRIMARY KEY, op TEXT NOT NULL, path TEXT NOT NULL,
                                  sha256 TEXT, size INTEGER, mtime_ms INTEGER);
CREATE INDEX IF NOT EXISTS outbox_path ON outbox(path);
CREATE TABLE IF NOT EXISTS cfg(path TEXT PRIMARY KEY, size INTEGER, mtime_ns INTEGER, sha256 TEXT);
"""


class Ledger:
    def __init__(self, path: str):
        os.makedirs(os.path.dirname(path), exist_ok=True)
        self.con = sqlite3.connect(path, isolation_level=None, check_same_thread=False, timeout=30)
        self.con.execute("PRAGMA journal_mode=WAL")
        self.con.execute("PRAGMA synchronous=FULL")  # tombstones must survive a crash
        self.con.executescript(_SCHEMA)
        try:
            os.chmod(path, 0o600)
        except OSError:
            pass

    def close(self) -> None:
        try:
            self.con.close()
        except sqlite3.Error:
            pass

    def get(self, key: str, default: str = "") -> str:
        row = self.con.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone()
        return row[0] if row else default

    def set(self, key: str, value: Any) -> None:
        self.con.execute("INSERT INTO meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
                         (key, str(value)))

    def pending(self) -> int:
        return int(self.con.execute("SELECT COUNT(*) FROM outbox").fetchone()[0])

    def enqueue(self, ops: Iterable[tuple[str, str, str | None, int | None, int | None]]) -> int:
        """Append ops in order. An op supersedes every queued op of the same
        path (each op states the whole file), so the queue stays one op per
        path, and the sequence order of what remains is preserved."""
        n = 0
        self.con.execute("BEGIN IMMEDIATE")
        try:
            seq = int(self.get("next_seq", "1") or 1)
            for op, path, sha, size, mtime_ms in ops:
                self.con.execute("DELETE FROM outbox WHERE path=?", (path,))
                self.con.execute("INSERT INTO outbox VALUES(?,?,?,?,?,?)", (seq, op, path, sha, size, mtime_ms))
                if op == "delete":
                    self.con.execute("DELETE FROM files WHERE path=?", (path,))
                else:
                    self.con.execute("INSERT OR REPLACE INTO files VALUES(?,?,?)", (path, sha, size))
                seq += 1
                n += 1
            self.set("next_seq", seq)
            self.con.execute("COMMIT")
        except BaseException:
            self.con.execute("ROLLBACK")
            raise
        return n

    def ack(self, published_seq: int) -> None:
        self.con.execute("DELETE FROM outbox WHERE seq<=?", (int(published_seq),))

    def reset(self, replica_id: str) -> None:
        """The copy starts over (new copy, or it lost track of us): forget what
        it was believed to hold; the next plan queues every file again."""
        self.con.execute("BEGIN IMMEDIATE")
        try:
            self.con.execute("DELETE FROM files")
            self.con.execute("DELETE FROM outbox")
            self.con.execute("DELETE FROM meta WHERE key='journal_seq'")
            self.set("replica_id", replica_id)
            self.con.execute("COMMIT")
        except BaseException:
            self.con.execute("ROLLBACK")
            raise


# ── mirror ────────────────────────────────────────────────────────────────

class Mirror:
    def __init__(self, vault_id: str, policy: Any, index_path: str, ledger_path: str, *,
                 max_file_bytes: int = MAX_FILE_BYTES, log: Callable[[str], None] | None = None):
        self.vault_id = vault_id
        self.policy = policy          # engine PathPolicy: the S1 path rules
        self.index_path = index_path
        self.ledger = Ledger(ledger_path)
        self.max_file_bytes = max_file_bytes
        self.log = log or (lambda m: None)
        self.skipped = 0

    def close(self) -> None:
        self.ledger.close()

    # ── what the copy should hold ──
    def _index(self) -> sqlite3.Connection:
        uri = Path(self.index_path).resolve().as_uri() + "?mode=ro"
        return sqlite3.connect(uri, uri=True, timeout=30)

    def _visible(self, path: str) -> bool:
        """Defence in depth: the index already applies the path policy; a
        stale row still never names an excluded path."""
        try:
            from_policy = self.policy.is_excluded(path)
        except Exception:
            return False
        if from_policy:
            return False
        segs = path.split("/")
        return not any(s in ("", ".", "..") or s.startswith(".") for s in segs)

    def _config_files(self) -> dict[str, tuple[str, int, int]]:
        """Allow-listed settings files of the config folder (read-only on the
        copy), hashed only when their stat moved."""
        out: dict[str, tuple[str, int, int]] = {}
        cfg_dir = self.policy.config_dir
        for name in CONFIG_FILES:
            absolute = self.policy.config_file(name)
            if not absolute:
                continue
            try:
                st = os.lstat(absolute)
            except OSError:
                continue
            if not _stat.S_ISREG(st.st_mode) or st.st_size > 2 * 1024 * 1024:
                continue
            rel = f"{cfg_dir}/{name}"
            row = self.ledger.con.execute("SELECT size, mtime_ns, sha256 FROM cfg WHERE path=?", (rel,)).fetchone()
            if row and row[0] == st.st_size and row[1] == st.st_mtime_ns:
                sha = row[2]
            else:
                try:
                    with open(absolute, "rb") as f:
                        sha = hashlib.sha256(f.read()).hexdigest()
                except OSError:
                    continue
                self.ledger.con.execute("INSERT OR REPLACE INTO cfg VALUES(?,?,?,?)", (rel, st.st_size, st.st_mtime_ns, sha))
            out[rel] = (sha, st.st_size, st.st_mtime_ns // 1_000_000)
        return out

    def plan(self) -> dict:
        """Queue the ops that take the copy from the ledger's state to the
        index's: deletes first (a file may replace a folder of the same name),
        then puts. Metadata only; no file is read except changed settings."""
        desired: dict[str, tuple[str, int, int]] = {}
        skipped = 0
        con = self._index()
        try:
            for path, size, mtime_ns, sha in con.execute("SELECT path, size, mtime_ns, sha256 FROM manifest"):
                if not sha or not isinstance(path, str) or not self._visible(path):
                    continue
                if int(size or 0) > self.max_file_bytes:
                    skipped += 1
                    continue
                desired[path] = (sha, int(size or 0), int(mtime_ns or 0) // 1_000_000)
        finally:
            con.close()
        desired.update(self._config_files())
        known = {p: (s, z) for p, s, z in self.ledger.con.execute("SELECT path, sha256, size FROM files")}
        deletes = sorted(p for p in known if p not in desired)
        puts = sorted(p for p, v in desired.items() if known.get(p) != (v[0], v[1]))
        ops = [("delete", p, None, None, None) for p in deletes]
        ops += [("put", p, desired[p][0], desired[p][1], desired[p][2]) for p in puts]
        n = self.ledger.enqueue(ops) if ops else 0
        self.skipped = skipped
        return {"enqueued": n, "pending": self.ledger.pending(), "skipped": skipped}

    def _device_state(self) -> tuple[int, int, int]:
        """(index_rev, journal head, journal floor) of the device index."""
        con = self._index()
        try:
            meta = dict(con.execute("SELECT key, value FROM meta WHERE key IN ('index_rev','journal_floor')").fetchall())
            head = con.execute("SELECT MAX(seq) FROM journal").fetchone()[0] or 0
            return int(meta.get("index_rev") or 0), int(head), int(meta.get("journal_floor") or 0)
        finally:
            con.close()

    def _journal(self, since: int) -> list[dict]:
        con = self._index()
        try:
            rows = con.execute("SELECT seq, event_id, op, path, old_path, revision, origin, write_id, ts FROM journal "
                               "WHERE seq > ? ORDER BY seq LIMIT ?", (since, JOURNAL_BATCH)).fetchall()
        finally:
            con.close()
        keys = ("seq", "event_id", "op", "path", "old_path", "revision", "origin", "write_id", "ts")
        return [dict(zip(keys, r)) for r in rows]

    # ── one sync round ──
    def sync(self, t: Any, *, budget_s: float = 600.0, sleep: Callable[[float], None] = time.sleep) -> dict:
        deadline = time.monotonic() + max(5.0, budget_s)
        try:
            return self._sync(t, deadline, sleep)
        except MirrorError as e:
            return {"state": e.code, "error": e.message[:300], "pending": self.ledger.pending()}
        except FileChanged as e:
            return {"state": "retry", "rescan": [e.path], "pending": self.ledger.pending()}

    def _call(self, t: Any, method: str, rel: str, sleep: Callable[[float], None], **kw: Any) -> Response:
        """One request with backpressure: 429 / 503 wait for Retry-After."""
        for attempt in range(MAX_BUSY_RETRIES):
            status, body, headers = t.request(method, rel, **kw)
            if status in (429, 503):
                try:
                    wait = float(headers.get("retry-after") or 2)
                except ValueError:
                    wait = 2.0
                sleep(min(30.0, max(0.5, wait)) * (1 + attempt * 0.5))
                continue
            if status in (401, 403):
                raise MirrorError("auth", str(body.get("message") or body.get("error") or status))
            if status in (507, 413) and body.get("error") in ("quota", "too_large"):
                raise MirrorError("quota", str(body.get("message") or "quota"))
            if status >= 500:
                raise MirrorError("retry", f"server {status}")
            return status, body, headers
        raise MirrorError("retry", "the Vault API is busy")

    def _sync(self, t: Any, deadline: float, sleep: Callable[[float], None]) -> dict:
        self.plan()
        index_rev, head, floor = self._device_state()
        journal_seq = int(self.ledger.get("journal_seq", "-1") or -1)
        replica_id = self.ledger.get("replica_id")
        if replica_id and journal_seq >= 0 and not self.ledger.pending() and journal_seq >= head \
                and self.ledger.get("published_index_rev") == str(index_rev):
            return {"state": "ready", "pending": 0, "noop": True, "skipped": self.skipped}

        status, b, _ = self._call(t, "POST", "begin", sleep, json_body={"replica_id": replica_id or None})
        if status != 200:
            raise MirrorError("error", f"begin {status}")
        if b.get("reset") or b.get("replica_id") != replica_id or int(b.get("staged_seq") or 0) >= int(self.ledger.get("next_seq", "1") or 1):
            if not b.get("reset"):
                status, b, _ = self._call(t, "POST", "begin", sleep, json_body={"replica_id": None, "reset": True})
                if status != 200:
                    raise MirrorError("error", f"begin {status}")
            self.ledger.reset(str(b.get("replica_id") or ""))
            self.plan()
            journal_seq = -1
        if journal_seq < 0:  # a new copy: its journal starts with recent history only
            journal_seq = max(floor, head - FRESH_JOURNAL_BACKLOG, 0)
        journal_seq = max(journal_seq, floor)
        staged = int(b.get("staged_seq") or 0)
        chunk = max(64 * 1024, min(int(b.get("chunk_size") or DEFAULT_CHUNK), int(b.get("max_chunk") or DEFAULT_CHUNK)))

        while True:
            rows = self.ledger.con.execute(
                "SELECT seq, op, path, sha256, size, mtime_ms FROM outbox WHERE seq > ? ORDER BY seq LIMIT ?",
                (staged, OPS_BATCH)).fetchall()
            if not rows:
                break
            ops = [{"seq": r[0], "op": r[1], "path": r[2], **({"sha256": r[3], "size": r[4], "mtime_ms": r[5]}
                                                              if r[1] == "put" else {})} for r in rows]
            self._upload_missing(t, ops, chunk, sleep)
            for attempt in range(3):
                status, body, _ = self._call(t, "POST", "ops", sleep, json_body={"ops": ops})
                if status == 409 and body.get("error") == "missing_blobs":
                    self._upload_missing(t, ops, chunk, sleep, only=set(body.get("missing") or []))
                    continue
                break
            if status != 200:
                raise MirrorError("error", f"ops {status} {body.get('error') or ''}".strip())
            staged = int(body.get("staged_seq") or staged)
            if time.monotonic() > deadline:
                return {"state": "partial", "pending": self.ledger.pending(), "stagedSeq": staged}

        events = self._journal(journal_seq)
        status, p, _ = self._call(t, "POST", "publish", sleep, json_body={
            "upto_seq": staged, "device_index_rev": index_rev,
            "journal": [{k: v for k, v in e.items() if k != "seq"} for e in events],
        })
        if status == 409 and p.get("error") == "staged_seq":
            return {"state": "retry", "pending": self.ledger.pending()}
        if status != 200:
            raise MirrorError("error", f"publish {status}")
        self.ledger.ack(int(p.get("published_seq") or staged))
        if events:
            self.ledger.set("journal_seq", events[-1]["seq"])
        else:
            self.ledger.set("journal_seq", journal_seq)
        self.ledger.set("published_index_rev", index_rev)
        more = len(events) >= JOURNAL_BATCH or self.ledger.pending() > 0
        return {
            "state": "partial" if more else "ready",
            "pending": self.ledger.pending(),
            "publishedSeq": int(p.get("published_seq") or staged),
            "replicaRev": int(p.get("replica_rev") or 0),
            "notes": int(p.get("notes") or 0),
            "bytes": int(p.get("bytes") or 0),
            "skipped": self.skipped,
        }

    # ── contents ──
    def _upload_missing(self, t: Any, ops: list[dict], chunk: int, sleep: Callable[[float], None],
                        only: set[str] | None = None) -> None:
        puts: dict[str, dict] = {}
        for o in ops:
            if o["op"] == "put" and (only is None or o["sha256"] in only):
                puts.setdefault(o["sha256"], o)
        if not puts:
            return
        shas = list(puts)
        missing: list[str] = []
        partial: dict[str, int] = {}
        for i in range(0, len(shas), 1000):
            status, body, _ = self._call(t, "POST", "blobs/check", sleep, json_body={"sha256": shas[i:i + 1000]})
            if status != 200:
                raise MirrorError("error", f"check {status}")
            missing += [s for s in body.get("missing") or [] if s in puts]
            partial.update({k: int(v) for k, v in (body.get("partial") or {}).items() if k in puts})
        for sha in missing:
            self._upload(t, puts[sha], int(partial.get(sha, 0)), chunk, sleep)

    def _open(self, path: str):
        """Open through the path policy: no links, no excluded path, and the
        handle must be the file lstat saw (a swap between check and open fails)."""
        segs = path.split("/")
        if len(segs) == 2 and segs[0] == self.policy.config_dir and segs[1] in CONFIG_FILES:
            absolute = self.policy.config_file(segs[1])
            if not absolute:
                raise FileChanged(path)
        else:
            if not self._visible(path):
                raise FileChanged(path)
            try:
                _norm, absolute = self.policy.check(path, allow_missing=False)
            except Exception:
                raise FileChanged(path) from None
        try:
            st = os.lstat(absolute)
            if not _stat.S_ISREG(st.st_mode):
                raise FileChanged(path)
            fd = os.open(absolute, os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0))
        except OSError:
            raise FileChanged(path) from None
        fst = os.fstat(fd)
        if (fst.st_ino, fst.st_dev) != (st.st_ino, st.st_dev) and fst.st_ino and st.st_ino:
            os.close(fd)
            raise FileChanged(path)
        return os.fdopen(fd, "rb")

    def _upload(self, t: Any, op: dict, offset: int, chunk: int, sleep: Callable[[float], None], depth: int = 0) -> None:
        if depth > 3:
            raise MirrorError("retry", "upload did not converge")
        sha, size = op["sha256"], int(op["size"])
        with self._open(op["path"]) as f:
            h = hashlib.sha256()
            pos = 0
            while pos < offset:  # resume: the prefix is hashed locally, not resent
                block = f.read(min(1 << 20, offset - pos))
                if not block:
                    raise FileChanged(op["path"])
                h.update(block)
                pos += len(block)
            while True:
                data = f.read(min(chunk, size - pos)) if pos < size else b""
                if pos < size and not data:
                    raise FileChanged(op["path"])
                h.update(data)
                last = pos + len(data) >= size
                if last and (h.hexdigest() != sha or f.read(1)):
                    raise FileChanged(op["path"])  # never send content that is not the queued one
                for attempt in range(2):
                    status, body, _ = self._call(t, "PUT", f"blobs/{sha}", sleep, data=data,
                                                 params={"offset": pos, "total": size},
                                                 headers={"x-mel-chunk-sha256": hashlib.sha256(data).hexdigest()})
                    if status == 422 and body.get("error") == "chunk_checksum" and attempt == 0:
                        continue
                    break
                if status == 409 and body.get("error") == "offset":
                    return self._upload(t, op, int(body.get("offset") or 0), chunk, sleep, depth + 1)
                if status == 422 and body.get("error") == "checksum_mismatch":
                    raise FileChanged(op["path"])
                if status != 200:
                    raise MirrorError("error", f"upload {status} {body.get('error') or ''}".strip())
                if body.get("complete"):
                    return None
                pos += len(data)
                if last:
                    raise MirrorError("error", "upload incomplete")