# systemd_ctypes
#
# Copyright (C) 2022 Allison Karlitskaya <allison.karlitskaya@redhat.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import ctypes
import os
import sys
from typing import ClassVar, Optional, Tuple, Union

from .inotify import inotify_event
from .librarywrapper import (
    Callback,
    Errno,
    Reference,
    ReferenceType,
    UserData,
    WeakReference,
    byref,
)
from .typing import Annotated


class Trampoline(ReferenceType):
    deferred: 'ClassVar[list[Callback] | None]' = None
    trampoline: Callback
    userdata: UserData = None

    def cancel(self) -> None:
        self._unref()
        self.value = None

    def __del__(self) -> None:
        # This might be the currently-dispatching callback — make sure we don't
        # destroy the trampoline before we return.  We drop the deferred list
        # from the event loop when we're sure we're not doing any dispatches.
        if Trampoline.deferred is not None:
            Trampoline.deferred.append(self.trampoline)
        if self.value is not None:
            self._unref()


class sd_bus_error(ctypes.Structure):
    # This is ABI, so we are safe to assume it doesn't change.
    # Unfortunately, we lack anything like sd_bus_error_new().
    _fields_ = (
        ("name", ctypes.c_char_p),
        ("message", ctypes.c_char_p),
        ("_need_free", ctypes.c_int),
    )

    def get(self) -> Tuple[str, str]:
        return self.name.decode(), self.message.decode()

    def set(self, name: str, message: str) -> None:
        result = libsystemd.sd_bus_error_set(byref(self), name, message)
        if result < 0:
            raise OSError(-result, f"sd_bus_error_set: {os.strerror(-result)}")

    def __del__(self) -> None:
        if self._b_needsfree_:
            libsystemd.sd_bus_error_free(byref(self))


class sd_id128(ctypes.Structure):
    # HACK: Pass-by-value of array-containing-structs is broken on Python
    # 3.6. See https://bugs.python.org/issue22273
    _fields_ = (
        [("bytes", ctypes.c_uint8 * 16)]
        if sys.version_info >= (3, 7, 0)
        else [("one", ctypes.c_uint64), ("two", ctypes.c_uint64)]
    )


class sd_event_source(Trampoline):
    ...


class sd_event(ReferenceType):
    def _add_inotify(
        self: 'sd_event',
        source: Reference[sd_event_source],
        path: str,
        event: int,
        callback: Callback,
        user_data: UserData,
    ) -> Union[None, Errno]:
        ...

    def _add_inotify_fd(
        self: 'sd_event',
        source: Reference[sd_event_source],
        fd: int,
        event: int,
        callback: Callback,
        user_data: UserData,
    ) -> Union[None, Errno]:
        ...

    # HACK: sd_event_add_inotify_fd() got added in 250, which is too new.  Fake it.
    def __fallback_add_inotify_fd(
        self: 'sd_event',
        source: Reference[sd_event_source],
        fd: int,
        event: int,
        callback: Callback,
        user_data: UserData,
    ) -> Union[None, Errno]:
        return self._add_inotify(source, f'/proc/self/fd/{fd}', event, callback, user_data)

    def dispatch(self: 'sd_event') -> Union[None, Errno]:
        ...

    def get_fd(self: 'sd_event') -> Union[int, Errno]:
        raise NotImplementedError

    def get_state(self: 'sd_event') -> Union[int, Errno]:
        raise NotImplementedError

    def loop(self: 'sd_event') -> Union[None, Errno]:
        ...

    def prepare(self: 'sd_event') -> Union[None, Errno]:
        ...

    def wait(
        self: 'sd_event', timeout: Annotated[int, ctypes.c_uint64]
    ) -> Union[None, Errno]:
        ...

    @staticmethod
    def _default(ret: Reference['sd_event']) -> Union[None, Errno]:
        ...


class sd_bus_slot(Trampoline):
    ...


class sd_bus_message(ReferenceType):
    def rewind(self: 'sd_bus_message', complete: bool) -> Union[None, Errno]:
        ...

    def _get_error(self: 'sd_bus_message') -> Reference[sd_bus_error]:
        raise NotImplementedError

    def has_signature(self: 'sd_bus_message', signature: str) -> Union[bool, Errno]:
        raise NotImplementedError

    def is_method_error(self: 'sd_bus_message', name: str) -> Union[bool, Errno]:
        raise NotImplementedError

    def _new_method_errnof(
            self: 'sd_bus_message',
            message: Reference['sd_bus_message'],
            error: int,
            format_str: str,
            first_arg: str
    ) -> Union[None, Errno]:
        ...

    def _new_method_errorf(
        self: 'sd_bus_message',
        m: Reference['sd_bus_message'],
        name: str,
        format_str: str,
        first_arg: str
    ) -> Union[None, Errno]:
        ...

    def _new_method_return(
        self: 'sd_bus_message', m: Reference['sd_bus_message']
    ) -> Union[None, Errno]:
        ...

    def seal(
        self: 'sd_bus_message',
        cookie: Annotated[int, ctypes.c_uint64],
        timeout: Annotated[int, ctypes.c_uint64],
    ) -> Union[None, Errno]:
        ...

    def _get_bus(self: 'sd_bus_message') -> WeakReference:
        raise NotImplementedError

    def get_destination(self: 'sd_bus_message') -> str:
        raise NotImplementedError

    def get_interface(self: 'sd_bus_message') -> str:
        raise NotImplementedError

    def get_member(self: 'sd_bus_message') -> str:
        raise NotImplementedError

    def get_path(self: 'sd_bus_message') -> str:
        raise NotImplementedError

    def get_sender(self: 'sd_bus_message') -> Optional[str]:
        raise NotImplementedError

    def get_signature(self: 'sd_bus_message', complete: bool) -> str:
        raise NotImplementedError

    def set_allow_interactive_authorization(self: 'sd_bus_message', b: bool) -> Union[None, Errno]:
        ...


class sd_bus(ReferenceType):
    def _add_match(
        self: 'sd_bus',
        slot: Reference[sd_bus_slot],
        match: str,
        handler: Callback,
        user_data: UserData,
    ) -> Union[None, Errno]:
        ...

    def _add_match_async(
        self: 'sd_bus',
        slot: Reference[sd_bus_slot],
        match: str,
        callback: Callback,
        install_callback: Callback,
        user_data: UserData,
    ) -> Union[None, Errno]:
        ...

    def _add_object(
        self: 'sd_bus',
        slot: Reference[sd_bus_slot],
        path: str,
        callback: Callback,
        user_data: UserData,
    ) -> Union[None, Errno]:
        ...

    def attach_event(
        self: 'sd_bus', event: Optional[sd_event], priority: int
    ) -> Union[None, Errno]:
        ...

    def _call(
        self: 'sd_bus',
        message: sd_bus_message,
        timeout: Annotated[int, ctypes.c_uint64],
        ret_error: Reference[sd_bus_error],
        reply: Reference[sd_bus_message],
    ) -> Union[None, Errno]:
        ...

    def _call_async(
        self: 'sd_bus',
        slot: Reference[sd_bus_slot],
        message: sd_bus_message,
        callback: Callback,
        user_data: UserData,
        timeout_usec: Annotated[int, ctypes.c_uint64],
    ) -> Union[None, Errno]:
        ...

    def flush(self: 'sd_bus') -> Union[None, Errno]:
        ...

    def get_fd(self: 'sd_bus') -> Union[int, Errno]:
        raise NotImplementedError

    def _message_new_method_call(
        self: 'sd_bus',
        message: Reference[sd_bus_message],
        destination: Optional[str],
        path: str,
        interface: str,
        member: str,
    ) -> Union[None, Errno]:
        ...

    def _message_new_signal(
        self: 'sd_bus',
        message: Reference[sd_bus_message],
        path: str,
        interface: str,
        member: str,
    ) -> Union[None, Errno]:
        ...

    def release_name(self: 'sd_bus', name: str) -> Union[None, Errno]:
        ...

    def request_name(
        self: 'sd_bus', name: str, flags: Annotated[int, ctypes.c_uint64]
    ) -> Union[None, Errno]:
        ...

    def set_address(self: 'sd_bus', address: str) -> Union[None, Errno]:
        ...

    def set_allow_interactive_authorization(self: 'sd_bus', b: bool) -> Union[None, Errno]:
        ...

    def set_bus_client(self: 'sd_bus', b: bool) -> Union[None, Errno]:
        ...

    def set_fd(self: 'sd_bus', input_fd: int, output_fd: int) -> Union[None, Errno]:
        ...

    def set_server(self: 'sd_bus', b: bool, bus_d: sd_id128) -> Union[None, Errno]:
        ...

    def start(self: 'sd_bus') -> Union[None, Errno]:
        ...

    def wait(
        self: 'sd_bus', timeout_usec: Annotated[int, ctypes.c_uint64]
    ) -> Union[None, Errno]:
        ...

    def send(
        self: 'sd_bus', message: sd_bus_message, cookie: Optional[Reference[ctypes.c_uint64]]
    ) -> Union[None, Errno]:
        ...

    @staticmethod
    def _default_system(ret: Reference['sd_bus']) -> Union[None, Errno]:
        ...

    @staticmethod
    def _default_user(ret: Reference['sd_bus']) -> Union[None, Errno]:
        ...

    @staticmethod
    def _new(ret: Reference['sd_bus']) -> Union[None, Errno]:
        ...


sd_bus_message_handler_t = ctypes.CFUNCTYPE(
    ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(sd_bus_error))
sd_event_inotify_handler_t = ctypes.CFUNCTYPE(
    ctypes.c_int, ctypes.c_void_p, ctypes.POINTER(inotify_event), ctypes.c_void_p)


libsystemd = ctypes.CDLL("libsystemd.so.0")
for cls in {
    sd_bus,
    sd_bus_message,
    sd_bus_slot,
    sd_event,
    sd_event_source,
}:
    cls._install_cfuncs(libsystemd)