"""Knowledge corpus automation (Phase 6).

Optional filesystem watches on corpus ``source_path`` roots and bounded
heartbeat staleness sweeps. Respects digest locks and concurrency caps.
"""

from __future__ import annotations

import logging
import os
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Any

_log = logging.getLogger(__name__)

_URL_RE = re.compile(r"^https?://", re.IGNORECASE)

try:
    from watchdog.events import FileSystemEventHandler  # type: ignore
    from watchdog.observers import Observer  # type: ignore

    _HAS_WATCHDOG = True
except Exception:  # pragma: no cover - polling fallback
    FileSystemEventHandler = object  # type: ignore
    Observer = None  # type: ignore
    _HAS_WATCHDOG = False

_POLL_INTERVAL_SECONDS = 4.0
_TARGET_SYNC_SECONDS = 60.0
_HEARTBEAT_CONFIG_POLL_MINUTES = 5.0

_notify_state_lock = threading.Lock()
_last_notified_stale_count: dict[str, int] = {}
_sweep_offset_lock = threading.Lock()
_sweep_offset = 0


def get_effective_config() -> dict:
    from core.config import get_knowledge_config, reload_config_if_changed

    reload_config_if_changed()
    return get_knowledge_config() or {}


def is_automation_enabled() -> bool:
    return bool(get_effective_config().get("automation_enabled", False))


def is_watch_enabled() -> bool:
    cfg = get_effective_config()
    return bool(cfg.get("automation_enabled", False)) and bool(cfg.get("watch_folders_enabled", False))


def is_heartbeat_enabled() -> bool:
    cfg = get_effective_config()
    return bool(cfg.get("automation_enabled", False)) and bool(
        cfg.get("heartbeat_stale_refresh_enabled", False)
    )


def knowledge_heartbeat_interval_minutes() -> float:
    """Dynamic scheduler interval: poll config when disabled, configured cadence when enabled."""
    if not is_heartbeat_enabled():
        return _HEARTBEAT_CONFIG_POLL_MINUTES
    cfg = get_effective_config()
    return max(30.0, float(cfg.get("heartbeat_interval_minutes", 360)))


def _local_watch_root(source_path: str) -> str | None:
    sp = (source_path or "").strip()
    if not sp or _URL_RE.match(sp):
        return None
    expanded = os.path.realpath(os.path.expanduser(sp))
    if os.path.isdir(expanded):
        return expanded
    parent = os.path.dirname(expanded)
    if parent and os.path.isdir(parent):
        return parent
    return None


def iter_watch_path_map() -> dict[str, list[str]]:
    """Map local watch roots to every corpus slug that uses that root."""
    from tools.knowledge_tools import list_knowledge_corpora

    path_to_slugs: dict[str, list[str]] = {}
    for row in list_knowledge_corpora():
        slug = str(row.get("slug") or "").strip()
        root = _local_watch_root(str(row.get("source_path") or ""))
        if not slug or not root:
            continue
        path_to_slugs.setdefault(root, []).append(slug)
    for slugs in path_to_slugs.values():
        slugs.sort()
    return path_to_slugs


def consolidate_watch_path_map(path_map: dict[str, list[str]]) -> dict[str, list[str]]:
    """Merge nested watch roots so one outer watch covers inner corpus folders."""
    if not path_map:
        return {}
    consolidated: dict[str, set[str]] = {}
    for path in sorted(path_map.keys(), key=len):
        slugs = set(path_map[path])
        parent: str | None = None
        for root in consolidated:
            if path == root or path.startswith(root + os.sep):
                parent = root
                break
        if parent is not None:
            consolidated[parent].update(slugs)
            continue
        drop = [root for root in consolidated if root.startswith(path + os.sep)]
        for root in drop:
            slugs.update(consolidated.pop(root))
        consolidated[path] = slugs
    return {path: sorted(slugs) for path, slugs in sorted(consolidated.items())}


def iter_watch_targets() -> list[tuple[str, str]]:
    """Return ``(slug, abs_path)`` pairs for local folder corpora."""
    out: list[tuple[str, str]] = []
    for path, slugs in consolidate_watch_path_map(iter_watch_path_map()).items():
        for slug in slugs:
            out.append((slug, path))
    return out


def _clear_notify_state(slug: str) -> None:
    with _notify_state_lock:
        _last_notified_stale_count.pop(slug, None)


def _notify_stale_corpus(slug: str, stale_count: int) -> bool:
    with _notify_state_lock:
        if _last_notified_stale_count.get(slug) == stale_count:
            return False

    delivered = False
    try:
        from utils.notifications import send_mac_notification

        send_mac_notification(
            title="Knowledge corpus stale",
            subtitle=slug,
            body=f"{stale_count} changed source(s). Open Knowledge Library to refresh.",
        )
        delivered = True
    except Exception:
        _log.debug("knowledge_automation notification failed", exc_info=True)
    try:
        from persistence import PROACTIVE_QUEUE_PATH
        from utils.proactive_queue import append_observation_queue

        proposed = f"refresh stale sources in the {slug} corpus ({stale_count} changed)"
        if append_observation_queue(
            os.path.expanduser(PROACTIVE_QUEUE_PATH),
            noticed=f"Knowledge corpus '{slug}' has {stale_count} stale source(s)",
            proposed_work=proposed,
            source="knowledge_automation",
            suggested_tier="in_thread",
            dedupe_key=f"knowledge_stale:{slug}:{stale_count}",
        ):
            delivered = True
    except Exception:
        _log.debug("knowledge_automation observation queue failed", exc_info=True)

    if delivered:
        with _notify_state_lock:
            _last_notified_stale_count[slug] = stale_count
        return True
    return False


def _refresh_started(result: dict[str, Any]) -> bool:
    if not result.get("ok"):
        return False
    status = str(result.get("status") or "").strip().lower()
    if status == "idle":
        return False
    if result.get("message") and "no stale sources" in str(result.get("message")).lower():
        return False
    return True


def _eligible_sweep_slugs() -> list[str]:
    from tools.knowledge_tools import get_knowledge_status, list_knowledge_corpora

    slugs: list[str] = []
    for row in list_knowledge_corpora():
        slug = str(row.get("slug") or "").strip()
        if not slug:
            continue
        digest_status = str(row.get("digest_status") or "").lower()
        if digest_status == "running":
            continue
        job_status = get_knowledge_status(slug)
        if job_status.get("running"):
            continue
        slugs.append(slug)
    return slugs


def _select_sweep_candidates(slugs: list[str], max_corpora: int) -> list[str]:
    global _sweep_offset
    if not slugs:
        return []
    if len(slugs) <= max_corpora:
        return slugs
    with _sweep_offset_lock:
        start = _sweep_offset % len(slugs)
        _sweep_offset = (start + max_corpora) % len(slugs)
    return [slugs[(start + i) % len(slugs)] for i in range(max_corpora)]


def _process_stale_slug(slug: str, *, auto_refresh: bool) -> dict[str, Any]:
    from tools.knowledge_tools import check_corpus_staleness, get_knowledge_status, refresh_digest_job

    status = get_knowledge_status(slug)
    if status.get("running"):
        return {"slug": slug, "skipped": "running"}
    stale = check_corpus_staleness(slug)
    count = int(stale.get("stale_count") or 0)
    if count <= 0:
        _clear_notify_state(slug)
        return {"slug": slug, "stale_count": 0}
    if auto_refresh:
        result = refresh_digest_job(slug, stale_only=True)
        if _refresh_started(result):
            _clear_notify_state(slug)
            return {
                "slug": slug,
                "stale_count": count,
                "refreshed": True,
                "status": result.get("status") or result.get("message"),
            }
        if result.get("ok"):
            stale_after = check_corpus_staleness(slug)
            return {
                "slug": slug,
                "stale_count": int(stale_after.get("stale_count") or 0),
                "idle": True,
            }
        return {"slug": slug, "stale_count": count, "error": str(result.get("error") or "refresh failed")}
    if _notify_stale_corpus(slug, count):
        return {"slug": slug, "stale_count": count, "notified": True}
    return {"slug": slug, "stale_count": count, "notify_skipped": True}


def run_bounded_staleness_maintenance(*, source: str = "heartbeat") -> dict[str, Any]:
    """Check up to ``max_corpora_per_run`` corpora; refresh or notify when stale."""
    if not is_heartbeat_enabled() and source == "heartbeat":
        return {"ok": True, "skipped": "heartbeat_disabled", "source": source}

    cfg = get_effective_config()
    max_corpora = max(1, min(50, int(cfg.get("max_corpora_per_run", 10))))
    max_concurrent = max(1, min(8, int(cfg.get("max_concurrent_checks", 2))))
    auto_refresh = bool(cfg.get("auto_refresh_stale", True))

    candidates = _select_sweep_candidates(_eligible_sweep_slugs(), max_corpora)

    checked = 0
    stale_total = 0
    refreshed: list[str] = []
    notified: list[str] = []
    skipped: list[str] = []
    errors: list[str] = []

    if not candidates:
        return {
            "ok": True,
            "source": source,
            "checked": 0,
            "stale_total": 0,
            "refreshed": refreshed,
            "notified": notified,
            "skipped": skipped,
            "errors": errors,
        }

    with ThreadPoolExecutor(max_workers=max_concurrent) as pool:
        futures = {pool.submit(_process_stale_slug, slug, auto_refresh=auto_refresh): slug for slug in candidates}
        for fut in as_completed(futures):
            checked += 1
            try:
                result = fut.result()
            except Exception as exc:
                errors.append(str(exc))
                continue
            if result.get("skipped"):
                skipped.append(str(result.get("slug") or ""))
                continue
            stale_total += int(result.get("stale_count") or 0)
            slug = str(result.get("slug") or "")
            if result.get("refreshed"):
                refreshed.append(slug)
            elif result.get("notified"):
                notified.append(slug)
            elif result.get("error"):
                errors.append(f"{slug}: {result['error']}")

    return {
        "ok": True,
        "source": source,
        "checked": checked,
        "stale_total": stale_total,
        "refreshed": refreshed,
        "notified": notified,
        "skipped": skipped,
        "errors": errors,
    }


def handle_watch_event(slug: str, changed_path: str) -> None:
    """Debounced handler for corpus path changes."""
    if not is_watch_enabled():
        return
    cfg = get_effective_config()
    auto_refresh = bool(cfg.get("auto_refresh_stale", True))
    try:
        result = _process_stale_slug(slug, auto_refresh=auto_refresh)
        if result.get("refreshed"):
            _log.info(
                "knowledge watch refreshed %s (%s)",
                slug,
                result.get("status") or "started",
            )
        elif result.get("notified"):
            _log.info("knowledge watch notified stale corpus %s", slug)
        elif result.get("error"):
            _log.warning("knowledge watch refresh failed for %s: %s", slug, result["error"])
    except Exception:
        _log.debug("knowledge watch handler failed for %s (%s)", slug, changed_path, exc_info=True)


@dataclass
class _CorpusWatchTarget:
    slugs: list[str]
    path: str
    debounce_sec: int


@dataclass
class _DebouncedSlugDispatcher:
    _last_fired: dict[tuple[str, str], float] = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def should_fire(self, slug: str, changed_path: str, debounce_sec: int) -> bool:
        key = (slug, changed_path)
        now = time.time()
        with self._lock:
            prev = self._last_fired.get(key, 0.0)
            if now - prev < debounce_sec:
                return False
            self._last_fired[key] = now
            return True


def _dispatch_watch_events(
    slugs: list[str],
    changed_path: str,
    debouncer: _DebouncedSlugDispatcher,
    debounce_sec: int,
) -> None:
    for slug in slugs:
        if debouncer.should_fire(slug, changed_path, debounce_sec):
            handle_watch_event(slug, changed_path)


def _normalize_watch_path(path: str) -> str:
    if not path:
        return ""
    return os.path.realpath(os.path.expanduser(path))


def _path_is_relevant(changed_path: str, watch_root: str, *, ignore_hidden: bool) -> bool:
    if not changed_path:
        return False
    base = os.path.basename(changed_path)
    if ignore_hidden and base.startswith("."):
        return False
    root = _normalize_watch_path(watch_root)
    try:
        candidate = _normalize_watch_path(changed_path)
    except OSError:
        parent = os.path.dirname(changed_path)
        candidate = _normalize_watch_path(parent) if parent else changed_path
    try:
        return os.path.commonpath([root, candidate]) == root
    except ValueError:
        return False


class _CorpusWatchHandler(FileSystemEventHandler):
    def __init__(
        self,
        target: _CorpusWatchTarget,
        debouncer: _DebouncedSlugDispatcher,
        *,
        ignore_hidden: bool,
    ) -> None:
        super().__init__()
        self.target = target
        self.debouncer = debouncer
        self.ignore_hidden = ignore_hidden

    def _emit(self, path: str) -> None:
        if not _path_is_relevant(path, self.target.path, ignore_hidden=self.ignore_hidden):
            return
        _dispatch_watch_events(self.target.slugs, path, self.debouncer, self.target.debounce_sec)

    def on_created(self, event):  # type: ignore[override]
        if not getattr(event, "is_directory", False):
            self._emit(getattr(event, "src_path", ""))

    def on_modified(self, event):  # type: ignore[override]
        if not getattr(event, "is_directory", False):
            self._emit(getattr(event, "src_path", ""))

    def on_moved(self, event):  # type: ignore[override]
        if not getattr(event, "is_directory", False):
            self._emit(getattr(event, "dest_path", ""))

    def on_deleted(self, event):  # type: ignore[override]
        if not getattr(event, "is_directory", False):
            self._emit(getattr(event, "src_path", ""))


class KnowledgeCorpusWatchManager:
    """Watch corpus source folders; uses watchdog when available."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._observer = None
        self._poll_thread: threading.Thread | None = None
        self._maintain_thread: threading.Thread | None = None
        self._stop_event = threading.Event()
        self._targets: list[_CorpusWatchTarget] = []
        self._debouncer = _DebouncedSlugDispatcher()
        self._poll_state: dict[str, dict[str, float]] = {}
        self._poll_baselines_done: set[str] = set()
        self._target_fingerprint: tuple[tuple[str, tuple[str, ...]], ...] = ()
        self._using_watchdog = False

    def _observer_alive(self) -> bool:
        if self._observer is None:
            return False
        try:
            return self._observer.is_alive()
        except Exception:
            return False

    def is_running(self) -> bool:
        if self._maintain_thread is not None and self._maintain_thread.is_alive():
            return True
        if self._observer_alive():
            return True
        return bool(self._poll_thread and self._poll_thread.is_alive())

    def _build_targets(self) -> list[_CorpusWatchTarget]:
        cfg = get_effective_config()
        debounce = max(5, min(3600, int(cfg.get("watch_debounce_seconds", 30))))
        targets: list[_CorpusWatchTarget] = []
        for path, slugs in consolidate_watch_path_map(iter_watch_path_map()).items():
            if not os.path.isdir(path):
                _log.warning("knowledge watch: path missing, skipping %s (%s)", slugs, path)
                continue
            targets.append(_CorpusWatchTarget(slugs=list(slugs), path=path, debounce_sec=debounce))
        return targets

    @staticmethod
    def _fingerprint(targets: list[_CorpusWatchTarget]) -> tuple[tuple[str, tuple[str, ...]], ...]:
        return tuple((t.path, tuple(t.slugs)) for t in targets)

    def _start_observer(self, targets: list[_CorpusWatchTarget]) -> None:
        cfg = get_effective_config()
        ignore_hidden = bool(cfg.get("watch_ignore_hidden", True))
        if not (_HAS_WATCHDOG and Observer is not None):
            return
        self._observer = Observer()
        for target in targets:
            self._observer.schedule(
                _CorpusWatchHandler(target, self._debouncer, ignore_hidden=ignore_hidden),
                target.path,
                recursive=True,
            )
        self._observer.start()
        self._using_watchdog = True
        _log.info("knowledge watch started with watchdog (%d paths)", len(targets))

    def _stop_observer(self, timeout: float = 5.0) -> None:
        if self._observer is None:
            self._using_watchdog = False
            return
        try:
            self._observer.stop()
            self._observer.join(timeout=timeout)
        except Exception:
            _log.debug("knowledge watch observer stop failed", exc_info=True)
        self._observer = None
        self._using_watchdog = False

    def _sync_watch_runtime_locked(self) -> None:
        if self._stop_event.is_set():
            return
        if not is_watch_enabled():
            self._stop_observer()
            self._targets = []
            self._target_fingerprint = ()
            return

        new_targets = self._build_targets()
        new_fp = self._fingerprint(new_targets)
        old_fp = self._target_fingerprint

        if not new_targets:
            self._stop_observer()
            self._targets = []
            self._target_fingerprint = ()
            return

        self._targets = new_targets
        self._target_fingerprint = new_fp

        if _HAS_WATCHDOG and Observer is not None:
            if not self._observer_alive() or new_fp != old_fp:
                self._stop_observer()
                self._start_observer(new_targets)
            return

        if self._poll_thread is None or not self._poll_thread.is_alive():
            self._poll_thread = threading.Thread(
                target=self._poll_loop,
                daemon=True,
                name="knowledge-corpus-watch-poll",
            )
            self._poll_thread.start()
            _log.info("knowledge watch started in polling mode (%d paths)", len(new_targets))

    def ensure_service(self) -> int:
        """Start maintain loop once; sync watch runtime from current config/corpora."""
        with self._lock:
            if self._maintain_thread is None or not self._maintain_thread.is_alive():
                self._stop_event.clear()
                self._maintain_thread = threading.Thread(
                    target=self._maintain_loop,
                    daemon=True,
                    name="knowledge-corpus-watch-maintain",
                )
                self._maintain_thread.start()
            self._sync_watch_runtime_locked()
            return len(self._targets)

    def start(self) -> int:
        return self.ensure_service()

    def stop(self, timeout: float = 5.0) -> None:
        self._stop_event.set()
        with self._lock:
            self._stop_observer(timeout=timeout)
            poll_thread = self._poll_thread
            maintain_thread = self._maintain_thread
            self._poll_thread = None
            self._maintain_thread = None
            self._targets = []
            self._target_fingerprint = ()
            self._poll_state = {}
            self._poll_baselines_done = set()
        if poll_thread is not None:
            poll_thread.join(timeout=timeout)
        if maintain_thread is not None:
            maintain_thread.join(timeout=timeout)

    def _maintain_loop(self) -> None:
        while not self._stop_event.is_set():
            try:
                if not self._stop_event.is_set():
                    with self._lock:
                        if self._stop_event.is_set():
                            break
                        self._sync_watch_runtime_locked()
            except Exception:
                _log.debug("knowledge watch maintain loop failed", exc_info=True)
            if self._stop_event.wait(_TARGET_SYNC_SECONDS):
                break

    def _poll_scan_target(
        self,
        target: _CorpusWatchTarget,
        *,
        ignore_hidden: bool,
        baseline: bool,
    ) -> None:
        state = self._poll_state.setdefault(target.path, {})
        seen: set[str] = set()
        for root, dirs, files in os.walk(target.path):
            if ignore_hidden:
                dirs[:] = [d for d in dirs if not d.startswith(".")]
                files = [f for f in files if not f.startswith(".")]
            for name in files:
                full = os.path.join(root, name)
                seen.add(full)
                try:
                    mtime = os.path.getmtime(full)
                except OSError:
                    continue
                prev = state.get(full)
                if prev is None:
                    state[full] = mtime
                    if not baseline:
                        _dispatch_watch_events(target.slugs, full, self._debouncer, target.debounce_sec)
                    continue
                if mtime > prev:
                    state[full] = mtime
                    _dispatch_watch_events(target.slugs, full, self._debouncer, target.debounce_sec)
        deleted = [path for path in list(state.keys()) if path not in seen]
        for path in deleted:
            state.pop(path, None)
            _dispatch_watch_events(target.slugs, path, self._debouncer, target.debounce_sec)

    def _poll_loop(self) -> None:
        while not self._stop_event.is_set():
            if not is_watch_enabled():
                if self._stop_event.wait(_POLL_INTERVAL_SECONDS):
                    break
                continue
            cfg = get_effective_config()
            ignore_hidden = bool(cfg.get("watch_ignore_hidden", True))
            with self._lock:
                if self._stop_event.is_set():
                    break
                targets = list(self._targets)
                active_paths = {target.path for target in targets}
                self._poll_baselines_done.intersection_update(active_paths)
            for target in targets:
                if self._stop_event.is_set():
                    break
                try:
                    baseline = target.path not in self._poll_baselines_done
                    self._poll_scan_target(target, ignore_hidden=ignore_hidden, baseline=baseline)
                    if baseline:
                        self._poll_baselines_done.add(target.path)
                except Exception:
                    _log.debug("knowledge watch poll failed for %s", target.slugs, exc_info=True)
            if self._stop_event.wait(_POLL_INTERVAL_SECONDS):
                break
