"""Peer-originated "learning" proposals: persistent, reviewable inbox.

Phase B of the drives-research-and-mesh-learning plan. Proposals arrive as
``peer_proposal`` mesh messages, get validated/rate-limited/idempotency-checked,
and are appended to ``~/.mugi/mesh/proposals.jsonl`` (mode ``0o600``) for the
user to promote (→ drive) or dismiss. Promotion happens on a main-agent-capable
code path, never inside ``heartbeat_drive_action``.

Security stance: **peer-originated, review before act**. Transport encryption
does not imply content trust. Nothing in this module calls ``create_drive`` or
touches workspace files on its own — callers must apply normal tool policy.
"""
from __future__ import annotations

import hashlib
import json
import logging
import os
import threading
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Iterable

try:
    import fcntl
    _HAS_FLOCK = True
except ImportError:
    _HAS_FLOCK = False

from utils.redaction import cap_str

from mesh.peers import MESH_DIR

_log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Schema constants and field caps
# ---------------------------------------------------------------------------

SCHEMA_VERSION = 1
SUPPORTED_SCHEMA_VERSIONS = frozenset({1})

PROPOSAL_KINDS = frozenset({
    "skill",
    "plugin",
    "research",
    "knowledge",
    "drive",
    "note",
})

STATE_PENDING = "pending"
STATE_ACCEPTED = "accepted"
STATE_DISMISSED = "dismissed"
STATE_SUPERSEDED = "superseded"
STATE_FAILED = "failed"

_VALID_STATES = frozenset({
    STATE_PENDING, STATE_ACCEPTED, STATE_DISMISSED, STATE_SUPERSEDED, STATE_FAILED,
})

MAX_TITLE_LEN = 200
MAX_BODY_LEN = 4000
MAX_SUGGESTED_ACTION_LEN = 1000
MAX_URL_LEN = 2000
MAX_PROPOSAL_ID_LEN = 128
MAX_PROPOSAL_BYTES = 12 * 1024  # full row
MAX_PEER_NAME_LEN = 120
MAX_ENTRIES = 5_000

# Rate limiting default (overridden by mesh config).
DEFAULT_RATE_PER_PEER_PER_HOUR = 20


# Reason codes used by validation / ingress; kept lowercase for telemetry.
REASON_SCHEMA = "schema"
REASON_OVERSIZE = "oversize"
REASON_RATE_LIMITED = "rate_limited"
REASON_UNTRUSTED_PEER = "untrusted_peer"
REASON_DISABLED = "disabled"
REASON_IDEMPOTENT_HIT = "idempotent_hit"
REASON_UNSUPPORTED_SCHEMA = "unsupported_schema"


# ---------------------------------------------------------------------------
# Paths and locking
# ---------------------------------------------------------------------------

PROPOSALS_PATH = os.path.join(MESH_DIR, "proposals.jsonl")
_LOCK_PATH = PROPOSALS_PATH + ".lock"

_proc_lock = threading.Lock()


def _ensure_mesh_dir() -> None:
    try:
        os.makedirs(MESH_DIR, exist_ok=True, mode=0o700)
    except OSError:
        pass


def _utcnow_iso() -> str:
    now = datetime.now(timezone.utc)
    return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"


class _FileLock:
    """Context manager combining the process-wide threading lock with fcntl flock."""

    def __init__(self, path: str):
        self.path = path
        self._fd: int | None = None
        self._proc_acquired = False

    def __enter__(self) -> "_FileLock":
        _proc_lock.acquire()
        self._proc_acquired = True
        try:
            if _HAS_FLOCK:
                _ensure_mesh_dir()
                fd = os.open(self.path, os.O_CREAT | os.O_RDWR, 0o600)
                try:
                    fcntl.flock(fd, fcntl.LOCK_EX)
                except OSError:
                    os.close(fd)
                    raise
                self._fd = fd
            return self
        except Exception:
            if self._proc_acquired:
                _proc_lock.release()
                self._proc_acquired = False
            raise

    def __exit__(self, *_exc: Any) -> None:
        try:
            if self._fd is not None and _HAS_FLOCK:
                fcntl.flock(self._fd, fcntl.LOCK_UN)
                os.close(self._fd)
        finally:
            self._fd = None
            if self._proc_acquired:
                _proc_lock.release()
                self._proc_acquired = False


# ---------------------------------------------------------------------------
# Payload validation (fail closed)
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class ValidationResult:
    ok: bool
    reason: str = ""
    detail: str = ""
    normalized: dict | None = None


def _is_str_nonempty(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def validate_payload(raw: Any) -> ValidationResult:
    """Validate a peer_proposal payload; returns a normalized dict on success.

    ``raw`` must be a dict with:
      - kind: str in PROPOSAL_KINDS
      - title: non-empty str (capped)
      - body: non-empty str (capped)
      - suggested_action: optional str (capped)
      - url: optional str (capped)
      - schema_version: optional int (defaults to current)
      - proposal_id: optional stable idempotency key (capped)
    Extra keys are dropped. Non-JSON-serializable fields are rejected.
    """
    if not isinstance(raw, dict):
        return ValidationResult(False, REASON_SCHEMA, "payload must be an object")

    schema_version = raw.get("schema_version", SCHEMA_VERSION)
    try:
        schema_version = int(schema_version)
    except (TypeError, ValueError):
        return ValidationResult(False, REASON_SCHEMA, "schema_version must be an integer")
    if schema_version > SCHEMA_VERSION:
        return ValidationResult(False, REASON_UNSUPPORTED_SCHEMA, f"schema_version {schema_version} is newer than supported")
    if schema_version not in SUPPORTED_SCHEMA_VERSIONS:
        return ValidationResult(False, REASON_UNSUPPORTED_SCHEMA, f"schema_version {schema_version} is not supported")

    kind = str(raw.get("kind") or "").strip().lower()
    if kind not in PROPOSAL_KINDS:
        return ValidationResult(False, REASON_SCHEMA, f"kind {kind!r} not in {sorted(PROPOSAL_KINDS)}")

    title = raw.get("title")
    if not _is_str_nonempty(title):
        return ValidationResult(False, REASON_SCHEMA, "title is required")
    body = raw.get("body")
    if not _is_str_nonempty(body):
        return ValidationResult(False, REASON_SCHEMA, "body is required")

    suggested_action = raw.get("suggested_action") or ""
    if not isinstance(suggested_action, str):
        return ValidationResult(False, REASON_SCHEMA, "suggested_action must be a string")
    url = raw.get("url") or ""
    if not isinstance(url, str):
        return ValidationResult(False, REASON_SCHEMA, "url must be a string")

    proposal_id_hint = raw.get("proposal_id") or ""
    if not isinstance(proposal_id_hint, str):
        return ValidationResult(False, REASON_SCHEMA, "proposal_id must be a string")

    normalized = {
        "schema_version": schema_version,
        "kind": kind,
        "title": cap_str(title, MAX_TITLE_LEN),
        "body": cap_str(body, MAX_BODY_LEN),
        "suggested_action": cap_str(suggested_action, MAX_SUGGESTED_ACTION_LEN),
        "url": cap_str(url, MAX_URL_LEN),
        "proposal_id_hint": cap_str(proposal_id_hint, MAX_PROPOSAL_ID_LEN),
    }

    try:
        encoded = json.dumps(normalized, ensure_ascii=True)
    except (TypeError, ValueError) as exc:
        return ValidationResult(False, REASON_SCHEMA, f"payload not JSON-serializable: {exc}")
    if len(encoded.encode("utf-8")) > MAX_PROPOSAL_BYTES:
        return ValidationResult(False, REASON_OVERSIZE, "payload exceeds max proposal size")

    return ValidationResult(True, normalized=normalized)


def compute_proposal_id(*, peer_fingerprint: str, payload: dict) -> str:
    """Compute a stable proposal id. Uses the peer hint when present.

    When ``proposal_id`` is provided (as ``proposal_id_hint`` after validation),
    the id is a 32-hex digest of ``(peer_fingerprint, hint)`` so the same
    string from two different peers is never treated as the same proposal.

    Without a hint, falls back to a content hash keyed on (peer_fingerprint,
    kind, title, body, suggested_action, url), so transport retries stay
    idempotent when the peer omits ``proposal_id``.
    """
    hint = (payload.get("proposal_id_hint") or "").strip() if isinstance(payload, dict) else ""
    if hint:
        h = hashlib.sha256()
        h.update((peer_fingerprint or "").encode("utf-8"))
        h.update(b"\x1f")
        h.update(hint.encode("utf-8"))
        return h.hexdigest()[:32]
    h = hashlib.sha256()
    h.update((peer_fingerprint or "").encode("utf-8"))
    for key in ("kind", "title", "body", "suggested_action", "url"):
        h.update(b"\x1f")
        h.update(str((payload or {}).get(key, "")).encode("utf-8"))
    return h.hexdigest()[:32]


# ---------------------------------------------------------------------------
# Read / write helpers
# ---------------------------------------------------------------------------

def _iter_rows(path: str) -> Iterable[dict]:
    if not os.path.isfile(path):
        return []
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
    except OSError:
        return []
    out: list[dict] = []
    for raw in lines:
        line = raw.strip()
        if not line:
            continue
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(row, dict):
            out.append(row)
    return out


def _write_all_rows_locked(rows: list[dict]) -> None:
    """Atomic rewrite of the proposals file. Caller must hold the file lock."""
    _ensure_mesh_dir()
    tmp = PROPOSALS_PATH + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=True) + "\n")
    os.replace(tmp, PROPOSALS_PATH)
    try:
        os.chmod(PROPOSALS_PATH, 0o600)
    except OSError:
        pass


def _append_row_locked(row: dict) -> None:
    """Append a single row. Caller must hold the file lock."""
    _ensure_mesh_dir()
    encoded = json.dumps(row, ensure_ascii=True)
    with open(PROPOSALS_PATH, "a", encoding="utf-8") as f:
        f.write(encoded + "\n")
    try:
        os.chmod(PROPOSALS_PATH, 0o600)
    except OSError:
        pass


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class AppendResult:
    ok: bool
    proposal_id: str = ""
    reason: str = ""
    detail: str = ""
    row: dict | None = None


def _count_recent_from_peer(rows: list[dict], peer_fingerprint: str, window_seconds: int) -> int:
    cutoff = time.time() - window_seconds
    n = 0
    for row in rows:
        if str(row.get("peer_fingerprint") or "") != peer_fingerprint:
            continue
        try:
            row_ts = float(row.get("ts_unix") or 0.0)
        except (TypeError, ValueError):
            continue
        if row_ts >= cutoff:
            n += 1
    return n


def _find_existing(rows: list[dict], proposal_id: str) -> int:
    for idx, row in enumerate(rows):
        if str(row.get("proposal_id") or "") == proposal_id:
            return idx
    return -1


def append_proposal(
    *,
    peer_fingerprint: str,
    peer_name: str,
    payload: dict,
    conversation_id: str = "",
    rate_per_hour: int = DEFAULT_RATE_PER_PEER_PER_HOUR,
) -> AppendResult:
    """Validate + persist one proposal. Caller must check peer trust / feature flag.

    Idempotent on ``proposal_id``: a second call for the same id is a no-op and
    returns the existing row with ``reason=idempotent_hit``.
    """
    if not _is_str_nonempty(peer_fingerprint):
        return AppendResult(False, reason=REASON_SCHEMA, detail="peer_fingerprint required")

    validation = validate_payload(payload)
    if not validation.ok:
        return AppendResult(False, reason=validation.reason, detail=validation.detail)
    normalized = validation.normalized or {}

    proposal_id = compute_proposal_id(
        peer_fingerprint=peer_fingerprint,
        payload=normalized,
    )
    if len(proposal_id) > MAX_PROPOSAL_ID_LEN:
        proposal_id = proposal_id[:MAX_PROPOSAL_ID_LEN]

    with _FileLock(_LOCK_PATH):
        rows = list(_iter_rows(PROPOSALS_PATH))
        existing_idx = _find_existing(rows, proposal_id)
        if existing_idx >= 0:
            return AppendResult(
                True,
                proposal_id=proposal_id,
                reason=REASON_IDEMPOTENT_HIT,
                row=rows[existing_idx],
            )
        if rate_per_hour > 0:
            recent = _count_recent_from_peer(rows, peer_fingerprint, window_seconds=3600)
            if recent >= rate_per_hour:
                return AppendResult(
                    False,
                    proposal_id=proposal_id,
                    reason=REASON_RATE_LIMITED,
                    detail=f"peer sent {recent} proposals in the last hour",
                )

        row = {
            "schema_version": SCHEMA_VERSION,
            "proposal_id": proposal_id,
            "peer_fingerprint": peer_fingerprint,
            "peer_name": cap_str(peer_name or "", MAX_PEER_NAME_LEN),
            "conversation_id": cap_str(conversation_id or "", 128),
            "kind": normalized["kind"],
            "payload": {
                "title": normalized["title"],
                "body": normalized["body"],
                "suggested_action": normalized["suggested_action"],
                "url": normalized["url"],
                "schema_version": normalized["schema_version"],
            },
            "state": STATE_PENDING,
            "ts": _utcnow_iso(),
            "ts_unix": time.time(),
            "drive_id": None,
            "last_error": None,
            "row_id": uuid.uuid4().hex,
        }
        encoded = json.dumps(row, ensure_ascii=True)
        if len(encoded.encode("utf-8")) > MAX_PROPOSAL_BYTES:
            return AppendResult(
                False,
                proposal_id=proposal_id,
                reason=REASON_OVERSIZE,
                detail="row exceeds max proposal size after normalization",
            )

        if len(rows) >= MAX_ENTRIES:
            _write_all_rows_locked(rows[-(MAX_ENTRIES - 1):])

        _append_row_locked(row)

    return AppendResult(True, proposal_id=proposal_id, row=row)


def list_proposals(
    *,
    state: str | None = None,
    peer_fingerprint: str | None = None,
    limit: int = 200,
) -> list[dict]:
    limit = max(1, min(int(limit or 200), MAX_ENTRIES))
    with _FileLock(_LOCK_PATH):
        rows = list(_iter_rows(PROPOSALS_PATH))
    if state:
        rows = [r for r in rows if str(r.get("state") or "") == state]
    if peer_fingerprint:
        rows = [r for r in rows if str(r.get("peer_fingerprint") or "") == peer_fingerprint]
    rows.sort(key=lambda r: float(r.get("ts_unix") or 0.0), reverse=True)
    return rows[:limit]


def get_proposal(proposal_id: str) -> dict | None:
    if not _is_str_nonempty(proposal_id):
        return None
    with _FileLock(_LOCK_PATH):
        rows = list(_iter_rows(PROPOSALS_PATH))
    idx = _find_existing(rows, proposal_id)
    return rows[idx] if idx >= 0 else None


def set_proposal_state(
    proposal_id: str,
    new_state: str,
    *,
    drive_id: str | None = None,
    last_error: str | None = None,
) -> bool:
    if new_state not in _VALID_STATES:
        return False
    if not _is_str_nonempty(proposal_id):
        return False
    with _FileLock(_LOCK_PATH):
        rows = list(_iter_rows(PROPOSALS_PATH))
        idx = _find_existing(rows, proposal_id)
        if idx < 0:
            return False
        row = dict(rows[idx])
        row["state"] = new_state
        if drive_id is not None:
            row["drive_id"] = cap_str(drive_id, 128) or None
        if last_error is not None:
            row["last_error"] = cap_str(last_error, 400) or None
        row["state_ts"] = _utcnow_iso()
        rows[idx] = row
        _write_all_rows_locked(rows)
    return True


def clear_all() -> int:
    """Remove every proposal row. Returns the number removed."""
    with _FileLock(_LOCK_PATH):
        rows = list(_iter_rows(PROPOSALS_PATH))
        count = len(rows)
        _write_all_rows_locked([])
    return count


def export_all() -> list[dict]:
    """Return a deep-copy snapshot of all rows (for privacy triage / export)."""
    with _FileLock(_LOCK_PATH):
        rows = list(_iter_rows(PROPOSALS_PATH))
    return [dict(r) for r in rows]
