use crate::{
SqliteSyncError, SqliteSyncResult, SQLITE_SYNC_SCHEMA_V1, SQLITE_SYNC_SCHEMA_V2,
SQLITE_SYNC_SCHEMA_V3, SQLITE_SYNC_SCHEMA_V4, SQLITE_SYNC_SCHEMA_V5,
};
use rusqlite::{Connection, TransactionBehavior};
const APPLICATION_ID: i64 = 0x4150_4353;
pub(crate) fn validate_identity(connection: &Connection) -> SqliteSyncResult<()> {
let version = schema_version(connection)?;
let application_id: i64 = connection
.pragma_query_value(None, "application_id", |row| row.get(0))
.map_err(SqliteSyncError::database)?;
if version > SQLITE_SYNC_SCHEMA_V5
|| (version == 0 && (application_id != 0 || has_user_tables(connection)?))
|| (version > 0 && application_id != APPLICATION_ID)
{
return Err(SqliteSyncError::UpdateRequired);
}
Ok(())
}
pub(crate) fn migrate(
connection: &mut Connection,
config: &crate::SqliteSyncConfig,
) -> SqliteSyncResult<()> {
validate_identity(connection)?;
let version = schema_version(connection)?;
if version > SQLITE_SYNC_SCHEMA_V5 {
return Err(SqliteSyncError::UpdateRequired);
}
if version == 0 && has_user_tables(connection)? {
return Err(SqliteSyncError::UpdateRequired);
}
if version == 0 {
create_schema_v2(connection)?;
migrate_v2_to_v3(connection)?;
crate::object_schema::migrate(connection)?;
}
if version != 0 {
let application_id: i64 = connection
.pragma_query_value(None, "application_id", |row| row.get(0))
.map_err(SqliteSyncError::database)?;
if application_id != APPLICATION_ID {
return Err(SqliteSyncError::UpdateRequired);
}
if version == SQLITE_SYNC_SCHEMA_V1 {
migrate_v1_to_v2(connection)?;
}
if schema_version(connection)? == SQLITE_SYNC_SCHEMA_V2 {
migrate_v2_to_v3(connection)?;
}
if schema_version(connection)? == SQLITE_SYNC_SCHEMA_V3 {
crate::object_schema::migrate(connection)?;
}
}
if schema_version(connection)? == SQLITE_SYNC_SCHEMA_V4 {
migrate_v4_to_v5(connection, config)?;
}
if schema_version(connection)? != SQLITE_SYNC_SCHEMA_V5 {
return Err(SqliteSyncError::UpdateRequired);
}
reconcile_outbox_mode(connection, config)?;
let application_id: i64 = connection
.pragma_query_value(None, "application_id", |row| row.get(0))
.map_err(SqliteSyncError::database)?;
if application_id != APPLICATION_ID {
return Err(SqliteSyncError::UpdateRequired);
}
Ok(())
}
fn migrate_v4_to_v5(
connection: &mut Connection,
config: &crate::SqliteSyncConfig,
) -> SqliteSyncResult<()> {
let queue_file = config
.separate_outbox_path
.as_deref()
.and_then(std::path::Path::file_name)
.and_then(std::ffi::OsStr::to_str);
if config.separate_outbox_path.is_some() && queue_file.is_none() {
return Err(SqliteSyncError::UnsafePath);
}
let mode = if queue_file.is_some() {
"separate"
} else {
"local"
};
let transaction = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(SqliteSyncError::database)?;
transaction
.execute_batch(
"CREATE TABLE appcore_sync_storage_mode (
singleton INTEGER PRIMARY KEY CHECK(singleton=1),
outbox_mode TEXT NOT NULL CHECK(outbox_mode IN ('local','separate')),
outbox_file TEXT
);",
)
.map_err(SqliteSyncError::database)?;
transaction
.execute(
"INSERT INTO appcore_sync_storage_mode(singleton,outbox_mode,outbox_file)
VALUES(1,?1,?2)",
rusqlite::params![mode, queue_file],
)
.map_err(SqliteSyncError::database)?;
transaction
.pragma_update(None, "user_version", SQLITE_SYNC_SCHEMA_V5)
.map_err(SqliteSyncError::database)?;
transaction.commit().map_err(SqliteSyncError::database)
}
fn reconcile_outbox_mode(
connection: &mut Connection,
config: &crate::SqliteSyncConfig,
) -> SqliteSyncResult<()> {
let existing: (String, Option<String>) = connection
.query_row(
"SELECT outbox_mode,outbox_file FROM appcore_sync_storage_mode WHERE singleton=1",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.map_err(SqliteSyncError::database)?;
let requested_file = config
.separate_outbox_path
.as_deref()
.and_then(std::path::Path::file_name)
.and_then(std::ffi::OsStr::to_str);
if config.separate_outbox_path.is_some() && requested_file.is_none() {
return Err(SqliteSyncError::UnsafePath);
}
match (existing.0.as_str(), requested_file) {
("local", None) => Ok(()),
("separate", Some(file)) if existing.1.as_deref() == Some(file) => Ok(()),
("local", Some(file)) => {
connection
.execute(
"UPDATE appcore_sync_storage_mode
SET outbox_mode='separate',outbox_file=?1 WHERE singleton=1",
[file],
)
.map_err(SqliteSyncError::database)?;
Ok(())
}
_ => Err(SqliteSyncError::UpdateRequired),
}
}
fn schema_version(connection: &Connection) -> SqliteSyncResult<u32> {
connection
.pragma_query_value(None, "user_version", |row| row.get(0))
.map_err(SqliteSyncError::database)
}
fn has_user_tables(connection: &Connection) -> SqliteSyncResult<bool> {
connection
.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%')",
[],
|row| row.get(0),
)
.map_err(SqliteSyncError::database)
}
fn create_schema_v2(connection: &mut Connection) -> SqliteSyncResult<()> {
let transaction = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(SqliteSyncError::database)?;
transaction
.execute_batch(
"CREATE TABLE appcore_replication_log (
log_index INTEGER PRIMARY KEY AUTOINCREMENT,
source_sequence INTEGER NOT NULL CHECK(source_sequence >= 0),
payload BLOB NOT NULL,
previous_hash TEXT NOT NULL,
record_hash TEXT NOT NULL
);
CREATE UNIQUE INDEX appcore_replication_sequence
ON appcore_replication_log(source_sequence)
WHERE source_sequence > 0;
CREATE TABLE 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)
);
CREATE TABLE appcore_sync_checkpoint (
peer_id TEXT PRIMARY KEY,
sequence INTEGER NOT NULL CHECK(sequence >= 0),
batch_hash TEXT NOT NULL
) WITHOUT ROWID;
CREATE TABLE appcore_sync_tombstone (
namespace TEXT NOT NULL,
opaque_key TEXT NOT NULL,
deleted_sequence INTEGER NOT NULL CHECK(deleted_sequence > 0),
payload_hash TEXT NOT NULL,
expires_at_ms INTEGER NOT NULL CHECK(expires_at_ms > 0),
PRIMARY KEY(namespace, opaque_key)
) WITHOUT ROWID;
CREATE INDEX appcore_sync_tombstone_expiry
ON appcore_sync_tombstone(expires_at_ms);",
)
.map_err(SqliteSyncError::database)?;
transaction
.pragma_update(None, "application_id", APPLICATION_ID)
.map_err(SqliteSyncError::database)?;
transaction
.pragma_update(None, "user_version", SQLITE_SYNC_SCHEMA_V2)
.map_err(SqliteSyncError::database)?;
transaction.commit().map_err(SqliteSyncError::database)
}
fn migrate_v1_to_v2(connection: &mut Connection) -> SqliteSyncResult<()> {
let transaction = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(SqliteSyncError::database)?;
transaction
.execute_batch(
"ALTER TABLE appcore_sync_outbox
ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0 CHECK(attempts >= 0);
ALTER TABLE appcore_sync_outbox
ADD COLUMN next_ready_at_ms INTEGER NOT NULL DEFAULT 0
CHECK(next_ready_at_ms >= 0);",
)
.map_err(SqliteSyncError::database)?;
transaction
.pragma_update(None, "user_version", SQLITE_SYNC_SCHEMA_V2)
.map_err(SqliteSyncError::database)?;
transaction.commit().map_err(SqliteSyncError::database)
}
fn migrate_v2_to_v3(connection: &mut Connection) -> SqliteSyncResult<()> {
let transaction = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(SqliteSyncError::database)?;
transaction
.execute_batch(
"CREATE TABLE appcore_sync_conflict (
conflict_id TEXT PRIMARY KEY CHECK(length(conflict_id) <= 128),
evidence TEXT NOT NULL CHECK(length(CAST(evidence AS BLOB)) <= 8192),
resolution TEXT CHECK(length(CAST(resolution AS BLOB)) <= 4096)
) WITHOUT ROWID;
CREATE TABLE appcore_object_write_intent (
position INTEGER PRIMARY KEY AUTOINCREMENT,
intent_id TEXT NOT NULL UNIQUE CHECK(length(intent_id) <= 128),
installation_id TEXT NOT NULL CHECK(length(installation_id) <= 128),
object_key TEXT NOT NULL CHECK(length(object_key) <= 128),
envelope BLOB NOT NULL CHECK(length(envelope) BETWEEN 1 AND 1048576),
envelope_hash TEXT NOT NULL CHECK(length(envelope_hash) = 64)
);",
)
.map_err(SqliteSyncError::database)?;
transaction
.pragma_update(None, "user_version", SQLITE_SYNC_SCHEMA_V3)
.map_err(SqliteSyncError::database)?;
transaction.commit().map_err(SqliteSyncError::database)
}