kglite 0.16.5

//! Change data capture — an opt-in, in-process stream of the changes a graph
//! publishes, addressed by stateless `(epoch, seq)` cursors.
//!
//! ## Where events come from
//!
//! From the **write-capture buffer the write-ahead log already uses**
//! ([`crate::graph::storage::recording`]), never from a second capture path.
//! `enable` installs that wrapper if the graph has none (a durable graph
//! already carries it), and every mutation that crosses the `GraphWrite` seam
//! buffers a [`RawOp`]. At a commit boundary the buffer is drained, resolved
//! against **final** state, and published as [`CdcEvent`]s.
//!
//! ## The no-phantom invariant
//!
//! **A change that was not committed must never appear in the stream.** This
//! is the property the design is arranged around, and it is why events are
//! derived at the drain rather than at the write:
//!
//! - A **failed statement** rolls its writes back and truncates the ops it
//!   buffered (`dir_graph::rollback`), so the drain never sees them.
//! - A **rolled-back transaction** drops its working copy; the fork's buffer
//!   dies with it, undrained. `RecordingGraph::Clone` starting a fork with an
//!   empty buffer is what makes that clean rather than merely likely.
//! - A **held reader** forces the writer to fork copy-on-write. The fork
//!   shares this log through its `Arc`, so the writer's commit publishes once,
//!   into the one log the reader's and writer's handles both see.
//!
//! The cost of that arrangement is that CDC has exactly the coverage the WAL
//! has: **a change is published where a durable graph would flush a frame**,
//! and a caller driving a bare `DirGraph` has to say where its commits are, by
//! calling [`drain_at_commit`] — the same obligation the durable paths already
//! carry (see `KnowledgeGraph::flush_wal` and `Session::log_working_commit`).
//! An unpublished commit is a *missing* event; there is no arrangement here
//! that can invent one.
//!
//! ## What is deliberately not here
//!
//! - **Persistence.** The log is `#[serde(skip)]` runtime state: a `.kgl` save
//!   writes none of it and a load starts a new epoch, so a cursor never
//!   silently addresses different data. See [`CdcLog`].
//! - **Disk storage mode.** Refused at `enable`; a disk graph's change
//!   boundary is the generation publish, not this buffer.

mod event;
mod log;
mod selector;
#[cfg(test)]
mod tests;

pub use event::{CdcChange, CdcEvent, CdcEventKind, EdgeState, NodeState};
pub use log::{CdcEnrichment, CdcHandoff, CdcLog, CdcStatus, DEFAULT_CAPACITY, MAX_CAPACITY};
pub use selector::{needs_before_images, parse_selectors, CdcSelector};

use crate::error::KgError;
use crate::graph::dir_graph::DirGraph;
use crate::graph::storage::mode::{live_storage_mode, StorageMode};
use crate::graph::storage::recording::RawOp;
use std::sync::{Arc, Mutex};

/// The shared handle `DirGraph` holds. `Clone` shares it — a copy-on-write
/// view, a transaction fork and the graph they came from all publish into and
/// read from one log — while `DirGraph::independent_copy` re-mints a fresh one
/// (new epoch, empty ring), exactly as it re-mints `graph_id`.
pub type CdcHandle = Arc<Mutex<CdcLog>>;

/// Lock a log handle, tolerating poisoning like every other lock in the
/// engine: a panicking publisher must not take the stream down with it.
fn lock(handle: &CdcHandle) -> std::sync::MutexGuard<'_, CdcLog> {
    handle
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// Start (or reconfigure) change data capture on `graph`.
///
/// Installs the write-capture wrapper if the graph has none — without claiming
/// write-ahead-log ownership, so a durable open is still possible afterwards
/// and the durable-only duplicate-id refusal is not imposed on a graph that
/// keeps no log ([`RecordingGraph::is_wal_owner`](crate::graph::storage::recording::RecordingGraph::is_wal_owner)).
///
/// **Re-enabling an enabled log reconfigures it in place** and keeps the epoch,
/// so live consumer cursors survive a capacity or enrichment change; a shrink
/// evicts from the front and shows up as `earliest` advancing, like any other
/// eviction.
///
/// `enable` is **declarative, not incremental**: every argument it does not
/// receive takes its default, so a re-enable that names only `capacity` also
/// resets enrichment to [`CdcEnrichment::Off`]. That is the same rule capacity
/// has always followed (a bare `enable()` on a running log resizes it to
/// `DEFAULT_CAPACITY`), and one rule for both knobs is what keeps the call
/// readable: what you pass is what the log ends up configured as.
///
/// # Cost
///
/// Capture is not free on the write path: every mutation buffers a `RawOp`,
/// and a wrapped backend gives up the checkpoint-free-mutation fast path, so
/// an enabled graph pays the same statement-checkpoint cost a durable graph
/// always pays. Graphs with capture *off* are untouched by any of this —
/// that separation is what the perf gate protects.
///
/// # Refusals
///
/// - **Disk storage mode.** A disk graph commits by publishing an immutable
///   generation, so the `GraphWrite` buffer this stream is derived from does
///   not describe its change boundary.
/// - **A capacity of 0, or one above [`MAX_CAPACITY`].**
// `KgError` deliberately carries structured context; boxing it here would give
// this one lifecycle call a different error type from every other engine entry
// point a binding maps.
#[allow(clippy::result_large_err)]
pub fn enable(
    graph: &mut DirGraph,
    capacity: Option<usize>,
    enrichment: CdcEnrichment,
) -> Result<CdcStatus, KgError> {
    if live_storage_mode(graph) == StorageMode::Disk {
        return Err(KgError::Argument(
            "change data capture is not supported for storage='disk'. A disk graph \
             commits by publishing an immutable generation, so its change boundary is \
             that publish rather than the per-mutation write capture this stream is \
             derived from — enabling here would report a stream that silently missed \
             writes. Use an in-memory or mapped graph for change data capture."
                .to_string(),
        ));
    }
    let capacity = match capacity {
        None => DEFAULT_CAPACITY,
        Some(0) => {
            return Err(KgError::Argument(
                "change-data-capture capacity must be at least 1 event; a capacity of 0 \
                 would evict every event before a consumer could read it."
                    .to_string(),
            ))
        }
        Some(requested) if requested > MAX_CAPACITY => {
            return Err(KgError::Argument(format!(
                "change-data-capture capacity {requested} exceeds the maximum of \
                 {MAX_CAPACITY} events. The log is held in memory, so its bound is a \
                 memory bound; keep a retention window a consumer can actually catch up \
                 across, not an unbounded one."
            )))
        }
        Some(requested) => requested,
    };

    let status = match &graph.cdc {
        Some(handle) => {
            let mut log = lock(handle);
            log.reconfigure(capacity, enrichment);
            log.status()
        }
        None => {
            graph.graph.wrap_for_capture();
            let log = CdcLog::new(capacity, enrichment);
            let status = log.status();
            graph.cdc = Some(Arc::new(Mutex::new(log)));
            status
        }
    };
    // The wrapper, not the log, decides what a write captures: the read that
    // builds a before-image happens at the write, long before anything looks
    // at the log. Setting it here — after the log is installed — keeps the two
    // in step through every re-enable.
    graph
        .graph
        .set_capture_before(enrichment == CdcEnrichment::Full);
    Ok(status)
}

/// Stop change data capture, discard the log, and hand back whether it was on.
///
/// **The capture wrapper comes off with it — unless a write-ahead log owns
/// it.** That is the whole rule, and both halves matter: capture costs a
/// buffered op per mutation and the checkpoint-free-mutation fast path, so a
/// disable that left the wrapper behind would leave a permanent tax on a graph
/// the caller believes is back to normal; and unwrapping a WAL-owned wrapper
/// would silently stop logging a durable graph, which is data loss. A durable
/// graph therefore keeps its wrapper here and simply loses its stream.
///
/// Buffered ops die with the wrapper, and that is not a lost publish: they
/// belong to a commit boundary that has not been reached, so they were never
/// publishable — and the log they would have gone to is being dropped.
pub fn disable(graph: &mut DirGraph) -> bool {
    // The epoch ends here, so this is where its end is worth recording: a
    // consumer that arrives later holding one of its cursors gets told where
    // it stopped rather than only that it is gone. Kept on the graph so a
    // subsequent save carries it into the file too.
    if let Some(status) = status(graph) {
        graph.cdc_handoff = Some(CdcHandoff {
            epoch: status.epoch,
            last_seq: status.current,
        });
    }
    let was_enabled = graph.cdc.take().is_some();
    // A WAL-owned wrapper survives `unwrap_capture_if_unowned`, so clearing
    // the flag is what stops a durable graph paying for before-images once
    // the consumer that asked for them is gone.
    graph.graph.set_capture_before(false);
    graph.graph.unwrap_capture_if_unowned();
    was_enabled
}

/// This graph's log addressing state, or `None` when capture is off.
pub fn status(graph: &DirGraph) -> Option<CdcStatus> {
    graph.cdc.as_ref().map(|handle| lock(handle).status())
}

/// Read events after the cursor position `from` (exclusive), oldest first,
/// keeping only those any of `selectors` matches.
///
/// `None` when capture is off. An empty `selectors` is no filter.
///
/// **Filtering happens before the clone-out, and `limit` after the
/// filtering.** Both matter: an event the caller did not ask for is never
/// copied out of the ring, and `limit` bounds the rows the caller *receives*
/// rather than the window they were drawn from — a limit applied first would
/// make a selective query return nothing while its matches sat one event past
/// the window, with no way for the caller to tell that from "caught up".
pub fn read(
    graph: &DirGraph,
    from: u64,
    limit: Option<usize>,
    selectors: &[CdcSelector],
) -> Option<Vec<CdcEvent>> {
    graph.cdc.as_ref().map(|handle| {
        lock(handle)
            .since(from, None)
            .into_iter()
            .filter(|event| selector::selected(selectors, event))
            .take(limit.unwrap_or(usize::MAX))
            .cloned()
            .collect()
    })
}

/// Publish one commit's drained capture buffer as CDC events.
///
/// A no-op when capture is off, and the **only** way events reach the log.
/// Call it with ops that have been drained (so no other consumer can publish
/// them again) and with `graph` in its post-commit state (so the after-state
/// resolution reads what the commit left behind) — the same two preconditions
/// [`resolve_ops`](crate::graph::storage::recording::resolve_ops) has.
pub fn publish_drained(graph: &DirGraph, raw: &[RawOp]) {
    if raw.is_empty() {
        return;
    }
    let Some(handle) = graph.cdc.as_ref() else {
        return;
    };
    let events = event::events_from_raw(raw, &graph.graph, &graph.interner, |idx| {
        graph.secondary_label_names(idx)
    });
    if events.is_empty() {
        return;
    }
    lock(handle).append(events);
}

/// Drain the capture buffer at a commit boundary, publishing what it holds,
/// and hand the raw ops back to the write-ahead log owner.
///
/// This is the drain primitive for every owner that has no fail-closed
/// requirement of its own; `Session::log_working_commit` keeps its own drain
/// because it must distinguish "nothing captured" from "the capture seam is
/// gone" and refuse the commit in the second case.
///
/// Returns an empty vector when the graph carries no capture layer at all.
pub fn drain_at_commit(graph: &mut DirGraph) -> Vec<RawOp> {
    let raw = graph
        .graph
        .recording_mut()
        .map(|recording| recording.take_ops())
        .unwrap_or_default();
    publish_drained(graph, &raw);
    raw
}