//! Bounded asynchronous systemd D-Bus restart adapter.
//!
//! zbus's `async-io` backend owns its socket-driving thread. This module uses
//! a small standard-library executor to poll each caller-side future only
//! until the shared monotonic deadline; it does not use zbus's blocking signal
//! iterator, whose `next` call has no application deadline.

use std::future::{Future, poll_fn};
use std::pin::{Pin, pin};
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use std::thread;
use std::time::Instant;

use zbus::connection::Builder;
use zbus::export::futures_core::Stream;
use zbus::message::Type;
use zbus::{Connection, MatchRule, MessageStream, Proxy};

use super::{
    JobResult, RestartDeadline, RestartExecution, RestartFailure, RestartJobTracker,
    RestartOutcome, RestartPhase, ServiceEvent,
};

const SYSTEMD_NAME: &str = "org.freedesktop.systemd1";
const SYSTEMD_PATH: &str = "/org/freedesktop/systemd1";
const SYSTEMD_MANAGER: &str = "org.freedesktop.systemd1.Manager";
const DBUS_NAME: &str = "org.freedesktop.DBus";
const DBUS_INTERFACE: &str = "org.freedesktop.DBus";
const SYSTEMD_PRIVATE_ADDRESS: &str = "unix:path=/run/systemd/private";

/// Resolve the configured unit's primary ID, request a restart of exactly
/// `configured_unit`, and wait for its matching systemd `JobRemoved` result.
/// The caller owns backup ordering and subsequent runtime verification, and
/// must pass the same deadline to those stages.
/// Restart a configured unit and retain whether a timed-out request was submitted.
pub(crate) fn restart_systemd_unit_report(
    configured_unit: &str,
    deadline: RestartDeadline,
) -> RestartExecution {
    restart_with_transport_report(&mut SystemdBus::default(), configured_unit, deadline)
}

trait RestartTransport {
    fn connect(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure>;
    fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure>;
    fn resolve_primary_unit(
        &mut self,
        configured_unit: &str,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure>;
    fn restart_unit(
        &mut self,
        unit: &str,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure>;
    fn next_event(&mut self, deadline: RestartDeadline) -> Result<ServiceEvent, RestartFailure>;
}

#[cfg(test)]
fn restart_with_transport(
    transport: &mut impl RestartTransport,
    configured_unit: &str,
    deadline: RestartDeadline,
) -> RestartOutcome {
    restart_with_transport_report(transport, configured_unit, deadline).outcome
}

fn restart_with_transport_report(
    transport: &mut impl RestartTransport,
    configured_unit: &str,
    deadline: RestartDeadline,
) -> RestartExecution {
    if configured_unit.is_empty() || configured_unit.as_bytes().contains(&0) {
        return execution(
            RestartOutcome::Failed(RestartFailure::RequestFailed),
            false,
            RestartPhase::RequestRestart,
        );
    }

    if let Err(failure) = transport.connect(deadline) {
        return execution(
            RestartOutcome::Failed(failure),
            false,
            RestartPhase::Connect,
        );
    }
    if deadline.is_expired_at(Instant::now()) {
        return execution(
            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete),
            false,
            RestartPhase::Connect,
        );
    }
    if let Err(failure) = transport.subscribe(deadline) {
        return execution(
            RestartOutcome::Failed(failure),
            false,
            RestartPhase::Subscribe,
        );
    }
    if deadline.is_expired_at(Instant::now()) {
        return execution(
            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete),
            false,
            RestartPhase::Subscribe,
        );
    }
    let primary_unit = match transport.resolve_primary_unit(configured_unit, deadline) {
        Ok(unit) if !unit.is_empty() => unit,
        Ok(_) => {
            return execution(
                RestartOutcome::Failed(RestartFailure::UnitResolutionFailed),
                false,
                RestartPhase::ResolveUnit,
            );
        }
        Err(failure) => {
            return execution(
                RestartOutcome::Failed(failure),
                false,
                RestartPhase::ResolveUnit,
            );
        }
    };
    if deadline.is_expired_at(Instant::now()) {
        return execution(
            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete),
            false,
            RestartPhase::ResolveUnit,
        );
    }

    let job_path = match transport.restart_unit(configured_unit, deadline) {
        Ok(path) => path,
        Err(failure) => {
            let submitted = matches!(
                failure,
                RestartFailure::WaitTimedOutMayStillComplete
                    | RestartFailure::BusDisconnected
                    | RestartFailure::ManagerOwnerChanged
            );
            return execution(
                RestartOutcome::Failed(failure),
                submitted,
                RestartPhase::RequestRestart,
            );
        }
    };
    let mut tracker = RestartJobTracker::new(primary_unit, deadline);
    let mut outcome = tracker.job_submitted(job_path, Instant::now());
    while outcome == RestartOutcome::Waiting {
        let event = match transport.next_event(deadline) {
            Ok(event) => event,
            Err(failure) => {
                return execution(
                    RestartOutcome::Failed(failure),
                    true,
                    RestartPhase::WaitForJob,
                );
            }
        };
        outcome = tracker.observe(event, Instant::now());
    }
    execution(outcome, true, RestartPhase::WaitForJob)
}

fn execution(outcome: RestartOutcome, submitted: bool, phase: RestartPhase) -> RestartExecution {
    RestartExecution {
        outcome,
        submitted,
        phase,
    }
}

#[derive(Default)]
struct SystemdBus {
    connection: Option<Connection>,
    private_peer: bool,
    manager: Option<Proxy<'static>>,
    job_events: Option<JobSignalStream>,
    owner_events: Option<zbus::MessageStream>,
}

enum JobSignalStream {
    Bus(Box<zbus::proxy::SignalStream<'static>>),
    Peer(Box<MessageStream>),
}

/// The small systemd unit API used by the shared resolver. Keeping method
/// selection and canonical-ID lookup here lets synthetic tests exercise the
/// same resolution flow as the production adapter.
trait UnitLoadApi {
    type UnitPath;

    fn load_unit(
        &mut self,
        configured_unit: &str,
        deadline: RestartDeadline,
    ) -> Result<Self::UnitPath, RestartFailure>;
    fn load_state(
        &mut self,
        unit_path: &Self::UnitPath,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure>;
    fn canonical_id(
        &mut self,
        unit_path: &Self::UnitPath,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure>;
}

fn resolve_loaded_unit(
    api: &mut impl UnitLoadApi,
    configured_unit: &str,
    deadline: RestartDeadline,
) -> Result<String, RestartFailure> {
    let unit_path = api.load_unit(configured_unit, deadline)?;
    if !unit_is_loaded(&api.load_state(&unit_path, deadline)?) {
        return Err(RestartFailure::UnitResolutionFailed);
    }
    api.canonical_id(&unit_path, deadline)
}

impl UnitLoadApi for SystemdBus {
    type UnitPath = zbus::zvariant::OwnedObjectPath;

    fn load_unit(
        &mut self,
        configured_unit: &str,
        deadline: RestartDeadline,
    ) -> Result<Self::UnitPath, RestartFailure> {
        self.call_manager("LoadUnit", &(configured_unit,), deadline)
            .map_err(map_unit_resolution_failure)?
            .body()
            .deserialize()
            .map_err(|_| RestartFailure::UnitResolutionFailed)
    }

    fn load_state(
        &mut self,
        unit_path: &Self::UnitPath,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure> {
        self.unit_property(unit_path, "LoadState", deadline)
    }

    fn canonical_id(
        &mut self,
        unit_path: &Self::UnitPath,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure> {
        self.unit_property(unit_path, "Id", deadline)
    }
}

impl SystemdBus {
    fn call_manager<B: serde::ser::Serialize + zbus::zvariant::DynamicType>(
        &self,
        method: &'static str,
        body: &B,
        deadline: RestartDeadline,
    ) -> Result<zbus::message::Message, RestartFailure> {
        let connection = self
            .connection
            .as_ref()
            .ok_or(RestartFailure::BusDisconnected)?;
        let destination = manager_destination(self.private_peer);
        drive(
            connection.call_method(
                destination,
                SYSTEMD_PATH,
                Some(SYSTEMD_MANAGER),
                method,
                body,
            ),
            deadline,
        )?
        .map_err(map_bus_error)
    }

    fn unit_property(
        &self,
        unit_path: &zbus::zvariant::OwnedObjectPath,
        property: &'static str,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure> {
        let connection = self
            .connection
            .as_ref()
            .ok_or(RestartFailure::BusDisconnected)?;
        if !self.private_peer {
            let unit = drive(
                Proxy::new_owned(
                    connection.clone(),
                    SYSTEMD_NAME,
                    unit_path.clone(),
                    "org.freedesktop.systemd1.Unit",
                ),
                deadline,
            )?
            .map_err(map_unit_resolution_error)?;
            return drive(unit.get_property::<String>(property), deadline)?
                .map_err(map_unit_resolution_error);
        }
        let message = drive(
            connection.call_method(
                Option::<&str>::None,
                unit_path.clone(),
                Some("org.freedesktop.DBus.Properties"),
                "Get",
                &("org.freedesktop.systemd1.Unit", property),
            ),
            deadline,
        )?
        .map_err(map_unit_resolution_error)?;
        let value: zbus::zvariant::OwnedValue = message
            .body()
            .deserialize()
            .map_err(|_| RestartFailure::UnitResolutionFailed)?;
        String::try_from(value).map_err(|_| RestartFailure::UnitResolutionFailed)
    }
}

fn manager_destination(private_peer: bool) -> Option<&'static str> {
    (!private_peer).then_some(SYSTEMD_NAME)
}

fn owner_monitoring_enabled(private_peer: bool) -> bool {
    !private_peer
}

fn select_connection<T>(
    system_bus: Result<T, (RestartFailure, bool)>,
    private_peer: impl FnOnce() -> Result<T, RestartFailure>,
) -> Result<(T, bool), RestartFailure> {
    match system_bus {
        Ok(connection) => Ok((connection, false)),
        // Once private fallback is attempted, keep its typed failure: it is
        // the selected transport's actionable result.
        Err((_failure, true)) => private_peer().map(|connection| (connection, true)),
        Err((failure, false)) => Err(failure),
    }
}

impl RestartTransport for SystemdBus {
    fn connect(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
        let bus_connection = connect_system_bus(deadline);
        let ((connection, manager), private_peer) = select_connection(bus_connection, || {
            let connection = drive(
                Builder::address(SYSTEMD_PRIVATE_ADDRESS)
                    .map_err(map_bus_error)?
                    .p2p()
                    .auth_mechanism(zbus::connection::AuthMechanism::External)
                    .build(),
                deadline,
            )?
            .map_err(map_private_builder_error)?;
            let uid = drive(connection.peer_creds(), deadline)?
                .map_err(map_peer_creds_error)?
                .unix_user_id();
            if !is_root_peer_uid(uid) {
                return Err(RestartFailure::PermissionDenied);
            }
            Ok((connection, None))
        })?;
        self.private_peer = private_peer;
        self.manager = manager;
        self.connection = Some(connection);
        Ok(())
    }

    fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
        let connection = self
            .connection
            .as_ref()
            .ok_or(RestartFailure::SubscriptionFailed)?;
        let job_events = if self.private_peer {
            let rule = MatchRule::builder()
                .msg_type(Type::Signal)
                .path(SYSTEMD_PATH)
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .interface(SYSTEMD_MANAGER)
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .member("JobRemoved")
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .build();
            JobSignalStream::Peer(Box::new(
                drive(
                    MessageStream::for_match_rule(rule, connection, Some(8)),
                    deadline,
                )?
                .map_err(|_| RestartFailure::SubscriptionFailed)?,
            ))
        } else {
            let manager = self
                .manager
                .as_ref()
                .ok_or(RestartFailure::SubscriptionFailed)?;
            JobSignalStream::Bus(Box::new(
                drive(manager.receive_signal("JobRemoved"), deadline)?
                    .map_err(|_| RestartFailure::SubscriptionFailed)?,
            ))
        };

        // Peer mode uses zbus's local p2p match and must not call the bus daemon.
        // The bus mode additionally watches owner changes before Subscribe.
        self.job_events = Some(job_events);
        if owner_monitoring_enabled(self.private_peer) {
            let owner_rule = MatchRule::builder()
                .msg_type(Type::Signal)
                .sender(DBUS_NAME)
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .interface(DBUS_INTERFACE)
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .member("NameOwnerChanged")
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .add_arg(SYSTEMD_NAME)
                .map_err(|_| RestartFailure::SubscriptionFailed)?
                .build();
            self.owner_events = Some(
                drive(
                    MessageStream::for_match_rule(owner_rule, connection, Some(1)),
                    deadline,
                )?
                .map_err(|_| RestartFailure::SubscriptionFailed)?,
            );
        }
        self.call_manager("Subscribe", &(), deadline)?;
        Ok(())
    }

    fn resolve_primary_unit(
        &mut self,
        configured_unit: &str,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure> {
        resolve_loaded_unit(self, configured_unit, deadline)
    }

    fn restart_unit(
        &mut self,
        unit: &str,
        deadline: RestartDeadline,
    ) -> Result<String, RestartFailure> {
        let connection = self
            .connection
            .as_ref()
            .ok_or(RestartFailure::BusDisconnected)?;
        let request_body = (unit, "replace");
        let job_path: zbus::zvariant::OwnedObjectPath = if self.private_peer {
            self.call_manager("RestartUnit", &request_body, deadline)?
                .body()
                .deserialize()
                .map_err(|_| RestartFailure::RequestFailed)?
        } else {
            let manager = self.manager.as_ref().ok_or(RestartFailure::RequestFailed)?;
            let owner_events = self
                .owner_events
                .as_mut()
                .ok_or(RestartFailure::BusDisconnected)?;
            let request =
                manager.call::<_, _, zbus::zvariant::OwnedObjectPath>("RestartUnit", &request_body);
            drive(
                async {
                    let mut request = pin!(request);
                    let mut closed = pin!(connection.closed());
                    poll_fn(|cx| {
                        if connection.is_closed() {
                            return Poll::Ready(Err(RestartFailure::BusDisconnected));
                        }
                        if let Poll::Ready(event) =
                            Stream::poll_next(Pin::new(&mut *owner_events), cx)
                        {
                            return Poll::Ready(match event {
                                None => Err(RestartFailure::BusDisconnected),
                                Some(Err(error)) => Err(map_bus_error(error)),
                                Some(Ok(_)) => Err(RestartFailure::ManagerOwnerChanged),
                            });
                        }
                        if let Poll::Ready(result) = Future::poll(request.as_mut(), cx) {
                            return Poll::Ready(result.map_err(map_bus_error));
                        }
                        if Future::poll(closed.as_mut(), cx).is_ready() {
                            return Poll::Ready(Err(RestartFailure::BusDisconnected));
                        }
                        Poll::Pending
                    })
                    .await
                },
                deadline,
            )??
        };
        Ok(job_path.to_string())
    }

    fn next_event(&mut self, deadline: RestartDeadline) -> Result<ServiceEvent, RestartFailure> {
        let connection = self
            .connection
            .as_ref()
            .ok_or(RestartFailure::BusDisconnected)?;
        let job_events = self
            .job_events
            .as_mut()
            .ok_or(RestartFailure::BusDisconnected)?;
        let mut owner_events = self.owner_events.as_mut();
        drive(
            async {
                let mut closed = pin!(connection.closed());
                poll_fn(|cx| {
                    if connection.is_closed() {
                        return Poll::Ready(Ok(ServiceEvent::Disconnected));
                    }
                    if let Some(owner_events) = owner_events.as_mut()
                        && let Poll::Ready(message) = Stream::poll_next(Pin::new(owner_events), cx)
                    {
                        return Poll::Ready(match message {
                            None => Ok(ServiceEvent::Disconnected),
                            Some(Err(error)) => Err(map_bus_error(error)),
                            Some(Ok(message)) => {
                                let decoded: Result<(String, String, String), _> =
                                    message.body().deserialize();
                                match decoded {
                                    Ok((name, _, _)) if name == SYSTEMD_NAME => {
                                        Ok(ServiceEvent::ManagerOwnerChanged)
                                    }
                                    Ok(_) => Ok(ServiceEvent::JobRemoved {
                                        job_path: String::new(),
                                        unit: String::new(),
                                        result: JobResult::Other,
                                    }),
                                    Err(_) => Err(RestartFailure::RequestFailed),
                                }
                            }
                        });
                    }
                    let message = match job_events {
                        JobSignalStream::Bus(stream) => {
                            Stream::poll_next(Pin::new(stream), cx).map(|message| message.map(Ok))
                        }
                        JobSignalStream::Peer(stream) => Stream::poll_next(Pin::new(stream), cx),
                    };
                    if let Poll::Ready(message) = message {
                        return Poll::Ready(match message {
                            None => Ok(ServiceEvent::Disconnected),
                            Some(Err(error)) => Err(map_bus_error(error)),
                            Some(Ok(message)) => {
                                let decoded: Result<
                                    (u32, zbus::zvariant::OwnedObjectPath, String, String),
                                    _,
                                > = message.body().deserialize();
                                match decoded {
                                    Ok((_, path, unit, result)) => Ok(ServiceEvent::JobRemoved {
                                        job_path: path.to_string(),
                                        unit,
                                        result: map_job_result(&result),
                                    }),
                                    Err(_) => Err(RestartFailure::RequestFailed),
                                }
                            }
                        });
                    }
                    if Future::poll(closed.as_mut(), cx).is_ready() {
                        return Poll::Ready(Ok(ServiceEvent::Disconnected));
                    }
                    Poll::Pending
                })
                .await
            },
            deadline,
        )?
    }
}

fn connect_system_bus(
    deadline: RestartDeadline,
) -> Result<(Connection, Option<Proxy<'static>>), (RestartFailure, bool)> {
    let connection = match drive(Connection::system(), deadline) {
        Err(failure) => return Err((failure, false)),
        Ok(Err(error)) => {
            return Err((map_connect_error(&error), private_fallback_allowed(&error)));
        }
        Ok(Ok(connection)) => connection,
    };
    let manager = match drive(
        Proxy::new_owned(
            connection.clone(),
            SYSTEMD_NAME,
            SYSTEMD_PATH,
            SYSTEMD_MANAGER,
        ),
        deadline,
    ) {
        Err(failure) => return Err((failure, false)),
        Ok(Err(error)) => {
            return Err((map_connect_error(&error), private_fallback_allowed(&error)));
        }
        Ok(Ok(manager)) => manager,
    };
    Ok((connection, Some(manager)))
}

fn unit_is_loaded(load_state: &str) -> bool {
    load_state == "loaded"
}

fn map_job_result(result: &str) -> JobResult {
    match result {
        "done" => JobResult::Done,
        "failed" => JobResult::Failed,
        "dependency" => JobResult::Dependency,
        "timeout" => JobResult::Timeout,
        "canceled" => JobResult::Canceled,
        "skipped" => JobResult::Skipped,
        _ => JobResult::Other,
    }
}

fn map_bus_error(error: zbus::Error) -> RestartFailure {
    match error {
        zbus::Error::MethodError(name, _, _)
            if name.as_str() == "org.freedesktop.DBus.Error.AccessDenied"
                || name.as_str() == "org.freedesktop.systemd1.AccessDenied" =>
        {
            RestartFailure::PermissionDenied
        }
        zbus::Error::InputOutput(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
            RestartFailure::PermissionDenied
        }
        zbus::Error::InputOutput(_) => RestartFailure::BusDisconnected,
        _ => RestartFailure::RequestFailed,
    }
}

fn map_connect_error(error: &zbus::Error) -> RestartFailure {
    match error {
        zbus::Error::Connection(error, _) | zbus::Error::InputOutput(error)
            if error.kind() == std::io::ErrorKind::PermissionDenied =>
        {
            RestartFailure::PermissionDenied
        }
        zbus::Error::Connection(_, _) | zbus::Error::InputOutput(_) => {
            RestartFailure::BusDisconnected
        }
        // A handshake rejection is an authentication/protocol failure, not an
        // unavailable transport. Never bypass it via the private socket.
        _ => RestartFailure::RequestFailed,
    }
}

fn map_private_builder_error(error: zbus::Error) -> RestartFailure {
    map_connect_error(&error)
}

fn private_fallback_allowed(error: &zbus::Error) -> bool {
    match error {
        zbus::Error::Connection(io, _) | zbus::Error::InputOutput(io) => matches!(
            io.kind(),
            std::io::ErrorKind::NotFound
                | std::io::ErrorKind::ConnectionRefused
                | std::io::ErrorKind::ConnectionReset
                | std::io::ErrorKind::ConnectionAborted
                | std::io::ErrorKind::NotConnected
                | std::io::ErrorKind::BrokenPipe
        ),
        _ => false,
    }
}

fn is_root_peer_uid(uid: Option<u32>) -> bool {
    uid == Some(0)
}

fn map_peer_creds_error(error: std::io::Error) -> RestartFailure {
    if error.kind() == std::io::ErrorKind::PermissionDenied {
        RestartFailure::PermissionDenied
    } else {
        RestartFailure::BusDisconnected
    }
}

fn map_unit_resolution_error(error: zbus::Error) -> RestartFailure {
    match map_bus_error(error) {
        RestartFailure::PermissionDenied => RestartFailure::PermissionDenied,
        RestartFailure::BusDisconnected => RestartFailure::BusDisconnected,
        _ => RestartFailure::UnitResolutionFailed,
    }
}

fn map_unit_resolution_failure(failure: RestartFailure) -> RestartFailure {
    match failure {
        RestartFailure::RequestFailed => RestartFailure::UnitResolutionFailed,
        other => other,
    }
}

struct ThreadWake(thread::Thread);

impl Wake for ThreadWake {
    fn wake(self: Arc<Self>) {
        self.0.unpark();
    }

    fn wake_by_ref(self: &Arc<Self>) {
        self.0.unpark();
    }
}

/// Poll a future on this thread, parking only for the remaining operation
/// budget. zbus's async-io connection executor continues driving socket I/O
/// independently and wakes this thread when the future can make progress.
fn drive<F: Future>(future: F, deadline: RestartDeadline) -> Result<F::Output, RestartFailure> {
    let waker = Waker::from(Arc::new(ThreadWake(thread::current())));
    let mut context = Context::from_waker(&waker);
    let mut future = pin!(future);
    loop {
        if deadline.is_expired_at(Instant::now()) {
            return Err(RestartFailure::WaitTimedOutMayStillComplete);
        }
        match Future::poll(future.as_mut(), &mut context) {
            Poll::Ready(output) => return Ok(output),
            Poll::Pending => {
                let remaining = deadline.remaining_at(Instant::now());
                if remaining.is_zero() {
                    return Err(RestartFailure::WaitTimedOutMayStillComplete);
                }
                thread::park_timeout(remaining);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::VecDeque;
    use std::time::Duration;

    struct FakeBus {
        calls: Vec<&'static str>,
        deadlines: Vec<(&'static str, RestartDeadline)>,
        events: VecDeque<ServiceEvent>,
        failure: Option<(&'static str, RestartFailure)>,
        configured_unit: &'static str,
        job_path: &'static str,
        primary_unit: &'static str,
    }

    impl Default for FakeBus {
        fn default() -> Self {
            Self {
                calls: Vec::new(),
                deadlines: Vec::new(),
                events: VecDeque::new(),
                failure: None,
                configured_unit: "wg-quick@wg0.service",
                job_path: "",
                primary_unit: "wg-quick@wg0.service",
            }
        }
    }

    impl FakeBus {
        fn stage(&mut self, stage: &'static str) -> Result<(), RestartFailure> {
            self.calls.push(stage);
            match self.failure {
                Some((failed_stage, failure)) if failed_stage == stage => Err(failure),
                _ => Ok(()),
            }
        }
    }

    impl RestartTransport for FakeBus {
        fn connect(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
            self.deadlines.push(("connect", deadline));
            self.stage("connect")
        }

        fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
            self.deadlines.push(("subscribe", deadline));
            self.stage("subscribe")
        }

        fn resolve_primary_unit(
            &mut self,
            configured_unit: &str,
            deadline: RestartDeadline,
        ) -> Result<String, RestartFailure> {
            assert_eq!(configured_unit, self.configured_unit);
            self.deadlines.push(("resolve", deadline));
            self.stage("resolve")?;
            Ok(self.primary_unit.to_owned())
        }

        fn restart_unit(
            &mut self,
            unit: &str,
            deadline: RestartDeadline,
        ) -> Result<String, RestartFailure> {
            assert_eq!(unit, self.configured_unit);
            self.deadlines.push(("restart", deadline));
            self.stage("restart")?;
            Ok(self.job_path.to_owned())
        }

        fn next_event(
            &mut self,
            deadline: RestartDeadline,
        ) -> Result<ServiceEvent, RestartFailure> {
            self.deadlines.push(("event", deadline));
            self.stage("event")?;
            self.events
                .pop_front()
                .ok_or(RestartFailure::BusDisconnected)
        }
    }

    fn deadline() -> RestartDeadline {
        RestartDeadline::new(Instant::now(), Duration::from_secs(2))
    }

    #[test]
    fn transport_selection_falls_back_only_before_restart_setup() {
        let (selected, private) =
            select_connection(Ok("bus"), || Ok("private")).expect("available bus transport");
        assert_eq!(selected, "bus");
        assert!(!private);

        let (selected, private) = select_connection(
            Err::<&str, _>((RestartFailure::BusDisconnected, true)),
            || Ok("private"),
        )
        .expect("private peer fallback");
        assert_eq!(selected, "private");
        assert!(private);
        assert_eq!(manager_destination(true), None);
        assert_eq!(manager_destination(false), Some(SYSTEMD_NAME));
        assert!(!owner_monitoring_enabled(true));
        assert!(owner_monitoring_enabled(false));

        let private_denied =
            select_connection::<&str>(Err((RestartFailure::BusDisconnected, true)), || {
                Err(RestartFailure::PermissionDenied)
            });
        assert_eq!(private_denied, Err(RestartFailure::PermissionDenied));

        let private_timeout =
            select_connection::<&str>(Err((RestartFailure::BusDisconnected, true)), || {
                Err(RestartFailure::WaitTimedOutMayStillComplete)
            });
        assert_eq!(
            private_timeout,
            Err(RestartFailure::WaitTimedOutMayStillComplete)
        );

        let mut private_attempted = false;
        let denied_bus =
            select_connection::<&str>(Err((RestartFailure::PermissionDenied, false)), || {
                private_attempted = true;
                Ok("private")
            });
        assert_eq!(denied_bus, Err(RestartFailure::PermissionDenied));
        assert!(!private_attempted);

        let timeout_bus = select_connection::<&str>(
            Err((RestartFailure::WaitTimedOutMayStillComplete, false)),
            || {
                private_attempted = true;
                Ok("private")
            },
        );
        assert_eq!(
            timeout_bus,
            Err(RestartFailure::WaitTimedOutMayStillComplete)
        );
        assert!(!private_attempted);
    }

    #[test]
    fn private_fallback_is_limited_to_missing_or_refused_bus_endpoint() {
        let unavailable = zbus::Error::Connection(
            Arc::new(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)),
            zbus::Address::try_from("unix:path=/run/dbus/system_bus_socket")
                .expect("valid synthetic bus address"),
        );
        assert!(private_fallback_allowed(&unavailable));
        assert_eq!(
            map_connect_error(&unavailable),
            RestartFailure::BusDisconnected
        );
        assert!(private_fallback_allowed(&zbus::Error::InputOutput(
            Arc::new(std::io::Error::from(std::io::ErrorKind::ConnectionReset))
        )));

        let denied = zbus::Error::Connection(
            Arc::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
            zbus::Address::try_from("unix:path=/run/dbus/system_bus_socket")
                .expect("valid synthetic bus address"),
        );
        assert!(!private_fallback_allowed(&denied));
        assert_eq!(map_connect_error(&denied), RestartFailure::PermissionDenied);
        assert_eq!(
            map_private_builder_error(denied),
            RestartFailure::PermissionDenied
        );
        assert!(!private_fallback_allowed(&zbus::Error::InputOutput(
            Arc::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied))
        )));
        assert!(!private_fallback_allowed(&zbus::Error::Handshake(
            "synthetic authentication rejection".into()
        )));
        assert!(!is_root_peer_uid(None));
        assert!(!is_root_peer_uid(Some(1000)));
        assert!(is_root_peer_uid(Some(0)));
        assert_eq!(
            map_peer_creds_error(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
            RestartFailure::PermissionDenied
        );
        assert_eq!(
            map_peer_creds_error(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
            RestartFailure::BusDisconnected
        );
    }

    fn removed(path: &str, unit: &str, result: JobResult) -> ServiceEvent {
        ServiceEvent::JobRemoved {
            job_path: path.to_owned(),
            unit: unit.to_owned(),
            result,
        }
    }

    #[test]
    fn subscribes_before_restart_and_uses_returned_job_path() {
        let mut fake = FakeBus {
            job_path: "/job/actual",
            primary_unit: "wg-quick@wg0.service",
            events: VecDeque::from([
                removed("/job/other", "wg-quick@wg0.service", JobResult::Done),
                removed("/job/actual", "wg-quick@wg0.service", JobResult::Done),
            ]),
            ..FakeBus::default()
        };
        assert_eq!(
            restart_with_transport(&mut fake, "wg-quick@wg0.service", deadline()),
            RestartOutcome::Succeeded
        );
        assert_eq!(
            fake.calls,
            [
                "connect",
                "subscribe",
                "resolve",
                "restart",
                "event",
                "event"
            ]
        );
    }

    #[test]
    fn fast_completion_and_failure_results_are_processed_without_a_bus() {
        let mut fast = FakeBus {
            job_path: "/job/fast",
            primary_unit: "wg-quick@wg0.service",
            events: VecDeque::from([removed(
                "/job/fast",
                "wg-quick@wg0.service",
                JobResult::Done,
            )]),
            ..FakeBus::default()
        };
        assert_eq!(
            restart_with_transport(&mut fast, "wg-quick@wg0.service", deadline()),
            RestartOutcome::Succeeded
        );

        let mut failed = FakeBus {
            job_path: "/job/failed",
            primary_unit: "wg-quick@wg0.service",
            events: VecDeque::from([removed(
                "/job/failed",
                "wg-quick@wg0.service",
                JobResult::Canceled,
            )]),
            ..FakeBus::default()
        };
        assert_eq!(
            restart_with_transport(&mut failed, "wg-quick@wg0.service", deadline()),
            RestartOutcome::Failed(RestartFailure::JobCanceled)
        );
    }

    #[test]
    fn setup_and_permission_failures_stop_before_restart() {
        for (stage, failure) in [
            ("connect", RestartFailure::BusDisconnected),
            ("subscribe", RestartFailure::SubscriptionFailed),
            ("resolve", RestartFailure::UnitResolutionFailed),
            ("restart", RestartFailure::PermissionDenied),
        ] {
            let mut fake = FakeBus {
                job_path: "/job/never",
                primary_unit: "wg-quick@wg0.service",
                failure: Some((stage, failure)),
                ..FakeBus::default()
            };
            assert_eq!(
                restart_with_transport(&mut fake, "wg-quick@wg0.service", deadline()),
                RestartOutcome::Failed(failure)
            );
            if stage != "restart" {
                assert!(!fake.calls.contains(&"restart"));
            }
        }
    }

    #[test]
    fn resolves_alias_before_restart_and_uses_one_shared_deadline() {
        let deadline = deadline();
        let mut fake = FakeBus {
            job_path: "/job/alias",
            configured_unit: "wireguard-vpn.service",
            primary_unit: "wg-quick@wg0.service",
            events: VecDeque::from([removed(
                "/job/alias",
                "wg-quick@wg0.service",
                JobResult::Done,
            )]),
            ..FakeBus::default()
        };
        assert_eq!(
            restart_with_transport(&mut fake, "wireguard-vpn.service", deadline),
            RestartOutcome::Succeeded
        );
        assert_eq!(
            fake.calls,
            ["connect", "subscribe", "resolve", "restart", "event"]
        );
        assert!(fake.deadlines.iter().all(|(_, passed)| *passed == deadline));
    }

    #[test]
    fn systemd_load_states_allow_only_loaded_units_to_reach_restart() {
        assert!(unit_is_loaded("loaded"));
        for state in ["not-found", "bad-setting", "error", "masked", "stub"] {
            assert!(!unit_is_loaded(state), "unexpected load state: {state}");
        }
    }

    #[test]
    fn unit_resolution_keeps_permission_timeout_and_disconnect_categories() {
        assert_eq!(
            map_unit_resolution_failure(RestartFailure::RequestFailed),
            RestartFailure::UnitResolutionFailed
        );
        for failure in [
            RestartFailure::PermissionDenied,
            RestartFailure::WaitTimedOutMayStillComplete,
            RestartFailure::BusDisconnected,
        ] {
            assert_eq!(map_unit_resolution_failure(failure), failure);
        }
    }

    struct FakeUnitLoadApi {
        calls: Vec<String>,
        configured_unit_seen: Option<String>,
        load_state: Result<String, RestartFailure>,
        canonical_id: Result<String, RestartFailure>,
        load_unit_result: Result<String, RestartFailure>,
    }

    impl UnitLoadApi for FakeUnitLoadApi {
        type UnitPath = String;

        fn load_unit(
            &mut self,
            configured_unit: &str,
            _deadline: RestartDeadline,
        ) -> Result<Self::UnitPath, RestartFailure> {
            self.calls.push("LoadUnit".to_owned());
            self.configured_unit_seen = Some(configured_unit.to_owned());
            self.load_unit_result.clone()
        }

        fn load_state(
            &mut self,
            unit_path: &Self::UnitPath,
            _deadline: RestartDeadline,
        ) -> Result<String, RestartFailure> {
            self.calls.push(format!("LoadState:{unit_path}"));
            self.load_state.clone()
        }

        fn canonical_id(
            &mut self,
            unit_path: &Self::UnitPath,
            _deadline: RestartDeadline,
        ) -> Result<String, RestartFailure> {
            self.calls.push(format!("Id:{unit_path}"));
            self.canonical_id.clone()
        }
    }

    impl FakeUnitLoadApi {
        fn loaded(unit_path: &str, canonical_id: &str) -> Self {
            Self {
                calls: Vec::new(),
                configured_unit_seen: None,
                load_state: Ok("loaded".to_owned()),
                canonical_id: Ok(canonical_id.to_owned()),
                load_unit_result: Ok(unit_path.to_owned()),
            }
        }
    }

    #[test]
    fn production_unit_resolver_loads_unloaded_alias_and_returns_canonical_id() {
        let mut resolver = FakeUnitLoadApi::loaded(
            "/org/freedesktop/systemd1/unit/wg_2dquick_40wg0_2eservice",
            "wg-quick@wg0.service",
        );

        let canonical = resolve_loaded_unit(&mut resolver, "wireguard-vpn.service", deadline())
            .expect("LoadUnit loads existing unloaded alias");

        assert_eq!(
            resolver.configured_unit_seen.as_deref(),
            Some("wireguard-vpn.service")
        );
        assert_eq!(canonical, "wg-quick@wg0.service");
        assert_eq!(
            resolver.calls,
            [
                "LoadUnit",
                "LoadState:/org/freedesktop/systemd1/unit/wg_2dquick_40wg0_2eservice",
                "Id:/org/freedesktop/systemd1/unit/wg_2dquick_40wg0_2eservice",
            ]
        );
    }

    #[test]
    fn production_unit_resolver_stops_for_missing_units_and_access_denial() {
        let mut missing = FakeUnitLoadApi::loaded("/unit/missing", "unused");
        missing.load_state = Ok("not-found".to_owned());
        assert_eq!(
            resolve_loaded_unit(&mut missing, "missing.service", deadline()),
            Err(RestartFailure::UnitResolutionFailed)
        );
        assert_eq!(missing.calls, ["LoadUnit", "LoadState:/unit/missing"]);

        let mut load_error = FakeUnitLoadApi::loaded("/unit/unused", "unused");
        load_error.load_unit_result = Err(RestartFailure::UnitResolutionFailed);
        assert_eq!(
            resolve_loaded_unit(&mut load_error, "absent.service", deadline()),
            Err(RestartFailure::UnitResolutionFailed)
        );
        assert_eq!(load_error.calls, ["LoadUnit"]);

        let mut denied = FakeUnitLoadApi::loaded("/unit/denied", "unused");
        denied.load_unit_result = Err(RestartFailure::PermissionDenied);
        assert_eq!(
            resolve_loaded_unit(&mut denied, "restricted.service", deadline()),
            Err(RestartFailure::PermissionDenied)
        );
        assert_eq!(denied.calls, ["LoadUnit"]);
    }

    #[test]
    fn alias_resolution_and_access_denial_stop_before_restart() {
        let mut alias = FakeBus {
            configured_unit: "wireguard-vpn.service",
            primary_unit: "wg-quick@wg0.service",
            job_path: "/job/alias",
            events: VecDeque::from([removed(
                "/job/alias",
                "wg-quick@wg0.service",
                JobResult::Done,
            )]),
            ..FakeBus::default()
        };
        let execution =
            restart_with_transport_report(&mut alias, "wireguard-vpn.service", deadline());
        assert_eq!(execution.outcome, RestartOutcome::Succeeded);
        assert_eq!(execution.phase, RestartPhase::WaitForJob);
        assert_eq!(
            alias.calls[0..4],
            ["connect", "subscribe", "resolve", "restart"]
        );

        let mut denied = FakeBus {
            failure: Some(("resolve", RestartFailure::PermissionDenied)),
            ..FakeBus::default()
        };
        let denied_unit = denied.configured_unit;
        let execution = restart_with_transport_report(&mut denied, denied_unit, deadline());
        assert_eq!(
            execution.outcome,
            RestartOutcome::Failed(RestartFailure::PermissionDenied)
        );
        assert_eq!(execution.phase, RestartPhase::ResolveUnit);
        assert!(!denied.calls.contains(&"restart"));
    }

    #[test]
    fn owner_change_while_restart_reply_is_pending_is_reported_as_submitted() {
        let mut fake = FakeBus {
            failure: Some(("restart", RestartFailure::ManagerOwnerChanged)),
            ..FakeBus::default()
        };

        let execution =
            restart_with_transport_report(&mut fake, "wg-quick@wg0.service", deadline());

        assert_eq!(
            execution.outcome,
            RestartOutcome::Failed(RestartFailure::ManagerOwnerChanged)
        );
        assert_eq!(execution.phase, RestartPhase::RequestRestart);
        assert!(execution.submitted);
        assert_eq!(
            fake.calls.iter().filter(|call| **call == "restart").count(),
            1
        );
    }

    #[test]
    fn expired_shared_deadline_stops_between_setup_and_submission() {
        let mut fake = FakeBus {
            job_path: "/job/never",
            ..FakeBus::default()
        };
        let expired = RestartDeadline::new(Instant::now(), Duration::ZERO);
        assert_eq!(
            restart_with_transport(&mut fake, "wg-quick@wg0.service", expired),
            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete)
        );
        assert_eq!(fake.calls, ["connect"]);
    }

    #[test]
    fn result_mapping_accepts_only_the_documented_done_value() {
        assert_eq!(map_job_result("done"), JobResult::Done);
        for value in ["failed", "dependency", "timeout", "canceled", "skipped"] {
            assert_ne!(map_job_result(value), JobResult::Done);
        }
        assert_eq!(map_job_result("future-result"), JobResult::Other);
    }

    #[test]
    fn std_wake_park_driver_rejects_an_already_expired_future() {
        let expired = RestartDeadline::new(Instant::now(), Duration::ZERO);
        assert_eq!(
            drive(std::future::pending::<()>(), expired),
            Err(RestartFailure::WaitTimedOutMayStillComplete)
        );
    }

    #[test]
    fn owner_change_and_disconnect_fail_the_attempt() {
        for (event, failure) in [
            (
                ServiceEvent::ManagerOwnerChanged,
                RestartFailure::ManagerOwnerChanged,
            ),
            (ServiceEvent::Disconnected, RestartFailure::BusDisconnected),
        ] {
            let mut fake = FakeBus {
                job_path: "/job/actual",
                primary_unit: "wg-quick@wg0.service",
                events: VecDeque::from([event]),
                ..FakeBus::default()
            };
            assert_eq!(
                restart_with_transport(&mut fake, "wg-quick@wg0.service", deadline()),
                RestartOutcome::Failed(failure)
            );
        }
    }
}