kcode_commit_session/
lib.rs

1#![forbid(unsafe_code)]
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    fmt,
6    path::Path,
7    str::FromStr,
8};
9
10use chrono::{DateTime, Utc};
11use kcode_kweb_db::{
12    Error as KwebError, KwebDb, NodeData, NodeId, ObjectId, Owner, Provenance, TransactionId,
13};
14use rusqlite::{Connection, OptionalExtension, params};
15use serde::{Deserialize, Deserializer, Serialize, Serializer};
16use sha2::{Digest, Sha256};
17
18const DIGEST_VERSION: i64 = 2;
19const RECEIPT_TABLE: &str = "kmap_session_commit_receipts";
20
21/// Everything needed to materialize one sealed session.
22///
23/// Object payloads must already be in the permanent format expected by their
24/// readers. Keys in `objects` and `creates` are canonical `pending:N` IDs.
25#[derive(Clone, Debug)]
26pub struct CommitRequest {
27    pub idempotency_key: String,
28    pub author: String,
29    pub source_created_at: DateTime<Utc>,
30    pub archive: Vec<u8>,
31    pub objects: BTreeMap<String, Vec<u8>>,
32    pub creates: BTreeMap<String, PlannedNode>,
33    pub updates: BTreeMap<NodeId, PlannedNode>,
34}
35
36/// A node whose pending references will be resolved during commit.
37#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
38#[serde(rename_all = "camelCase")]
39pub struct PlannedNode {
40    pub short_name: String,
41    pub short_description: String,
42    pub long_description: String,
43    pub owner: String,
44    #[serde(default)]
45    pub fixed_connections: Vec<String>,
46    #[serde(default)]
47    pub recent_connections: Vec<String>,
48    #[serde(default)]
49    pub objects: Vec<String>,
50    /// The serialized name preserves existing Kennedy checkpoints.
51    #[serde(
52        default,
53        rename = "includeSessionObject",
54        alias = "attachSessionArchive"
55    )]
56    pub attach_session_archive: bool,
57}
58
59/// The stable identifiers allocated for a committed session.
60#[derive(Clone, Debug, Eq, PartialEq)]
61pub struct CommitReceipt {
62    pub transaction_id: Option<TransactionId>,
63    pub session_object_id: ObjectId,
64    pub node_ids: BTreeMap<String, NodeId>,
65    pub object_ids: BTreeMap<String, ObjectId>,
66}
67
68#[derive(Serialize, Deserialize)]
69#[serde(rename_all = "camelCase")]
70struct WireReceipt {
71    transaction_id: Option<String>,
72    session_object_id: String,
73    node_ids: BTreeMap<String, String>,
74    object_ids: BTreeMap<String, String>,
75}
76
77impl Serialize for CommitReceipt {
78    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
79    where
80        S: Serializer,
81    {
82        WireReceipt {
83            transaction_id: self.transaction_id.map(|id| id.to_string()),
84            session_object_id: self.session_object_id.to_string(),
85            node_ids: self
86                .node_ids
87                .iter()
88                .map(|(pending, id)| (pending.clone(), id.to_string()))
89                .collect(),
90            object_ids: self
91                .object_ids
92                .iter()
93                .map(|(pending, id)| (pending.clone(), id.to_string()))
94                .collect(),
95        }
96        .serialize(serializer)
97    }
98}
99
100impl<'de> Deserialize<'de> for CommitReceipt {
101    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
102    where
103        D: Deserializer<'de>,
104    {
105        let wire = WireReceipt::deserialize(deserializer)?;
106        Ok(Self {
107            transaction_id: wire
108                .transaction_id
109                .map(|value| TransactionId::from_str(&value).map_err(serde::de::Error::custom))
110                .transpose()?,
111            session_object_id: ObjectId::from_str(&wire.session_object_id)
112                .map_err(serde::de::Error::custom)?,
113            node_ids: parse_id_map(wire.node_ids)?,
114            object_ids: parse_id_map(wire.object_ids)?,
115        })
116    }
117}
118
119fn parse_id_map<T, E>(values: BTreeMap<String, String>) -> Result<BTreeMap<String, T>, E>
120where
121    T: FromStr,
122    T::Err: fmt::Display,
123    E: serde::de::Error,
124{
125    values
126        .into_iter()
127        .map(|(key, value)| T::from_str(&value).map(|id| (key, id)).map_err(E::custom))
128        .collect()
129}
130
131#[derive(Clone, Copy, Debug, Eq, PartialEq)]
132#[non_exhaustive]
133pub enum ErrorKind {
134    InvalidInput,
135    NotFound,
136    Conflict,
137    Internal,
138}
139
140#[derive(Debug)]
141pub struct Error {
142    kind: ErrorKind,
143    message: String,
144}
145
146impl Error {
147    pub fn kind(&self) -> ErrorKind {
148        self.kind
149    }
150
151    fn invalid(message: impl Into<String>) -> Self {
152        Self {
153            kind: ErrorKind::InvalidInput,
154            message: message.into(),
155        }
156    }
157
158    fn conflict(message: impl Into<String>) -> Self {
159        Self {
160            kind: ErrorKind::Conflict,
161            message: message.into(),
162        }
163    }
164
165    fn internal(error: impl fmt::Display) -> Self {
166        Self {
167            kind: ErrorKind::Internal,
168            message: error.to_string(),
169        }
170    }
171}
172
173impl fmt::Display for Error {
174    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
175        formatter.write_str(&self.message)
176    }
177}
178
179impl std::error::Error for Error {}
180
181impl From<KwebError> for Error {
182    fn from(error: KwebError) -> Self {
183        let kind = match error {
184            KwebError::InvalidInput(_) | KwebError::InvalidTransaction(_) => {
185                ErrorKind::InvalidInput
186            }
187            KwebError::NotFound(_) => ErrorKind::NotFound,
188            _ => ErrorKind::Internal,
189        };
190        Self {
191            kind,
192            message: error.to_string(),
193        }
194    }
195}
196
197/// Atomically materializes a sealed session into one Kweb transaction.
198///
199/// Repeating an identical request returns its receipt. Reusing the
200/// idempotency key for a different request returns `ErrorKind::Conflict`.
201/// The caller must serialize this operation with other writers of the receipt
202/// database; Kweb serializes its own transaction writer internally.
203pub fn commit_session(
204    database: &KwebDb,
205    receipt_database: &Path,
206    request: CommitRequest,
207) -> Result<CommitReceipt, Error> {
208    validate(&request)?;
209    let request_digest = request_digest(&request);
210    let receipts = open_receipts(receipt_database)?;
211    if let Some(receipt) = recover_or_replay(database, &receipts, &request, request_digest)? {
212        return Ok(receipt);
213    }
214
215    receipts
216        .execute(
217            "INSERT INTO kmap_session_commit_receipts(
218                 session_id,request_sha256,digest_version,prepared_json,result_json,
219                 started_at,committed_at
220             ) VALUES(?1,?2,?3,NULL,NULL,?4,NULL)",
221            params![
222                &request.idempotency_key,
223                request_digest.as_slice(),
224                DIGEST_VERSION,
225                Utc::now().to_rfc3339(),
226            ],
227        )
228        .map_err(Error::internal)?;
229
230    let mut transaction = database.start_transaction(Provenance {
231        author: request.author.clone(),
232        source: "kennedy-session".into(),
233        source_created_at: request.source_created_at,
234        data: format!("Kennedy session {}.", request.idempotency_key),
235    })?;
236
237    let mut object_ids = BTreeMap::new();
238    for (pending, payload) in request.objects {
239        object_ids.insert(pending, transaction.create_object(payload)?);
240    }
241    let archive = replace_pending_object_tokens(&request.archive, &object_ids);
242    let session_object_id = transaction.create_object(archive)?;
243
244    let mut node_ids = BTreeMap::new();
245    for pending in request.creates.keys() {
246        node_ids.insert(pending.clone(), transaction.reserve_node_id()?);
247    }
248    for (pending, data) in request.creates {
249        let resolved = resolve_node(data, &node_ids, &object_ids, session_object_id)?;
250        transaction.create_reserved_node(node_ids[&pending], resolved)?;
251    }
252    for (id, data) in request.updates {
253        let resolved = resolve_node(data, &node_ids, &object_ids, session_object_id)?;
254        transaction.update_node(id, resolved)?;
255    }
256
257    let mut receipt = CommitReceipt {
258        transaction_id: None,
259        session_object_id,
260        node_ids,
261        object_ids,
262    };
263    let prepared_json = serde_json::to_string(&receipt).map_err(Error::internal)?;
264    require_one(
265        receipts
266            .execute(
267                "UPDATE kmap_session_commit_receipts
268                 SET prepared_json=?2
269                 WHERE session_id=?1 AND prepared_json IS NULL AND result_json IS NULL",
270                params![&request.idempotency_key, prepared_json],
271            )
272            .map_err(Error::internal)?,
273        "session commit preparation receipt disappeared",
274    )?;
275
276    receipt.transaction_id = Some(transaction.finalize()?);
277    let result_json = serde_json::to_string(&receipt).map_err(Error::internal)?;
278    require_one(
279        receipts
280            .execute(
281                "UPDATE kmap_session_commit_receipts
282                 SET result_json=?2,committed_at=?3
283                 WHERE session_id=?1 AND result_json IS NULL",
284                params![
285                    &request.idempotency_key,
286                    result_json,
287                    Utc::now().to_rfc3339(),
288                ],
289            )
290            .map_err(Error::internal)?,
291        "session commit receipt disappeared during mutation",
292    )?;
293    Ok(receipt)
294}
295
296fn open_receipts(path: &Path) -> Result<Connection, Error> {
297    let connection = Connection::open(path).map_err(Error::internal)?;
298    connection
299        .execute_batch(
300            "PRAGMA foreign_keys=ON;
301             PRAGMA journal_mode=WAL;
302             PRAGMA busy_timeout=15000;
303             CREATE TABLE IF NOT EXISTS kmap_session_commit_receipts (
304                 session_id TEXT PRIMARY KEY,
305                 request_sha256 BLOB NOT NULL CHECK(length(request_sha256)=32),
306                 digest_version INTEGER NOT NULL DEFAULT 2,
307                 prepared_json TEXT,
308                 result_json TEXT,
309                 started_at TEXT NOT NULL,
310                 committed_at TEXT,
311                 CHECK((result_json IS NULL) = (committed_at IS NULL))
312             );",
313        )
314        .map_err(Error::internal)?;
315    let columns = {
316        let mut statement = connection
317            .prepare(&format!("PRAGMA table_info({RECEIPT_TABLE})"))
318            .map_err(Error::internal)?;
319        statement
320            .query_map([], |row| row.get::<_, String>(1))
321            .map_err(Error::internal)?
322            .collect::<rusqlite::Result<BTreeSet<_>>>()
323            .map_err(Error::internal)?
324    };
325    if !columns.contains("prepared_json") {
326        connection
327            .execute(
328                "ALTER TABLE kmap_session_commit_receipts ADD COLUMN prepared_json TEXT",
329                [],
330            )
331            .map_err(Error::internal)?;
332    }
333    if !columns.contains("digest_version") {
334        connection
335            .execute(
336                "ALTER TABLE kmap_session_commit_receipts
337                 ADD COLUMN digest_version INTEGER NOT NULL DEFAULT 1",
338                [],
339            )
340            .map_err(Error::internal)?;
341    }
342    Ok(connection)
343}
344
345fn recover_or_replay(
346    database: &KwebDb,
347    receipts: &Connection,
348    request: &CommitRequest,
349    request_digest: [u8; 32],
350) -> Result<Option<CommitReceipt>, Error> {
351    let existing = receipts
352        .query_row(
353            "SELECT request_sha256,digest_version,prepared_json,result_json
354             FROM kmap_session_commit_receipts WHERE session_id=?1",
355            [&request.idempotency_key],
356            |row| {
357                Ok((
358                    row.get::<_, Vec<u8>>(0)?,
359                    row.get::<_, i64>(1)?,
360                    row.get::<_, Option<String>>(2)?,
361                    row.get::<_, Option<String>>(3)?,
362                ))
363            },
364        )
365        .optional()
366        .map_err(Error::internal)?;
367    let Some((stored_digest, digest_version, prepared, result)) = existing else {
368        return Ok(None);
369    };
370    // Version 1 was Kennedy's serde-derived digest. Its private request types
371    // no longer exist here, so the already-unique key is the compatibility
372    // authority for those historical receipts.
373    if digest_version == DIGEST_VERSION && stored_digest.as_slice() != request_digest {
374        return Err(Error::conflict(
375            "idempotency key was already used for a different session commit",
376        ));
377    }
378    if let Some(result) = result {
379        return serde_json::from_str(&result)
380            .map(Some)
381            .map_err(Error::internal);
382    }
383    if let Some(prepared) = prepared {
384        let recovered: CommitReceipt = serde_json::from_str(&prepared).map_err(Error::internal)?;
385        match database.get_object(recovered.session_object_id) {
386            Ok(_) => {
387                require_one(
388                    receipts
389                        .execute(
390                            "UPDATE kmap_session_commit_receipts
391                             SET result_json=prepared_json,committed_at=?2
392                             WHERE session_id=?1 AND result_json IS NULL",
393                            params![&request.idempotency_key, Utc::now().to_rfc3339()],
394                        )
395                        .map_err(Error::internal)?,
396                    "session recovery receipt disappeared",
397                )?;
398                return Ok(Some(recovered));
399            }
400            Err(KwebError::NotFound(_)) => {}
401            Err(error) => return Err(error.into()),
402        }
403    }
404    receipts
405        .execute(
406            "DELETE FROM kmap_session_commit_receipts
407             WHERE session_id=?1 AND result_json IS NULL",
408            [&request.idempotency_key],
409        )
410        .map_err(Error::internal)?;
411    Ok(None)
412}
413
414fn validate(request: &CommitRequest) -> Result<(), Error> {
415    if request.idempotency_key.trim().is_empty() || request.idempotency_key.len() > 1024 {
416        return Err(Error::invalid(
417            "idempotency key must contain between 1 and 1024 bytes",
418        ));
419    }
420    let mut pending_ids = BTreeSet::new();
421    for pending in request.objects.keys().chain(request.creates.keys()) {
422        validate_pending_id(pending)?;
423        if !pending_ids.insert(pending) {
424            return Err(Error::invalid(format!(
425                "pending ID {pending} is used for both an object and a node"
426            )));
427        }
428    }
429    for node in request.creates.values().chain(request.updates.values()) {
430        validate_node(node, &request.creates, &request.objects)?;
431    }
432    Ok(())
433}
434
435fn validate_pending_id(value: &str) -> Result<(), Error> {
436    let number = value
437        .strip_prefix("pending:")
438        .and_then(|number| number.parse::<u64>().ok())
439        .filter(|number| *number > 0);
440    if number.is_none() || format!("pending:{}", number.unwrap_or_default()) != value {
441        return Err(Error::invalid(format!(
442            "{value:?} is not a canonical pending ID"
443        )));
444    }
445    Ok(())
446}
447
448fn validate_node(
449    node: &PlannedNode,
450    creates: &BTreeMap<String, PlannedNode>,
451    objects: &BTreeMap<String, Vec<u8>>,
452) -> Result<(), Error> {
453    if !matches!(node.owner.as_str(), "self" | "unowned") {
454        validate_node_ref(&node.owner, creates)?;
455    }
456    for value in node
457        .fixed_connections
458        .iter()
459        .chain(&node.recent_connections)
460    {
461        validate_node_ref(value, creates)?;
462    }
463    for value in &node.objects {
464        if value.starts_with("pending:") {
465            validate_pending_id(value)?;
466            if !objects.contains_key(value) {
467                return Err(Error::invalid(format!("unresolved pending object {value}")));
468            }
469        } else {
470            ObjectId::from_str(value).map_err(Error::from)?;
471        }
472    }
473    Ok(())
474}
475
476fn validate_node_ref(value: &str, creates: &BTreeMap<String, PlannedNode>) -> Result<(), Error> {
477    if value.starts_with("pending:") {
478        validate_pending_id(value)?;
479        if !creates.contains_key(value) {
480            return Err(Error::invalid(format!("unresolved pending node {value}")));
481        }
482        Ok(())
483    } else {
484        NodeId::from_str(value).map(|_| ()).map_err(Error::from)
485    }
486}
487
488fn resolve_node(
489    node: PlannedNode,
490    node_ids: &BTreeMap<String, NodeId>,
491    object_ids: &BTreeMap<String, ObjectId>,
492    session_object_id: ObjectId,
493) -> Result<NodeData, Error> {
494    let resolve_node_id = |value: &str| {
495        if value.starts_with("pending:") {
496            node_ids
497                .get(value)
498                .copied()
499                .ok_or_else(|| Error::invalid(format!("unresolved pending node {value}")))
500        } else {
501            NodeId::from_str(value).map_err(Error::from)
502        }
503    };
504    let owner = match node.owner.as_str() {
505        "unowned" => Owner::Unowned,
506        "self" => Owner::SelfNode,
507        value => Owner::Node(resolve_node_id(value)?),
508    };
509    let mut objects = node
510        .objects
511        .iter()
512        .map(|value| {
513            if value.starts_with("pending:") {
514                object_ids
515                    .get(value)
516                    .copied()
517                    .ok_or_else(|| Error::invalid(format!("unresolved pending object {value}")))
518            } else {
519                ObjectId::from_str(value).map_err(Error::from)
520            }
521        })
522        .collect::<Result<Vec<_>, _>>()?;
523    if node.attach_session_archive && !objects.contains(&session_object_id) {
524        objects.push(session_object_id);
525    }
526    Ok(NodeData {
527        short_name: replace_pending_object_tokens_in_text(&node.short_name, object_ids),
528        short_description: replace_pending_object_tokens_in_text(
529            &node.short_description,
530            object_ids,
531        ),
532        long_description: replace_pending_object_tokens_in_text(&node.long_description, object_ids),
533        owner,
534        fixed_connections: node
535            .fixed_connections
536            .iter()
537            .map(|value| resolve_node_id(value))
538            .collect::<Result<_, _>>()?,
539        recent_connections: node
540            .recent_connections
541            .iter()
542            .map(|value| resolve_node_id(value))
543            .collect::<Result<_, _>>()?,
544        objects,
545    })
546}
547
548fn replace_pending_object_tokens(bytes: &[u8], object_ids: &BTreeMap<String, ObjectId>) -> Vec<u8> {
549    let Ok(text) = std::str::from_utf8(bytes) else {
550        return bytes.to_vec();
551    };
552    replace_pending_object_tokens_in_text(text, object_ids).into_bytes()
553}
554
555fn replace_pending_object_tokens_in_text(
556    text: &str,
557    object_ids: &BTreeMap<String, ObjectId>,
558) -> String {
559    if object_ids.is_empty() || !text.contains("pending:") {
560        return text.into();
561    }
562    let mut output = String::with_capacity(text.len());
563    let mut cursor = 0;
564    while let Some(relative) = text[cursor..].find("pending:") {
565        let start = cursor + relative;
566        let number_start = start + "pending:".len();
567        let number_len = text[number_start..]
568            .bytes()
569            .take_while(u8::is_ascii_digit)
570            .count();
571        if number_len == 0 {
572            output.push_str(&text[cursor..number_start]);
573            cursor = number_start;
574            continue;
575        }
576        let end = number_start + number_len;
577        let token = &text[start..end];
578        let left_boundary = start == 0
579            || !text[..start]
580                .chars()
581                .next_back()
582                .is_some_and(|character| character.is_ascii_alphanumeric() || character == '_');
583        let right_boundary = text[end..]
584            .chars()
585            .next()
586            .is_none_or(|character| !character.is_ascii_alphanumeric() && character != '_');
587        if left_boundary
588            && right_boundary
589            && let Some(id) = object_ids.get(token)
590        {
591            output.push_str(&text[cursor..start]);
592            output.push_str(&id.to_string());
593            cursor = end;
594            continue;
595        }
596        output.push_str(&text[cursor..end]);
597        cursor = end;
598    }
599    output.push_str(&text[cursor..]);
600    output
601}
602
603fn request_digest(request: &CommitRequest) -> [u8; 32] {
604    let mut digest = Sha256::new();
605    digest.update(b"kcode-commit-session request v2\0");
606    hash_bytes(&mut digest, request.idempotency_key.as_bytes());
607    hash_bytes(&mut digest, request.author.as_bytes());
608    digest.update(request.source_created_at.timestamp().to_be_bytes());
609    digest.update(
610        request
611            .source_created_at
612            .timestamp_subsec_nanos()
613            .to_be_bytes(),
614    );
615    hash_bytes(&mut digest, &request.archive);
616    hash_u64(&mut digest, request.objects.len());
617    for (pending, bytes) in &request.objects {
618        hash_bytes(&mut digest, pending.as_bytes());
619        hash_bytes(&mut digest, bytes);
620    }
621    hash_u64(&mut digest, request.creates.len());
622    for (pending, node) in &request.creates {
623        hash_bytes(&mut digest, pending.as_bytes());
624        hash_node(&mut digest, node);
625    }
626    hash_u64(&mut digest, request.updates.len());
627    for (id, node) in &request.updates {
628        digest.update(id.to_bytes());
629        hash_node(&mut digest, node);
630    }
631    digest.finalize().into()
632}
633
634fn hash_node(digest: &mut Sha256, node: &PlannedNode) {
635    hash_bytes(digest, node.short_name.as_bytes());
636    hash_bytes(digest, node.short_description.as_bytes());
637    hash_bytes(digest, node.long_description.as_bytes());
638    hash_bytes(digest, node.owner.as_bytes());
639    hash_strings(digest, &node.fixed_connections);
640    hash_strings(digest, &node.recent_connections);
641    hash_strings(digest, &node.objects);
642    digest.update([u8::from(node.attach_session_archive)]);
643}
644
645fn hash_strings(digest: &mut Sha256, values: &[String]) {
646    hash_u64(digest, values.len());
647    for value in values {
648        hash_bytes(digest, value.as_bytes());
649    }
650}
651
652fn hash_bytes(digest: &mut Sha256, bytes: &[u8]) {
653    hash_u64(digest, bytes.len());
654    digest.update(bytes);
655}
656
657fn hash_u64(digest: &mut Sha256, value: usize) {
658    digest.update((value as u64).to_be_bytes());
659}
660
661fn require_one(updated: usize, message: &'static str) -> Result<(), Error> {
662    if updated == 1 {
663        Ok(())
664    } else {
665        Err(Error::internal(message))
666    }
667}
668
669#[cfg(test)]
670mod tests {
671    use super::*;
672
673    #[test]
674    fn planned_node_keeps_checkpoint_field_compatible() {
675        let node = PlannedNode {
676            short_name: "name".into(),
677            short_description: String::new(),
678            long_description: String::new(),
679            owner: "self".into(),
680            fixed_connections: Vec::new(),
681            recent_connections: Vec::new(),
682            objects: Vec::new(),
683            attach_session_archive: true,
684        };
685        let value = serde_json::to_value(&node).unwrap();
686        assert_eq!(value["includeSessionObject"], true);
687        assert_eq!(
688            serde_json::from_value::<PlannedNode>(serde_json::json!({
689                "shortName": "name",
690                "shortDescription": "",
691                "longDescription": "",
692                "owner": "self",
693                "attachSessionArchive": true
694            }))
695            .unwrap(),
696            node
697        );
698    }
699}
kcode_commit_session/lib.rs

kcode_commit_session/
lib.rs
