# cron_scheduler.py — Daemon thread that polls scheduled_jobs and executes due tasks.
# Started from mugi_backend.py on app launch. Runs tasks in fresh subagent sessions.
from __future__ import annotations

import logging
import os
import re
import threading
import time
from datetime import datetime, timedelta

_log = logging.getLogger(__name__)

_scheduler_thread: threading.Thread | None = None
_stop_event = threading.Event()
_POLL_INTERVAL_SECONDS = 30


def start_scheduler() -> None:
    """Start the cron scheduler daemon thread (idempotent)."""
    global _scheduler_thread
    if _scheduler_thread is not None and _scheduler_thread.is_alive():
        return
    _stop_event.clear()
    _scheduler_thread = threading.Thread(target=_scheduler_loop, daemon=True, name="cron-scheduler")
    _scheduler_thread.start()
    _log.info("Cron scheduler started.")


def stop_scheduler() -> None:
    """Signal the scheduler to stop."""
    _stop_event.set()
    if _scheduler_thread is not None:
        _scheduler_thread.join(timeout=10)
    _log.info("Cron scheduler stopped.")


def _scheduler_loop() -> None:
    """Main loop: poll for due jobs, execute them, update status."""
    while not _stop_event.is_set():
        try:
            _process_due_jobs()
        except Exception as e:
            _log.error("Cron scheduler error: %s", e)
        _stop_event.wait(timeout=_POLL_INTERVAL_SECONDS)


def _process_due_jobs() -> None:
    """Find and execute all active jobs whose next_run_at has passed."""
    try:
        from chat_storage import _get_connection
    except ImportError:
        return

    now_iso = datetime.now().isoformat()
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT id, prompt, schedule, delivery, origin_thread_id, repeat_count, "
            "runs_completed, timezone, cron_expr, webhook_url, retry_count, max_retries, "
            "last_error, model_override, kanban_assignee, kanban_tenant, kanban_metadata_json "
            "FROM scheduled_jobs WHERE status = 'active' AND next_run_at <= ?",
            (now_iso,),
        ).fetchall()
    finally:
        conn.close()

    for row in rows:
        (job_id, prompt, schedule, delivery, origin_tid, repeat_count,
         runs_completed, tz, cron_expr, webhook_url, retry_count,
         max_retries, last_error, model_override, kanban_assignee,
         kanban_tenant, kanban_metadata_json) = row

        _log.info("Executing scheduled job %s: %s", job_id, prompt[:60])
        try:
            from activity_bus import emit_activity_enriched
            emit_activity_enriched({
                "source": "cron",
                "state": "started",
                "summary": f"Cron job started: {prompt[:80]}",
                "thread_id": origin_tid or "",
                "job_id": str(job_id),
                "meta": {"delivery": delivery, "schedule": schedule},
            }, tags=["scheduled"])
        except Exception:
            pass

        _cron_t0 = time.time()
        try:
            if (kanban_assignee or "").strip():
                result = _drop_to_kanban(
                    job_id=str(job_id),
                    prompt=prompt,
                    assignee=str(kanban_assignee or "").strip(),
                    tenant=str(kanban_tenant or "default").strip() or "default",
                    metadata_json=kanban_metadata_json or "{}",
                    origin_thread_id=origin_tid,
                    next_run_at=now_iso,
                    model_override=model_override or "",
                )
            else:
                result = _execute_job(prompt, origin_thread_id=origin_tid, model_override=model_override)
            _deliver_result(job_id, prompt, result, delivery, origin_tid, webhook_url)
            try:
                from utils import task_ledger as _tl
                _tl.record_run(
                    _tl.KIND_CRON,
                    f"job_{job_id}",
                    thread_id=origin_tid or None,
                    status=_tl.STATUS_OK,
                    duration_ms=int((time.time() - _cron_t0) * 1000),
                    meta={"delivery": delivery, "result_chars": len(result or "")},
                )
            except Exception:
                _log.debug("ledger record (cron ok) failed", exc_info=True)
            try:
                from core import get_workspace_path
                from utils.agent_plane import record_proactive_turn_summary

                workspace = get_workspace_path()
                record_proactive_turn_summary(
                    workspace=workspace,
                    wake_kind="cron",
                    thread_id=(str(origin_tid or "").strip() or "default"),
                    status="HEARTBEAT_OK",
                    one_line_summary=(result or "").splitlines()[0][:220] if result else "",
                )
            except Exception:
                _log.warning("cron: failed to record proactive turn summary", exc_info=True)
            # Reset retry count on success
            _update_retry_count(job_id, 0, "")
            if retry_count and retry_count > 0:
                try:
                    from activity_bus import emit_system_event
                    emit_system_event(
                        f"Scheduled task recovered: {prompt[:100]}",
                        source="cron",
                        thread_id=origin_tid or "",
                    )
                except Exception:
                    pass
            try:
                from activity_bus import emit_activity_enriched
                emit_activity_enriched({
                    "source": "cron",
                    "state": "completed",
                    "summary": f"Cron job completed: {prompt[:80]}",
                    "thread_id": origin_tid or "",
                    "job_id": str(job_id),
                }, tags=["scheduled"])
            except Exception:
                pass
        except Exception as e:
            _log.error("Job %s execution failed: %s", job_id, e)
            try:
                from utils import task_ledger as _tl
                _tl.record_run(
                    _tl.KIND_CRON,
                    f"job_{job_id}",
                    thread_id=origin_tid or None,
                    status=_tl.STATUS_ERROR,
                    duration_ms=int((time.time() - _cron_t0) * 1000),
                    error_summary=str(e),
                )
            except Exception:
                _log.debug("ledger record (cron error) failed", exc_info=True)
            new_retry = (retry_count or 0) + 1
            if new_retry <= (max_retries or 2):
                backoff_minutes = 2 ** min(new_retry, 6)
                _log.info("Job %s: scheduling retry %d/%d in %d minutes",
                          job_id, new_retry, max_retries or 2, backoff_minutes)
                _schedule_retry(job_id, new_retry, str(e), backoff_minutes)
                continue
            else:
                _update_retry_count(job_id, new_retry, str(e)[:500])
                try:
                    from activity_bus import emit_system_event
                    emit_system_event(
                        f"Scheduled task failed after {max_retries or 2} retries: {prompt[:100]}",
                        source="cron",
                        thread_id=origin_tid or "",
                    )
                except Exception:
                    pass
            try:
                from activity_bus import emit_activity_enriched
                emit_activity_enriched({
                    "source": "cron",
                    "state": "error",
                    "summary": f"Cron job failed: {prompt[:80]}",
                    "thread_id": origin_tid or "",
                    "job_id": str(job_id),
                    "meta": {"error": str(e)[:300]},
                }, tags=["scheduled"])
            except Exception:
                pass

        runs_completed += 1
        _update_job_after_run(job_id, schedule, repeat_count, runs_completed,
                              cron_expr=cron_expr, timezone=tz)


def _drop_to_kanban(
    *,
    job_id: str,
    prompt: str,
    assignee: str,
    tenant: str,
    metadata_json: str,
    origin_thread_id: str | None,
    next_run_at: str,
    model_override: str,
) -> str:
    """Plan B: when a cron job has ``kanban_assignee`` set, drop a kanban task
    instead of running inline. Idempotency key ``cron:{job_id}:{run_at_iso}``
    means a missed-run replay returns the original task instead of duplicating.
    Returns a one-line confirmation string for the delivery channel.
    """
    import json as _json
    metadata: dict = {}
    try:
        parsed = _json.loads(metadata_json or "{}")
        if isinstance(parsed, dict):
            metadata = parsed
    except Exception:
        metadata = {}
    metadata.setdefault("source", "cron")
    metadata.setdefault("cron_job_id", job_id)
    title = (prompt or "").splitlines()[0][:240] or f"cron:{job_id}"
    try:
        import kanban_storage as ks
        task = ks.create_task(
            title=title,
            body=prompt or "",
            assignee=assignee or "general",
            tenant=tenant or "default",
            priority=3,
            metadata=metadata,
            origin_thread_id=str(origin_thread_id or ""),
            idempotency_key=f"cron:{job_id}:{next_run_at}",
            model_override=model_override or "",
        )
    except RuntimeError as exc:
        return f"[cron->kanban] disabled: {exc}"
    except Exception as exc:
        _log.warning("cron->kanban create_task failed for job %s", job_id, exc_info=True)
        return f"[cron->kanban] failed: {exc}"
    try:
        from utils.kanban_dispatcher import nudge_dispatcher
        nudge_dispatcher()
    except Exception:
        pass
    tid = (task or {}).get("id") if isinstance(task, dict) else "?"
    return f"[cron->kanban] queued task {tid} (assignee={assignee}, tenant={tenant})"


def _execute_job(prompt: str, origin_thread_id: str | None = None, model_override: str = "") -> str:
    """Run the job prompt in a fresh subagent session."""
    from subagent_runner import run_subagent
    from tool_policy import context_from_prefs_and_flags
    from utils.agent_plane import build_wake_context_snippet, get_agent_plane_config

    origin = str(origin_thread_id or "").strip()
    # Cron subagent runs are not charged against drive_engine proactive budget (drive_action only).
    kwargs: dict = {
        "task_description": prompt,
        "agent_type": "general",
        "timeout_seconds": 120,
        "policy_context": context_from_prefs_and_flags(scheduled_job=True, unattended_external=True, subagent_mode=True),
    }
    if origin:
        kwargs["parent_thread_id"] = origin
        cfg = get_agent_plane_config()
        if bool(cfg.get("enabled", True)):
            try:
                from core import get_workspace_path

                snippet = build_wake_context_snippet(
                    workspace=get_workspace_path(),
                    wake_kind="cron",
                    task_name="cron",
                )
                if snippet and snippet.strip():
                    kwargs["context_snippet"] = snippet
            except Exception as e:
                _log.warning("cron: wake context snippet unavailable (%s)", e)
    if model_override:
        kwargs["model_override"] = model_override
    return run_subagent(**kwargs)


def _deliver_result(
    job_id: str, prompt: str, result: str, delivery: str,
    origin_tid: str, webhook_url: str = "",
) -> None:
    """Enqueue the job result on the runner-owned delivery queue.

    The actual delivery (calling ``_deliver_to_*`` helpers below) is owned by
    ``mugi_runner``'s drain thread — see ``utils.delivery_queue``. Keeping
    ``_deliver_to_*`` helpers here so the drain thread can import them.
    """
    _enqueue_delivery(job_id, prompt, result, delivery, origin_tid, webhook_url)


def _enqueue_delivery(
    job_id: str, prompt: str, result: str, delivery: str,
    origin_tid: str, webhook_url: str = "",
) -> None:
    """Append one delivery entry to ``~/.mugi/delivery_queue.jsonl``."""
    from utils import delivery_queue as dq

    if delivery == "file":
        dq.enqueue(
            dq.KIND_FILE,
            {"job_id": str(job_id), "prompt": prompt, "result": result},
            run_id=str(job_id),
            origin_thread_id=origin_tid or None,
        )
    elif delivery == "notification":
        dq.enqueue(
            dq.KIND_NOTIFICATION,
            {"prompt": prompt, "result": result},
            run_id=str(job_id),
            origin_thread_id=origin_tid or None,
        )
    elif delivery == "webhook" and webhook_url:
        dq.enqueue(
            dq.KIND_WEBHOOK,
            {"url": webhook_url, "job_id": str(job_id), "prompt": prompt, "result": result},
            run_id=str(job_id),
            origin_thread_id=origin_tid or None,
        )
    else:
        dq.enqueue(
            dq.KIND_THREAD,
            {"thread_id": origin_tid or "", "result": result},
            run_id=str(job_id),
            origin_thread_id=origin_tid or None,
        )


def _deliver_to_thread(result: str, thread_id: str, banner: str = "Scheduled task result") -> None:
    """Append the result as an assistant message to the origin thread.

    ``banner`` customizes the bracketed prefix so timer fires surface as
    ``[Timer]`` rather than ``[Scheduled task result]``.
    """
    try:
        from chat_storage import append_messages
        tag = (str(banner or "Scheduled task result").strip() or "Scheduled task result")[:80]
        content = f"[{tag}]\n\n{result}"
        append_messages(thread_id, [{"role": "assistant", "content": content}])
    except Exception as e:
        _log.error("Failed to deliver to thread %s: %s", thread_id, e)


def _deliver_to_file(job_id: str, prompt: str, result: str) -> None:
    """Save the result to a file in the workspace."""
    try:
        from core import get_workspace_path
        workspace = get_workspace_path()
        output_dir = os.path.join(workspace, "scheduled_results")
        os.makedirs(output_dir, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"job_{job_id}_{timestamp}.md"
        filepath = os.path.join(output_dir, filename)
        content = (
            f"# Scheduled Task Result\n\n**Job:** {job_id}\n**Prompt:** {prompt}\n"
            f"**Completed:** {datetime.now().isoformat()}\n\n---\n\n{result}\n"
        )
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        os.chmod(filepath, 0o600)
    except Exception as e:
        _log.error("Failed to save result to file for job %s: %s", job_id, e)


def _deliver_notification(prompt: str, result: str) -> None:
    """Send a macOS notification with the result summary."""
    try:
        from utils.notifications import send_mac_notification
        body = result[:400] if result else "(no result)"
        send_mac_notification(
            title="Mugi: Scheduled Task",
            subtitle=prompt[:80],
            body=body,
        )
    except Exception as e:
        _log.error("Failed to send notification: %s", e)


_WEBHOOK_TIMEOUT_SECONDS = 15
_WEBHOOK_MAX_RESPONSE_BYTES = 64 * 1024


def _resolve_outbound_target(target: str) -> tuple[str | None, str | None]:
    """Resolve a cron webhook identifier to (slug, url).

    Accepts either a raw slug, a ``slug:<slug>`` prefix, or a full URL that
    matches a registered outbound webhook. Returns (None, None) when the
    target cannot be resolved to a registered entry, in which case we fall
    back to the legacy allowlist.
    """
    raw = (target or "").strip()
    if not raw:
        return None, None
    try:
        from persistence import outbound_webhooks_get, outbound_webhooks_list
    except Exception:
        return None, None

    slug_candidate = raw
    if raw.startswith("slug:"):
        slug_candidate = raw.split(":", 1)[1].strip()
    entry = outbound_webhooks_get(slug_candidate) if slug_candidate else None
    if entry:
        return str(entry.get("slug") or slug_candidate), str(entry.get("url") or "")

    if raw.startswith("http://") or raw.startswith("https://"):
        try:
            for row in outbound_webhooks_list():
                if str(row.get("url") or "") == raw:
                    return str(row.get("slug") or ""), raw
        except Exception:
            pass
    return None, None


def _validate_webhook_url(url: str) -> bool:
    """Accept registered outbound targets or the legacy allowlist.

    The outbound registry is the preferred path. The legacy config allowlist
    remains so pre-existing cron rows without a slug still deliver until users
    migrate them.
    """
    if not url:
        return False
    slug, _resolved = _resolve_outbound_target(url)
    if slug:
        return True
    try:
        from core.config import get_config
        cfg = get_config()
        allowed = (cfg.get("hooks") or {}).get("allowed_webhook_urls", [])
        if not allowed:
            return False
        from urllib.parse import urlparse
        parsed = urlparse(url)
        for pattern in allowed:
            if parsed.netloc == pattern or url.startswith(pattern):
                return True
    except Exception:
        pass
    return False


def _is_https_or_loopback(url: str) -> bool:
    from urllib.parse import urlparse
    try:
        parsed = urlparse(url)
    except Exception:
        return False
    if parsed.scheme == "https":
        return True
    if parsed.scheme == "http":
        host = (parsed.hostname or "").lower()
        if host in ("localhost", "127.0.0.1", "::1"):
            return True
    return False


class _NoRedirectHandler:
    """urllib handler factory that refuses any redirect."""

    @staticmethod
    def build():
        import urllib.request

        class _NR(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: D401
                raise urllib.error.HTTPError(
                    req.full_url, code, f"redirect refused ({code})", headers, fp,
                )
        return urllib.request.build_opener(_NR())


def _deliver_webhook(target: str, job_id: str, prompt: str, result: str) -> None:
    """POST the cron result to a registered outbound webhook target.

    ``target`` may be a slug, ``slug:<slug>``, or a URL that matches a
    registered entry. HMAC signing happens when a Keychain secret exists for
    the slug. All transport is hardened: no redirects, strict timeout, and
    response body is capped + discarded.
    """
    if not _validate_webhook_url(target):
        _log.warning("Webhook target %s not registered, skipping delivery", target)
        return

    slug, resolved_url = _resolve_outbound_target(target)
    url = resolved_url or target
    if not _is_https_or_loopback(url):
        _log.warning("Webhook %s rejected: non-https and non-loopback", url)
        return

    try:
        import hashlib
        import hmac as _hmac
        import json
        import secrets as _secrets
        import urllib.error
        import urllib.request

        delivery_id = _secrets.token_hex(12)
        payload = json.dumps({
            "job_id": job_id,
            "prompt": prompt[:200],
            "result": result[:4000],
            "delivery_id": delivery_id,
            "completed_at": datetime.now().isoformat(),
        }).encode("utf-8")

        headers = {
            "Content-Type": "application/json",
            "X-Mugi-Delivery-Id": delivery_id,
            "User-Agent": "Mugi-Webhook/1",
        }

        if slug:
            try:
                from persistence import keychain_get_outbound_webhook_hmac
                secret = keychain_get_outbound_webhook_hmac(slug)
                if secret:
                    digest = _hmac.new(
                        secret.encode("utf-8"), payload, hashlib.sha256,
                    ).hexdigest()
                    headers["X-Mugi-Signature"] = f"sha256={digest}"
            except Exception:
                _log.debug("HMAC sign for %s failed", slug, exc_info=True)

        req = urllib.request.Request(url, data=payload, headers=headers, method="POST")
        opener = _NoRedirectHandler.build()
        try:
            with opener.open(req, timeout=_WEBHOOK_TIMEOUT_SECONDS) as resp:
                resp.read(_WEBHOOK_MAX_RESPONSE_BYTES + 1)
                _log.info("Webhook delivered to %s (slug=%s): %d", url, slug or "-", resp.status)
        except urllib.error.HTTPError as e:
            _log.error("Webhook %s HTTP %s", url, e.code)
            raise
    except Exception as e:
        _log.error("Webhook delivery to %s failed: %s", url, e)
        raise


def _update_retry_count(job_id: str, retry_count: int, error: str) -> None:
    """Update retry count and last error on a job."""
    try:
        from chat_storage import _get_connection
        conn = _get_connection()
        try:
            conn.execute(
                "UPDATE scheduled_jobs SET retry_count = ?, last_error = ? WHERE id = ?",
                (retry_count, error[:500], job_id),
            )
            conn.commit()
        finally:
            conn.close()
    except Exception as e:
        _log.error("Failed to update retry count for %s: %s", job_id, e)


def _schedule_retry(job_id: str, retry_count: int, error: str, delay_minutes: int) -> None:
    """Schedule a retry by bumping next_run_at forward."""
    try:
        from chat_storage import _get_connection
        next_run = (datetime.now() + timedelta(minutes=delay_minutes)).isoformat()
        conn = _get_connection()
        try:
            conn.execute(
                "UPDATE scheduled_jobs SET retry_count = ?, last_error = ?, next_run_at = ? WHERE id = ?",
                (retry_count, error[:500], next_run, job_id),
            )
            conn.commit()
        finally:
            conn.close()
    except Exception as e:
        _log.error("Failed to schedule retry for %s: %s", job_id, e)


def _update_job_after_run(
    job_id: str, schedule: str, repeat_count: int, runs_completed: int,
    cron_expr: str = "", timezone: str = "",
) -> None:
    """Update the job: compute next run or mark completed."""
    try:
        from chat_storage import _get_connection
    except ImportError:
        return

    now_iso = datetime.now().isoformat()

    if runs_completed >= repeat_count:
        conn = _get_connection()
        try:
            conn.execute(
                "UPDATE scheduled_jobs SET status = 'completed', runs_completed = ?, last_run_at = ? WHERE id = ?",
                (runs_completed, now_iso, job_id),
            )
            conn.commit()
        finally:
            conn.close()
        return

    next_run = _compute_next_run(schedule, cron_expr=cron_expr, timezone=timezone)
    conn = _get_connection()
    try:
        conn.execute(
            "UPDATE scheduled_jobs SET runs_completed = ?, last_run_at = ?, next_run_at = ?, retry_count = 0 WHERE id = ?",
            (runs_completed, now_iso, next_run.isoformat() if next_run else now_iso, job_id),
        )
        conn.commit()
    finally:
        conn.close()


def _compute_next_run(schedule: str, cron_expr: str = "", timezone: str = "") -> datetime:
    """Compute the next run time from the schedule string or cron expression.

    When a timezone is specified, the returned datetime is in system local time
    so that ``_process_due_jobs`` (which compares against ``datetime.now()``)
    fires the job at the correct wall-clock moment.
    """
    now = datetime.now()

    # Cron expression takes priority
    if cron_expr:
        try:
            from croniter import croniter
            if timezone:
                try:
                    from zoneinfo import ZoneInfo
                    tz = ZoneInfo(timezone)
                    now_in_tz = datetime.now(tz)
                    next_aware = croniter(cron_expr, now_in_tz).get_next(datetime)
                    if next_aware.tzinfo is None:
                        next_aware = next_aware.replace(tzinfo=tz)
                    return next_aware.astimezone().replace(tzinfo=None)
                except Exception:
                    pass
            return croniter(cron_expr, now).get_next(datetime)
        except Exception as e:
            _log.warning("croniter failed for '%s': %s", cron_expr, e)

    s = schedule.strip().lower()

    # Daily at HH:MM
    daily = re.match(r"daily\s+at\s+(\d{1,2}):(\d{2})", s)
    if daily:
        h, m = int(daily.group(1)), int(daily.group(2))
        next_dt = (now + timedelta(days=1)).replace(hour=h, minute=m, second=0, microsecond=0)
        return next_dt

    # Relative intervals
    rel = re.match(r"in\s+(\d+)\s+(minute|hour|day|min|hr)s?", s)
    if rel:
        n = int(rel.group(1))
        unit = rel.group(2)
        if unit in ("minute", "min"):
            return now + timedelta(minutes=n)
        elif unit in ("hour", "hr"):
            return now + timedelta(hours=n)
        elif unit == "day":
            return now + timedelta(days=n)

    # Fallback: 24 hours from now
    return now + timedelta(hours=24)


def run_job_now(job_id: str) -> str:
    """Mark an active job due immediately. Returns ``scheduled``, ``inactive``, or ``not_found``."""
    jid = (job_id or "").strip()
    if not jid:
        return "not_found"
    try:
        from chat_storage import _get_connection

        conn = _get_connection()
        try:
            row = conn.execute(
                "SELECT status FROM scheduled_jobs WHERE id = ?",
                (jid,),
            ).fetchone()
            if not row:
                return "not_found"
            if str(row[0] or "") != "active":
                return "inactive"
            now_iso = datetime.now().isoformat()
            conn.execute(
                "UPDATE scheduled_jobs SET next_run_at = ? WHERE id = ? AND status = 'active'",
                (now_iso, jid),
            )
            conn.commit()
            return "scheduled"
        finally:
            conn.close()
    except Exception:
        _log.warning("run_job_now failed for %s", job_id, exc_info=True)
        return "not_found"
