appcore_sync_sqlite/
objects.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: objects.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//! Bounded canonical opaque objects sharing the provider's transaction and database.
12use crate::{SqliteSyncError as E, SqliteSyncResult as R, SqliteSyncStore};
13use rusqlite::{params, Connection, OptionalExtension};
14use sha2::{Digest, Sha256};
15pub(crate) const MAX_OBJECTS: u64 = 100_000;
16pub(crate) const MAX_CHANGES: u64 = 1_000_000;
17pub(crate) const MAX_BYTES: usize = 1024 * 1024;
18/// Opaque provider-owned record. Metadata is opaque too; callers own its confidentiality.
19#[derive(Clone, PartialEq, Eq)]
20pub struct OpaqueObjectV1 {
21    /// Installation scope, validated as a bounded identifier.
22    pub installation_id: String,
23    /// Opaque key, never a filesystem path interpreted by this provider.
24    pub object_key: String,
25    /// Encrypted/opaque envelope, at most 1 MiB; empty only for a tombstone.
26    pub envelope: Vec<u8>,
27    /// Opaque routing metadata, at most 4 KiB. Never put secrets here.
28    pub metadata: Vec<u8>,
29    /// A deletion preserves identity and revision, with an empty envelope.
30    pub tombstone: bool,
31}
32impl std::fmt::Debug for OpaqueObjectV1 {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        f.debug_struct("OpaqueObjectV1")
35            .field("envelope_bytes", &self.envelope.len())
36            .field("metadata_bytes", &self.metadata.len())
37            .field("tombstone", &self.tombstone)
38            .finish_non_exhaustive()
39    }
40}
41impl OpaqueObjectV1 {
42    /// Validates bounded identities and bytes, without interpreting or authenticating content.
43    pub fn validate(&self) -> R<()> {
44        validate_id(&self.installation_id)?;
45        validate_id(&self.object_key)?;
46        if self.metadata.len() > 4096
47            || self.envelope.len() > MAX_BYTES
48            || self.tombstone != self.envelope.is_empty()
49        {
50            return Err(E::CapacityExceeded("opaque object bytes"));
51        }
52        Ok(())
53    }
54}
55/// Compare-and-set precondition; no implicit last-writer-wins mutation exists.
56#[derive(Clone, Copy, Debug, PartialEq, Eq)]
57pub enum ObjectRevisionV1 {
58    /// The key must never have existed, including as a tombstone.
59    Absent,
60    /// The key must have this exact durable revision.
61    Exact(u64),
62}
63/// A record and its technical revision; digests are integrity checks, not authentication.
64#[derive(Clone, Debug, PartialEq, Eq)]
65pub struct OpaqueObjectRecordV1 {
66    /// Caller-owned opaque object.
67    pub object: OpaqueObjectV1,
68    /// Monotonically increasing provider revision.
69    pub revision: u64,
70    /// SHA-256 of the opaque envelope.
71    pub envelope_hash: String,
72}
73/// Stable database identity plus its current object revision.
74#[derive(Clone, Debug, PartialEq, Eq)]
75pub struct ObjectHeadV1 {
76    /// Persistent generation; a replaced database requires a fresh reader projection.
77    pub generation: String,
78    /// Global committed object revision, zero for an empty object store.
79    pub revision: u64,
80}
81/// Durable result of an object mutation or exact idempotent replay.
82#[derive(Clone, Copy, Debug, PartialEq, Eq)]
83pub struct ObjectWriteResultV1 {
84    /// The original commit revision, also returned on replay.
85    pub revision: u64,
86    /// True only when this call created a new revision.
87    pub changed: bool,
88}
89/// Canonical opaque objects, independent of application schemas and filesystem publication.
90#[derive(Clone)]
91pub struct SqliteOpaqueObjectStoreV1(pub(crate) SqliteSyncStore);
92impl SqliteSyncStore {
93    /// Returns a handle to canonical objects in the same database as the sync state.
94    pub fn opaque_objects_v1(&self) -> SqliteOpaqueObjectStoreV1 {
95        SqliteOpaqueObjectStoreV1(self.clone())
96    }
97}
98impl SqliteOpaqueObjectStoreV1 {
99    /// Reads generation/revision without loading any envelope or altering the database.
100    pub fn head(&self) -> R<ObjectHeadV1> {
101        self.0.with_connection(|connection| head(connection))
102    }
103    /// Reads and hash-checks one bounded current record; absent is distinct from deleted.
104    pub fn get(&self, installation: &str, key: &str) -> R<Option<OpaqueObjectRecordV1>> {
105        validate_id(installation)?;
106        validate_id(key)?;
107        self.0
108            .with_connection(|connection| read_object(connection, installation, key))
109    }
110}
111pub(crate) fn head(connection: &Connection) -> R<ObjectHeadV1> {
112    connection
113        .query_row(
114            "SELECT generation,revision FROM appcore_object_head WHERE singleton=1",
115            [],
116            |row| {
117                Ok(ObjectHeadV1 {
118                    generation: row.get(0)?,
119                    revision: row.get(1)?,
120                })
121            },
122        )
123        .map_err(E::database)
124}
125pub(crate) fn read_object(
126    connection: &Connection,
127    installation: &str,
128    key: &str,
129) -> R<Option<OpaqueObjectRecordV1>> {
130    let mut statement=connection.prepare("SELECT revision,tombstone,length(envelope),length(metadata),envelope_hash,envelope,metadata FROM appcore_opaque_object WHERE installation_id=?1 AND object_key=?2").map_err(E::database)?;
131    let mut rows = statement
132        .query(params![installation, key])
133        .map_err(E::database)?;
134    let Some(row) = rows.next().map_err(E::database)? else {
135        return Ok(None);
136    };
137    let length: usize = row.get(2).map_err(E::database)?;
138    let metadata_length: usize = row.get(3).map_err(E::database)?;
139    if length > MAX_BYTES || metadata_length > 4096 {
140        return Err(E::IntegrityFailed);
141    }
142    let object = OpaqueObjectV1 {
143        installation_id: installation.into(),
144        object_key: key.into(),
145        envelope: row.get(5).map_err(E::database)?,
146        metadata: row.get(6).map_err(E::database)?,
147        tombstone: row.get(1).map_err(E::database)?,
148    };
149    object.validate().map_err(|_| E::IntegrityFailed)?;
150    let hash: String = row.get(4).map_err(E::database)?;
151    if hash != digest(&object.envelope) {
152        return Err(E::IntegrityFailed);
153    }
154    Ok(Some(OpaqueObjectRecordV1 {
155        object,
156        revision: row.get(0).map_err(E::database)?,
157        envelope_hash: hash,
158    }))
159}
160pub(crate) fn validate_id(value: &str) -> R<()> {
161    if value.is_empty()
162        || value.len() > 128
163        || !value
164            .bytes()
165            .all(|byte| byte.is_ascii_alphanumeric() || b"-_.:".contains(&byte))
166    {
167        return Err(E::CorruptRecord("opaque object identity"));
168    }
169    Ok(())
170}
171pub(crate) fn digest(bytes: &[u8]) -> String {
172    Sha256::digest(bytes)
173        .iter()
174        .map(|byte| format!("{byte:02x}"))
175        .collect()
176}
177pub(crate) fn fingerprint(object: &OpaqueObjectV1) -> String {
178    let mut hash = Sha256::new();
179    for part in [
180        object.installation_id.as_bytes(),
181        object.object_key.as_bytes(),
182        object.metadata.as_slice(),
183        object.envelope.as_slice(),
184    ] {
185        hash.update((part.len() as u64).to_be_bytes());
186        hash.update(part);
187    }
188    hash.update([u8::from(object.tombstone)]);
189    hash.finalize()
190        .iter()
191        .map(|byte| format!("{byte:02x}"))
192        .collect()
193}
194pub(crate) fn mutate(
195    connection: &Connection,
196    object: &OpaqueObjectV1,
197    mutation_id: &str,
198    expected: ObjectRevisionV1,
199) -> R<ObjectWriteResultV1> {
200    object.validate()?;
201    validate_id(mutation_id)?;
202    let fingerprint = fingerprint(object);
203    let replay:Option<(u64,String)>=connection.query_row("SELECT revision,record_hash FROM appcore_object_change WHERE installation_id=?1 AND mutation_id=?2",params![object.installation_id,mutation_id],|row|Ok((row.get(0)?,row.get(1)?))).optional().map_err(E::database)?;
204    if let Some((revision, hash)) = replay {
205        if hash != fingerprint {
206            return Err(E::CorruptRecord("opaque mutation replay"));
207        }
208        return Ok(ObjectWriteResultV1 {
209            revision,
210            changed: false,
211        });
212    }
213    let existing: Option<u64> = connection
214        .query_row(
215            "SELECT revision FROM appcore_opaque_object WHERE installation_id=?1 AND object_key=?2",
216            params![object.installation_id, object.object_key],
217            |row| row.get(0),
218        )
219        .optional()
220        .map_err(E::database)?;
221    if existing
222        != match expected {
223            ObjectRevisionV1::Absent => None,
224            ObjectRevisionV1::Exact(revision) => Some(revision),
225        }
226    {
227        return Err(E::CorruptRecord("opaque object precondition"));
228    }
229    let (head, count, changes): (u64, u64, u64) = connection
230        .query_row(
231            "SELECT revision,object_count,change_count FROM appcore_object_head WHERE singleton=1",
232            [],
233            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
234        )
235        .map_err(E::database)?;
236    if existing.is_none() && count >= MAX_OBJECTS || changes >= MAX_CHANGES {
237        return Err(E::CapacityExceeded("opaque objects or changes"));
238    }
239    let revision = head
240        .checked_add(1)
241        .filter(|value| *value <= i64::MAX as u64)
242        .ok_or(E::CapacityExceeded("object revision"))?;
243    connection.execute("INSERT INTO appcore_opaque_object(installation_id,object_key,revision,tombstone,envelope,metadata,envelope_hash) VALUES(?1,?2,?3,?4,?5,?6,?7) ON CONFLICT(installation_id,object_key) DO UPDATE SET revision=excluded.revision,tombstone=excluded.tombstone,envelope=excluded.envelope,metadata=excluded.metadata,envelope_hash=excluded.envelope_hash",params![object.installation_id,object.object_key,revision,object.tombstone,object.envelope,object.metadata,digest(&object.envelope)]).map_err(E::database)?;
244    connection.execute("INSERT INTO appcore_object_change(revision,installation_id,object_key,mutation_id,record_hash) VALUES(?1,?2,?3,?4,?5)",params![revision,object.installation_id,object.object_key,mutation_id,fingerprint]).map_err(E::database)?;
245    connection.execute("UPDATE appcore_object_head SET revision=?1,object_count=object_count+?2,change_count=change_count+1 WHERE singleton=1",params![revision,u8::from(existing.is_none())]).map_err(E::database)?;
246    Ok(ObjectWriteResultV1 {
247        revision,
248        changed: true,
249    })
250}
appcore_sync_sqlite/objects.rs

appcore_sync_sqlite/
objects.rs
