live

A live builder for incremental graph construction with label-to-ID registry.

Unlike the static labeled builder which follows a “Build-Freeze-Analyze” pattern, LiveBuilder provides a Transaction-style API that tracks pending changes. This allows efficient synchronization of an existing Graph with new labeled edges in $O(\Delta E)$ time, where $\Delta E$ is the number of new edges since last sync.

Use Cases

REPL environments: Incrementally build and analyze graphs
UI editors: Add nodes/edges interactively without rebuilding
Streaming data: Ingest new relationships as they arrive
Large graphs: Avoid $O(E)$ rebuild for single-edge updates

Guarantees

ID Stability: Once a label is mapped to a NodeId, that mapping is immutable
Idempotency: Calling sync with no pending changes is effectively free
Opaque Integration: Uses the same ID generation as static builders

Important: Managing the Pending Queue

The LiveBuilder queues changes in memory until sync() is called. In streaming scenarios, if you add edges continuously without syncing, the pending queue will grow unbounded and consume memory.

Best Practice: Sync periodically based on your workload:

// For high-frequency streaming (e.g., Kafka consumer)
// Sync every N messages or every T seconds
let #(builder, graph) = case live.pending_count(builder) > 1000 {
  True -> live.sync(builder, graph)
  False -> #(builder, graph)
}

// For batch processing
// Build up a batch, then sync once
let builder = list.fold(batch, builder, fn(b, edge) {
  live.add_edge(b, edge.0, edge.1, edge.2)
})
let #(builder, graph) = live.sync(builder, graph)

Recovery: If you need to discard pending changes without applying them, use purge_pending() (abandon changes) or checkpoint() (keep registry).

Limitations

Memory: Pending changes are stored in memory until synced
No Persistence: The pending queue is lost if the process crashes
Single-threaded: Not designed for concurrent updates from multiple actors

Example

import yog/builder/live
import yog/pathfinding/dijkstra as pathfinding
import gleam/int

// Initial setup - build base graph
let builder = live.new() |> live.add_edge("A", "B", 10)
let #(builder, graph) = live.sync(builder, yog.directed())

// Incremental update - add new edge efficiently
let builder = builder |> live.add_edge("B", "C", 5)
let #(builder, graph) = live.sync(builder, graph)  // O(1) for just this edge!

// Use with algorithms - get IDs from registry
let assert Ok(a_id) = live.get_id(builder, "A")
let assert Ok(c_id) = live.get_id(builder, "C")
let path = pathfinding.shortest_path(graph, a_id, c_id, ...)

Types

LiveBuilder

opaque </>

Opaque type for the live builder.

Tracks a registry of label-to-ID mappings and pending transitions that haven’t been applied to a graph yet.

pub opaque type LiveBuilder(n, e)

Values

add_edge

</>

pub fn add_edge(
  builder: LiveBuilder(n, e),
  from src_label: n,
  to dst_label: n,
  with weight: e,
) -> LiveBuilder(n, e)

Adds an edge between two labeled nodes.

If either label doesn’t exist in the registry, new nodes are created. The edge is queued as a pending transition and will be applied on next sync.

Example

let builder =
  live.new()
  |> live.add_edge("home", "work", 10)
  |> live.add_edge("work", "gym", 5)

Complexity

$O(\log N)$ per label lookup/insert

add_simple_edge

</>

pub fn add_simple_edge(
  builder: LiveBuilder(n, Int),
  from src_label: n,
  to dst_label: n,
) -> LiveBuilder(n, Int)

Adds a simple edge with weight 1 between two labeled nodes.

This is a convenience function for graphs with integer weights where a default weight of 1 is appropriate (e.g., unweighted graphs, hop counts).

Example

let builder =
  live.directed()
  |> live.add_simple_edge("A", "B")
  |> live.add_simple_edge("B", "C")

add_unweighted_edge

</>

pub fn add_unweighted_edge(
  builder: LiveBuilder(n, Nil),
  from src_label: n,
  to dst_label: n,
) -> LiveBuilder(n, Nil)

Adds an unweighted edge between two labeled nodes.

This is a convenience function for graphs where edges have no meaningful weight. Uses Nil as the edge data type.

Example

let builder: live.LiveBuilder(String, Nil) =
  live.directed()
  |> live.add_unweighted_edge("A", "B")

all_labels

</>

pub fn all_labels(builder: LiveBuilder(n, e)) -> List(n)

Returns all labels that have been registered.

Example

let labels = live.all_labels(builder)
// ["A", "B", "C"]

checkpoint

</>

pub fn checkpoint(
  builder: LiveBuilder(n, e),
) -> LiveBuilder(n, e)

Marks a checkpoint by discarding pending transitions.

This is useful when you want to “commit” progress and discard the pending queue without applying it to a graph. The registry is preserved.

Compare: purge_pending() has the same effect but different intent. Use checkpoint() for progress tracking, purge_pending() for error recovery.

Example

let builder =
  live.new()
  |> live.add_edge("A", "B", 10)
  |> live.checkpoint()  // Discard the pending edge

// builder now has no pending changes
let #(builder, graph) = live.sync(builder, yog.directed())
// graph is empty - the edge was discarded

directed

</>

pub fn directed() -> LiveBuilder(n, e)

Creates a new live builder with a directed graph type in mind.

This is a convenience function - the builder itself doesn’t store the graph type, but it helps document intent.

Example

let builder = live.directed()

from_labeled

</>

pub fn from_labeled(
  labeled_builder: labeled.Builder(n, e),
) -> LiveBuilder(n, e)

Creates a live builder from an existing labeled builder.

This allows migration from static to incremental building. The pending list starts empty - call sync separately to apply changes.

Note: This uses the labeled builder’s exported registry. The label→ID mappings are preserved exactly.

Example

import yog/builder/labeled

// Start with static builder
let static = labeled.directed() |> labeled.add_edge("A", "B", 10)
let graph = labeled.to_graph(static)

// Convert to live for incremental updates
let live_builder = live.from_labeled(static)
let live_builder = live.add_edge(live_builder, "B", "C", 5)
let #(live_builder, graph) = live.sync(live_builder, graph)

get_id

</>

pub fn get_id(
  builder: LiveBuilder(n, e),
  label: n,
) -> Result(Int, Nil)

Looks up the node ID for a given label.

Returns Ok(id) if the label has been registered, Error(Nil) if not. Use this to get node IDs for use with graph algorithms.

Note: After remove_node, this will return Error(Nil) for that label.

Example

let assert Ok(home_id) = live.get_id(builder, "home")
let path = dijkstra.shortest_path(graph, home_id, ...)

Complexity

$O(\log N)$ for dict lookup

new

</>

pub fn new() -> LiveBuilder(n, e)

Creates a new empty live builder.

Example

let builder = live.new()

node_count

</>

pub fn node_count(builder: LiveBuilder(n, e)) -> Int

Returns the number of registered labels (nodes).

Example

live.node_count(builder)  // 5

pending_count

</>

pub fn pending_count(builder: LiveBuilder(n, e)) -> Int

Returns the number of pending transitions.

Useful for debugging or deciding whether to sync based on batch size.

Example

// Sync every 1000 changes to avoid unbounded growth
case live.pending_count(builder) > 1000 {
  True -> live.sync(builder, graph)
  False -> #(builder, graph)
}

purge_pending

</>

pub fn purge_pending(
  builder: LiveBuilder(n, e),
) -> LiveBuilder(n, e)

Discards all pending transitions without applying them.

This is a “hard reset” that abandons all queued changes. The registry (label→ID mappings) is preserved. Use this when you detect an error in your batch and want to start fresh.

Compare: checkpoint() also clears pending but is intended for marking progress. purge_pending() is for abandoning work.

Example

// Queue some changes
let builder = live.add_edge(builder, "A", "B", 10)

// Oops, wrong data! Purge and start over.
let builder = live.purge_pending(builder)
// builder has no pending changes, registry still has A and B

remove_edge

</>

pub fn remove_edge(
  builder: LiveBuilder(n, e),
  from src_label: n,
  to dst_label: n,
) -> LiveBuilder(n, e)

Removes an edge between two labeled nodes.

If either label doesn’t exist, no transition is queued. The removal is queued as a pending transition.

Example

let builder =
  builder
  |> live.remove_edge("A", "B")

Complexity

$O(\log N)$ per label lookup

remove_node

</>

pub fn remove_node(
  builder: LiveBuilder(n, e),
  label: n,
) -> LiveBuilder(n, e)

Removes a node by its label.

The node and all its connected edges are removed. The ID is NOT reused. The label is removed from the registry so a future add would get a new ID.

Warning: Removing a node invalidates any cached IDs for that label. After sync, get_id(builder, label) will return Error(Nil).

Example

let builder =
  builder
  |> live.remove_node("obsolete_node")

Complexity

$O(\log N)$ for dict lookup/removal

sync

</>

pub fn sync(
  builder: LiveBuilder(n, e),
  graph: model.Graph(n, e),
) -> #(LiveBuilder(n, e), model.Graph(n, e))

Synchronizes pending changes to the given graph.

Applies all pending transitions (in order) to the provided graph, then returns a new builder with an empty pending list and the updated graph.

Note: If there are no pending changes, this returns immediately with the same builder and graph (effectively O(1)).

Warning: The graph type (directed/undirected) is determined by the input graph. Make sure to provide a graph of the correct type on first sync.

Performance: For large pending queues (>1000 items), consider syncing more frequently to avoid memory pressure and reduce sync latency.

Example

// Initial build
let builder = live.new() |> live.add_edge("A", "B", 10)
let #(builder, graph) = live.sync(builder, yog.directed())

// Incremental update
let builder = builder |> live.add_edge("B", "C", 5)
let #(builder, graph) = live.sync(builder, graph)  // Only applies B->C!

Complexity

$O(\Delta E)$ where $\Delta E$ is the number of pending transitions

undirected

</>

pub fn undirected() -> LiveBuilder(n, e)

Creates a new live builder with an undirected graph type in mind.

This is a convenience function - the builder itself doesn’t store the graph type, but it helps document intent.

Example

let builder = live.undirected()