rstango 0.4.0

//! Implementation of the DServer class.

use std::time::Duration;
use std::sync::Arc;

use tracing::{debug, warn};

use crate::error::{reason, Error, ErrorKind, Result};
use crate::value::{CommandDataKind, Scalar, Spectrum};
use crate::pool;

use super::polling::{self, Poller, PolledKind, MIN_PERIOD};
use super::def::{ClassDef, CommandDef};
use super::server::Server;
use super::dev::{AnyDev, Dev};


/// Class name of the admin class: DServer.
pub const CLASS_NAME: &str = "DServer";


/// The model of the admin device (`DServer`), which holds nothing: its initialization boots
/// the server, see [`Admin::init`].
pub struct Admin {

}

impl Admin {

    /// The definition of the admin class, which is the whole administration interface of
    /// a Tango server: every command below is one a client, `Astor` and `Jive` in
    /// particular, expects to find on `dserver/exe/inst` and nowhere else.
    /// 
    /// The commands every device has, `State`, `Status` and `Init`, are not among them:
    /// the server answers those itself, on the admin device as on any other.
    /// 
    /// TODO: Only the commands of the server itself are implemented: the others are
    /// declared with the signature a client expects and answer that they have no execution
    /// function until they are written.
    /// 
    /// TODO: The commands of cppTango that no specification has, and that are therefore
    /// left out until they are wanted: `QueryEventSystem`, `EnableEventSystemPerfMon` and
    /// the whole `*Telemetry*` family, all of them from cppTango 10.
    pub fn def() -> ClassDef<Self> {
        ClassDef::new_async(CLASS_NAME, Self::init)
            .doc("The administration device of that server, one per server, through which \
                its devices are restarted, polled, logged and subscribed to.")

            // --- The server itself --- //

            .command(CommandDef::new("DevRestart")
                .in_doc("Device name")
                .exec_async(Self::dev_restart))

            .command(CommandDef::new("RestartServer")
                .exec_async(Self::restart_server))

            .command(CommandDef::new("QueryClass")
                .out_doc("Device server class(es) list")
                .exec(Self::query_class))

            .command(CommandDef::new("QueryDevice")
                .out_doc("Device server device(s) list")
                .exec(Self::query_device))

            .command(command("QuerySubDevice", CommandDataKind::Void, CommandDataKind::StringArray)
                .out_doc("Device server sub device(s) list"))

            .command(CommandDef::new("Kill")
                .exec_async(Self::kill))

            // --- Polling --- //

            .command(CommandDef::new("PolledDevice")
                .out_doc("Polled device name list")
                .exec(Self::polled_device))

            .command(CommandDef::new("DevPollStatus")
                .in_doc("Device name")
                .out_doc("Device polling status")
                .exec(Self::dev_poll_status))

            .command(CommandDef::new("AddObjPolling")
                .in_doc("Long[0]=Upd period. Str[0]=Device name. Str[1]=Object type. \
                    Str[2]=Object name")
                .exec_async(Self::add_obj_polling))

            .command(CommandDef::new("UpdObjPollingPeriod")
                .in_doc("Long[0]=Upd period. Str[0]=Device name. Str[1]=Object type. \
                    Str[2]=Object name")
                .exec_async(Self::upd_obj_polling_period))

            .command(CommandDef::new("RemObjPolling")
                .in_doc("Str[0]=Device name. Str[1]=Object type. Str[2]=Object name")
                .exec_async(Self::rem_obj_polling))

            .command(CommandDef::new("StartPolling")
                .exec(Self::start_polling))

            .command(CommandDef::new("StopPolling")
                .exec(Self::stop_polling))

            // --- Logging --- //

            .command(command("AddLoggingTarget", CommandDataKind::StringArray, CommandDataKind::Void)
                .in_doc("Str[i]=Device-name. Str[i+1]=Target-type::Target-name"))

            .command(command("RemoveLoggingTarget", CommandDataKind::StringArray, CommandDataKind::Void)
                .in_doc("Str[i]=Device-name. Str[i+1]=Target-type::Target-name"))

            .command(command("GetLoggingTarget", CommandDataKind::String, CommandDataKind::StringArray)
                .in_doc("Device name")
                .out_doc("Logging target list"))

            .command(command("SetLoggingLevel", CommandDataKind::LongStringArray, CommandDataKind::Void)
                .in_doc("Long[i]=Logging Level. Str[i]=Device name."))

            .command(command("GetLoggingLevel", CommandDataKind::StringArray, CommandDataKind::LongStringArray)
                .in_doc("Device list")
                .out_doc("Long[i]=Logging Level. Str[i]=Device name."))

            .command(command("StartLogging", CommandDataKind::Void, CommandDataKind::Void))

            .command(command("StopLogging", CommandDataKind::Void, CommandDataKind::Void))

            // --- Events --- //

            .command(command("EventSubscriptionChange", CommandDataKind::StringArray, CommandDataKind::Long)
                .in_doc("Event consumer wants to subscribe to")
                .out_doc("Tango lib release"))

            .command(CommandDef::new("ZmqEventSubscriptionChange")
                .exec_async(Self::zmq_event_subscription_change)
                .in_doc("Event consumer wants to subscribe to.\n\
                    device name, attribute/pipe name, action (\"subscribe\"), event name, \
                    <Tango client IDL version>\n\
                    event name can take the following values:\n\
                    \x20   \"change\",\n\
                    \x20   \"alarm\",\n\
                    \x20   \"periodic\",\n\
                    \x20   \"archive\",\n\
                    \x20   \"user_event\",\n\
                    \x20   \"attr_conf\",\n\
                    \x20   \"data_ready\",\n\
                    \x20   \"intr_change\",\n\
                    \x20   \"pipe\"\n\
                    \"info\" can also be used as single parameter to retrieve information \
                    about the heartbeat and event pub endpoints.")
                .out_doc("Str[0] = Heartbeat pub endpoint - Str[1] = Event pub endpoint\n\
                    ...\n\
                    Str[n] = Alternate Heartbeat pub endpoint - Str[n+1] = Alternate Event \
                    pub endpoint\n\
                    Str[n+1] = event name used by this server as zmq topic to send events\n\
                    Str[n+2] = channel name used by this server to send heartbeat events\n\
                    Long[0] = Tango lib release - Long[1] = Device IDL release\n\
                    Long[2] = Subscriber HWM - Long[3] = Multicast rate\n\
                    Long[4] = Multicast IVL - Long[5] = ZMQ release"))

            .command(CommandDef::new("EventConfirmSubscription")
                .exec_async(Self::event_confirm_subscription)
                .in_doc("Str[0] = dev1 name, Str[1] = att1 name, Str[2] = event name, \
                    Str[3] = dev2 name, Str[4] = att2 name, Str[5] = event name,..."))

            // Only a server running without a database answers this itself; every other
            // one has its channel in the database for a client to read there.
            .command(command("QueryEventChannelIOR", CommandDataKind::Void, CommandDataKind::String)
                .out_doc("Device server event channel IOR"))

            // --- Locking --- //

            .command(command("LockDevice", CommandDataKind::LongStringArray, CommandDataKind::Void)
                .in_doc("Str[0] = Device name. Long[0] = Lock validity"))

            .command(command("UnLockDevice", CommandDataKind::LongStringArray, CommandDataKind::Long)
                .in_doc("Str[x] = Device name(s). Long[0] = Force flag")
                .out_doc("Device global lock counter"))

            .command(command("ReLockDevices", CommandDataKind::StringArray, CommandDataKind::Void)
                .in_doc("Device(s) name"))

            .command(command("DevLockStatus", CommandDataKind::String, CommandDataKind::LongStringArray)
                .in_doc("Device name")
                .out_doc("Device locking status"))

            // --- What a wizard asks a class of --- //

            .command(command("QueryWizardClassProperty", CommandDataKind::String, CommandDataKind::StringArray)
                .in_doc("Class name")
                .out_doc("Class property list (name - description and default value)"))

            .command(command("QueryWizardDevProperty", CommandDataKind::String, CommandDataKind::StringArray)
                .in_doc("Class name")
                .out_doc("Device property list (name - description and default value)"))
    }

    /// The initialization of the admin device is a bit special since it is itself 
    /// responsible for querying the database, if any, to find all devices and their 
    /// classes, and then dynamically add the devices to the server in order to boot
    /// them.
    /// 
    /// The devices added here are not initialized by this function: they are new devices
    /// of the server, and the synchronization that initialized this admin device picks
    /// them up in its next batch, see [`super::server::Server::sync_async`]. So the whole boot of a
    /// server is the initialization of its admin device, and the admin device is the only
    /// device a server needs to be given.
    async fn init(dev: &Dev<Self>) -> Result<Self> {

        // Failing here fails the admin device itself, which is the right place for it:
        // without that list the server has nothing to serve, and a client asking the admin
        // device what is wrong is told exactly that.
        add_listed_devs(dev.server()).await?;
        Ok(Self {})

    }

    /// Stop the server, see [`Server::stop_async`], from the pool: stopping deletes the admin
    /// device, which waits for this very request, so this request is answered first and the
    /// server stops on its own after it. Every failure of the stop is logged by the stop.
    async fn kill(&self, dev: &Dev<Self>, _: ()) -> Result<()> {
        let dev = dev.arc();
        drop(pool::spawn(move || {
            let _ = dev.server().stop();
            Ok(())
        }));
        Ok(())
    }

    /// Restart a device of this server: deleted and initialized again, unexported and
    /// exported again, which is what the device gets from its class anew, see
    /// [`Server::sync_async`]. The admin device is not among them, see
    /// [`Self::restart_server`].
    /// 
    /// The name may be given as a client knows it, with the database or the protocol in
    /// front of it, and options behind it.
    async fn dev_restart(&self, dev: &Dev<Self>, name: Scalar<str>) -> Result<()> {

        let server = dev.server();
        let name = local_dev_name(name.value());

        let target = server.any_dev(name)
            .filter(|target| !target.class().name().eq_ignore_ascii_case(CLASS_NAME))
            .ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_DeviceNotFound))
                .with_description(format!("Device {name:?} is not served by this server")))?;

        let (name, class) = (Arc::clone(target.name()), Arc::clone(target.class().name()));
        drop(target);

        server.remove_dev(&name);
        server.add_dev(&class, &name)?;
        server.sync_async().await

    }

    /// Restart every device of this server but the admin one: all removed, then added
    /// again from what the database lists now, which is how a server picks up the devices
    /// added to it or removed from it in the database.
    /// 
    /// The request is answered once the devices are restarted, which a client waiting for
    /// the restart to be done can rely on, so it takes as long as initializing every device
    /// does.
    async fn restart_server(&self, dev: &Dev<Self>, _: ()) -> Result<()> {
        let server = dev.server();
        for dev in server.any_devs() {
            if !dev.class().name().eq_ignore_ascii_case(CLASS_NAME) {
                server.remove_dev(dev.name());
            }
        }
        add_listed_devs(server).await?;
        server.sync_async().await
    }

    /// Have an event published, which is entirely up to the backend serving this server,
    /// see [`crate::backend::BackendServer::zmq_event_subscription_change`].
    async fn zmq_event_subscription_change(&self, dev: &Dev<Self>, args: Spectrum<str>) -> Result<(Spectrum<i32>, Spectrum<str>)> {
        dev.server().backend_checked()?.zmq_event_subscription_change(args).await
    }

    /// Keep the events subscribed to published, which is entirely up to the backend
    /// serving this server, see [`crate::backend::BackendServer::event_confirm_subscription`].
    async fn event_confirm_subscription(&self, dev: &Dev<Self>, args: Spectrum<str>) -> Result<()> {
        dev.server().backend_checked()?.event_confirm_subscription(args).await
    }

    // The polling of the devices of this server, see [`super::polling`]. A change of the
    // polling of a device is written to the database first, in its `polled_attr` or
    // `polled_cmd` property, so that the next initialization of the device finds it, and
    // a change the database refused is not made.

    /// The names of the devices of this server polled for anything.
    fn polled_device(&self, dev: &Dev<Self>, _: ()) -> Result<Vec<String>> {
        Ok(poller(dev.server())?.polled_devs().iter()
            .map(|name| name.to_string())
            .collect())
    }

    /// The polling status of a device of this server, one text per object polled, see
    /// [`polling::format_status`], none for a device that is not polled.
    fn dev_poll_status(&self, dev: &Dev<Self>, name: Scalar<str>) -> Result<Vec<String>> {
        let target = find_dev(dev.server(), name.value())?;
        Ok(poller(dev.server())?.status(target.name()).iter()
            .map(|status| polling::format_status(status, target.polled_history(status.kind, &status.name).as_ref()))
            .collect())
    }

    /// Poll an object of a device of this server, which must not be polled already.
    async fn add_obj_polling(&self, dev: &Dev<Self>, (longs, strs): (Spectrum<i32>, Spectrum<str>)) -> Result<()> {

        let server = dev.server();
        let poller = poller(server)?;
        let period = period_arg(&longs)?;
        let (target, kind, name) = polled_target(server, &strs)?;

        if poller.period(target.name(), kind, &name).is_some() {
            return Err(Error::new(ErrorKind::Reason(reason::API_AlreadyPolled))
                .with_description(format!("The {} {name:?} of device {:?} is already polled", 
                    kind.as_str(), target.name())));
        }

        store_polled(server, target.name(), kind, &name, Some(period)).await?;
        poller.add(&target, kind, name, period)

    }

    /// Poll an object of a device of this server at another period, which must be polled.
    async fn upd_obj_polling_period(&self, dev: &Dev<Self>, (longs, strs): (Spectrum<i32>, Spectrum<str>)) -> Result<()> {

        let server = dev.server();
        let poller = poller(server)?;
        let period = period_arg(&longs)?;
        let (target, kind, name) = polled_target(server, &strs)?;

        if poller.period(target.name(), kind, &name).is_none() {
            return Err(Error::new(ErrorKind::Reason(reason::API_PollObjNotFound))
                .with_description(format!("The {} {name:?} of device {:?} is not polled", 
                    kind.as_str(), target.name())));
        }

        store_polled(server, target.name(), kind, &name, Some(period)).await?;
        poller.set_period(target.name(), kind, &name, period)

    }

    /// Stop polling an object of a device of this server, which must be polled. An object
    /// its definition asks to be polled is polled again by the next initialization of the
    /// device, see [`super::polling`].
    async fn rem_obj_polling(&self, dev: &Dev<Self>, strs: Spectrum<str>) -> Result<()> {

        let server = dev.server();
        let poller = poller(server)?;
        let (target, kind, name) = polled_target(server, &strs)?;

        if poller.period(target.name(), kind, &name).is_none() {
            return Err(Error::new(ErrorKind::Reason(reason::API_PollObjNotFound))
                .with_description(format!("The {} {name:?} of device {:?} is not polled", 
                    kind.as_str(), target.name())));
        }

        store_polled(server, target.name(), kind, &name, None).await?;
        poller.remove(target.name(), kind, &name)

    }

    /// Poll again what is polled, after a [`Self::stop_polling`].
    fn start_polling(&self, dev: &Dev<Self>, _: ()) -> Result<()> {
        poller(dev.server())?.resume();
        Ok(())
    }

    /// Stop polling anything until [`Self::start_polling`], what is polled being kept,
    /// and so is the database.
    fn stop_polling(&self, dev: &Dev<Self>, _: ()) -> Result<()> {
        poller(dev.server())?.pause();
        Ok(())
    }

    /// The classes of this server, the admin one aside, in the order they were made.
    fn query_class(&self, dev: &Dev<Self>, _: ()) -> Result<Vec<String>> {
        Ok(dev.server().any_classes().iter()
            .filter(|class| !class.name().eq_ignore_ascii_case(CLASS_NAME))
            .map(|class| class.name().to_string())
            .collect())
    }

    /// The devices of this server, the admin one aside, each as `Class::name`, class by
    /// class.
    fn query_device(&self, dev: &Dev<Self>, _: ()) -> Result<Vec<String>> {
        Ok(dev.server().any_devs().iter()
            .filter(|dev| !dev.class().name().eq_ignore_ascii_case(CLASS_NAME))
            .map(|dev| format!("{}::{}", dev.class().name(), dev.name()))
            .collect())
    }

}

/// The name of a device of this server out of a name as a client gives it: what is behind
/// a `#`, the options of the name, is dropped, and so is what is in front of the name for
/// a full one, `tango://host:port/` or the protocol it is reached with.
fn local_dev_name(name: &str) -> &str {
    let name = name.split_once('#').map_or(name, |(name, _)| name);
    match name.split_once("://") {
        Some((_, rest)) => rest.split_once('/').map_or(rest, |(_, name)| name),
        None => name,
    }
}

/// The device of that server of that name, as a client gives it, see [`local_dev_name`].
fn find_dev(server: &Server, name: &str) -> Result<Arc<dyn AnyDev>> {
    let name = local_dev_name(name);
    server.any_dev(name)
        .ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_DeviceNotFound))
            .with_description(format!("Device {name:?} is not served by this server")))
}

/// The polling of that server, failing once it is not served anymore.
fn poller(server: &Server) -> Result<Poller> {
    server.poller()
        .ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_ServerNotRunning))
            .with_description(format!("Server {:?} is not served anymore", server.name())))
}

/// The device, the kind and the name of the object a polling command is about, which are
/// `Str[0]` to `Str[2]`, the name being given back as the definition of the object gives
/// it: an object the device does not have, or one that cannot be polled, is refused.
fn polled_target(server: &Server, strs: &Spectrum<str>) -> Result<(Arc<dyn AnyDev>, PolledKind, Arc<str>)> {

    let mut strs = strs.iter();
    let (Some(dev), Some(kind), Some(name)) = (strs.next(), strs.next(), strs.next()) else {
        return Err(Error::new(ErrorKind::Reason(reason::API_WrongNumberOfArgs))
            .with_description("Expected the device name, the object type and the object name"));
    };

    let target = find_dev(server, dev)?;
    let kind = PolledKind::parse(kind)?;
    let name = target.check_pollable(kind, name)?;
    Ok((target, kind, name))

}

/// The polling period a polling command gives, which is `Long[0]` in milliseconds: a
/// period of zero, which cppTango takes for a polling that the device triggers itself, is
/// not supported.
fn period_arg(longs: &Spectrum<i32>) -> Result<Duration> {

    let Some(&period) = longs.as_slice().first() else {
        return Err(Error::new(ErrorKind::Reason(reason::API_WrongNumberOfArgs))
            .with_description("Expected the polling period"));
    };

    let period = u64::try_from(period).ok()
        .filter(|&period| period > 0)
        .map(Duration::from_millis)
        .ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_NotSupported))
            .with_description(format!("Polling period {period} ms is not supported, a polling \
                triggered by the device in particular")))?;

    if period < MIN_PERIOD {
        return Err(Error::new(ErrorKind::Reason(reason::API_NotSupported))
            .with_description(format!("Polling period {period:?} is below the minimum of {MIN_PERIOD:?}")));
    }

    Ok(period)

}

/// Write the period of that object of that device, or that it has none, in the
/// `polled_attr` or `polled_cmd` property of the device, keeping what it lists of the
/// other objects, the property being deleted once it lists nothing.
async fn store_polled(server: &Server, dev: &str, kind: PolledKind, name: &str, period: Option<Duration>) -> Result<()> {

    let prop = match kind {
        PolledKind::Attr => "polled_attr",
        PolledKind::Command => "polled_cmd",
    };

    let database = server.database();
    let listed = database.get_dev_prop_async::<Spectrum<str>, _>(dev, prop).await?;

    let mut values = Vec::new();
    let mut listed = listed.iter();
    while let (Some(other), Some(other_period)) = (listed.next(), listed.next()) {
        if !other.eq_ignore_ascii_case(name) {
            values.push(other.to_string());
            values.push(other_period.to_string());
        }
    }

    if let Some(period) = period {
        values.push(name.to_string());
        values.push(period.as_millis().to_string());
    }

    if values.is_empty() {
        database.delete_dev_prop_async(dev, prop).await
    } else {
        database.set_dev_prop_async(dev, (prop, values)).await
    }

}

/// Add to that server the devices the database lists for it, each with the class it is an
/// instance of, the admin device aside: they are initialized and exported by the next
/// synchronization, see [`Server::sync_async`].
async fn add_listed_devs(server: &Server) -> Result<()> {

    let listed = server.database().list_server_dev_async(server.name()).await?;

    for dev in &listed {

        let (name, class) = (dev.name.value(), dev.value.value());

        // The admin device is itself registered in the database, as a device of the
        // DServer class of this server, and it is the very device running this.
        if class.eq_ignore_ascii_case(CLASS_NAME) {
            continue;
        }

        // A device of a class this executable does not define, or a device the server
        // already has, is a misconfiguration of that device alone: the others are
        // still added and the server still boots, which is what makes a server usable
        // while one of its devices is being fixed.
        if let Err(err) = server.add_dev(class, name) {
            warn!(dev = %name, %class, %err, "Failed to add the device listed by the database");
            continue;
        }

        debug!(dev = %name, %class, "Added the device listed by the database");

    }

    Ok(())

}

/// A command of the administration interface, which is a name and the two kinds it carries:
/// none of them is implemented, so nothing derives those kinds and each has to be given.
fn command(name: &'static str, in_kind: CommandDataKind, out_kind: CommandDataKind) -> CommandDef<Admin> {
    CommandDef::new(name)
        .in_kind(in_kind)
        .out_kind(out_kind)
}

#[cfg(test)]
mod tests {

    use super::*;

    #[test]
    fn local_names() {
        assert_eq!(local_dev_name("a/b/c"), "a/b/c");
        assert_eq!(local_dev_name("a/b/c#dbase=no"), "a/b/c");
        assert_eq!(local_dev_name("tango://host:10000/a/b/c#dbase=no"), "a/b/c");
    }

}