// =============================================================================
// #######
// ### ### F: objects.rs
// ## ## ## ## P: AppCore-Runtime
// ## ##
// C: 2026/10/12 00:00:00 by dnettoRaw
// ## ## ## ## U: 2026/10/12 00:00:00 by dnettoRaw
// ########### S: 1.0.2-rc
// =============================================================================
//! Bounded canonical opaque objects sharing the provider's transaction and database.
use crate::{SqliteSyncError as E, SqliteSyncResult as R, SqliteSyncStore};
use rusqlite::{params, Connection, OptionalExtension};
use sha2::{Digest, Sha256};
pub(crate) const MAX_OBJECTS: u64 = 100_000;
pub(crate) const MAX_CHANGES: u64 = 1_000_000;
pub(crate) const MAX_BYTES: usize = 1024 * 1024;
/// Opaque provider-owned record. Metadata is opaque too; callers own its confidentiality.
#[derive(Clone, PartialEq, Eq)]
pub struct OpaqueObjectV1 {
/// Installation scope, validated as a bounded identifier.
pub installation_id: String,
/// Opaque key, never a filesystem path interpreted by this provider.
pub object_key: String,
/// Encrypted/opaque envelope, at most 1 MiB; empty only for a tombstone.
pub envelope: Vec<u8>,
/// Opaque routing metadata, at most 4 KiB. Never put secrets here.
pub metadata: Vec<u8>,
/// A deletion preserves identity and revision, with an empty envelope.
pub tombstone: bool,
}
impl std::fmt::Debug for OpaqueObjectV1 {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OpaqueObjectV1")
.field("envelope_bytes", &self.envelope.len())
.field("metadata_bytes", &self.metadata.len())
.field("tombstone", &self.tombstone)
.finish_non_exhaustive()
}
}
impl OpaqueObjectV1 {
/// Validates bounded identities and bytes, without interpreting or authenticating content.
pub fn validate(&self) -> R<()> {
validate_id(&self.installation_id)?;
validate_id(&self.object_key)?;
if self.metadata.len() > 4096
|| self.envelope.len() > MAX_BYTES
|| self.tombstone != self.envelope.is_empty()
{
return Err(E::CapacityExceeded("opaque object bytes"));
}
Ok(())
}
}
/// Compare-and-set precondition; no implicit last-writer-wins mutation exists.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ObjectRevisionV1 {
/// The key must never have existed, including as a tombstone.
Absent,
/// The key must have this exact durable revision.
Exact(u64),
}
/// A record and its technical revision; digests are integrity checks, not authentication.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OpaqueObjectRecordV1 {
/// Caller-owned opaque object.
pub object: OpaqueObjectV1,
/// Monotonically increasing provider revision.
pub revision: u64,
/// SHA-256 of the opaque envelope.
pub envelope_hash: String,
}
/// Stable database identity plus its current object revision.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ObjectHeadV1 {
/// Persistent generation; a replaced database requires a fresh reader projection.
pub generation: String,
/// Global committed object revision, zero for an empty object store.
pub revision: u64,
}
/// Durable result of an object mutation or exact idempotent replay.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ObjectWriteResultV1 {
/// The original commit revision, also returned on replay.
pub revision: u64,
/// True only when this call created a new revision.
pub changed: bool,
}
/// Canonical opaque objects, independent of application schemas and filesystem publication.
#[derive(Clone)]
pub struct SqliteOpaqueObjectStoreV1(pub(crate) SqliteSyncStore);
impl SqliteSyncStore {
/// Returns a handle to canonical objects in the same database as the sync state.
pub fn opaque_objects_v1(&self) -> SqliteOpaqueObjectStoreV1 {
SqliteOpaqueObjectStoreV1(self.clone())
}
}
impl SqliteOpaqueObjectStoreV1 {
/// Reads generation/revision without loading any envelope or altering the database.
pub fn head(&self) -> R<ObjectHeadV1> {
self.0.with_connection(|connection| head(connection))
}
/// Reads and hash-checks one bounded current record; absent is distinct from deleted.
pub fn get(&self, installation: &str, key: &str) -> R<Option<OpaqueObjectRecordV1>> {
validate_id(installation)?;
validate_id(key)?;
self.0
.with_connection(|connection| read_object(connection, installation, key))
}
}
pub(crate) fn head(connection: &Connection) -> R<ObjectHeadV1> {
connection
.query_row(
"SELECT generation,revision FROM appcore_object_head WHERE singleton=1",
[],
|row| {
Ok(ObjectHeadV1 {
generation: row.get(0)?,
revision: row.get(1)?,
})
},
)
.map_err(E::database)
}
pub(crate) fn read_object(
connection: &Connection,
installation: &str,
key: &str,
) -> R<Option<OpaqueObjectRecordV1>> {
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)?;
let mut rows = statement
.query(params![installation, key])
.map_err(E::database)?;
let Some(row) = rows.next().map_err(E::database)? else {
return Ok(None);
};
let length: usize = row.get(2).map_err(E::database)?;
let metadata_length: usize = row.get(3).map_err(E::database)?;
if length > MAX_BYTES || metadata_length > 4096 {
return Err(E::IntegrityFailed);
}
let object = OpaqueObjectV1 {
installation_id: installation.into(),
object_key: key.into(),
envelope: row.get(5).map_err(E::database)?,
metadata: row.get(6).map_err(E::database)?,
tombstone: row.get(1).map_err(E::database)?,
};
object.validate().map_err(|_| E::IntegrityFailed)?;
let hash: String = row.get(4).map_err(E::database)?;
if hash != digest(&object.envelope) {
return Err(E::IntegrityFailed);
}
Ok(Some(OpaqueObjectRecordV1 {
object,
revision: row.get(0).map_err(E::database)?,
envelope_hash: hash,
}))
}
pub(crate) fn validate_id(value: &str) -> R<()> {
if value.is_empty()
|| value.len() > 128
|| !value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"-_.:".contains(&byte))
{
return Err(E::CorruptRecord("opaque object identity"));
}
Ok(())
}
pub(crate) fn digest(bytes: &[u8]) -> String {
Sha256::digest(bytes)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
pub(crate) fn fingerprint(object: &OpaqueObjectV1) -> String {
let mut hash = Sha256::new();
for part in [
object.installation_id.as_bytes(),
object.object_key.as_bytes(),
object.metadata.as_slice(),
object.envelope.as_slice(),
] {
hash.update((part.len() as u64).to_be_bytes());
hash.update(part);
}
hash.update([u8::from(object.tombstone)]);
hash.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
pub(crate) fn mutate(
connection: &Connection,
object: &OpaqueObjectV1,
mutation_id: &str,
expected: ObjectRevisionV1,
) -> R<ObjectWriteResultV1> {
object.validate()?;
validate_id(mutation_id)?;
let fingerprint = fingerprint(object);
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)?;
if let Some((revision, hash)) = replay {
if hash != fingerprint {
return Err(E::CorruptRecord("opaque mutation replay"));
}
return Ok(ObjectWriteResultV1 {
revision,
changed: false,
});
}
let existing: Option<u64> = connection
.query_row(
"SELECT revision FROM appcore_opaque_object WHERE installation_id=?1 AND object_key=?2",
params![object.installation_id, object.object_key],
|row| row.get(0),
)
.optional()
.map_err(E::database)?;
if existing
!= match expected {
ObjectRevisionV1::Absent => None,
ObjectRevisionV1::Exact(revision) => Some(revision),
}
{
return Err(E::CorruptRecord("opaque object precondition"));
}
let (head, count, changes): (u64, u64, u64) = connection
.query_row(
"SELECT revision,object_count,change_count FROM appcore_object_head WHERE singleton=1",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.map_err(E::database)?;
if existing.is_none() && count >= MAX_OBJECTS || changes >= MAX_CHANGES {
return Err(E::CapacityExceeded("opaque objects or changes"));
}
let revision = head
.checked_add(1)
.filter(|value| *value <= i64::MAX as u64)
.ok_or(E::CapacityExceeded("object revision"))?;
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)?;
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)?;
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)?;
Ok(ObjectWriteResultV1 {
revision,
changed: true,
})
}