appcore-sync-sqlite 1.0.1-rc.2

// =============================================================================
//        #######
//     ###       ###     F: object_pages.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
// =============================================================================

//! Read-only bounded current-state pages and incremental invalidations.
use crate::objects::{head, read_object, validate_id};
use crate::{
    ObjectHeadV1, OpaqueObjectRecordV1, SqliteOpaqueObjectStoreV1, SqliteSyncError as E,
    SqliteSyncResult as R,
};
use rusqlite::params;
/// Incremental page cursor plus current values, not historical values as of the cursor.
/// Values are observed atomically within this page; a newer object revision may exceed `through`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ObjectChangesPageV1 {
    /// Generation/revision observed in the same read transaction.
    pub head: ObjectHeadV1,
    /// Resume after this change revision. Always advances or reports completion.
    pub cursor: u64,
    /// Whether the requested range has been fully covered.
    pub complete: bool,
    /// Current hash-checked objects for the selected changes, including tombstones.
    pub records: Vec<OpaqueObjectRecordV1>,
}
/// Bounded lexicographic scan of current objects, including tombstones.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ObjectsPageV1 {
    /// Generation/revision observed by this page, not a frozen multi-page snapshot.
    pub head: ObjectHeadV1,
    /// Last admitted key when more records exist; otherwise no continuation.
    pub next_key: Option<String>,
    /// Current hash-checked records. Catch up changes from the head captured before scanning.
    pub records: Vec<OpaqueObjectRecordV1>,
}
impl SqliteOpaqueObjectStoreV1 {
    /// Reads current values for a fixed change range, deduplicated by key within the range.
    /// `generation` must match the prior head; source replacement fails closed.
    /// Metadata is admitted before BLOB reads. Too small a budget fails instead of returning a stalled page.
    pub fn changes(
        &self,
        installation: &str,
        generation: &str,
        after: u64,
        through: u64,
        limit: usize,
        max_bytes: usize,
    ) -> R<ObjectChangesPageV1> {
        validate_page(installation, limit, max_bytes)?;
        self.0.with_connection(|connection| {
            let transaction=connection.transaction().map_err(E::database)?;
            let current=head(&transaction)?;
            if current.generation!=generation || after>through || through>current.revision {return Err(E::CorruptRecord("opaque object precondition"));}
            let mut statement=transaction.prepare("SELECT max(c.revision),c.object_key,length(o.envelope)+length(o.metadata) FROM appcore_object_change c JOIN appcore_opaque_object o ON o.installation_id=c.installation_id AND o.object_key=c.object_key WHERE c.installation_id=?1 AND c.revision>?2 AND c.revision<=?3 GROUP BY c.object_key ORDER BY max(c.revision) LIMIT ?4").map_err(E::database)?;
            let rows=statement.query_map(params![installation,after,through,limit+1],|row|Ok((row.get::<_,u64>(0)?,row.get::<_,String>(1)?,row.get::<_,usize>(2)?))).map_err(E::database)?;
            let selected=rows.collect::<Result<Vec<_>,_>>().map_err(E::database)?;
            let mut bytes=0usize;let mut records=Vec::new();let mut cursor=after;let mut complete=true;
            for (position,key,length) in selected {
                if records.len()==limit || bytes.saturating_add(length)>max_bytes {complete=false;break;}
                bytes+=length;cursor=position;
                records.push(read_object(&transaction,installation,&key)?.ok_or(E::IntegrityFailed)?);
            }
            if !complete && records.is_empty() {return Err(E::CapacityExceeded("object page bytes"));}
            if complete {cursor=through;}
            Ok(ObjectChangesPageV1 {head:current,cursor,complete,records})
        })
    }
    /// Scans current records; callers must catch up changes after the initial scan head.
    pub fn scan(
        &self,
        installation: &str,
        after: Option<&str>,
        limit: usize,
        max_bytes: usize,
    ) -> R<ObjectsPageV1> {
        validate_page(installation, limit, max_bytes)?;
        if let Some(key) = after {
            validate_id(key)?;
        }
        self.0.with_connection(|connection| {
            let transaction=connection.transaction().map_err(E::database)?;
            let current=head(&transaction)?;
            let mut statement=transaction.prepare("SELECT object_key,length(envelope)+length(metadata) FROM appcore_opaque_object WHERE installation_id=?1 AND object_key>?2 ORDER BY object_key LIMIT ?3").map_err(E::database)?;
            let rows=statement.query_map(params![installation,after.unwrap_or(""),limit+1],|row|Ok((row.get::<_,String>(0)?,row.get::<_,usize>(1)?))).map_err(E::database)?;
            let selected=rows.collect::<Result<Vec<_>,_>>().map_err(E::database)?;
            let mut bytes=0usize;let mut records=Vec::new();let mut last=None;let mut more=false;
            for (key,length) in selected {
                if records.len()==limit || bytes.saturating_add(length)>max_bytes {more=true;break;}
                bytes+=length;last=Some(key.clone());records.push(read_object(&transaction,installation,&key)?.ok_or(E::IntegrityFailed)?);
            }
            if more && records.is_empty() {return Err(E::CapacityExceeded("object page bytes"));}
            Ok(ObjectsPageV1 {head:current,next_key:if more {last}else{None},records})
        })
    }
}
fn validate_page(installation: &str, limit: usize, max_bytes: usize) -> R<()> {
    validate_id(installation)?;
    if limit == 0 || limit > 1000 || max_bytes == 0 || max_bytes > 16 * 1024 * 1024 {
        return Err(E::CapacityExceeded("object page"));
    }
    Ok(())
}