liminal-cli 0.3.1

//! The live rail: a liminal server through the Rust SDK, one connection per
//! read or request, closed before the answer is returned.

use std::future::Future;
use std::pin::pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::task::{Context, Poll, Waker};

use liminal_sdk::{
    ConnectionPoolConfig, ConversationHandle, RemoteConfig, RemoteConversationHandle,
    lys_pass_credential,
};

use crate::rail_link::{self, Subscription};
use serde_json::Value;

use super::{Counts, Rail, RailError};
use crate::config::{Config, Credential, Limits};
use crate::lys::LysAgent;

/// The console's topology subject, the directory's surface (SRV-027 R3).
pub const TOPOLOGY_SUBJECT: &str = "console.topology";

/// The live rail.
pub struct Live {
    server: String,
    channel: String,
    source: Source,
    limits: Limits,
    counts: Counts,
    conversations: AtomicU64,
}

impl std::fmt::Debug for Live {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Live")
            .field("server", &self.server)
            .field("channel", &self.channel)
            .field("source", &self.source)
            .field("limits", &self.limits)
            .finish_non_exhaustive()
    }
}

impl Live {
    /// A rail on `config`'s server and pass.
    ///
    /// # Errors
    /// Names a `[lys]` table that cannot ask Lys (see [`LysAgent::new`]).
    pub fn new(config: &Config) -> Result<Self, &'static str> {
        let source = match &config.credential {
            Credential::Token(token) => Source::Fixed(lys_pass_credential(token.as_str())),
            Credential::Lys(table) => {
                Source::Lys(LysAgent::new(table, config.limits.reply_timeout())?)
            }
            Credential::Open => Source::Open,
        };
        Ok(Self {
            server: config.server.clone(),
            channel: config.channel.clone(),
            source,
            limits: config.limits,
            counts: Counts::default(),
            conversations: AtomicU64::new(0),
        })
    }

    /// The credential the next connection presents: the fixed pass, the
    /// agent's pass from Lys, renewed when it is near expiry, or nothing at
    /// all from an open seat (ACCESS-010 R3).
    fn credential(&self) -> Result<Vec<u8>, RailError> {
        match &self.source {
            Source::Fixed(bytes) => Ok(bytes.clone()),
            Source::Open => Ok(Vec::new()),
            Source::Lys(agent) => agent
                .pass()
                .map(|pass| lys_pass_credential(&pass))
                .map_err(|error| RailError::Refused(error.0)),
        }
    }
}

/// Where a connection's pass comes from.
enum Source {
    /// One pass, read at start.
    Fixed(Vec<u8>),
    /// The agent's own pass, asked of Lys and renewed.
    Lys(LysAgent),
    /// No credential: an open seat on a loopback exchange (ACCESS-010 R3).
    Open,
}

impl Rail for Live {
    fn topology(&self) -> Result<Value, RailError> {
        let stream = self.admit(|credential| {
            Counts::add(&self.counts.connections_opened);
            Subscription::open(&self.server, TOPOLOGY_SUBJECT, credential).map_err(|error| {
                Counts::add(&self.counts.connections_closed);
                RailError::Refused(error.to_string())
            })
        })?;
        Counts::add(&self.counts.subscribes);
        let record = stream.recv_timeout(self.limits.reply_timeout());
        drop(stream);
        Counts::add(&self.counts.connections_closed);
        let record = record.map_err(|_| RailError::NoRecord {
            millis: self.limits.reply_timeout_ms,
        })?;
        Counts::add(&self.counts.records);
        if record.payload().len() > self.limits.record_bytes {
            return Err(RailError::RecordTooLarge {
                limit: self.limits.record_bytes,
            });
        }
        serde_json::from_slice(record.payload()).map_err(|error| RailError::Unreadable {
            message: error.to_string(),
        })
    }

    fn request(
        &self,
        subject: &str,
        input: &Value,
        reply: bool,
    ) -> Result<Option<Value>, RailError> {
        self.conversations.fetch_add(1, Ordering::SeqCst);
        Counts::add(&self.counts.connections_opened);
        let outcome = self.request_once(subject, input, reply);
        Counts::add(&self.counts.connections_closed);
        outcome
    }

    fn pass_for(&self, audience: &str) -> Result<String, RailError> {
        match &self.source {
            Source::Fixed(_) => Err(RailError::Refused(
                "pass_for needs a [lys] table: a ticket on a fixed token cannot ask Lys for a pass"
                    .to_owned(),
            )),
            Source::Open => Err(RailError::Refused(
                "pass_for needs a [lys] table: an open seat cannot ask Lys for a pass".to_owned(),
            )),
            Source::Lys(agent) => agent
                .pass_for(audience)
                .map_err(|error| RailError::Refused(error.0)),
        }
    }

    fn tap(&self, channel: &str) -> Result<super::TapSubscription, RailError> {
        self.admit(|credential| {
            super::live_tap::open(&self.server, channel, credential, self.limits.record_bytes)
        })
    }

    fn server_address(&self) -> &str {
        &self.server
    }

    fn counts(&self) -> &Counts {
        &self.counts
    }

    fn request_bytes(&self) -> usize {
        self.limits.request_bytes
    }

    fn flow_steps(&self) -> usize {
        self.limits.flow_steps
    }
}

impl Live {
    fn request_once(
        &self,
        subject: &str,
        input: &Value,
        reply: bool,
    ) -> Result<Option<Value>, RailError> {
        let mut attempted = false;
        let config = self.admit(|credential| {
            if attempted {
                Counts::add(&self.counts.connections_closed);
                Counts::add(&self.counts.connections_opened);
            }
            attempted = true;
            RemoteConfig::new(
                self.server.clone(),
                self.channel.clone(),
                subject.to_owned(),
                ConnectionPoolConfig::new(1, 10, 16),
            )
            .and_then(|config| rail_link::connect(&self.server, config, credential))
            .map_err(|error| RailError::Refused(error.to_string()))
        })?;
        let conversation = RemoteConversationHandle::new(&config);
        Counts::add(&self.counts.requests);
        if !reply {
            conversation
                .send(input)
                .map_err(|error| RailError::Refused(error.to_string()))?;
            return Ok(None);
        }
        conversation
            .request(input)
            .map_err(|error| RailError::Refused(error.to_string()))?;
        let answer: Value = block_on(conversation.receive())
            .and_then(|received| received.map_err(|error| error.to_string()))
            .map_err(RailError::Refused)?;
        Counts::add(&self.counts.replies);
        Ok(Some(answer))
    }
}

/// Drives the SDK's ready-or-never future once: the synchronous transport
/// has the reply in hand by the time `receive` is polled.
fn block_on<F: Future>(future: F) -> Result<F::Output, String> {
    let mut future = pin!(future);
    let waker = Waker::noop();
    let mut context = Context::from_waker(waker);
    match future.as_mut().poll(&mut context) {
        Poll::Ready(value) => Ok(value),
        Poll::Pending => Err("the synchronous transport parked unexpectedly".to_owned()),
    }
}

#[path = "pass_renewal.rs"]
mod pass_renewal;

#[cfg(test)]
#[path = "live_tests.rs"]
mod tests;

impl std::fmt::Debug for Source {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            Self::Fixed(_) => "FixedCredential",
            Self::Lys(_) => "LysCredential",
            Self::Open => "OpenCredential",
        })
    }
}