kcode_kweb_manager/
lib.rs

1//! Kennedy's typed application manager for one privately owned Kweb database.
2
3use std::{
4    fmt,
5    path::{Path, PathBuf},
6    str::FromStr,
7    sync::{Arc, Mutex},
8};
9
10use chrono::Utc;
11use kcode_commit_session::{CommitReceipt, CommitRequest};
12use kcode_kweb_db::{
13    Error as KwebError, KwebDb, Node, NodeHistory, NodeId, ObjectId, Owner, Provenance,
14};
15use kcode_server_object_envelopes::{StoredProvenance, decode_provenance, encode_provenance};
16use rusqlite::{Connection, OptionalExtension, params};
17use sha2::{Digest, Sha256};
18
19const MAX_EMBEDDED_PROVENANCE_BYTES: usize = 1024 * 1024;
20
21/// Result returned by `kcode-kweb-manager`.
22pub type Result<T> = std::result::Result<T, Error>;
23
24/// Stable error category for application adapters.
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
26#[non_exhaustive]
27pub enum ErrorKind {
28    /// The caller supplied an invalid identifier or mutation.
29    InvalidInput,
30    /// A requested node, object, or provenance value does not exist.
31    NotFound,
32    /// Idempotency state or a concurrent database operation conflicts.
33    Conflict,
34    /// Persistence is unavailable, corrupt, or otherwise failed internally.
35    Internal,
36}
37
38/// Error returned by a Kweb manager operation.
39#[derive(Debug)]
40pub struct Error {
41    kind: ErrorKind,
42    message: String,
43}
44
45impl Error {
46    /// Returns the stable category suitable for transport mapping.
47    pub fn kind(&self) -> ErrorKind {
48        self.kind
49    }
50
51    fn invalid(message: impl Into<String>) -> Self {
52        Self {
53            kind: ErrorKind::InvalidInput,
54            message: message.into(),
55        }
56    }
57
58    fn conflict(message: impl Into<String>) -> Self {
59        Self {
60            kind: ErrorKind::Conflict,
61            message: message.into(),
62        }
63    }
64
65    fn internal(error: impl fmt::Display) -> Self {
66        Self {
67            kind: ErrorKind::Internal,
68            message: error.to_string(),
69        }
70    }
71}
72
73impl fmt::Display for Error {
74    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
75        formatter.write_str(&self.message)
76    }
77}
78
79impl std::error::Error for Error {}
80
81impl From<KwebError> for Error {
82    fn from(error: KwebError) -> Self {
83        let kind = match error {
84            KwebError::InvalidInput(_) | KwebError::InvalidTransaction(_) => {
85                ErrorKind::InvalidInput
86            }
87            KwebError::NotFound(_) => ErrorKind::NotFound,
88            KwebError::Busy(_) => ErrorKind::Conflict,
89            KwebError::Io(_)
90            | KwebError::Corrupt(_)
91            | KwebError::InvalidConfig(_)
92            | KwebError::OfflineUpgradeRequired(_) => ErrorKind::Internal,
93        };
94        Self {
95            kind,
96            message: error.to_string(),
97        }
98    }
99}
100
101impl From<kcode_commit_session::Error> for Error {
102    fn from(error: kcode_commit_session::Error) -> Self {
103        let kind = match error.kind() {
104            kcode_commit_session::ErrorKind::InvalidInput => ErrorKind::InvalidInput,
105            kcode_commit_session::ErrorKind::NotFound => ErrorKind::NotFound,
106            kcode_commit_session::ErrorKind::Conflict => ErrorKind::Conflict,
107            _ => ErrorKind::Internal,
108        };
109        Self {
110            kind,
111            message: error.to_string(),
112        }
113    }
114}
115
116/// One idempotent provenance-object creation.
117#[derive(Clone, Debug, Eq, PartialEq)]
118pub struct CreateProvenance {
119    /// Stable 16-byte request identity encoded as lowercase hexadecimal.
120    pub idempotency_id: String,
121    /// Application provenance envelope stored as the object payload.
122    pub value: StoredProvenance,
123    /// Kweb provenance describing the transaction that stores the envelope.
124    pub storage_provenance: Provenance,
125}
126
127/// Node fields editable by ordinary Kennedy Kweb mutations.
128///
129/// Object references are intentionally absent. New nodes start without
130/// objects, and updates preserve the current object list.
131#[derive(Clone, Debug, Eq, PartialEq)]
132pub struct NodeContents {
133    pub short_name: String,
134    pub short_description: String,
135    pub long_description: String,
136    pub owner: Owner,
137    pub fixed_connections: Vec<NodeId>,
138    pub recent_connections: Vec<NodeId>,
139}
140
141impl NodeContents {
142    fn into_data(self, objects: Vec<ObjectId>) -> kcode_kweb_db::NodeData {
143        kcode_kweb_db::NodeData {
144            short_name: self.short_name,
145            short_description: self.short_description,
146            long_description: self.long_description,
147            owner: self.owner,
148            fixed_connections: self.fixed_connections,
149            recent_connections: self.recent_connections,
150            objects,
151        }
152    }
153}
154
155/// Shared input for idempotent node creation and update.
156#[derive(Clone, Debug, Eq, PartialEq)]
157pub struct NodeWrite {
158    /// Stable 16-byte request identity encoded as lowercase hexadecimal.
159    pub idempotency_id: String,
160    /// Stored application provenance used to describe the Kweb transaction.
161    pub provenance_id: ObjectId,
162    /// Author or model attribution recorded on the Kweb transaction.
163    pub author: String,
164    /// Complete ordinary node fields.
165    pub contents: NodeContents,
166}
167
168/// Cloneable sole owner of one live Kweb database handle.
169#[derive(Clone)]
170pub struct KwebManager {
171    database: Arc<KwebDb>,
172    receipt_database: PathBuf,
173    receipts: Arc<Mutex<Connection>>,
174}
175
176impl KwebManager {
177    /// Takes ownership of an open Kweb database and opens its receipt lane.
178    pub fn open(database: KwebDb, receipt_database: impl AsRef<Path>) -> Result<Self> {
179        let receipt_database = receipt_database.as_ref().to_path_buf();
180        let receipts = Connection::open(&receipt_database).map_err(Error::internal)?;
181        receipts
182            .execute_batch(
183                "PRAGMA busy_timeout=15000;
184                 CREATE TABLE IF NOT EXISTS kmap_idempotency_receipts (
185                     idempotency_id TEXT PRIMARY KEY CHECK(length(idempotency_id)=32),
186                     operation TEXT NOT NULL,
187                     digest_version INTEGER NOT NULL DEFAULT 1,
188                     request_sha256 BLOB NOT NULL CHECK(length(request_sha256)=32),
189                     result_id TEXT CHECK(result_id IS NULL OR length(result_id)=8),
190                     started_at TEXT NOT NULL,
191                     committed_at TEXT,
192                     CHECK((result_id IS NULL) = (committed_at IS NULL))
193                 );
194                 DROP TABLE IF EXISTS kmap_object_provenance;",
195            )
196            .map_err(Error::internal)?;
197        ensure_digest_version_column(&receipts)?;
198        Ok(Self {
199            database: Arc::new(database),
200            receipt_database,
201            receipts: Arc::new(Mutex::new(receipts)),
202        })
203    }
204
205    /// Loads one current node.
206    pub fn get_node(&self, id: NodeId) -> Result<Node> {
207        self.database.get_node(id).map_err(Error::from)
208    }
209
210    /// Loads one node's complete visible history.
211    pub fn get_node_history(&self, id: NodeId) -> Result<NodeHistory> {
212        self.database.get_node_history(id).map_err(Error::from)
213    }
214
215    /// Loads the exact opaque bytes of one Kweb object.
216    pub fn get_object(&self, id: ObjectId) -> Result<Vec<u8>> {
217        self.database.get_object(id).map_err(Error::from)
218    }
219
220    /// Loads exact bytes and provenance from the signed creating Kweb transaction.
221    pub fn get_object_with_provenance(&self, id: ObjectId) -> Result<(Vec<u8>, Provenance)> {
222        self.database
223            .get_object_with_provenance(id)
224            .map_err(Error::from)
225    }
226
227    /// Idempotently stores one application provenance envelope.
228    pub fn create_provenance(&self, request: CreateProvenance) -> Result<ObjectId> {
229        validate_idempotency_id(&request.idempotency_id)?;
230        let encoded =
231            encode_provenance(&request.value).map_err(|error| Error::invalid(error.to_string()))?;
232        let legacy_digest = legacy_provenance_request_digest(&request);
233        let digest = provenance_request_digest(&encoded, &request.storage_provenance);
234        let storage_provenance = request.storage_provenance.clone();
235        let result = self.with_idempotency(
236            &request.idempotency_id,
237            "create_provenance",
238            VersionedDigest::v2(digest, legacy_digest),
239            |result_id| {
240                let id = ObjectId::from_str(result_id).map_err(|error| {
241                    Error::internal(format!("invalid stored provenance receipt: {error}"))
242                })?;
243                let (stored_bytes, creating_provenance) =
244                    self.database.get_object_with_provenance(id)?;
245                let stored_value = decode_provenance(&stored_bytes).map_err(|error| {
246                    Error::internal(format!("invalid stored provenance object {id}: {error}"))
247                })?;
248                Ok(stored_value == request.value
249                    && creating_provenance == request.storage_provenance)
250            },
251            || {
252                let mut transaction = self.database.start_transaction(storage_provenance)?;
253                let id = transaction.create_object(encoded)?;
254                transaction.finalize()?;
255                Ok(id.to_string())
256            },
257        )?;
258        let id = ObjectId::from_str(&result).map_err(|error| {
259            Error::internal(format!("invalid stored provenance receipt: {error}"))
260        })?;
261        Ok(id)
262    }
263
264    /// Idempotently creates a node without object attachments.
265    pub fn create_node(&self, request: NodeWrite) -> Result<Node> {
266        validate_idempotency_id(&request.idempotency_id)?;
267        let digest = node_request_digest("create_node", None, &request);
268        let result =
269            self.with_idempotency(
270                &request.idempotency_id,
271                "create_node",
272                VersionedDigest::v1(digest),
273                |_| Ok(true),
274                || {
275                    let provenance = self.load_provenance(request.provenance_id)?;
276                    let mut transaction = self.database.start_transaction(
277                        transaction_provenance(&provenance, request.provenance_id, request.author),
278                    )?;
279                    let id = transaction.create_node(request.contents.into_data(Vec::new()))?;
280                    transaction.finalize()?;
281                    Ok(id.to_string())
282                },
283            )?;
284        let id = NodeId::from_str(&result)
285            .map_err(|error| Error::internal(format!("invalid stored node receipt: {error}")))?;
286        self.get_node(id)
287    }
288
289    /// Idempotently updates ordinary node fields while retaining its objects.
290    pub fn update_node(&self, id: NodeId, request: NodeWrite) -> Result<Node> {
291        validate_idempotency_id(&request.idempotency_id)?;
292        let digest = node_request_digest("update_node", Some(id), &request);
293        self.with_idempotency(
294            &request.idempotency_id,
295            "update_node",
296            VersionedDigest::v1(digest),
297            |_| Ok(true),
298            || {
299                let provenance = self.load_provenance(request.provenance_id)?;
300                let objects = self.database.get_node(id)?.data.objects;
301                let mut transaction = self.database.start_transaction(transaction_provenance(
302                    &provenance,
303                    request.provenance_id,
304                    request.author,
305                ))?;
306                transaction.update_node(id, request.contents.into_data(objects))?;
307                transaction.finalize()?;
308                Ok(id.to_string())
309            },
310        )?;
311        self.get_node(id)
312    }
313
314    /// Stores exact opaque object bytes in one Kweb transaction.
315    pub fn store_object(&self, provenance: Provenance, bytes: Vec<u8>) -> Result<ObjectId> {
316        let mut transaction = self.database.start_transaction(provenance)?;
317        let id = transaction.create_object(bytes)?;
318        transaction.finalize()?;
319        Ok(id)
320    }
321
322    /// Atomically commits one complete session through `kcode-commit-session`.
323    pub fn commit_session(&self, request: CommitRequest) -> Result<CommitReceipt> {
324        let _receipt_lane = self
325            .receipts
326            .lock()
327            .map_err(|_| Error::internal("Kweb manager idempotency mutex is poisoned"))?;
328        kcode_commit_session::commit_session(&self.database, &self.receipt_database, request)
329            .map_err(Error::from)
330    }
331
332    fn load_provenance(&self, id: ObjectId) -> Result<StoredProvenance> {
333        let bytes = self.database.get_object(id)?;
334        decode_provenance(&bytes).map_err(|error| {
335            Error::internal(format!("invalid stored provenance object {id}: {error}"))
336        })
337    }
338
339    fn with_idempotency(
340        &self,
341        idempotency_id: &str,
342        operation: &'static str,
343        digest: VersionedDigest,
344        legacy_result_matches: impl FnOnce(&str) -> Result<bool>,
345        mutation: impl FnOnce() -> Result<String>,
346    ) -> Result<String> {
347        let receipts = self
348            .receipts
349            .lock()
350            .map_err(|_| Error::internal("Kweb manager idempotency mutex is poisoned"))?;
351        let existing = receipts
352            .query_row(
353                "SELECT operation,digest_version,request_sha256,result_id
354                 FROM kmap_idempotency_receipts WHERE idempotency_id=?1",
355                [idempotency_id],
356                |row| {
357                    Ok((
358                        row.get::<_, String>(0)?,
359                        row.get::<_, i64>(1)?,
360                        row.get::<_, Vec<u8>>(2)?,
361                        row.get::<_, Option<String>>(3)?,
362                    ))
363                },
364            )
365            .optional()
366            .map_err(Error::internal)?;
367        if let Some((stored_operation, stored_version, stored_hash, result_id)) = existing {
368            let (expected_hash, verify_legacy_result) = digest.for_version(stored_version)?;
369            if stored_operation != operation || stored_hash.as_slice() != expected_hash {
370                return Err(Error::conflict(
371                    "idempotency_id was already used for a different Kweb manager mutation",
372                ));
373            }
374            let result_id = result_id.ok_or_else(|| {
375                Error::conflict(
376                    "a prior Kweb manager mutation with this idempotency_id has an unknown outcome; offline recovery is required",
377                )
378            })?;
379            if verify_legacy_result && !legacy_result_matches(&result_id)? {
380                return Err(Error::conflict(
381                    "idempotency_id was already used for different provenance contents",
382                ));
383            }
384            return Ok(result_id);
385        }
386
387        receipts
388            .execute(
389                "INSERT INTO kmap_idempotency_receipts(
390                     idempotency_id,operation,digest_version,request_sha256,
391                     result_id,started_at,committed_at
392                 ) VALUES(?1,?2,?3,?4,NULL,?5,NULL)",
393                params![
394                    idempotency_id,
395                    operation,
396                    digest.current_version,
397                    digest.current.as_slice(),
398                    now_text(),
399                ],
400            )
401            .map_err(Error::internal)?;
402        let result_id = mutation()?;
403        let updated = receipts
404            .execute(
405                "UPDATE kmap_idempotency_receipts
406                 SET result_id=?2,committed_at=?3
407                 WHERE idempotency_id=?1 AND result_id IS NULL",
408                params![idempotency_id, &result_id, now_text()],
409            )
410            .map_err(Error::internal)?;
411        if updated != 1 {
412            return Err(Error::internal(
413                "Kweb manager idempotency receipt disappeared during mutation",
414            ));
415        }
416        Ok(result_id)
417    }
418}
419
420fn ensure_digest_version_column(receipts: &Connection) -> Result<()> {
421    let present = receipts
422        .query_row(
423            "SELECT COUNT(*) FROM pragma_table_info('kmap_idempotency_receipts')
424             WHERE name='digest_version'",
425            [],
426            |row| row.get::<_, i64>(0),
427        )
428        .map_err(Error::internal)?;
429    if present == 0 {
430        receipts
431            .execute(
432                "ALTER TABLE kmap_idempotency_receipts
433                 ADD COLUMN digest_version INTEGER NOT NULL DEFAULT 1",
434                [],
435            )
436            .map_err(Error::internal)?;
437    }
438    Ok(())
439}
440
441struct VersionedDigest {
442    current_version: i64,
443    current: [u8; 32],
444    legacy: Option<[u8; 32]>,
445}
446
447impl VersionedDigest {
448    fn v1(current: [u8; 32]) -> Self {
449        Self {
450            current_version: 1,
451            current,
452            legacy: None,
453        }
454    }
455
456    fn v2(current: [u8; 32], legacy: [u8; 32]) -> Self {
457        Self {
458            current_version: 2,
459            current,
460            legacy: Some(legacy),
461        }
462    }
463
464    fn for_version(&self, version: i64) -> Result<(&[u8; 32], bool)> {
465        if version == self.current_version {
466            return Ok((&self.current, false));
467        }
468        if version == 1
469            && let Some(legacy) = &self.legacy
470        {
471            return Ok((legacy, true));
472        }
473        Err(Error::internal(format!(
474            "unsupported Kweb manager idempotency digest version {version}"
475        )))
476    }
477}
478
479fn now_text() -> String {
480    Utc::now().to_rfc3339()
481}
482
483fn validate_idempotency_id(value: &str) -> Result<()> {
484    if value.len() != 32
485        || !value
486            .bytes()
487            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
488    {
489        return Err(Error::invalid(
490            "idempotency_id must encode 16 bytes as lowercase hexadecimal",
491        ));
492    }
493    Ok(())
494}
495
496struct RequestDigest(Sha256);
497
498impl RequestDigest {
499    fn new(operation: &str) -> Self {
500        let mut hash = Sha256::new();
501        hash.update(b"kennedy kmap idempotency v1\0");
502        let mut value = Self(hash);
503        value.field(operation.as_bytes());
504        value
505    }
506
507    fn new_v2(operation: &str) -> Self {
508        let mut hash = Sha256::new();
509        hash.update(b"kennedy kweb manager idempotency v2\0");
510        let mut value = Self(hash);
511        value.field(operation.as_bytes());
512        value
513    }
514
515    fn field(&mut self, bytes: &[u8]) {
516        self.0.update((bytes.len() as u64).to_be_bytes());
517        self.0.update(bytes);
518    }
519
520    fn finish(self) -> [u8; 32] {
521        self.0.finalize().into()
522    }
523}
524
525fn provenance_request_digest(encoded: &[u8], storage: &Provenance) -> [u8; 32] {
526    let mut hash = RequestDigest::new_v2("create_provenance");
527    hash.field(encoded);
528    hash.field(storage.author.as_bytes());
529    hash.field(storage.source.as_bytes());
530    hash.field(&storage.source_created_at.timestamp().to_be_bytes());
531    hash.field(
532        &storage
533            .source_created_at
534            .timestamp_subsec_nanos()
535            .to_be_bytes(),
536    );
537    hash.field(storage.data.as_bytes());
538    hash.finish()
539}
540
541// Existing receipts used this incomplete digest. It remains only to replay an
542// exact legacy result after independently checking the stored Kweb object and
543// its signed creating provenance against the complete request.
544fn legacy_provenance_request_digest(request: &CreateProvenance) -> [u8; 32] {
545    let mut hash = RequestDigest::new("create_provenance");
546    hash.field(request.value.data.as_bytes());
547    hash.field(request.value.source.as_bytes());
548    hash.field(&request.value.source_created_at.timestamp().to_be_bytes());
549    hash.field(
550        &request
551            .value
552            .source_created_at
553            .timestamp_subsec_nanos()
554            .to_be_bytes(),
555    );
556    hash.field(b"");
557    hash.field(&(request.value.artifacts.len() as u64).to_be_bytes());
558    for artifact in &request.value.artifacts {
559        hash.field(artifact.original_filename.as_bytes());
560        hash.field(artifact.media_type.as_bytes());
561        hash.field(&artifact.sha256);
562    }
563    hash.field(request.storage_provenance.author.as_bytes());
564    hash.field(request.storage_provenance.source.as_bytes());
565    hash.field(
566        &request
567            .storage_provenance
568            .source_created_at
569            .timestamp()
570            .to_be_bytes(),
571    );
572    hash.field(
573        &request
574            .storage_provenance
575            .source_created_at
576            .timestamp_subsec_nanos()
577            .to_be_bytes(),
578    );
579    hash.field(request.storage_provenance.data.as_bytes());
580    hash.finish()
581}
582
583fn node_request_digest(operation: &str, id: Option<NodeId>, request: &NodeWrite) -> [u8; 32] {
584    let mut hash = RequestDigest::new(operation);
585    hash.field(&id.map(NodeId::to_bytes).unwrap_or([0; 6]));
586    hash.field(&request.provenance_id.to_bytes());
587    hash.field(request.author.as_bytes());
588    hash.field(request.contents.short_name.as_bytes());
589    hash.field(request.contents.short_description.as_bytes());
590    hash.field(request.contents.long_description.as_bytes());
591    match request.contents.owner {
592        Owner::Unowned => hash.field(&[0]),
593        Owner::SelfNode => hash.field(&[1]),
594        Owner::Node(owner) => {
595            hash.field(&[2]);
596            hash.field(&owner.to_bytes());
597        }
598    }
599    hash.field(&(request.contents.fixed_connections.len() as u64).to_be_bytes());
600    for connection in &request.contents.fixed_connections {
601        hash.field(&connection.to_bytes());
602    }
603    hash.field(&(request.contents.recent_connections.len() as u64).to_be_bytes());
604    for connection in &request.contents.recent_connections {
605        hash.field(&connection.to_bytes());
606    }
607    hash.finish()
608}
609
610fn transaction_provenance(
611    stored: &StoredProvenance,
612    object_id: ObjectId,
613    author: String,
614) -> Provenance {
615    let data = if stored.data.len() <= MAX_EMBEDDED_PROVENANCE_BYTES {
616        stored.data.clone()
617    } else {
618        format!("Kennedy provenance is stored in object {object_id}.")
619    };
620    Provenance {
621        author,
622        source: stored.source.clone(),
623        source_created_at: stored.source_created_at,
624        data,
625    }
626}
627
628#[cfg(test)]
629mod tests {
630    use std::{
631        collections::BTreeMap,
632        fs,
633        sync::{
634            Arc,
635            atomic::{AtomicU64, Ordering},
636        },
637    };
638
639    use chrono::{Duration, TimeZone, Utc};
640    use kcode_commit_session::CommitRequest;
641    use kcode_kweb_db::{Config, NodeData, NoopGossip, WriterId};
642    use kcode_server_object_envelopes::{StoredArtifact, decode_provenance};
643
644    use super::*;
645
646    static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1);
647
648    struct TestDirectory(PathBuf);
649
650    impl TestDirectory {
651        fn new() -> Self {
652            let sequence = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed);
653            let path = std::env::temp_dir().join(format!(
654                "kcode-kweb-manager-test-{}-{sequence}",
655                std::process::id()
656            ));
657            fs::create_dir_all(&path).unwrap();
658            Self(path)
659        }
660
661        fn join(&self, value: &str) -> PathBuf {
662            self.0.join(value)
663        }
664    }
665
666    impl Drop for TestDirectory {
667        fn drop(&mut self) {
668            fs::remove_dir_all(&self.0).unwrap();
669        }
670    }
671
672    fn config() -> Config {
673        let signing_key = [7; 32];
674        Config {
675            signing_key,
676            writers_by_priority: vec![WriterId::from_signing_key(&signing_key)],
677            gossip: Arc::new(NoopGossip),
678        }
679    }
680
681    fn timestamp() -> chrono::DateTime<Utc> {
682        Utc.with_ymd_and_hms(2026, 7, 28, 12, 0, 0).unwrap()
683    }
684
685    fn transaction_provenance_for(label: &str) -> Provenance {
686        Provenance {
687            author: "test".into(),
688            source: label.into(),
689            source_created_at: timestamp(),
690            data: format!("{label} transaction"),
691        }
692    }
693
694    fn provenance_request(idempotency_id: &str, data: &str) -> CreateProvenance {
695        CreateProvenance {
696            idempotency_id: idempotency_id.into(),
697            value: StoredProvenance {
698                data: data.into(),
699                source: "test-source".into(),
700                source_created_at: timestamp(),
701                artifacts: Vec::new(),
702            },
703            storage_provenance: transaction_provenance_for("provenance-storage"),
704        }
705    }
706
707    fn artifact() -> StoredArtifact {
708        StoredArtifact {
709            object_id: ObjectId::from_bytes([0x80, 1, 2, 3, 4, 5]).unwrap(),
710            original_filename: "source.txt".into(),
711            media_type: "text/plain".into(),
712            role: "source".into(),
713            byte_length: 12,
714            sha256: [7; 32],
715        }
716    }
717
718    fn node_write(idempotency_id: &str, provenance_id: ObjectId, short_name: &str) -> NodeWrite {
719        NodeWrite {
720            idempotency_id: idempotency_id.into(),
721            provenance_id,
722            author: "test-model".into(),
723            contents: NodeContents {
724                short_name: short_name.into(),
725                short_description: "short".into(),
726                long_description: "long".into(),
727                owner: Owner::SelfNode,
728                fixed_connections: Vec::new(),
729                recent_connections: Vec::new(),
730            },
731        }
732    }
733
734    #[test]
735    fn provenance_and_node_mutations_are_idempotent_and_typed() {
736        let directory = TestDirectory::new();
737        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
738        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
739
740        let create_request =
741            provenance_request("00000000000000000000000000000001", "source material");
742        let provenance_id = kmap.create_provenance(create_request.clone()).unwrap();
743        assert_eq!(
744            kmap.create_provenance(create_request).unwrap(),
745            provenance_id
746        );
747        let stored = decode_provenance(&kmap.get_object(provenance_id).unwrap()).unwrap();
748        assert_eq!(stored.data, "source material");
749
750        let conflict = kmap
751            .create_provenance(provenance_request(
752                "00000000000000000000000000000001",
753                "different material",
754            ))
755            .unwrap_err();
756        assert_eq!(conflict.kind(), ErrorKind::Conflict);
757
758        let write = node_write(
759            "00000000000000000000000000000002",
760            provenance_id,
761            "Created node",
762        );
763        let node = kmap.create_node(write.clone()).unwrap();
764        assert_eq!(kmap.create_node(write).unwrap(), node);
765        assert_eq!(kmap.get_node(node.id).unwrap(), node);
766        assert!(node.data.objects.is_empty());
767    }
768
769    #[test]
770    fn provenance_idempotency_includes_every_storage_provenance_field() {
771        let directory = TestDirectory::new();
772        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
773        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
774
775        let request = provenance_request("00000000000000000000000000000005", "source material");
776        let provenance_id = kmap.create_provenance(request.clone()).unwrap();
777        assert_eq!(
778            kmap.create_provenance(request.clone()).unwrap(),
779            provenance_id
780        );
781
782        let mut changed_author = request.clone();
783        changed_author
784            .storage_provenance
785            .author
786            .push_str("-changed");
787        let mut changed_source = request.clone();
788        changed_source
789            .storage_provenance
790            .source
791            .push_str("-changed");
792        let mut changed_timestamp = request.clone();
793        changed_timestamp.storage_provenance.source_created_at += Duration::nanoseconds(1);
794        let mut changed_data = request.clone();
795        changed_data.storage_provenance.data.push_str("-changed");
796
797        for changed in [
798            changed_author,
799            changed_source,
800            changed_timestamp,
801            changed_data,
802        ] {
803            let conflict = kmap.create_provenance(changed).unwrap_err();
804            assert_eq!(conflict.kind(), ErrorKind::Conflict);
805        }
806
807        let (bytes, creating_provenance) = kmap.get_object_with_provenance(provenance_id).unwrap();
808        assert_eq!(decode_provenance(&bytes).unwrap(), request.value);
809        assert_eq!(creating_provenance, request.storage_provenance);
810    }
811
812    #[test]
813    fn provenance_idempotency_includes_every_artifact_field() {
814        let directory = TestDirectory::new();
815        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
816        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
817        let mut request = provenance_request("00000000000000000000000000000006", "source material");
818        request.value.artifacts.push(artifact());
819        let provenance_id = kmap.create_provenance(request.clone()).unwrap();
820        assert_eq!(
821            kmap.create_provenance(request.clone()).unwrap(),
822            provenance_id
823        );
824
825        let mut changed_object = request.clone();
826        changed_object.value.artifacts[0].object_id =
827            ObjectId::from_bytes([0x80, 1, 2, 3, 4, 6]).unwrap();
828        let mut changed_filename = request.clone();
829        changed_filename.value.artifacts[0].original_filename = "other.txt".into();
830        let mut changed_media_type = request.clone();
831        changed_media_type.value.artifacts[0].media_type = "application/json".into();
832        let mut changed_role = request.clone();
833        changed_role.value.artifacts[0].role = "transcript".into();
834        let mut changed_size = request.clone();
835        changed_size.value.artifacts[0].byte_length += 1;
836        let mut changed_sha256 = request;
837        changed_sha256.value.artifacts[0].sha256[0] ^= 1;
838
839        for changed in [
840            changed_object,
841            changed_filename,
842            changed_media_type,
843            changed_role,
844            changed_size,
845            changed_sha256,
846        ] {
847            let conflict = kmap.create_provenance(changed).unwrap_err();
848            assert_eq!(conflict.kind(), ErrorKind::Conflict);
849        }
850    }
851
852    #[test]
853    fn legacy_provenance_receipts_replay_exactly_and_drop_duplicate_storage() {
854        let directory = TestDirectory::new();
855        let receipt_path = directory.join("application.sqlite3");
856        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
857        let mut request = provenance_request("00000000000000000000000000000007", "legacy source");
858        request.value.artifacts.push(artifact());
859        let encoded = encode_provenance(&request.value).unwrap();
860        let mut transaction = database
861            .start_transaction(request.storage_provenance.clone())
862            .unwrap();
863        let object_id = transaction.create_object(encoded).unwrap();
864        transaction.finalize().unwrap();
865
866        let receipts = Connection::open(&receipt_path).unwrap();
867        receipts
868            .execute_batch(
869                "CREATE TABLE kmap_idempotency_receipts (
870                     idempotency_id TEXT PRIMARY KEY CHECK(length(idempotency_id)=32),
871                     operation TEXT NOT NULL,
872                     request_sha256 BLOB NOT NULL CHECK(length(request_sha256)=32),
873                     result_id TEXT CHECK(result_id IS NULL OR length(result_id)=8),
874                     started_at TEXT NOT NULL,
875                     committed_at TEXT,
876                     CHECK((result_id IS NULL) = (committed_at IS NULL))
877                 );
878                 CREATE TABLE kmap_object_provenance(object_id TEXT PRIMARY KEY);",
879            )
880            .unwrap();
881        receipts
882            .execute(
883                "INSERT INTO kmap_idempotency_receipts(
884                     idempotency_id,operation,request_sha256,result_id,started_at,committed_at
885                 ) VALUES(?1,'create_provenance',?2,?3,?4,?4)",
886                params![
887                    &request.idempotency_id,
888                    legacy_provenance_request_digest(&request).as_slice(),
889                    object_id.to_string(),
890                    now_text(),
891                ],
892            )
893            .unwrap();
894        drop(receipts);
895
896        let kmap = KwebManager::open(database, &receipt_path).unwrap();
897        assert_eq!(kmap.create_provenance(request.clone()).unwrap(), object_id);
898        let mut changed_role = request;
899        changed_role.value.artifacts[0].role = "different".into();
900        let conflict = kmap.create_provenance(changed_role).unwrap_err();
901        assert_eq!(conflict.kind(), ErrorKind::Conflict);
902
903        let receipts = Connection::open(receipt_path).unwrap();
904        let duplicate_table_count = receipts
905            .query_row(
906                "SELECT COUNT(*) FROM sqlite_master
907                 WHERE type='table' AND name='kmap_object_provenance'",
908                [],
909                |row| row.get::<_, i64>(0),
910            )
911            .unwrap();
912        assert_eq!(duplicate_table_count, 0);
913        let digest_version = receipts
914            .query_row(
915                "SELECT digest_version FROM kmap_idempotency_receipts
916                 WHERE idempotency_id=?1",
917                [&"00000000000000000000000000000007"],
918                |row| row.get::<_, i64>(0),
919            )
920            .unwrap();
921        assert_eq!(digest_version, 1);
922    }
923
924    #[test]
925    fn ordinary_updates_preserve_existing_object_attachments() {
926        let directory = TestDirectory::new();
927        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
928        let mut transaction = database
929            .start_transaction(transaction_provenance_for("seed"))
930            .unwrap();
931        let object_id = transaction.create_object(b"attachment".to_vec()).unwrap();
932        let node_id = transaction
933            .create_node(NodeData {
934                short_name: "Before".into(),
935                short_description: String::new(),
936                long_description: String::new(),
937                owner: Owner::SelfNode,
938                fixed_connections: Vec::new(),
939                recent_connections: Vec::new(),
940                objects: vec![object_id],
941            })
942            .unwrap();
943        transaction.finalize().unwrap();
944
945        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
946        let (stored_bytes, stored_provenance) = kmap.get_object_with_provenance(object_id).unwrap();
947        assert_eq!(stored_bytes, b"attachment");
948        assert_eq!(stored_provenance, transaction_provenance_for("seed"));
949
950        let provenance_id = kmap
951            .create_provenance(provenance_request(
952                "00000000000000000000000000000003",
953                "update source",
954            ))
955            .unwrap();
956        let updated = kmap
957            .update_node(
958                node_id,
959                node_write("00000000000000000000000000000004", provenance_id, "After"),
960            )
961            .unwrap();
962        assert_eq!(updated.data.short_name, "After");
963        assert_eq!(updated.data.objects, vec![object_id]);
964    }
965
966    #[test]
967    fn object_storage_and_session_commits_share_the_owned_database() {
968        let directory = TestDirectory::new();
969        let database = KwebDb::open(directory.join("kweb"), config()).unwrap();
970        let kmap = KwebManager::open(database, directory.join("application.sqlite3")).unwrap();
971
972        let opaque_provenance = transaction_provenance_for("opaque-object");
973        let object_id = kmap
974            .store_object(opaque_provenance.clone(), b"opaque bytes".to_vec())
975            .unwrap();
976        assert_eq!(kmap.get_object(object_id).unwrap(), b"opaque bytes");
977        let (opaque_bytes, retained_provenance) =
978            kmap.get_object_with_provenance(object_id).unwrap();
979        assert_eq!(opaque_bytes, b"opaque bytes");
980        assert_eq!(retained_provenance, opaque_provenance);
981
982        let request = CommitRequest {
983            idempotency_key: "session-test".into(),
984            author: "test-model".into(),
985            source_created_at: timestamp(),
986            archive: b"{\"events\":[]}".to_vec(),
987            objects: BTreeMap::new(),
988            creates: BTreeMap::new(),
989            updates: BTreeMap::new(),
990        };
991        let first = kmap.commit_session(request.clone()).unwrap();
992        assert_eq!(
993            kmap.get_object(first.session_object_id).unwrap(),
994            b"{\"events\":[]}"
995        );
996        let (archive, provenance) = kmap
997            .get_object_with_provenance(first.session_object_id)
998            .unwrap();
999        assert_eq!(archive, b"{\"events\":[]}");
1000        assert_eq!(provenance.author, "test-model");
1001        assert_eq!(provenance.source, "kennedy-session");
1002        assert_eq!(kmap.commit_session(request).unwrap(), first);
1003    }
1004}
kcode_kweb_manager/lib.rs

kcode_kweb_manager/
lib.rs
