#[allow(clippy::missing_panics_doc, clippy::must_use_candidate)]
pub mod apogee_core_v1 {
    hyprwire::include_protocol!("apogee_core_v1");
    pub use server::*;
    pub use spec::*;
}

mod condition;
mod config;
mod ecs;
mod environment;
mod logging;
mod notify;
mod unit;

use anyhow::Context as _;
use apogee_core_v1::{apogee_manager_v1, apogee_reload_v1};
use bevy_ecs::message;
use bevy_ecs::prelude::*;
use bevy_ecs::system::SystemParam;
use calloop::generic;
use clap::Parser;
use hyprwire::server;
use std::collections::{BTreeMap, HashSet};
use std::os::fd::AsRawFd;
use std::{fs, io, path, process, sync};
use unit::{service, task};

#[derive(SystemSet, Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum ReconcileSet {
    Input,
    Removal,
    Lifecycle,
    TaskLifecycle,
    Sync,
}

#[derive(Parser)]
struct Args {
    #[arg(long)]
    config_path: path::PathBuf,
    #[arg(long, default_value_t = log::LevelFilter::Info)]
    log_level: log::LevelFilter,
}

struct Apogee {
    world: World,
}

impl hyprwire::Handler<apogee_core_v1::ApogeeCoreV1> for Apogee {
    fn protocol_filter(&mut self, client: server::ServerClient) -> bool {
        client.creds().uid.is_root() || client.creds().uid == rustix::process::getuid()
    }

    fn bind(&mut self, object: apogee_manager_v1::ApogeeManagerV1) {
        for name in self.world.resource::<ecs::ServiceRegistry>().by_name.keys() {
            object.send_service_added(name);
        }
        for name in self.world.resource::<ecs::TaskRegistry>().by_name.keys() {
            object.send_task_added(name);
        }
        for name in self
            .world
            .resource::<ecs::ConditionRegistry>()
            .by_name
            .keys()
        {
            object.send_condition_added(name);
        }
        object.send_done();

        self.world.resource_mut::<ecs::Managers>().0.push(object);
    }
}

impl hyprwire::Dispatch<apogee_manager_v1::ApogeeManagerV1> for Apogee {
    #[allow(clippy::panic)]
    fn event(
        &mut self,
        _object: &apogee_manager_v1::ApogeeManagerV1,
        event: <apogee_manager_v1::ApogeeManagerV1 as hyprwire::Object>::Event<'_>,
    ) {
        match event {
            apogee_manager_v1::Event::Reload { apogee_reload_v1 } => {
                self.world
                    .trigger(ecs::ReloadRequested(Some(apogee_reload_v1)));
            }
            apogee_manager_v1::Event::GetService {
                apogee_service_v1,
                name,
            } => {
                let service = self
                    .world
                    .resource::<ecs::ServiceRegistry>()
                    .by_name
                    .get(name.as_str())
                    .copied();
                if let Some(service) = service {
                    self.world
                        .entity_mut(service)
                        .get_mut::<service::Objects>()
                        .unwrap_or_else(|| {
                            panic!("service {service:?} is missing its object bindings")
                        })
                        .0
                        .push(apogee_service_v1.clone());
                    self.world
                        .resource_mut::<ecs::ServiceRegistry>()
                        .by_object
                        .insert(apogee_service_v1, service);
                } else {
                    apogee_service_v1.send_destroyed();
                }
            }
            apogee_manager_v1::Event::GetCondition {
                apogee_condition_v1,
                name,
            } => {
                let condition = self
                    .world
                    .resource::<ecs::ConditionRegistry>()
                    .by_name
                    .get(name.as_str())
                    .copied();
                if let Some(condition) = condition {
                    self.world
                        .entity_mut(condition)
                        .get_mut::<condition::Objects>()
                        .unwrap_or_else(|| {
                            panic!("condition {condition:?} is missing its object bindings")
                        })
                        .0
                        .push(apogee_condition_v1.clone());
                    self.world
                        .resource_mut::<ecs::ConditionRegistry>()
                        .by_object
                        .insert(apogee_condition_v1, condition);
                } else {
                    apogee_condition_v1.send_destroyed();
                }
            }
            apogee_manager_v1::Event::GetTask {
                apogee_task_v1,
                name,
            } => {
                let task = self
                    .world
                    .resource::<ecs::TaskRegistry>()
                    .by_name
                    .get(name.as_str())
                    .copied();
                if let Some(task) = task {
                    self.world
                        .entity_mut(task)
                        .get_mut::<task::Objects>()
                        .unwrap_or_else(|| {
                            panic!("condition {task:?} is missing its object bindings")
                        })
                        .0
                        .push(apogee_task_v1.clone());
                    self.world
                        .resource_mut::<ecs::TaskRegistry>()
                        .by_object
                        .insert(apogee_task_v1, task);
                } else {
                    apogee_task_v1.send_destroyed();
                }
            }
            apogee_manager_v1::Event::GetEnvironment {
                apogee_environment_v1,
            } => {
                let environment = &self.world.resource::<environment::Environment>().0;
                for (name, value) in environment {
                    let name = name.to_string_lossy();
                    let value = value.to_string_lossy();
                    apogee_environment_v1.send_getenv(name.as_ref(), value.as_ref());
                }
                apogee_environment_v1.send_can_modify(1);
                apogee_environment_v1.send_done();
            }
        }
    }
}

fn spawn_process(
    loop_handle: &ecs::LoopHandle,
    global_environment: &environment::Environment,
    service_environment: &BTreeMap<String, String>,
    service: Entity,
    service_name: sync::Arc<str>,
    command: &str,
    args: &[String],
    notify: &mut dyn notify::Notify,
) -> anyhow::Result<unit::Process> {
    let mut cmd = process::Command::new(command);
    cmd.args(args)
        .env_clear()
        .envs(&global_environment.0)
        .envs(service_environment)
        .stdout(process::Stdio::piped())
        .stderr(process::Stdio::piped());
    let notify_token = notify.configure_command(loop_handle, service, &mut cmd)?;

    let mut child = match cmd.spawn() {
        Ok(child) => child,
        Err(error) => {
            if let Some(token) = notify_token {
                loop_handle.0.remove(token);
            }
            return Err(error).with_context(|| format!("failed to spawn {command}"));
        }
    };
    let id = unit::ProcessId(child.id());

    let stdout = child.stdout.take().context("child stdout was not piped")?;
    let flags = rustix::fs::fcntl_getfl(&stdout)?;
    rustix::fs::fcntl_setfl(&stdout, flags | rustix::fs::OFlags::NONBLOCK)?;

    let stderr = child.stderr.take().context("child stderr was not piped")?;
    let flags = rustix::fs::fcntl_getfl(&stderr)?;
    rustix::fs::fcntl_setfl(&stderr, flags | rustix::fs::OFlags::NONBLOCK)?;

    let stdout_token = register_service_output(
        loop_handle,
        stdout,
        sync::Arc::clone(&service_name),
        id,
        log::Level::Info,
        apogee_core_v1::LogStream::Stdout,
    )?;
    let stderr_token = match register_service_output(
        loop_handle,
        stderr,
        sync::Arc::clone(&service_name),
        id,
        log::Level::Warn,
        apogee_core_v1::LogStream::Stderr,
    ) {
        Ok(token) => token,
        Err(error) => {
            loop_handle.0.remove(stdout_token);
            return Err(error.context("failed to register child stderr"));
        }
    };

    let setup = || -> anyhow::Result<calloop::RegistrationToken> {
        let pidfd = rustix::process::pidfd_open(
            rustix::process::Pid::from_child(&child),
            rustix::process::PidfdFlags::empty(),
        )
        .context("failed to open pidfd")?;

        loop_handle
            .0
            .insert_source(
                generic::Generic::new(pidfd, calloop::Interest::READ, calloop::Mode::Level),
                move |_, _, apogee| {
                    if apogee.world.get::<task::Task>(service).is_some() {
                        apogee.world.trigger(task::Exited {
                            task: service,
                            process: id,
                            reason: unit::StopReason::Exited,
                        });
                    } else {
                        apogee.world.trigger(service::Exited {
                            service,
                            process: id,
                            reason: unit::StopReason::Exited,
                        });
                    }
                    apogee.world.flush();
                    Ok(calloop::PostAction::Remove)
                },
            )
            .map_err(|error| error.error)
            .context("failed to register pidfd")
    };

    match setup() {
        Ok(token) => Ok(unit::Process {
            id,
            child,
            token,
            notify_token,
        }),
        Err(error) => {
            if let Some(token) = notify_token {
                loop_handle.0.remove(token);
            }
            loop_handle.0.remove(stdout_token);
            loop_handle.0.remove(stderr_token);
            if let Err(kill_error) = child.kill() {
                log::debug!("process may already have exited after setup failed: {kill_error}");
            }
            if let Err(wait_error) = child.wait() {
                log::warn!("failed to reap process after setup failed: {wait_error}");
            }
            Err(error)
        }
    }
}

fn register_service_output(
    loop_handle: &ecs::LoopHandle,
    reader: impl std::os::fd::AsFd + io::Read + 'static,
    service_name: sync::Arc<str>,
    process: unit::ProcessId,
    level: log::Level,
    stream: apogee_core_v1::LogStream,
) -> anyhow::Result<calloop::RegistrationToken> {
    let mut buffer = Vec::new();
    let mut logger = logging::ServiceLogger::new(&service_name, process.0)
        .map_err(|error| anyhow::anyhow!(error))?;
    loop_handle
        .0
        .insert_source(
            generic::Generic::new(reader, calloop::Interest::READ, calloop::Mode::Level),
            move |_, reader, state| {
                let reader = unsafe { reader.get_mut() };
                let mut chunk = [0; 4096];
                loop {
                    match reader.read(&mut chunk) {
                        Ok(0) => {
                            if !buffer.is_empty() {
                                logger.emit(level, message(&buffer));
                            }
                            return Ok(calloop::PostAction::Remove);
                        }
                        Ok(length) => {
                            buffer.extend_from_slice(&chunk[..length]);
                            while let Some(position) = buffer.iter().position(|byte| *byte == b'\n')
                            {
                                let line = buffer.drain(..=position).collect::<Vec<_>>();
                                let line: sync::Arc<str> = message(&line).into();
                                logger.emit(level, &line);
                                state.world.write_message(unit::ServiceOutput {
                                    stream,
                                    process,
                                    line,
                                });
                            }
                        }
                        Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                            return Ok(calloop::PostAction::Continue);
                        }
                        Err(error) => {
                            log::warn!(
                                "{}[{}]: failed reading service output: {error}",
                                service_name,
                                process.0
                            );
                            return Ok(calloop::PostAction::Remove);
                        }
                    }
                }
            },
        )
        .map_err(|error| anyhow::anyhow!(error.error))
        .context("failed to register child stdout/stderr")
}

fn message(bytes: &[u8]) -> String {
    String::from_utf8_lossy(bytes)
        .trim_end_matches(['\r', '\n'])
        .to_owned()
}

fn handle_restart_requested(
    restart: On<ecs::RestartRequested>,
    mut commands: Commands,
    mut services: Query<
        (
            &unit::Name,
            &unit::Command,
            &unit::Args,
            &unit::Conditions,
            &service::Ready,
            &unit::Running,
            &unit::Environment,
            &apogee_core_v1::ServiceState,
            &mut service::RestartCount,
            &config::Notify,
        ),
        (With<service::Service>, Without<unit::Process>),
    >,
    fulfilled: Query<(), With<condition::Fulfilled>>,
    loop_handle: NonSend<ecs::LoopHandle>,
    global_env: Res<environment::Environment>,
    mut systemd_notify: ResMut<notify::Systemd>,
) {
    let Ok((
        name,
        command,
        args,
        conditions,
        ready,
        running,
        environment,
        state,
        mut restart_count,
        notify,
    )) = services.get_mut(restart.0)
    else {
        return;
    };

    if !matches!(*state, apogee_core_v1::ServiceState::Restarting) {
        return;
    }

    if !conditions
        .0
        .iter()
        .all(|condition| fulfilled.contains(*condition))
    {
        commands
            .entity(restart.0)
            .insert(apogee_core_v1::ServiceState::Waiting);
        return;
    }

    log::info!(
        "restarting service {:?}: {} ({} {:?})",
        restart.0,
        name.0,
        command.0,
        args.0
    );

    let notifier: &mut dyn notify::Notify = match notify {
        config::Notify::None => &mut notify::Dummy,
        config::Notify::Systemd => &mut *systemd_notify,
        config::Notify::S6 => &mut notify::S6,
    };
    match spawn_process(
        &loop_handle,
        &global_env,
        &environment.0,
        restart.0,
        sync::Arc::clone(&name.0),
        &command.0,
        &args.0,
        notifier,
    ) {
        Ok(process) => {
            restart_count.0 = restart_count.0.saturating_add(1);
            log::info!("Starting {}[{}]", name.0, process.id.0);
            commands.entity(restart.0).insert((
                process,
                unit::Timestamp(unit::unix_timestamp()),
                apogee_core_v1::ServiceState::Running,
            ));
            if *notify == config::Notify::None {
                commands.entity(ready.0).insert(condition::Fulfilled);
            }
            commands.entity(running.0).insert(condition::Fulfilled);
        }
        Err(error) => {
            log::error!("failed to restart {}: {error:#}", name.0);
            commands
                .entity(restart.0)
                .insert(apogee_core_v1::ServiceState::Failed);
        }
    }
}

#[derive(SystemParam)]
struct UnitParams<'w, 's> {
    commands: Query<'w, 's, &'static unit::Command>,
    descriptions: Query<'w, 's, &'static unit::Description>,
    args: Query<'w, 's, &'static unit::Args>,
    environments: Query<'w, 's, &'static unit::Environment>,
    conditions: Query<'w, 's, &'static unit::Conditions>,
    running: Query<'w, 's, &'static unit::Running>,
}

#[derive(SystemParam)]
struct ServiceParams<'w, 's> {
    ready: Query<'w, 's, &'static service::Ready>,
    restart_delays: Query<'w, 's, &'static service::RestartDelay>,
    restart_policies: Query<'w, 's, &'static apogee_core_v1::RestartPolicy>,
    notifies: Query<'w, 's, &'static config::Notify>,
    existing: Query<'w, 's, (Entity, &'static unit::Name), With<service::Service>>,
}

#[derive(SystemParam)]
struct TaskParams<'w, 's> {
    successful: Query<'w, 's, &'static task::Successful>,
    existing: Query<'w, 's, (Entity, &'static unit::Name), With<task::Task>>,
}

fn handle_reload(
    reload: On<ecs::ReloadRequested>,
    mut commands: Commands,
    unit: UnitParams,
    service: ServiceParams,
    task: TaskParams,
    config: Res<config::Config>,
    mut services: ResMut<ecs::ServiceRegistry>,
    mut conditions: ResMut<ecs::ConditionRegistry>,
    mut tasks: ResMut<ecs::TaskRegistry>,
    managers: Res<ecs::Managers>,
) {
    let UnitParams {
        commands: unit_commands,
        descriptions: unit_descriptions,
        args: unit_args,
        environments: unit_environments,
        conditions: unit_conditions,
        running: unit_running,
    } = unit;
    let ServiceParams {
        ready: service_ready,
        restart_delays: service_restart_delays,
        restart_policies: service_restart_policies,
        notifies: service_notifies,
        existing: existing_services,
    } = service;
    let TaskParams {
        successful: task_successful,
        existing: existing_tasks,
    } = task;

    match config.load() {
        Ok(config) => {
            for configured_condition in config.conditions {
                condition::resolve(
                    &mut commands,
                    &mut conditions,
                    &managers,
                    configured_condition.name,
                    configured_condition.immutable,
                );
            }

            let desired: HashSet<_> = config
                .services
                .iter()
                .map(|service| service.name.as_ref())
                .collect();
            for (entity, name) in existing_services.iter() {
                if !desired.contains(&*name.0) {
                    commands.entity(entity).remove::<service::Service>();
                }
            }

            for service in config.services {
                let resolved_conditions = service
                    .conditions
                    .iter()
                    .map(|condition| {
                        condition::resolve(
                            &mut commands,
                            &mut conditions,
                            &managers,
                            sync::Arc::clone(&condition.name),
                            condition.immutable,
                        )
                    })
                    .collect::<Vec<_>>();
                let ready = condition::resolve(
                    &mut commands,
                    &mut conditions,
                    &managers,
                    service.ready.name,
                    service.ready.immutable,
                );
                let running = condition::resolve(
                    &mut commands,
                    &mut conditions,
                    &managers,
                    service.running.name,
                    service.running.immutable,
                );

                if let Some(entity) = services.by_name.get(&service.name) {
                    if unit_descriptions
                        .get(*entity)
                        .is_ok_and(|current| current.0 != service.description)
                    {
                        commands
                            .entity(*entity)
                            .insert(unit::Description(service.description));
                    }
                    if unit_commands
                        .get(*entity)
                        .is_ok_and(|current| current.0.as_str() != service.command)
                    {
                        commands
                            .entity(*entity)
                            .insert(unit::Command(service.command));
                    }
                    if unit_args
                        .get(*entity)
                        .is_ok_and(|current| current.0 != service.args)
                    {
                        commands.entity(*entity).insert(unit::Args(service.args));
                    }
                    if unit_environments
                        .get(*entity)
                        .is_ok_and(|current| current.0 != service.env)
                    {
                        commands
                            .entity(*entity)
                            .insert(unit::Environment(service.env));
                    }
                    if unit_conditions
                        .get(*entity)
                        .is_ok_and(|current| current.0 != resolved_conditions)
                    {
                        commands
                            .entity(*entity)
                            .insert(unit::Conditions(resolved_conditions));
                    }
                    if service_ready
                        .get(*entity)
                        .is_ok_and(|current| current.0 != ready)
                    {
                        commands.entity(*entity).insert(service::Ready(ready));
                    }
                    if unit_running
                        .get(*entity)
                        .is_ok_and(|current| current.0 != running)
                    {
                        commands.entity(*entity).insert(unit::Running(running));
                    }
                    if service_restart_delays
                        .get(*entity)
                        .is_ok_and(|current| current.0 != service.restart.delay)
                    {
                        commands
                            .entity(*entity)
                            .insert(service::RestartDelay(service.restart.delay));
                    }
                    if service_restart_policies
                        .get(*entity)
                        .is_ok_and(|current| *current != service.restart.on)
                    {
                        commands.entity(*entity).insert(service.restart.on);
                    }
                    if service_notifies.get(*entity) != Ok(&service.notify) {
                        commands.entity(*entity).insert(service.notify);
                    }
                } else {
                    service::spawn(
                        &mut commands,
                        &mut services,
                        &managers,
                        service.name,
                        service.description,
                        service.command,
                        service.args,
                        resolved_conditions,
                        ready,
                        running,
                        service.env,
                        service.restart.delay,
                        service.restart.on,
                        service.notify,
                    );
                }
            }

            let desired: HashSet<_> = config.tasks.iter().map(|task| task.name.as_ref()).collect();
            for (entity, name) in existing_tasks.iter() {
                if !desired.contains(&*name.0) {
                    commands.entity(entity).remove::<task::Task>();
                }
            }

            for task in config.tasks {
                let resolved_conditions = task
                    .conditions
                    .iter()
                    .map(|condition| {
                        condition::resolve(
                            &mut commands,
                            &mut conditions,
                            &managers,
                            sync::Arc::clone(&condition.name),
                            condition.immutable,
                        )
                    })
                    .collect::<Vec<_>>();
                let successful = condition::resolve(
                    &mut commands,
                    &mut conditions,
                    &managers,
                    task.successful.name,
                    task.successful.immutable,
                );
                let running = condition::resolve(
                    &mut commands,
                    &mut conditions,
                    &managers,
                    task.running.name,
                    task.running.immutable,
                );

                if let Some(entity) = tasks.by_name.get(&task.name) {
                    if unit_descriptions
                        .get(*entity)
                        .is_ok_and(|current| current.0 != task.description)
                    {
                        commands
                            .entity(*entity)
                            .insert(unit::Description(task.description));
                    }
                    if unit_commands
                        .get(*entity)
                        .is_ok_and(|current| current.0.as_str() != task.command)
                    {
                        commands.entity(*entity).insert(unit::Command(task.command));
                    }
                    if unit_args
                        .get(*entity)
                        .is_ok_and(|current| current.0 != task.args)
                    {
                        commands.entity(*entity).insert(unit::Args(task.args));
                    }
                    if unit_environments
                        .get(*entity)
                        .is_ok_and(|current| current.0 != task.env)
                    {
                        commands.entity(*entity).insert(unit::Environment(task.env));
                    }
                    if unit_conditions
                        .get(*entity)
                        .is_ok_and(|current| current.0 != resolved_conditions)
                    {
                        commands
                            .entity(*entity)
                            .insert(unit::Conditions(resolved_conditions));
                    }
                    if task_successful
                        .get(*entity)
                        .is_ok_and(|current| current.0 != successful)
                    {
                        commands
                            .entity(*entity)
                            .insert(task::Successful(successful));
                    }
                    if unit_running
                        .get(*entity)
                        .is_ok_and(|current| current.0 != running)
                    {
                        commands.entity(*entity).insert(unit::Running(running));
                    }
                } else {
                    task::spawn(
                        &mut commands,
                        &mut tasks,
                        &managers,
                        task.name,
                        task.description,
                        task.command,
                        task.args,
                        resolved_conditions,
                        successful,
                        running,
                        task.env,
                    );
                }
            }

            if let Some(object) = reload.0.as_ref() {
                object.send_finished();
            }
        }
        Err(e) => {
            log::error!("failed to load service configuration: {e:#}");
            if let Some(object) = reload.0.as_ref() {
                object.send_failed(apogee_core_v1::ReloadError::Io, e.to_string());
            }
        }
    }
}

fn service_modified(
    mut commands: Commands,
    mut services: Query<
        (
            Entity,
            &unit::Name,
            &unit::Command,
            &unit::Args,
            &unit::Conditions,
            Ref<service::Service>,
            Option<&mut unit::Process>,
            &service::Ready,
            &unit::Running,
            &unit::Environment,
            &config::Notify,
        ),
        Or<(
            Changed<unit::Command>,
            Changed<unit::Args>,
            Changed<unit::Environment>,
            Changed<unit::Conditions>,
            Changed<service::Ready>,
            Changed<unit::Running>,
            Changed<config::Notify>,
        )>,
    >,
    fulfilled: Query<(), With<condition::Fulfilled>>,
    loop_handle: NonSend<ecs::LoopHandle>,
    global_env: Res<environment::Environment>,
    mut systemd_notify: ResMut<notify::Systemd>,
) {
    for (
        entity,
        name,
        command,
        args,
        conditions,
        service,
        process,
        ready,
        running,
        environment,
        notify,
    ) in services.iter_mut()
    {
        if service.is_added() {
            continue;
        }

        log::info!(
            "restarting service {entity:?}: {} ({} {:?})",
            name.0,
            command.0,
            args.0
        );

        if let Some(mut process) = process {
            process.remove_sources(&loop_handle);
            log::info!("Stopping {}[{}], sending SIGKILL ...", name.0, process.id.0);
            if let Err(error) = process.child.kill() {
                log::debug!("old process may already have exited: {error}");
            }

            let event = service::Exited {
                service: entity,
                process: process.id,
                reason: unit::StopReason::Restarted,
            };
            commands.queue(move |world: &mut World| {
                world.trigger(event);
                world.flush();
            });
        }

        if conditions
            .0
            .iter()
            .all(|condition| fulfilled.contains(*condition))
        {
            let notifier: &mut dyn notify::Notify = match notify {
                config::Notify::None => &mut notify::Dummy,
                config::Notify::Systemd => &mut *systemd_notify,
                config::Notify::S6 => &mut notify::S6,
            };
            match spawn_process(
                &loop_handle,
                &global_env,
                &environment.0,
                entity,
                sync::Arc::clone(&name.0),
                &command.0,
                &args.0,
                notifier,
            ) {
                Ok(process) => {
                    commands.entity(entity).insert((
                        process,
                        unit::Timestamp(unit::unix_timestamp()),
                        apogee_core_v1::ServiceState::Running,
                    ));

                    log::info!("restarted service: {}", name.0);
                    if *notify == config::Notify::None {
                        commands.entity(ready.0).insert(condition::Fulfilled);
                    }
                    commands.entity(running.0).insert(condition::Fulfilled);
                }
                Err(error) => {
                    log::error!("failed to restart {}: {error:#}", name.0);
                    commands
                        .entity(entity)
                        .insert(apogee_core_v1::ServiceState::Failed);
                }
            }
        } else {
            commands
                .entity(entity)
                .insert(apogee_core_v1::ServiceState::Waiting);
        }
    }
}

fn start_services(
    mut commands: Commands,
    added_services: Query<(), Added<service::Service>>,
    added_conditions: Query<(), Added<condition::Fulfilled>>,
    services: Query<
        (
            Entity,
            &unit::Name,
            &unit::Command,
            &unit::Args,
            &unit::Conditions,
            &service::Ready,
            &unit::Running,
            &unit::Environment,
            &config::Notify,
        ),
        (With<service::Service>, Without<unit::Process>),
    >,
    fulfilled: Query<(), With<condition::Fulfilled>>,
    loop_handle: NonSend<ecs::LoopHandle>,
    loop_signal: Res<ecs::LoopSignal>,
    global_env: Res<environment::Environment>,
    mut systemd_notify: ResMut<notify::Systemd>,
) {
    if added_services.is_empty() && added_conditions.is_empty() {
        return;
    }

    let any_started = services
        .iter()
        .filter(|(_, _, _, _, conditions, _, _, _, _)| {
            conditions
                .0
                .iter()
                .all(|condition| fulfilled.contains(*condition))
        })
        .fold(
            false,
            |any_started, (entity, name, command, args, _, ready, running, environment, notify)| {
                let notifier: &mut dyn notify::Notify = match notify {
                    config::Notify::None => &mut notify::Dummy,
                    config::Notify::Systemd => &mut *systemd_notify,
                    config::Notify::S6 => &mut notify::S6,
                };
                (match spawn_process(
                    &loop_handle,
                    &global_env,
                    &environment.0,
                    entity,
                    sync::Arc::clone(&name.0),
                    &command.0,
                    &args.0,
                    notifier,
                ) {
                    Ok(process) => {
                        log::info!("Starting {}[{}]", name.0, process.id.0);
                        commands.entity(entity).insert((
                            process,
                            unit::Timestamp(unit::unix_timestamp()),
                            apogee_core_v1::ServiceState::Running,
                        ));
                        if *notify == config::Notify::None {
                            commands.entity(ready.0).insert(condition::Fulfilled);
                        }
                        commands.entity(running.0).insert(condition::Fulfilled);
                        true
                    }
                    Err(error) => {
                        log::error!("failed to start {}: {error:#}", name.0);
                        commands
                            .entity(entity)
                            .insert(apogee_core_v1::ServiceState::Failed);
                        false
                    }
                }) || any_started
            },
        );

    if any_started {
        loop_signal.0.wakeup();
    }
}

fn start_tasks(
    mut commands: Commands,
    added_tasks: Query<(), Added<task::Task>>,
    added_conditions: Query<(), Added<condition::Fulfilled>>,
    tasks: Query<
        (
            Entity,
            &unit::Name,
            &unit::Command,
            &unit::Args,
            &unit::Conditions,
            &unit::Running,
            &unit::Environment,
            &apogee_core_v1::TaskState,
        ),
        (With<task::Task>, Without<unit::Process>),
    >,
    fulfilled: Query<(), With<condition::Fulfilled>>,
    loop_handle: NonSend<ecs::LoopHandle>,
    loop_signal: Res<ecs::LoopSignal>,
    global_env: Res<environment::Environment>,
) {
    if added_tasks.is_empty() && added_conditions.is_empty() {
        return;
    }

    let any_started = tasks
        .iter()
        .filter(|(_, _, _, _, conditions, _, _, state)| {
            matches!(state, apogee_core_v1::TaskState::Waiting)
                && conditions
                    .0
                    .iter()
                    .all(|condition| fulfilled.contains(*condition))
        })
        .fold(
            false,
            |any_started, (entity, name, command, args, _, running, environment, _)| {
                (match crate::spawn_process(
                    &loop_handle,
                    &global_env,
                    &environment.0,
                    entity,
                    sync::Arc::clone(&name.0),
                    &command.0,
                    &args.0,
                    &mut notify::Dummy,
                ) {
                    Ok(process) => {
                        log::info!("Starting task {}[{}]", name.0, process.id.0);
                        commands.entity(entity).insert((
                            process,
                            unit::Timestamp(unit::unix_timestamp()),
                            apogee_core_v1::TaskState::Running,
                        ));
                        commands.entity(running.0).insert(condition::Fulfilled);
                        true
                    }
                    Err(error) => {
                        log::error!("failed to start task {}: {error:#}", name.0);
                        commands
                            .entity(entity)
                            .insert(apogee_core_v1::TaskState::Failed);
                        false
                    }
                }) || any_started
            },
        );

    if any_started {
        loop_signal.0.wakeup();
    }
}

fn sync_managers(mut managers: ResMut<ecs::Managers>) {
    managers.0.retain(|manager| manager.try_client().is_some());
    for manager in &managers.0 {
        manager.send_done();
    }
}

fn main() -> anyhow::Result<()> {
    let args = Args::parse();

    logging::init_syslog(syslog::Facility::LOG_DAEMON, args.log_level)?;

    let mut event_loop = calloop::EventLoop::try_new()?;

    let mut world = World::new();
    message::MessageRegistry::register_message::<unit::ServiceOutput>(&mut world);
    let mut schedule = Schedule::new(ecs::Reconcile);
    schedule.add_systems((
        (message::message_update_system, unit::store_service_output)
            .chain()
            .in_set(ReconcileSet::Input),
        (service::removed, task::removed)
            .after(ReconcileSet::Input)
            .in_set(ReconcileSet::Removal),
        (start_services, service_modified)
            .after(ReconcileSet::Removal)
            .in_set(ReconcileSet::Lifecycle),
        start_tasks
            .after(ReconcileSet::Lifecycle)
            .in_set(ReconcileSet::TaskLifecycle),
        (unit::sync_objects, condition::sync_objects)
            .after(ReconcileSet::TaskLifecycle)
            .in_set(ReconcileSet::Sync),
        sync_managers.after(ReconcileSet::Sync),
    ));
    world.init_resource::<ecs::ServiceRegistry>();
    world.init_resource::<ecs::ConditionRegistry>();
    world.init_resource::<ecs::TaskRegistry>();
    world.init_resource::<ecs::Managers>();
    world.init_resource::<notify::Systemd>();
    world.init_resource::<environment::Environment>();
    world.insert_resource(config::Config(args.config_path));
    world.insert_non_send(ecs::LoopHandle(event_loop.handle()));
    world.insert_resource(ecs::LoopSignal(event_loop.get_signal()));

    world.add_observer(handle_reload);
    world.add_observer(service::handle_exited);
    world.add_observer(task::handle_exited);
    world.add_observer(handle_restart_requested);
    world.add_observer(condition::handle_start);
    world.add_observer(condition::handle_stop);
    world.add_observer(service::handle_start);
    world.add_observer(service::handle_stop);
    world.add_observer(service::handle_restart);
    world.add_observer(task::handle_start);
    world.add_observer(task::handle_stop);
    world.add_observer(task::handle_restart);
    world.add_observer(unit::handle_new_log_object);
    world.flush();

    world.trigger(ecs::ReloadRequested(None));

    schedule.run(&mut world);

    let mut apogee = Apogee { world };

    let runtime_dir = dirs::runtime_dir()
        .unwrap_or_else(|| path::PathBuf::from("run"))
        .join("apogee");
    fs::create_dir_all(&runtime_dir)?;

    let socket_path = runtime_dir.join("apogee.sock");
    let mut socket = server::Server::bind(&socket_path)?;

    socket.add_implementation::<apogee_core_v1::ApogeeCoreV1, _>(&mut apogee, 1);

    let fd_wrapper = unsafe { generic::FdWrapper::new(socket.extract_loop_fd().as_raw_fd()) };
    let source = generic::Generic::new(
        fd_wrapper,
        calloop::Interest {
            readable: true,
            writable: false,
        },
        calloop::Mode::Level,
    );
    event_loop
        .handle()
        .insert_source(source, move |_, _, state| {
            _ = socket.dispatch_events(state, false);
            Ok(calloop::PostAction::Continue)
        })
        .map_err(|err| anyhow::anyhow!("failed to insert VT watcher: {err:?}"))?;

    event_loop.run(None, &mut apogee, |state| {
        schedule.run(&mut state.world);
    })?;

    Ok(())
}

hyprwire::delegate_noop!(Apogee: apogee_reload_v1::ApogeeReloadV1);