wecker/api/schema.py

import strawberry
from typing import List, Optional
import os
import shlex
import sys
import time
import subprocess
import signal
from pathlib import Path
from api.crontab_manager import CrontabManager
from common import PID_FILE
from styles import get_style


def get_manager():
    tabfile = os.getenv("TABFILE")
    return CrontabManager(tabfile=tabfile)


def is_wecker_ringing() -> bool:
    """Check if the wecker process is currently running via the PID file."""
    if not os.path.exists(PID_FILE):
        return False
    try:
        with open(PID_FILE) as f:
            pid = int(f.read().strip())
        os.kill(pid, 0)
        # Guard against PID reuse: verify the running process is actually wecker.py.
        cmdline_path = f"/proc/{pid}/cmdline"
        with open(cmdline_path, "rb") as f:
            cmdline = f.read().replace(b"\0", b" ").decode()
        return "wecker.py" in cmdline
    except (ValueError, ProcessLookupError, PermissionError, OSError, FileNotFoundError):
        return False


def _project_root() -> Path:
    """Locate the project root by searching upward for pyproject.toml."""
    start = Path(__file__).resolve()
    for parent in start.parents:
        if (parent / "pyproject.toml").exists():
            return parent
    # Fallback for unusual layouts; keep a sensible default.
    return start.parent.parent


def _default_command(style: str) -> str:
    """Return the default shell command to run wecker.py from crontab."""
    project_root = _project_root()
    python_exec = sys.executable
    runtime_dir = f"/run/user/{os.getuid()}"
    return (
        f"cd {project_root} && "
        f"XDG_RUNTIME_DIR={runtime_dir} SDL_AUDIODRIVER=pulse "
        f"{python_exec} wecker.py --style {shlex.quote(style)} >> wecker.log 2>&1"
    )


def _start_wecker_process(style: str) -> subprocess.Popen:
    """Start wecker.py without invoking a shell."""
    project_root = _project_root()
    env = os.environ.copy()
    env.setdefault("SDL_AUDIODRIVER", "pulse")
    env.setdefault("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
    return subprocess.Popen(
        [sys.executable, str(project_root / "wecker.py"), "--style", style],
        cwd=project_root,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
        env=env,
    )


@strawberry.type
class Alarm:
    id: str
    cron_expression: str
    is_enabled: bool
    style: str


def _alarm_from_dict(data: dict) -> Alarm:
    """Map a CrontabManager alarm dict to the public GraphQL Alarm type."""
    return Alarm(
        id=data["id"],
        cron_expression=data["cron_expression"],
        is_enabled=data["is_enabled"],
        style=data["style"],
    )


def _load_alarms() -> List[Alarm]:
    """Load all alarms from the crontab and map them to GraphQL types."""
    manager = get_manager()
    return [_alarm_from_dict(a) for a in manager.get_alarms()]


@strawberry.type
class Query:
    @strawberry.field
    def is_ringing(self) -> bool:
        return is_wecker_ringing()

    @strawberry.field
    def get_alarms(self) -> List[Alarm]:
        return _load_alarms()

    @strawberry.field
    def get_alarm(self, id: str) -> Optional[Alarm]:
        for alarm in _load_alarms():
            if alarm.id == id:
                return alarm
        return None


@strawberry.type
class Mutation:
    @strawberry.field
    def set_alarm(
        self,
        cron_expression: str,
        is_enabled: bool = True,
        id: Optional[str] = None,
        style: str = "simple",
    ) -> Alarm:
        get_style(style)  # validate; raises ValueError on unknown style
        command = _default_command(style)

        manager = get_manager()
        new_id = manager.set_alarm(
            alarm_id=id,
            cron_expression=cron_expression,
            command=command,
            is_enabled=is_enabled,
        )
        return Alarm(
            id=new_id,
            cron_expression=cron_expression,
            is_enabled=is_enabled,
            style=style,
        )

    @strawberry.field
    def delete_alarm(self, id: str) -> bool:
        manager = get_manager()
        # Verify it exists
        alarms = manager.get_alarms()
        exists = any(a["id"] == id for a in alarms)
        if exists:
            manager.delete_alarm(id)
            return True
        return False

    @strawberry.field
    def start_ringing(self, style: str = "simple") -> bool:
        """Start the wecker alarm if it is not already ringing.
        Returns True if started, False if already ringing."""
        if is_wecker_ringing():
            return False
        get_style(style)  # validate; raises ValueError on unknown style
        _start_wecker_process(style)
        return True

    @strawberry.field
    def stop_ringing(self) -> bool:
        """Stop the wecker alarm if it is currently ringing.
        Returns True if stopped, False if not ringing."""
        if not is_wecker_ringing():
            return False
        try:
            with open(PID_FILE) as f:
                pid = int(f.read().strip())
            os.kill(pid, signal.SIGTERM)
            # Give the process a short grace period, then escalate to SIGKILL.
            for _ in range(20):
                time.sleep(0.1)
                try:
                    os.kill(pid, 0)
                except ProcessLookupError:
                    break
            else:
                os.kill(pid, signal.SIGKILL)
            if os.path.exists(PID_FILE):
                os.remove(PID_FILE)
            return True
        except (ValueError, ProcessLookupError, PermissionError, OSError):
            # PID file is stale or process already dead
            if os.path.exists(PID_FILE):
                os.remove(PID_FILE)
            return False


schema = strawberry.Schema(query=Query, mutation=Mutation)