appcore_sync_sqlite/
materialization.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: materialization.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/10/12 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/10/12 00:00:00 by dnettoRaw
8//      ###########      S: 1.0.2-rc
9// =============================================================================
10
11//! `SQLite` write-ahead intents for recoverable opaque-object materialization.
12//! This is not an atomic filesystem/SQLite transaction. Deployment maps opaque
13//! keys to safe paths and syncs/revalidates bytes before acknowledging the intent.
14use crate::{SqliteSyncError, SqliteSyncResult, SqliteSyncStore};
15use rusqlite::{params, Connection, OptionalExtension};
16use sha2::{Digest, Sha256};
17
18const MAX_INTENTS: usize = 4_096;
19const MAX_PAYLOAD_BYTES: usize = 1024 * 1024;
20/// Bounded immutable materialization intent, scoped to an installation.
21#[derive(Clone, PartialEq, Eq)]
22pub struct ObjectWriteIntentV1 {
23    /// Opaque idempotency identity, 1..=128 safe identifier bytes.
24    pub intent_id: String,
25    /// Installation allowed to replay this intent.
26    pub installation_id: String,
27    /// Opaque tenant-scoped object key; never an unchecked filesystem path.
28    pub object_key: String,
29    /// Opaque envelope bytes, at most 1 MiB. Deployment authenticates them.
30    pub envelope: Vec<u8>,
31}
32impl std::fmt::Debug for ObjectWriteIntentV1 {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        f.debug_struct("ObjectWriteIntentV1")
35            .field("envelope_bytes", &self.envelope.len())
36            .finish_non_exhaustive()
37    }
38}
39impl ObjectWriteIntentV1 {
40    /// Checks identity and payload limits without interpreting document content.
41    pub fn validate(&self) -> SqliteSyncResult<()> {
42        for id in [&self.intent_id, &self.installation_id, &self.object_key] {
43            validate_id(id)?;
44        }
45        if self.envelope.is_empty() || self.envelope.len() > MAX_PAYLOAD_BYTES {
46            return Err(SqliteSyncError::CapacityExceeded("object envelope"));
47        }
48        Ok(())
49    }
50    /// Digest of opaque envelope bytes. Prefer encrypted envelopes; never export
51    /// predictable plaintext digests as diagnostics or treat them as authentication.
52    pub fn envelope_sha256(&self) -> String {
53        digest(&self.envelope)
54    }
55}
56impl SqliteSyncStore {
57    /// Returns a FIFO bounded page of outstanding intents for one installation.
58    /// Materialize idempotently in this order under application-owned object locks.
59    pub fn pending_object_writes(
60        &self,
61        installation: &str,
62        limit: usize,
63        max_bytes: usize,
64    ) -> SqliteSyncResult<Vec<ObjectWriteIntentV1>> {
65        validate_id(installation)?;
66        if limit > 1_024 || max_bytes > 16 * 1024 * 1024 {
67            return Err(SqliteSyncError::CapacityExceeded("object page"));
68        }
69        self.with_connection(|connection| {
70            let transaction = connection.transaction().map_err(SqliteSyncError::database)?;
71            let mut statement = transaction.prepare(
72                "SELECT intent_id, object_key, length(envelope) FROM appcore_object_write_intent WHERE installation_id=?1 ORDER BY position LIMIT ?2"
73            ).map_err(SqliteSyncError::database)?;
74            let rows = statement.query_map(params![installation, limit as i64], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, i64>(2)?)))
75                .map_err(SqliteSyncError::database)?;
76            let mut result = Vec::new();
77            let mut bytes = 0usize;
78            for row in rows {
79                let (id, key, length) = row.map_err(SqliteSyncError::database)?;
80                let length = usize::try_from(length).map_err(|_| SqliteSyncError::IntegrityFailed)?;
81                if length > MAX_PAYLOAD_BYTES { return Err(SqliteSyncError::IntegrityFailed); }
82                if bytes.checked_add(length).is_none_or(|value| value > max_bytes) { break; }
83                let (envelope, hash): (Vec<u8>, String) = transaction.query_row(
84                    "SELECT envelope, envelope_hash FROM appcore_object_write_intent WHERE intent_id=?1", [&id], |row| Ok((row.get(0)?, row.get(1)?))
85                ).map_err(SqliteSyncError::database)?;
86                if digest(&envelope) != hash { return Err(SqliteSyncError::IntegrityFailed); }
87                let intent = ObjectWriteIntentV1 { intent_id: id, installation_id: installation.into(), object_key: key, envelope };
88                intent.validate()?;
89                bytes += length;
90                result.push(intent);
91            }
92            Ok(result)
93        })
94    }
95    /// Removes only the installation's oldest intent after the caller has synced
96    /// and verified the actual materialized envelope. Same receipt replay returns
97    /// false after removal; an ID must never be reused once an intent is completed.
98    /// This records caller evidence, not an independent filesystem verification.
99    pub fn acknowledge_object_write(
100        &self,
101        installation: &str,
102        id: &str,
103        observed_sha256: &str,
104    ) -> SqliteSyncResult<bool> {
105        validate_id(installation)?;
106        validate_id(id)?;
107        self.with_connection(|connection| {
108            let transaction = connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate).map_err(SqliteSyncError::database)?;
109            let owner: Option<String> = transaction.query_row(
110                "SELECT installation_id FROM appcore_object_write_intent WHERE intent_id=?1", [id], |row| row.get(0)
111            ).optional().map_err(SqliteSyncError::database)?;
112            let Some(owner) = owner else { return Ok(false); };
113            if owner != installation { return Err(SqliteSyncError::CorruptRecord("object installation")); }
114            let existing: Option<(String, String)> = transaction.query_row(
115                "SELECT intent_id, envelope_hash FROM appcore_object_write_intent WHERE installation_id=?1 ORDER BY position LIMIT 1", [installation],
116                |row| Ok((row.get(0)?, row.get(1)?))
117            ).optional().map_err(SqliteSyncError::database)?;
118            let Some((front, hash)) = existing else { return Ok(false); };
119            if front != id || hash != observed_sha256 { return Err(SqliteSyncError::CorruptRecord("object receipt")); }
120            transaction.execute("DELETE FROM appcore_object_write_intent WHERE intent_id=?1 AND installation_id=?2", params![id, installation]).map_err(SqliteSyncError::database)?;
121            transaction.commit().map_err(SqliteSyncError::database)?;
122            Ok(true)
123        })
124    }
125}
126pub(crate) fn stage(
127    connection: &Connection,
128    intent: &ObjectWriteIntentV1,
129) -> SqliteSyncResult<bool> {
130    intent.validate()?;
131    let existing: Option<(String, String, Vec<u8>)> = connection.query_row(
132        "SELECT installation_id, object_key, CASE WHEN length(envelope) BETWEEN 1 AND 1048576 THEN envelope ELSE NULL END FROM appcore_object_write_intent WHERE intent_id=?1", [&intent.intent_id],
133        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))
134    ).optional().map_err(SqliteSyncError::database)?;
135    if let Some((installation, key, bytes)) = existing {
136        if installation != intent.installation_id
137            || key != intent.object_key
138            || bytes != intent.envelope
139        {
140            return Err(SqliteSyncError::CorruptRecord("object intent"));
141        }
142        return Ok(false);
143    }
144    let count: i64 = connection
145        .query_row(
146            "SELECT COUNT(*) FROM appcore_object_write_intent",
147            [],
148            |row| row.get(0),
149        )
150        .map_err(SqliteSyncError::database)?;
151    if count >= MAX_INTENTS as i64 {
152        return Err(SqliteSyncError::CapacityExceeded("object intent"));
153    }
154    connection.execute("INSERT INTO appcore_object_write_intent(intent_id, installation_id, object_key, envelope, envelope_hash) VALUES (?1, ?2, ?3, ?4, ?5)",
155        params![intent.intent_id, intent.installation_id, intent.object_key, intent.envelope, digest(&intent.envelope)]).map_err(SqliteSyncError::database)?;
156    Ok(true)
157}
158pub(crate) fn validate_records(connection: &Connection) -> SqliteSyncResult<()> {
159    let mut statement = connection.prepare("SELECT intent_id, installation_id, object_key, CASE WHEN length(envelope) BETWEEN 1 AND 1048576 THEN envelope ELSE NULL END, envelope_hash FROM appcore_object_write_intent ORDER BY position LIMIT 4097").map_err(SqliteSyncError::database)?;
160    let rows = statement
161        .query_map([], |row| {
162            Ok((
163                ObjectWriteIntentV1 {
164                    intent_id: row.get(0)?,
165                    installation_id: row.get(1)?,
166                    object_key: row.get(2)?,
167                    envelope: row.get(3)?,
168                },
169                row.get::<_, String>(4)?,
170            ))
171        })
172        .map_err(SqliteSyncError::database)?;
173    for (index, row) in rows.enumerate() {
174        let (intent, hash) = row.map_err(SqliteSyncError::database)?;
175        if index >= MAX_INTENTS || intent.validate().is_err() || digest(&intent.envelope) != hash {
176            return Err(SqliteSyncError::IntegrityFailed);
177        }
178    }
179    Ok(())
180}
181fn digest(bytes: &[u8]) -> String {
182    Sha256::digest(bytes)
183        .iter()
184        .map(|byte| format!("{byte:02x}"))
185        .collect()
186}
187fn validate_id(id: &str) -> SqliteSyncResult<()> {
188    if id.is_empty()
189        || id.len() > 128
190        || !id
191            .bytes()
192            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':'))
193    {
194        return Err(SqliteSyncError::CorruptRecord("object identity"));
195    }
196    Ok(())
197}
appcore_sync_sqlite/materialization.rs

appcore_sync_sqlite/
materialization.rs
