///|
/// Core workflow graph node for bitflow execution planning.
pub struct FlowNode {
  id : String
  required : Bool
  depends_on : Array[String]
}

///|
pub fn new_node(
  id : String,
  depends_on : Array[String],
  required? : Bool = true,
) -> FlowNode {
  { id, required, depends_on }
}

///|
fn node_id_set(nodes : Array[FlowNode]) -> Map[String, Bool] {
  let ids : Map[String, Bool] = {}
  for node in nodes {
    ids[node.id] = true
  }
  ids
}

///|
fn node_map(nodes : Array[FlowNode]) -> Map[String, FlowNode] {
  let result : Map[String, FlowNode] = {}
  for node in nodes {
    result[node.id] = node
  }
  result
}

///|
pub fn topological_nodes(nodes : Array[FlowNode]) -> Array[FlowNode] {
  let ids = node_id_set(nodes)
  let mapped = node_map(nodes)
  let indegree : Map[String, Int] = {}
  let dependents : Map[String, Array[String]] = {}
  for node in nodes {
    indegree[node.id] = 0
    dependents[node.id] = []
  }
  for node in nodes {
    let mut deg = indegree.get(node.id).unwrap_or(0)
    for dep in node.depends_on {
      if ids.get(dep) is Some(_) {
        deg += 1
        let next = dependents.get(dep).unwrap_or([])
        next.push(node.id)
        dependents[dep] = next
      }
    }
    indegree[node.id] = deg
  }
  let queue : Array[String] = []
  for node in nodes {
    if indegree.get(node.id).unwrap_or(0) == 0 {
      queue.push(node.id)
    }
  }
  let order_ids : Array[String] = []
  let visited : Map[String, Bool] = {}
  while queue.length() > 0 {
    let current = queue.pop().unwrap_or("")
    if current.length() == 0 || visited.get(current) is Some(_) {
      continue
    }
    visited[current] = true
    order_ids.push(current)
    let followers = dependents.get(current).unwrap_or([])
    for follower in followers {
      let next_deg = indegree.get(follower).unwrap_or(0) - 1
      indegree[follower] = next_deg
      if next_deg == 0 {
        queue.push(follower)
      }
    }
  }
  let ordered : Array[FlowNode] = []
  for id in order_ids {
    if mapped.get(id) is Some(node) {
      ordered.push(node)
    }
  }
  for node in nodes {
    if visited.get(node.id) is None {
      ordered.push(node)
    }
  }
  ordered
}

///|
pub fn has_dependency_cycle(nodes : Array[FlowNode]) -> Bool {
  let ids = node_id_set(nodes)
  let indegree : Map[String, Int] = {}
  let dependents : Map[String, Array[String]] = {}
  for node in nodes {
    indegree[node.id] = 0
    dependents[node.id] = []
  }
  for node in nodes {
    let mut deg = indegree.get(node.id).unwrap_or(0)
    for dep in node.depends_on {
      if ids.get(dep) is Some(_) {
        deg += 1
        let next = dependents.get(dep).unwrap_or([])
        next.push(node.id)
        dependents[dep] = next
      }
    }
    indegree[node.id] = deg
  }
  let queue : Array[String] = []
  for node in nodes {
    if indegree.get(node.id).unwrap_or(0) == 0 {
      queue.push(node.id)
    }
  }
  let mut visited_count = 0
  while queue.length() > 0 {
    let current = queue.pop().unwrap_or("")
    if current.length() == 0 {
      continue
    }
    visited_count += 1
    let followers = dependents.get(current).unwrap_or([])
    for follower in followers {
      let next_deg = indegree.get(follower).unwrap_or(0) - 1
      indegree[follower] = next_deg
      if next_deg == 0 {
        queue.push(follower)
      }
    }
  }
  visited_count < nodes.length()
}

///|
pub fn graph_issues(nodes : Array[FlowNode]) -> Array[String] {
  let issues : Array[String] = []
  let seen_ids : Map[String, Bool] = {}
  for node in nodes {
    if node.id.length() == 0 {
      issues.push("node id is empty")
    } else if seen_ids.get(node.id) is Some(_) {
      issues.push("duplicate node id '\{node.id}'")
    } else {
      seen_ids[node.id] = true
    }
  }
  let ids = node_id_set(nodes)
  for node in nodes {
    for dep in node.depends_on {
      if ids.get(dep) is None {
        issues.push("node '\{node.id}' depends on unknown node '\{dep}'")
      }
    }
  }
  if has_dependency_cycle(nodes) {
    issues.push("dependency graph has a cycle")
  }
  issues
}

///|
pub fn expand_affected_nodes(
  nodes : Array[FlowNode],
  initial : Map[String, Bool],
) -> Map[String, Bool] {
  let known_ids = node_id_set(nodes)
  let neighbors : Map[String, Array[String]] = {}
  for node in nodes {
    neighbors[node.id] = []
  }
  for node in nodes {
    let seen_neighbor : Map[String, Bool] = {}
    let next = neighbors.get(node.id).unwrap_or([])
    for dep in node.depends_on {
      if known_ids.get(dep) is Some(_) && seen_neighbor.get(dep) is None {
        next.push(dep)
        seen_neighbor[dep] = true
      }
    }
    neighbors[node.id] = next
    for dep in node.depends_on {
      if known_ids.get(dep) is Some(_) {
        let dep_next = neighbors.get(dep).unwrap_or([])
        if dep_next.contains(node.id) {
          continue
        }
        dep_next.push(node.id)
        neighbors[dep] = dep_next
      }
    }
  }
  let affected : Map[String, Bool] = {}
  let queue : Array[String] = []
  for node in nodes {
    if initial.get(node.id) is Some(_) {
      affected[node.id] = true
      queue.push(node.id)
    }
  }
  while queue.length() > 0 {
    let current = queue.pop().unwrap_or("")
    if current.length() == 0 {
      continue
    }
    for neighbor in neighbors.get(current).unwrap_or([]) {
      if affected.get(neighbor) is Some(_) {
        continue
      }
      affected[neighbor] = true
      queue.push(neighbor)
    }
  }
  affected
}

///|
pub fn flow_cache_key(task : String, node_id : String) -> String {
  task + "::" + node_id
}

///|
pub fn flow_fingerprint(
  task : String,
  task_cmd : String,
  node : FlowNode,
  signatures : Map[String, String],
) -> String {
  let buf = StringBuilder::new()
  buf.write_string("task=")
  buf.write_string(task)
  buf.write_string("\nnode=")
  buf.write_string(node.id)
  buf.write_string("\ncommand=")
  buf.write_string(task_cmd)
  buf.write_string("\nself=")
  buf.write_string(signatures.get(node.id).unwrap_or("missing"))
  for dep in node.depends_on {
    buf.write_string("\ndep:")
    buf.write_string(dep)
    buf.write_string("=")
    buf.write_string(signatures.get(dep).unwrap_or("missing"))
  }
  buf.to_string()
}