kcode-kweb-manager 0.1.1

//! Kennedy's typed application manager for one privately owned Kweb database.

use std::{
    fmt,
    path::{Path, PathBuf},
    str::FromStr,
    sync::{Arc, Mutex},
};

use chrono::Utc;
use kcode_commit_session::{CommitReceipt, CommitRequest};
use kcode_kweb_db::{
    Error as KwebError, KwebDb, Node, NodeHistory, NodeId, ObjectId, Owner, Provenance,
};
use kcode_server_object_envelopes::{StoredProvenance, decode_provenance, encode_provenance};
use rusqlite::{Connection, OptionalExtension, params};
use sha2::{Digest, Sha256};

const MAX_EMBEDDED_PROVENANCE_BYTES: usize = 1024 * 1024;

/// Result returned by `kcode-kweb-manager`.
pub type Result<T> = std::result::Result<T, Error>;

/// Stable error category for application adapters.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum ErrorKind {
    /// The caller supplied an invalid identifier or mutation.
    InvalidInput,
    /// A requested node, object, or provenance value does not exist.
    NotFound,
    /// Idempotency state or a concurrent database operation conflicts.
    Conflict,
    /// Persistence is unavailable, corrupt, or otherwise failed internally.
    Internal,
}

/// Error returned by a Kweb manager operation.
#[derive(Debug)]
pub struct Error {
    kind: ErrorKind,
    message: String,
}

impl Error {
    /// Returns the stable category suitable for transport mapping.
    pub fn kind(&self) -> ErrorKind {
        self.kind
    }

    fn invalid(message: impl Into<String>) -> Self {
        Self {
            kind: ErrorKind::InvalidInput,
            message: message.into(),
        }
    }

    fn conflict(message: impl Into<String>) -> Self {
        Self {
            kind: ErrorKind::Conflict,
            message: message.into(),
        }
    }

    fn internal(error: impl fmt::Display) -> Self {
        Self {
            kind: ErrorKind::Internal,
            message: error.to_string(),
        }
    }
}

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

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

impl From<KwebError> for Error {
    fn from(error: KwebError) -> Self {
        let kind = match error {
            KwebError::InvalidInput(_) | KwebError::InvalidTransaction(_) => {
                ErrorKind::InvalidInput
            }
            KwebError::NotFound(_) => ErrorKind::NotFound,
            KwebError::Busy(_) => ErrorKind::Conflict,
            KwebError::Io(_)
            | KwebError::Corrupt(_)
            | KwebError::InvalidConfig(_)
            | KwebError::OfflineUpgradeRequired(_) => ErrorKind::Internal,
        };
        Self {
            kind,
            message: error.to_string(),
        }
    }
}

impl From<kcode_commit_session::Error> for Error {
    fn from(error: kcode_commit_session::Error) -> Self {
        let kind = match error.kind() {
            kcode_commit_session::ErrorKind::InvalidInput => ErrorKind::InvalidInput,
            kcode_commit_session::ErrorKind::NotFound => ErrorKind::NotFound,
            kcode_commit_session::ErrorKind::Conflict => ErrorKind::Conflict,
            _ => ErrorKind::Internal,
        };
        Self {
            kind,
            message: error.to_string(),
        }
    }
}

/// One idempotent provenance-object creation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CreateProvenance {
    /// Stable 16-byte request identity encoded as lowercase hexadecimal.
    pub idempotency_id: String,
    /// Application provenance envelope stored as the object payload.
    pub value: StoredProvenance,
    /// Kweb provenance describing the transaction that stores the envelope.
    pub storage_provenance: Provenance,
}

/// Node fields editable by ordinary Kennedy Kweb mutations.
///
/// Object references are intentionally absent. New nodes start without
/// objects, and updates preserve the current object list.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct NodeContents {
    pub short_name: String,
    pub short_description: String,
    pub long_description: String,
    pub owner: Owner,
    pub fixed_connections: Vec<NodeId>,
    pub recent_connections: Vec<NodeId>,
}

impl NodeContents {
    fn into_data(self, objects: Vec<ObjectId>) -> kcode_kweb_db::NodeData {
        kcode_kweb_db::NodeData {
            short_name: self.short_name,
            short_description: self.short_description,
            long_description: self.long_description,
            owner: self.owner,
            fixed_connections: self.fixed_connections,
            recent_connections: self.recent_connections,
            objects,
        }
    }
}

/// Shared input for idempotent node creation and update.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct NodeWrite {
    /// Stable 16-byte request identity encoded as lowercase hexadecimal.
    pub idempotency_id: String,
    /// Stored application provenance used to describe the Kweb transaction.
    pub provenance_id: ObjectId,
    /// Author or model attribution recorded on the Kweb transaction.
    pub author: String,
    /// Complete ordinary node fields.
    pub contents: NodeContents,
}

/// Cloneable sole owner of one live Kweb database handle.
#[derive(Clone)]
pub struct KwebManager {
    database: Arc<KwebDb>,
    receipt_database: PathBuf,
    receipts: Arc<Mutex<Connection>>,
}

impl KwebManager {
    /// Takes ownership of an open Kweb database and opens its receipt lane.
    pub fn open(database: KwebDb, receipt_database: impl AsRef<Path>) -> Result<Self> {
        let receipt_database = receipt_database.as_ref().to_path_buf();
        let receipts = Connection::open(&receipt_database).map_err(Error::internal)?;
        receipts
            .execute_batch(
                "PRAGMA busy_timeout=15000;
                 CREATE TABLE IF NOT EXISTS kmap_idempotency_receipts (
                     idempotency_id TEXT PRIMARY KEY CHECK(length(idempotency_id)=32),
                     operation TEXT NOT NULL,
                     digest_version INTEGER NOT NULL DEFAULT 1,
                     request_sha256 BLOB NOT NULL CHECK(length(request_sha256)=32),
                     result_id TEXT CHECK(result_id IS NULL OR length(result_id)=8),
                     started_at TEXT NOT NULL,
                     committed_at TEXT,
                     CHECK((result_id IS NULL) = (committed_at IS NULL))
                 );
                 DROP TABLE IF EXISTS kmap_object_provenance;",
            )
            .map_err(Error::internal)?;
        ensure_digest_version_column(&receipts)?;
        Ok(Self {
            database: Arc::new(database),
            receipt_database,
            receipts: Arc::new(Mutex::new(receipts)),
        })
    }

    /// Loads one current node.
    pub fn get_node(&self, id: NodeId) -> Result<Node> {
        self.database.get_node(id).map_err(Error::from)
    }

    /// Loads one node's complete visible history.
    pub fn get_node_history(&self, id: NodeId) -> Result<NodeHistory> {
        self.database.get_node_history(id).map_err(Error::from)
    }

    /// Loads the exact opaque bytes of one Kweb object.
    pub fn get_object(&self, id: ObjectId) -> Result<Vec<u8>> {
        self.database.get_object(id).map_err(Error::from)
    }

    /// Loads exact bytes and provenance from the signed creating Kweb transaction.
    pub fn get_object_with_provenance(&self, id: ObjectId) -> Result<(Vec<u8>, Provenance)> {
        self.database
            .get_object_with_provenance(id)
            .map_err(Error::from)
    }

    /// Idempotently stores one application provenance envelope.
    pub fn create_provenance(&self, request: CreateProvenance) -> Result<ObjectId> {
        validate_idempotency_id(&request.idempotency_id)?;
        let encoded =
            encode_provenance(&request.value).map_err(|error| Error::invalid(error.to_string()))?;
        let legacy_digest = legacy_provenance_request_digest(&request);
        let digest = provenance_request_digest(&encoded, &request.storage_provenance);
        let storage_provenance = request.storage_provenance.clone();
        let result = self.with_idempotency(
            &request.idempotency_id,
            "create_provenance",
            VersionedDigest::v2(digest, legacy_digest),
            |result_id| {
                let id = ObjectId::from_str(result_id).map_err(|error| {
                    Error::internal(format!("invalid stored provenance receipt: {error}"))
                })?;
                let (stored_bytes, creating_provenance) =
                    self.database.get_object_with_provenance(id)?;
                let stored_value = decode_provenance(&stored_bytes).map_err(|error| {
                    Error::internal(format!("invalid stored provenance object {id}: {error}"))
                })?;
                Ok(stored_value == request.value
                    && creating_provenance == request.storage_provenance)
            },
            || {
                let mut transaction = self.database.start_transaction(storage_provenance)?;
                let id = transaction.create_object(encoded)?;
                transaction.finalize()?;
                Ok(id.to_string())
            },
        )?;
        let id = ObjectId::from_str(&result).map_err(|error| {
            Error::internal(format!("invalid stored provenance receipt: {error}"))
        })?;
        Ok(id)
    }

    /// Idempotently creates a node without object attachments.
    pub fn create_node(&self, request: NodeWrite) -> Result<Node> {
        validate_idempotency_id(&request.idempotency_id)?;
        let digest = node_request_digest("create_node", None, &request);
        let result =
            self.with_idempotency(
                &request.idempotency_id,
                "create_node",
                VersionedDigest::v1(digest),
                |_| Ok(true),
                || {
                    let provenance = self.load_provenance(request.provenance_id)?;
                    let mut transaction = self.database.start_transaction(
                        transaction_provenance(&provenance, request.provenance_id, request.author),
                    )?;
                    let id = transaction.create_node(request.contents.into_data(Vec::new()))?;
                    transaction.finalize()?;
                    Ok(id.to_string())
                },
            )?;
        let id = NodeId::from_str(&result)
            .map_err(|error| Error::internal(format!("invalid stored node receipt: {error}")))?;
        self.get_node(id)
    }

    /// Idempotently updates ordinary node fields while retaining its objects.
    pub fn update_node(&self, id: NodeId, request: NodeWrite) -> Result<Node> {
        validate_idempotency_id(&request.idempotency_id)?;
        let digest = node_request_digest("update_node", Some(id), &request);
        self.with_idempotency(
            &request.idempotency_id,
            "update_node",
            VersionedDigest::v1(digest),
            |_| Ok(true),
            || {
                let provenance = self.load_provenance(request.provenance_id)?;
                let objects = self.database.get_node(id)?.data.objects;
                let mut transaction = self.database.start_transaction(transaction_provenance(
                    &provenance,
                    request.provenance_id,
                    request.author,
                ))?;
                transaction.update_node(id, request.contents.into_data(objects))?;
                transaction.finalize()?;
                Ok(id.to_string())
            },
        )?;
        self.get_node(id)
    }

    /// Stores exact opaque object bytes in one Kweb transaction.
    pub fn store_object(&self, provenance: Provenance, bytes: Vec<u8>) -> Result<ObjectId> {
        let mut transaction = self.database.start_transaction(provenance)?;
        let id = transaction.create_object(bytes)?;
        transaction.finalize()?;
        Ok(id)
    }

    /// Atomically commits one complete session through `kcode-commit-session`.
    pub fn commit_session(&self, request: CommitRequest) -> Result<CommitReceipt> {
        let _receipt_lane = self
            .receipts
            .lock()
            .map_err(|_| Error::internal("Kweb manager idempotency mutex is poisoned"))?;
        kcode_commit_session::commit_session(&self.database, &self.receipt_database, request)
            .map_err(Error::from)
    }

    fn load_provenance(&self, id: ObjectId) -> Result<StoredProvenance> {
        let bytes = self.database.get_object(id)?;
        decode_provenance(&bytes).map_err(|error| {
            Error::internal(format!("invalid stored provenance object {id}: {error}"))
        })
    }

    fn with_idempotency(
        &self,
        idempotency_id: &str,
        operation: &'static str,
        digest: VersionedDigest,
        legacy_result_matches: impl FnOnce(&str) -> Result<bool>,
        mutation: impl FnOnce() -> Result<String>,
    ) -> Result<String> {
        let receipts = self
            .receipts
            .lock()
            .map_err(|_| Error::internal("Kweb manager idempotency mutex is poisoned"))?;
        let existing = receipts
            .query_row(
                "SELECT operation,digest_version,request_sha256,result_id
                 FROM kmap_idempotency_receipts WHERE idempotency_id=?1",
                [idempotency_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, i64>(1)?,
                        row.get::<_, Vec<u8>>(2)?,
                        row.get::<_, Option<String>>(3)?,
                    ))
                },
            )
            .optional()
            .map_err(Error::internal)?;
        if let Some((stored_operation, stored_version, stored_hash, result_id)) = existing {
            let (expected_hash, verify_legacy_result) = digest.for_version(stored_version)?;
            if stored_operation != operation || stored_hash.as_slice() != expected_hash {
                return Err(Error::conflict(
                    "idempotency_id was already used for a different Kweb manager mutation",
                ));
            }
            let result_id = result_id.ok_or_else(|| {
                Error::conflict(
                    "a prior Kweb manager mutation with this idempotency_id has an unknown outcome; offline recovery is required",
                )
            })?;
            if verify_legacy_result && !legacy_result_matches(&result_id)? {
                return Err(Error::conflict(
                    "idempotency_id was already used for different provenance contents",
                ));
            }
            return Ok(result_id);
        }

        receipts
            .execute(
                "INSERT INTO kmap_idempotency_receipts(
                     idempotency_id,operation,digest_version,request_sha256,
                     result_id,started_at,committed_at
                 ) VALUES(?1,?2,?3,?4,NULL,?5,NULL)",
                params![
                    idempotency_id,
                    operation,
                    digest.current_version,
                    digest.current.as_slice(),
                    now_text(),
                ],
            )
            .map_err(Error::internal)?;
        let result_id = mutation()?;
        let updated = receipts
            .execute(
                "UPDATE kmap_idempotency_receipts
                 SET result_id=?2,committed_at=?3
                 WHERE idempotency_id=?1 AND result_id IS NULL",
                params![idempotency_id, &result_id, now_text()],
            )
            .map_err(Error::internal)?;
        if updated != 1 {
            return Err(Error::internal(
                "Kweb manager idempotency receipt disappeared during mutation",
            ));
        }
        Ok(result_id)
    }
}

fn ensure_digest_version_column(receipts: &Connection) -> Result<()> {
    let present = receipts
        .query_row(
            "SELECT COUNT(*) FROM pragma_table_info('kmap_idempotency_receipts')
             WHERE name='digest_version'",
            [],
            |row| row.get::<_, i64>(0),
        )
        .map_err(Error::internal)?;
    if present == 0 {
        receipts
            .execute(
                "ALTER TABLE kmap_idempotency_receipts
                 ADD COLUMN digest_version INTEGER NOT NULL DEFAULT 1",
                [],
            )
            .map_err(Error::internal)?;
    }
    Ok(())
}

struct VersionedDigest {
    current_version: i64,
    current: [u8; 32],
    legacy: Option<[u8; 32]>,
}

impl VersionedDigest {
    fn v1(current: [u8; 32]) -> Self {
        Self {
            current_version: 1,
            current,
            legacy: None,
        }
    }

    fn v2(current: [u8; 32], legacy: [u8; 32]) -> Self {
        Self {
            current_version: 2,
            current,
            legacy: Some(legacy),
        }
    }

    fn for_version(&self, version: i64) -> Result<(&[u8; 32], bool)> {
        if version == self.current_version {
            return Ok((&self.current, false));
        }
        if version == 1
            && let Some(legacy) = &self.legacy
        {
            return Ok((legacy, true));
        }
        Err(Error::internal(format!(
            "unsupported Kweb manager idempotency digest version {version}"
        )))
    }
}

fn now_text() -> String {
    Utc::now().to_rfc3339()
}

fn validate_idempotency_id(value: &str) -> Result<()> {
    if value.len() != 32
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(Error::invalid(
            "idempotency_id must encode 16 bytes as lowercase hexadecimal",
        ));
    }
    Ok(())
}

struct RequestDigest(Sha256);

impl RequestDigest {
    fn new(operation: &str) -> Self {
        let mut hash = Sha256::new();
        hash.update(b"kennedy kmap idempotency v1\0");
        let mut value = Self(hash);
        value.field(operation.as_bytes());
        value
    }

    fn new_v2(operation: &str) -> Self {
        let mut hash = Sha256::new();
        hash.update(b"kennedy kweb manager idempotency v2\0");
        let mut value = Self(hash);
        value.field(operation.as_bytes());
        value
    }

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

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

fn provenance_request_digest(encoded: &[u8], storage: &Provenance) -> [u8; 32] {
    let mut hash = RequestDigest::new_v2("create_provenance");
    hash.field(encoded);
    hash.field(storage.author.as_bytes());
    hash.field(storage.source.as_bytes());
    hash.field(&storage.source_created_at.timestamp().to_be_bytes());
    hash.field(
        &storage
            .source_created_at
            .timestamp_subsec_nanos()
            .to_be_bytes(),
    );
    hash.field(storage.data.as_bytes());
    hash.finish()
}

// Existing receipts used this incomplete digest. It remains only to replay an
// exact legacy result after independently checking the stored Kweb object and
// its signed creating provenance against the complete request.
fn legacy_provenance_request_digest(request: &CreateProvenance) -> [u8; 32] {
    let mut hash = RequestDigest::new("create_provenance");
    hash.field(request.value.data.as_bytes());
    hash.field(request.value.source.as_bytes());
    hash.field(&request.value.source_created_at.timestamp().to_be_bytes());
    hash.field(
        &request
            .value
            .source_created_at
            .timestamp_subsec_nanos()
            .to_be_bytes(),
    );
    hash.field(b"");
    hash.field(&(request.value.artifacts.len() as u64).to_be_bytes());
    for artifact in &request.value.artifacts {
        hash.field(artifact.original_filename.as_bytes());
        hash.field(artifact.media_type.as_bytes());
        hash.field(&artifact.sha256);
    }
    hash.field(request.storage_provenance.author.as_bytes());
    hash.field(request.storage_provenance.source.as_bytes());
    hash.field(
        &request
            .storage_provenance
            .source_created_at
            .timestamp()
            .to_be_bytes(),
    );
    hash.field(
        &request
            .storage_provenance
            .source_created_at
            .timestamp_subsec_nanos()
            .to_be_bytes(),
    );
    hash.field(request.storage_provenance.data.as_bytes());
    hash.finish()
}

fn node_request_digest(operation: &str, id: Option<NodeId>, request: &NodeWrite) -> [u8; 32] {
    let mut hash = RequestDigest::new(operation);
    hash.field(&id.map(NodeId::to_bytes).unwrap_or([0; 6]));
    hash.field(&request.provenance_id.to_bytes());
    hash.field(request.author.as_bytes());
    hash.field(request.contents.short_name.as_bytes());
    hash.field(request.contents.short_description.as_bytes());
    hash.field(request.contents.long_description.as_bytes());
    match request.contents.owner {
        Owner::Unowned => hash.field(&[0]),
        Owner::SelfNode => hash.field(&[1]),
        Owner::Node(owner) => {
            hash.field(&[2]);
            hash.field(&owner.to_bytes());
        }
    }
    hash.field(&(request.contents.fixed_connections.len() as u64).to_be_bytes());
    for connection in &request.contents.fixed_connections {
        hash.field(&connection.to_bytes());
    }
    hash.field(&(request.contents.recent_connections.len() as u64).to_be_bytes());
    for connection in &request.contents.recent_connections {
        hash.field(&connection.to_bytes());
    }
    hash.finish()
}

fn transaction_provenance(
    stored: &StoredProvenance,
    object_id: ObjectId,
    author: String,
) -> Provenance {
    let data = if stored.data.len() <= MAX_EMBEDDED_PROVENANCE_BYTES {
        stored.data.clone()
    } else {
        format!("Kennedy provenance is stored in object {object_id}.")
    };
    Provenance {
        author,
        source: stored.source.clone(),
        source_created_at: stored.source_created_at,
        data,
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        fs,
        sync::{
            Arc,
            atomic::{AtomicU64, Ordering},
        },
    };

    use chrono::{Duration, TimeZone, Utc};
    use kcode_commit_session::CommitRequest;
    use kcode_kweb_db::{Config, NodeData, NoopGossip, WriterId};
    use kcode_server_object_envelopes::{StoredArtifact, decode_provenance};

    use super::*;

    static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1);

    struct TestDirectory(PathBuf);

    impl TestDirectory {
        fn new() -> Self {
            let sequence = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed);
            let path = std::env::temp_dir().join(format!(
                "kcode-kweb-manager-test-{}-{sequence}",
                std::process::id()
            ));
            fs::create_dir_all(&path).unwrap();
            Self(path)
        }

        fn join(&self, value: &str) -> PathBuf {
            self.0.join(value)
        }
    }

    impl Drop for TestDirectory {
        fn drop(&mut self) {
            fs::remove_dir_all(&self.0).unwrap();
        }
    }

    fn config() -> Config {
        let signing_key = [7; 32];
        Config {
            signing_key,
            writers_by_priority: vec![WriterId::from_signing_key(&signing_key)],
            gossip: Arc::new(NoopGossip),
        }
    }

    fn timestamp() -> chrono::DateTime<Utc> {
        Utc.with_ymd_and_hms(2026, 7, 28, 12, 0, 0).unwrap()
    }

    fn transaction_provenance_for(label: &str) -> Provenance {
        Provenance {
            author: "test".into(),
            source: label.into(),
            source_created_at: timestamp(),
            data: format!("{label} transaction"),
        }
    }

    fn provenance_request(idempotency_id: &str, data: &str) -> CreateProvenance {
        CreateProvenance {
            idempotency_id: idempotency_id.into(),
            value: StoredProvenance {
                data: data.into(),
                source: "test-source".into(),
                source_created_at: timestamp(),
                artifacts: Vec::new(),
            },
            storage_provenance: transaction_provenance_for("provenance-storage"),
        }
    }

    fn artifact() -> StoredArtifact {
        StoredArtifact {
            object_id: ObjectId::from_bytes([0x80, 1, 2, 3, 4, 5]).unwrap(),
            original_filename: "source.txt".into(),
            media_type: "text/plain".into(),
            role: "source".into(),
            byte_length: 12,
            sha256: [7; 32],
        }
    }

    fn node_write(idempotency_id: &str, provenance_id: ObjectId, short_name: &str) -> NodeWrite {
        NodeWrite {
            idempotency_id: idempotency_id.into(),
            provenance_id,
            author: "test-model".into(),
            contents: NodeContents {
                short_name: short_name.into(),
                short_description: "short".into(),
                long_description: "long".into(),
                owner: Owner::SelfNode,
                fixed_connections: Vec::new(),
                recent_connections: Vec::new(),
            },
        }
    }

    #[test]
    fn provenance_and_node_mutations_are_idempotent_and_typed() {
        let directory = TestDirectory::new();
        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();

        let create_request =
            provenance_request("00000000000000000000000000000001", "source material");
        let provenance_id = kmap.create_provenance(create_request.clone()).unwrap();
        assert_eq!(
            kmap.create_provenance(create_request).unwrap(),
            provenance_id
        );
        let stored = decode_provenance(&kmap.get_object(provenance_id).unwrap()).unwrap();
        assert_eq!(stored.data, "source material");

        let conflict = kmap
            .create_provenance(provenance_request(
                "00000000000000000000000000000001",
                "different material",
            ))
            .unwrap_err();
        assert_eq!(conflict.kind(), ErrorKind::Conflict);

        let write = node_write(
            "00000000000000000000000000000002",
            provenance_id,
            "Created node",
        );
        let node = kmap.create_node(write.clone()).unwrap();
        assert_eq!(kmap.create_node(write).unwrap(), node);
        assert_eq!(kmap.get_node(node.id).unwrap(), node);
        assert!(node.data.objects.is_empty());
    }

    #[test]
    fn provenance_idempotency_includes_every_storage_provenance_field() {
        let directory = TestDirectory::new();
        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();

        let request = provenance_request("00000000000000000000000000000005", "source material");
        let provenance_id = kmap.create_provenance(request.clone()).unwrap();
        assert_eq!(
            kmap.create_provenance(request.clone()).unwrap(),
            provenance_id
        );

        let mut changed_author = request.clone();
        changed_author
            .storage_provenance
            .author
            .push_str("-changed");
        let mut changed_source = request.clone();
        changed_source
            .storage_provenance
            .source
            .push_str("-changed");
        let mut changed_timestamp = request.clone();
        changed_timestamp.storage_provenance.source_created_at += Duration::nanoseconds(1);
        let mut changed_data = request.clone();
        changed_data.storage_provenance.data.push_str("-changed");

        for changed in [
            changed_author,
            changed_source,
            changed_timestamp,
            changed_data,
        ] {
            let conflict = kmap.create_provenance(changed).unwrap_err();
            assert_eq!(conflict.kind(), ErrorKind::Conflict);
        }

        let (bytes, creating_provenance) = kmap.get_object_with_provenance(provenance_id).unwrap();
        assert_eq!(decode_provenance(&bytes).unwrap(), request.value);
        assert_eq!(creating_provenance, request.storage_provenance);
    }

    #[test]
    fn provenance_idempotency_includes_every_artifact_field() {
        let directory = TestDirectory::new();
        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
        let mut request = provenance_request("00000000000000000000000000000006", "source material");
        request.value.artifacts.push(artifact());
        let provenance_id = kmap.create_provenance(request.clone()).unwrap();
        assert_eq!(
            kmap.create_provenance(request.clone()).unwrap(),
            provenance_id
        );

        let mut changed_object = request.clone();
        changed_object.value.artifacts[0].object_id =
            ObjectId::from_bytes([0x80, 1, 2, 3, 4, 6]).unwrap();
        let mut changed_filename = request.clone();
        changed_filename.value.artifacts[0].original_filename = "other.txt".into();
        let mut changed_media_type = request.clone();
        changed_media_type.value.artifacts[0].media_type = "application/json".into();
        let mut changed_role = request.clone();
        changed_role.value.artifacts[0].role = "transcript".into();
        let mut changed_size = request.clone();
        changed_size.value.artifacts[0].byte_length += 1;
        let mut changed_sha256 = request;
        changed_sha256.value.artifacts[0].sha256[0] ^= 1;

        for changed in [
            changed_object,
            changed_filename,
            changed_media_type,
            changed_role,
            changed_size,
            changed_sha256,
        ] {
            let conflict = kmap.create_provenance(changed).unwrap_err();
            assert_eq!(conflict.kind(), ErrorKind::Conflict);
        }
    }

    #[test]
    fn legacy_provenance_receipts_replay_exactly_and_drop_duplicate_storage() {
        let directory = TestDirectory::new();
        let receipt_path = directory.join("application.sqlite3");
        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
        let mut request = provenance_request("00000000000000000000000000000007", "legacy source");
        request.value.artifacts.push(artifact());
        let encoded = encode_provenance(&request.value).unwrap();
        let mut transaction = database
            .start_transaction(request.storage_provenance.clone())
            .unwrap();
        let object_id = transaction.create_object(encoded).unwrap();
        transaction.finalize().unwrap();

        let receipts = Connection::open(&receipt_path).unwrap();
        receipts
            .execute_batch(
                "CREATE TABLE kmap_idempotency_receipts (
                     idempotency_id TEXT PRIMARY KEY CHECK(length(idempotency_id)=32),
                     operation TEXT NOT NULL,
                     request_sha256 BLOB NOT NULL CHECK(length(request_sha256)=32),
                     result_id TEXT CHECK(result_id IS NULL OR length(result_id)=8),
                     started_at TEXT NOT NULL,
                     committed_at TEXT,
                     CHECK((result_id IS NULL) = (committed_at IS NULL))
                 );
                 CREATE TABLE kmap_object_provenance(object_id TEXT PRIMARY KEY);",
            )
            .unwrap();
        receipts
            .execute(
                "INSERT INTO kmap_idempotency_receipts(
                     idempotency_id,operation,request_sha256,result_id,started_at,committed_at
                 ) VALUES(?1,'create_provenance',?2,?3,?4,?4)",
                params![
                    &request.idempotency_id,
                    legacy_provenance_request_digest(&request).as_slice(),
                    object_id.to_string(),
                    now_text(),
                ],
            )
            .unwrap();
        drop(receipts);

        let kmap = KwebManager::open(database, &receipt_path).unwrap();
        assert_eq!(kmap.create_provenance(request.clone()).unwrap(), object_id);
        let mut changed_role = request;
        changed_role.value.artifacts[0].role = "different".into();
        let conflict = kmap.create_provenance(changed_role).unwrap_err();
        assert_eq!(conflict.kind(), ErrorKind::Conflict);

        let receipts = Connection::open(receipt_path).unwrap();
        let duplicate_table_count = receipts
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master
                 WHERE type='table' AND name='kmap_object_provenance'",
                [],
                |row| row.get::<_, i64>(0),
            )
            .unwrap();
        assert_eq!(duplicate_table_count, 0);
        let digest_version = receipts
            .query_row(
                "SELECT digest_version FROM kmap_idempotency_receipts
                 WHERE idempotency_id=?1",
                [&"00000000000000000000000000000007"],
                |row| row.get::<_, i64>(0),
            )
            .unwrap();
        assert_eq!(digest_version, 1);
    }

    #[test]
    fn ordinary_updates_preserve_existing_object_attachments() {
        let directory = TestDirectory::new();
        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
        let mut transaction = database
            .start_transaction(transaction_provenance_for("seed"))
            .unwrap();
        let object_id = transaction.create_object(b"attachment".to_vec()).unwrap();
        let node_id = transaction
            .create_node(NodeData {
                short_name: "Before".into(),
                short_description: String::new(),
                long_description: String::new(),
                owner: Owner::SelfNode,
                fixed_connections: Vec::new(),
                recent_connections: Vec::new(),
                objects: vec![object_id],
            })
            .unwrap();
        transaction.finalize().unwrap();

        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
        let (stored_bytes, stored_provenance) = kmap.get_object_with_provenance(object_id).unwrap();
        assert_eq!(stored_bytes, b"attachment");
        assert_eq!(stored_provenance, transaction_provenance_for("seed"));

        let provenance_id = kmap
            .create_provenance(provenance_request(
                "00000000000000000000000000000003",
                "update source",
            ))
            .unwrap();
        let updated = kmap
            .update_node(
                node_id,
                node_write("00000000000000000000000000000004", provenance_id, "After"),
            )
            .unwrap();
        assert_eq!(updated.data.short_name, "After");
        assert_eq!(updated.data.objects, vec![object_id]);
    }

    #[test]
    fn object_storage_and_session_commits_share_the_owned_database() {
        let directory = TestDirectory::new();
        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();

        let opaque_provenance = transaction_provenance_for("opaque-object");
        let object_id = kmap
            .store_object(opaque_provenance.clone(), b"opaque bytes".to_vec())
            .unwrap();
        assert_eq!(kmap.get_object(object_id).unwrap(), b"opaque bytes");
        let (opaque_bytes, retained_provenance) =
            kmap.get_object_with_provenance(object_id).unwrap();
        assert_eq!(opaque_bytes, b"opaque bytes");
        assert_eq!(retained_provenance, opaque_provenance);

        let request = CommitRequest {
            idempotency_key: "session-test".into(),
            author: "test-model".into(),
            source_created_at: timestamp(),
            archive: b"{\"events\":[]}".to_vec(),
            objects: BTreeMap::new(),
            creates: BTreeMap::new(),
            updates: BTreeMap::new(),
        };
        let first = kmap.commit_session(request.clone()).unwrap();
        assert_eq!(
            kmap.get_object(first.session_object_id).unwrap(),
            b"{\"events\":[]}"
        );
        let (archive, provenance) = kmap
            .get_object_with_provenance(first.session_object_id)
            .unwrap();
        assert_eq!(archive, b"{\"events\":[]}");
        assert_eq!(provenance.author, "test-model");
        assert_eq!(provenance.source, "kennedy-session");
        assert_eq!(kmap.commit_session(request).unwrap(), first);
    }
}