taktwerk 0.8.1

//! The OPC UA side of `taktwerk record`: one recorder's pull nodes on an engine's server, as
//! [`taktwerk_core::record::wire`] describes them.

use opcua::types::{
    AttributeId, BrowseDescription, BrowseDirection, BrowseResultMask, CallMethodRequest,
    NodeClass, NodeClassMask, NodeId, ReadValueId, ReferenceTypeId, TimestampsToReturn, Variant,
};
use taktwerk_core::record::{Event, EventKind, Manifest};

use crate::ua::Conn;

/// One `read_rows` result.
#[derive(Debug)]
pub struct RowBatch {
    /// The engine process.
    pub engine_run: String,
    /// Where the next read starts.
    pub next_cycle: u64,
    /// Encoded rows.
    pub rows: Vec<u8>,
}

/// One `read_events` result.
#[derive(Debug)]
pub struct EventBatch {
    /// The engine process.
    pub engine_run: String,
    /// Oldest sequence number the log holds.
    pub oldest_seq: u64,
    /// Where the next read starts.
    pub next_seq: u64,
    /// The events.
    pub events: Vec<Event>,
}

/// A recorder on a connected server.
pub struct Remote {
    conn: Conn,
    folder: NodeId,
    manifest: NodeId,
    read_rows: NodeId,
    read_events: NodeId,
}

/// The ids of the recorders under `<prefix>.recorder`.
async fn recorder_ids(conn: &Conn, ns: u16, prefix: &str) -> Result<Vec<String>, String> {
    let description = BrowseDescription {
        node_id: NodeId::new(ns, format!("{prefix}.recorder/")),
        browse_direction: BrowseDirection::Forward,
        reference_type_id: ReferenceTypeId::Organizes.into(),
        include_subtypes: true,
        node_class_mask: NodeClassMask::OBJECT.bits(),
        result_mask: BrowseResultMask::All as u32,
    };
    let results = conn
        .session
        .browse(&[description], 0, None)
        .await
        .map_err(|e| format!("browse: {e}"))?;
    Ok(results
        .into_iter()
        .flat_map(|r| r.references.unwrap_or_default())
        .filter(|r| r.node_class == NodeClass::Object)
        .map(|r| r.browse_name.name.as_ref().to_owned())
        .collect())
}

impl Remote {
    /// Find recorder `id` (or the only one) under `<prefix>.recorder` in `namespace`. `Err` with
    /// `fatal` set when retrying cannot help.
    pub async fn open(
        conn: Conn,
        namespace: &str,
        prefix: &str,
        id: Option<&str>,
    ) -> Result<Self, (Box<Conn>, String, bool)> {
        let ns = match conn.session.get_namespace_index(namespace).await {
            Ok(ns) => ns,
            Err(e) => {
                return Err((
                    Box::new(conn),
                    format!("namespace `{namespace}`: {e}"),
                    false,
                ));
            }
        };
        let ids = match recorder_ids(&conn, ns, prefix).await {
            Ok(ids) => ids,
            Err(e) => return Err((Box::new(conn), e, false)),
        };
        let id = match (id, ids.as_slice()) {
            (Some(id), _) if ids.iter().any(|i| i == id) => id.to_owned(),
            (_, []) => {
                let msg = format!("no recorder under `{prefix}.recorder`");
                return Err((Box::new(conn), msg, true));
            }
            (Some(id), _) => {
                let msg = format!("no recorder `{id}`; the engine has {}", ids.join(", "));
                return Err((Box::new(conn), msg, true));
            }
            (None, [only]) => only.clone(),
            (None, _) => {
                let msg = format!(
                    "several recorders ({}); pick one with --recorder",
                    ids.join(", ")
                );
                return Err((Box::new(conn), msg, true));
            }
        };
        let base = format!("{prefix}.recorder.{id}");
        Ok(Self {
            conn,
            folder: NodeId::new(ns, format!("{base}/")),
            manifest: NodeId::new(ns, format!("{base}.manifest")),
            read_rows: NodeId::new(ns, format!("{base}.read_rows")),
            read_events: NodeId::new(ns, format!("{base}.read_events")),
        })
    }

    /// The connection, for closing.
    pub fn into_conn(self) -> Conn {
        self.conn
    }

    /// Whether the connection is gone for good.
    pub fn ended(&self) -> bool {
        self.conn.ended()
    }

    /// The recorder's manifest.
    pub async fn manifest(&self) -> Result<Manifest, String> {
        let dv = self
            .conn
            .session
            .read(
                &[ReadValueId::new(self.manifest.clone(), AttributeId::Value)],
                TimestampsToReturn::Neither,
                0.0,
            )
            .await
            .map_err(|e| format!("read manifest: {e}"))?
            .into_iter()
            .next();
        let Some(Variant::String(json)) = dv.and_then(|dv| dv.value) else {
            return Err(format!("{}: no manifest", self.manifest));
        };
        serde_json::from_str(json.as_ref()).map_err(|e| format!("manifest: {e}"))
    }

    async fn call(&self, method: &NodeId, args: Vec<Variant>) -> Result<Vec<Variant>, String> {
        let r = self
            .conn
            .session
            .call_one(CallMethodRequest {
                object_id: self.folder.clone(),
                method_id: method.clone(),
                input_arguments: Some(args),
            })
            .await
            .map_err(|e| format!("call {method}: {e}"))?;
        if !r.status_code.is_good() {
            return Err(format!("call {method}: {}", r.status_code));
        }
        Ok(r.output_arguments.unwrap_or_default())
    }

    /// `read_rows(from_cycle, max_rows)`.
    pub async fn read_rows(&self, from_cycle: u64, max_rows: u32) -> Result<RowBatch, String> {
        let out = self
            .call(
                &self.read_rows,
                vec![Variant::from(from_cycle), Variant::from(max_rows)],
            )
            .await?;
        match <[Variant; 4]>::try_from(out) {
            Ok(
                [
                    Variant::String(run),
                    Variant::UInt64(_oldest),
                    Variant::UInt64(next_cycle),
                    rows,
                ],
            ) => {
                let rows = match rows {
                    Variant::ByteString(b) => b.value.unwrap_or_default(),
                    Variant::Empty => Vec::new(),
                    _ => return Err("read_rows: rows are not a ByteString".to_owned()),
                };
                Ok(RowBatch {
                    engine_run: run.as_ref().to_owned(),
                    next_cycle,
                    rows,
                })
            }
            _ => Err("read_rows: unexpected outputs".to_owned()),
        }
    }

    /// `read_events(from_seq, max)`.
    pub async fn read_events(&self, from_seq: u64, max: u32) -> Result<EventBatch, String> {
        let out = self
            .call(
                &self.read_events,
                vec![Variant::from(from_seq), Variant::from(max)],
            )
            .await?;
        let bad = || "read_events: unexpected outputs".to_owned();
        let Ok(
            [
                Variant::String(run),
                Variant::UInt64(oldest_seq),
                Variant::UInt64(next_seq),
                cycles,
                kinds,
                sources,
                details,
            ],
        ) = <[Variant; 7]>::try_from(out)
        else {
            return Err(bad());
        };
        let cycles: Vec<u64> = array(cycles)
            .into_iter()
            .map(|v| match v {
                Variant::UInt64(c) => Ok(c),
                _ => Err(bad()),
            })
            .collect::<Result<_, _>>()?;
        let text = |v: Variant| -> Result<Vec<String>, String> {
            array(v)
                .into_iter()
                .map(|v| match v {
                    Variant::String(s) => Ok(s.as_ref().to_owned()),
                    _ => Err(bad()),
                })
                .collect()
        };
        let (kinds, sources, details) = (text(kinds)?, text(sources)?, text(details)?);
        if kinds.len() != cycles.len()
            || sources.len() != cycles.len()
            || details.len() != cycles.len()
        {
            return Err(bad());
        }
        let first = next_seq.saturating_sub(cycles.len() as u64);
        let mut events = Vec::with_capacity(cycles.len());
        for (i, (((cycle, kind), source), detail)) in cycles
            .into_iter()
            .zip(kinds)
            .zip(sources)
            .zip(details)
            .enumerate()
        {
            let Some(kind) = EventKind::parse(&kind) else {
                tracing::warn!(%kind, "unknown event kind skipped");
                continue;
            };
            events.push(Event {
                seq: first + i as u64,
                cycle,
                kind,
                source,
                detail,
            });
        }
        Ok(EventBatch {
            engine_run: run.as_ref().to_owned(),
            oldest_seq,
            next_seq,
            events,
        })
    }
}

/// The elements of an array variant; an empty or null array is none.
fn array(v: Variant) -> Vec<Variant> {
    match v {
        Variant::Array(a) => a.values,
        _ => Vec::new(),
    }
}