"""Cross-platform resident scheduler for project-owned workflows."""

from __future__ import annotations

import calendar
import json
import os
import subprocess
import sys
import time
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Protocol
from zoneinfo import ZoneInfo

from gyxx_flow.catalog import ScheduleEntry, WorkflowCatalog
from gyxx_flow.core.artifacts import atomic_write_json

_TMALL_BAIBU_WORKFLOW_ID = "product.tmall_baibu_apply"
_TMALL_BAIBU_SCHEDULE_IMPORT_MODE = "without_hyperlinks"


def latest_due_slot(schedule: ScheduleEntry, now: datetime) -> datetime:
    """Return the latest scheduled wall-clock slot at or before ``now``."""
    if now.tzinfo is None:
        raise ValueError("scheduler time must be timezone-aware")

    candidates = [
        _latest_due_slot_at(schedule, now, wall_clock)
        for wall_clock in schedule.effective_times
    ]
    eligible = [candidate for candidate in candidates if candidate <= now]
    if eligible:
        return max(eligible)
    return min(candidates)


def _latest_due_slot_at(
    schedule: ScheduleEntry,
    now: datetime,
    wall_clock: str,
) -> datetime:
    hour, minute = (int(part) for part in wall_clock.split(":"))

    def at(day: date) -> datetime:
        return datetime(day.year, day.month, day.day, hour, minute, tzinfo=now.tzinfo)

    if schedule.kind == "daily":
        candidate = at(now.date())
        return candidate if candidate <= now else at(now.date() - timedelta(days=1))

    if schedule.kind == "weekly":
        for offset in range(8):
            candidate = at(now.date() - timedelta(days=offset))
            if candidate.strftime("%A") in schedule.days and candidate <= now:
                return candidate
        raise ValueError(f"weekly schedule has no due slot: {schedule.workflow_id}")

    if schedule.kind == "monthly":
        if schedule.day_of_month is None:
            raise ValueError(f"monthly schedule has no day: {schedule.workflow_id}")
        year, month = now.year, now.month
        for _ in range(24):
            if schedule.day_of_month <= calendar.monthrange(year, month)[1]:
                candidate = datetime(year, month, schedule.day_of_month, hour, minute, tzinfo=now.tzinfo)
                if candidate <= now:
                    return candidate
            month -= 1
            if month == 0:
                year -= 1
                month = 12
        raise ValueError(f"monthly schedule has no due slot: {schedule.workflow_id}")

    if schedule.kind == "interval_days":
        if not schedule.anchor_date or not schedule.every_days:
            raise ValueError(f"interval schedule is incomplete: {schedule.workflow_id}")
        anchor = at(date.fromisoformat(schedule.anchor_date))
        if now < anchor:
            return anchor
        elapsed_days = (now.date() - anchor.date()).days
        periods = elapsed_days // schedule.every_days
        candidate = anchor + timedelta(days=periods * schedule.every_days)
        if candidate > now:
            candidate -= timedelta(days=schedule.every_days)
        return candidate

    raise ValueError(f"unsupported schedule kind: {schedule.kind}")


class SchedulerProcess(Protocol):
    pid: int

    def poll(self) -> int | None: ...


class WorkflowLauncher(Protocol):
    def launch(
        self,
        workflow_id: str,
        slot: datetime,
        business_date: date,
    ) -> SchedulerProcess: ...
    def close(self, process: SchedulerProcess) -> None: ...


class SubprocessWorkflowLauncher:
    """Launch workflows through the installed gyxx CLI and retain per-run logs."""

    def __init__(self, project_root: Path, data_root: Path, python_executable: Path | None = None) -> None:
        self.project_root = Path(project_root).resolve()
        self.data_root = Path(data_root).resolve()
        self.python_executable = Path(python_executable or sys.executable).resolve()
        self.log_root = self.data_root / "logs" / "scheduler"
        self._streams: dict[int, object] = {}

    def launch(
        self,
        workflow_id: str,
        slot: datetime,
        business_date: date,
    ) -> subprocess.Popen:
        self.log_root.mkdir(parents=True, exist_ok=True)
        stamp = slot.strftime("%Y%m%dT%H%M%S%z")
        log_path = self.log_root / f"{workflow_id}-{stamp}.log"
        stream = log_path.open("a", encoding="utf-8", newline="\n")
        environment = {
            **os.environ,
            "GYXX_PROJECT_ROOT": str(self.project_root),
            "GYXX_DATA_ROOT": str(self.data_root),
            # Keep the concrete scheduled slot available to the workflow.
            # A workflow may have more than one daily slot; the business date
            # alone cannot distinguish the initial collection from a later
            # same-day retry.
            "GYXX_SCHEDULE_SLOT": slot.isoformat(),
            "GYXX_SCHEDULE_TIME": slot.strftime("%H:%M"),
        }
        kwargs: dict[str, object] = {}
        if os.name == "nt":
            kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
        command = [
            str(self.python_executable),
            "-m",
            "gyxx_flow",
            "run",
            workflow_id,
            "--date",
            business_date.isoformat(),
        ]
        if workflow_id == _TMALL_BAIBU_WORKFLOW_ID:
            command.extend(
                [
                    "--tmall-baibu-import-mode",
                    _TMALL_BAIBU_SCHEDULE_IMPORT_MODE,
                ]
            )
        command.append("--execute")
        try:
            process = subprocess.Popen(
                command,
                cwd=self.project_root,
                env=environment,
                stdin=subprocess.DEVNULL,
                stdout=stream,
                stderr=subprocess.STDOUT,
                shell=False,
                **kwargs,
            )
        except Exception:
            stream.close()
            raise
        self._streams[process.pid] = stream
        return process

    def close(self, process: SchedulerProcess) -> None:
        stream = self._streams.pop(process.pid, None)
        if stream is not None:
            stream.close()  # type: ignore[attr-defined]


@dataclass(slots=True)
class _ActiveRun:
    process: SchedulerProcess
    slot: datetime


class PythonScheduler:
    """Evaluate catalog schedules with persistent at-most-once slot tracking."""

    def __init__(
        self,
        catalog: WorkflowCatalog,
        data_root: Path,
        *,
        launcher: WorkflowLauncher,
        misfire_grace_seconds: int = 6 * 60 * 60,
        dry_run: bool = False,
        catalog_loader: Callable[[], WorkflowCatalog] | None = None,
        maintenance: Callable[[], object] | None = None,
    ) -> None:
        if misfire_grace_seconds < 0:
            raise ValueError("misfire grace must be non-negative")
        self.catalog = catalog
        self.timezone = ZoneInfo(catalog.timezone)
        self.data_root = Path(data_root).resolve()
        self.state_path = self.data_root / "state" / "scheduler" / "state.json"
        self.launcher = launcher
        self.misfire_grace_seconds = misfire_grace_seconds
        self.dry_run = dry_run
        self._catalog_loader = catalog_loader
        self._maintenance = maintenance
        self._state = self._load_state()
        self._active: dict[str, _ActiveRun] = {}

    def _reload_catalog(self) -> None:
        if self._catalog_loader is None:
            return
        catalog = self._catalog_loader()
        self.catalog = catalog
        self.timezone = ZoneInfo(catalog.timezone)

    def _load_state(self) -> dict:
        if not self.state_path.exists():
            return {"schema_version": 1, "updated_at": None, "jobs": {}}
        try:
            payload = json.loads(self.state_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            raise ValueError(f"cannot load scheduler state: {exc}") from exc
        if payload.get("schema_version") != 1 or not isinstance(payload.get("jobs"), dict):
            raise ValueError("invalid scheduler state")
        return payload

    def _save_state(self, now: datetime) -> None:
        self._state["updated_at"] = now.astimezone(timezone.utc).isoformat(timespec="seconds")
        atomic_write_json(self.state_path, self._state)

    def _reap(self, now: datetime) -> None:
        changed = False
        for workflow_id, active in list(self._active.items()):
            return_code = active.process.poll()
            if return_code is None:
                continue
            self.launcher.close(active.process)
            job = self._state["jobs"][workflow_id]
            job.update({
                "status": "success" if return_code == 0 else "failed",
                "exit_code": return_code,
                "finished_at": now.astimezone(timezone.utc).isoformat(timespec="seconds"),
            })
            del self._active[workflow_id]
            changed = True
        if changed:
            self._save_state(now)

    def tick(self, now: datetime | None = None) -> list[str]:
        self._reload_catalog()
        current = (now or datetime.now(self.timezone)).astimezone(self.timezone)
        self._reap(current)
        if not self.dry_run and self._maintenance is not None:
            try:
                self._maintenance()
            except Exception:
                # Maintenance must never prevent scheduled business workflows
                # from being evaluated. The maintenance owner records its own
                # redacted status where appropriate.
                pass
        due: list[tuple[str, datetime, date]] = []
        state_changed = False
        for workflow in self.catalog.scheduled_workflows():
            schedule = self.catalog.schedule_for(workflow.workflow_id)
            if not schedule.enabled:
                continue
            slot = latest_due_slot(schedule, current)
            age = (current - slot).total_seconds()
            if age < 0 or age > self.misfire_grace_seconds:
                continue
            previous = self._state["jobs"].get(workflow.workflow_id, {})
            if previous.get("status") == "running":
                pid = int(previous.get("pid") or 0)
                if pid and _pid_exists(pid):
                    continue
                previous.update({
                    "status": "interrupted",
                    "finished_at": current.astimezone(timezone.utc).isoformat(timespec="seconds"),
                })
                state_changed = True
            if previous.get("slot") == slot.isoformat() or workflow.workflow_id in self._active:
                continue
            business_date = slot.date() + timedelta(
                days=schedule.business_date_offset_days
            )
            due.append((workflow.workflow_id, slot, business_date))

        if state_changed and not self.dry_run:
            self._save_state(current)

        if self.dry_run:
            return [workflow_id for workflow_id, _slot, _date in due]

        started: list[str] = []
        for workflow_id, slot, business_date in due:
            started_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
            try:
                process = self.launcher.launch(workflow_id, slot, business_date)
            except Exception as exc:
                self._state["jobs"][workflow_id] = {
                    "slot": slot.isoformat(), "status": "launch_failed",
                    "business_date": business_date.isoformat(),
                    "started_at": started_at, "error": type(exc).__name__,
                }
                self._save_state(current)
                continue
            self._active[workflow_id] = _ActiveRun(process, slot)
            self._state["jobs"][workflow_id] = {
                "slot": slot.isoformat(), "status": "running", "pid": process.pid,
                "business_date": business_date.isoformat(),
                "started_at": started_at,
            }
            self._save_state(current)
            started.append(workflow_id)
        return started

    def wait_for_active(self, timeout_seconds: float | None = None, poll_seconds: float = 0.2) -> bool:
        deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds
        while self._active:
            self._reap(datetime.now(self.timezone))
            if not self._active:
                return True
            if deadline is not None and time.monotonic() >= deadline:
                return False
            time.sleep(poll_seconds)
        return True

    def status(self) -> dict:
        self._reap(datetime.now(self.timezone))
        return json.loads(json.dumps(self._state))


class SchedulerInstanceLock:
    """Single-instance PID lock with stale-owner recovery."""

    def __init__(self, path: Path) -> None:
        self.path = Path(path).resolve()
        self._owned = False

    def __enter__(self) -> "SchedulerInstanceLock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        for _ in range(2):
            try:
                descriptor = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            except FileExistsError:
                try:
                    pid = int(json.loads(self.path.read_text(encoding="utf-8")).get("pid", 0))
                except Exception:
                    pid = 0
                if pid and _pid_exists(pid):
                    raise RuntimeError(f"scheduler is already running (pid={pid})")
                self.path.unlink(missing_ok=True)
                continue
            with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
                json.dump({"pid": os.getpid(), "started_at": datetime.now(timezone.utc).isoformat(timespec="seconds")}, stream)
            self._owned = True
            return self
        raise RuntimeError("cannot acquire scheduler instance lock")

    def __exit__(self, exc_type, exc, traceback) -> None:  # type: ignore[no-untyped-def]
        if self._owned:
            self.path.unlink(missing_ok=True)
            self._owned = False


def _pid_exists(pid: int) -> bool:
    if pid <= 0:
        return False
    if os.name == "nt":
        # Windows 的 os.kill(pid, 0) 并非可靠的只读探测；使用查询句柄，
        # 避免向被检查进程发送控制信号。
        import ctypes

        process_query_limited_information = 0x1000
        handle = ctypes.windll.kernel32.OpenProcess(  # type: ignore[attr-defined]
            process_query_limited_information, False, pid
        )
        if not handle:
            return False
        ctypes.windll.kernel32.CloseHandle(handle)  # type: ignore[attr-defined]
        return True
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError:
        return False
    return True