#!/usr/bin/env python3
#
# Copyright (c) 2026 Stormux
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.

"""Wayland workspace backends for normalized compositor state tracking."""

from __future__ import annotations

import os
from collections.abc import Mapping, Sequence
from typing import Any

from gi.repository import GLib

from . import debug
from .compositor_state_types import (
    DESKTOP_TRANSITION_FINISHED,
    DESKTOP_TRANSITION_STARTED,
    WORKSPACE_STATE_CHANGED,
)
from .wayland_protocols import ext_workspace_v1
from .wnck_support import get_session_type


class NullWorkspaceBackend:
    """No-op backend used when no real workspace backend is available."""

    name = "null"

    def __init__(self) -> None:
        self._emitSignal = None

    def is_available(self, session_type: str | None = None) -> bool:
        return True

    def activate(self, emit_signal: Any = None) -> None:
        self._emitSignal = emit_signal

    def deactivate(self, emit_signal: Any = None) -> None:
        self._emitSignal = None


class WaylandSharedProtocolsBackend:
    """Runtime-optional backend for the ext-workspace shared Wayland protocol."""

    name = "wayland-shared-protocols"

    def __init__(
        self,
        *,
        environment: Mapping[str, str] | None = None,
        protocols: Any = None,
    ) -> None:
        self._environment = environment if environment is not None else os.environ
        self._protocols = protocols or ext_workspace_v1
        self._emitSignal = None
        self._display = None
        self._registry = None
        self._workspaceManager = None
        self._workspaceStates: dict[Any, dict[str, Any]] = {}
        self._batchDirty = False
        self._transitionPending = False
        self._dispatchSourceId = 0

    def is_available(self, session_type: str | None = None) -> bool:
        effectiveSessionType = (session_type or get_session_type()).strip().lower()
        return (
            effectiveSessionType == "wayland"
            and bool((self._environment.get("WAYLAND_DISPLAY") or "").strip())
            and bool(self._protocols.has_runtime_support())
        )

    def activate(self, emit_signal: Any = None) -> None:
        self.deactivate()
        self._emitSignal = emit_signal

        if emit_signal is None or not self.is_available():
            return

        displayClass = self._protocols.get_display_class()
        if displayClass is None:
            return

        try:
            self._display = displayClass()
            connect = getattr(self._display, "connect", None)
            if callable(connect):
                connect()
            get_registry = getattr(self._display, "get_registry", None)
            if not callable(get_registry):
                raise RuntimeError("Wayland display missing registry access")
            self._registry = get_registry()
            self._bind_listener(self._registry, "global", self._handle_registry_global)
            self._bind_listener(self._registry, "global_remove", self._handle_registry_global_remove)
            self._roundtrip()
            self._install_dispatch_watch()
        except Exception as error:
            msg = f"COMPOSITOR STATE: Wayland backend activation failed: {error}"
            debug.printMessage(debug.LEVEL_WARNING, msg, True)
            self.deactivate()

    def deactivate(self, emit_signal: Any = None) -> None:
        self._workspaceStates = {}
        self._batchDirty = False
        self._transitionPending = False
        self._remove_dispatch_watch()
        self._safe_close_proxy(self._workspaceManager)
        self._safe_close_proxy(self._registry)
        self._workspaceManager = None
        self._registry = None

        if self._display is not None:
            disconnect = getattr(self._display, "disconnect", None)
            if callable(disconnect):
                try:
                    disconnect()
                except Exception:
                    pass
        self._display = None
        self._emitSignal = None

    def _bind_listener(self, target: Any, event_name: str, callback: Any) -> bool:
        dispatcher = getattr(target, "dispatcher", None)
        if dispatcher is not None:
            try:
                dispatcher[event_name] = callback
                return True
            except Exception:
                pass

        add_listener = getattr(target, "add_listener", None)
        if callable(add_listener):
            try:
                add_listener(**{event_name: callback})
                return True
            except TypeError:
                try:
                    add_listener(event_name, callback)
                    return True
                except TypeError:
                    pass

        attributeName = f"on_{event_name}"
        if hasattr(target, attributeName):
            setattr(target, attributeName, callback)
            return True

        return False

    def _roundtrip(self) -> None:
        if self._display is None:
            return

        for methodName in ("roundtrip", "dispatch", "dispatch_pending"):
            method = getattr(self._display, methodName, None)
            if callable(method):
                method()
                break

    def _install_dispatch_watch(self) -> None:
        if self._display is None or self._dispatchSourceId:
            return

        getFd = getattr(self._display, "get_fd", None)
        if not callable(getFd):
            return

        try:
            displayFd = int(getFd())
        except Exception:
            return

        if displayFd < 0:
            return

        self._dispatchSourceId = GLib.io_add_watch(
            displayFd,
            GLib.PRIORITY_DEFAULT,
            GLib.IO_IN | GLib.IO_ERR | GLib.IO_HUP,
            self._dispatch_display_events,
        )

    def _remove_dispatch_watch(self) -> None:
        if not self._dispatchSourceId:
            return

        try:
            GLib.source_remove(self._dispatchSourceId)
        except Exception:
            pass

        self._dispatchSourceId = 0

    def _dispatch_display_events(self, _fd: int, condition: Any) -> bool:
        if condition & (GLib.IO_ERR | GLib.IO_HUP):
            self.deactivate()
            return False

        dispatch = getattr(self._display, "dispatch", None)
        if not callable(dispatch):
            self.deactivate()
            return False

        try:
            while True:
                dispatched = dispatch(block=False)
                if not dispatched:
                    break
        except TypeError:
            try:
                dispatch()
            except Exception:
                self.deactivate()
                return False
        except Exception:
            self.deactivate()
            return False

        return True

    def _safe_close_proxy(self, proxy: Any) -> None:
        if proxy is None:
            return

        for methodName in ("destroy", "release"):
            method = getattr(proxy, methodName, None)
            if callable(method):
                try:
                    method()
                except Exception:
                    pass
                return

    def _handle_registry_global(self, *args: Any) -> None:
        if self._workspaceManager is not None:
            return

        globalName, interfaceName, interfaceVersion = self._parse_registry_global_args(args)
        if interfaceName != self._protocols.INTERFACE_NAME:
            return

        registry = self._registry
        if registry is None:
            return

        bindVersion = min(interfaceVersion, self._protocols.INTERFACE_VERSION)
        try:
            self._workspaceManager = self._protocols.bind_workspace_manager(
                registry,
                globalName,
                bindVersion,
            )
        except Exception:
            self._workspaceManager = None

        if self._workspaceManager is None:
            return

        self._bind_listener(self._workspaceManager, "workspace", self._handle_workspace_manager_workspace)
        self._bind_listener(self._workspaceManager, "done", self._handle_workspace_manager_done)
        self._bind_listener(self._workspaceManager, "finished", self._handle_workspace_manager_finished)

    def _handle_registry_global_remove(self, *args: Any) -> None:
        return

    def _handle_workspace_manager_workspace(self, *args: Any) -> None:
        workspaceHandle = args[-1] if args else None
        if workspaceHandle is None or workspaceHandle in self._workspaceStates:
            return

        self._workspaceStates[workspaceHandle] = {"active": False}
        self._bind_listener(workspaceHandle, "id", self._handle_workspace_id)
        self._bind_listener(workspaceHandle, "state", self._handle_workspace_state)
        self._bind_listener(workspaceHandle, "removed", self._handle_workspace_removed)

    def _handle_workspace_id(self, workspaceHandle: Any, workspaceId: str, *_args: Any) -> None:
        workspaceState = self._workspaceStates.setdefault(workspaceHandle, {"active": False})
        workspaceState["workspace_id"] = (workspaceId or "").strip()
        self._batchDirty = True

    def _handle_workspace_state(self, workspaceHandle: Any, stateValue: Any, value: Any = None, *_args: Any) -> None:
        workspaceState = self._workspaceStates.setdefault(workspaceHandle, {"active": False})
        workspaceState["active"] = self._workspace_is_active(stateValue, value)
        self._batchDirty = True
        self._emit_transition_started("workspace-state-update")

    def _handle_workspace_removed(self, workspaceHandle: Any, *_args: Any) -> None:
        self._workspaceStates.pop(workspaceHandle, None)
        self._batchDirty = True
        self._emit_transition_started("workspace-removed")

    def _handle_workspace_manager_done(self, *_args: Any) -> None:
        if not self._batchDirty:
            return

        activeWorkspaceIds = self._active_workspace_ids()
        self._emit_signal(WORKSPACE_STATE_CHANGED, activeWorkspaceIds, "workspace-batch-done")
        self._emit_signal(DESKTOP_TRANSITION_FINISHED, activeWorkspaceIds, "workspace-batch-done")
        self._batchDirty = False
        self._transitionPending = False

    def _handle_workspace_manager_finished(self, *_args: Any) -> None:
        self._workspaceManager = None

    def _workspace_is_active(self, stateValue: Any, value: Any = None) -> bool:
        if isinstance(stateValue, str):
            return stateValue.strip().lower() == "active" and bool(value)

        if isinstance(stateValue, Mapping):
            return bool(stateValue.get("active"))

        if isinstance(stateValue, Sequence) and not isinstance(stateValue, (str, bytes, bytearray)):
            return self._protocols.ACTIVE_STATE_VALUE in stateValue

        return bool(stateValue)

    def _active_workspace_ids(self) -> set[str]:
        workspaceIds = set()
        for workspaceHandle, workspaceState in self._workspaceStates.items():
            if not workspaceState.get("active"):
                continue
            workspaceId = (workspaceState.get("workspace_id") or "").strip()
            if not workspaceId:
                workspaceId = self._workspace_id(workspaceHandle)
            workspaceIds.add(workspaceId)
        return workspaceIds

    def _workspace_id(self, workspaceHandle: Any) -> str:
        return f"workspace:{id(workspaceHandle)}"

    def _emit_transition_started(self, reason: str) -> None:
        if self._transitionPending:
            return
        self._transitionPending = True
        self._emit_signal(DESKTOP_TRANSITION_STARTED, self._active_workspace_ids(), reason)

    def _emit_signal(self, signalType: str, workspaceIds: set[str], reason: str) -> None:
        if callable(self._emitSignal):
            self._emitSignal(signalType, workspaceIds, reason)
            return

    def _parse_registry_global_args(self, args: Sequence[Any]) -> tuple[int, str, int]:
        if len(args) >= 4:
            _, globalName, interfaceName, interfaceVersion = args[-4:]
        elif len(args) >= 3:
            globalName, interfaceName, interfaceVersion = args[-3:]
        else:
            return (-1, "", 0)

        try:
            parsedName = int(globalName)
        except (TypeError, ValueError):
            parsedName = -1

        try:
            parsedVersion = int(interfaceVersion)
        except (TypeError, ValueError):
            parsedVersion = self._protocols.INTERFACE_VERSION

        return parsedName, str(interfaceName or ""), parsedVersion
