"""The daemon: device list, session, and the JSON control socket the panel talks to.

Protocol (newline-delimited JSON):
  in   {"cmd": "cast", "device": "cast:<uuid>"}          … see handle()
  out  {"type": "state", …}                              pushed on every change
"""
from __future__ import annotations

import asyncio
import contextlib
import fcntl
import json
import os
import shlex
import shutil
import signal
import time

from . import VERSION
from .backends import AirPlayBackend, for_kind
from .backends.base import BackendError
from .capture import PortalError, drop_legacy_token
from .config import PIPELINE_KEYS, Config
from .devices import Device
from .discovery import Discovery
from .session import Session
from .stream import StreamServer
from .util import (firewall_argv, firewall_command, firewall_tool, log, runtime_dir,
                   terminal_argv)


NO_AGENT = "no polkit agent"   # pkexec had nowhere to ask; fall back to a terminal


class Daemon:
    def __init__(self) -> None:
        self.cfg = Config()
        self.discovery = Discovery(on_change=self.push_state)
        self.stream = StreamServer(on_client=self._on_client)
        self.session = Session(self)
        self.backends: dict[str, object] = {}
        self.clients: set[asyncio.StreamWriter] = set()
        self.error = ""
        self.busy = ""
        self.pairing = ""          # device id a pairing PIN is expected for
        self.firewall_blocked = False   # a cast failed with nothing on the stream
        self._push_handle: asyncio.TimerHandle | None = None
        self._cast_task: asyncio.Task | None = None
        self._rescan_task: asyncio.Task | None = None
        self._stopping = False

    # ---- backends ----
    def backend_for(self, device: Device):
        cls = for_kind(device.kind)
        if cls is None:
            raise BackendError(f"no backend for {device.kind}")
        backend = self.backends.get(device.kind)
        if backend is None:
            backend = cls(self)
            self.backends[device.kind] = backend
        return backend

    def note_backend_error(self, message: str) -> None:
        self.error = message
        self.push_state()

    # ---- state ----
    def build_state(self) -> dict:
        devices = sorted(
            self.discovery.devices.values(), key=lambda d: (d.kind != "cast", d.name.lower())
        )
        return {
            "type": "state",
            "version": VERSION,
            "devices": [d.to_json() for d in devices],
            "session": self.session.to_json(),
            "settings": dict(self.cfg.values),
            "error": self.error,
            "busy": self.busy,
            "pairing": self.pairing,
            "capabilities": {
                "airplay": AirPlayBackend.available(),
                "port": self.stream.port or int(self.cfg["port"]),
            },
            "firewall": self.firewall_state(),
            "time": time.time(),
        }

    def firewall_state(self) -> dict:
        """What the panel needs to offer "open the port" after a failed cast."""
        tool = firewall_tool()
        host = self.session.device.host if self.session.device else ""
        port = self.stream.port or int(self.cfg["port"])
        return {
            "tool": tool,
            # "blocked": a cast just failed with nothing on the stream.
            # "verified": some receiver has fetched it at least once on this
            # machine, which is the only proof the port is really open.
            "blocked": bool(tool) and self.firewall_blocked,
            "verified": bool(self.cfg["portVerified"]),
            "port": port,
            "command": firewall_command(tool, port, host) if tool else "",
        }

    async def open_firewall(self, host: str = "") -> dict:
        """Let receivers on this network reach the stream port, once.

        Casting is pull-based - the receiver opens the connection to us - so a
        default-deny firewall has to be told about the port. Installing does not
        need root, so this is where it is asked for: one password dialog before
        the first cast, and the rule persists.
        """
        tool = firewall_tool(refresh=True)
        if not tool:
            self.firewall_blocked = False
            self.cfg.update({"portVerified": True})
            self.push_state()
            return {"ok": True, "note": "no firewall is running"}
        if not host:
            host = self.session.device.host if self.session.device else ""
        port = self.stream.port or int(self.cfg["port"])
        argv_list = firewall_argv(tool, port, host)
        self.busy = f"Opening port {port}…"
        self.push_state(now=True)
        try:
            ok, err = await self._pkexec(argv_list)
            if not ok and err == NO_AGENT:
                ok, err = await self._sudo_in_terminal(argv_list, port)
        finally:
            self.busy = ""
            self.push_state()
        if not ok:
            log.warning("open firewall: %s", err)
            # The panel only sees state, so a cancelled or ignored prompt has to
            # land there too - otherwise the button just looks broken.
            self.error = (
                f"the port was not opened: {err}" if err != NO_AGENT
                else "no way to ask for a password on this machine"
            )
            self.push_state(now=True)
            return {"ok": False, "error": err}
        log.info("opened port %d in %s", port, tool)
        self.firewall_blocked = False
        self.error = ""
        # The rule is in: stop warning about a port that is now open. A receiver
        # actually reaching us keeps it that way (see _on_client).
        self.cfg.update({"portVerified": True})
        self.push_state(now=True)
        return {"ok": True, "note": f"port {port} is open"}

    async def _pkexec(self, argv_list: list[list[str]]) -> tuple[bool, str]:
        if shutil.which("pkexec") is None:
            return False, NO_AGENT
        for argv in argv_list:
            exe = shutil.which(argv[0])
            if exe is None:
                return False, f"{argv[0]} is not installed"
            proc = await asyncio.create_subprocess_exec(
                "pkexec", exe, *argv[1:],
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT,
            )
            try:
                out, _ = await asyncio.wait_for(proc.communicate(), 180)
            except asyncio.TimeoutError:
                proc.kill()
                return False, "the password prompt timed out"
            text = out.decode(errors="replace").strip()
            # 126 = dismissed or refused, 127 = pkexec had no agent to ask.
            if proc.returncode == 127 or "authentication agent" in text.lower():
                return False, NO_AGENT
            if proc.returncode == 126:
                return False, "cancelled"
            if proc.returncode != 0:
                return False, text.splitlines()[-1] if text else f"failed ({proc.returncode})"
        return True, ""

    async def _sudo_in_terminal(self, argv_list: list[list[str]], port: int) -> tuple[bool, str]:
        """No polkit agent: run sudo in a terminal window and watch for the marker."""
        mark = runtime_dir() / "firewall-ok"
        with contextlib.suppress(FileNotFoundError):
            mark.unlink()
        script = " && ".join(shlex.join(a) for a in argv_list) + f" && : > {shlex.quote(str(mark))}"
        inner = (
            f"echo 'screencast needs port {port} open so your TV can fetch the stream.'; "
            f"echo; sudo sh -c {shlex.quote(script)} "
            "|| { echo; echo 'not opened'; read -r _; }"
        )
        argv = terminal_argv(["sh", "-c", inner])
        if argv is None:
            return False, "no terminal to ask for a password in; run: " + " && ".join(
                "sudo " + shlex.join(a) for a in argv_list)
        proc = await asyncio.create_subprocess_exec(
            *argv, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL
        )
        try:
            await asyncio.wait_for(proc.wait(), 300)
        except asyncio.TimeoutError:
            proc.kill()
        # The terminal may fork, so trust the marker rather than the exit code.
        for _ in range(20):
            if mark.exists():
                mark.unlink()
                return True, ""
            await asyncio.sleep(0.5)
        return False, "the port was not opened"

    def push_state(self, now: bool = False) -> None:
        """Coalesce the bursts discovery and polling produce into one push.

        `now` skips the coalescing: anything a click just changed is feedback,
        and 80ms of it queued behind the work is 80ms of a dead-looking panel.
        """
        if now:
            if self._push_handle is not None:
                self._push_handle.cancel()
                self._push_handle = None
            self._push_now()
            return
        if self._push_handle is not None:
            return
        loop = asyncio.get_event_loop()
        self._push_handle = loop.call_later(0.08, self._push_now)

    def _push_now(self) -> None:
        self._push_handle = None
        if not self.clients:
            return
        line = (json.dumps(self.build_state()) + "\n").encode()
        for writer in list(self.clients):
            try:
                writer.write(line)
            except Exception:  # noqa: BLE001
                self.clients.discard(writer)

    def _on_client(self, peer: str, connected: bool) -> None:
        if connected:
            self.firewall_blocked = False
            if not self.cfg["portVerified"]:
                self.cfg.update({"portVerified": True})
        self.session.note_client(peer, connected)

    # ---- commands ----
    def find_device(self, key: str) -> Device:
        key = (key or "").strip()
        devices = self.discovery.devices
        if key in devices:
            return devices[key]
        lowered = key.lower()
        for dev in devices.values():
            if dev.name.lower() == lowered or dev.id.lower() == lowered:
                return dev
        for dev in devices.values():
            if lowered and lowered in dev.name.lower():
                return dev
        raise BackendError(f"no device matching “{key}”")

    async def handle(self, msg: dict) -> dict:
        cmd = str(msg.get("cmd", ""))
        if cmd == "get":
            return {"ok": True}
        if cmd == "rescan":
            if self._rescan_task is None or self._rescan_task.done():
                self._rescan_task = asyncio.create_task(self._run_rescan())
            return {"ok": True, "started": True}
        if cmd == "cast":
            device = self.find_device(str(msg.get("device", "")))
            self.error = ""
            self.busy = f"Connecting to {device.name}…"
            self.cfg.update({"lastDevice": device.id})
            self.push_state(now=True)
            # A firewall we have never been through is a cast that is going to
            # fail in forty seconds for a reason we already know. Ask for the
            # port up front instead - one dialog, once, and then it is done.
            if firewall_tool() and not self.cfg["portVerified"]:
                await self.open_firewall(device.host)
                self.busy = f"Connecting to {device.name}…"
                self.error = ""
                self.push_state(now=True)
            # Starting can sit for a minute on the portal's screen picker, so it
            # runs on its own task: the panel has to stay able to send "stop".
            await self._cancel_cast()
            self._cast_task = asyncio.create_task(
                self._run_cast(device, self.backend_for(device))
            )
            return {"ok": True, "started": True}
        if cmd == "openFirewall":
            return await self.open_firewall(str(msg.get("host", "")))
        if cmd == "stop":
            self.busy = "Stopping…"
            self.push_state(now=True)
            try:
                await self._cancel_cast()
                await self.session.stop()
            finally:
                self.busy = ""
                self.push_state(now=True)
            return {"ok": True}
        if cmd == "set":
            changed = self.cfg.update(msg.get("settings") or {})
            if changed & {"port"} and self.session.active:
                await self.session.stop()
            elif changed & PIPELINE_KEYS and self.session.active:
                # Quality settings only exist in the running pipeline: restart it.
                device, backend = self.session.device, self.session.backend
                if device is not None and backend is not None:
                    await self.session.start(device, backend)
            self.push_state()
            return {"ok": True, "changed": sorted(changed)}
        if cmd in ("volume", "mute", "transport"):
            device = (
                self.find_device(str(msg["device"])) if msg.get("device") else self.session.device
            )
            if device is None:
                raise BackendError("nothing is being cast")
            backend = self.backend_for(device)
            if cmd == "volume":
                await backend.set_volume(device, float(msg.get("value", 0)))
                device.volume = float(msg.get("value", 0))
            elif cmd == "mute":
                await backend.set_muted(device, bool(msg.get("value")))
                device.muted = bool(msg.get("value"))
            else:
                await backend.transport(device, str(msg.get("action", "")))
            self.push_state()
            return {"ok": True}
        if cmd == "repick":
            await self.session.repick_source()
            self.push_state()
            return {"ok": True}
        if cmd == "pair":
            device = self.find_device(str(msg.get("device", "")))
            backend = self.backend_for(device)
            if not hasattr(backend, "pair_begin"):
                raise BackendError(f"{device.name} does not need pairing")
            self.busy = f"Asking {device.name} for a code…"
            self.push_state(now=True)
            try:
                await backend.pair_begin(device)
            finally:
                self.busy = ""
            self.pairing = device.id
            self.push_state()
            return {"ok": True}
        if cmd == "pairPin":
            if not self.pairing:
                raise BackendError("no pairing in progress")
            device = self.find_device(self.pairing)
            backend = self.backend_for(device)
            self.busy = f"Pairing with {device.name}…"
            self.push_state(now=True)
            try:
                ok = await backend.pair_pin(str(msg.get("pin", "")))
            finally:
                self.busy = ""
            self.pairing = ""
            self.error = "" if ok else f"{device.name} rejected that code"
            self.push_state()
            return {"ok": ok, "paired": ok}
        if cmd == "pairCancel":
            for backend in self.backends.values():
                if hasattr(backend, "pair_cancel"):
                    await backend.pair_cancel()
            self.pairing = ""
            self.push_state()
            return {"ok": True}
        if cmd == "quit":
            asyncio.create_task(self.shutdown())
            return {"ok": True}
        raise BackendError(f"unknown command “{cmd}”")

    async def _run_rescan(self) -> None:
        self.busy = "Looking for displays…"
        self.push_state(now=True)
        try:
            await self.discovery.rescan()
        finally:
            self.busy = ""
            self.push_state(now=True)

    async def _run_cast(self, device: Device, backend) -> None:
        try:
            await self.session.start(device, backend)
        except asyncio.CancelledError:
            await self.session.stop(quiet=True)
            raise
        finally:
            self.busy = ""
            self.push_state()

    async def _cancel_cast(self) -> None:
        task, self._cast_task = self._cast_task, None
        if task is not None and not task.done():
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await task

    # ---- control socket ----
    async def _serve_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        self.clients.add(writer)
        pending: set[asyncio.Task] = set()
        try:
            writer.write((json.dumps(self.build_state()) + "\n").encode())
            while True:
                line = await reader.readline()
                if not line:
                    break
                try:
                    msg = json.loads(line)
                except json.JSONDecodeError:
                    continue
                # Each command runs on its own task. Reading the next line must
                # not wait on this one: a rescan takes six seconds and a cast can
                # sit on the screen picker for a minute, and a panel whose next
                # click is stuck in the socket looks broken.
                task = asyncio.create_task(self._run_command(msg, writer))
                pending.add(task)
                task.add_done_callback(pending.discard)
        except (ConnectionResetError, BrokenPipeError, asyncio.IncompleteReadError):
            pass   # a CLI client that read one line and walked away
        finally:
            for task in pending:
                task.cancel()
            self.clients.discard(writer)
            with contextlib.suppress(Exception):
                writer.close()

    async def _run_command(self, msg: dict, writer: asyncio.StreamWriter) -> None:
        try:
            reply = await self.handle(msg)
        except asyncio.CancelledError:
            raise
        except (BackendError, PortalError) as exc:
            reply = {"ok": False, "error": str(exc)}
            self.error = str(exc)
        except Exception as exc:  # noqa: BLE001 - one bad command must not kill the daemon
            log.exception("command failed")
            reply = {"ok": False, "error": str(exc)}
            self.error = str(exc)
        reply["type"] = "reply"
        reply["cmd"] = msg.get("cmd", "")
        with contextlib.suppress(Exception):
            writer.write((json.dumps(reply) + "\n").encode())
        self._push_now()

    # ---- lifecycle ----
    async def run(self) -> int:
        lock_path = runtime_dir() / "lock"
        lock = open(lock_path, "w")          # noqa: SIM115 - held for the process lifetime
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            log.error("another screencast-server is already running")
            return 3
        drop_legacy_token()
        lock.write(str(os.getpid()))
        lock.flush()

        sock_path = runtime_dir() / "ctl.sock"
        with contextlib.suppress(FileNotFoundError):
            os.unlink(sock_path)
        # Create it private rather than chmod-ing after the bind: in the /tmp
        # fallback that gap is a window another local user can connect through.
        old_umask = os.umask(0o077)
        try:
            server = await asyncio.start_unix_server(self._serve_client, path=str(sock_path))
        finally:
            os.umask(old_umask)
        os.chmod(sock_path, 0o600)
        log.info("screencast-server %s listening on %s", VERSION, sock_path)

        await self.discovery.start()
        self._done = asyncio.get_running_loop().create_future()
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            loop.add_signal_handler(sig, lambda: asyncio.create_task(self.shutdown()))
        try:
            await self._done
        finally:
            server.close()
            with contextlib.suppress(Exception):
                await server.wait_closed()
            with contextlib.suppress(FileNotFoundError):
                os.unlink(sock_path)
            fcntl.flock(lock, fcntl.LOCK_UN)
            lock.close()
        return 0

    async def shutdown(self) -> None:
        if self._stopping:
            return
        self._stopping = True
        log.info("shutting down")
        # zeroconf and pychromecast both run non-daemon threads that can outlive
        # a clean shutdown; never let one of them keep the socket (and the lock)
        # from being handed to the next instance.
        asyncio.get_running_loop().call_later(5, lambda: os._exit(0))
        with contextlib.suppress(Exception):
            await self._cancel_cast()
        with contextlib.suppress(Exception):
            await self.session.stop(quiet=True)
        for backend in self.backends.values():
            with contextlib.suppress(Exception):
                await backend.close()
        with contextlib.suppress(Exception):
            await self.stream.stop()
        with contextlib.suppress(Exception):
            await self.discovery.stop()
        if not self._done.done():
            self._done.set_result(None)