appcore-sync-sqlite 1.0.1-rc.4

// =============================================================================
//        #######
//     ###       ###     F: separate_outbox.rs
//    ##   ## ##   ##    P: AppCore-Runtime
//         ## ##
//                       C: 2026/11/14 00:00:00 by dnettoRaw
//    ##   ## ##   ##    U: 2026/11/14 00:00:00 by dnettoRaw
//      ###########
// =============================================================================

//! Owns the optional pending-only outbox database and its atomic handoff.

use crate::{SqliteSyncConfig, SqliteSyncError, SqliteSyncResult};
use rusqlite::{Connection, TransactionBehavior};

const QUEUE_APPLICATION_ID: i64 = 0x4150_4351;
const QUEUE_SCHEMA_VERSION: u32 = 1;

#[cfg(test)]
#[path = "separate_outbox_tests.rs"]
mod tests;

pub(crate) fn attach(connection: &Connection, config: &SqliteSyncConfig) -> SqliteSyncResult<()> {
    let Some(path) = config.separate_outbox_path.as_ref() else {
        return Ok(());
    };
    let path = path.to_str().ok_or(SqliteSyncError::UnsafePath)?;
    connection
        .execute("ATTACH DATABASE ?1 AS sync_queue", [path])
        .map_err(SqliteSyncError::database)?;
    Ok(())
}

pub(crate) fn initialize(
    connection: &mut Connection,
    config: &SqliteSyncConfig,
) -> SqliteSyncResult<()> {
    if config.separate_outbox_path.is_none() {
        return Ok(());
    }
    let version: u32 = connection
        .pragma_query_value(Some("sync_queue"), "user_version", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    let application_id: i64 = connection
        .pragma_query_value(Some("sync_queue"), "application_id", |row| row.get(0))
        .map_err(SqliteSyncError::database)?;
    let has_tables: bool = connection
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM sync_queue.sqlite_schema
             WHERE type='table' AND name NOT LIKE 'sqlite_%')",
            [],
            |row| row.get(0),
        )
        .map_err(SqliteSyncError::database)?;
    if version == 0 && application_id == 0 && !has_tables {
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(SqliteSyncError::database)?;
        transaction
            .execute_batch(
                "CREATE TABLE sync_queue.appcore_sync_outbox (
                   position INTEGER PRIMARY KEY AUTOINCREMENT,
                   batch_id TEXT NOT NULL UNIQUE,
                   encoded BLOB NOT NULL,
                   attempts INTEGER NOT NULL DEFAULT 0 CHECK(attempts >= 0),
                   next_ready_at_ms INTEGER NOT NULL DEFAULT 0 CHECK(next_ready_at_ms >= 0)
                 );",
            )
            .map_err(SqliteSyncError::database)?;
        transaction
            .pragma_update(Some("sync_queue"), "application_id", QUEUE_APPLICATION_ID)
            .and_then(|_| {
                transaction.pragma_update(Some("sync_queue"), "user_version", QUEUE_SCHEMA_VERSION)
            })
            .map_err(SqliteSyncError::database)?;
        transaction.commit().map_err(SqliteSyncError::database)?;
        return Ok(());
    }
    let expected_table: bool = connection
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM sync_queue.sqlite_schema
             WHERE type='table' AND name='appcore_sync_outbox')",
            [],
            |row| row.get(0),
        )
        .map_err(SqliteSyncError::database)?;
    if version != QUEUE_SCHEMA_VERSION
        || application_id != QUEUE_APPLICATION_ID
        || !expected_table
        || has_unowned_tables(connection)?
    {
        return Err(SqliteSyncError::UpdateRequired);
    }
    Ok(())
}

fn has_unowned_tables(connection: &Connection) -> SqliteSyncResult<bool> {
    connection
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM sync_queue.sqlite_schema
             WHERE type='table' AND name NOT LIKE 'sqlite_%'
               AND name!='appcore_sync_outbox')",
            [],
            |row| row.get(0),
        )
        .map_err(SqliteSyncError::database)
}

/// Moves any pre-existing main-database queue only after an exact attached copy is proven.
///
/// Both attached files participate in the same SQLite transaction. Any identity,
/// sequence or payload mismatch aborts before deleting the source rows.
pub(crate) fn relocate_existing_outbox(
    connection: &mut Connection,
    config: &SqliteSyncConfig,
) -> SqliteSyncResult<()> {
    if config.separate_outbox_path.is_none() {
        return Ok(());
    }
    let legacy_count: i64 = connection
        .query_row("SELECT count(*) FROM main.appcore_sync_outbox", [], |row| {
            row.get(0)
        })
        .map_err(SqliteSyncError::database)?;
    if legacy_count == 0 {
        return Ok(());
    }
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(SqliteSyncError::database)?;
    transaction
        .execute_batch(
            "INSERT OR IGNORE INTO sync_queue.appcore_sync_outbox
               (position,batch_id,encoded,attempts,next_ready_at_ms)
             SELECT position,batch_id,encoded,attempts,next_ready_at_ms
             FROM main.appcore_sync_outbox ORDER BY position;",
        )
        .map_err(SqliteSyncError::database)?;
    let mismatch: bool = transaction
        .query_row(
            "SELECT EXISTS(
               SELECT 1 FROM main.appcore_sync_outbox old
               LEFT JOIN sync_queue.appcore_sync_outbox new
                 ON new.position=old.position
               WHERE new.position IS NULL
                  OR new.batch_id!=old.batch_id
                  OR new.encoded!=old.encoded
                  OR new.attempts!=old.attempts
                  OR new.next_ready_at_ms!=old.next_ready_at_ms
             ) OR EXISTS(
               SELECT 1 FROM sync_queue.appcore_sync_outbox new
               LEFT JOIN main.appcore_sync_outbox old
                 ON old.position=new.position
               WHERE old.position IS NULL
                  OR old.batch_id!=new.batch_id
                  OR old.encoded!=new.encoded
                  OR old.attempts!=new.attempts
                  OR old.next_ready_at_ms!=new.next_ready_at_ms
             )",
            [],
            |row| row.get(0),
        )
        .map_err(SqliteSyncError::database)?;
    if mismatch {
        return Err(SqliteSyncError::IntegrityFailed);
    }
    transaction
        .execute("DELETE FROM main.appcore_sync_outbox", [])
        .map_err(SqliteSyncError::database)?;
    transaction.commit().map_err(SqliteSyncError::database)
}