kmp-adapter-embedded 0.1.10

Embedded edition storage adapters: every KMP kernel port on a single local store file (redb by default, SQLite opt-in)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use kmp_domain::{ContextEventStore, ContextUpdatedEvent, IdempotentOutcome, PortError};

use super::engine::{Key, Table};
use super::serdes::{AggregateRecord, decode, encode};
use super::store::{EmbeddedKernelStore, aggregate_key};

impl ContextEventStore for EmbeddedKernelStore {
    async fn append(
        &self,
        event: ContextUpdatedEvent,
        expected_revision: u64,
    ) -> Result<u64, PortError> {
        self.run(move |store| {
            let mut tx = store.begin_write()?;

            let key = aggregate_key(&event.root_node_id, &event.role);
            let current = match tx.get(Table::Aggregates, Key::Str(&key))? {
                Some(raw) => decode::<AggregateRecord>("aggregate head", &raw)?.revision,
                None => 0,
            };
            if current != expected_revision {
                return Err(PortError::Conflict(format!(
                    "expected revision {expected_revision}, current is {current}"
                )));
            }
            let new_revision = current + 1;

            // Stamp the assigned revision on the stored event so replay
            // derives projections with the same revision the aggregate
            // recorded.
            let mut event = event;
            event.revision = new_revision;

            let aggregate_bytes = encode(
                "aggregate head",
                &AggregateRecord {
                    revision: new_revision,
                    content_hash: event.content_hash.clone(),
                },
            )?;
            tx.insert(Table::Aggregates, Key::Str(&key), &aggregate_bytes)?;

            let next_sequence = tx
                .last_u64(Table::EventLog)?
                .map_or(1, |(sequence, _)| sequence + 1);
            let event_bytes = encode("context event", &event)?;
            tx.insert(Table::EventLog, Key::U64(next_sequence), &event_bytes)?;

            if let Some(idempotency_key) = event.idempotency_key.as_deref() {
                let outcome_bytes = encode(
                    "idempotency outcome",
                    &IdempotentOutcome {
                        revision: new_revision,
                        content_hash: event.content_hash.clone(),
                        logical_digest: event.logical_digest.clone(),
                    },
                )?;
                tx.insert(
                    Table::Idempotency,
                    Key::Str(idempotency_key),
                    &outcome_bytes,
                )?;
            }

            tx.commit()?;
            Ok(new_revision)
        })
        .await
    }

    async fn current_revision(&self, root_node_id: &str, role: &str) -> Result<u64, PortError> {
        let key = aggregate_key(root_node_id, role);
        self.run(move |store| {
            let tx = store.begin_read()?;
            match tx.get(Table::Aggregates, Key::Str(&key))? {
                Some(raw) => Ok(decode::<AggregateRecord>("aggregate head", &raw)?.revision),
                None => Ok(0),
            }
        })
        .await
    }

    async fn current_content_hash(
        &self,
        root_node_id: &str,
        role: &str,
    ) -> Result<Option<String>, PortError> {
        let key = aggregate_key(root_node_id, role);
        self.run(move |store| {
            let tx = store.begin_read()?;
            match tx.get(Table::Aggregates, Key::Str(&key))? {
                Some(raw) => Ok(Some(
                    decode::<AggregateRecord>("aggregate head", &raw)?.content_hash,
                )),
                None => Ok(None),
            }
        })
        .await
    }

    async fn find_by_idempotency_key(
        &self,
        key: &str,
    ) -> Result<Option<IdempotentOutcome>, PortError> {
        let key = key.to_string();
        self.run(move |store| {
            let tx = store.begin_read()?;
            match tx.get(Table::Idempotency, Key::Str(&key))? {
                Some(raw) => Ok(Some(decode("idempotency outcome", &raw)?)),
                None => Ok(None),
            }
        })
        .await
    }
}

impl EmbeddedKernelStore {
    /// The event log, read synchronously. The migration path uses this
    /// before any runtime exists around the source copy.
    pub(crate) fn read_event_log_blocking(&self) -> Result<Vec<ContextUpdatedEvent>, PortError> {
        self.read_event_log()
    }

    /// Reads the full append-only event log in sequence order (audit and
    /// replay surface).
    pub(crate) fn read_event_log(&self) -> Result<Vec<ContextUpdatedEvent>, PortError> {
        let tx = self.begin_read()?;
        tx.scan_u64(Table::EventLog)?
            .into_iter()
            .map(|(_, raw)| decode("context event", &raw))
            .collect()
    }
}