fux 0.10.0

//! Manager socket contract: list, resolve and kill workspaces. Uses the control preface and
//! newline-delimited JSON, one request per connection.

#[cfg(test)]
use crate::proto::control::write_frame;
use anyhow::{Context, Result, bail};
use std::os::unix::net::UnixStream;
use std::path::Path;
use std::time::{Duration, Instant};

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "request", rename_all = "kebab-case", deny_unknown_fields)]
pub enum ManagerRequest {
    Create {
        name: String,
    },
    Final {
        instance: String,
        pane: crate::ids::PaneId,
    },
    /// Attach to `name`, creating it when missing. `None` applies the documented default rule.
    Resolve {
        name: Option<String>,
    },
    List,
    Kill {
        name: String,
    },
    /// The server's identity, version, runtime directory and limits.
    Info,
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "reply", rename_all = "kebab-case", deny_unknown_fields)]
pub enum ManagerReply {
    Final {
        result: crate::proto::control::Reply,
    },
    Attach {
        descriptor: super::Descriptor,
    },
    Names {
        names: Vec<String>,
    },
    Info {
        info: Box<crate::proto::control::ServerInfo>,
    },
    Failed {
        message: String,
    },
}

pub const MANAGER_DEADLINE: Duration = Duration::from_secs(15);

pub fn manager_request(path: &Path, request: &ManagerRequest) -> Result<ManagerReply> {
    manager_request_until(path, request, Instant::now() + MANAGER_DEADLINE)
}

/// One absolute deadline covers negotiation, write and response, including slow peers.
pub fn manager_request_until(
    path: &Path,
    request: &ManagerRequest,
    deadline: Instant,
) -> Result<ManagerReply> {
    let remaining = || {
        deadline
            .checked_duration_since(Instant::now())
            .filter(|duration| !duration.is_zero())
            .ok_or_else(|| anyhow::anyhow!("manager request timed out"))
    };
    let mut stream = crate::proto::socket::connect_local(path, deadline)
        .with_context(|| format!("connecting to manager socket {}", path.display()))?;
    crate::proto::socket::negotiate_client_with_timeout(&mut stream, remaining()?)
        .context("authenticating the manager socket and negotiating its control protocol")?;
    crate::proto::control::write_frame_until(&mut stream, request, deadline)
        .context("sending manager request")?;
    stream.set_read_timeout(Some(remaining()?))?;
    let reply = read_json_frame(&mut stream, remaining()?).context("receiving manager reply")?;
    serde_json::from_slice(&reply).context("decoding manager reply")
}

/// One newline-delimited frame read byte by byte, so repeated calls on one stream each get the
/// next frame.
pub fn read_json_frame(stream: &mut UnixStream, deadline: Duration) -> Result<Vec<u8>> {
    local_ipc::FrameReader::bytewise(crate::proto::control::MAX_FRAME_BYTES)
        .next_frame(stream, Instant::now() + deadline)
        .map_err(|error| match error {
            local_ipc::FrameError::TimedOut => anyhow::anyhow!("manager response timed out"),
            local_ipc::FrameError::Closed => {
                anyhow::anyhow!("manager closed before a complete response")
            }
            local_ipc::FrameError::Oversize => {
                anyhow::anyhow!("manager response exceeds frame limit")
            }
            local_ipc::FrameError::Io(error) => error.into(),
        })
}

pub fn workspace_names(path: &Path) -> Result<Vec<String>> {
    match manager_request(path, &ManagerRequest::List)? {
        ManagerReply::Names { names } => {
            anyhow::ensure!(
                names.len() <= crate::config::MAX_WORKSPACES,
                "too many workspaces in manager reply"
            );
            for name in &names {
                crate::ids::validate_workspace_name(name)?;
            }
            Ok(names)
        }
        ManagerReply::Failed { message } => bail!("{message}"),
        ManagerReply::Attach { .. } | ManagerReply::Info { .. } | ManagerReply::Final { .. } => {
            bail!("manager did not return a workspace list")
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn reply_reader_accepts_a_complete_frame() -> Result<()> {
        let (mut reader, mut writer) = UnixStream::pair()?;
        write_frame(
            &mut writer,
            &ManagerReply::Names {
                names: vec!["one".into()],
            },
        )?;
        drop(writer);
        let bytes = read_json_frame(&mut reader, Duration::from_secs(1))?;
        assert!(matches!(
            serde_json::from_slice::<ManagerReply>(&bytes),
            Ok(ManagerReply::Names { .. })
        ));
        Ok(())
    }
}