mod records;
mod sql;
use crate::internal::identity_recover_local_state::helpers::StoreResult;
use records::{
merge_recovered_contact, merge_recovered_contact_handle_binding, merge_recovered_group,
merge_recovered_group_member, merge_recovered_message, merge_recovered_relationship_event,
normalize_recover_owner_dids, normalize_recovered_contact_handle_binding_row,
normalize_recovered_contact_row, normalize_recovered_group_member_row,
normalize_recovered_group_row, normalize_recovered_message_row,
normalize_recovered_relationship_event_row,
};
use rusqlite::Transaction;
use sql::{
count_rows_for_owners, delete_rows_for_owners, query_maps_with_param, query_one_map2,
query_one_map3, store_file_exists, upsert_recovered_contact,
upsert_recovered_contact_handle_binding, upsert_recovered_group, upsert_recovered_group_member,
upsert_recovered_message, upsert_recovered_relationship_event, zero_counts,
};
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
const STORE_MERGE_TABLES: &[&str] = &[
"messages",
"contacts",
"contact_handle_bindings",
"relationship_events",
"groups",
"group_members",
];
const E2EE_CLEANUP_TABLES: &[&str] = &["e2ee_outbox", "e2ee_sessions"];
pub(crate) fn merge_recovered_handle_local_state<S: AsRef<str>>(
sqlite_path: &Path,
old_owner_dids: &[S],
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
) -> StoreResult<(BTreeMap<String, i64>, BTreeMap<String, i64>)> {
if !store_file_exists(sqlite_path) {
return Ok((
zero_counts(STORE_MERGE_TABLES),
zero_counts(E2EE_CLEANUP_TABLES),
));
}
let mut connection = rusqlite::Connection::open(sqlite_path)
.map_err(crate::internal::local_state::local_state_unavailable)?;
crate::internal::local_state::configure(&connection)?;
crate::internal::local_state::schema::ensure_schema(&connection)?;
merge_recovered_handle_local_state_for_connection(
&mut connection,
old_owner_dids,
new_owner_did,
final_owner_identity_id,
final_credential_name,
)
}
pub(crate) fn merge_recovered_handle_local_state_for_connection<S: AsRef<str>>(
connection: &mut rusqlite::Connection,
old_owner_dids: &[S],
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
) -> StoreResult<(BTreeMap<String, i64>, BTreeMap<String, i64>)> {
let store_merge = zero_counts(STORE_MERGE_TABLES);
let e2ee_cleanup = zero_counts(E2EE_CLEANUP_TABLES);
let owners = normalize_recover_owner_dids(old_owner_dids, new_owner_did);
let _ = final_credential_name;
crate::internal::local_state::schema::record_identity_did_history_transition(
connection,
final_owner_identity_id,
new_owner_did,
&owners,
)?;
Ok((store_merge, e2ee_cleanup))
}
fn merge_recovered_messages(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
old_owner_set: &BTreeSet<String>,
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
) -> StoreResult<i64> {
let mut count = 0;
for old_owner in old_owner_dids {
let rows = query_maps_with_param(
transaction,
"SELECT * FROM messages WHERE owner_did = ?1 ORDER BY COALESCE(sent_at, stored_at) ASC, msg_id ASC",
old_owner,
)?;
count += rows.len() as i64;
for row in rows {
let mut record = normalize_recovered_message_row(
&row,
old_owner_set,
new_owner_did,
final_owner_identity_id,
final_credential_name,
);
if let Some(existing) = query_one_map2(
transaction,
"SELECT * FROM messages WHERE owner_identity_id = ?1 AND msg_id = ?2",
final_owner_identity_id,
&record.msg_id,
)? {
record = merge_recovered_message(&existing, record);
}
upsert_recovered_message(transaction, &record)?;
}
}
delete_rows_for_owners(transaction, "messages", old_owner_dids)?;
Ok(count)
}
fn merge_recovered_contacts(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
affected_handles: &mut BTreeSet<String>,
) -> StoreResult<i64> {
let mut count = 0;
for old_owner in old_owner_dids {
let rows = query_maps_with_param(
transaction,
"SELECT * FROM contacts WHERE owner_did = ?1 ORDER BY COALESCE(last_seen_at, first_seen_at, connected_at) ASC, did ASC",
old_owner,
)?;
count += rows.len() as i64;
for row in rows {
let mut record = normalize_recovered_contact_row(
&row,
new_owner_did,
final_owner_identity_id,
final_credential_name,
);
if !record.handle.trim().is_empty() {
affected_handles.insert(record.handle.trim().to_string());
}
if let Some(existing) = query_one_map2(
transaction,
"SELECT * FROM contacts WHERE owner_identity_id = ?1 AND did = ?2",
final_owner_identity_id,
&record.did,
)? {
record = merge_recovered_contact(&existing, record);
}
upsert_recovered_contact(transaction, &record)?;
}
}
delete_rows_for_owners(transaction, "contacts", old_owner_dids)?;
Ok(count)
}
fn merge_recovered_contact_handle_bindings(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
affected_handles: &mut BTreeSet<String>,
) -> StoreResult<i64> {
let mut count = 0;
for old_owner in old_owner_dids {
let rows = query_maps_with_param(
transaction,
"SELECT * FROM contact_handle_bindings WHERE owner_did = ?1 ORDER BY COALESCE(last_seen_at, first_seen_at) ASC, handle ASC, did ASC",
old_owner,
)?;
count += rows.len() as i64;
for row in rows {
let mut record = normalize_recovered_contact_handle_binding_row(
&row,
new_owner_did,
final_owner_identity_id,
final_credential_name,
);
if !record.handle.trim().is_empty() {
affected_handles.insert(record.handle.trim().to_string());
}
if let Some(existing) = query_one_map3(
transaction,
"SELECT * FROM contact_handle_bindings WHERE owner_identity_id = ?1 AND handle = ?2 AND did = ?3",
final_owner_identity_id,
&record.handle,
&record.did,
)? {
record = merge_recovered_contact_handle_binding(&existing, record);
}
upsert_recovered_contact_handle_binding(transaction, &record)?;
}
}
delete_rows_for_owners(transaction, "contact_handle_bindings", old_owner_dids)?;
Ok(count)
}
fn merge_recovered_relationship_events(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
) -> StoreResult<i64> {
let mut count = 0;
for old_owner in old_owner_dids {
let rows = query_maps_with_param(
transaction,
"SELECT * FROM relationship_events WHERE owner_did = ?1 ORDER BY COALESCE(updated_at, created_at) ASC, event_id ASC",
old_owner,
)?;
count += rows.len() as i64;
for row in rows {
let mut record = normalize_recovered_relationship_event_row(
&row,
new_owner_did,
final_owner_identity_id,
final_credential_name,
);
if let Some(existing) = query_one_map2(
transaction,
"SELECT * FROM relationship_events WHERE owner_identity_id = ?1 AND event_id = ?2",
final_owner_identity_id,
&record.event_id,
)? {
record = merge_recovered_relationship_event(&existing, record);
}
upsert_recovered_relationship_event(transaction, &record)?;
}
}
delete_rows_for_owners(transaction, "relationship_events", old_owner_dids)?;
Ok(count)
}
fn merge_recovered_groups(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
old_owner_set: &BTreeSet<String>,
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
) -> StoreResult<i64> {
let mut count = 0;
for old_owner in old_owner_dids {
let rows = query_maps_with_param(
transaction,
"SELECT * FROM groups WHERE owner_did = ?1 ORDER BY COALESCE(remote_updated_at, last_message_at, stored_at) ASC, group_id ASC",
old_owner,
)?;
count += rows.len() as i64;
for row in rows {
let mut record = normalize_recovered_group_row(
&row,
old_owner_set,
new_owner_did,
final_owner_identity_id,
final_credential_name,
);
if let Some(existing) = query_one_map2(
transaction,
"SELECT * FROM groups WHERE owner_identity_id = ?1 AND group_id = ?2",
final_owner_identity_id,
&record.group_id,
)? {
record = merge_recovered_group(&existing, record);
}
upsert_recovered_group(transaction, &record)?;
}
}
delete_rows_for_owners(transaction, "groups", old_owner_dids)?;
Ok(count)
}
fn merge_recovered_group_members(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
old_owner_set: &BTreeSet<String>,
new_owner_did: &str,
final_owner_identity_id: &str,
final_credential_name: &str,
) -> StoreResult<i64> {
let mut count = 0;
for old_owner in old_owner_dids {
let rows = query_maps_with_param(
transaction,
"SELECT * FROM group_members WHERE owner_did = ?1 ORDER BY group_id ASC, user_id ASC",
old_owner,
)?;
count += rows.len() as i64;
for row in rows {
let mut record = normalize_recovered_group_member_row(
&row,
old_owner_set,
new_owner_did,
final_owner_identity_id,
final_credential_name,
);
if let Some(existing) = query_one_map3(
transaction,
"SELECT * FROM group_members WHERE owner_identity_id = ?1 AND group_id = ?2 AND user_id = ?3",
final_owner_identity_id,
&record.group_id,
&record.user_id,
)? {
record = merge_recovered_group_member(&existing, record);
}
upsert_recovered_group_member(transaction, &record)?;
}
}
delete_rows_for_owners(transaction, "group_members", old_owner_dids)?;
Ok(count)
}
fn clear_recovered_owner_e2ee_data(
transaction: &Transaction<'_>,
old_owner_dids: &[String],
) -> StoreResult<BTreeMap<String, i64>> {
let mut result = zero_counts(E2EE_CLEANUP_TABLES);
for table in E2EE_CLEANUP_TABLES {
let total = count_rows_for_owners(transaction, table, old_owner_dids)?;
result.insert((*table).to_string(), total);
delete_rows_for_owners(transaction, table, old_owner_dids)?;
}
Ok(result)
}
mod helpers {
use std::time::{SystemTime, UNIX_EPOCH};
use time::OffsetDateTime;
pub(super) type StoreResult<T> = crate::ImResult<T>;
pub(super) fn now_utc() -> String {
let value = OffsetDateTime::now_utc();
format!(
"{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
value.year(),
u8::from(value.month()),
value.day(),
value.hour(),
value.minute(),
value.second()
)
}
pub(super) fn make_thread_id(_my_did: &str, peer_did: &str, group_id: &str) -> String {
let group_id = group_id.trim();
if !group_id.is_empty() {
return crate::internal::local_state::owner_scope::group_conversation_id(group_id);
}
let peer_did = peer_did.trim();
if !peer_did.is_empty() {
return crate::internal::local_state::owner_scope::direct_conversation_id(peer_did);
}
crate::internal::local_state::owner_scope::direct_conversation_id("")
}
pub(super) fn normalize_owner_did(value: &str) -> String {
value.trim().to_string()
}
pub(super) fn normalize_credential_name(value: &str) -> String {
value.trim().to_string()
}
pub(super) fn normalize_optional_string(value: &str) -> Option<String> {
let value = value.trim();
if value.is_empty() {
None
} else {
Some(value.to_string())
}
}
pub(super) fn normalize_optional_bool(value: Option<bool>) -> Option<i64> {
value.map(bool_to_int)
}
pub(super) fn normalize_optional_int64(value: Option<i64>) -> Option<i64> {
value
}
pub(super) fn normalize_optional_float64(value: Option<f64>) -> Option<f64> {
value
}
pub(super) fn normalize_metadata(value: &str) -> Option<String> {
normalize_optional_string(value)
}
pub(super) fn default_bool_value(value: Option<bool>) -> i64 {
value.map(bool_to_int).unwrap_or(0)
}
pub(super) fn bool_to_int(value: bool) -> i64 {
if value {
1
} else {
0
}
}
pub(super) fn default_string(value: String, fallback: &str) -> String {
if value.trim().is_empty() {
fallback.to_string()
} else {
value
}
}
#[allow(dead_code)]
pub(super) fn generate_id() -> String {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
format!("local-{nanos}")
}
}
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::Connection;
use std::collections::BTreeMap;
use std::path::Path;
const STORE_KEYS: &[&str] = &[
"messages",
"contacts",
"contact_handle_bindings",
"relationship_events",
"groups",
"group_members",
];
const E2EE_KEYS: &[&str] = &["e2ee_outbox", "e2ee_sessions"];
#[test]
fn recover_merge_missing_database_returns_zero_counts_and_does_not_create_db() {
let temp = tempfile::TempDir::new().expect("temp dir");
let sqlite_path = sqlite_path(temp.path());
let (store_merge, e2ee_cleanup) = merge_recovered_handle_local_state(
&sqlite_path,
&["did:old"],
"did:new",
"final-owner-id",
"final-credential",
)
.expect("missing database is a soft no-op");
assert_zero_counts(&store_merge, STORE_KEYS);
assert_zero_counts(&e2ee_cleanup, E2EE_KEYS);
assert!(
!sqlite_path.exists(),
"missing-DB no-op must not create the SQLite file"
);
}
#[test]
fn recover_merge_existing_empty_database_ensures_schema_and_returns_zero_counts() {
let temp = tempfile::TempDir::new().expect("temp dir");
let sqlite_path = sqlite_path(temp.path());
std::fs::create_dir_all(sqlite_path.parent().unwrap()).expect("data dir");
drop(Connection::open(&sqlite_path).expect("create empty sqlite"));
let (store_merge, e2ee_cleanup) = merge_recovered_handle_local_state(
&sqlite_path,
&["did:old"],
"did:new",
"final-owner-id",
"final-credential",
)
.expect("existing empty database should be migrated before merge");
assert_zero_counts(&store_merge, STORE_KEYS);
assert_zero_counts(&e2ee_cleanup, E2EE_KEYS);
let verify = Connection::open(&sqlite_path).expect("open database");
assert_eq!(
crate::internal::local_state::schema::current_schema_version(&verify)
.expect("schema version"),
crate::internal::local_state::schema::SCHEMA_VERSION
);
for table in STORE_KEYS.iter().chain(["e2ee_outbox"].iter()) {
assert_table_exists(&verify, table);
}
}
#[test]
fn recover_merge_moves_only_target_owners_and_cleans_only_old_owner_e2ee() {
let temp = tempfile::TempDir::new().expect("temp dir");
let sqlite_path = sqlite_path(temp.path());
let connection = create_database(&sqlite_path);
seed_recover_main_rows(&connection).expect("seed rows");
drop(connection);
let (store_merge, e2ee_cleanup) = merge_recovered_handle_local_state(
&sqlite_path,
&[
" did:owner:old-1 ",
"",
"did:owner:new",
"did:owner:old-2",
"did:owner:old-1",
],
"did:owner:new",
"final-owner-id",
"final-credential",
)
.expect("merge recovered local state");
assert_zero_counts(&store_merge, STORE_KEYS);
assert_zero_counts(&e2ee_cleanup, E2EE_KEYS);
let verify = Connection::open(&sqlite_path).expect("open database");
assert_owner_count(&verify, "messages", "did:owner:old-1", 1);
assert_owner_count(&verify, "messages", "did:owner:old-2", 1);
assert_owner_count(&verify, "contacts", "did:owner:old-1", 1);
assert_owner_count(&verify, "contacts", "did:owner:old-2", 1);
assert_owner_count(&verify, "contact_handle_bindings", "did:owner:old-1", 1);
assert_owner_count(&verify, "contact_handle_bindings", "did:owner:old-2", 1);
assert_owner_count(&verify, "relationship_events", "did:owner:old-1", 1);
assert_owner_count(&verify, "groups", "did:owner:old-2", 1);
assert_owner_count(&verify, "group_members", "did:owner:old-2", 1);
assert_owner_count(&verify, "messages", "did:owner:new", 0);
assert_owner_count(&verify, "contacts", "did:owner:new", 0);
assert_owner_count(&verify, "contact_handle_bindings", "did:owner:new", 0);
assert_owner_count(&verify, "relationship_events", "did:owner:new", 0);
assert_owner_count(&verify, "groups", "did:owner:new", 0);
assert_owner_count(&verify, "group_members", "did:owner:new", 0);
assert_owner_count(&verify, "messages", "did:owner:other", 1);
assert_owner_count(&verify, "contacts", "did:owner:other", 1);
assert_owner_count(&verify, "e2ee_outbox", "did:owner:new", 1);
assert_owner_count(&verify, "e2ee_sessions", "did:owner:new", 1);
assert_owner_count(&verify, "e2ee_outbox", "did:owner:old-1", 1);
assert_owner_count(&verify, "e2ee_sessions", "did:owner:old-2", 1);
assert_owner_count(&verify, "e2ee_outbox", "did:owner:other", 1);
assert_eq!(
scalar_string(
&verify,
"SELECT receiver_did FROM messages WHERE owner_did = 'did:owner:old-1' AND msg_id = 'msg-old-1'",
),
"did:owner:old-1"
);
assert_eq!(
scalar_string(
&verify,
"SELECT thread_id FROM messages WHERE owner_did = 'did:owner:old-1' AND msg_id = 'msg-old-1'",
),
expected_thread_id("did:owner:old-1", "did:peer:bob", "")
);
assert_eq!(
scalar_string(
&verify,
"SELECT sender_did FROM messages WHERE owner_did = 'did:owner:old-2' AND msg_id = 'msg-old-2'",
),
"did:owner:old-2"
);
assert_eq!(
scalar_string(
&verify,
"SELECT credential_name FROM messages WHERE owner_did = 'did:owner:old-2' AND msg_id = 'msg-old-2'",
),
"old-credential-2"
);
assert_eq!(
scalar_string(
&verify,
"SELECT did FROM contact_handle_bindings WHERE owner_did = 'did:owner:old-2' AND handle = 'alice' AND is_current = 1",
),
"did:peer:new"
);
assert_eq!(
scalar_opt_string(
&verify,
"SELECT handle FROM contacts WHERE owner_did = 'did:owner:old-1' AND did = 'did:peer:old'",
),
Some("alice".to_string())
);
assert_eq!(
scalar_opt_string(
&verify,
"SELECT handle FROM contacts WHERE owner_did = 'did:owner:old-2' AND did = 'did:peer:new'",
),
Some("alice".to_string())
);
assert_eq!(
scalar_string(
&verify,
"SELECT group_owner_did FROM groups WHERE owner_did = 'did:owner:old-2' AND group_id = 'group:one'",
),
"did:owner:old-2"
);
assert_eq!(
scalar_string(
&verify,
"SELECT member_did FROM group_members WHERE owner_did = 'did:owner:old-2' AND group_id = 'group:one' AND user_id = 'user-z'",
),
"did:owner:old-2"
);
assert_did_history(&verify, "final-owner-id", "did:owner:new", "current");
assert_did_history(&verify, "final-owner-id", "did:owner:old-1", "previous");
assert_did_history(&verify, "final-owner-id", "did:owner:old-2", "previous");
}
#[test]
fn recover_merge_counts_source_rows_before_conflicts_and_applies_merge_algebra() {
let temp = tempfile::TempDir::new().expect("temp dir");
let sqlite_path = sqlite_path(temp.path());
let connection = create_database(&sqlite_path);
seed_recover_conflict_rows(&connection).expect("seed rows");
drop(connection);
let (store_merge, e2ee_cleanup) = merge_recovered_handle_local_state(
&sqlite_path,
&["did:owner:old"],
"did:owner:new",
"final-owner-id",
"final-credential",
)
.expect("merge recovered conflicts");
assert_zero_counts(&store_merge, STORE_KEYS);
assert_zero_counts(&e2ee_cleanup, E2EE_KEYS);
let verify = Connection::open(&sqlite_path).expect("open database");
for table in STORE_KEYS {
assert_owner_count(&verify, table, "did:owner:old", 1);
}
assert_eq!(
scalar_string(&verify, "SELECT content FROM messages WHERE owner_did = 'did:owner:old' AND msg_id = 'msg-conflict'"),
"incoming content"
);
assert_eq!(
scalar_i64(&verify, "SELECT server_seq FROM messages WHERE owner_did = 'did:owner:old' AND msg_id = 'msg-conflict'"),
9
);
assert_eq!(
scalar_string(&verify, "SELECT sent_at FROM messages WHERE owner_did = 'did:owner:old' AND msg_id = 'msg-conflict'"),
"2026-01-03T00:00:00Z"
);
assert_eq!(
scalar_i64(&verify, "SELECT is_e2ee FROM messages WHERE owner_did = 'did:owner:old' AND msg_id = 'msg-conflict'"),
1
);
assert_eq!(
scalar_i64(&verify, "SELECT is_read FROM messages WHERE owner_did = 'did:owner:old' AND msg_id = 'msg-conflict'"),
0
);
assert_eq!(
scalar_string(&verify, "SELECT name FROM contacts WHERE owner_did = 'did:owner:old' AND did = 'did:peer:conflict'"),
"Incoming Alice"
);
assert_eq!(
scalar_string(&verify, "SELECT first_seen_at FROM contacts WHERE owner_did = 'did:owner:old' AND did = 'did:peer:conflict'"),
"2026-01-01T00:00:00Z"
);
assert_eq!(
scalar_string(&verify, "SELECT last_seen_at FROM contacts WHERE owner_did = 'did:owner:old' AND did = 'did:peer:conflict'"),
"2026-01-10T00:00:00Z"
);
assert_eq!(
scalar_i64(&verify, "SELECT followed FROM contacts WHERE owner_did = 'did:owner:old' AND did = 'did:peer:conflict'"),
0
);
assert_eq!(
scalar_i64(&verify, "SELECT messaged FROM contacts WHERE owner_did = 'did:owner:old' AND did = 'did:peer:conflict'"),
1
);
assert_eq!(
scalar_string(&verify, "SELECT first_seen_at FROM contact_handle_bindings WHERE owner_did = 'did:owner:old' AND handle = 'alice' AND did = 'did:peer:conflict'"),
"2026-01-01T00:00:00Z"
);
assert_eq!(
scalar_string(&verify, "SELECT last_seen_at FROM contact_handle_bindings WHERE owner_did = 'did:owner:old' AND handle = 'alice' AND did = 'did:peer:conflict'"),
"2026-01-10T00:00:00Z"
);
assert_eq!(
scalar_string(&verify, "SELECT name FROM groups WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict'"),
"Incoming Group"
);
assert_eq!(
scalar_string(&verify, "SELECT group_owner_did FROM groups WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict'"),
"did:owner:old"
);
assert_eq!(
scalar_i64(&verify, "SELECT member_count FROM groups WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict'"),
7
);
assert_eq!(
scalar_string(&verify, "SELECT remote_created_at FROM groups WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict'"),
"2026-01-01T00:00:00Z"
);
assert_eq!(
scalar_string(&verify, "SELECT remote_updated_at FROM groups WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict'"),
"2026-01-10T00:00:00Z"
);
assert_eq!(
scalar_string(&verify, "SELECT member_did FROM group_members WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict' AND user_id = 'user-conflict'"),
"did:owner:old"
);
assert_eq!(
scalar_i64(&verify, "SELECT sent_message_count FROM group_members WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict' AND user_id = 'user-conflict'"),
8
);
assert_eq!(
scalar_string(&verify, "SELECT joined_at FROM group_members WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict' AND user_id = 'user-conflict'"),
"2026-01-01T00:00:00Z"
);
assert_eq!(
scalar_string(&verify, "SELECT last_synced_at FROM group_members WHERE owner_did = 'did:owner:old' AND group_id = 'group-conflict' AND user_id = 'user-conflict'"),
"2026-01-10T00:00:00Z"
);
assert_did_history(&verify, "final-owner-id", "did:owner:new", "current");
assert_did_history(&verify, "final-owner-id", "did:owner:old", "previous");
}
#[test]
fn recover_merge_current_contact_handle_uses_latest_binding_then_did_desc_tiebreak() {
let temp = tempfile::TempDir::new().expect("temp dir");
let sqlite_path = sqlite_path(temp.path());
let connection = create_database(&sqlite_path);
seed_contact_with_binding(
&connection,
"did:owner:old-a",
"did:peer:a",
"same",
"2026-02-01T00:00:00Z",
"2026-02-10T00:00:00Z",
)
.expect("seed a");
seed_contact_with_binding(
&connection,
"did:owner:old-z",
"did:peer:z",
"same",
"2026-02-01T00:00:00Z",
"2026-02-10T00:00:00Z",
)
.expect("seed z");
seed_contact_with_binding(
&connection,
"did:owner:old-old",
"did:peer:old",
"same",
"2026-02-01T00:00:00Z",
"2026-02-09T00:00:00Z",
)
.expect("seed old");
drop(connection);
merge_recovered_handle_local_state(
&sqlite_path,
&["did:owner:old-a", "did:owner:old-z", "did:owner:old-old"],
"did:owner:new",
"final-owner-id",
"final-credential",
)
.expect("merge contacts");
let verify = Connection::open(&sqlite_path).expect("open database");
assert_owner_count(&verify, "contacts", "did:owner:new", 0);
assert_did_history(&verify, "final-owner-id", "did:owner:new", "current");
assert_did_history(&verify, "final-owner-id", "did:owner:old-a", "previous");
assert_did_history(&verify, "final-owner-id", "did:owner:old-z", "previous");
assert_did_history(&verify, "final-owner-id", "did:owner:old-old", "previous");
assert_eq!(
scalar_opt_string(
&verify,
"SELECT handle FROM contacts WHERE owner_did = 'did:owner:old-z' AND did = 'did:peer:z'",
),
Some("same".to_string())
);
for (owner_did, did) in [
("did:owner:old-a", "did:peer:a"),
("did:owner:old-old", "did:peer:old"),
] {
assert_eq!(
scalar_opt_string(
&verify,
&format!(
"SELECT handle FROM contacts WHERE owner_did = '{owner_did}' AND did = '{did}'"
),
),
Some("same".to_string()),
"{did} should remain in its original owner identity"
);
}
}
#[test]
fn recover_merge_relationship_events_move_by_global_event_id_not_owner_scope() {
let temp = tempfile::TempDir::new().expect("temp dir");
let sqlite_path = sqlite_path(temp.path());
let connection = create_database(&sqlite_path);
connection
.execute(
r#"
INSERT INTO relationship_events (
event_id, owner_identity_id, owner_did, target_did, target_handle, event_type, source_type,
reason, score, status, created_at, updated_at, metadata, credential_name
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)
"#,
rusqlite::params![
"evt-global",
owner_identity_id("did:owner:old"),
"did:owner:old",
"did:peer:target",
"target",
"follow",
"contacts",
"because",
0.42_f64,
"accepted",
"2026-03-01T00:00:00Z",
"2026-03-02T00:00:00Z",
"{\"old\":true}",
"old-credential",
],
)
.expect("insert relationship event");
drop(connection);
let (store_merge, e2ee_cleanup) = merge_recovered_handle_local_state(
&sqlite_path,
&["did:owner:old"],
"did:owner:new",
"final-owner-id",
"final-credential",
)
.expect("merge relationship event");
assert_zero_counts(&store_merge, STORE_KEYS);
assert_zero_counts(&e2ee_cleanup, E2EE_KEYS);
let verify = Connection::open(&sqlite_path).expect("open database");
assert_eq!(
scalar_i64(
&verify,
"SELECT COUNT(*) FROM relationship_events WHERE event_id = 'evt-global'",
),
1
);
assert_eq!(
scalar_string(
&verify,
"SELECT owner_did FROM relationship_events WHERE event_id = 'evt-global'",
),
"did:owner:old"
);
assert_eq!(
scalar_string(
&verify,
"SELECT credential_name FROM relationship_events WHERE event_id = 'evt-global'",
),
"old-credential"
);
assert_eq!(
scalar_opt_f64(
&verify,
"SELECT score FROM relationship_events WHERE event_id = 'evt-global'",
),
Some(0.42)
);
assert_did_history(&verify, "final-owner-id", "did:owner:new", "current");
assert_did_history(&verify, "final-owner-id", "did:owner:old", "previous");
}
fn seed_recover_main_rows(connection: &Connection) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO messages (
msg_id, owner_identity_id, owner_did, thread_id, direction, sender_did, receiver_did,
content_type, content, sent_at, stored_at, credential_name
) VALUES (?1, ?2, ?3, ?4, 0, ?5, ?6, 'text', ?7, ?8, ?8, ?9)
"#,
rusqlite::params![
"msg-old-1",
owner_identity_id("did:owner:old-1"),
"did:owner:old-1",
expected_thread_id("did:owner:old-1", "did:peer:bob", ""),
"did:peer:bob",
"did:owner:old-1",
"hello old one",
"2026-04-20T10:00:00Z",
"old-credential-1",
],
)?;
connection.execute(
r#"
INSERT INTO messages (
msg_id, owner_identity_id, owner_did, thread_id, direction, sender_did, receiver_did,
content_type, content, sent_at, stored_at, credential_name
) VALUES (?1, ?2, ?3, ?4, 1, ?5, ?6, 'text', ?7, ?8, ?8, ?9)
"#,
rusqlite::params![
"msg-old-2",
owner_identity_id("did:owner:old-2"),
"did:owner:old-2",
expected_thread_id("did:owner:old-2", "did:peer:bob", ""),
"did:owner:old-2",
"did:peer:bob",
"hello old two",
"2026-04-20T11:00:00Z",
"old-credential-2",
],
)?;
connection.execute(
r#"
INSERT INTO messages (
msg_id, owner_identity_id, owner_did, thread_id, direction, sender_did, receiver_did,
content_type, content, sent_at, stored_at, credential_name
) VALUES (?1, ?2, ?3, ?4, 0, ?5, ?6, 'text', ?7, ?8, ?8, ?9)
"#,
rusqlite::params![
"msg-other",
owner_identity_id("did:owner:other"),
"did:owner:other",
expected_thread_id("did:owner:other", "did:peer:bob", ""),
"did:peer:bob",
"did:owner:other",
"hello other",
"2026-04-20T12:00:00Z",
"other-credential",
],
)?;
seed_contact_with_binding(
connection,
"did:owner:old-1",
"did:peer:old",
"alice",
"2026-04-20T10:00:00Z",
"2026-04-20T10:00:00Z",
)?;
seed_contact_with_binding(
connection,
"did:owner:old-2",
"did:peer:new",
"alice",
"2026-04-20T11:00:00Z",
"2026-04-20T11:00:00Z",
)?;
seed_contact_with_binding(
connection,
"did:owner:other",
"did:peer:else",
"else",
"2026-04-20T09:00:00Z",
"2026-04-20T09:00:00Z",
)?;
connection.execute(
r#"
INSERT INTO relationship_events (
event_id, owner_identity_id, owner_did, target_did, target_handle, event_type, status,
created_at, updated_at, credential_name
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8, ?9)
"#,
rusqlite::params![
"evt-1",
owner_identity_id("did:owner:old-1"),
"did:owner:old-1",
"did:peer:old",
"alice",
"follow",
"pending",
"2026-04-20T10:00:00Z",
"old-credential-1",
],
)?;
connection.execute(
r#"
INSERT INTO groups (
owner_identity_id, owner_did, group_id, group_did, name, group_mode, group_owner_did,
membership_status, last_message_at, stored_at, credential_name
) VALUES (?1, ?2, ?3, ?4, ?5, 'general', ?6, 'active', ?7, ?7, ?8)
"#,
rusqlite::params![
owner_identity_id("did:owner:old-2"),
"did:owner:old-2",
"group:one",
"did:group:one",
"Group One",
"did:owner:old-2",
"2026-04-20T11:30:00Z",
"old-credential-2",
],
)?;
connection.execute(
r#"
INSERT INTO group_members (
owner_identity_id, owner_did, group_id, user_id, member_did, member_handle, status,
sent_message_count, last_synced_at, credential_name
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'active', 3, ?7, ?8)
"#,
rusqlite::params![
owner_identity_id("did:owner:old-2"),
"did:owner:old-2",
"group:one",
"user-z",
"did:owner:old-2",
"zhuocheng",
"2026-04-20T11:30:00Z",
"old-credential-2",
],
)?;
seed_e2ee_outbox(
connection,
"out-old",
"did:owner:old-1",
"did:peer:bob",
"old secret",
)?;
ensure_legacy_e2ee_sessions_table(connection)?;
seed_e2ee_session(connection, "did:owner:old-2", "did:peer:bob", "session-old")?;
seed_e2ee_outbox(
connection,
"out-new",
"did:owner:new",
"did:peer:bob",
"new secret",
)?;
seed_e2ee_session(connection, "did:owner:new", "did:peer:bob", "session-new")?;
seed_e2ee_outbox(
connection,
"out-other",
"did:owner:other",
"did:peer:bob",
"other secret",
)?;
Ok(())
}
fn seed_recover_conflict_rows(connection: &Connection) -> rusqlite::Result<()> {
seed_message_conflict(
connection,
"did:owner:new",
"did:peer:conflict",
"did:owner:new",
"existing content",
4,
"2026-01-02T00:00:00Z",
false,
true,
"existing-credential",
)?;
seed_message_conflict(
connection,
"did:owner:old",
"did:owner:old",
"did:peer:conflict",
"incoming content",
9,
"2026-01-03T00:00:00Z",
true,
false,
"old-credential",
)?;
seed_contact_conflict(
connection,
"did:owner:new",
"Existing Alice",
"2026-01-05T00:00:00Z",
"2026-01-06T00:00:00Z",
true,
false,
)?;
seed_contact_conflict(
connection,
"did:owner:old",
"Incoming Alice",
"2026-01-01T00:00:00Z",
"2026-01-10T00:00:00Z",
false,
true,
)?;
seed_binding_conflict(
connection,
"did:owner:new",
"2026-01-05T00:00:00Z",
"2026-01-06T00:00:00Z",
"existing-binding",
)?;
seed_binding_conflict(
connection,
"did:owner:old",
"2026-01-01T00:00:00Z",
"2026-01-10T00:00:00Z",
"incoming-binding",
)?;
connection.execute(
r#"
INSERT INTO relationship_events (
event_id, owner_identity_id, owner_did, target_did, target_handle, event_type, source_type,
reason, score, status, created_at, updated_at, credential_name
) VALUES ('evt-conflict', 'old-id', 'did:owner:old', 'did:peer:conflict', 'alice', 'follow', 'contacts',
'old reason', 2.5, 'pending', '2026-01-02T00:00:00Z', '2026-01-03T00:00:00Z', 'old-credential')
"#,
[],
)?;
seed_group_conflict(
connection,
"did:owner:new",
"Existing Group",
"did:owner:someone",
2,
"2026-01-05T00:00:00Z",
"2026-01-06T00:00:00Z",
)?;
seed_group_conflict(
connection,
"did:owner:old",
"Incoming Group",
"did:owner:old",
7,
"2026-01-01T00:00:00Z",
"2026-01-10T00:00:00Z",
)?;
seed_group_member_conflict(
connection,
"did:owner:new",
"did:owner:someone",
3,
"2026-01-05T00:00:00Z",
"2026-01-06T00:00:00Z",
)?;
seed_group_member_conflict(
connection,
"did:owner:old",
"did:owner:old",
8,
"2026-01-01T00:00:00Z",
"2026-01-10T00:00:00Z",
)?;
Ok(())
}
fn seed_message_conflict(
connection: &Connection,
owner_did: &str,
sender_did: &str,
receiver_did: &str,
content: &str,
server_seq: i64,
at: &str,
is_e2ee: bool,
is_read: bool,
credential_name: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO messages (
msg_id, owner_identity_id, owner_did, thread_id, direction, sender_did, receiver_did,
content_type, content, title, server_seq, sent_at, stored_at, is_e2ee,
is_read, sender_name, metadata, credential_name
) VALUES (?1, ?2, ?3, ?4, 0, ?5, ?6, 'text', ?7, 'incoming title', ?8, ?9, ?9, ?10, ?11, 'sender', '{"kind":"msg"}', ?12)
"#,
rusqlite::params![
"msg-conflict",
owner_identity_id(owner_did),
owner_did,
expected_thread_id(owner_did, "did:peer:conflict", ""),
sender_did,
receiver_did,
content,
server_seq,
at,
if is_e2ee { 1 } else { 0 },
if is_read { 1 } else { 0 },
credential_name,
],
)?;
Ok(())
}
fn seed_contact_conflict(
connection: &Connection,
owner_did: &str,
name: &str,
first_seen_at: &str,
last_seen_at: &str,
followed: bool,
messaged: bool,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO contacts (
owner_identity_id, owner_did, did, name, handle, followed, messaged, first_seen_at, last_seen_at, metadata
) VALUES (?1, ?2, 'did:peer:conflict', ?3, 'alice', ?4, ?5, ?6, ?7, '{"kind":"contact"}')
"#,
rusqlite::params![
owner_identity_id(owner_did),
owner_did,
name,
if followed { 1 } else { 0 },
if messaged { 1 } else { 0 },
first_seen_at,
last_seen_at,
],
)?;
Ok(())
}
fn seed_binding_conflict(
connection: &Connection,
owner_did: &str,
first_seen_at: &str,
last_seen_at: &str,
metadata: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO contact_handle_bindings (
owner_identity_id, owner_did, handle, did, is_current, first_seen_at, last_seen_at, metadata, credential_name
) VALUES (?1, ?2, 'alice', 'did:peer:conflict', 1, ?3, ?4, ?5, 'old-credential')
"#,
rusqlite::params![
owner_identity_id(owner_did),
owner_did,
first_seen_at,
last_seen_at,
metadata
],
)?;
Ok(())
}
fn seed_group_conflict(
connection: &Connection,
owner_did: &str,
name: &str,
group_owner_did: &str,
member_count: i64,
remote_created_at: &str,
remote_updated_at: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO groups (
owner_identity_id, owner_did, group_id, group_did, name, group_mode, group_owner_did,
membership_status, member_count, remote_created_at, remote_updated_at,
stored_at, credential_name
) VALUES (?1, ?2, 'group-conflict', 'did:group:conflict', ?3, 'general', ?4, 'active', ?5, ?6, ?7, ?7, 'old-credential')
"#,
rusqlite::params![
owner_identity_id(owner_did),
owner_did,
name,
group_owner_did,
member_count,
remote_created_at,
remote_updated_at,
],
)?;
Ok(())
}
fn seed_group_member_conflict(
connection: &Connection,
owner_did: &str,
member_did: &str,
sent_message_count: i64,
joined_at: &str,
last_synced_at: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO group_members (
owner_identity_id, owner_did, group_id, user_id, member_did, member_handle, status,
joined_at, sent_message_count, last_synced_at, credential_name
) VALUES (?1, ?2, 'group-conflict', 'user-conflict', ?3, 'member', 'active', ?4, ?5, ?6, 'old-credential')
"#,
rusqlite::params![
owner_identity_id(owner_did),
owner_did,
member_did,
joined_at,
sent_message_count,
last_synced_at,
],
)?;
Ok(())
}
fn seed_contact_with_binding(
connection: &Connection,
owner_did: &str,
did: &str,
handle: &str,
first_seen_at: &str,
last_seen_at: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO contacts (
owner_identity_id, owner_did, did, handle, first_seen_at, last_seen_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
"#,
rusqlite::params![
owner_identity_id(owner_did),
owner_did,
did,
handle,
first_seen_at,
last_seen_at
],
)?;
connection.execute(
r#"
INSERT INTO contact_handle_bindings (
owner_identity_id, owner_did, handle, did, is_current, first_seen_at, last_seen_at, credential_name
) VALUES (?1, ?2, ?3, ?4, 1, ?5, ?6, 'old-credential')
"#,
rusqlite::params![
owner_identity_id(owner_did),
owner_did,
handle,
did,
first_seen_at,
last_seen_at
],
)?;
Ok(())
}
fn seed_e2ee_outbox(
connection: &Connection,
outbox_id: &str,
owner_did: &str,
peer_did: &str,
plaintext: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO e2ee_outbox (
outbox_id, owner_identity_id, owner_did, peer_did, plaintext, created_at, updated_at, credential_name
) VALUES (?1, ?2, ?3, ?4, ?5, '2026-04-20T10:00:00Z', '2026-04-20T10:00:00Z', 'old-credential')
"#,
rusqlite::params![
outbox_id,
owner_identity_id(owner_did),
owner_did,
peer_did,
plaintext
],
)?;
Ok(())
}
fn seed_e2ee_session(
connection: &Connection,
owner_did: &str,
peer_did: &str,
session_id: &str,
) -> rusqlite::Result<()> {
connection.execute(
r#"
INSERT INTO e2ee_sessions (
owner_did, peer_did, session_id, is_initiator, send_chain_key, recv_chain_key,
send_seq, recv_seq, expires_at, created_at, active_at, peer_confirmed,
credential_name, updated_at
) VALUES (?1, ?2, ?3, 1, 'send-key', 'recv-key', 0, 0, NULL, '2026-04-20T10:00:00Z', NULL, 0, 'old-credential', '2026-04-20T10:00:00Z')
"#,
rusqlite::params![owner_did, peer_did, session_id],
)?;
Ok(())
}
fn ensure_legacy_e2ee_sessions_table(connection: &Connection) -> rusqlite::Result<()> {
connection.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS e2ee_sessions (
owner_did TEXT NOT NULL DEFAULT '',
peer_did TEXT NOT NULL,
session_id TEXT NOT NULL,
is_initiator INTEGER NOT NULL DEFAULT 0,
send_chain_key TEXT NOT NULL,
recv_chain_key TEXT NOT NULL,
send_seq INTEGER NOT NULL DEFAULT 0,
recv_seq INTEGER NOT NULL DEFAULT 0,
expires_at REAL,
created_at TEXT NOT NULL,
active_at TEXT,
peer_confirmed INTEGER NOT NULL DEFAULT 0,
credential_name TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL,
PRIMARY KEY (owner_did, peer_did),
UNIQUE (owner_did, session_id)
);
"#,
)
}
fn create_database(sqlite_path: &Path) -> Connection {
crate::internal::local_state::open_writable(sqlite_path).expect("open database")
}
fn owner_identity_id(owner_did: &str) -> &'static str {
match owner_did {
"did:owner:new" => "final-owner-id",
"did:owner:old" => "old-id",
"did:owner:old-1" => "old-1-id",
"did:owner:old-2" => "old-2-id",
"did:owner:old-a" => "old-a-id",
"did:owner:old-z" => "old-z-id",
"did:owner:old-old" => "old-old-id",
"did:owner:other" => "other-id",
_ => "test-owner-id",
}
}
fn assert_zero_counts(counts: &BTreeMap<String, i64>, keys: &[&str]) {
assert_eq!(counts.len(), keys.len(), "count keys: {counts:?}");
for key in keys {
assert_eq!(counts[*key], 0, "zero count for {key}");
}
}
fn assert_table_exists(connection: &Connection, table: &str) {
let count: i64 = connection
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
rusqlite::params![table],
|row| row.get(0),
)
.expect("table lookup");
assert_eq!(count, 1, "{table} should exist");
}
fn assert_owner_count(connection: &Connection, table: &str, owner_did: &str, want: i64) {
let got = scalar_i64(
connection,
&format!("SELECT COUNT(*) FROM {table} WHERE owner_did = '{owner_did}'"),
);
assert_eq!(got, want, "owner count for {table}/{owner_did}");
}
fn assert_did_history(
connection: &Connection,
owner_identity_id: &str,
did: &str,
status: &str,
) {
let got = scalar_string(
connection,
&format!(
"SELECT status FROM identity_did_history WHERE owner_identity_id = '{owner_identity_id}' AND did = '{did}'"
),
);
assert_eq!(
got, status,
"DID history status for {owner_identity_id}/{did}"
);
}
fn scalar_string(connection: &Connection, sql: &str) -> String {
connection
.query_row(sql, [], |row| row.get::<_, String>(0))
.expect(sql)
}
fn scalar_opt_string(connection: &Connection, sql: &str) -> Option<String> {
connection
.query_row(sql, [], |row| row.get::<_, Option<String>>(0))
.expect(sql)
}
fn scalar_i64(connection: &Connection, sql: &str) -> i64 {
connection
.query_row(sql, [], |row| row.get::<_, i64>(0))
.expect(sql)
}
fn scalar_opt_f64(connection: &Connection, sql: &str) -> Option<f64> {
connection
.query_row(sql, [], |row| row.get::<_, Option<f64>>(0))
.expect(sql)
}
fn sqlite_path(root: &Path) -> std::path::PathBuf {
root.join("local").join("im.sqlite")
}
fn expected_thread_id(my_did: &str, peer_did: &str, group_id: &str) -> String {
helpers::make_thread_id(my_did, peer_did, group_id)
}
}