gru/gree-controller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.11.2

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-09-02 23:10:44 +02:00
parent 563523d2e2
commit 76bbd39378
16 changed files with 196 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/flows.rs
+85 -14
View File
@@ -4,6 +4,8 @@ struct FlowInput {
#[serde(default = "yes")]
enabled: bool,
#[serde(default)]
draft: bool,
#[serde(default)]
description: String,
#[serde(default)]
nodes: Vec<crate::models::FlowNode>,
@@ -107,9 +109,8 @@ fn validate_shared_input_source(kind: &str, config: &Value, state: &AppState) ->
Ok(())
}
fn validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
fn validate_flow_draft_graph(input: &FlowInput) -> Result<(), AppError> {
if input.name.trim().is_empty() { return Err(AppError::BadRequest("flow name is required".into())); }
if input.nodes.is_empty() { return Err(AppError::BadRequest("flow needs at least one block".into())); }
let mut ids = std::collections::HashSet::<String>::new();
for node in &input.nodes {
if node.id.trim().is_empty() || !ids.insert(node.id.clone()) { return Err(AppError::BadRequest("flow contains duplicate or empty block IDs".into())); }
@@ -117,14 +118,26 @@ fn validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
return Err(AppError::BadRequest(format!("unsupported flow block: {}", node.kind)));
}
}
if !input.nodes.iter().any(|node| flow_action_kind(&node.kind)) { return Err(AppError::BadRequest("flow needs at least one action block".into())); }
let mut edge_ids = std::collections::HashSet::<String>::new();
let mut connections = std::collections::HashSet::<(String, String)>::new();
for edge in &input.edges {
if edge.id.trim().is_empty() || !edge_ids.insert(edge.id.clone()) || !connections.insert((edge.from.clone(), edge.to.clone())) {
return Err(AppError::BadRequest("flow contains duplicate or empty connection IDs".into()));
}
if edge.from == edge.to || !ids.contains(edge.from.as_str()) || !ids.contains(edge.to.as_str()) {
if !ids.contains(edge.from.as_str()) || !ids.contains(edge.to.as_str()) {
return Err(AppError::BadRequest("flow contains a connection to a missing block".into()));
}
}
Ok(())
}
fn validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
validate_flow_draft_graph(input)?;
if input.nodes.is_empty() { return Err(AppError::BadRequest("flow needs at least one block".into())); }
let ids = input.nodes.iter().map(|node| node.id.clone()).collect::<std::collections::HashSet<_>>();
if !input.nodes.iter().any(|node| flow_action_kind(&node.kind)) { return Err(AppError::BadRequest("flow needs at least one action block".into())); }
for edge in &input.edges {
if edge.from == edge.to {
return Err(AppError::BadRequest("flow contains an invalid connection".into()));
}
let from = input.nodes.iter().find(|node| node.id == edge.from).expect("validated Flow source");
@@ -154,8 +167,7 @@ fn validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
}
}
// Reject cycles. Flow is deliberately a DAG: finite evaluation, deterministic topological order,
// and no hidden state machine semantics unless a dedicated stateful block is introduced later.
// Reject cycles for executable Flows. Drafts intentionally allow unfinished wiring and are never compiled.
let mut outgoing = std::collections::HashMap::<String, Vec<String>>::new();
for edge in &input.edges { outgoing.entry(edge.from.clone()).or_default().push(edge.to.clone()); }
fn visit(id: &str, outgoing: &std::collections::HashMap<String, Vec<String>>, temp: &mut std::collections::HashSet<String>, done: &mut std::collections::HashSet<String>) -> bool {
@@ -542,23 +554,34 @@ fn compile_flow(state: &AppState, mut flow: crate::models::Flow) -> Result<(crat
}
fn flow_from_input(id: String, input: FlowInput, created_at: chrono::DateTime<Utc>, revision: u64) -> crate::models::Flow {
let draft = input.draft;
crate::models::Flow {
id, name: input.name.trim().into(), enabled: input.enabled, description: input.description.trim().into(), nodes: input.nodes, edges: input.edges,
id, name: input.name.trim().into(), enabled: if draft { false } else { input.enabled }, draft,
description: input.description.trim().into(), nodes: input.nodes, edges: input.edges,
summary: String::new(), compiled_schedule_ids: vec![], compiled_automation_ids: vec![], revision, created_at, updated_at: Utc::now(),
}
}
fn prepare_draft_flow(mut flow: crate::models::Flow) -> crate::models::Flow {
flow.enabled = false;
flow.draft = true;
flow.compiled_schedule_ids.clear();
flow.compiled_automation_ids.clear();
flow.summary = format!("Draft · {} blocks", flow.nodes.len());
flow
}
async fn list_flows(State(state): State<AppState>) -> Result<Json<Vec<crate::models::Flow>>, AppError> { Ok(Json(state.db.list_flows()?)) }
async fn get_flow(State(state): State<AppState>, Path(id): Path<String>) -> Result<Json<crate::models::Flow>, AppError> { state.db.get_flow(&id)?.map(Json).ok_or_else(|| AppError::NotFound(format!("flow {id}"))) }
async fn create_flow(State(state): State<AppState>, Json(input): Json<FlowInput>) -> Result<(StatusCode, Json<crate::models::Flow>), AppError> {
validate_flow_graph(&input)?;
if input.draft { validate_flow_draft_graph(&input)?; } else { validate_flow_graph(&input)?; }
let _configuration_guard = state.lock_configuration_operation().await;
let _automation_guard = state.lock_automation_operation().await;
let _schedule_guard = state.lock_schedule_operation().await;
let _cycle_guard = state.lock_zone_control_cycle().await;
let flow = flow_from_input(Uuid::new_v4().to_string(), input, Utc::now(), 1);
let (flow, schedules, automations) = compile_flow(&state, flow)?;
let (flow, schedules, automations) = if flow.draft { (prepare_draft_flow(flow), vec![], vec![]) } else { compile_flow(&state, flow)? };
state.db.replace_flow_outputs(&flow, &schedules, &automations)?;
for zone_id in schedules.iter().map(|s| s.zone_id.as_str()).collect::<std::collections::HashSet<_>>() { refresh_zone_override_boundary(&state, zone_id).await?; }
state.log("info", "flow.created", &format!("Created Flow {}", flow.name), json!({"flow_id": flow.id, "revision": flow.revision, "schedules": schedules.len(), "automations": automations.len()}));
@@ -567,7 +590,7 @@ async fn create_flow(State(state): State<AppState>, Json(input): Json<FlowInput>
}
async fn update_flow(State(state): State<AppState>, Path(id): Path<String>, Json(input): Json<FlowInput>) -> Result<Json<crate::models::Flow>, AppError> {
validate_flow_graph(&input)?;
if input.draft { validate_flow_draft_graph(&input)?; } else { validate_flow_graph(&input)?; }
let _configuration_guard = state.lock_configuration_operation().await;
let _automation_guard = state.lock_automation_operation().await;
let _schedule_guard = state.lock_schedule_operation().await;
@@ -578,7 +601,7 @@ async fn update_flow(State(state): State<AppState>, Path(id): Path<String>, Json
let old_zone_ids: std::collections::HashSet<String> = state.db.list_schedules()?.into_iter().filter(|s| s.flow_id.as_deref() == Some(id.as_str())).map(|s| s.zone_id).collect();
let next_revision = existing.revision.saturating_add(1).max(1);
let flow = flow_from_input(id, input, existing.created_at, next_revision);
let (flow, schedules, automations) = compile_flow(&state, flow)?;
let (flow, schedules, automations) = if flow.draft { (prepare_draft_flow(flow), vec![], vec![]) } else { compile_flow(&state, flow)? };
state.db.replace_flow_outputs(&flow, &schedules, &automations)?;
let mut zone_ids = old_zone_ids; zone_ids.extend(schedules.iter().map(|s| s.zone_id.clone()));
for zone_id in zone_ids { refresh_zone_override_boundary(&state, &zone_id).await?; }
@@ -610,6 +633,7 @@ async fn export_flow(State(state): State<AppState>, Path(id): Path<String>) -> R
"flow": {
"name": flow.name,
"enabled": flow.enabled,
"draft": flow.draft,
"description": flow.description,
"nodes": flow.nodes,
"edges": flow.edges
@@ -624,13 +648,13 @@ async fn import_flow(State(state): State<AppState>, Json(document): Json<Value>)
let payload = document.get("flow").cloned().unwrap_or(document);
let mut input: FlowInput = serde_json::from_value(payload).map_err(|err| AppError::BadRequest(format!("invalid Flow import: {err}")))?;
input.expected_revision = None;
validate_flow_graph(&input)?;
if input.draft { validate_flow_draft_graph(&input)?; } else { validate_flow_graph(&input)?; }
let _configuration_guard = state.lock_configuration_operation().await;
let _automation_guard = state.lock_automation_operation().await;
let _schedule_guard = state.lock_schedule_operation().await;
let _cycle_guard = state.lock_zone_control_cycle().await;
let flow = flow_from_input(Uuid::new_v4().to_string(), input, Utc::now(), 1);
let (flow, schedules, automations) = compile_flow(&state, flow)?;
let (flow, schedules, automations) = if flow.draft { (prepare_draft_flow(flow), vec![], vec![]) } else { compile_flow(&state, flow)? };
state.db.replace_flow_outputs(&flow, &schedules, &automations)?;
for zone_id in schedules.iter().map(|s| s.zone_id.as_str()).collect::<std::collections::HashSet<_>>() { refresh_zone_override_boundary(&state, zone_id).await?; }
state.log("info", "flow.imported", &format!("Imported Flow {}", flow.name), json!({"flow_id": flow.id, "revision": flow.revision}));
@@ -699,7 +723,7 @@ async fn simulate_flow(State(state): State<AppState>, Json(input): Json<FlowSimu
};
let preview_id = input.flow_id.clone().filter(|v| !v.trim().is_empty()).unwrap_or_else(|| format!("dry-run-{}", Uuid::new_v4()));
let preview = crate::models::Flow {
id: preview_id.clone(), name: input.flow.name.trim().into(), enabled: input.flow.enabled, description: input.flow.description.trim().into(),
id: preview_id.clone(), name: input.flow.name.trim().into(), enabled: input.flow.enabled, draft: false, description: input.flow.description.trim().into(),
nodes: input.flow.nodes.clone(), edges: input.flow.edges.clone(), summary: String::new(), compiled_schedule_ids: vec![], compiled_automation_ids: vec![], revision: 0,
created_at: Utc::now(), updated_at: Utc::now(),
};
@@ -749,3 +773,50 @@ async fn flow_logs(State(state): State<AppState>, Path(id): Path<String>, Query(
}).take(limit).collect::<Vec<_>>();
Ok(Json(json!({"events": events})))
}
#[cfg(test)]
mod flow_draft_tests {
use super::*;
fn unfinished_input() -> FlowInput {
FlowInput {
name: "Unfinished".into(),
enabled: true,
draft: true,
description: String::new(),
nodes: vec![crate::models::FlowNode {
id: "condition".into(),
kind: "constant".into(),
x: 0.0,
y: 0.0,
config: json!({"value": true}),
}],
edges: vec![],
expected_revision: None,
}
}
#[test]
fn draft_graph_accepts_flow_without_action() {
let input = unfinished_input();
assert!(validate_flow_draft_graph(&input).is_ok());
assert!(validate_flow_graph(&input).is_err());
}
#[test]
fn draft_graph_still_rejects_missing_edge_endpoints() {
let mut input = unfinished_input();
input.edges.push(crate::models::FlowEdge { id: "edge".into(), from: "condition".into(), to: "missing".into() });
assert!(validate_flow_draft_graph(&input).is_err());
}
#[test]
fn draft_flow_is_forced_disabled_and_has_no_compiled_outputs() {
let flow = flow_from_input("draft-id".into(), unfinished_input(), chrono::Utc::now(), 1);
let flow = prepare_draft_flow(flow);
assert!(flow.draft);
assert!(!flow.enabled);
assert!(flow.compiled_schedule_ids.is_empty());
assert!(flow.compiled_automation_ids.is_empty());
}
}
src/api/settings.rs
+7
View File
@@ -202,6 +202,13 @@ fn validate_configuration_export(export: &ConfigurationExport) -> Result<(), App
let schedules: std::collections::HashSet<&str> = export.schedules.iter().map(|item| item.id.as_str()).collect();
let automations: std::collections::HashSet<&str> = export.automations.iter().map(|item| item.id.as_str()).collect();
let flows: std::collections::HashSet<&str> = export.flows.iter().map(|item| item.id.as_str()).collect();
let draft_flows: std::collections::HashSet<&str> = export.flows.iter().filter(|item| item.draft).map(|item| item.id.as_str()).collect();
if export.flows.iter().any(|item| item.draft && (item.enabled || !item.compiled_schedule_ids.is_empty() || !item.compiled_automation_ids.is_empty()))
|| export.schedules.iter().any(|item| item.flow_id.as_deref().is_some_and(|id| draft_flows.contains(id)))
|| export.automations.iter().any(|item| item.flow_id.as_deref().is_some_and(|id| draft_flows.contains(id)))
{
return Err(AppError::BadRequest("import contains an executable Flow draft".into()));
}
if devices.len() != export.devices.len() || zones.len() != export.zones.len()
|| schedules.len() != export.schedules.len() || automations.len() != export.automations.len() || flows.len() != export.flows.len()
|| devices.contains("") || zones.contains("") || schedules.contains("") || automations.contains("") || flows.contains("")
src/models/flow.rs
+3
View File
@@ -37,6 +37,9 @@ pub struct Flow {
pub name: String,
#[serde(default = "default_true")]
pub enabled: bool,
/// Work-in-progress Flow saved without executable outputs. Drafts are always disabled.
#[serde(default)]
pub draft: bool,
#[serde(default)]
pub description: String,
#[serde(default)]