kweb_db_core/

lib.rs

//! Storage-only primitives for a provenance-backed knowledge web.
//!
//! [`Kmap`] owns one SQLite database and a sibling immutable-artifact store.
//! It records current knowledge nodes, append-only provenance history, ordered
//! connection identifier arrays, and permanent mutation-idempotency receipts.
//! The caller owns concurrency, graph policy, user/root mappings, HTTP, and all
//! multi-call orchestration.
//!
//! # Minimal example
//!
//! ```
//! use kweb_db_core::{IdempotencyId, Kmap, NewProvenance};
//!
//! let mut kweb = Kmap::open(":memory:")?;
//! let provenance_id = kweb.create_provenance(
//!     IdempotencyId::random(),
//!     NewProvenance {
//!         data: "Imported source material".into(),
//!         source: "example".into(),
//!         source_created_at: "2026-07-18T00:00:00Z".into(),
//!     },
//! )?;
//! assert_eq!(kweb.get_provenance(provenance_id)?.data, "Imported source material");
//! # Ok::<(), kweb_db_core::Error>(())
//! ```
//!
//! See the crate README and `Specification.md` in the published package for
//! validation rules, persistence details, and deliberate limitations.

#![deny(missing_docs)]

use std::{
    collections::HashSet,
    fmt,
    fs::{self, File, OpenOptions},
    io::{Read, Write},
    path::{Component, Path, PathBuf},
};

use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use chrono::{DateTime, Utc};
use rand::RngCore;
use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
use sha2::{Digest, Sha256};

const SCHEMA: &str = include_str!("../schema.sql");
const INLINE_PROVENANCE_LIMIT: usize = 256 * 1024;

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
/// A durable 20-byte knowledge-node identifier.
///
/// Its display, parsing, and Serde representation is exactly 40 lowercase
/// hexadecimal characters.
pub struct NodeId([u8; 20]);

impl NodeId {
    /// Generates a cryptographically random identifier.
    pub fn random() -> Self {
        let mut bytes = [0_u8; 20];
        rand::rng().fill_bytes(&mut bytes);
        Self(bytes)
    }

    /// Parses exactly 40 lowercase hexadecimal characters.
    pub fn from_hex(value: &str) -> Result<Self, Error> {
        if value.len() != 40
            || !value
                .bytes()
                .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        {
            return Err(Error::InvalidInput(
                "Identifiers must be 40 lowercase hexadecimal characters.".into(),
            ));
        }
        let decoded =
            hex::decode(value).map_err(|_| Error::InvalidInput("Invalid identifier.".into()))?;
        let bytes: [u8; 20] = decoded
            .try_into()
            .map_err(|_| Error::InvalidInput("Invalid identifier.".into()))?;
        Ok(Self(bytes))
    }

    /// Returns the 40-character lowercase hexadecimal representation.
    pub fn to_hex(self) -> String {
        hex::encode(self.0)
    }

    fn as_bytes(&self) -> &[u8] {
        &self.0
    }

    fn from_blob(value: Vec<u8>) -> rusqlite::Result<Self> {
        let length = value.len();
        let bytes = value.try_into().map_err(|_| {
            rusqlite::Error::FromSqlConversionFailure(
                length,
                rusqlite::types::Type::Blob,
                "Kmap identifier is not 20 bytes".into(),
            )
        })?;
        Ok(Self(bytes))
    }
}

impl fmt::Display for NodeId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&hex::encode(self.0))
    }
}

impl Serialize for NodeId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&self.to_string())
    }
}

impl<'de> Deserialize<'de> for NodeId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::from_hex(&value).map_err(de::Error::custom)
    }
}

/// The identifier of an immutable provenance record.
pub type ProvenanceId = NodeId;

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
/// A caller-generated 16-byte identifier for one logical mutation.
///
/// Reuse this value only when retrying the exact same operation and semantic
/// input. Reusing it for another request returns [`Error::Conflict`].
pub struct IdempotencyId([u8; 16]);

impl IdempotencyId {
    /// Generates a cryptographically random idempotency identifier.
    pub fn random() -> Self {
        let mut bytes = [0_u8; 16];
        rand::rng().fill_bytes(&mut bytes);
        Self(bytes)
    }

    /// Parses exactly 32 lowercase hexadecimal characters.
    pub fn from_hex(value: &str) -> Result<Self, Error> {
        if value.len() != 32
            || !value
                .bytes()
                .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        {
            return Err(Error::InvalidInput(
                "Idempotency identifiers must be 32 lowercase hexadecimal characters.".into(),
            ));
        }
        let decoded = hex::decode(value)
            .map_err(|_| Error::InvalidInput("Invalid idempotency identifier.".into()))?;
        let bytes: [u8; 16] = decoded
            .try_into()
            .map_err(|_| Error::InvalidInput("Invalid idempotency identifier.".into()))?;
        Ok(Self(bytes))
    }

    /// Returns the 32-character lowercase hexadecimal representation.
    pub fn to_hex(self) -> String {
        hex::encode(self.0)
    }

    fn as_bytes(&self) -> &[u8] {
        &self.0
    }
}

impl fmt::Display for IdempotencyId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&hex::encode(self.0))
    }
}

impl Serialize for IdempotencyId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&self.to_string())
    }
}

impl<'de> Deserialize<'de> for IdempotencyId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::from_hex(&value).map_err(de::Error::custom)
    }
}

#[derive(Clone, Debug, Serialize)]
/// The complete current state of a knowledge node.
pub struct Node {
    /// Durable node identifier.
    pub id: NodeId,
    /// Trimmed display name containing 4–50 Unicode characters.
    pub short_name: String,
    /// Trimmed summary containing at most 200 Unicode characters.
    pub short_description: String,
    /// Full caller-controlled description containing at most 1,000 words.
    pub long_description: String,
    /// Opaque caller attribution for the latest mutation.
    pub last_modified_by: String,
    /// Library-generated RFC 3339 timestamp of the latest mutation.
    pub last_modified_at: String,
    /// Owner node, self, or `None` for an explicitly unowned node.
    pub owner_node_id: Option<NodeId>,
    /// Ordered fixed-connection identifiers with no library-assigned policy.
    pub fixed_connections: Vec<NodeId>,
    /// Ordered recent-connection identifiers with no library-assigned policy.
    pub recent_connections: Vec<NodeId>,
}

#[derive(Clone, Copy, Debug)]
/// Ownership state supplied during a node create or update.
pub enum Owner {
    /// Store no owner.
    Unowned,
    /// Resolve ownership to the node being created or updated.
    SelfNode,
    /// Store another existing node as the owner.
    Node(NodeId),
}

#[derive(Clone, Debug)]
/// Complete input for [`Kmap::create_node`].
pub struct CreateNode {
    /// Caller-selected durable node identifier.
    pub id: NodeId,
    /// Existing provenance supporting the initial node state.
    pub provenance_id: ProvenanceId,
    /// Requested ownership state.
    pub owner: Owner,
    /// Trimmed display name containing 4–50 Unicode characters.
    pub short_name: String,
    /// Trimmed summary containing at most 200 Unicode characters.
    pub short_description: String,
    /// Full description containing at most 1,000 words.
    pub long_description: String,
    /// Opaque 1–200 character caller attribution.
    pub model_attribution: String,
    /// Complete ordered replacement array of fixed connections.
    pub fixed_connections: Vec<NodeId>,
    /// Complete ordered replacement array of recent connections.
    pub recent_connections: Vec<NodeId>,
}

#[derive(Clone, Debug)]
/// Complete replacement input for [`Kmap::update_node`].
pub struct UpdateNode {
    /// Existing provenance supporting the new node state.
    pub provenance_id: ProvenanceId,
    /// Requested ownership state.
    pub owner: Owner,
    /// Trimmed display name containing 4–50 Unicode characters.
    pub short_name: String,
    /// Trimmed summary containing at most 200 Unicode characters.
    pub short_description: String,
    /// Full description containing at most 1,000 words.
    pub long_description: String,
    /// Opaque 1–200 character caller attribution.
    pub model_attribution: String,
    /// Complete ordered replacement array of fixed connections.
    pub fixed_connections: Vec<NodeId>,
    /// Complete ordered replacement array of recent connections.
    pub recent_connections: Vec<NodeId>,
}

#[derive(Clone, Debug, Deserialize)]
/// Input for creating one immutable provenance record.
pub struct NewProvenance {
    /// Arbitrary UTF-8 source material.
    pub data: String,
    /// Trimmed source label containing 1–200 characters.
    pub source: String,
    /// Caller-supplied RFC 3339 creation time of the source material.
    pub source_created_at: String,
}

#[derive(Clone, Debug)]
/// One immutable byte artifact attached to new provenance.
pub struct NewProvenanceArtifact {
    /// Safe original basename to preserve in the stored filename.
    pub original_filename: String,
    /// Printable media type containing 1–200 characters.
    pub media_type: String,
    /// Caller-defined semantic role containing 1–100 characters.
    pub role: String,
    /// Complete artifact bytes.
    pub data: Vec<u8>,
}

#[derive(Clone, Debug)]
/// Artifact-storage input for [`Kmap::create_provenance_with_storage`].
pub struct ProvenanceStorage {
    /// Basename used if the main provenance text is stored externally.
    pub data_filename: String,
    /// Ordered immutable artifacts attached to the provenance.
    pub artifacts: Vec<NewProvenanceArtifact>,
}

#[derive(Clone, Debug, Serialize)]
/// Metadata for one stored provenance artifact.
pub struct ProvenanceArtifact {
    /// Two-component path relative to [`Kmap::artifact_path`].
    pub relative_path: String,
    /// Original caller-supplied basename.
    pub original_filename: String,
    /// Caller-supplied media type.
    pub media_type: String,
    /// Semantic role within the provenance record.
    pub role: String,
    /// Exact stored byte length.
    pub byte_length: u64,
    /// Lowercase hexadecimal SHA-256 digest of the stored bytes.
    pub sha256: String,
}

#[derive(Clone, Debug, Serialize)]
/// One immutable provenance record and its ordered artifact metadata.
pub struct Provenance {
    /// Durable provenance identifier.
    pub id: ProvenanceId,
    /// Complete UTF-8 source material, loaded transparently when external.
    pub data: String,
    /// Trimmed source label.
    pub source: String,
    /// Caller-supplied RFC 3339 source creation time.
    pub source_created_at: String,
    /// Ordered metadata for attached artifacts and external source data.
    pub artifacts: Vec<ProvenanceArtifact>,
}

#[derive(Clone, Debug, Serialize)]
/// Additively extensible statistics over current knowledge-node text.
///
/// Fields are private so new metrics can be added without breaking Rust callers.
/// Serialized consumers must ignore unknown fields.
pub struct Stats {
    node_count: u64,
    full_node_characters: u64,
    full_node_words: u64,
    estimated_full_node_tokens: u64,
    long_description_characters: u64,
    long_description_words: u64,
    estimated_long_description_tokens: u64,
}

impl Stats {
    /// Returns the number of current knowledge nodes.
    pub fn node_count(&self) -> u64 {
        self.node_count
    }

    /// Returns Unicode-character count across all three node text fields.
    pub fn full_node_characters(&self) -> u64 {
        self.full_node_characters
    }

    /// Returns whitespace-delimited word count across all node text fields.
    pub fn full_node_words(&self) -> u64 {
        self.full_node_words
    }

    /// Returns the full-text character count divided by four, rounded up.
    pub fn estimated_full_node_tokens(&self) -> u64 {
        self.estimated_full_node_tokens
    }

    /// Returns Unicode-character count for long descriptions only.
    pub fn long_description_characters(&self) -> u64 {
        self.long_description_characters
    }

    /// Returns whitespace-delimited word count for long descriptions only.
    pub fn long_description_words(&self) -> u64 {
        self.long_description_words
    }

    /// Returns the long-description character count divided by four, rounded up.
    pub fn estimated_long_description_tokens(&self) -> u64 {
        self.estimated_long_description_tokens
    }
}

#[derive(Debug)]
/// Error returned by Kweb DB Core storage operations.
pub enum Error {
    /// Caller input failed validation.
    InvalidInput(String),
    /// A required stored object was not found.
    NotFound(&'static str),
    /// Stored state or idempotency reuse conflicts with the request.
    Conflict(String),
    /// SQLite returned an unexpected error.
    Database(rusqlite::Error),
    /// Filesystem artifact storage returned an unexpected error.
    Io(std::io::Error),
}

impl fmt::Display for Error {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidInput(message) | Self::Conflict(message) => formatter.write_str(message),
            Self::NotFound(kind) => write!(formatter, "{kind} not found."),
            Self::Database(error) => write!(formatter, "Kmap database error: {error}"),
            Self::Io(error) => write!(formatter, "Kmap artifact storage error: {error}"),
        }
    }
}

impl std::error::Error for Error {}

impl From<rusqlite::Error> for Error {
    fn from(error: rusqlite::Error) -> Self {
        Self::Database(error)
    }
}

impl From<std::io::Error> for Error {
    fn from(error: std::io::Error) -> Self {
        Self::Io(error)
    }
}

#[derive(Clone, Debug)]
struct ArtifactStore {
    root: PathBuf,
}

#[derive(Clone, Debug)]
struct StoredArtifact {
    relative_path: String,
    original_filename: String,
    media_type: String,
    byte_length: u64,
    sha256: [u8; 32],
}

impl ArtifactStore {
    fn new(root: impl AsRef<Path>) -> Self {
        Self {
            root: root.as_ref().to_owned(),
        }
    }

    fn root(&self) -> &Path {
        &self.root
    }

    fn store(
        &self,
        suffix: &str,
        original_filename: &str,
        media_type: &str,
        data: &[u8],
    ) -> Result<StoredArtifact, Error> {
        let original_filename = validate_artifact_filename(original_filename)?;
        let media_type = validate_media_type(media_type)?;
        let stored_filename = suffixed_filename(&original_filename, suffix)?;
        let shard = suffix
            .get(..2)
            .ok_or_else(|| Error::InvalidInput("Artifact suffix is too short.".into()))?;
        let relative_path = format!("{shard}/{stored_filename}");
        let directory = self.root.join(shard);
        create_private_directory(&self.root)?;
        create_private_directory(&directory)?;
        let final_path = directory.join(&stored_filename);
        let sha256: [u8; 32] = Sha256::digest(data).into();
        if final_path.exists() {
            verify_artifact(&final_path, data.len() as u64, sha256)?;
        } else {
            let temporary_suffix = URL_SAFE_NO_PAD.encode(&IdempotencyId::random().0[..9]);
            let temporary = directory.join(format!(".{stored_filename}.{temporary_suffix}.tmp"));
            let mut file = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&temporary)?;
            set_file_private(&file)?;
            let write_result = (|| -> Result<(), Error> {
                file.write_all(data)?;
                file.sync_all()?;
                fs::rename(&temporary, &final_path)?;
                sync_directory(&directory)?;
                Ok(())
            })();
            if write_result.is_err() {
                let _ = fs::remove_file(&temporary);
            }
            write_result?;
        }
        Ok(StoredArtifact {
            relative_path,
            original_filename,
            media_type,
            byte_length: data.len() as u64,
            sha256,
        })
    }

    fn read(&self, relative_path: &str) -> Result<Vec<u8>, Error> {
        let path = resolve_artifact_path(&self.root, relative_path)?;
        fs::read(path).map_err(Error::from)
    }
}

/// A serialized handle to one Kweb DB Core SQLite database and artifact directory.
///
/// The caller must serialize access and must not expect multi-call atomicity.
pub struct Kmap {
    db: Connection,
    artifacts: ArtifactStore,
}

impl Kmap {
    /// Opens or creates a database and derives its sibling artifact directory.
    ///
    /// `kweb.sqlite3` derives `kweb-provenance-artifacts`. This enables foreign
    /// keys, WAL journaling, a five-second busy timeout, and initializes the
    /// current schema when opening a new database.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, Error> {
        let path = path.as_ref();
        let artifact_path = default_artifact_path(path);
        Self::open_with_artifacts(path, artifact_path)
    }

    /// Opens or creates a database with an explicit artifact-directory path.
    pub fn open_with_artifacts(
        path: impl AsRef<Path>,
        artifact_path: impl AsRef<Path>,
    ) -> Result<Self, Error> {
        let db = Connection::open(path)?;
        db.execute_batch(
            "PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;",
        )?;
        db.execute_batch(SCHEMA)?;
        Ok(Self {
            db,
            artifacts: ArtifactStore::new(artifact_path),
        })
    }

    /// Returns the configured artifact-directory path.
    pub fn artifact_path(&self) -> &Path {
        self.artifacts.root()
    }

    /// Creates immutable provenance without explicit attached artifacts.
    ///
    /// Main data larger than 256 KiB is stored externally. An exact replay of
    /// `idempotency_id` and `input` returns the original identifier without a
    /// second write.
    pub fn create_provenance(
        &mut self,
        idempotency_id: IdempotencyId,
        input: NewProvenance,
    ) -> Result<ProvenanceId, Error> {
        self.create_provenance_internal(idempotency_id, input, None, Vec::new())
    }

    /// Creates immutable provenance with ordered attached artifacts.
    ///
    /// `storage.data_filename` names the main data if it crosses the 256 KiB
    /// external-storage threshold. An exact replay returns the original ID.
    pub fn create_provenance_with_storage(
        &mut self,
        idempotency_id: IdempotencyId,
        input: NewProvenance,
        storage: ProvenanceStorage,
    ) -> Result<ProvenanceId, Error> {
        self.create_provenance_internal(
            idempotency_id,
            input,
            Some(storage.data_filename),
            storage.artifacts,
        )
    }

    fn create_provenance_internal(
        &mut self,
        idempotency_id: IdempotencyId,
        input: NewProvenance,
        data_filename: Option<String>,
        artifacts: Vec<NewProvenanceArtifact>,
    ) -> Result<ProvenanceId, Error> {
        let source = input.source.trim().to_owned();
        if source.is_empty() || source.chars().count() > 200 {
            return Err(Error::InvalidInput(
                "Source must contain 1 to 200 characters.".into(),
            ));
        }
        DateTime::parse_from_rfc3339(&input.source_created_at).map_err(|_| {
            Error::InvalidInput("source_created_at must be an RFC 3339 timestamp.".into())
        })?;
        let data_filename = data_filename
            .map(|value| validate_artifact_filename(&value))
            .transpose()?;
        let artifacts = validate_new_artifacts(artifacts)?;
        let request_hash = provenance_request_hash(
            &input.data,
            &source,
            &input.source_created_at,
            data_filename.as_deref(),
            &artifacts,
        );
        if let Some(id) = replay_receipt_on_connection(
            &self.db,
            idempotency_id,
            "create_provenance",
            request_hash,
        )? {
            return Ok(id);
        }
        let externalize_data = input.data.len() > INLINE_PROVENANCE_LIMIT;
        let mut stored = Vec::with_capacity(artifacts.len() + usize::from(externalize_data));
        for (index, artifact) in artifacts.iter().enumerate() {
            stored.push((
                artifact.role.clone(),
                self.artifacts.store(
                    &artifact_suffix(idempotency_id, index as u64),
                    &artifact.original_filename,
                    &artifact.media_type,
                    &artifact.data,
                )?,
            ));
        }
        let data_artifact = if externalize_data {
            let filename =
                data_filename.unwrap_or_else(|| default_data_filename(&source, &input.data));
            Some(self.artifacts.store(
                &artifact_suffix(idempotency_id, artifacts.len() as u64),
                &filename,
                if looks_like_json(&input.data) {
                    "application/json"
                } else {
                    "text/plain; charset=utf-8"
                },
                input.data.as_bytes(),
            )?)
        } else {
            None
        };
        let tx = self
            .db
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        if let Some(id) = replay_receipt(&tx, idempotency_id, "create_provenance", request_hash)? {
            tx.commit()?;
            return Ok(id);
        }
        let id = NodeId::random();
        for (_, artifact) in &stored {
            insert_artifact(&tx, artifact)?;
        }
        if let Some(artifact) = &data_artifact {
            insert_artifact(&tx, artifact)?;
        }
        tx.execute(
            "INSERT INTO data_provenance_nodes(id,data,data_artifact_path,source,source_created_at) VALUES(?1,?2,?3,?4,?5)",
            params![
                id.as_bytes(),
                (!externalize_data).then_some(input.data),
                data_artifact.as_ref().map(|artifact| artifact.relative_path.as_str()),
                source,
                input.source_created_at
            ],
        )?;
        for (position, (role, artifact)) in stored.iter().enumerate() {
            link_artifact(&tx, id, artifact, role, position)?;
        }
        if let Some(artifact) = &data_artifact {
            link_artifact(&tx, id, artifact, "source-data", stored.len())?;
        }
        record_receipt(&tx, idempotency_id, "create_provenance", request_hash, id)?;
        tx.commit()?;
        Ok(id)
    }

    /// Creates a knowledge node, its first history entry, and both connections.
    ///
    /// All referenced provenance, owner, and connection nodes must exist. An
    /// exact idempotent replay adds no history and returns the current node.
    pub fn create_node(
        &mut self,
        idempotency_id: IdempotencyId,
        input: CreateNode,
    ) -> Result<Node, Error> {
        let (short_name, short_description, model_attribution) = validate_node_input(
            &input.short_name,
            &input.short_description,
            &input.long_description,
            &input.model_attribution,
        )?;
        validate_connection_lists(&input.fixed_connections, &input.recent_connections)?;
        let owner = resolve_owner(input.owner, input.id);
        let request_hash = node_request_hash(
            input.id,
            input.provenance_id,
            owner,
            &short_name,
            &short_description,
            &input.long_description,
            &model_attribution,
            &input.fixed_connections,
            &input.recent_connections,
        );
        let tx = self
            .db
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        if let Some(id) = replay_receipt(&tx, idempotency_id, "create_node", request_hash)? {
            tx.commit()?;
            return self.get_node(id);
        }
        require_provenance(&tx, input.provenance_id)?;
        require_owner(&tx, owner, input.id)?;
        require_connections(&tx, &input.fixed_connections, input.id)?;
        require_connections(&tx, &input.recent_connections, input.id)?;
        let modified_at = Utc::now().to_rfc3339();
        let history_id = NodeId::random();
        tx.execute(
            "INSERT INTO knowledge_nodes(id,short_name,short_description,long_description,history_head_id,owner_node_id,last_modified_by,last_modified_at) VALUES(?1,?2,?3,?4,NULL,?5,?6,?7)",
            params![input.id.as_bytes(), short_name, short_description, input.long_description, owner.map(|id| id.0.to_vec()), model_attribution, modified_at],
        ).map_err(map_constraint("Knowledge node already exists."))?;
        tx.execute(
            "INSERT INTO data_history_nodes(id,knowledge_node_id,previous_history_id,provenance_id) VALUES(?1,?2,NULL,?3)",
            params![history_id.as_bytes(), input.id.as_bytes(), input.provenance_id.as_bytes()],
        )?;
        tx.execute(
            "UPDATE knowledge_nodes SET history_head_id=?1 WHERE id=?2",
            params![history_id.as_bytes(), input.id.as_bytes()],
        )?;
        replace_connections(
            &tx,
            input.id,
            &input.fixed_connections,
            &input.recent_connections,
        )?;
        record_receipt(&tx, idempotency_id, "create_node", request_hash, input.id)?;
        tx.commit()?;
        self.get_node(input.id)
    }

    /// Replaces a knowledge node and appends one provenance-history entry.
    ///
    /// Both connection arrays are complete replacements. An exact idempotent
    /// replay appends no history and returns the current node.
    pub fn update_node(
        &mut self,
        idempotency_id: IdempotencyId,
        id: NodeId,
        input: UpdateNode,
    ) -> Result<Node, Error> {
        let (short_name, short_description, model_attribution) = validate_node_input(
            &input.short_name,
            &input.short_description,
            &input.long_description,
            &input.model_attribution,
        )?;
        validate_connection_lists(&input.fixed_connections, &input.recent_connections)?;
        let owner = resolve_owner(input.owner, id);
        let request_hash = node_request_hash(
            id,
            input.provenance_id,
            owner,
            &short_name,
            &short_description,
            &input.long_description,
            &model_attribution,
            &input.fixed_connections,
            &input.recent_connections,
        );
        let tx = self
            .db
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        if let Some(result_id) = replay_receipt(&tx, idempotency_id, "update_node", request_hash)? {
            tx.commit()?;
            return self.get_node(result_id);
        }
        require_provenance(&tx, input.provenance_id)?;
        let previous = tx
            .query_row(
                "SELECT history_head_id FROM knowledge_nodes WHERE id=?1",
                [id.as_bytes()],
                |row| row.get::<_, Option<Vec<u8>>>(0),
            )
            .optional()?
            .ok_or(Error::NotFound("Knowledge node"))?;
        require_owner(&tx, owner, id)?;
        require_connections(&tx, &input.fixed_connections, id)?;
        require_connections(&tx, &input.recent_connections, id)?;
        let modified_at = Utc::now().to_rfc3339();
        let history_id = NodeId::random();
        tx.execute(
            "INSERT INTO data_history_nodes(id,knowledge_node_id,previous_history_id,provenance_id) VALUES(?1,?2,?3,?4)",
            params![history_id.as_bytes(), id.as_bytes(), previous, input.provenance_id.as_bytes()],
        )?;
        tx.execute(
            "UPDATE knowledge_nodes SET short_name=?1,short_description=?2,long_description=?3,history_head_id=?4,owner_node_id=?5,last_modified_by=?6,last_modified_at=?7 WHERE id=?8",
            params![short_name, short_description, input.long_description, history_id.as_bytes(), owner.map(|id| id.0.to_vec()), model_attribution, modified_at, id.as_bytes()],
        )?;
        replace_connections(&tx, id, &input.fixed_connections, &input.recent_connections)?;
        record_receipt(&tx, idempotency_id, "update_node", request_hash, id)?;
        tx.commit()?;
        self.get_node(id)
    }

    /// Returns a complete knowledge node by durable identifier.
    pub fn get_node(&self, id: NodeId) -> Result<Node, Error> {
        let row = self
            .db
            .query_row(
                "SELECT short_name,short_description,long_description,last_modified_by,last_modified_at,owner_node_id FROM knowledge_nodes WHERE id=?1",
                [id.as_bytes()],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, Option<Vec<u8>>>(5)?,
                    ))
                },
            )
            .optional()?
            .ok_or(Error::NotFound("Knowledge node"))?;
        Ok(Node {
            id,
            short_name: row.0,
            short_description: row.1,
            long_description: row.2,
            last_modified_by: row.3,
            last_modified_at: row.4,
            owner_node_id: row.5.map(NodeId::from_blob).transpose()?,
            fixed_connections: read_connections(&self.db, "fixed_connections", id)?,
            recent_connections: read_connections(&self.db, "recent_connections", id)?,
        })
    }

    /// Returns immutable provenance and ordered artifact metadata.
    ///
    /// Externally stored main UTF-8 data is read transparently. Explicit
    /// attached artifact bytes are not loaded into the result.
    pub fn get_provenance(&self, id: ProvenanceId) -> Result<Provenance, Error> {
        let (data, data_artifact_path, source, source_created_at) = self
            .db
            .query_row(
                "SELECT data,data_artifact_path,source,source_created_at FROM data_provenance_nodes WHERE id=?1",
                [id.as_bytes()],
                |row| {
                    Ok((
                        row.get::<_, Option<String>>(0)?,
                        row.get::<_, Option<String>>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                    ))
                },
            )
            .optional()?
            .ok_or(Error::NotFound("Provenance"))?;
        let data = match (data, data_artifact_path) {
            (Some(data), None) => data,
            (None, Some(path)) => String::from_utf8(self.artifacts.read(&path)?).map_err(|_| {
                Error::Conflict("Provenance source artifact is not valid UTF-8.".into())
            })?,
            _ => {
                return Err(Error::Conflict(
                    "Provenance has invalid inline/artifact storage state.".into(),
                ));
            }
        };
        Ok(Provenance {
            id,
            data,
            source,
            source_created_at,
            artifacts: read_provenance_artifacts(&self.db, id)?,
        })
    }

    /// Returns the node's complete provenance history, newest first.
    ///
    /// The traversal detects a corrupt cycle and returns [`Error::Conflict`].
    pub fn get_node_history(&self, id: NodeId) -> Result<Vec<ProvenanceId>, Error> {
        let mut current = self
            .db
            .query_row(
                "SELECT history_head_id FROM knowledge_nodes WHERE id=?1",
                [id.as_bytes()],
                |row| row.get::<_, Option<Vec<u8>>>(0),
            )
            .optional()?
            .ok_or(Error::NotFound("Knowledge node"))?;
        let mut seen = HashSet::new();
        let mut provenance_ids = Vec::new();
        while let Some(history_blob) = current {
            let history_id = NodeId::from_blob(history_blob)?;
            if !seen.insert(history_id) {
                return Err(Error::Conflict("History chain contains a cycle.".into()));
            }
            let (previous, provenance) = self.db.query_row(
                "SELECT previous_history_id,provenance_id FROM data_history_nodes WHERE id=?1 AND knowledge_node_id=?2",
                params![history_id.as_bytes(), id.as_bytes()],
                |row| Ok((row.get::<_, Option<Vec<u8>>>(0)?, row.get::<_, Vec<u8>>(1)?)),
            )?;
            provenance_ids.push(NodeId::from_blob(provenance)?);
            current = previous;
        }
        Ok(provenance_ids)
    }

    /// Calculates extensible statistics over current node text.
    pub fn stats(&self) -> Result<Stats, Error> {
        let mut statement = self
            .db
            .prepare("SELECT short_name,short_description,long_description FROM knowledge_nodes")?;
        let rows = statement.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
            ))
        })?;
        let mut stats = Stats {
            node_count: 0,
            full_node_characters: 0,
            full_node_words: 0,
            estimated_full_node_tokens: 0,
            long_description_characters: 0,
            long_description_words: 0,
            estimated_long_description_tokens: 0,
        };
        for row in rows {
            let (name, short, long) = row?;
            stats.node_count = stats.node_count.saturating_add(1);
            stats.full_node_characters = stats
                .full_node_characters
                .saturating_add(count_characters(&name))
                .saturating_add(count_characters(&short))
                .saturating_add(count_characters(&long))
                .saturating_add(2);
            stats.full_node_words = stats
                .full_node_words
                .saturating_add(count_words(&name))
                .saturating_add(count_words(&short))
                .saturating_add(count_words(&long));
            stats.long_description_characters = stats
                .long_description_characters
                .saturating_add(count_characters(&long));
            stats.long_description_words = stats
                .long_description_words
                .saturating_add(count_words(&long));
        }
        stats.estimated_full_node_tokens = stats.full_node_characters.div_ceil(4);
        stats.estimated_long_description_tokens = stats.long_description_characters.div_ceil(4);
        Ok(stats)
    }
}

fn default_artifact_path(database: &Path) -> PathBuf {
    if database == Path::new(":memory:") {
        return std::env::temp_dir().join(format!(
            "kweb-provenance-artifacts-{}",
            URL_SAFE_NO_PAD.encode(&IdempotencyId::random().0[..9])
        ));
    }
    let parent = database.parent().unwrap_or_else(|| Path::new("."));
    let stem = database
        .file_stem()
        .and_then(|value| value.to_str())
        .filter(|value| !value.is_empty())
        .unwrap_or("kweb");
    parent.join(format!("{stem}-provenance-artifacts"))
}

fn validate_artifact_filename(value: &str) -> Result<String, Error> {
    if value.is_empty() || value.len() > 220 || value.contains(['/', '\\']) {
        return Err(Error::InvalidInput(
            "Artifact filenames must be non-empty basenames no longer than 220 bytes.".into(),
        ));
    }
    let mut components = Path::new(value).components();
    if !matches!(components.next(), Some(Component::Normal(_))) || components.next().is_some() {
        return Err(Error::InvalidInput(
            "Artifact filenames must contain exactly one safe path component.".into(),
        ));
    }
    Ok(value.to_owned())
}

fn validate_media_type(value: &str) -> Result<String, Error> {
    let value = value.trim();
    if value.is_empty() || value.len() > 200 || value.chars().any(char::is_control) {
        return Err(Error::InvalidInput(
            "Artifact media types must contain 1 to 200 printable characters.".into(),
        ));
    }
    Ok(value.to_owned())
}

fn validate_artifact_role(value: &str) -> Result<String, Error> {
    let value = value.trim();
    if value.is_empty() || value.chars().count() > 100 {
        return Err(Error::InvalidInput(
            "Artifact roles must contain 1 to 100 characters.".into(),
        ));
    }
    Ok(value.to_owned())
}

fn validate_new_artifacts(
    artifacts: Vec<NewProvenanceArtifact>,
) -> Result<Vec<NewProvenanceArtifact>, Error> {
    artifacts
        .into_iter()
        .map(|artifact| {
            Ok(NewProvenanceArtifact {
                original_filename: validate_artifact_filename(&artifact.original_filename)?,
                media_type: validate_media_type(&artifact.media_type)?,
                role: validate_artifact_role(&artifact.role)?,
                data: artifact.data,
            })
        })
        .collect()
}

fn artifact_suffix(idempotency_id: IdempotencyId, position: u64) -> String {
    let mut hash = Sha256::new();
    hash.update(b"kweb-provenance-artifact-name-v1\0");
    hash.update(idempotency_id.as_bytes());
    hash.update(position.to_be_bytes());
    let digest = hash.finalize();
    URL_SAFE_NO_PAD.encode(&digest[..9])
}

fn suffixed_filename(original: &str, suffix: &str) -> Result<String, Error> {
    let path = Path::new(original);
    let extension = path.extension().and_then(|value| value.to_str());
    let stem = path
        .file_stem()
        .and_then(|value| value.to_str())
        .ok_or_else(|| Error::InvalidInput("Artifact filename is not valid UTF-8.".into()))?;
    let filename = match extension {
        Some(extension) if !extension.is_empty() => format!("{stem}.{suffix}.{extension}"),
        _ => format!("{original}.{suffix}"),
    };
    if filename.len() > 255 {
        return Err(Error::InvalidInput(
            "Artifact filename is too long after adding its random suffix.".into(),
        ));
    }
    Ok(filename)
}

fn default_data_filename(source: &str, data: &str) -> String {
    let safe_source = source
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_') {
                character
            } else {
                '-'
            }
        })
        .collect::<String>();
    let extension = if looks_like_json(data) { "json" } else { "txt" };
    format!("{safe_source}-provenance.{extension}")
}

fn looks_like_json(data: &str) -> bool {
    let data = data.trim();
    (data.starts_with('{') && data.ends_with('}')) || (data.starts_with('[') && data.ends_with(']'))
}

fn resolve_artifact_path(root: &Path, relative_path: &str) -> Result<PathBuf, Error> {
    let mut components = Path::new(relative_path).components();
    let shard = match components.next() {
        Some(Component::Normal(value)) => value,
        _ => {
            return Err(Error::Conflict(
                "Stored artifact path has an invalid shard.".into(),
            ));
        }
    };
    let filename = match components.next() {
        Some(Component::Normal(value)) if components.next().is_none() => value,
        _ => {
            return Err(Error::Conflict(
                "Stored artifact path has an invalid filename.".into(),
            ));
        }
    };
    let shard = shard
        .to_str()
        .filter(|value| value.len() == 2)
        .ok_or_else(|| Error::Conflict("Stored artifact shard is invalid.".into()))?;
    let filename = filename
        .to_str()
        .ok_or_else(|| Error::Conflict("Stored artifact filename is invalid.".into()))?;
    Ok(root.join(shard).join(filename))
}

fn verify_artifact(path: &Path, byte_length: u64, sha256: [u8; 32]) -> Result<(), Error> {
    let metadata = fs::metadata(path)?;
    if !metadata.is_file() || metadata.len() != byte_length {
        return Err(Error::Conflict(format!(
            "Existing artifact {} does not match its idempotent write.",
            path.display()
        )));
    }
    let mut file = File::open(path)?;
    let mut digest = Sha256::new();
    let mut buffer = [0_u8; 64 * 1024];
    loop {
        let read = file.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        digest.update(&buffer[..read]);
    }
    if <[u8; 32]>::from(digest.finalize()) != sha256 {
        return Err(Error::Conflict(format!(
            "Existing artifact {} failed its idempotent hash check.",
            path.display()
        )));
    }
    Ok(())
}

fn create_private_directory(path: &Path) -> Result<(), Error> {
    fs::create_dir_all(path)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
    }
    Ok(())
}

fn set_file_private(file: &File) -> Result<(), Error> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        file.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    Ok(())
}

fn sync_directory(path: &Path) -> Result<(), Error> {
    File::open(path)?.sync_all()?;
    Ok(())
}

fn insert_artifact(tx: &Transaction<'_>, artifact: &StoredArtifact) -> Result<(), Error> {
    let byte_length = i64::try_from(artifact.byte_length)
        .map_err(|_| Error::InvalidInput("Artifact is too large for SQLite metadata.".into()))?;
    tx.execute(
        "INSERT INTO provenance_artifacts(relative_path,original_filename,media_type,byte_length,sha256,created_at) VALUES(?1,?2,?3,?4,?5,?6)",
        params![
            artifact.relative_path,
            artifact.original_filename,
            artifact.media_type,
            byte_length,
            artifact.sha256.as_slice(),
            Utc::now().to_rfc3339()
        ],
    )?;
    Ok(())
}

fn link_artifact(
    tx: &Transaction<'_>,
    provenance_id: ProvenanceId,
    artifact: &StoredArtifact,
    role: &str,
    position: usize,
) -> Result<(), Error> {
    tx.execute(
        "INSERT INTO provenance_artifact_links(provenance_id,artifact_path,role,position) VALUES(?1,?2,?3,?4)",
        params![
            provenance_id.as_bytes(),
            artifact.relative_path,
            role,
            position as i64
        ],
    )?;
    Ok(())
}

fn read_provenance_artifacts(
    db: &Connection,
    provenance_id: ProvenanceId,
) -> Result<Vec<ProvenanceArtifact>, Error> {
    let mut statement = db.prepare(
        "SELECT a.relative_path,a.original_filename,a.media_type,l.role,a.byte_length,a.sha256
         FROM provenance_artifact_links l
         JOIN provenance_artifacts a ON a.relative_path=l.artifact_path
         WHERE l.provenance_id=?1 ORDER BY l.position",
    )?;
    Ok(statement
        .query_map([provenance_id.as_bytes()], |row| {
            let byte_length = row.get::<_, i64>(4)?;
            Ok(ProvenanceArtifact {
                relative_path: row.get(0)?,
                original_filename: row.get(1)?,
                media_type: row.get(2)?,
                role: row.get(3)?,
                byte_length: u64::try_from(byte_length)
                    .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(4, byte_length))?,
                sha256: hex::encode(row.get::<_, Vec<u8>>(5)?),
            })
        })?
        .collect::<Result<Vec<_>, _>>()?)
}

fn replay_receipt_on_connection(
    db: &Connection,
    id: IdempotencyId,
    operation_kind: &'static str,
    request_hash: [u8; 32],
) -> Result<Option<NodeId>, Error> {
    let receipt = db
        .query_row(
            "SELECT operation_kind,request_hash,result_id FROM idempotency_receipts WHERE id=?1",
            [id.as_bytes()],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, Vec<u8>>(1)?,
                    row.get::<_, Vec<u8>>(2)?,
                ))
            },
        )
        .optional()?;
    let Some((stored_kind, stored_hash, result_id)) = receipt else {
        return Ok(None);
    };
    if stored_kind != operation_kind || stored_hash.as_slice() != request_hash {
        return Err(Error::Conflict(
            "Idempotency identifier was already used for a different mutation.".into(),
        ));
    }
    Ok(Some(NodeId::from_blob(result_id)?))
}

fn validate_node_input(
    name: &str,
    short: &str,
    long: &str,
    attribution: &str,
) -> Result<(String, String, String), Error> {
    let name = name.trim().to_owned();
    let short = short.trim().to_owned();
    let attribution = attribution.trim().to_owned();
    if !(4..=50).contains(&name.chars().count()) {
        return Err(Error::InvalidInput(
            "Short name must contain 4 to 50 characters.".into(),
        ));
    }
    if short.chars().count() > 200 {
        return Err(Error::InvalidInput(
            "Short description must not exceed 200 characters.".into(),
        ));
    }
    if long.split_whitespace().count() > 1000 {
        return Err(Error::InvalidInput(
            "Long description must not exceed 1000 words.".into(),
        ));
    }
    if attribution.is_empty() || attribution.chars().count() > 200 {
        return Err(Error::InvalidInput(
            "model_attribution must contain 1 to 200 characters.".into(),
        ));
    }
    Ok((name, short, attribution))
}

fn validate_connection_lists(fixed: &[NodeId], recent: &[NodeId]) -> Result<(), Error> {
    for (name, ids) in [("fixed_connections", fixed), ("recent_connections", recent)] {
        let unique = ids.iter().collect::<HashSet<_>>();
        if unique.len() != ids.len() {
            return Err(Error::InvalidInput(format!(
                "{name} must not contain duplicate identifiers."
            )));
        }
    }
    Ok(())
}

fn resolve_owner(owner: Owner, node_id: NodeId) -> Option<NodeId> {
    match owner {
        Owner::Unowned => None,
        Owner::SelfNode => Some(node_id),
        Owner::Node(id) => Some(id),
    }
}

fn replay_receipt(
    tx: &Transaction<'_>,
    id: IdempotencyId,
    operation_kind: &'static str,
    request_hash: [u8; 32],
) -> Result<Option<NodeId>, Error> {
    let receipt = tx
        .query_row(
            "SELECT operation_kind,request_hash,result_id FROM idempotency_receipts WHERE id=?1",
            [id.as_bytes()],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, Vec<u8>>(1)?,
                    row.get::<_, Vec<u8>>(2)?,
                ))
            },
        )
        .optional()?;
    let Some((stored_kind, stored_hash, result_id)) = receipt else {
        return Ok(None);
    };
    if stored_kind != operation_kind || stored_hash.as_slice() != request_hash {
        return Err(Error::Conflict(
            "Idempotency identifier was already used for a different mutation.".into(),
        ));
    }
    Ok(Some(NodeId::from_blob(result_id)?))
}

fn record_receipt(
    tx: &Transaction<'_>,
    id: IdempotencyId,
    operation_kind: &'static str,
    request_hash: [u8; 32],
    result_id: NodeId,
) -> Result<(), Error> {
    tx.execute(
        "INSERT INTO idempotency_receipts(id,operation_kind,request_hash,result_id,committed_at) VALUES(?1,?2,?3,?4,?5)",
        params![
            id.as_bytes(),
            operation_kind,
            request_hash.as_slice(),
            result_id.as_bytes(),
            Utc::now().to_rfc3339(),
        ],
    )?;
    Ok(())
}

struct RequestHasher(Sha256);

impl RequestHasher {
    fn new() -> Self {
        Self(Sha256::new())
    }

    fn field(&mut self, value: &[u8]) {
        self.0.update((value.len() as u64).to_be_bytes());
        self.0.update(value);
    }

    fn id(&mut self, value: NodeId) {
        self.field(value.as_bytes());
    }

    fn optional_id(&mut self, value: Option<NodeId>) {
        match value {
            Some(value) => {
                self.field(&[1]);
                self.id(value);
            }
            None => self.field(&[0]),
        }
    }

    fn ids(&mut self, values: &[NodeId]) {
        self.field(&(values.len() as u64).to_be_bytes());
        for value in values {
            self.id(*value);
        }
    }

    fn finish(self) -> [u8; 32] {
        self.0.finalize().into()
    }
}

fn provenance_request_hash(
    data: &str,
    source: &str,
    source_created_at: &str,
    data_filename: Option<&str>,
    artifacts: &[NewProvenanceArtifact],
) -> [u8; 32] {
    let mut hash = RequestHasher::new();
    hash.field(data.as_bytes());
    hash.field(source.as_bytes());
    hash.field(source_created_at.as_bytes());
    match data_filename {
        Some(value) => {
            hash.field(&[1]);
            hash.field(value.as_bytes());
        }
        None => hash.field(&[0]),
    }
    hash.field(&(artifacts.len() as u64).to_be_bytes());
    for artifact in artifacts {
        hash.field(artifact.original_filename.as_bytes());
        hash.field(artifact.media_type.as_bytes());
        hash.field(artifact.role.as_bytes());
        hash.field(Sha256::digest(&artifact.data).as_slice());
    }
    hash.finish()
}

#[allow(clippy::too_many_arguments)]
fn node_request_hash(
    id: NodeId,
    provenance_id: ProvenanceId,
    owner: Option<NodeId>,
    short_name: &str,
    short_description: &str,
    long_description: &str,
    model_attribution: &str,
    fixed_connections: &[NodeId],
    recent_connections: &[NodeId],
) -> [u8; 32] {
    let mut hash = RequestHasher::new();
    hash.id(id);
    hash.id(provenance_id);
    hash.optional_id(owner);
    hash.field(short_name.as_bytes());
    hash.field(short_description.as_bytes());
    hash.field(long_description.as_bytes());
    hash.field(model_attribution.as_bytes());
    hash.ids(fixed_connections);
    hash.ids(recent_connections);
    hash.finish()
}

fn require_provenance(tx: &Transaction<'_>, id: ProvenanceId) -> Result<(), Error> {
    let exists: bool = tx.query_row(
        "SELECT EXISTS(SELECT 1 FROM data_provenance_nodes WHERE id=?1)",
        [id.as_bytes()],
        |row| row.get(0),
    )?;
    if exists {
        Ok(())
    } else {
        Err(Error::NotFound("Provenance"))
    }
}

fn require_owner(
    tx: &Transaction<'_>,
    owner: Option<NodeId>,
    creating: NodeId,
) -> Result<(), Error> {
    let Some(owner) = owner else {
        return Ok(());
    };
    if owner == creating {
        return Ok(());
    }
    let exists: bool = tx.query_row(
        "SELECT EXISTS(SELECT 1 FROM knowledge_nodes WHERE id=?1)",
        [owner.as_bytes()],
        |row| row.get(0),
    )?;
    if exists {
        Ok(())
    } else {
        Err(Error::NotFound("Owner knowledge node"))
    }
}

fn require_connections(
    tx: &Transaction<'_>,
    ids: &[NodeId],
    creating: NodeId,
) -> Result<(), Error> {
    for id in ids {
        if *id == creating {
            continue;
        }
        let exists: bool = tx.query_row(
            "SELECT EXISTS(SELECT 1 FROM knowledge_nodes WHERE id=?1)",
            [id.as_bytes()],
            |row| row.get(0),
        )?;
        if !exists {
            return Err(Error::NotFound("Connected knowledge node"));
        }
    }
    Ok(())
}

fn replace_connections(
    tx: &Transaction<'_>,
    source: NodeId,
    fixed: &[NodeId],
    recent: &[NodeId],
) -> Result<(), Error> {
    for table in ["fixed_connections", "recent_connections"] {
        tx.execute(
            &format!("DELETE FROM {table} WHERE source_node_id=?1"),
            [source.as_bytes()],
        )?;
    }
    for (table, ids) in [("fixed_connections", fixed), ("recent_connections", recent)] {
        for (position, target) in ids.iter().enumerate() {
            tx.execute(
                &format!(
                    "INSERT INTO {table}(source_node_id,target_node_id,position) VALUES(?1,?2,?3)"
                ),
                params![source.as_bytes(), target.as_bytes(), position as i64],
            )?;
        }
    }
    Ok(())
}

fn read_connections(db: &Connection, table: &str, source: NodeId) -> Result<Vec<NodeId>, Error> {
    let mut statement = db.prepare(&format!(
        "SELECT target_node_id FROM {table} WHERE source_node_id=?1 ORDER BY position"
    ))?;
    Ok(statement
        .query_map([source.as_bytes()], |row| {
            NodeId::from_blob(row.get::<_, Vec<u8>>(0)?)
        })?
        .collect::<Result<Vec<_>, _>>()?)
}

fn map_constraint(message: &'static str) -> impl FnOnce(rusqlite::Error) -> Error {
    move |error| {
        if error.sqlite_error_code() == Some(rusqlite::ErrorCode::ConstraintViolation) {
            Error::Conflict(message.into())
        } else {
            Error::Database(error)
        }
    }
}

fn count_characters(value: &str) -> u64 {
    u64::try_from(value.chars().count()).unwrap_or(u64::MAX)
}

fn count_words(value: &str) -> u64 {
    u64::try_from(value.split_whitespace().count()).unwrap_or(u64::MAX)
}

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

    fn kmap() -> Kmap {
        Kmap::open(":memory:").unwrap()
    }

    fn provenance(kmap: &mut Kmap, data: &str) -> ProvenanceId {
        kmap.create_provenance(
            IdempotencyId::random(),
            NewProvenance {
                data: data.into(),
                source: "test".into(),
                source_created_at: "2026-07-18T00:00:00Z".into(),
            },
        )
        .unwrap()
    }

    fn create(kmap: &mut Kmap, id: NodeId, provenance_id: ProvenanceId, name: &str) -> Node {
        kmap.create_node(
            IdempotencyId::random(),
            CreateNode {
                id,
                provenance_id,
                owner: Owner::SelfNode,
                short_name: name.into(),
                short_description: String::new(),
                long_description: String::new(),
                model_attribution: "test-model".into(),
                fixed_connections: vec![],
                recent_connections: vec![],
            },
        )
        .unwrap()
    }

    #[test]
    fn creates_updates_and_reads_history() {
        let mut kmap = kmap();
        let created_from = provenance(&mut kmap, "created");
        let updated_from = provenance(&mut kmap, "updated");
        let id = NodeId::random();
        create(&mut kmap, id, created_from, "Test Node");
        let updated = kmap
            .update_node(
                IdempotencyId::random(),
                id,
                UpdateNode {
                    provenance_id: updated_from,
                    owner: Owner::SelfNode,
                    short_name: "Updated Node".into(),
                    short_description: "Updated.".into(),
                    long_description: "Complete updated text.".into(),
                    model_attribution: "new-model".into(),
                    fixed_connections: vec![],
                    recent_connections: vec![],
                },
            )
            .unwrap();
        assert_eq!(updated.short_name, "Updated Node");
        assert_eq!(updated.owner_node_id, Some(id));
        assert_eq!(
            kmap.get_node_history(id).unwrap(),
            vec![updated_from, created_from]
        );
    }

    #[test]
    fn required_idempotency_ids_replay_mutations_and_reject_reuse() {
        let mut kmap = kmap();
        let provenance_operation = IdempotencyId::random();
        let provenance_input = NewProvenance {
            data: "source material".into(),
            source: "test".into(),
            source_created_at: "2026-07-18T00:00:00Z".into(),
        };
        let provenance_id = kmap
            .create_provenance(provenance_operation, provenance_input.clone())
            .unwrap();
        assert_eq!(
            kmap.create_provenance(provenance_operation, provenance_input.clone())
                .unwrap(),
            provenance_id
        );

        let node_id = NodeId::random();
        let create_operation = IdempotencyId::random();
        let create_input = CreateNode {
            id: node_id,
            provenance_id,
            owner: Owner::SelfNode,
            short_name: "Idempotent Node".into(),
            short_description: String::new(),
            long_description: "Initial state.".into(),
            model_attribution: "test-model".into(),
            fixed_connections: vec![],
            recent_connections: vec![],
        };
        kmap.create_node(create_operation, create_input.clone())
            .unwrap();
        kmap.create_node(create_operation, create_input).unwrap();
        assert_eq!(kmap.get_node_history(node_id).unwrap().len(), 1);

        let update_operation = IdempotencyId::random();
        let update_input = UpdateNode {
            provenance_id,
            owner: Owner::SelfNode,
            short_name: "Idempotent Node".into(),
            short_description: "Updated once.".into(),
            long_description: "Updated state.".into(),
            model_attribution: "test-model".into(),
            fixed_connections: vec![],
            recent_connections: vec![],
        };
        let first = kmap
            .update_node(update_operation, node_id, update_input.clone())
            .unwrap();
        let replay = kmap
            .update_node(update_operation, node_id, update_input.clone())
            .unwrap();
        assert_eq!(first.last_modified_at, replay.last_modified_at);
        assert_eq!(kmap.get_node_history(node_id).unwrap().len(), 2);

        let mut changed_input = update_input;
        changed_input.long_description = "Different state.".into();
        assert!(matches!(
            kmap.update_node(update_operation, node_id, changed_input),
            Err(Error::Conflict(_))
        ));
        assert_eq!(kmap.get_node_history(node_id).unwrap().len(), 2);
        let receipt_count: i64 = kmap
            .db
            .query_row("SELECT COUNT(*) FROM idempotency_receipts", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(receipt_count, 3);
    }

    #[test]
    fn externalizes_provenance_and_preserves_suffixed_original_filenames() {
        let directory = std::env::temp_dir().join(format!(
            "kweb-artifact-test-{}",
            URL_SAFE_NO_PAD.encode(IdempotencyId::random().0)
        ));
        let database = directory.join("kweb.sqlite3");
        let artifacts = directory.join("kweb-provenance-artifacts");
        fs::create_dir(&directory).unwrap();
        let mut kmap = Kmap::open_with_artifacts(&database, &artifacts).unwrap();
        let operation = IdempotencyId::random();
        let input = NewProvenance {
            data: "x".repeat(INLINE_PROVENANCE_LIMIT + 1),
            source: "conversation-history".into(),
            source_created_at: "2026-07-18T00:00:00Z".into(),
        };
        let storage = ProvenanceStorage {
            data_filename: "conversation-archive.json".into(),
            artifacts: vec![NewProvenanceArtifact {
                original_filename: "telegram-vnote.wav".into(),
                media_type: "audio/wav".into(),
                role: "media".into(),
                data: b"voice note".to_vec(),
            }],
        };
        let id = kmap
            .create_provenance_with_storage(operation, input.clone(), storage.clone())
            .unwrap();
        assert_eq!(
            kmap.create_provenance_with_storage(operation, input.clone(), storage)
                .unwrap(),
            id
        );

        let provenance = kmap.get_provenance(id).unwrap();
        assert_eq!(provenance.data, input.data);
        assert_eq!(provenance.artifacts.len(), 2);
        let media = &provenance.artifacts[0];
        assert_eq!(media.original_filename, "telegram-vnote.wav");
        assert_eq!(media.role, "media");
        let (shard, filename) = media.relative_path.split_once('/').unwrap();
        let suffix = filename
            .strip_prefix("telegram-vnote.")
            .unwrap()
            .strip_suffix(".wav")
            .unwrap();
        assert_eq!(suffix.len(), 12);
        assert!(
            suffix
                .bytes()
                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
        );
        assert_eq!(shard, &suffix[..2]);
        assert_eq!(
            fs::read(artifacts.join(&media.relative_path)).unwrap(),
            b"voice note"
        );
        assert_eq!(provenance.artifacts[1].role, "source-data");
        let artifact_count: i64 = kmap
            .db
            .query_row("SELECT COUNT(*) FROM provenance_artifacts", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(artifact_count, 2);
        let externalized: bool = kmap
            .db
            .query_row(
                "SELECT data IS NULL AND data_artifact_path IS NOT NULL FROM data_provenance_nodes WHERE id=?1",
                [id.as_bytes()],
                |row| row.get(0),
            )
            .unwrap();
        assert!(externalized);
        drop(kmap);
        fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn stores_connection_ids_without_connection_policy() {
        let mut kmap = kmap();
        let source = provenance(&mut kmap, "source");
        let first = NodeId::random();
        let second = NodeId::random();
        create(&mut kmap, first, source, "First Node");
        create(&mut kmap, second, source, "Second Node");
        let parent = NodeId::random();
        let node = kmap
            .create_node(
                IdempotencyId::random(),
                CreateNode {
                    id: parent,
                    provenance_id: source,
                    owner: Owner::SelfNode,
                    short_name: "Parent Node".into(),
                    short_description: String::new(),
                    long_description: String::new(),
                    model_attribution: "test-model".into(),
                    fixed_connections: vec![second],
                    recent_connections: vec![first, second],
                },
            )
            .unwrap();
        assert_eq!(node.fixed_connections, vec![second]);
        assert_eq!(node.recent_connections, vec![first, second]);
    }

    #[test]
    fn stats_are_typed_and_serialize_additively() {
        let mut kmap = kmap();
        let source = provenance(&mut kmap, "source");
        kmap.create_node(
            IdempotencyId::random(),
            CreateNode {
                id: NodeId::random(),
                provenance_id: source,
                owner: Owner::SelfNode,
                short_name: "Alpha".into(),
                short_description: "Short note".into(),
                long_description: "Long description here".into(),
                model_attribution: "test-model".into(),
                fixed_connections: vec![],
                recent_connections: vec![],
            },
        )
        .unwrap();
        kmap.create_node(
            IdempotencyId::random(),
            CreateNode {
                id: NodeId::random(),
                provenance_id: source,
                owner: Owner::SelfNode,
                short_name: "Unicode 🦀".into(),
                short_description: "Second".into(),
                long_description: "More words".into(),
                model_attribution: "test-model".into(),
                fixed_connections: vec![],
                recent_connections: vec![],
            },
        )
        .unwrap();
        let stats = kmap.stats().unwrap();
        assert_eq!(stats.node_count(), 2);
        assert_eq!(stats.full_node_characters(), 65);
        assert_eq!(stats.full_node_words(), 11);
        assert_eq!(stats.estimated_full_node_tokens(), 17);
        assert_eq!(stats.long_description_characters(), 31);
        assert_eq!(stats.long_description_words(), 5);
        assert_eq!(stats.estimated_long_description_tokens(), 8);
        let json = serde_json::to_value(stats).unwrap();
        assert_eq!(json["node_count"], 2);
        assert!(json.get("estimated_full_node_tokens").is_some());
    }
}