"""Private same-UID Unix RPC: one outstanding request per tool_call, no polling."""
from __future__ import annotations

import asyncio
import contextlib
import json
import os
import socket
import struct
from pathlib import Path

from pydantic import BaseModel, Field, ValidationError

from .models import ControlError


def approval_socket_path():
    directory = Path(os.environ.get("DEVICE_LAB_CONTROL_DIR", "~/.config/esp32-device-gateway/runtime")).expanduser()
    return directory / "approvals.sock"


class PiApprovalRequest(BaseModel):
    session_id: str = Field(min_length=1, max_length=160)
    runtime_id: str = Field(min_length=1, max_length=80)
    tool_call_id: str = Field(min_length=1, max_length=200)
    cwd: str = Field(min_length=1, max_length=1024)
    kind: str = Field(pattern="^(command|file_change|tool|action)$")
    title: str = Field(min_length=1, max_length=160)
    detail: str = Field(max_length=4000)


class PiApprovalServer:
    def __init__(self, registry, projects, path=None):
        self.registry, self.projects = registry, projects
        self.path = Path(path) if path else approval_socket_path()
        self.server = None
        self.clients = set()
        self.tasks = set()

    async def start(self):
        self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        if self.path.parent.stat().st_uid != os.getuid() or self.path.parent.stat().st_mode & 0o077:
            raise RuntimeError("Pi approval socket directory must be private to this user (0700)")
        if self.path.exists():
            try:
                _, writer = await asyncio.wait_for(asyncio.open_unix_connection(self.path), 1)
            except (ConnectionRefusedError, FileNotFoundError):
                self.path.unlink(missing_ok=True)
            else:
                writer.close()
                await writer.wait_closed()
                raise RuntimeError("Pi approval endpoint already has a Gateway owner")
        self.server = await asyncio.start_unix_server(self.handle, self.path, limit=16384)
        self.path.chmod(0o600)

    async def handle(self, reader, writer):
        task = asyncio.current_task()
        self.tasks.add(task)
        self.clients.add(writer)
        approval = None
        eof = decision = None
        try:
            peer = writer.get_extra_info("socket").getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
            if struct.unpack("3i", peer)[1] != os.getuid():
                return
            line = await asyncio.wait_for(reader.readline(), 5)
            if not line:
                return
            value = PiApprovalRequest.model_validate_json(line)
            project = next((p for p in self.projects.list() if p.path == str(Path(value.cwd).resolve())), None)
            if project is None:
                raise ControlError("approval project is not registered", 403)
            approval = self.registry.create(agent="pi", session_id=value.session_id, project_id=project.id,
                kind=value.kind, title=value.title, detail=value.detail, owner="pi:"+value.runtime_id,
                request_identity=json.dumps([value.session_id, value.tool_call_id]))
            writer.write((json.dumps({"approval_id": approval.approval_id, "state": "PENDING"})+"\n").encode())
            await writer.drain()
            decision = asyncio.create_task(self.registry.wait(approval))
            eof = asyncio.create_task(reader.read(1))
            done, _ = await asyncio.wait([decision, eof], return_when=asyncio.FIRST_COMPLETED)
            if eof in done:
                self.registry.cancel(approval.approval_id)
                return
            state = decision.result()
            writer.write((json.dumps({"approval_id": approval.approval_id,
                "decision": "approve" if state == "APPROVED" else "reject" if state == "REJECTED" else "cancel", "state": state})+"\n").encode())
            await writer.drain()
        except (ValueError, ValidationError, ControlError, OSError, TimeoutError):
            with contextlib.suppress(OSError):
                writer.write(b'{"decision":"cancel","state":"CANCELLED"}\n')
                await writer.drain()
        finally:
            if approval:
                self.registry.cancel(approval.approval_id)
            for pending in (eof, decision):
                if pending:
                    pending.cancel()
                    with contextlib.suppress(asyncio.CancelledError):
                        await pending
            writer.close()
            with contextlib.suppress(OSError):
                await writer.wait_closed()
            self.clients.discard(writer)
            self.tasks.discard(task)

    async def close(self):
        if self.server:
            self.server.close()
            await self.server.wait_closed()
            for writer in list(self.clients):
                writer.close()
            for task in list(self.tasks):
                task.cancel()
            await asyncio.gather(*list(self.tasks), return_exceptions=True)
            self.path.unlink(missing_ok=True)