1use crate::cli::LinkArgs;
7use crate::errors::CliError;
8use crate::output::terminal;
9use crate::output::{CommandResult, HumanRenderable, OutputContext};
10use serde::{Deserialize, Serialize};
11use std::collections::HashMap;
12use std::path::{Path, PathBuf};
13
14#[derive(Debug, Serialize)]
16pub struct LinkResult {
17 pub all_linked: bool,
19 pub flows: Vec<FlowLinkResult>,
21}
22
23#[derive(Debug, Serialize)]
25pub struct FlowLinkResult {
26 pub name: String,
28 pub linked: bool,
30 pub component_count: usize,
32 pub edge_count: usize,
34 pub diagnostics: Vec<String>,
36}
37
38impl HumanRenderable for LinkResult {
39 fn render_human(&self, ctx: &OutputContext) {
40 for flow in &self.flows {
41 if flow.linked {
42 terminal::print_success(
43 ctx,
44 &format!(
45 "Flow \"{}\" links successfully ({} components, {} edges, 0 errors)",
46 flow.name, flow.component_count, flow.edge_count
47 ),
48 );
49 } else {
50 terminal::print_failure(ctx, &format!("Flow \"{}\" has linking errors", flow.name));
51 for d in &flow.diagnostics {
52 eprintln!(" {d}");
53 }
54 }
55 }
56 }
57}
58
59pub async fn execute(
63 args: &LinkArgs,
64 ctx: &OutputContext,
65) -> Result<CommandResult<LinkResult>, CliError> {
66 let manifest_path = &args.manifest;
67
68 if !manifest_path.exists() {
69 return Err(CliError::Config {
70 detail: format!("Manifest not found: {}", manifest_path.display()),
71 file: Some(manifest_path.display().to_string()),
72 suggestion: "Run this command from a Torvyn project directory.".into(),
73 });
74 }
75
76 let manifest_content = std::fs::read_to_string(manifest_path).map_err(|e| CliError::Io {
77 detail: e.to_string(),
78 path: Some(manifest_path.display().to_string()),
79 })?;
80
81 let manifest = torvyn_config::ComponentManifest::from_toml_str(
82 &manifest_content,
83 manifest_path.to_str().unwrap_or("Torvyn.toml"),
84 )
85 .map_err(|errors| CliError::Config {
86 detail: format!("Manifest has {} error(s)", errors.len()),
87 file: Some(manifest_path.display().to_string()),
88 suggestion: "Run `torvyn check` first.".into(),
89 })?;
90
91 if !manifest.has_flows() {
92 return Err(CliError::Config {
93 detail: "No flows defined in manifest".into(),
94 file: Some(manifest_path.display().to_string()),
95 suggestion: "Add a [flow.*] section to your Torvyn.toml.".into(),
96 });
97 }
98
99 let _project_dir = manifest_path.parent().unwrap_or(Path::new("."));
100 let mut flow_results = Vec::new();
101 let mut all_linked = true;
102
103 for (flow_name, flow_value) in &manifest.flow {
107 if let Some(ref filter) = args.flow {
109 if flow_name != filter {
110 continue;
111 }
112 }
113
114 ctx.print_debug(&format!("Linking flow: {flow_name}"));
115
116 let flow_def: FlowDef = flow_value
118 .clone()
119 .try_into()
120 .map_err(|e: toml::de::Error| CliError::Config {
121 detail: format!("Invalid flow definition for '{flow_name}': {e}"),
122 file: Some(manifest_path.display().to_string()),
123 suggestion: "Check the [flow] section in your Torvyn.toml.".into(),
124 })?;
125
126 let mut topo = torvyn_linker::PipelineTopology::new(flow_name.clone());
128
129 for (node_name, node_def) in &flow_def.nodes {
131 let role = match node_def.interface.as_deref() {
132 Some(iface) if iface.contains("source") => torvyn_types::ComponentRole::Source,
133 Some(iface) if iface.contains("sink") => torvyn_types::ComponentRole::Sink,
134 Some(iface) if iface.contains("filter") => torvyn_types::ComponentRole::Filter,
135 Some(iface) if iface.contains("router") => torvyn_types::ComponentRole::Router,
136 _ => torvyn_types::ComponentRole::Processor,
137 };
138
139 topo.add_node(torvyn_linker::TopologyNode {
140 name: node_name.clone(),
141 role,
142 artifact_path: PathBuf::from(&node_def.component),
143 config: node_def.config.clone(),
144 capability_grants: vec![],
145 });
146 }
147
148 for edge_def in &flow_def.edges {
150 let (from_node, from_port) = parse_port_ref(&edge_def.from);
151 let (to_node, to_port) = parse_port_ref(&edge_def.to);
152
153 topo.add_edge(torvyn_linker::TopologyEdge {
154 from_node,
155 from_port,
156 to_node,
157 to_port,
158 queue_depth: 64,
159 backpressure_policy: Default::default(),
160 });
161 }
162
163 let node_count = topo.nodes.len();
165 let edge_count = topo.edges.len();
166
167 let mut linker = torvyn_linker::PipelineLinker::new();
168 let link_result = linker.link_topology_only(&topo);
169
170 let (linked, diags) = match link_result {
171 Ok(_) => (true, vec![]),
172 Err(e) => {
173 let diag_strs = match &e {
174 torvyn_linker::LinkerError::LinkFailed(report) => report
175 .errors
176 .iter()
177 .map(|d| d.message.clone())
178 .collect::<Vec<_>>(),
179 other => vec![other.to_string()],
180 };
181 (false, diag_strs)
182 }
183 };
184
185 if !linked {
186 all_linked = false;
187 }
188
189 flow_results.push(FlowLinkResult {
190 name: flow_name.clone(),
191 linked,
192 component_count: node_count,
193 edge_count,
194 diagnostics: diags,
195 });
196 }
197
198 let result = LinkResult {
199 all_linked,
200 flows: flow_results,
201 };
202
203 if !all_linked {
204 let err_msgs: Vec<String> = result
205 .flows
206 .iter()
207 .filter(|f| !f.linked)
208 .flat_map(|f| f.diagnostics.clone())
209 .collect();
210 return Err(CliError::Link {
211 detail: "One or more flows failed to link".into(),
212 diagnostics: err_msgs,
213 });
214 }
215
216 Ok(CommandResult {
217 success: true,
218 command: "link".into(),
219 data: result,
220 warnings: vec![],
221 })
222}
223
224#[derive(Debug, Deserialize)]
226struct FlowDef {
227 #[serde(default)]
229 nodes: HashMap<String, NodeDef>,
230 #[serde(default)]
232 edges: Vec<EdgeDef>,
233}
234
235#[derive(Debug, Deserialize)]
237struct NodeDef {
238 component: String,
240 #[serde(default)]
242 interface: Option<String>,
243 #[serde(default)]
245 config: Option<String>,
246}
247
248#[derive(Debug, Deserialize)]
250struct EdgeDef {
251 from: String,
253 to: String,
255}
256
257fn parse_port_ref(s: &str) -> (String, String) {
259 match s.split_once(':') {
260 Some((node, port)) => (node.to_string(), port.to_string()),
261 None => (s.to_string(), "default".to_string()),
262 }
263}