frame-core 0.3.0

use beamr::atom::Atom;
use beamr::ets::{OwnedTerm, copy_term_to_ets};
use beamr::process::heap::Heap;
use beamr::scheduler::Scheduler;
use beamr::term::Term;
use beamr::term::boxed::{Tuple, write_tuple};

use crate::component::ComponentId;
use crate::error::RegistryError;

#[derive(Clone, Copy)]
pub(super) struct CommandTags {
    namespace: Atom,
    barrier: Atom,
    send: Atom,
    stop: Atom,
}

impl CommandTags {
    pub(super) fn new(scheduler: &Scheduler) -> Self {
        Self {
            namespace: scheduler.atom_table().intern("$frame.supervision.command"),
            barrier: scheduler.atom_table().intern("barrier"),
            send: scheduler.atom_table().intern("send"),
            stop: scheduler.atom_table().intern("stop"),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum HostCommand {
    Barrier,
    Send {
        pid: u64,
        message: i64,
        expects_reply: bool,
    },
    Stop,
}

pub(super) enum CommandDecode {
    Command(HostCommand),
    NotCommand,
    Malformed(String),
}

pub(super) fn send_host_command(
    id: ComponentId,
    scheduler: &Scheduler,
    supervisor: u64,
    tags: CommandTags,
    command: HostCommand,
) -> Result<(), RegistryError> {
    let command_name = command.name();
    let message =
        encode(command, tags).map_err(|detail| RegistryError::SupervisorProtocol { id, detail })?;
    scheduler
        .send_to_mailbox(supervisor, message)
        .map_err(|source| RegistryError::CommandDelivery {
            id,
            command: command_name,
            source,
        })
}

pub(super) fn decode(message: Term, tags: CommandTags) -> CommandDecode {
    let Some(tuple) = Tuple::new(message) else {
        return CommandDecode::NotCommand;
    };
    if tuple.get(0).and_then(Term::as_atom) != Some(tags.namespace) {
        return CommandDecode::NotCommand;
    }
    let Some(kind) = tuple.get(1).and_then(Term::as_atom) else {
        return CommandDecode::Malformed("command kind was not an atom".to_owned());
    };
    if kind == tags.barrier {
        return if tuple.arity() == 2 {
            CommandDecode::Command(HostCommand::Barrier)
        } else {
            CommandDecode::Malformed("barrier command had the wrong arity".to_owned())
        };
    }
    if kind == tags.stop {
        return if tuple.arity() == 2 {
            CommandDecode::Command(HostCommand::Stop)
        } else {
            CommandDecode::Malformed("stop command had the wrong arity".to_owned())
        };
    }
    if kind != tags.send || tuple.arity() != 5 {
        return CommandDecode::Malformed("unknown command kind or arity".to_owned());
    }
    match (
        tuple.get(2).and_then(Term::as_pid),
        tuple.get(3).and_then(Term::as_small_int),
        tuple.get(4).and_then(Term::as_atom),
    ) {
        (Some(pid), Some(message), Some(expects_reply))
            if expects_reply == Atom::TRUE || expects_reply == Atom::FALSE =>
        {
            CommandDecode::Command(HostCommand::Send {
                pid,
                message,
                expects_reply: expects_reply == Atom::TRUE,
            })
        }
        _ => CommandDecode::Malformed("send command carried malformed fields".to_owned()),
    }
}

fn encode(command: HostCommand, tags: CommandTags) -> Result<OwnedTerm, String> {
    let elements = match command {
        HostCommand::Barrier => vec![Term::atom(tags.namespace), Term::atom(tags.barrier)],
        HostCommand::Send {
            pid,
            message,
            expects_reply,
        } => vec![
            Term::atom(tags.namespace),
            Term::atom(tags.send),
            Term::pid(pid),
            Term::small_int(message),
            Term::atom(if expects_reply {
                Atom::TRUE
            } else {
                Atom::FALSE
            }),
        ],
        HostCommand::Stop => vec![Term::atom(tags.namespace), Term::atom(tags.stop)],
    };
    let mut heap = Heap::new(elements.len() + 1);
    let words = heap
        .alloc_slice(elements.len() + 1)
        .map_err(|error| format!("command tuple allocation failed: {error}"))?;
    let tuple = write_tuple(words, &elements)
        .ok_or_else(|| "command tuple construction failed".to_owned())?;
    copy_term_to_ets(tuple).map_err(|error| format!("command ownership failed: {error}"))
}

impl HostCommand {
    const fn name(self) -> &'static str {
        match self {
            Self::Barrier => "barrier",
            Self::Send { .. } => "send",
            Self::Stop => "stop",
        }
    }
}

#[cfg(test)]
mod tests {
    use beamr::module::ModuleRegistry;
    use beamr::scheduler::{SchedulerConfig, SchedulerServices};

    use super::*;

    fn scheduler() -> Result<Scheduler, Box<dyn std::error::Error>> {
        Ok(crate::composition::compose_scheduler(
            SchedulerConfig::default(),
            SchedulerServices::minimal(),
            std::sync::Arc::new(ModuleRegistry::new()),
        )?)
    }

    #[test]
    fn child_reply_and_exit_shapes_cannot_decode_as_commands()
    -> Result<(), Box<dyn std::error::Error>> {
        let scheduler = scheduler()?;
        let tags = CommandTags::new(&scheduler);
        assert!(matches!(
            decode(Term::small_int(0), tags),
            CommandDecode::NotCommand
        ));

        let mut heap = Heap::new(4);
        let words = heap.alloc_slice(4)?;
        let exit = write_tuple(
            words,
            &[
                Term::atom(Atom::EXIT),
                Term::pid(7),
                Term::atom(Atom::ERROR),
            ],
        )
        .ok_or_else(|| std::io::Error::other("EXIT tuple construction failed"))?;
        assert!(matches!(decode(exit, tags), CommandDecode::NotCommand));
        scheduler.shutdown();
        Ok(())
    }
}