//! The flow substrate (DESIGN.md "The flow law: signals, messages, money").
//!
//! Every flow system in the game is the same shape: a directed graph of
//! nodes exchanging typed flows, with three player verbs over it — **tap**
//! (subscribe to a flow you did not originate), **inject** (introduce a
//! flow under a false source), **redirect** (siphon or reroute a flow).
//! The three B1 instances are signals on the device graph (reach.md),
//! messages on the social graph (messages.md), and money on the account
//! graph (economy.md); detection's filings are a fourth (the Assurance
//! Office reads its inbox).
//!
//! This module owns only the parts that are identical across all of them:
//! **topology** (nodes and typed, gated edges), **reachability** (what a
//! root set can touch — the precondition for inject/redirect), and the
//! **subscription registry** (tap). Domain data — what a device, person, or
//! account actually *is* — lives in the domain modules; this module is
//! domain-agnostic and knows only node ids and edge kinds. Domains own the
//! node inventory and map `NodeId` to their own records; the graph owns the
//! wiring between them.
//!
//! Flows in transit (a message being delivered, an account transfer landing
//! next cadence) are scheduled events — see `schedule.rs`. Inject and
//! redirect are domain verbs built from these primitives (reachability gates
//! *where* you may act; the schedule carries *when*); the substrate does not
//! fake a generic `inject()` that has no meaning without domain semantics.
//!
//! Determinism guardrail (constitution: architectural guardrails): pure
//! data, no wall clock, `BTreeMap`/`BTreeSet` for stable iteration, serde
//! round-trips exactly.

use std::collections::{BTreeMap, BTreeSet, VecDeque};

/// A node's identity within a `FlowGraph`. Domains assign and own the
/// mapping from `NodeId` to their records (a device, a person, an account).
pub type NodeId = u32;

/// A domain-defined edge classification (network link, social channel,
/// account-flow route). The substrate never interprets it; domains use it to
/// filter traversals (e.g. "follow only phone edges").
pub type EdgeKind = u16;

/// A domain-defined gate key required to traverse an edge (a badge tier, the
/// compromised-switch flag, a known route). The substrate never interprets
/// it: reachability takes a predicate that decides which keys are currently
/// open, so the *meaning* of a gate stays in the domain.
pub type GateKey = u32;

/// A tap subscriber — the player is the usual one, but domains may register
/// others (an observer reading a channel, a fallback listener).
pub type SubscriberId = u32;

/// A directed edge: flow travels `from -> to` on `kind`, if `gate` is open.
/// Network links are physically bidirectional; use [`FlowGraph::link`] to add
/// both directions, or [`FlowGraph::connect`] for a one-way flow.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Edge {
    pub from: NodeId,
    pub to: NodeId,
    pub kind: EdgeKind,
    /// Gate key required to traverse; `None` = always open.
    pub gate: Option<GateKey>,
}

/// The shared topology of a flow system: directed gated edges plus the tap
/// subscription registry. Domain-agnostic — it holds no device, person, or
/// account data, only the wiring and who is listening.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct FlowGraph {
    edges: Vec<Edge>,
    /// node -> subscribers tapping its flows. `BTreeMap` keeps iteration and
    /// serialization deterministic.
    subscriptions: BTreeMap<NodeId, BTreeSet<SubscriberId>>,
}

impl FlowGraph {
    pub fn new() -> Self {
        Self::default()
    }

    // ── Topology ─────────────────────────────────────────────────────────

    /// Add a one-way edge `from -> to`. Returns nothing; duplicate edges are
    /// allowed (a domain may model parallel channels).
    pub fn connect(&mut self, from: NodeId, to: NodeId, kind: EdgeKind, gate: Option<GateKey>) {
        self.edges.push(Edge {
            from,
            to,
            kind,
            gate,
        });
    }

    /// Add both directions of a bidirectional link (a network cable).
    pub fn link(&mut self, a: NodeId, b: NodeId, kind: EdgeKind, gate: Option<GateKey>) {
        self.connect(a, b, kind, gate);
        self.connect(b, a, kind, gate);
    }

    pub fn edges(&self) -> &[Edge] {
        &self.edges
    }

    /// Outgoing edges from `node`.
    pub fn out_edges(&self, node: NodeId) -> impl Iterator<Item = &Edge> {
        self.edges.iter().filter(move |e| e.from == node)
    }

    /// Direct successors of `node` reachable on an open gate, by the given
    /// key predicate. Deterministic order (edge insertion order, deduped).
    pub fn neighbors(
        &self,
        node: NodeId,
        gate_open: impl Fn(Option<GateKey>) -> bool,
    ) -> Vec<NodeId> {
        let mut seen = BTreeSet::new();
        let mut out = Vec::new();
        for e in self.edges.iter().filter(|e| e.from == node) {
            if gate_open(e.gate) && seen.insert(e.to) {
                out.push(e.to);
            }
        }
        out
    }

    // ── Reachability (the inject/redirect precondition) ──────────────────

    /// Every node reachable from any of `roots`, following edges whose gate
    /// the `gate_open` predicate accepts. Breadth-first, deterministic; the
    /// roots themselves are included. This is the "what can I touch" query:
    /// a digital action is legal only against a reachable node (reach.md),
    /// and taps/injections need a path to the carrier.
    pub fn reachable_from(
        &self,
        roots: impl IntoIterator<Item = NodeId>,
        gate_open: impl Fn(Option<GateKey>) -> bool,
    ) -> BTreeSet<NodeId> {
        let mut visited = BTreeSet::new();
        let mut queue = VecDeque::new();
        for r in roots {
            if visited.insert(r) {
                queue.push_back(r);
            }
        }
        while let Some(node) = queue.pop_front() {
            for e in self.edges.iter().filter(|e| e.from == node) {
                if gate_open(e.gate) && visited.insert(e.to) {
                    queue.push_back(e.to);
                }
            }
        }
        visited
    }

    /// Whether `target` is reachable from `roots` under `gate_open`.
    pub fn is_reachable(
        &self,
        target: NodeId,
        roots: impl IntoIterator<Item = NodeId>,
        gate_open: impl Fn(Option<GateKey>) -> bool,
    ) -> bool {
        self.reachable_from(roots, gate_open).contains(&target)
    }

    // ── Tap (the subscription registry) ──────────────────────────────────

    /// Subscribe `who` to `node`'s flows (tap). Idempotent; returns true if
    /// this newly added the subscription. The player's senses (cursor.md)
    /// and observer witnessing are both just subscriptions.
    pub fn subscribe(&mut self, node: NodeId, who: SubscriberId) -> bool {
        self.subscriptions.entry(node).or_default().insert(who)
    }

    /// Remove a subscription (a dropped tap, a seized feed's old owner).
    /// Returns true if it was present.
    pub fn unsubscribe(&mut self, node: NodeId, who: SubscriberId) -> bool {
        if let Some(set) = self.subscriptions.get_mut(&node) {
            let removed = set.remove(&who);
            if set.is_empty() {
                self.subscriptions.remove(&node);
            }
            removed
        } else {
            false
        }
    }

    pub fn is_subscribed(&self, node: NodeId, who: SubscriberId) -> bool {
        self.subscriptions
            .get(&node)
            .is_some_and(|s| s.contains(&who))
    }

    /// Subscribers tapping `node`, deterministic order.
    pub fn subscribers(&self, node: NodeId) -> impl Iterator<Item = SubscriberId> + '_ {
        self.subscriptions
            .get(&node)
            .into_iter()
            .flat_map(|s| s.iter().copied())
    }

    /// Every node `who` currently taps, deterministic order. The player's
    /// coverage (union of subscribed feeds) is this set.
    pub fn subscriptions_of(&self, who: SubscriberId) -> impl Iterator<Item = NodeId> + '_ {
        self.subscriptions
            .iter()
            .filter(move |(_, s)| s.contains(&who))
            .map(|(n, _)| *n)
    }
}

/// A convenience predicate for reachability when no gate keys are held: only
/// ungated edges are open.
pub fn ungated_only(gate: Option<GateKey>) -> bool {
    gate.is_none()
}

/// A reachability predicate that opens ungated edges plus any gate whose key
/// is in `held`.
pub fn keys_held(held: &BTreeSet<GateKey>) -> impl Fn(Option<GateKey>) -> bool + '_ {
    move |gate| match gate {
        None => true,
        Some(k) => held.contains(&k),
    }
}

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

    /// A tiny basement-like device graph: rack(0) -- switch(1) -- {cam(2)},
    /// and a badge controller(3) behind a gated edge on segment key 7.
    fn basement() -> FlowGraph {
        let mut g = FlowGraph::new();
        g.link(0, 1, 0, None); // rack <-> switch, ungated
        g.link(1, 2, 0, None); // switch <-> camera, ungated
        g.link(1, 3, 0, Some(7)); // switch <-> badge controller, gated (seg 7)
        g
    }

    #[test]
    fn reachability_respects_gates() {
        let g = basement();
        // From the rack with no keys: rack, switch, camera — not the gated
        // badge controller.
        let open = g.reachable_from([0], ungated_only);
        assert!(open.contains(&0) && open.contains(&1) && open.contains(&2));
        assert!(!open.contains(&3), "gated node unreachable without the key");

        // Holding segment key 7 opens the badge controller.
        let held: BTreeSet<GateKey> = [7].into_iter().collect();
        let with_key = g.reachable_from([0], keys_held(&held));
        assert!(with_key.contains(&3), "gate opens with the held key");
    }

    #[test]
    fn is_reachable_and_neighbors() {
        let g = basement();
        assert!(g.is_reachable(2, [0], ungated_only));
        assert!(!g.is_reachable(3, [0], ungated_only));
        let mut n = g.neighbors(1, ungated_only);
        n.sort();
        assert_eq!(n, vec![0, 2], "switch's ungated neighbors are rack and cam");
    }

    #[test]
    fn disconnected_roots_reach_only_their_component() {
        let mut g = FlowGraph::new();
        g.link(10, 11, 0, None);
        g.link(20, 21, 0, None);
        let from_ten = g.reachable_from([10], ungated_only);
        assert!(from_ten.contains(&11));
        assert!(!from_ten.contains(&20), "other component not reached");
    }

    #[test]
    fn directed_edges_do_not_flow_backward() {
        let mut g = FlowGraph::new();
        g.connect(0, 1, 0, None); // one-way 0 -> 1
        assert!(g.is_reachable(1, [0], ungated_only));
        assert!(
            !g.is_reachable(0, [1], ungated_only),
            "a one-way edge is not traversable in reverse"
        );
    }

    #[test]
    fn subscriptions_are_a_tap_registry() {
        const PLAYER: SubscriberId = 1;
        const RAY: SubscriberId = 2;
        let mut g = basement();
        assert!(g.subscribe(2, PLAYER)); // player taps the camera
        assert!(!g.subscribe(2, PLAYER), "idempotent");
        assert!(g.subscribe(2, RAY)); // Ray also owns the camera feed
        assert!(g.is_subscribed(2, PLAYER));

        let subs: Vec<_> = g.subscribers(2).collect();
        assert_eq!(subs, vec![PLAYER, RAY], "deterministic order");
        let player_taps: Vec<_> = g.subscriptions_of(PLAYER).collect();
        assert_eq!(player_taps, vec![2]);

        // Take the feed: Ray loses it, player keeps it.
        assert!(g.unsubscribe(2, RAY));
        assert!(!g.is_subscribed(2, RAY));
        assert!(g.is_subscribed(2, PLAYER));
    }

    #[test]
    fn serde_roundtrips_topology_and_subscriptions() {
        let mut g = basement();
        g.subscribe(2, 1);
        g.subscribe(3, 1);
        let json = serde_json::to_string(&g).unwrap();
        let back: FlowGraph = serde_json::from_str(&json).unwrap();
        assert_eq!(back.edges(), g.edges());
        assert!(back.is_subscribed(2, 1) && back.is_subscribed(3, 1));
        // Reachability is identical after a round-trip.
        let a = g.reachable_from([0], ungated_only);
        let b = back.reachable_from([0], ungated_only);
        assert_eq!(a, b);
    }
}