synapse_application/
runtime.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use serde_json::{Value, json};
5use soma_fleet::{HostEndpoint, HostId, HostRecord, HostRepository, TopologySnapshot};
6use soma_infra::{
7    ComposeInspector, ComposeProjectRef, DockerClientProvider, FilesystemQueryInspector,
8    HostInspector, HostSystemInspector, LogReader, ProcessInspector, ZfsInspector,
9};
10use soma_ops::{AccessClass, OperationName, Timestamp};
11use tokio_util::sync::CancellationToken;
12
13use crate::runtime_params::optional_str;
14use crate::{ExecutionError, SynapseCatalog};
15
16const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
17
18/// Product-owned ports required to execute every canonical read operation.
19pub struct SynapseReadPorts {
20    /// Immutable fleet topology source.
21    pub hosts: Arc<dyn HostRepository>,
22    /// Core host identity and resource inspection.
23    pub host: Arc<dyn HostInspector>,
24    /// Host services, network, mounts, ports, usage, and doctor checks.
25    pub host_system: Arc<dyn HostSystemInspector>,
26    /// Revision-bound Docker client provider.
27    pub docker: Arc<dyn DockerClientProvider>,
28    /// Compose inspection engine.
29    pub compose: Arc<dyn ComposeInspector>,
30    /// Bounded filesystem read, tree, find, and tail engine.
31    pub filesystem: Arc<dyn FilesystemQueryInspector>,
32    /// Process inspection engine.
33    pub processes: Arc<dyn ProcessInspector>,
34    /// Operating-system log reader.
35    pub logs: Arc<dyn LogReader>,
36    /// ZFS inspection engine.
37    pub zfs: Arc<dyn ZfsInspector>,
38}
39
40/// Canonical Synapse read-operation runtime.
41pub struct SynapseReadRuntime {
42    pub(crate) catalog: &'static SynapseCatalog,
43    pub(crate) ports: SynapseReadPorts,
44    default_host: Option<HostId>,
45    timeout: Duration,
46}
47
48impl SynapseReadRuntime {
49    /// Creates a runtime using the checked-in canonical catalog.
50    #[must_use]
51    pub fn new(ports: SynapseReadPorts) -> Self {
52        Self {
53            catalog: SynapseCatalog::embedded(),
54            ports,
55            default_host: None,
56            timeout: DEFAULT_TIMEOUT,
57        }
58    }
59
60    /// Sets the product default host used when a schema permits omission.
61    #[must_use]
62    pub fn with_default_host(mut self, host: HostId) -> Self {
63        self.default_host = Some(host);
64        self
65    }
66
67    /// Sets the per-operation command deadline budget.
68    #[must_use]
69    pub fn with_timeout(mut self, timeout: Duration) -> Self {
70        if !timeout.is_zero() {
71            self.timeout = timeout;
72        }
73        self
74    }
75
76    /// Executes one schema-validated canonical read operation.
77    pub async fn execute(
78        &self,
79        operation: &OperationName,
80        parameters: &Value,
81        cancellation: &CancellationToken,
82    ) -> Result<Value, ExecutionError> {
83        let spec = self
84            .catalog
85            .operation(operation)
86            .ok_or_else(|| crate::CompatibilityError::UnknownOperation(operation.clone()))?;
87        if spec.access() != AccessClass::Read {
88            return Err(ExecutionError::UnsupportedOperation(operation.clone()));
89        }
90        self.catalog.validate_parameters(operation, parameters)?;
91        let result = match operation.as_str().split('.').next().unwrap_or_default() {
92            "product" => self.execute_product(operation, parameters)?,
93            "docker" | "container" => {
94                self.execute_docker(operation, parameters, cancellation)
95                    .await?
96            }
97            "host" | "fleet" => {
98                self.execute_host(operation, parameters, cancellation)
99                    .await?
100            }
101            "compose" | "processes" | "zfs" | "logs" => {
102                self.execute_observability(operation, parameters, cancellation)
103                    .await?
104            }
105            "files" | "filesystem" => {
106                self.execute_files(operation, parameters, cancellation)
107                    .await?
108            }
109            _ => return Err(ExecutionError::UnsupportedOperation(operation.clone())),
110        };
111        self.catalog.validate_result(operation, &result)?;
112        Ok(result)
113    }
114
115    fn execute_product(
116        &self,
117        operation: &OperationName,
118        parameters: &Value,
119    ) -> Result<Value, ExecutionError> {
120        if operation.as_str() != "product.help" {
121            return Err(ExecutionError::UnsupportedOperation(operation.clone()));
122        }
123        let topic = optional_str(parameters, "topic")?;
124        let operations = self
125            .catalog
126            .operations()
127            .filter(|spec| topic.is_none_or(|topic| spec.name().as_str().starts_with(topic)))
128            .map(|spec| {
129                json!({
130                    "name": spec.name().as_str(),
131                    "summary": format!("Canonical {} operation", spec.name())
132                })
133            })
134            .collect::<Vec<_>>();
135        let mut names = self
136            .catalog
137            .operations()
138            .filter_map(|spec| spec.name().as_str().split('.').next())
139            .collect::<std::collections::BTreeSet<_>>();
140        if let Some(topic) = topic {
141            names.retain(|name| name.starts_with(topic) || topic.starts_with(name));
142        }
143        let topics = names
144            .into_iter()
145            .map(|name| json!({"name": name, "summary": format!("{name} operations")}))
146            .collect::<Vec<_>>();
147        Ok(json!({"topics": topics, "operations": operations}))
148    }
149
150    pub(crate) fn deadline(&self) -> Timestamp {
151        let millis = i64::try_from(self.timeout.as_millis()).unwrap_or(i64::MAX);
152        Timestamp::from_unix_millis(Timestamp::now().unix_millis().saturating_add(millis))
153    }
154
155    pub(crate) async fn resolve_host(
156        &self,
157        parameters: &Value,
158    ) -> Result<HostRecord, ExecutionError> {
159        let snapshot = self.ports.hosts.snapshot().await?;
160        if let Some(name) = optional_str(parameters, "host")? {
161            return self.resolve_host_name_from_snapshot(&snapshot, name);
162        }
163        if let Some(default) = &self.default_host {
164            return snapshot
165                .get(default)
166                .cloned()
167                .ok_or_else(|| ExecutionError::HostNotFound(default.to_string()));
168        }
169        if snapshot.len() == 1 {
170            return Ok(snapshot.hosts().next().expect("single host exists").clone());
171        }
172        let local = snapshot
173            .hosts()
174            .filter(|host| matches!(host.endpoint(), HostEndpoint::Local))
175            .collect::<Vec<_>>();
176        if local.len() == 1 {
177            return Ok(local[0].clone());
178        }
179        Err(ExecutionError::HostRequired)
180    }
181
182    pub(crate) async fn resolve_host_name(&self, name: &str) -> Result<HostRecord, ExecutionError> {
183        let snapshot = self.ports.hosts.snapshot().await?;
184        self.resolve_host_name_from_snapshot(&snapshot, name)
185    }
186
187    fn resolve_host_name_from_snapshot(
188        &self,
189        snapshot: &TopologySnapshot,
190        name: &str,
191    ) -> Result<HostRecord, ExecutionError> {
192        let id = HostId::new(name).map_err(|error| ExecutionError::InvalidParameter {
193            field: "host".into(),
194            message: error.to_string(),
195        })?;
196        snapshot
197            .get(&id)
198            .cloned()
199            .ok_or_else(|| ExecutionError::HostNotFound(name.to_owned()))
200    }
201
202    pub(crate) async fn resolve_project(
203        &self,
204        host: &HostRecord,
205        name: &str,
206        cancellation: &CancellationToken,
207    ) -> Result<ComposeProjectRef, ExecutionError> {
208        let projects = self
209            .ports
210            .compose
211            .list_projects(host, self.deadline(), cancellation)
212            .await?;
213        let project = projects
214            .into_iter()
215            .find(|project| project.name == name)
216            .ok_or_else(|| ExecutionError::ProjectNotFound {
217                host: host.id().to_string(),
218                project: name.to_owned(),
219            })?;
220        let config = project.config_files.into_iter().next().ok_or_else(|| {
221            ExecutionError::ProjectNotFound {
222                host: host.id().to_string(),
223                project: name.to_owned(),
224            }
225        })?;
226        Ok(ComposeProjectRef::new(name, config)?)
227    }
228
229    pub(crate) async fn topology_items(&self) -> Result<Value, ExecutionError> {
230        let snapshot = self.ports.hosts.snapshot().await?;
231        let items = snapshot
232            .hosts()
233            .map(|host| {
234                json!({
235                    "id": host.id(),
236                    "revision": host.revision(),
237                    "endpoint": host.endpoint(),
238                    "labels": host.labels().collect::<Vec<_>>(),
239                    "capabilities": host.capabilities().collect::<Vec<_>>()
240                })
241            })
242            .collect::<Vec<_>>();
243        crate::runtime_result::items(items, snapshot.len(), false)
244    }
245}
synapse_application/runtime.rs

synapse_application/
runtime.rs
