"""Live Cursor CLI store.db + Desktop composer providers (extracted)."""

from __future__ import annotations

import json
import os
import sqlite3
import uuid
from dataclasses import replace
from datetime import datetime, timezone
from typing import Any, Mapping

from ..bounds import DEFAULT_BOUNDS, ReadBudget
from ..diagnostics import DiagnosticError
from ..model import Query, Session, SessionSummary, Turn
from ..paths import canonical_root, canonicalize_cwd, is_within, same_cwd
from ..sanitize import sanitize_turn_record
from ..snapshot import private_sqlite_connection, query_only_live_sqlite, stable_read_bytes

# Reuse shared helpers from the fixture/adapter module without circular import of ADAPTER.
from . import cursor as _cursor

LIVE_CLI_FORMAT = _cursor.LIVE_CLI_FORMAT
LIVE_DESKTOP_FORMAT = _cursor.LIVE_DESKTOP_FORMAT
_regular_directory = _cursor._regular_directory
_cwd_hash = _cursor._cwd_hash
_within = _cursor._within
_object = _cursor._object
_DuplicateKey = _cursor._DuplicateKey


def _bounded_names(directory: str, scanned: list[int]) -> list[str]:
    """Enumerate one directory without materializing an unbounded live store."""
    values: list[str] = []
    try:
        with os.scandir(directory) as entries:
            for entry in entries:
                scanned[0] += 1
                if scanned[0] > DEFAULT_BOUNDS.scanned_records:
                    # Do not return a lexical prefix: it cannot support an
                    # honest newest-session selection.
                    raise DiagnosticError.limit_exceeded()
                values.append(entry.name)
    except DiagnosticError:
        raise
    except OSError as error:
        raise DiagnosticError.source_busy(provider=LIVE_CLI_FORMAT) from error
    values.sort()
    return values


def _ms_to_rfc3339(value: object) -> str | None:
    if type(value) is not int:
        return None
    seconds = value / 1000 if value >= 1_577_836_800_000 else value
    try:
        return datetime.fromtimestamp(seconds, timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z")
    except (OverflowError, OSError, ValueError):
        return None


def _desktop_app_storage_dirs() -> list[str]:
    home = os.path.expanduser("~")
    return [
        os.path.join(home, "Library", "Application Support", "Cursor", "User", "globalStorage"),
        os.path.join(home, ".config", "Cursor", "User", "globalStorage"),
    ]


def _content_to_text(value: object) -> str | None:
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        chunks: list[str] = []
        for item in value:
            if isinstance(item, str):
                chunks.append(item)
            elif isinstance(item, Mapping) and isinstance(item.get("text"), str):
                chunks.append(item["text"])
        return "\n".join(chunks) if chunks else None
    return None


def _list_live_cli_stores(root: str, query: Query) -> list[SessionSummary]:
    chats = os.path.join(root, "chats")
    if not _regular_directory(chats, root):
        return []
    prefer = _cwd_hash(query.cwd) if query.cwd else None
    scanned = [0]
    # A cwd query maps to one deterministic bucket, so avoid enumerating every
    # unrelated project. Otherwise enumerate under one aggregate bound.
    hash_names = [prefer] if prefer is not None else _bounded_names(chats, scanned)
    values: list[SessionSummary] = []
    metadata_bytes = 0
    for name in hash_names:
        bucket = os.path.join(chats, name)
        if not _regular_directory(bucket, root):
            continue
        session_names = _bounded_names(bucket, scanned)
        for session_name in session_names:
            try:
                session_id = str(uuid.UUID(session_name))
            except ValueError:
                continue
            session_dir = os.path.join(bucket, session_name)
            store = os.path.join(session_dir, "store.db")
            if not os.path.isfile(store) or os.path.islink(store):
                continue
            if not is_within(store, root):
                continue
            meta_path = os.path.join(session_dir, "meta.json")
            cwd: str | None = None
            title: str | None = None
            created_at: str | None = None
            updated_at: str | None = None
            if os.path.lexists(meta_path):
                try:
                    read = stable_read_bytes(
                        meta_path,
                        root=root,
                        max_bytes=min(DEFAULT_BOUNDS.record_bytes, 256 * 1024),
                        budget=None,
                    )
                    metadata_bytes += len(read.data)
                    if metadata_bytes > DEFAULT_BOUNDS.source_read_bytes:
                        raise DiagnosticError.limit_exceeded()
                    meta = json.loads(read.data.decode("utf-8"))
                    if isinstance(meta, dict):
                        if isinstance(meta.get("cwd"), str):
                            cwd = canonicalize_cwd(meta["cwd"])
                        title = meta.get("title") if isinstance(meta.get("title"), str) else None
                        created_at = _ms_to_rfc3339(meta.get("createdAtMs"))
                        updated_at = _ms_to_rfc3339(meta.get("updatedAtMs"))
                except DiagnosticError as error:
                    if error.code == "E_LIMIT_EXCEEDED":
                        raise
                except (OSError, UnicodeDecodeError, json.JSONDecodeError):
                    pass
            if query.cwd is not None:
                if cwd is not None and not same_cwd(cwd, query.cwd):
                    continue
                if cwd is None and prefer is None:
                    continue
                if cwd is None:
                    cwd = canonicalize_cwd(query.cwd)
            if updated_at is None:
                try:
                    updated_at = datetime.fromtimestamp(
                        os.lstat(store).st_mtime, timezone.utc
                    ).isoformat(timespec="microseconds").replace("+00:00", "Z")
                except OSError:
                    continue
            if not _within(updated_at, query, session_id):
                continue
            values.append(
                SessionSummary(
                    source="cursor",
                    session_id=session_id,
                    source_path=store,
                    title=title,
                    cwd=cwd,
                    created_at=created_at,
                    updated_at=updated_at,
                    provider=LIVE_CLI_FORMAT,
                )
            )
    # Rank newest first so latest is not UUID/name-order capped.
    values.sort(
        key=lambda item: (
            item.updated_at is None,
            item.updated_at or "",
            item.session_id,
        ),
        reverse=True,
    )
    return values


def _show_live_cli_store(
    path: str, root: str, session_id: str, budget: ReadBudget, *, max_tool_chars: int
) -> Session:
    if not is_within(path, root) or not path.endswith("store.db"):
        raise DiagnosticError.unsafe_path()
    try:
        size = os.path.getsize(path)
    except OSError as error:
        raise DiagnosticError("E_SOURCE_BUSY", source="cursor", provider=LIVE_CLI_FORMAT) from error
    ctx = (
        query_only_live_sqlite(path, root=root, provider=LIVE_CLI_FORMAT)
        if size > DEFAULT_BOUNDS.sqlite_snapshot_bytes
        else private_sqlite_connection(path, root=root, provider=LIVE_CLI_FORMAT)
    )
    warnings: list[str] = []
    turns: list[Turn] = []
    turn_bounds = replace(DEFAULT_BOUNDS, tool_output_chars=max_tool_chars)
    cwd: str | None = None
    title: str | None = None
    created_at: str | None = None
    updated_at: str | None = None
    meta_path = os.path.join(os.path.dirname(path), "meta.json")
    if os.path.lexists(meta_path):
        try:
            read = stable_read_bytes(
                meta_path,
                root=root,
                max_bytes=min(DEFAULT_BOUNDS.record_bytes, 256 * 1024),
                budget=budget,
            )
            meta = json.loads(read.data.decode("utf-8"))
            if isinstance(meta, dict):
                if isinstance(meta.get("cwd"), str):
                    cwd = canonicalize_cwd(meta["cwd"])
                title = meta.get("title") if isinstance(meta.get("title"), str) else None
                created_at = _ms_to_rfc3339(meta.get("createdAtMs"))
                updated_at = _ms_to_rfc3339(meta.get("updatedAtMs"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, DiagnosticError):
            warnings.append("W_STALE_INDEX")
    row_cap = min(budget.limits.transcript_records, DEFAULT_BOUNDS.transcript_records)
    rows: list[Any] = []
    with ctx as connection:
        try:
            tables = {row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")}
            if "blobs" not in tables:
                raise DiagnosticError("E_UNSUPPORTED_FORMAT", source="cursor", provider=LIVE_CLI_FORMAT)
            # Deterministic order: rowid insertion order (stable), then id.
            # Fetch transcript_records + 1 so a longer store fails closed instead of
            # silently returning a prefix (issue #11).
            rows = connection.execute(
                "SELECT id, data FROM blobs ORDER BY rowid ASC, id ASC LIMIT ?",
                (row_cap + 1,),
            ).fetchall()
        except sqlite3.Error as error:
            raise DiagnosticError("E_CORRUPT_RECORD", source="cursor", provider=LIVE_CLI_FORMAT) from error
    if len(rows) > row_cap:
        raise DiagnosticError.limit_exceeded()
    for row in rows:
        if len(row) != 2 or not isinstance(row[0], str) or not isinstance(row[1], (bytes, memoryview)):
            continue
        data = bytes(row[1])
        max_blob = min(budget.limits.record_bytes, DEFAULT_BOUNDS.record_bytes)
        if len(data) > max_blob:
            raise DiagnosticError.limit_exceeded()
        # Charge each persisted blob row against the transcript window and source budget.
        budget.consume_transcript_records()
        budget.consume_bytes(len(data))
        try:
            payload = json.loads(data.decode("utf-8"), object_pairs_hook=_object)
        except (UnicodeDecodeError, json.JSONDecodeError, _DuplicateKey, RecursionError):
            warnings.append("W_BINARY_OMITTED")
            continue
        if not isinstance(payload, dict):
            continue
        role = payload.get("role")
        if role not in {"user", "assistant", "tool"}:
            if role == "system":
                continue
            continue
        text = _content_to_text(payload.get("content"))
        if text is None:
            continue
        turn, turn_warnings = sanitize_turn_record(
            {
                "role": role,
                "content": text,
                "tool_name": payload.get("name") if isinstance(payload.get("name"), str) else None,
            },
            ordinal=len(turns),
            bounds=turn_bounds,
        )
        warnings.extend(turn_warnings)
        if turn is not None:
            budget.consume_turns()
            turns.append(turn)
    if updated_at is None:
        try:
            updated_at = datetime.fromtimestamp(os.lstat(path).st_mtime, timezone.utc).isoformat(
                timespec="microseconds"
            ).replace("+00:00", "Z")
        except OSError:
            updated_at = None
    return Session(
        source="cursor",
        session_id=session_id,
        source_path=path,
        title=title,
        cwd=cwd,
        created_at=created_at,
        updated_at=updated_at,
        last_user_request=next((turn.content for turn in reversed(turns) if turn.role == "user"), None),
        last_assistant_action=next((turn.content for turn in reversed(turns) if turn.role == "assistant"), None),
        turns=tuple(turns),
        warnings=tuple(dict.fromkeys(warnings)),
    )


def _list_live_desktop(query: Query) -> list[SessionSummary]:
    values: list[SessionSummary] = []
    for storage in _desktop_app_storage_dirs():
        database = os.path.join(storage, "state.vscdb")
        if not os.path.isfile(database) or os.path.islink(database):
            continue
        try:
            root = canonical_root(storage)
        except DiagnosticError:
            continue
        if not is_within(database, root):
            continue
        try:
            size = os.path.getsize(database)
        except OSError:
            continue
        try:
            ctx = (
                query_only_live_sqlite(database, root=root, provider=LIVE_DESKTOP_FORMAT)
                if size > DEFAULT_BOUNDS.sqlite_snapshot_bytes
                else private_sqlite_connection(database, root=root, provider=LIVE_DESKTOP_FORMAT)
            )
            with ctx as connection:
                tables = {
                    row[0]
                    for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")
                }
                if "composerHeaders" not in tables:
                    continue
                rows = connection.execute(
                    "SELECT composerId, workspaceId, createdAt, lastUpdatedAt, isArchived, isSubagent, value "
                    "FROM composerHeaders WHERE IFNULL(isSubagent,0)=0 AND IFNULL(isArchived,0)=0 "
                    "ORDER BY lastUpdatedAt DESC LIMIT ?",
                    (DEFAULT_BOUNDS.listed_sessions * 4,),
                ).fetchall()
        except DiagnosticError:
            continue
        except sqlite3.Error:
            continue
        for row in rows:
            if len(values) >= DEFAULT_BOUNDS.listed_sessions:
                break
            if len(row) != 7 or not isinstance(row[0], str):
                continue
            composer_id = row[0]
            try:
                uuid.UUID(composer_id)
            except ValueError:
                if composer_id in {"empty-state-draft"}:
                    continue
            value_raw = row[6]
            cwd: str | None = None
            title: str | None = None
            if isinstance(value_raw, str):
                try:
                    payload = json.loads(value_raw)
                except json.JSONDecodeError:
                    payload = None
                if isinstance(payload, dict):
                    title = payload.get("name") if isinstance(payload.get("name"), str) else None
                    if title is None and isinstance(payload.get("subtitle"), str):
                        title = payload["subtitle"]
                    wi = payload.get("workspaceIdentifier")
                    if isinstance(wi, dict):
                        uri = wi.get("uri")
                        if isinstance(uri, dict) and isinstance(uri.get("fsPath"), str):
                            try:
                                cwd = canonicalize_cwd(uri["fsPath"])
                            except DiagnosticError:
                                cwd = None
                        elif isinstance(uri, dict) and isinstance(uri.get("path"), str):
                            try:
                                cwd = canonicalize_cwd(uri["path"])
                            except DiagnosticError:
                                cwd = None
            if query.cwd is not None:
                if cwd is None or not same_cwd(cwd, query.cwd):
                    continue
            updated_at = _ms_to_rfc3339(row[3])
            if updated_at is None or not _within(updated_at, query, composer_id):
                continue
            values.append(
                SessionSummary(
                    source="cursor",
                    session_id=composer_id,
                    source_path=database,
                    title=title,
                    cwd=cwd,
                    created_at=_ms_to_rfc3339(row[2]),
                    updated_at=updated_at,
                    provider=LIVE_DESKTOP_FORMAT,
                )
            )
    return values


def _show_live_desktop(
    path: str, session_id: str, budget: ReadBudget, *, max_tool_chars: int
) -> Session:
    """List-level metadata + optional composerData text; bubble graph not fully restored."""
    storage = os.path.dirname(path)
    root = canonical_root(storage)
    if not is_within(path, root):
        raise DiagnosticError.unsafe_path()
    try:
        size = os.path.getsize(path)
    except OSError as error:
        raise DiagnosticError("E_SOURCE_BUSY", source="cursor", provider=LIVE_DESKTOP_FORMAT) from error
    ctx = (
        query_only_live_sqlite(path, root=root, provider=LIVE_DESKTOP_FORMAT)
        if size > DEFAULT_BOUNDS.sqlite_snapshot_bytes
        else private_sqlite_connection(path, root=root, provider=LIVE_DESKTOP_FORMAT)
    )
    warnings: list[str] = ["W_MISSING_BLOB"]  # full bubble chain not claimed
    with ctx as connection:
        try:
            row = connection.execute(
                "SELECT composerId, createdAt, lastUpdatedAt, value FROM composerHeaders WHERE composerId=?",
                (session_id,),
            ).fetchone()
            text_blob = None
            if "cursorDiskKV" in {
                r[0] for r in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")
            }:
                # Two-phase length gate: measure before loading full value (issue #11).
                len_row = connection.execute(
                    "SELECT length(CAST(value AS BLOB)) FROM cursorDiskKV WHERE key=?",
                    (f"composerData:{session_id}",),
                ).fetchone()
                if len_row is not None and isinstance(len_row[0], int):
                    blob_len = len_row[0]
                    remaining = min(
                        max(0, budget.limits.source_read_bytes - budget.bytes_read),
                        max(0, DEFAULT_BOUNDS.source_read_bytes - budget.bytes_read),
                        DEFAULT_BOUNDS.record_bytes,
                    )
                    if blob_len < 0 or blob_len > remaining or blob_len > DEFAULT_BOUNDS.record_bytes:
                        raise DiagnosticError.limit_exceeded()
                    text_blob = connection.execute(
                        "SELECT value FROM cursorDiskKV WHERE key=?",
                        (f"composerData:{session_id}",),
                    ).fetchone()
                    # Stash measured length for charge after fetch.
                    text_blob = (text_blob[0], blob_len) if text_blob is not None else None
        except sqlite3.Error as error:
            raise DiagnosticError("E_CORRUPT_RECORD", source="cursor", provider=LIVE_DESKTOP_FORMAT) from error
    if row is None:
        raise DiagnosticError("E_NO_MATCH", source="cursor", provider=LIVE_DESKTOP_FORMAT)
    cwd = None
    title = None
    if isinstance(row[3], str):
        try:
            payload = json.loads(row[3])
            if isinstance(payload, dict):
                title = payload.get("name") if isinstance(payload.get("name"), str) else None
                wi = payload.get("workspaceIdentifier")
                if isinstance(wi, dict) and isinstance(wi.get("uri"), dict):
                    fs = wi["uri"].get("fsPath") or wi["uri"].get("path")
                    if isinstance(fs, str):
                        cwd = canonicalize_cwd(fs)
        except (json.JSONDecodeError, DiagnosticError):
            pass
    turns: list[Turn] = []
    turn_bounds = replace(DEFAULT_BOUNDS, tool_output_chars=max_tool_chars)
    structured = False
    if text_blob and len(text_blob) >= 1 and isinstance(text_blob[0], (str, bytes)):
        raw_bytes = (
            text_blob[0]
            if isinstance(text_blob[0], (bytes, bytearray, memoryview))
            else text_blob[0].encode("utf-8")
        )
        actual_len = len(raw_bytes)
        expected_len = text_blob[1] if len(text_blob) >= 2 and isinstance(text_blob[1], int) else actual_len
        # Re-check after fetch: concurrent updates between length probe and value load
        # must not admit an oversized body under a stale length charge.
        remaining = min(
            max(0, budget.limits.source_read_bytes - budget.bytes_read),
            max(0, DEFAULT_BOUNDS.source_read_bytes - budget.bytes_read),
            DEFAULT_BOUNDS.record_bytes,
        )
        if actual_len > remaining or actual_len > DEFAULT_BOUNDS.record_bytes:
            raise DiagnosticError.limit_exceeded()
        if expected_len != actual_len and actual_len > min(expected_len, remaining):
            raise DiagnosticError.limit_exceeded()
        budget.consume_bytes(actual_len)
        raw = raw_bytes.decode("utf-8") if not isinstance(text_blob[0], str) else text_blob[0]
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            data = None
        if isinstance(data, dict):
            extracted = _composer_data_turns(data)
            if extracted:
                structured = True
                for raw_turn in extracted:
                    if len(turns) >= DEFAULT_BOUNDS.normalized_turns:
                        warnings.append("W_TRUNCATED")
                        break
                    turn, tw = sanitize_turn_record(raw_turn, ordinal=len(turns), bounds=turn_bounds)
                    warnings.extend(tw)
                    if turn is not None:
                        budget.consume_turns()
                        turns.append(turn)
            elif isinstance(data.get("text"), str) and data["text"].strip():
                turn, tw = sanitize_turn_record(
                    {"role": "user", "content": data["text"]},
                    ordinal=0,
                    bounds=turn_bounds,
                )
                warnings.extend(tw)
                if turn is not None:
                    budget.consume_turns()
                    turns.append(turn)
    # Full bubble graph / parent links still not claimed even when multi-turn text is recovered.
    if not structured:
        warnings = ["W_MISSING_BLOB", *warnings]
    else:
        warnings = ["W_MISSING_BLOB", *warnings]  # graph not claimed; multi-turn text is best-effort
    return Session(
        source="cursor",
        session_id=session_id,
        source_path=path,
        title=title,
        cwd=cwd,
        created_at=_ms_to_rfc3339(row[1]),
        updated_at=_ms_to_rfc3339(row[2]),
        last_user_request=next((turn.content for turn in reversed(turns) if turn.role == "user"), None),
        last_assistant_action=next(
            (turn.content for turn in reversed(turns) if turn.role == "assistant"), None
        ),
        turns=tuple(turns),
        warnings=tuple(dict.fromkeys(warnings)),
    )


def _composer_data_turns(data: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Best-effort multi-turn extraction from composerData JSON (no full bubble graph)."""

    def _role(value: object) -> str | None:
        if not isinstance(value, str):
            return None
        lowered = value.casefold()
        if lowered in {"user", "human", "humanity"}:
            return "user"
        if lowered in {"assistant", "ai", "bot", "model"}:
            return "assistant"
        if lowered in {"tool", "function"}:
            return "tool"
        return None

    def _text(value: object) -> str | None:
        return _content_to_text(value)

    candidates: list[Any] = []
    for key in (
        "conversation",
        "messages",
        "bubbles",
        "fullConversationHeadersMaybe",
        "conversationMap",
        "tabs",
    ):
        if key in data:
            candidates.append(data[key])
    # Nested common shapes
    for key in ("composerState", "state", "data"):
        nested = data.get(key)
        if isinstance(nested, Mapping):
            for sub in ("conversation", "messages", "bubbles"):
                if sub in nested:
                    candidates.append(nested[sub])

    values: list[dict[str, Any]] = []
    seen_content: set[str] = set()

    def _append(role: str, content: str) -> None:
        content = content.strip()
        if not content:
            return
        fingerprint = f"{role}:{content[:200]}"
        if fingerprint in seen_content:
            return
        seen_content.add(fingerprint)
        values.append({"role": role, "content": content})

    def _walk(node: object, depth: int = 0) -> None:
        if depth > 8 or len(values) >= DEFAULT_BOUNDS.normalized_turns:
            return
        if isinstance(node, list):
            for item in node:
                _walk(item, depth + 1)
            return
        if not isinstance(node, Mapping):
            return
        role = _role(node.get("role") or node.get("type") or node.get("bubbleType") or node.get("sender"))
        text = _text(node.get("text") or node.get("content") or node.get("richText") or node.get("rawText"))
        if role and text:
            _append(role, text)
        # Map of id → bubble
        if all(isinstance(v, Mapping) for v in node.values()) and len(node) <= 500:
            for v in node.values():
                _walk(v, depth + 1)
            return
        for key in ("messages", "bubbles", "children", "items", "conversation"):
            if key in node:
                _walk(node[key], depth + 1)

    for candidate in candidates:
        _walk(candidate)
    return values