topodb 0.0.1

//! Scoped reads: point lookup, label scan, and k-hop temporal traversal.
//! Every entry point here takes a `&ScopeSet` (directly, or via
//! `TraversalQuery::scopes`) — there is no unscoped read path.

use crate::db::Db;
use crate::error::TopoError;
use crate::graph::Snapshot;
use crate::ids::{EdgeId, NodeId, ScopeSet};
use crate::state::{EdgeRecord, NodeRecord};
use smol_str::SmolStr;
use std::collections::{HashSet, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};

/// Which adjacency to walk from each frontier node.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Direction {
    Out,
    In,
    Both,
}

/// A bounded, scoped, temporal breadth-first traversal request.
#[derive(Debug, Clone)]
pub struct TraversalQuery {
    pub scopes: ScopeSet,
    pub seeds: Vec<NodeId>,
    /// Hop budget. Must be `1..=4` — `0` or `>4` is rejected.
    pub max_hops: u8,
    /// `None` matches every edge type.
    pub edge_types: Option<Vec<SmolStr>>,
    pub direction: Direction,
    /// `None` means "now" — read once, at traversal start, from the wall
    /// clock (this is a read path; only writes must never embed wall-clock
    /// time).
    pub as_of: Option<i64>,
}

/// Result of a traversal: every in-scope seed plus everything reached,
/// deduped, with the full edge records (from the snapshot's `edges` map) for
/// every traversed edge.
#[derive(Debug, Clone, Default)]
pub struct Subgraph {
    pub nodes: Vec<NodeRecord>,
    pub edges: Vec<EdgeRecord>,
}

fn now_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system clock before UNIX epoch")
        .as_millis() as i64
}

impl Db {
    /// Point lookup, scoped: `None` both when the node doesn't exist and when
    /// it exists but is outside `scopes` — the two are indistinguishable to
    /// the caller, by design (no way to detect out-of-scope data via absence
    /// timing/shape).
    #[must_use]
    pub fn node(&self, scopes: &ScopeSet, id: NodeId) -> Option<NodeRecord> {
        let snap = self.snapshot();
        snap.nodes.get(&id).filter(|n| scopes.contains(n.scope)).cloned()
    }

    /// All nodes with the given `label`, restricted to `scopes`. Order is
    /// unspecified (snapshot iteration order).
    #[must_use]
    pub fn nodes_by_label(&self, scopes: &ScopeSet, label: &str) -> Vec<NodeRecord> {
        let snap = self.snapshot();
        snap.nodes
            .values()
            .filter(|n| n.label == label && scopes.contains(n.scope))
            .cloned()
            .collect()
    }

    /// Bounded (`1..=4` hops), scoped, temporal BFS from `q.seeds` over a
    /// single `snapshot()` — a consistent view for the whole traversal.
    pub fn traverse(&self, q: &TraversalQuery) -> Result<Subgraph, TopoError> {
        if q.max_hops == 0 || q.max_hops > 4 {
            return Err(TopoError::Rejected(format!(
                "max_hops must be in 1..=4, got {}",
                q.max_hops
            )));
        }

        let snap = self.snapshot();
        let t = q.as_of.unwrap_or_else(now_ms);

        let mut visited: HashSet<NodeId> = HashSet::new();
        let mut result_edges: HashSet<EdgeId> = HashSet::new();
        let mut frontier: VecDeque<(NodeId, u8)> = VecDeque::new();

        for &seed in &q.seeds {
            if let Some(n) = snap.nodes.get(&seed) {
                if q.scopes.contains(n.scope) && visited.insert(seed) {
                    frontier.push_back((seed, 0));
                }
            }
        }

        while let Some((node, hop)) = frontier.pop_front() {
            if hop >= q.max_hops {
                continue;
            }
            for entry in adjacency_entries(&snap, node, q.direction) {
                if !edge_traversable(&q.scopes, q.edge_types.as_deref(), entry, t) {
                    continue;
                }
                let Some(other) = snap.nodes.get(&entry.other) else { continue };
                if !q.scopes.contains(other.scope) {
                    continue;
                }
                result_edges.insert(entry.edge);
                if visited.insert(entry.other) {
                    frontier.push_back((entry.other, hop + 1));
                }
            }
        }

        let nodes = visited
            .iter()
            .filter_map(|id| snap.nodes.get(id).cloned())
            .collect();
        let edges = result_edges
            .iter()
            .filter_map(|id| snap.edges.get(id).cloned())
            .collect();

        Ok(Subgraph { nodes, edges })
    }
}

/// Collects the `AdjEntry`s to walk from `node` given `direction`. `Both`
/// walks `out` and `inn`; an edge reachable both ways (i.e. it appears in
/// both maps, which never happens for the *same* traversal step since one
/// side is keyed by `from` and the other by `to`) still only ever
/// contributes one `AdjEntry` per direction here — de-duplication of the
/// resulting node/edge id happens via `visited`/`result_edges` in the caller.
fn adjacency_entries(
    snap: &Snapshot,
    node: NodeId,
    direction: Direction,
) -> Vec<&crate::graph::AdjEntry> {
    let mut out = Vec::new();
    if matches!(direction, Direction::Out | Direction::Both) {
        if let Some(v) = snap.out.get(&node) {
            out.extend(v.iter());
        }
    }
    if matches!(direction, Direction::In | Direction::Both) {
        if let Some(v) = snap.inn.get(&node) {
            out.extend(v.iter());
        }
    }
    out
}

fn edge_traversable(
    scopes: &ScopeSet,
    edge_types: Option<&[SmolStr]>,
    entry: &crate::graph::AdjEntry,
    t: i64,
) -> bool {
    if !scopes.contains(entry.scope) {
        return false;
    }
    if let Some(types) = edge_types {
        if !types.iter().any(|ty| ty == &entry.ty) {
            return false;
        }
    }
    entry.valid_from <= t && entry.valid_to.is_none_or(|vt| t < vt)
}