journal_sdk/
lib.rs

1#![doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/README.md"))]
2
3pub use crate::config::Config;
4use crate::evaluator::{Evaluator, Primitive, Type, json2lisp, lisp2json, obj2str};
5use crate::extensions::crypto::{
6    primitive_s7_crypto_generate, primitive_s7_crypto_sign, primitive_s7_crypto_verify,
7};
8use crate::extensions::system::{primitive_s7_system_time_unix, primitive_s7_system_time_utc};
9use crate::persistor::{MemoryPersistor, PERSISTOR, Persistor};
10use crate::cache::{
11    strict_cache_get, strict_cache_put, OverlayPersistor, ResolvedNode, ResolveSource,
12    resolve_branch_with, resolve_node_with, resolve_stump_with,
13};
14pub use crate::persistor::{SIZE, Word};
15use libc;
16use log::{debug, info, warn};
17use once_cell::sync::Lazy;
18use serde_json::Value;
19use std::collections::{HashMap, HashSet};
20use std::sync::{Arc, Mutex, RwLock};
21use std::time::Instant;
22
23use evaluator as s7;
24use sha2::{Digest, Sha256};
25use std::ffi::{CStr, CString};
26
27mod config;
28mod cache;
29pub mod evaluator;
30mod persistor;
31mod extensions {
32    pub mod crypto;
33    pub mod system;
34}
35
36pub static JOURNAL: Lazy<Journal> = Lazy::new(|| Journal::new());
37
38pub(crate) const SYNC_NODE_TAG: i64 = 0;
39
40const GENESIS_STR: &str = "(lambda (*sync-state* query) (cons (eval query) *sync-state*))";
41
42pub(crate) const NULL: Word = [
43    0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
44];
45
46pub(crate) struct Session {
47    pub(crate) record: Word,
48    pub(crate) state: Word,
49    pub(crate) persistor: MemoryPersistor,
50    pub(crate) cache: Arc<Mutex<HashMap<(String, String, Vec<u8>), Vec<u8>>>>,
51    pub(crate) strict_env_loc: Option<s7::s7_int>,
52    pub(crate) strict_loader_locs: HashMap<Word, s7::s7_int>,
53    pub(crate) strict_overlay_persistor: Option<MemoryPersistor>,
54    pub(crate) strict_overlay_handles: HashSet<Word>,
55    pub(crate) external_called: bool,
56}
57
58impl Session {
59    fn new(
60        record: Word,
61        state: Word,
62        persistor: MemoryPersistor,
63        cache: Arc<Mutex<HashMap<(String, String, Vec<u8>), Vec<u8>>>>,
64    ) -> Self {
65        Self {
66            record,
67            state,
68            persistor,
69            cache,
70            strict_env_loc: None,
71            strict_loader_locs: HashMap::new(),
72            strict_overlay_persistor: None,
73            strict_overlay_handles: HashSet::new(),
74            external_called: false,
75        }
76    }
77}
78
79pub(crate) static SESSIONS: Lazy<RwLock<HashMap<usize, Session>>> =
80    Lazy::new(|| RwLock::new(HashMap::new()));
81
82struct CallOnDrop<F: FnMut()>(F);
83
84impl<F: FnMut()> Drop for CallOnDrop<F> {
85    fn drop(&mut self) {
86        (self.0)();
87    }
88}
89
90#[derive(Debug)]
91pub struct JournalAccessError(pub Word);
92
93static LOCK: Mutex<()> = Mutex::new(());
94static RUNS: usize = 1;
95
96fn escape_scheme_string(value: &str) -> String {
97    value.replace('\\', "\\\\").replace('\"', "\\\"")
98}
99
100fn truncate_for_log(value: &str, limit: usize) -> String {
101    let truncated: String = value.chars().take(limit).collect();
102    if value.chars().count() > limit {
103        format!("{truncated} ...")
104    } else {
105        truncated
106    }
107}
108
109fn warn_on_error_result(query: &str, output: &str) {
110    if output.starts_with("(error ") {
111        warn!(
112            "Evaluation returned error form. Query: {} Result: {}",
113            truncate_for_log(query, 256),
114            truncate_for_log(output, 256),
115        );
116    }
117}
118
119/// Journals are the primary way that application developers
120/// interact with the synchronic web.
121///
122/// Conceptually, a Journal is a
123/// service that interacts with users and other Journals (nodes) to
124/// persist synchronic web state. Behind the schemes, it is
125/// responsible for two capabilities:
126///
127/// * __Persistence__: managing bytes on the global hash graph
128///
129/// * __Evaluation__: executing code in the global Scheme environment
130///
131/// __Records__ are the primary way that developers interface with
132/// Journals. A Record is a mapping between a constant identifier and
133/// mutable state. Both identifiers and state are represented as
134/// fixed-size __Words__ that the outputs of a cryptographic hash
135/// function. When a new record is created, the Journal returns a
136/// record secret that is the second hash preimage of the identifier.
137/// This is intended to be used so that applications can bootstrap
138/// records into increasingly sophisticated notions of identity.
139pub struct Journal {
140    client: reqwest::Client,
141}
142
143impl Journal {
144    fn new() -> Self {
145        match PERSISTOR.root_new(
146            NULL,
147            PERSISTOR
148                .branch_set(
149                    PERSISTOR
150                        .leaf_set(GENESIS_STR.as_bytes().to_vec())
151                        .expect("Failed to create genesis leaf"),
152                    NULL,
153                    NULL,
154                )
155                .expect("Failed to create genesis branch"),
156        ) {
157            Ok(_) => Self {
158                client: reqwest::Client::new(),
159            },
160            Err(_) => Self {
161                client: reqwest::Client::new(),
162            },
163        }
164    }
165
166    /// Evaluate a Scheme expression within a Record
167    ///
168    /// # Examples
169    /// ```
170    /// use journal_sdk::JOURNAL;
171    ///
172    /// // Simple expression
173    /// let output = JOURNAL.evaluate("(+ 1 2)");
174    /// assert!(output == "3");
175    ///
176    /// // Complex expression
177    /// let output = JOURNAL.evaluate(
178    ///     "(begin (define (add2 x) (+ x 2)) (add2 1))",
179    /// );
180    /// assert!(output == "3");
181    pub fn evaluate(&self, query: &str) -> String {
182        self.evaluate_record(NULL, query)
183    }
184
185    pub fn evaluate_json(&self, query: Value) -> Value {
186        match json2lisp(&query) {
187            Ok(scheme_query) => {
188                let result = self.evaluate_record(NULL, scheme_query.as_str());
189                match lisp2json(result.as_str()) {
190                    Ok(json_result) => json_result,
191                    Err(_) => {
192                        log::warn!("Failed to parse Scheme to JSON. Result: {}", result);
193                        lisp2json("(error 'parse-error \"Failed to parse Scheme to JSON\")")
194                    }
195                    .expect("Error parsing the JSON error message"),
196                }
197            }
198            Err(_) => {
199                let query_str = serde_json::to_string(&query)
200                    .unwrap_or_else(|_| "<unprintable json>".to_string());
201                log::warn!("Failed to parse JSON to Scheme. Query: {}", query_str);
202                lisp2json("(error 'parse-error \"Failed to parse JSON to Scheme\")")
203            }
204            .expect("Error parsing the JSON error message"),
205        }
206    }
207
208    /// Convert a Scheme expression into its JSON representation without evaluation.
209    ///
210    /// # Examples
211    /// ```
212    /// use journal_sdk::JOURNAL;
213    /// use serde_json::json;
214    ///
215    /// let output = JOURNAL.scheme_to_json("(+ 1 2)");
216    /// assert_eq!(output, json!(["+", 1, 2]));
217    /// ```
218    pub fn scheme_to_json(&self, query: &str) -> Value {
219        match lisp2json(query) {
220            Ok(json_result) => json_result,
221            Err(_) => {
222                log::warn!("Failed to parse Scheme to JSON. Query: {}", query);
223                lisp2json("(error 'parse-error \"Failed to parse Scheme to JSON\")")
224            }
225            .expect("Error parsing the JSON error message"),
226        }
227    }
228
229    /// Convert a JSON expression into its Scheme representation without evaluation.
230    ///
231    /// # Examples
232    /// ```
233    /// use journal_sdk::JOURNAL;
234    /// use serde_json::json;
235    ///
236    /// let output = JOURNAL.json_to_scheme(json!(["+", 1, 2]));
237    /// assert_eq!(output, "(+ 1 2)");
238    /// ```
239    pub fn json_to_scheme(&self, query: Value) -> String {
240        match json2lisp(&query) {
241            Ok(scheme_result) => scheme_result,
242            Err(_) => {
243                let query_str = serde_json::to_string(&query)
244                    .unwrap_or_else(|_| "<unprintable json>".to_string());
245                log::warn!("Failed to parse JSON to Scheme. Query: {}", query_str);
246                "(error 'parse-error \"Failed to parse JSON to Scheme\")".to_string()
247            }
248        }
249    }
250
251    fn evaluate_record(&self, record: Word, query: &str) -> String {
252        let mut runs = 0;
253        let cache = Arc::new(Mutex::new(HashMap::new()));
254
255        let start = Instant::now();
256        debug!(
257            "Evaluating ({})",
258            query.chars().take(128).collect::<String>(),
259        );
260
261        loop {
262            let _lock1 = if runs >= RUNS {
263                Some(LOCK.lock().expect("Failed to acquire concurrency lock"))
264            } else {
265                None
266            };
267
268            let (state_old, record_temp) = {
269                let _lock2 = match _lock1 {
270                    Some(_) => None,
271                    None => Some(LOCK.lock().expect("Failed to acquire secondary lock")),
272                };
273                let state_old = PERSISTOR
274                    .root_get(record)
275                    .expect("Failed to get current state");
276                let record_temp = PERSISTOR
277                    .root_temp(state_old)
278                    .expect("Failed to create temporary record");
279                (state_old, record_temp)
280            };
281
282            let _record_dropper = CallOnDrop(|| {
283                PERSISTOR
284                    .root_delete(record_temp)
285                    .expect("Failed to delete temporary record");
286            });
287
288            let genesis_branch = PERSISTOR
289                .branch_get(state_old)
290                .expect("Failed to get genesis branch");
291
292            let genesis_func = PERSISTOR
293                .leaf_get(genesis_branch.0)
294                .expect("Failed to get genesis function")
295                .to_vec();
296
297            let genesis_str = String::from_utf8_lossy(&genesis_func);
298
299            let evaluator = Evaluator::new(
300                vec![(SYNC_NODE_TAG, type_s7_sync_node())]
301                    .into_iter()
302                    .collect(),
303                vec![
304                    primitive_s7_sync_hash(),
305                    primitive_s7_sync_null(),
306                    primitive_s7_sync_state(),
307                    primitive_s7_sync_stub(),
308                    primitive_s7_sync_is_node(),
309                    primitive_s7_sync_is_pair(),
310                    primitive_s7_sync_is_stub(),
311                    primitive_s7_sync_is_null(),
312                    primitive_s7_sync_digest(),
313                    primitive_s7_sync_cons(),
314                    primitive_s7_sync_car(),
315                    primitive_s7_sync_cdr(),
316                    primitive_s7_sync_cut(),
317                    primitive_s7_sync_create(),
318                    primitive_s7_sync_delete(),
319                    primitive_s7_sync_all(),
320                    primitive_s7_sync_call(),
321                    primitive_s7_sync_eval(),
322                    primitive_s7_sync_remote(),
323                    primitive_s7_sync_http(),
324                    primitive_s7_crypto_generate(),
325                    primitive_s7_crypto_sign(),
326                    primitive_s7_crypto_verify(),
327                    primitive_s7_system_time_unix(),
328                    primitive_s7_system_time_utc(),
329                ],
330            );
331
332            let persistor_initial = MemoryPersistor::new();
333
334            match PERSISTOR.branch_get(state_old) {
335                Ok((left, right, digest)) => persistor_initial
336                    .branch_set(left, right, digest)
337                    .expect("Could not set state root branch to session persistor"),
338                Err(_) => panic!("Could not set state root branch to session persistor"),
339            };
340
341            SESSIONS
342                .write()
343                .expect("Failed to acquire sessions lock")
344                .insert(
345                    evaluator.sc as usize,
346                    Session::new(record, state_old, persistor_initial, cache.clone()),
347                );
348
349            let _session_dropper = CallOnDrop(|| {
350                let mut session = SESSIONS
351                    .write()
352                    .expect("Failed to acquire sessions lock for cleanup");
353                if let Some(session) = session.remove(&(evaluator.sc as usize)) {
354                    if let Some(loc) = session.strict_env_loc {
355                        unsafe {
356                            s7::s7_gc_unprotect_at(evaluator.sc, loc);
357                        }
358                    }
359                    for loc in session.strict_loader_locs.into_values() {
360                        unsafe {
361                            s7::s7_gc_unprotect_at(evaluator.sc, loc);
362                        }
363                    }
364                    if let Some(overlay) = session.strict_overlay_persistor {
365                        for handle in session.strict_overlay_handles {
366                            let _ = overlay.root_delete(handle);
367                        }
368                    }
369                }
370            });
371
372            let expr = format!(
373                "((eval {}) (sync-state) (read (open-input-string \"{}\")))",
374                genesis_str,
375                escape_scheme_string(query),
376            );
377
378            let result = evaluator.evaluate(expr.as_str());
379            runs += 1;
380
381            let (persistor, overlay_persistor, external_called) = {
382                let session = SESSIONS.read().expect("Failed to acquire sessions lock");
383                let session = session
384                    .get(&(evaluator.sc as usize))
385                    .expect("Session not found in SESSIONS map");
386                (
387                    session.persistor.clone(),
388                    session.strict_overlay_persistor.clone(),
389                    session.external_called,
390                )
391            };
392
393            let (output, state_new) = match result.starts_with("(error '") {
394                true => (result, state_old),
395                false => match result.rfind('.') {
396                    Some(index) => match *&result[(index + 16)..(result.len() - 3)]
397                        .split(' ')
398                        .collect::<Vec<&str>>()
399                        .iter()
400                        .map(|x| x.parse::<u8>().expect("Failed to parse state byte"))
401                        .collect::<Vec<u8>>()
402                        .try_into()
403                    {
404                        Ok(state_new) => (String::from(&result[1..(index - 1)]), state_new),
405                        Err(_) => (
406                            String::from("(error 'sync-format \"Invalid return format\")"),
407                            state_old,
408                        ),
409                    },
410                    None => (
411                        String::from("(error 'sync-format \"Invalid return format\")"),
412                        state_old,
413                    ),
414                },
415            };
416
417            if external_called && state_old != state_new {
418                let output = String::from(
419                    "(error 'external-state-error \"Request called an external function and changed state\")",
420                );
421                warn_on_error_result(query, output.as_str());
422                debug!(
423                    "Completed ({:?}) {} -> {}",
424                    start.elapsed(),
425                    query.chars().take(128).collect::<String>(),
426                    output,
427                );
428                return output;
429            }
430
431            match state_old == state_new {
432                true => {
433                    warn_on_error_result(query, output.as_str());
434                    debug!(
435                        "Completed ({:?}) {} -> {}",
436                        start.elapsed(),
437                        query.chars().take(128).collect::<String>(),
438                        output,
439                    );
440                    return output;
441                }
442                false => match state_old
443                    == PERSISTOR
444                        .root_get(record)
445                        .expect("Failed to get record state for comparison")
446                {
447                    true => {
448                        {
449                            let _lock2 = match _lock1 {
450                                Some(_) => None,
451                                None => {
452                                    Some(LOCK.lock().expect("Failed to acquire secondary lock"))
453                                }
454                            };
455
456                            let overlay_source = OverlayPersistor {
457                                primary: persistor.clone(),
458                                overlay: overlay_persistor.clone(),
459                            };
460                            // Commits must see both the session graph and any cache-backed
461                            // overlay roots attached by strict cache hits in this session.
462                            match PERSISTOR.root_set(record, state_old, state_new, &overlay_source) {
463                                Ok(_) => {
464                                    warn_on_error_result(query, output.as_str());
465                                    debug!(
466                                        "Completed ({:?}) {} -> {}",
467                                        start.elapsed(),
468                                        query.chars().take(128).collect::<String>(),
469                                        output,
470                                    );
471                                    return output;
472                                }
473                                Err(_) => {
474                                    info!(
475                                        "Rerunning (x{}) due to concurrency collision: {}",
476                                        runs,
477                                        query.chars().take(128).collect::<String>(),
478                                    );
479                                    continue;
480                                }
481                            }
482                        }
483                    }
484                    false => {
485                        info!(
486                            "Rerunning (x{}) due to concurrency collision: {}",
487                            runs,
488                            query.chars().take(128).collect::<String>(),
489                        );
490                        continue;
491                    }
492                },
493            }
494        }
495    }
496}
497
498pub(crate) unsafe fn sync_error(sc: *mut s7::s7_scheme, string: &str) -> s7::s7_pointer {
499    unsafe {
500        let c_string = CString::new(string).expect("Failed to create CString from string");
501
502        s7::s7_error(
503            sc,
504            s7::s7_make_symbol(sc, c"sync-web-error".as_ptr()),
505            s7::s7_list(sc, 1, s7::s7_make_string(sc, c_string.as_ptr())),
506        )
507    }
508}
509
510fn mark_external_called(sc: *mut s7::s7_scheme) {
511    let mut sessions = SESSIONS
512        .write()
513        .expect("Failed to acquire sessions lock for external call tracking");
514    let session = sessions
515        .get_mut(&(sc as usize))
516        .expect("Session not found for given context");
517    session.external_called = true;
518}
519
520fn type_s7_sync_node() -> Type {
521    unsafe extern "C" fn free(_sc: *mut s7::s7_scheme, obj: s7::s7_pointer) -> s7::s7_pointer {
522        unsafe {
523            sync_heap_free(s7::s7_c_object_value(obj));
524            std::ptr::null_mut()
525        }
526    }
527
528    unsafe extern "C" fn mark(_sc: *mut s7::s7_scheme, _obj: s7::s7_pointer) -> s7::s7_pointer {
529        std::ptr::null_mut()
530    }
531
532    unsafe extern "C" fn is_equal(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
533        unsafe {
534            match sync_is_node(s7::s7_cadr(args)) {
535                true => {
536                    let word1 = sync_heap_read(s7::s7_c_object_value(s7::s7_car(args)));
537                    let word2 = sync_heap_read(s7::s7_c_object_value(s7::s7_cadr(args)));
538                    s7::s7_make_boolean(sc, word1 == word2)
539                }
540                false => s7::s7_wrong_type_arg_error(
541                    sc,
542                    c"equal?".as_ptr(),
543                    2,
544                    s7::s7_cadr(args),
545                    c"a sync-node".as_ptr(),
546                ),
547            }
548        }
549    }
550
551    unsafe extern "C" fn to_string(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
552        unsafe {
553            string_to_s7(
554                sc,
555                format!(
556                    "(sync-node #u({}))",
557                    sync_heap_read(s7::s7_c_object_value(s7::s7_car(args)))
558                        .iter()
559                        .map(|&byte| byte.to_string())
560                        .collect::<Vec<String>>()
561                        .join(" "),
562                )
563                .as_str(),
564            )
565        }
566    }
567
568    Type::new(c"sync-node", free, mark, is_equal, to_string)
569}
570
571fn primitive_s7_sync_stub() -> Primitive {
572    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
573        unsafe {
574            let bv = s7::s7_car(args);
575
576            if !s7::s7_is_byte_vector(bv) || s7::s7_vector_length(bv) as usize != SIZE {
577                return s7::s7_wrong_type_arg_error(
578                    sc,
579                    c"sync-cut".as_ptr(),
580                    1,
581                    s7::s7_car(args),
582                    c"a hash-sized byte-vector".as_ptr(),
583                );
584            }
585
586            let mut digest = [0 as u8; SIZE];
587            for i in 0..SIZE {
588                digest[i] = s7::s7_byte_vector_ref(bv, i as i64);
589            }
590
591            let persistor = {
592                let session = SESSIONS.read().expect("Failed to acquire SESSIONS lock");
593                &session
594                    .get(&(sc as usize))
595                    .expect("Session not found for given context")
596                    .persistor
597                    .clone()
598            };
599
600            match persistor.stump_set(digest) {
601                Ok(stump) => s7::s7_make_c_object(sc, SYNC_NODE_TAG, sync_heap_make(stump)),
602                Err(_) => sync_error(sc, "Journal is unable to create stub node (sync-stub)"),
603            }
604        }
605    }
606
607    Primitive::new(
608        code,
609        c"sync-stub",
610        c"(sync-stub digest) create a sync stub from the provided byte-vector",
611        1,
612        0,
613        false,
614    )
615}
616
617fn primitive_s7_sync_hash() -> Primitive {
618    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
619        unsafe {
620            let data_bv = s7::s7_car(args);
621
622            // check the input arguments
623            if !s7::s7_is_byte_vector(data_bv) {
624                return s7::s7_wrong_type_arg_error(
625                    sc,
626                    c"sync-hash".as_ptr(),
627                    1,
628                    data_bv,
629                    c"a byte-vector".as_ptr(),
630                );
631            }
632
633            // convert to rust data types
634            let mut data = vec![];
635            for i in 0..s7::s7_vector_length(data_bv) {
636                data.push(s7::s7_byte_vector_ref(data_bv, i as i64))
637            }
638
639            let digest = Sha256::digest(data).to_vec();
640            let digest_bv = s7::s7_make_byte_vector(sc, SIZE as i64, 1, std::ptr::null_mut());
641            for i in 0..SIZE {
642                s7::s7_byte_vector_set(digest_bv, i as i64, digest[i]);
643            }
644            digest_bv
645        }
646    }
647
648    Primitive::new(
649        code,
650        c"sync-hash",
651        c"(sync-hash bv) compute the SHA-256 digest of a byte vector",
652        1,
653        0,
654        false,
655    )
656}
657
658fn primitive_s7_sync_state() -> Primitive {
659    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
660        unsafe {
661            if !s7::s7_is_null(sc, args) {
662                return s7::s7_wrong_number_of_args_error(
663                    sc,
664                    c"sync-state".as_ptr(),
665                    args,
666                );
667            }
668
669            let state = {
670                let session = SESSIONS.read().expect("Failed to acquire sessions lock");
671                session
672                    .get(&(sc as usize))
673                    .expect("Session not found for sync-state")
674                    .state
675            };
676
677            s7::s7_make_c_object(sc, SYNC_NODE_TAG, sync_heap_make(state))
678        }
679    }
680
681    Primitive::new(
682        code,
683        c"sync-state",
684        c"(sync-state) returns the current session state as a sync-node",
685        0,
686        0,
687        false,
688    )
689}
690
691fn primitive_s7_sync_is_node() -> Primitive {
692    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
693        unsafe { s7::s7_make_boolean(sc, sync_is_node(s7::s7_car(args))) }
694    }
695
696    Primitive::new(
697        code,
698        c"sync-node?",
699        c"(sync-node?) returns whether the object is a sync node",
700        1,
701        0,
702        false,
703    )
704}
705
706fn primitive_s7_sync_null() -> Primitive {
707    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, _args: s7::s7_pointer) -> s7::s7_pointer {
708        unsafe { s7::s7_make_c_object(sc, SYNC_NODE_TAG, sync_heap_make(NULL)) }
709    }
710
711    Primitive::new(
712        code,
713        c"sync-null",
714        c"(sync-null) returns the null synchronic node",
715        0,
716        0,
717        false,
718    )
719}
720
721fn primitive_s7_sync_is_null() -> Primitive {
722    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
723        unsafe {
724            let arg = s7::s7_car(args);
725            if sync_is_node(arg) {
726                let word = sync_heap_read(s7::s7_c_object_value(arg));
727                for i in 0..SIZE {
728                    if word[i] != 0 {
729                        return s7::s7_make_boolean(sc, false);
730                    }
731                }
732                s7::s7_make_boolean(sc, true)
733            } else if s7::s7_is_byte_vector(arg) {
734                s7::s7_make_boolean(sc, false)
735            } else {
736                s7::s7_wrong_type_arg_error(
737                    sc,
738                    c"sync-null?".as_ptr(),
739                    1,
740                    arg,
741                    c"a sync-node or byte-vector".as_ptr(),
742                )
743            }
744        }
745    }
746
747    Primitive::new(
748        code,
749        c"sync-null?",
750        c"(sync-null? sp) returns whether a sync-node or byte-vector is equal to sync-null",
751        1,
752        0,
753        false,
754    )
755}
756
757fn primitive_s7_sync_is_pair() -> Primitive {
758    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
759        unsafe {
760            let arg = s7::s7_car(args);
761            if sync_is_node(arg) {
762                let word = sync_heap_read(s7::s7_c_object_value(arg));
763                s7::s7_make_boolean(sc, sync_branch_children(sc, word).is_ok())
764            } else if s7::s7_is_byte_vector(arg) {
765                s7::s7_make_boolean(sc, false)
766            } else {
767                s7::s7_wrong_type_arg_error(
768                    sc,
769                    c"sync-pair?".as_ptr(),
770                    1,
771                    arg,
772                    c"a sync-node or byte-vector".as_ptr(),
773                )
774            }
775        }
776    }
777
778    Primitive::new(
779        code,
780        c"sync-pair?",
781        c"(sync-pair? sp) returns whether a sync-node or byte-vector is a pair",
782        1,
783        0,
784        false,
785    )
786}
787
788fn primitive_s7_sync_is_stub() -> Primitive {
789    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
790        unsafe {
791            let arg = s7::s7_car(args);
792            if sync_is_node(arg) {
793                let word = sync_heap_read(s7::s7_c_object_value(arg));
794                let (persistor, overlay) = session_storage_for(sc);
795                s7::s7_make_boolean(
796                    sc,
797                    resolve_stump_with(&persistor, overlay.as_ref(), word).is_some(),
798                )
799            } else if s7::s7_is_byte_vector(arg) {
800                s7::s7_make_boolean(sc, false)
801            } else {
802                s7::s7_wrong_type_arg_error(
803                    sc,
804                    c"sync-stub?".as_ptr(),
805                    1,
806                    arg,
807                    c"a sync-node or byte-vector".as_ptr(),
808                )
809            }
810        }
811    }
812
813    Primitive::new(
814        code,
815        c"sync-stub?",
816        c"(sync-stub? sp) returns whether a sync-node or byte-vector is a stub",
817        1,
818        0,
819        false,
820    )
821}
822
823fn primitive_s7_sync_digest() -> Primitive {
824    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
825        unsafe {
826            let arg = s7::s7_car(args);
827            if sync_is_node(arg) {
828                let word = sync_heap_read(s7::s7_c_object_value(arg));
829                let digest = sync_digest(sc, word).expect("Failed to obtain digest");
830                let bv = s7::s7_make_byte_vector(sc, SIZE as i64, 1, std::ptr::null_mut());
831                for i in 0..SIZE {
832                    s7::s7_byte_vector_set(bv, i as i64, digest[i]);
833                }
834                bv
835            } else if s7::s7_is_byte_vector(arg) {
836                let mut data = vec![];
837                for i in 0..s7::s7_vector_length(arg) {
838                    data.push(s7::s7_byte_vector_ref(arg, i as i64))
839                }
840                let digest = Sha256::digest(data);
841                let bv = s7::s7_make_byte_vector(sc, SIZE as i64, 1, std::ptr::null_mut());
842                for i in 0..SIZE {
843                    s7::s7_byte_vector_set(bv, i as i64, digest[i]);
844                }
845                bv
846            } else {
847                s7::s7_wrong_type_arg_error(
848                    sc,
849                    c"sync-digest".as_ptr(),
850                    1,
851                    arg,
852                    c"a sync-node or byte-vector".as_ptr(),
853                )
854            }
855        }
856    }
857
858    Primitive::new(
859        code,
860        c"sync-digest",
861        c"(sync-digest value) returns the digest of a sync-node or byte-vector as a byte-vector",
862        1,
863        0,
864        false,
865    )
866}
867
868fn primitive_s7_sync_cons() -> Primitive {
869    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
870        unsafe {
871            let persistor = {
872                let session = SESSIONS.read().expect("Failed to acquire sessions lock");
873                &session
874                    .get(&(sc as usize))
875                    .expect("Session not found for sync-cons")
876                    .persistor
877                    .clone()
878            };
879
880            let handle_arg = |obj, number| {
881                if sync_is_node(obj) {
882                    Ok(sync_heap_read(s7::s7_c_object_value(obj)))
883                } else if s7::s7_is_byte_vector(obj) {
884                    let mut content = vec![];
885                    for i in 0..s7::s7_vector_length(obj) {
886                        content.push(s7::s7_byte_vector_ref(obj, i as i64))
887                    }
888                    match persistor.leaf_set(content) {
889                        Ok(atom) => Ok(atom),
890                        Err(_) => Err(sync_error(
891                            sc,
892                            "Journal is unable to add leaf node (sync-cons)",
893                        )),
894                    }
895                } else {
896                    Err(s7::s7_wrong_type_arg_error(
897                        sc,
898                        c"sync-cons".as_ptr(),
899                        number,
900                        obj,
901                        c"a byte vector or a sync node".as_ptr(),
902                    ))
903                }
904            };
905
906            match (
907                handle_arg(s7::s7_car(args), 1),
908                handle_arg(s7::s7_cadr(args), 2),
909            ) {
910                (Ok(left), Ok(right)) => match (sync_digest(sc, left), sync_digest(sc, right)) {
911                    (Ok(digest_left), Ok(digest_right)) => {
912                        let mut joined = [0 as u8; SIZE * 2];
913                        joined[..SIZE].copy_from_slice(&digest_left);
914                        joined[SIZE..].copy_from_slice(&digest_right);
915                        let digest = Word::from(Sha256::digest(joined));
916
917                        match persistor.branch_set(left, right, digest) {
918                            Ok(pair) => {
919                                s7::s7_make_c_object(sc, SYNC_NODE_TAG, sync_heap_make(pair))
920                            }
921                            Err(_) => {
922                                sync_error(sc, "Journal is unable to add pair node (sync-cons)")
923                            }
924                        }
925                    }
926                    _ => sync_error(sc, "Journal is unable to obtain node digests (sync-cons)"),
927                },
928                (Err(left), _) => left,
929                (_, Err(right)) => right,
930            }
931        }
932    }
933
934    Primitive::new(
935        code,
936        c"sync-cons",
937        c"(sync-cons first rest) construct a new sync pair node",
938        2,
939        0,
940        false,
941    )
942}
943
944fn primitive_s7_sync_car() -> Primitive {
945    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
946        unsafe {
947            if !sync_is_node(s7::s7_car(args)) {
948                return s7::s7_wrong_type_arg_error(
949                    sc,
950                    c"sync-car".as_ptr(),
951                    1,
952                    s7::s7_car(args),
953                    c"a sync-pair".as_ptr(),
954                );
955            }
956            sync_cxr(sc, args, c"sync-car", |children| children.0)
957        }
958    }
959
960    Primitive::new(
961        code,
962        c"sync-car",
963        c"(sync-car pair) retrieve the first element of a sync pair",
964        1,
965        0,
966        false,
967    )
968}
969
970fn primitive_s7_sync_cdr() -> Primitive {
971    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
972        unsafe {
973            if !sync_is_node(s7::s7_car(args)) {
974                return s7::s7_wrong_type_arg_error(
975                    sc,
976                    c"sync-cdr".as_ptr(),
977                    1,
978                    s7::s7_car(args),
979                    c"a sync-pair".as_ptr(),
980                );
981            }
982            sync_cxr(sc, args, c"sync-cdr", |children| children.1)
983        }
984    }
985
986    Primitive::new(
987        code,
988        c"sync-cdr",
989        c"(sync-cdr pair) retrieve the second element of a sync pair",
990        1,
991        0,
992        false,
993    )
994}
995
996fn primitive_s7_sync_cut() -> Primitive {
997    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
998        unsafe {
999            let arg = s7::s7_car(args);
1000
1001            let handle_digest = |digest| {
1002                let persistor = {
1003                    let session = SESSIONS.read().expect("Failed to acquire SESSIONS lock");
1004                    &session
1005                        .get(&(sc as usize))
1006                        .expect("Session not found for given context")
1007                        .persistor
1008                        .clone()
1009                };
1010                match persistor.stump_set(digest) {
1011                    Ok(stump) => s7::s7_make_c_object(sc, SYNC_NODE_TAG, sync_heap_make(stump)),
1012                    Err(_) => sync_error(sc, "Journal is unable to add stub node (sync-cut)"),
1013                }
1014            };
1015
1016            if s7::s7_is_byte_vector(arg) {
1017                let mut content = vec![];
1018                for i in 0..s7::s7_vector_length(arg) {
1019                    content.push(s7::s7_byte_vector_ref(arg, i as i64))
1020                }
1021                handle_digest(Word::from(Sha256::digest(Sha256::digest(&content))))
1022            } else if sync_is_node(arg) {
1023                match sync_digest(sc, sync_heap_read(s7::s7_c_object_value(arg))) {
1024                    Ok(digest) => handle_digest(digest),
1025                    Err(_) => sync_error(sc, "Journal does not recognize input node (sync-cut)"),
1026                }
1027            } else {
1028                s7::s7_wrong_type_arg_error(
1029                    sc,
1030                    c"sync-cut".as_ptr(),
1031                    1,
1032                    s7::s7_car(args),
1033                    c"a sync-node or byte-vector".as_ptr(),
1034                )
1035            }
1036        }
1037    }
1038
1039    Primitive::new(
1040        code,
1041        c"sync-cut",
1042        c"(sync-cut value) obtain the stub of a sync-node",
1043        1,
1044        0,
1045        false,
1046    )
1047}
1048
1049fn primitive_s7_sync_create() -> Primitive {
1050    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
1051        unsafe {
1052            let id = s7::s7_car(args);
1053
1054            if !s7::s7_is_byte_vector(id) || s7::s7_vector_length(id) as usize != SIZE {
1055                return s7::s7_wrong_type_arg_error(
1056                    sc,
1057                    c"sync-create".as_ptr(),
1058                    1,
1059                    id,
1060                    c"a hash-sized byte-vector".as_ptr(),
1061                );
1062            }
1063
1064            let mut record: Word = [0 as u8; SIZE];
1065
1066            for i in 0..SIZE {
1067                record[i as usize] = s7::s7_byte_vector_ref(id, i as i64)
1068            }
1069
1070            debug!("Adding record: {}", hex::encode(record));
1071
1072            match PERSISTOR.root_new(
1073                record,
1074                PERSISTOR
1075                    .branch_set(
1076                        PERSISTOR
1077                            .leaf_set(GENESIS_STR.as_bytes().to_vec())
1078                            .expect("Failed to create genesis leaf for new record"),
1079                        NULL,
1080                        NULL,
1081                    )
1082                    .expect("Failed to create genesis branch for new record"),
1083            ) {
1084                Ok(_) => s7::s7_make_boolean(sc, true),
1085                Err(_) => s7::s7_error(
1086                    sc,
1087                    s7::s7_make_symbol(sc, c"sync-web-error".as_ptr()),
1088                    s7::s7_list(
1089                        sc,
1090                        1,
1091                        s7::s7_make_string(sc, c"record ID is already in use".as_ptr()),
1092                    ),
1093                ),
1094            }
1095        }
1096    }
1097
1098    Primitive::new(
1099        code,
1100        c"sync-create",
1101        c"(sync-create id) create a new synchronic record with the given 32-byte ID",
1102        1,
1103        0,
1104        false,
1105    )
1106}
1107
1108fn primitive_s7_sync_delete() -> Primitive {
1109    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
1110        unsafe {
1111            let id = s7::s7_car(args);
1112
1113            if !s7::s7_is_byte_vector(id) || s7::s7_vector_length(id) as usize != SIZE {
1114                return s7::s7_wrong_type_arg_error(
1115                    sc,
1116                    c"sync-delete".as_ptr(),
1117                    1,
1118                    id,
1119                    c"a hash-sized byte-vector".as_ptr(),
1120                );
1121            }
1122
1123            let mut record: Word = [0 as u8; SIZE];
1124
1125            for i in 0..s7::s7_vector_length(id) {
1126                record[i as usize] = s7::s7_byte_vector_ref(id, i as i64)
1127            }
1128
1129            if record == NULL {
1130                return s7::s7_error(
1131                    sc,
1132                    s7::s7_make_symbol(sc, c"sync-web-error".as_ptr()),
1133                    s7::s7_list(
1134                        sc,
1135                        1,
1136                        s7::s7_make_string(sc, c"cannot delete the root record".as_ptr()),
1137                    ),
1138                );
1139            }
1140
1141            debug!("Deleting record: {}", hex::encode(record));
1142
1143            match PERSISTOR.root_delete(record) {
1144                Ok(_) => s7::s7_make_boolean(sc, true),
1145                Err(_) => s7::s7_error(
1146                    sc,
1147                    s7::s7_make_symbol(sc, c"sync-web-error".as_ptr()),
1148                    s7::s7_list(
1149                        sc,
1150                        1,
1151                        s7::s7_make_string(sc, c"record ID does not exist".as_ptr()),
1152                    ),
1153                ),
1154            }
1155        }
1156    }
1157
1158    Primitive::new(
1159        code,
1160        c"sync-delete",
1161        c"(sync-delete id) delete the synchronic record with the given 32-byte ID",
1162        1,
1163        0,
1164        false,
1165    )
1166}
1167
1168fn primitive_s7_sync_all() -> Primitive {
1169    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, _args: s7::s7_pointer) -> s7::s7_pointer {
1170        unsafe {
1171            let mut list = s7::s7_list(sc, 0);
1172
1173            for record in PERSISTOR.root_list().into_iter().rev() {
1174                let bv = s7::s7_make_byte_vector(sc, SIZE as i64, 1, std::ptr::null_mut());
1175                for i in 0..SIZE {
1176                    s7::s7_byte_vector_set(bv, i as i64, record[i]);
1177                }
1178
1179                list = s7::s7_cons(sc, bv, list)
1180            }
1181
1182            list
1183        }
1184    }
1185
1186    Primitive::new(
1187        code,
1188        c"sync-all",
1189        c"(sync-all) list all synchronic record IDs in ascending order",
1190        0,
1191        0,
1192        false,
1193    )
1194}
1195
1196fn primitive_s7_sync_call() -> Primitive {
1197    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
1198        unsafe {
1199            mark_external_called(sc);
1200
1201            let message_expr = s7::s7_car(args);
1202            let blocking = s7::s7_cadr(args);
1203
1204            if !s7::s7_is_boolean(blocking) {
1205                return s7::s7_wrong_type_arg_error(
1206                    sc,
1207                    c"sync-call".as_ptr(),
1208                    2,
1209                    blocking,
1210                    c"a boolean".as_ptr(),
1211                );
1212            }
1213
1214            let record = match s7::s7_is_null(sc, s7::s7_cddr(args)) {
1215                true => {
1216                    let session = SESSIONS.read().expect("Failed to acquire sessions lock");
1217                    session
1218                        .get(&(sc as usize))
1219                        .expect("Session number not found in sessions map")
1220                        .record
1221                }
1222                false => {
1223                    let bv = s7::s7_caddr(args);
1224                    // check the input arguments
1225                    if !s7::s7_is_byte_vector(bv) || s7::s7_vector_length(bv) as usize != SIZE {
1226                        return s7::s7_wrong_type_arg_error(
1227                            sc,
1228                            c"sync-call".as_ptr(),
1229                            3,
1230                            bv,
1231                            c"a hash-sized byte-vector".as_ptr(),
1232                        );
1233                    }
1234
1235                    let mut record = [0 as u8; SIZE];
1236                    for i in 0..SIZE {
1237                        record[i] = s7::s7_byte_vector_ref(bv, i as i64);
1238                    }
1239                    record
1240                }
1241            };
1242
1243            match PERSISTOR.root_get(record) {
1244                Ok(_) => {
1245                    let message = obj2str(sc, message_expr);
1246                    if s7::s7_boolean(sc, blocking) {
1247                        let result = JOURNAL.evaluate_record(record, message.as_str());
1248                        let c_result = CString::new(format!("(quote {})", result))
1249                            .expect("Failed to create C string from journal evaluation result");
1250                        s7::s7_eval_c_string(sc, c_result.as_ptr())
1251                    } else {
1252                        tokio::spawn(async move {
1253                            JOURNAL.evaluate_record(record, message.as_str());
1254                        });
1255                        s7::s7_make_boolean(sc, true)
1256                    }
1257                }
1258                Err(_) => s7::s7_error(
1259                    sc,
1260                    s7::s7_make_symbol(sc, c"sync-web-error".as_ptr()),
1261                    s7::s7_list(
1262                        sc,
1263                        1,
1264                        s7::s7_make_string(sc, c"record ID does not exist".as_ptr()),
1265                    ),
1266                ),
1267            }
1268        }
1269    }
1270
1271    Primitive::new(
1272        code,
1273        c"sync-call",
1274        c"(sync-call query blocking? id) query the provided record ID or self if ID not provided",
1275        2,
1276        1,
1277        false,
1278    )
1279}
1280
1281fn primitive_s7_sync_eval() -> Primitive {
1282    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
1283        unsafe {
1284            let expression = s7::s7_gc_protect_via_stack(sc, s7::s7_car(args));
1285            let strict = if s7::s7_is_null(sc, s7::s7_cdr(args)) {
1286                true
1287            } else {
1288                let strict = s7::s7_cadr(args);
1289                if !s7::s7_is_boolean(strict) {
1290                    s7::s7_gc_unprotect_via_stack(sc, expression);
1291                    return s7::s7_wrong_type_arg_error(
1292                        sc,
1293                        c"sync-eval".as_ptr(),
1294                        2,
1295                        strict,
1296                        c"a boolean".as_ptr(),
1297                    );
1298                }
1299                s7::s7_boolean(sc, strict)
1300            };
1301            if !sync_is_node(expression) {
1302                s7::s7_gc_unprotect_via_stack(sc, expression);
1303                return s7::s7_wrong_type_arg_error(
1304                    sc,
1305                    c"sync-eval".as_ptr(),
1306                    1,
1307                    expression,
1308                    c"a sync-node".as_ptr(),
1309                );
1310            }
1311
1312            let expression_word = sync_heap_read(s7::s7_c_object_value(expression));
1313            if strict {
1314                if let Some(result) = strict_cache_get(sc, expression_word) {
1315                    s7::s7_gc_unprotect_via_stack(sc, expression);
1316                    return result;
1317                }
1318                let header_word = match sync_branch_children(sc, expression_word) {
1319                    Ok((left, _)) => left,
1320                    Err(err) => {
1321                        s7::s7_gc_unprotect_via_stack(sc, expression);
1322                        return sync_error(
1323                            sc,
1324                            format!("sync-eval first argument should be a sync-node with a byte-vector header ({})", err).as_str(),
1325                        );
1326                    }
1327                };
1328                let eval_env = strict_sync_eval_env_cached(sc);
1329                let loader = match strict_loader_lookup(sc, header_word) {
1330                    Some(loader) => loader,
1331                    None => {
1332                        let header = s7::s7_gc_protect_via_stack(
1333                            sc,
1334                            sync_cxr(
1335                                sc,
1336                                s7::s7_list(sc, 1, expression),
1337                                c"sync-eval",
1338                                |children| children.0,
1339                            ),
1340                        );
1341                        if !s7::s7_is_byte_vector(header) {
1342                            s7::s7_gc_unprotect_via_stack(sc, header);
1343                            s7::s7_gc_unprotect_via_stack(sc, expression);
1344                            return sync_error(sc, "sync-eval first argument should be a sync-node with a byte-vector header");
1345                        }
1346                        let mut bytes = vec![39];
1347                        for i in 0..s7::s7_vector_length(header) {
1348                            bytes.push(s7::s7_byte_vector_ref(header, i));
1349                        }
1350                        bytes.push(0);
1351                        let loader_expr = match CString::from_vec_with_nul(bytes) {
1352                            Ok(c_string) => s7::s7_gc_protect_via_stack(sc, s7::s7_eval_c_string(sc, c_string.as_ptr())),
1353                            Err(_) => {
1354                                s7::s7_gc_unprotect_via_stack(sc, header);
1355                                s7::s7_gc_unprotect_via_stack(sc, expression);
1356                                return s7::s7_error(
1357                                    sc,
1358                                    s7::s7_make_symbol(sc, c"encoding-error".as_ptr()),
1359                                    s7::s7_list(
1360                                        sc,
1361                                        1,
1362                                        s7::s7_make_string(sc, c"Byte vector string is malformed".as_ptr()),
1363                                    ),
1364                                );
1365                            }
1366                        };
1367                        let loader = s7::s7_gc_protect_via_stack(sc, s7::s7_eval(sc, loader_expr, eval_env));
1368                        let cached = strict_loader_cache_store(sc, header_word, loader);
1369                        s7::s7_gc_unprotect_via_stack(sc, loader);
1370                        s7::s7_gc_unprotect_via_stack(sc, loader_expr);
1371                        s7::s7_gc_unprotect_via_stack(sc, header);
1372                        cached
1373                    }
1374                };
1375                let result = s7::s7_gc_protect_via_stack(
1376                    sc,
1377                    s7::s7_apply_function(sc, loader, s7::s7_list(sc, 1, expression)),
1378                );
1379                let cached = strict_cache_put(sc, expression_word, result);
1380                s7::s7_gc_unprotect_via_stack(sc, expression);
1381                s7::s7_gc_unprotect_via_stack(sc, result);
1382                return cached;
1383            } else {
1384                let header = s7::s7_gc_protect_via_stack(
1385                    sc,
1386                    sync_cxr(
1387                        sc,
1388                        s7::s7_list(sc, 1, expression),
1389                        c"sync-eval",
1390                        |children| children.0,
1391                    ),
1392                );
1393                if !s7::s7_is_byte_vector(header) {
1394                    s7::s7_gc_unprotect_via_stack(sc, header);
1395                    s7::s7_gc_unprotect_via_stack(sc, expression);
1396                    return sync_error(sc, "sync-eval first argument should be a sync-node with a byte-vector header");
1397                }
1398                let mut bytes = vec![39];
1399                for i in 0..s7::s7_vector_length(header) {
1400                    bytes.push(s7::s7_byte_vector_ref(header, i));
1401                }
1402                bytes.push(0);
1403                let loader_expr = match CString::from_vec_with_nul(bytes) {
1404                    Ok(c_string) => s7::s7_gc_protect_via_stack(sc, s7::s7_eval_c_string(sc, c_string.as_ptr())),
1405                    Err(_) => {
1406                        s7::s7_gc_unprotect_via_stack(sc, header);
1407                        s7::s7_gc_unprotect_via_stack(sc, expression);
1408                        return s7::s7_error(
1409                            sc,
1410                            s7::s7_make_symbol(sc, c"encoding-error".as_ptr()),
1411                            s7::s7_list(
1412                                sc,
1413                                1,
1414                                s7::s7_make_string(sc, c"Byte vector string is malformed".as_ptr()),
1415                            ),
1416                        );
1417                    }
1418                };
1419                let eval_env = s7::s7_gc_protect_via_stack(sc, s7::s7_curlet(sc));
1420                let loader = s7::s7_gc_protect_via_stack(sc, s7::s7_eval(sc, loader_expr, eval_env));
1421                let result = s7::s7_gc_protect_via_stack(
1422                    sc,
1423                    s7::s7_apply_function(sc, loader, s7::s7_list(sc, 1, expression)),
1424                );
1425                s7::s7_gc_unprotect_via_stack(sc, eval_env);
1426                s7::s7_gc_unprotect_via_stack(sc, loader);
1427                s7::s7_gc_unprotect_via_stack(sc, loader_expr);
1428                s7::s7_gc_unprotect_via_stack(sc, header);
1429                s7::s7_gc_unprotect_via_stack(sc, expression);
1430                s7::s7_gc_unprotect_via_stack(sc, result);
1431                return result;
1432            };
1433        }
1434    }
1435
1436    Primitive::new(
1437        code,
1438        c"sync-eval",
1439        c"(sync-eval node (strict? #t)) evaluate a sync-node strictly, or load it when strict? is #f",
1440        1,
1441        3,
1442        false,
1443    )
1444}
1445
1446fn primitive_s7_sync_http() -> Primitive {
1447    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
1448        unsafe {
1449            mark_external_called(sc);
1450
1451            let vec2s7 = |vector: Vec<u8>| {
1452                let bv = s7::s7_make_byte_vector(sc, vector.len() as i64, 1, std::ptr::null_mut());
1453                for i in 0..vector.len() {
1454                    s7::s7_byte_vector_set(bv, i as i64, vector[i]);
1455                }
1456                bv
1457            };
1458
1459            let method = obj2str(sc, s7::s7_car(args));
1460            let url = obj2str(sc, s7::s7_cadr(args));
1461
1462            let body = if s7::s7_list_length(sc, args) >= 3 {
1463                obj2str(sc, s7::s7_caddr(args))
1464            } else {
1465                String::from("")
1466            };
1467
1468            let cache_mutex = {
1469                let session = SESSIONS.read().expect("Failed to acquire sessions lock");
1470                session
1471                    .get(&(sc as usize))
1472                    .expect("Session ID not found in active sessions")
1473                    .cache
1474                    .clone()
1475            };
1476
1477            let mut cache = cache_mutex
1478                .lock()
1479                .expect("Failed to acquire cache mutex lock");
1480
1481            let key = (method.clone(), url.clone(), body.as_bytes().to_vec());
1482
1483            match cache.get(&key) {
1484                Some(bytes) => {
1485                    debug!("Cache hit on key {:?}", key);
1486                    vec2s7(bytes.to_vec())
1487                }
1488                None => {
1489                    let result = tokio::task::block_in_place(move || {
1490                        tokio::runtime::Handle::current().block_on(async move {
1491                            match method.to_lowercase() {
1492                                method if method == "get" => {
1493                                    JOURNAL
1494                                        .client
1495                                        .get(&url[1..url.len() - 1])
1496                                        .send()
1497                                        .await?
1498                                        .bytes()
1499                                        .await
1500                                }
1501                                method if method == "post" => {
1502                                    JOURNAL
1503                                        .client
1504                                        .post(&url[1..url.len() - 1])
1505                                        .body(String::from(&body[1..body.len() - 1]))
1506                                        .send()
1507                                        .await?
1508                                        .bytes()
1509                                        .await
1510                                }
1511                                _ => {
1512                                    panic!("Unsupported HTTP method")
1513                                }
1514                            }
1515                        })
1516                    });
1517
1518                    match result {
1519                        Ok(vector) => {
1520                            cache.insert(key, vector.to_vec());
1521                            vec2s7(vector.to_vec())
1522                        }
1523                        Err(_) => {
1524                            sync_error(sc, "Journal is unable to fulfill HTTP request (sync-http)")
1525                        }
1526                    }
1527                }
1528            }
1529        }
1530    }
1531
1532    Primitive::new(
1533        code,
1534        c"sync-http",
1535        c"(sync-http method url . data) make an http request where method is 'get or 'post",
1536        2,
1537        2,
1538        false,
1539    )
1540}
1541
1542fn primitive_s7_sync_remote() -> Primitive {
1543    unsafe extern "C" fn code(sc: *mut s7::s7_scheme, args: s7::s7_pointer) -> s7::s7_pointer {
1544        unsafe {
1545            mark_external_called(sc);
1546
1547            let vec2s7 = |mut vector: Vec<u8>| {
1548                vector.insert(0, 39); // add quote character so that it evaluates correctly
1549                vector.push(0);
1550                let c_string = CString::from_vec_with_nul(vector)
1551                    .expect("Failed to create C string from vector");
1552                s7::s7_eval_c_string(sc, c_string.as_ptr())
1553            };
1554
1555            let url = obj2str(sc, s7::s7_car(args));
1556
1557            let body = obj2str(sc, s7::s7_cadr(args));
1558
1559            let cache_mutex = {
1560                let session = SESSIONS.read().expect("Failed to acquire session lock");
1561                session
1562                    .get(&(sc as usize))
1563                    .expect("Failed to get session from map")
1564                    .cache
1565                    .clone()
1566            };
1567
1568            let mut cache = cache_mutex.lock().expect("Failed to acquire cache lock");
1569
1570            let key = (String::from("post"), url.clone(), body.as_bytes().to_vec());
1571
1572            match cache.get(&key) {
1573                Some(bytes) => {
1574                    debug!("Cache hit on key {:?}", key);
1575                    vec2s7(bytes.to_vec())
1576                }
1577                None => {
1578                    let result = tokio::task::block_in_place(move || {
1579                        tokio::runtime::Handle::current().block_on(async move {
1580                            JOURNAL
1581                                .client
1582                                .post(&url[1..url.len() - 1])
1583                                .body(body)
1584                                .send()
1585                                .await?
1586                                .bytes()
1587                                .await
1588                        })
1589                    });
1590
1591                    match result {
1592                        Ok(bytes) => {
1593                            cache.insert(key, bytes.to_vec());
1594                            vec2s7(bytes.to_vec())
1595                        }
1596                        Err(_) => {
1597                            sync_error(sc, "Journal is unable to query remote peer (sync-remote)")
1598                        }
1599                    }
1600                }
1601            }
1602        }
1603    }
1604
1605    Primitive::new(
1606        code,
1607        c"sync-remote",
1608        c"(sync-remote url data) make a post http request with the data payload)",
1609        2,
1610        0,
1611        false,
1612    )
1613}
1614unsafe fn string_to_s7(sc: *mut s7::s7_scheme, string: &str) -> s7::s7_pointer {
1615    unsafe {
1616        let c_string = CString::new(string).expect("Failed to create CString from string");
1617        let s7_string = s7::s7_make_string(sc, c_string.as_ptr());
1618        s7::s7_object_to_string(sc, s7_string, false)
1619    }
1620}
1621
1622unsafe fn sync_heap_make(word: Word) -> *mut libc::c_void {
1623    unsafe {
1624        let ptr = libc::malloc(SIZE);
1625        let array: &mut [u8] = std::slice::from_raw_parts_mut(ptr as *mut u8, SIZE);
1626        for i in 0..SIZE {
1627            array[i] = word[i] as u8;
1628        }
1629        ptr
1630    }
1631}
1632
1633pub(crate) unsafe fn sync_heap_read(ptr: *mut libc::c_void) -> Word {
1634    unsafe {
1635        std::slice::from_raw_parts_mut(ptr as *mut u8, SIZE)
1636            .try_into()
1637            .expect("Failed to convert slice to Word array")
1638    }
1639}
1640
1641unsafe fn sync_heap_free(ptr: *mut libc::c_void) {
1642    unsafe {
1643        libc::free(ptr);
1644    }
1645}
1646
1647pub(crate) unsafe fn sync_is_node(obj: s7::s7_pointer) -> bool {
1648    unsafe { s7::s7_is_c_object(obj) && s7::s7_c_object_type(obj) == SYNC_NODE_TAG }
1649}
1650
1651unsafe fn sync_cxr(
1652    sc: *mut s7::s7_scheme,
1653    args: s7::s7_pointer,
1654    name: &CStr,
1655    selector: fn((Word, Word)) -> Word,
1656) -> s7::s7_pointer {
1657    unsafe {
1658        let node = s7::s7_car(args);
1659        let word = sync_heap_read(s7::s7_c_object_value(node));
1660        let (persistor, overlay) = session_storage_for(sc);
1661
1662        let child_return = |word| {
1663            let node_return = |word| s7::s7_make_c_object(sc, SYNC_NODE_TAG, sync_heap_make(word));
1664
1665            let vector_return = |vector: Vec<u8>| {
1666                let bv = s7::s7_make_byte_vector(sc, vector.len() as i64, 1, std::ptr::null_mut());
1667                for i in 0..vector.len() {
1668                    s7::s7_byte_vector_set(bv, i as i64, vector[i]);
1669                }
1670                bv
1671            };
1672
1673            if word == NULL {
1674                return node_return(word);
1675            }
1676
1677            match resolve_node_with(&persistor, overlay.as_ref(), word) {
1678                Some(ResolvedNode::Branch((left, right, digest), ResolveSource::Global)) => {
1679                    persistor
1680                        .branch_set(left, right, digest)
1681                        .expect("Failed to add branch to session persistor");
1682                    node_return(word)
1683                }
1684                Some(ResolvedNode::Branch(_, _)) => node_return(word),
1685                Some(ResolvedNode::Leaf(content, ResolveSource::Global)) => {
1686                    persistor
1687                        .leaf_set(content.clone())
1688                        .expect("Failed to add leaf to session persistor");
1689                    vector_return(content)
1690                }
1691                Some(ResolvedNode::Leaf(content, _)) => vector_return(content),
1692                Some(ResolvedNode::Stump(digest, ResolveSource::Global)) => {
1693                    persistor
1694                        .stump_set(digest)
1695                        .expect("Failed to add stump to session persistor");
1696                    node_return(word)
1697                }
1698                Some(ResolvedNode::Stump(_, _)) => node_return(word),
1699                None => sync_error(
1700                    sc,
1701                    format!(
1702                        "Cannot retrieve items for node that is not a sync-pair ({})",
1703                        name.to_string_lossy()
1704                    )
1705                    .as_str(),
1706                ),
1707            }
1708        };
1709
1710        match sync_is_node(node) {
1711            true => match resolve_branch_with(&persistor, overlay.as_ref(), word) {
1712                Some(((left, right, _), _)) => child_return(selector((left, right))),
1713                None => sync_error(
1714                    sc,
1715                    format!(
1716                        "Journal cannot retrieve leaf byte-vector ({})",
1717                        name.to_string_lossy()
1718                    )
1719                    .as_str(),
1720                ),
1721            },
1722            false => {
1723                s7::s7_wrong_type_arg_error(sc, name.as_ptr(), 1, node, c"a sync-node".as_ptr())
1724            }
1725        }
1726    }
1727}
1728
1729unsafe fn sync_digest(sc: *mut s7::s7_scheme, word: Word) -> Result<Word, String> {
1730    let (persistor, overlay) = session_storage_for(sc);
1731
1732    if word == NULL {
1733        Ok(NULL)
1734    } else {
1735        match resolve_node_with(&persistor, overlay.as_ref(), word) {
1736            Some(ResolvedNode::Branch((_, _, digest), _)) => Ok(digest),
1737            Some(ResolvedNode::Leaf(_, _)) => Ok(word),
1738            Some(ResolvedNode::Stump(digest, _)) => Ok(digest),
1739            None => Err("Digest not found in persistor".to_string()),
1740        }
1741    }
1742}
1743
1744unsafe fn sync_branch_children(sc: *mut s7::s7_scheme, word: Word) -> Result<(Word, Word), String> {
1745    let (persistor, overlay) = session_storage_for(sc);
1746
1747    if let Some(((left, right, _), _)) = resolve_branch_with(&persistor, overlay.as_ref(), word) {
1748        Ok((left, right))
1749    } else {
1750        Err("Node is not a sync-pair".to_string())
1751    }
1752}
1753
1754pub(crate) fn session_persistor_for(sc: *mut s7::s7_scheme) -> MemoryPersistor {
1755    let session = SESSIONS.read().expect("Failed to acquire SESSIONS lock");
1756    session
1757        .get(&(sc as usize))
1758        .expect("Session not found for given context")
1759        .persistor
1760        .clone()
1761}
1762
1763fn session_storage_for(sc: *mut s7::s7_scheme) -> (MemoryPersistor, Option<MemoryPersistor>) {
1764    let session = SESSIONS.read().expect("Failed to acquire SESSIONS lock");
1765    let session = session
1766        .get(&(sc as usize))
1767        .expect("Session not found for given context");
1768    (
1769        session.persistor.clone(),
1770        session.strict_overlay_persistor.clone(),
1771    )
1772}
1773
1774unsafe fn strict_sync_eval_env(sc: *mut s7::s7_scheme) -> s7::s7_pointer {
1775    unsafe {
1776        let unsafe_names = [
1777            c"curlet",
1778            c"cutlet",
1779            c"funclet",
1780            c"inlet",
1781            c"load",
1782            c"open-input-string",
1783            c"openlet",
1784            c"owlet",
1785            c"outlet",
1786            c"read",
1787            c"varlet",
1788            c"sync-all",
1789            c"sync-call",
1790            c"sync-create",
1791            c"sync-delete",
1792            c"sync-state",
1793            c"sync-http",
1794            c"sync-remote",
1795            c"random-byte-vector",
1796        ];
1797        let mut form = String::from("(let ((e (sublet (rootlet))))");
1798        for name in unsafe_names {
1799            let symbol = name.to_string_lossy();
1800            form.push_str(&format!(
1801                " (varlet e '{0} (lambda args (error 'unsafe-error \"{0} unavailable in strict sync-eval\")))",
1802                symbol
1803            ));
1804        }
1805        form.push_str(" e)");
1806        let c_form = CString::new(form).expect("Failed to build strict sync-eval environment form");
1807        s7::s7_eval_c_string(sc, c_form.as_ptr())
1808    }
1809}
1810
1811unsafe fn strict_sync_eval_env_cached(sc: *mut s7::s7_scheme) -> s7::s7_pointer {
1812    unsafe {
1813        if let Some(loc) = {
1814            let sessions = SESSIONS.read().expect("Failed to acquire sessions lock");
1815            sessions
1816                .get(&(sc as usize))
1817                .and_then(|session| session.strict_env_loc)
1818        } {
1819            return s7::s7_gc_protected_at(sc, loc);
1820        }
1821
1822        let env = strict_sync_eval_env(sc);
1823        let loc = s7::s7_gc_protect(sc, env);
1824
1825        let existing = {
1826            let mut sessions = SESSIONS.write().expect("Failed to acquire sessions lock");
1827            let session = sessions
1828                .get_mut(&(sc as usize))
1829                .expect("Session not found for strict env caching");
1830            match session.strict_env_loc {
1831                Some(existing) => Some(existing),
1832                None => {
1833                    session.strict_env_loc = Some(loc);
1834                    None
1835                }
1836            }
1837        };
1838
1839        if let Some(existing) = existing {
1840            s7::s7_gc_unprotect_at(sc, loc);
1841            s7::s7_gc_protected_at(sc, existing)
1842        } else {
1843            env
1844        }
1845    }
1846}
1847
1848unsafe fn strict_loader_lookup(sc: *mut s7::s7_scheme, header_word: Word) -> Option<s7::s7_pointer> {
1849    unsafe {
1850        let loc = {
1851            let sessions = SESSIONS.read().expect("Failed to acquire sessions lock");
1852            sessions
1853                .get(&(sc as usize))
1854                .and_then(|session| session.strict_loader_locs.get(&header_word).copied())
1855        }?;
1856        Some(s7::s7_gc_protected_at(sc, loc))
1857    }
1858}
1859
1860unsafe fn strict_loader_cache_store(
1861    sc: *mut s7::s7_scheme,
1862    header_word: Word,
1863    loader: s7::s7_pointer,
1864) -> s7::s7_pointer {
1865    unsafe {
1866        let loc = s7::s7_gc_protect(sc, loader);
1867        let existing = {
1868            let mut sessions = SESSIONS.write().expect("Failed to acquire sessions lock");
1869            let session = sessions
1870                .get_mut(&(sc as usize))
1871                .expect("Session not found for strict loader caching");
1872            session.strict_loader_locs.insert(header_word, loc)
1873        };
1874
1875        if let Some(existing) = existing {
1876            s7::s7_gc_unprotect_at(sc, loc);
1877            s7::s7_gc_protected_at(sc, existing)
1878        } else {
1879            loader
1880        }
1881    }
1882}
journal_sdk/lib.rs

journal_sdk/
lib.rs
