use crate::{SqliteSyncConfig, SqliteSyncError, SqliteSyncStore};
use appcore_core::NodeId;
use appcore_sync::{ReplicationLog, SyncMessage, SyncOutbox, SyncOutboxReceipt};
use rusqlite::Connection;
fn message(batch_id: &str, sequence: u64) -> SyncMessage {
SyncMessage::new(
batch_id.into(),
NodeId::new("sync-test-node").unwrap(),
sequence,
sequence,
100,
None,
vec![format!("event-{sequence}").into_bytes()],
)
}
fn split_config(root: &std::path::Path, local: &str, queue: &str) -> SqliteSyncConfig {
SqliteSyncConfig::new(root.join(local)).with_separate_outbox_path(root.join(queue))
}
#[test]
fn separate_outbox_reopens_attempts_and_acknowledges_without_local_payload_copy() {
let root = tempfile::tempdir().unwrap();
let local = root.path().join("local.sqlite3");
let queue = root.path().join("sync-pending.sqlite3");
let config = split_config(root.path(), "local.sqlite3", "sync-pending.sqlite3");
let store = SqliteSyncStore::open(config.clone()).unwrap();
let outbox = store.outbox();
let pending = message("pending-1", 1);
assert!(outbox.try_enqueue(pending.clone(), 8).unwrap());
assert_eq!(outbox.mark_attempt("pending-1", 500).unwrap(), 1);
assert_eq!(outbox.stats().unwrap().pending_messages, 1);
drop(outbox);
drop(store);
let reopened = SqliteSyncStore::open(config).unwrap();
let outbox = reopened.outbox();
assert_eq!(outbox.messages().unwrap(), vec![pending]);
assert_eq!(outbox.stats().unwrap().next_ready_at_ms, Some(500));
let receipt = SyncOutboxReceipt::new(vec!["pending-1".into()]).unwrap();
assert_eq!(outbox.acknowledge_receipt(&receipt).unwrap(), 1);
assert_eq!(outbox.len().unwrap(), 0);
let local_connection = Connection::open(local).unwrap();
let local_count: i64 = local_connection
.query_row("SELECT COUNT(*) FROM main.appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(local_count, 0);
let queue_connection = Connection::open(queue).unwrap();
let queue_count: i64 = queue_connection
.query_row("SELECT COUNT(*) FROM appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(queue_count, 0);
}
#[test]
fn composed_local_work_and_sync_enqueue_roll_back_together() {
let root = tempfile::tempdir().unwrap();
let store = SqliteSyncStore::open(split_config(
root.path(),
"local.sqlite3",
"sync-pending.sqlite3",
))
.unwrap();
let pending = message("atomic-1", 1);
let result = store.transaction(|work| {
work.append(b"local-state", 0)?;
assert!(work.enqueue(&pending)?);
Err::<(), _>(SqliteSyncError::DatabaseOperation)
});
assert!(result.is_err());
assert_eq!(store.replication_log().len().unwrap(), 0);
assert_eq!(store.outbox().len().unwrap(), 0);
store
.transaction(|work| {
work.append(b"local-state", 0)?;
assert!(work.enqueue(&pending)?);
Ok(())
})
.unwrap();
assert_eq!(store.replication_log().len().unwrap(), 1);
assert_eq!(store.outbox().messages().unwrap(), vec![pending]);
}
#[test]
fn legacy_outbox_migration_copies_then_removes_the_local_rows() {
let root = tempfile::tempdir().unwrap();
let local = root.path().join("local.sqlite3");
let legacy = SqliteSyncStore::open(SqliteSyncConfig::new(&local)).unwrap();
assert!(legacy
.outbox()
.try_enqueue(message("legacy-1", 1), 8)
.unwrap());
drop(legacy);
let split = SqliteSyncStore::open(split_config(
root.path(),
"local.sqlite3",
"sync-pending.sqlite3",
))
.unwrap();
assert_eq!(
split.outbox().messages().unwrap(),
vec![message("legacy-1", 1)]
);
let local_connection = Connection::open(local).unwrap();
let local_count: i64 = local_connection
.query_row("SELECT COUNT(*) FROM main.appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(local_count, 0);
}
#[test]
fn divergent_existing_queue_refuses_migration_without_deleting_either_copy() {
let root = tempfile::tempdir().unwrap();
let local = root.path().join("local.sqlite3");
let queue = root.path().join("sync-pending.sqlite3");
let legacy = SqliteSyncStore::open(SqliteSyncConfig::new(&local)).unwrap();
assert!(legacy
.outbox()
.try_enqueue(message("local-copy", 1), 8)
.unwrap());
drop(legacy);
let other_local = root.path().join("other-local.sqlite3");
let other =
SqliteSyncStore::open(SqliteSyncConfig::new(other_local).with_separate_outbox_path(&queue))
.unwrap();
assert!(other
.outbox()
.try_enqueue(message("queue-copy", 1), 8)
.unwrap());
drop(other);
let result = SqliteSyncStore::open(split_config(
root.path(),
"local.sqlite3",
"sync-pending.sqlite3",
));
assert!(matches!(result, Err(SqliteSyncError::IntegrityFailed)));
let local_connection = Connection::open(local).unwrap();
let local_count: i64 = local_connection
.query_row("SELECT COUNT(*) FROM main.appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(local_count, 1);
let queue_connection = Connection::open(queue).unwrap();
let queue_count: i64 = queue_connection
.query_row("SELECT COUNT(*) FROM appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(queue_count, 1);
}
#[test]
fn migration_rejects_extra_queue_only_records_without_deleting_either_side() {
let root = tempfile::tempdir().unwrap();
let local_path = root.path().join("local.sqlite3");
let queue_path = root.path().join("sync-pending.sqlite3");
let shared = message("shared", 1);
let local = SqliteSyncStore::open(SqliteSyncConfig::new(&local_path)).unwrap();
local.outbox().try_enqueue(shared.clone(), 8).unwrap();
drop(local);
let other = SqliteSyncStore::open(
SqliteSyncConfig::new(root.path().join("other.sqlite3"))
.with_separate_outbox_path(&queue_path),
)
.unwrap();
other.outbox().try_enqueue(shared.clone(), 8).unwrap();
other
.outbox()
.try_enqueue(message("queue-only", 2), 8)
.unwrap();
drop(other);
assert!(matches!(
SqliteSyncStore::open(
SqliteSyncConfig::new(&local_path).with_separate_outbox_path(&queue_path)
),
Err(SqliteSyncError::IntegrityFailed)
));
let local = Connection::open(local_path).unwrap();
let local_count: i64 = local
.query_row("SELECT count(*) FROM appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
let queue = Connection::open(queue_path).unwrap();
let queue_count: i64 = queue
.query_row("SELECT count(*) FROM appcore_sync_outbox", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(local_count, 1);
assert_eq!(queue_count, 2);
}
#[test]
fn paired_backup_restore_preserves_both_files_and_rejects_tampering() {
let root = tempfile::tempdir().unwrap();
let local = root.path().join("canonical-v1.sqlite3");
let queue = root.path().join("sync-pending.sqlite3");
let store =
SqliteSyncStore::open(SqliteSyncConfig::new(&local).with_separate_outbox_path(&queue))
.unwrap();
store
.transaction(|work| {
work.append(b"complete-local-state", 0)?;
work.enqueue(&message("backup-pending", 1))?;
Ok(())
})
.unwrap();
let bundle = root.path().join("backup-bundle");
let report = store.paired_online_backup(&bundle).unwrap();
assert_eq!(report.queue_entries, 1);
assert!(bundle.join("manifest.txt").is_file());
let restored_dir = root.path().join("restored-profile");
SqliteSyncStore::restore_paired_backup_to_new(&bundle, &restored_dir).unwrap();
let restored = SqliteSyncStore::open(
SqliteSyncConfig::new(restored_dir.join("canonical-v1.sqlite3"))
.with_separate_outbox_path(restored_dir.join("sync-pending.sqlite3")),
)
.unwrap();
assert_eq!(restored.replication_log().len().unwrap(), 1);
assert_eq!(
restored.outbox().messages().unwrap(),
vec![message("backup-pending", 1)]
);
let corrupted = root.path().join("corrupted-bundle");
std::fs::create_dir(&corrupted).unwrap();
for name in ["local.sqlite3", "sync-queue.sqlite3", "manifest.txt"] {
std::fs::copy(bundle.join(name), corrupted.join(name)).unwrap();
}
let queue_copy = corrupted.join("sync-queue.sqlite3");
let connection = Connection::open(&queue_copy).unwrap();
connection
.execute("UPDATE appcore_sync_outbox SET attempts=attempts+1", [])
.unwrap();
drop(connection);
assert_eq!(
SqliteSyncStore::restore_paired_backup_to_new(
&corrupted,
root.path().join("must-not-exist"),
)
.unwrap_err(),
SqliteSyncError::IntegrityFailed
);
assert!(!root.path().join("must-not-exist").exists());
}