#!/usr/bin/python3
"""Steam's login1 sleep notifications, backed by DroidDeck's session pause.

Android owns device power. Suspend here pauses the guest through SessionService;
the app's Resume action thaws it and completes PrepareForSleep(false). Steam
must see this service before startup or it can stay black after its idle timer.
"""
import os
from pathlib import Path
import re
import sys
import uuid

import gi

gi.require_version("Gio", "2.0")
from gi.repository import Gio, GLib  # noqa: E402

NAME = "org.freedesktop.login1"
ROOT = "/org/freedesktop/login1"
INTERFACE = NAME + ".Manager"
UPOWER = "org.freedesktop.UPower"
UPOWER_ROOT = "/org/freedesktop/UPower"
NATIVE_HOST_ACK_MS = 12000  # Android falls back after ten seconds; allow its freeze to finish.
INTROSPECTION = """
<node>
  <interface name="org.freedesktop.login1.Manager">
    <method name="Suspend"><arg type="b" direction="in"/></method>
    <method name="SuspendWithFlags"><arg type="t" direction="in"/></method>
    <method name="CanSuspend"><arg type="s" direction="out"/></method>
    <method name="Inhibit">
      <arg type="s" direction="in"/><arg type="s" direction="in"/>
      <arg type="s" direction="in"/><arg type="s" direction="in"/>
      <arg type="h" direction="out"/>
    </method>
    <method name="ListInhibitors"><arg type="a(ssssuu)" direction="out"/></method>
    <property name="BlockInhibited" type="s" access="read"/>
    <property name="DelayInhibited" type="s" access="read"/>
    <property name="InhibitDelayMaxUSec" type="t" access="read"/>
    <property name="PreparingForSleep" type="b" access="read"/>
    <signal name="PrepareForSleep"><arg type="b"/></signal>
  </interface>
</node>
"""
UPOWER_INTROSPECTION = """
<node>
  <interface name="org.freedesktop.UPower">
    <method name="Suspend"/>
  </interface>
</node>
"""


class LoginManager:
    def __init__(self, bus, directory):
        self.bus = bus
        self.directory = Path(directory)
        self.request = self.directory / "steam-sleep"
        self.state = self.directory / "steam-sleep-state"
        self.wake = self.directory / "steam-wake"
        self.native_request = self.directory / "steam-native-sleep"
        self.native_ready = self.directory / "steam-native-ready"
        self.pending = None
        self.timeout = None
        self.inhibitors = {}
        self.next_inhibitor = 0
        self.native = False
        self.waiting = set()
        self.steam_participated = False
        self.preparation_failed = False
        self.preparation_flushed = False
        info = Gio.DBusNodeInfo.new_for_xml(INTROSPECTION).interfaces[0]
        self.registration = bus.register_object(ROOT, info, self.method, self.property, None)
        # Steam can issue its suspend command through the legacy UPower service even
        # while listening to login1's sleep/resume notifications. Use one handshake.
        upower_info = Gio.DBusNodeInfo.new_for_xml(UPOWER_INTROSPECTION).interfaces[0]
        self.upower_registration = bus.register_object(UPOWER_ROOT, upower_info, self.method, None, None)
        self.monitor = Gio.File.new_for_path(str(self.directory)).monitor_directory(Gio.FileMonitorFlags.NONE, None)
        self.monitor.connect("changed", self.changed)

    def property(self, connection, sender, path, interface, name):
        if name == "PreparingForSleep":
            return GLib.Variant("b", self.pending is not None)
        if name == "InhibitDelayMaxUSec":
            return GLib.Variant("t", 10000000)
        if name in ("BlockInhibited", "DelayInhibited"):
            mode = "block" if name == "BlockInhibited" else "delay"
            kinds = {kind for item in self.inhibitors.values() if item[3] == mode
                     for kind in item[0].split(":")}
            return GLib.Variant("s", ":".join(sorted(kinds)))
        return None

    def notify(self, sleeping):
        self.bus.emit_signal(None, ROOT, INTERFACE, "PrepareForSleep", GLib.Variant("(b)", (sleeping,)))
        self.bus.emit_signal(None, ROOT, "org.freedesktop.DBus.Properties", "PropertiesChanged",
                             GLib.Variant("(sa{sv}as)", (INTERFACE, {
                                 "PreparingForSleep": GLib.Variant("b", sleeping),
                             }, [])))
        print("== login1: " + ("pausing the session" if sleeping else "session resumed"), flush=True)

    def method(self, connection, sender, path, interface, name, args, reply):
        if name == "CanSuspend":
            reply.return_value(GLib.Variant("(s)", ("yes",)))
        elif name == "Inhibit":
            self.inhibit(sender, args.unpack(), reply)
        elif name == "ListInhibitors":
            reply.return_value(GLib.Variant("(a(ssssuu))", ([item[:6] for item in self.inhibitors.values()],)))
        elif name in ("Suspend", "SuspendWithFlags"):
            if self.pending:
                reply.return_dbus_error(NAME + ".OperationInProgress", "The session is already pausing")
                return
            self.pending = uuid.uuid4().hex
            self.notify(True)
            reply.return_value(None)
            # Flush the signal and method reply without blocking this D-Bus callback. Once both are
            # on the wire, publishing the request lets the host safely freeze this service.
            self.bus.flush(None, self.publish_after_flush, self.pending)

    def peer(self, sender, method, signature):
        return self.bus.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus",
                                  "org.freedesktop.DBus", method, GLib.Variant("(s)", (sender,)),
                                  GLib.VariantType.new(signature), Gio.DBusCallFlags.NONE, 2000, None).unpack()[0]

    def inhibit(self, sender, args, reply):
        what, who, why, mode = args
        kinds = set(what.split(":"))
        valid = {"sleep", "shutdown", "idle", "handle-power-key", "handle-suspend-key",
                 "handle-hibernate-key", "handle-lid-switch"}
        if mode not in ("block", "delay") or not kinds <= valid or (mode == "delay" and not kinds <= {"sleep", "shutdown"}):
            reply.return_dbus_error("org.freedesktop.DBus.Error.InvalidArgs", "Invalid inhibitor type or mode")
            return
        try:
            uid = self.peer(sender, "GetConnectionUnixUser", "(u)")
            pid = self.peer(sender, "GetConnectionUnixProcessID", "(u)")
            read_fd, write_fd = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC)
            descriptors = Gio.UnixFDList.new()
            try:
                index = descriptors.append(write_fd)
            finally:
                os.close(write_fd)
            self.next_inhibitor += 1
            key = self.next_inhibitor
            self.inhibitors[key] = (what, who, why, mode, uid, pid, sender, read_fd)
            GLib.io_add_watch(read_fd, GLib.IOCondition.HUP | GLib.IOCondition.ERR,
                              self.release_inhibitor, key)
            if self.native and mode == "delay" and "sleep" in kinds:
                self.waiting.add(key)
                self.steam_participated |= who.lower() == "steam"
            reply.return_value_with_unix_fd_list(GLib.Variant("(h)", (index,)), descriptors)
            print("== login1: inhibitor acquired: " + who + " " + what + " " + mode, flush=True)
        except (GLib.Error, OSError) as error:
            reply.return_dbus_error(NAME + ".Failed", str(error))

    def release_inhibitor(self, fd, condition, key):
        item = self.inhibitors.pop(key)
        os.close(fd)
        print("== login1: inhibitor released: " + item[1], flush=True)
        if key in self.waiting:
            self.waiting.remove(key)
            # A disconnected/crashed client is not a preparation acknowledgement.
            try:
                alive = self.peer(item[6], "NameHasOwner", "(b)")
                os.kill(item[5], 0)
            except (GLib.Error, OSError):
                alive = False
            self.preparation_failed |= not alive
            self.finish_preparation()
        return False

    def publish_after_flush(self, connection, result, token):
        try:
            connection.flush_finish(result)
        except GLib.Error as error:
            print("== login1: cannot complete suspend handshake: " + str(error), file=sys.stderr)
            self.resume()
            return
        if self.pending == token:
            self.publish(self.request)

    def publish(self, request):
        self.check_state()
        if self.pending is None:
            return False
        try:
            staged = request.with_suffix(".tmp")
            staged.write_text(self.pending + "\n")
            staged.replace(request)
            if self.timeout is not None:
                GLib.source_remove(self.timeout)
            self.timeout = GLib.timeout_add_seconds(5, self.check_ack)
        except OSError as error:
            print("== login1: cannot request pause: " + str(error), file=sys.stderr)
            self.resume()
        return False

    def prepare_native(self):
        try:
            token = self.native_request.read_text().strip()
            self.native_request.unlink(missing_ok=True)
        except OSError:
            return
        if not re.fullmatch(r"[a-f0-9]{32}", token) or self.pending is not None:
            return
        # Android may have returned before this helper read its request.
        if self.read_state() == token + ":awake":
            return
        self.pending = token
        self.native = True
        self.waiting = {key for key, item in self.inhibitors.items()
                        if item[3] == "delay" and "sleep" in item[0].split(":")}
        self.steam_participated = any(self.inhibitors[key][1].lower() == "steam" for key in self.waiting)
        self.preparation_failed = False
        self.preparation_flushed = False
        self.timeout = GLib.timeout_add(NATIVE_HOST_ACK_MS, self.check_ack)
        self.notify(True)
        self.bus.flush(None, self.prepare_after_flush, token)

    def prepare_after_flush(self, connection, result, token):
        try:
            connection.flush_finish(result)
        except GLib.Error as error:
            print("== login1: cannot prepare native sleep: " + str(error), file=sys.stderr)
            self.resume()
            return
        self.check_state()
        if self.pending == token:
            self.preparation_flushed = True
            self.finish_preparation()

    def finish_preparation(self):
        self.check_state()
        if (self.native and self.pending and self.preparation_flushed and self.steam_participated
                and not self.waiting and not self.preparation_failed):
            self.native = False
            print("== login1: Steam acknowledged sleep preparation", flush=True)
            self.publish(self.native_ready)
        return False

    def changed(self, monitor, file, other, event):
        if file.get_basename() == self.native_request.name or (other and other.get_basename() == self.native_request.name):
            self.prepare_native()
        if file.get_basename() == self.wake.name or (other and other.get_basename() == self.wake.name):
            if self.wake.exists():
                self.wake.unlink(missing_ok=True)
                self.resume(force=True)
        if file.get_basename() == self.state.name or (other and other.get_basename() == self.state.name):
            self.check_state()

    def read_state(self):
        try:
            return self.state.read_text().strip()
        except OSError:
            return ""

    def check_state(self):
        if self.pending and self.read_state() == self.pending + ":awake":
            self.resume()

    def check_ack(self):
        self.timeout = None
        if self.pending and self.read_state() != self.pending + ":paused":
            # A missing host or a failed freeze must not strand Steam in black.
            self.request.unlink(missing_ok=True)
            self.native_ready.unlink(missing_ok=True)
            self.resume()
        return False

    def resume(self, force=False):
        if self.pending is None:
            if force:
                # A missed suspend command leaves Steam without the first half of
                # the handshake. Replay both notifications, without pausing again.
                self.notify(True)
                self.bus.flush(None, self.finish_recovery, None)
            return
        self.pending = None
        self.native = False
        self.waiting.clear()
        if self.timeout is not None:
            GLib.source_remove(self.timeout)
            self.timeout = None
        self.notify(False)

    def finish_recovery(self, connection, result, _data):
        try:
            connection.flush_finish(result)
        except GLib.Error as error:
            print("== login1: cannot flush wake recovery: " + str(error), file=sys.stderr)
        if self.pending is None:
            self.notify(False)


def main():
    directory = os.environ.get("BL_LAUNCH_DIR")
    if not directory or not Path(directory).is_dir():
        sys.exit("droiddeck-login1: no DroidDeck session directory")
    ready = Path(directory) / "steam-sleep-ready"
    ready.unlink(missing_ok=True)
    loop = GLib.MainLoop()
    manager = []
    acquired_names = set()

    def connected(bus, name):
        if not manager:
            manager.append(LoginManager(bus, directory))

    def acquired(bus, name):
        acquired_names.add(name)
        if acquired_names == {NAME, UPOWER}:
            ready.write_text("ready\n")

    def lost(bus, name):
        ready.unlink(missing_ok=True)
        loop.quit()

    owners = [Gio.bus_own_name(Gio.BusType.SYSTEM, name, Gio.BusNameOwnerFlags.NONE, connected, acquired, lost)
              for name in (NAME, UPOWER)]
    try:
        loop.run()
    finally:
        ready.unlink(missing_ok=True)
        for owner in owners:
            Gio.bus_unown_name(owner)


if __name__ == "__main__":
    main()
