[Fix] User deletion & migration

src/api/response.rs

@@ -28,7 +28,7 @@ pub struct UserResponse {
     pub username: String,
     pub public_key: String,
     pub user_id: i64,
-    pub iota_id: i64,
+    pub iota_id: Option<i64>,
     pub sub_level: i32,
     pub sub_end: i64,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -47,7 +47,7 @@ pub struct UsernameResponse {
     pub username: String,
     pub public_key: String,
     pub user_id: i64,
-    pub iota_id: i64,
+    pub iota_id: Option<i64>,
     pub sub_level: i32,
     pub sub_end: i64,
 }

src/db/user_repo.rs

@@ -101,7 +101,7 @@ const USER_COLUMNS: &str = "SELECT id, iota_id, username, display, status, prese
 #[derive(FromRow)]
 struct UserRow {
     id: i64,
-    iota_id: i64,
+    iota_id: Option<i64>,
     username: Vec<u8>,
     display: Option<Vec<u8>>,
     status: Option<Vec<u8>>,
@@ -124,7 +124,7 @@ impl TryFrom<UserRow> for User {
             |value| String::from_utf8(value).map_err(|error| sqlx::Error::Decode(Box::new(error)));
         Ok(User {
             id: row.id.into(),
-            iota_id: row.iota_id.into(),
+            iota_id: row.iota_id.map(IotaId::from),
             username: decode(row.username)?,
             display: row.display.map(decode).transpose()?,
             status: row.status.map(decode).transpose()?,
@@ -344,9 +344,9 @@ pub async fn change_presence_preference(id: UserId, value: String) -> Result<()>
     .await
 }
 
-pub async fn change_iota_id(id: UserId, value: IotaId) -> Result<()> {
+pub async fn change_iota_id(id: UserId, value: Option<IotaId>) -> Result<()> {
     sqlx::query("UPDATE users SET iota_id = ? WHERE id = ?")
-        .bind(value.0)
+        .bind(value.map(|id| id.0))
         .bind(id.0)
         .execute(&pool().await?)
         .await?;
@@ -368,6 +368,42 @@ pub async fn delete_user(id: UserId) -> Result<()> {
     Ok(())
 }
 
+/// Delete the central identity while retaining a durable instruction for the
+/// last hosting Iota. The pending row is intentionally independent of users:
+/// it must outlive the account row.
+pub async fn delete_user_with_pending_erasure(id: UserId) -> Result<Option<IotaId>> {
+    let mut tx = pool().await?.begin().await?;
+    let row = sqlx::query("SELECT iota_id FROM users WHERE id = ? FOR UPDATE")
+        .bind(id.0)
+        .fetch_optional(&mut *tx)
+        .await?
+        .ok_or(OmegaError::NotFound)?;
+    let iota_id: Option<i64> = row.get("iota_id");
+    if let Some(iota_id) = iota_id {
+        sqlx::query("INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)")
+            .bind(id.0)
+            .bind(iota_id)
+            .execute(&mut *tx)
+            .await?;
+    }
+    sqlx::query("DELETE FROM registration_leases WHERE user_id = ?").bind(id.0).execute(&mut *tx).await?;
+    sqlx::query("DELETE FROM users WHERE id = ?").bind(id.0).execute(&mut *tx).await?;
+    tx.commit().await?;
+    Ok(iota_id.map(IotaId::from))
+}
+
+pub async fn pending_erasures_for_iota(iota_id: IotaId) -> Result<Vec<UserId>> {
+    let rows = sqlx::query("SELECT user_id FROM pending_iota_user_erasure WHERE iota_id = ?")
+        .bind(iota_id.0).fetch_all(&pool().await?).await?;
+    Ok(rows.into_iter().map(|row| UserId::from(row.get::<i64, _>("user_id"))).collect())
+}
+
+pub async fn acknowledge_pending_erasure(user_id: UserId, iota_id: IotaId) -> Result<bool> {
+    let result = sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?")
+        .bind(user_id.0).bind(iota_id.0).execute(&pool().await?).await?;
+    Ok(result.rows_affected() == 1)
+}
+
 pub async fn change_keys(id: UserId, public_key: PublicKeyBundle) -> Result<()> {
     sqlx::query("UPDATE users SET public_key = ? WHERE id = ?")
         .bind(public_key.as_bytes())
@@ -438,7 +474,7 @@ pub async fn register_complete_user(
                 .map_err(OmegaError::from)?
             {
                 Some(existing)
-                    if existing.iota_id == iota_id
+                    if existing.iota_id == Some(iota_id)
                         && existing.username == username
                         && existing.public_key.as_bytes() == public_key.as_bytes()
                         && existing.token == token =>

src/models/user.rs

@@ -4,7 +4,7 @@ use mtp::crypto::PublicKeyBundle;
 #[derive(Clone, Debug, serde::Serialize)]
 pub struct User {
     pub id: UserId,
-    pub iota_id: IotaId,
+    pub iota_id: Option<IotaId>,
     pub username: String,
     pub display: Option<String>,
     pub status: Option<String>,

src/server/api.rs

@@ -41,7 +41,7 @@ fn user_response(user: crate::models::User) -> UserResponse {
         username: user.username,
         public_key: user.public_key.to_base64(),
         user_id: user.id.0,
-        iota_id: user.iota_id.0,
+        iota_id: user.iota_id.map(|id| id.0),
         sub_level: user.sub_level,
         sub_end: user.sub_end,
         display: user.display,
@@ -83,7 +83,15 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> {
                 omikron_id
             } else {
                 let user = get_by_user_id(UserId::from(id)).await?;
-                get_iota_primary_omikron_connection(user.iota_id.0).ok_or(OmegaError::NotFound)?
+                match user.iota_id {
+                    Some(iota_id) => get_iota_primary_omikron_connection(iota_id.0),
+                    None => get_random_omikron()
+                        .await
+                        .map_err(|_| OmegaError::NotFound)?
+                        .get_omikron_id()
+                        .await,
+                }
+                .ok_or(OmegaError::NotFound)?
             };
 
             // Database rows describe registered Omikrons. The public discovery
@@ -150,7 +158,7 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> {
                     username: user.username,
                     public_key: user.public_key.to_base64(),
                     user_id: user.id.0,
-                    iota_id: user.iota_id.0,
+                    iota_id: user.iota_id.map(|id| id.0),
                     sub_level: user.sub_level,
                     sub_end: user.sub_end,
                 }),

src/state.rs

@@ -1,18 +1,47 @@
 use crate::sql::user_online_tracker::PresenceTracker;
 use std::sync::Arc;
+use dashmap::DashMap;
+use std::time::{Duration, Instant};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum AccountChallengeOperation { Attach, Delete }
+
+#[derive(Clone, Debug)]
+pub struct AccountChallenge {
+    pub operation: AccountChallengeOperation,
+    pub user_id: i64,
+    pub requester_iota_id: i64,
+    pub nonce: u64,
+    pub created_at: Instant,
+}
 
 pub struct OmegaState {
     pub presence: Arc<PresenceTracker>,
+    challenges: DashMap<(AccountChallengeOperation, i64, i64), AccountChallenge>,
 }
 
 impl Default for OmegaState {
     fn default() -> Self {
         Self {
             presence: Arc::new(PresenceTracker::default()),
+            challenges: DashMap::new(),
         }
     }
 }
 
+impl OmegaState {
+    pub fn issue_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64) -> u64 {
+        let nonce = rand::random::<u64>();
+        self.challenges.insert((operation, user_id, requester_iota_id), AccountChallenge { operation, user_id, requester_iota_id, nonce, created_at: Instant::now() });
+        nonce
+    }
+
+    pub fn consume_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64, nonce: u64) -> bool {
+        self.challenges.remove(&(operation, user_id, requester_iota_id)).is_some_and(|(_, value)|
+            value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120))
+    }
+}
+
 impl OmegaState {
     pub fn new() -> Arc<Self> {
         Arc::new(Self::default())

src/transport/handlers/account.rs

@@ -2,8 +2,9 @@ use super::super::omikron_connection::{OmikronConnection, OmikronResult};
 use crate::{
     db::{iota_repo, user_repo},
     models::{IotaId, UserId},
+    state::AccountChallengeOperation,
 };
-use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
+use mtp::{codec::{CommunicationType, CommunicationValue, DataType, DataValue}, crypto::{verify_ed25519, verify_ml_dsa}};
 use std::sync::Arc;
 
 async fn delete(
@@ -23,12 +24,8 @@ pub async fn user(
     connection: Arc<OmikronConnection>,
     value: CommunicationValue,
 ) -> OmikronResult<()> {
-    delete(
-        connection,
-        value.clone(),
-        user_repo::delete_user(UserId::from(value.get_sender() as i64)),
-    )
-    .await
+    let user_id = UserId::from(value.get_sender() as i64);
+    complete_delete(connection, value, user_id).await
 }
 pub async fn iota(
     connection: Arc<OmikronConnection>,
@@ -41,3 +38,157 @@ pub async fn iota(
     )
     .await
 }
+
+pub async fn release_from_iota(
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|id| i64::try_from(id).ok())
+        .filter(|id| *id > 0)
+    else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
+            .await;
+    };
+    let requester = IotaId::from(value.get_sender() as i64);
+    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
+            .await;
+    };
+    if user.iota_id != Some(requester) {
+        return connection
+            .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
+            .await;
+    }
+    let previous_iota = user.iota_id;
+    match user_repo::change_iota_id(user.id, None).await {
+        Ok(()) => {
+            if let Some(iota) = previous_iota { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; }
+            connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await
+        },
+        Err(error) => connection
+            .send(&CommunicationValue::new(CommunicationType::ErrorInternal)
+                .with_id(value.get_id())
+                .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())))
+            .await,
+    }
+}
+
+fn lifecycle_payload(domain: &[u8], user_id: i64, iota_id: i64, nonce: u64) -> Vec<u8> {
+    let mut payload = Vec::with_capacity(domain.len() + 32);
+    payload.extend_from_slice(domain);
+    payload.extend_from_slice(&user_id.to_be_bytes());
+    payload.extend_from_slice(&iota_id.to_be_bytes());
+    payload.extend_from_slice(&nonce.to_be_bytes());
+    payload
+}
+
+pub async fn attach_begin(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else {
+        return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await;
+    };
+    if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() {
+        return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await;
+    }
+    let requester = value.get_sender() as i64;
+    let nonce = connection.state().issue_challenge(AccountChallengeOperation::Attach, user_id, requester);
+    connection.send(&CommunicationValue::new(CommunicationType::AttachUserChallenge).with_id(value.get_id())
+        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+        .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await
+}
+
+pub async fn attach_complete(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
+    let requester = value.get_sender() as i64;
+    let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    let signature = value.get_data(DataType::Signature).as_bytes();
+    let pq_signature = value.get_data(DataType::PqSignature).as_bytes();
+    let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    if !connection.state().consume_challenge(AccountChallengeOperation::Attach, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }
+    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; };
+    let payload = lifecycle_payload(b"tensamin:user-attach:v1\0", user_id, requester, nonce);
+    if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; }
+    let previous_iota = user.iota_id;
+    match user_repo::change_iota_id(user.id, Some(IotaId::from(requester))).await {
+        Ok(()) => {
+            if let Some(iota) = previous_iota.filter(|id| id.0 != requester) { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; }
+            crate::transport::omikron_manager::publish_iota_user_snapshot(requester).await;
+            connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await
+        },
+        Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await,
+    }
+}
+
+async fn complete_delete(connection: Arc<OmikronConnection>, value: CommunicationValue, user_id: UserId) -> OmikronResult<()> {
+    match user_repo::delete_user_with_pending_erasure(user_id).await {
+        Ok(iota_id) => {
+            let cleanup_pending = iota_id.is_some();
+            if let Some(iota_id) = iota_id {
+                crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id.0).await;
+                crate::transport::omikron_manager::deliver_pending_erasures(iota_id.0).await;
+            }
+            connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())
+                .add_typed_default(DataType::CleanupPending, DataValue::Bool(cleanup_pending))).await
+        }
+        Err(crate::error::OmegaError::NotFound) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await,
+        Err(error) => connection.send(&CommunicationValue::new(CommunicationType::ErrorInternal).with_id(value.get_id()).add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string()))).await,
+    }
+}
+
+pub async fn delete_credential_begin(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else {
+        return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await;
+    };
+    if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; }
+    let requester = value.get_sender() as i64;
+    let nonce = connection.state().issue_challenge(AccountChallengeOperation::Delete, user_id, requester);
+    connection.send(&CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge).with_id(value.get_id())
+        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+        .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await
+}
+
+pub async fn delete_credential_complete(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
+    let requester = value.get_sender() as i64;
+    let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    let (Some(signature), Some(pq_signature)) = (value.get_data(DataType::Signature).as_bytes(), value.get_data(DataType::PqSignature).as_bytes()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
+    if !connection.state().consume_challenge(AccountChallengeOperation::Delete, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }
+    let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; };
+    let payload = lifecycle_payload(b"tensamin:user-delete:v1\0", user_id, requester, nonce);
+    if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; }
+    complete_delete(connection, value, user.id).await
+}
+
+pub async fn erase_hosted_user_data_ack(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
+    let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
+    let iota_id = IotaId::from(value.get_sender() as i64);
+    match user_repo::acknowledge_pending_erasure(UserId::from(user_id), iota_id).await {
+        Ok(true) => connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await,
+        Ok(false) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await,
+        Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await,
+    }
+}
+
+/// New lifecycle operation names are intentionally fail-closed until their
+/// proof and durable-erasure handlers are enabled.  This explicit dispatch
+/// prevents either a bare Iota request or the legacy DeleteUser path from
+/// acquiring account-deletion authority during a staged rollout.
+pub async fn lifecycle_unavailable(
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    connection
+        .send(
+            &CommunicationValue::new(CommunicationType::ErrorNotAuthenticated)
+                .with_id(value.get_id())
+                .add_typed_default(
+                    DataType::ErrorType,
+                    DataValue::Str("user lifecycle proof handler is not enabled".into()),
+                ),
+        )
+        .await
+}

src/transport/handlers/presence.rs

@@ -53,7 +53,7 @@ fn states_for_users(state: &OmegaState, users: &[crate::models::User]) -> HashMa
                 user.id.0,
                 state
                     .presence
-                    .resolve_public_state(user.id.0, user.iota_id.0),
+                    .resolve_public_state(user.id.0, user.iota_id.map(|id| id.0).unwrap_or_default()),
             )
         })
         .collect()
@@ -69,7 +69,7 @@ fn changed_states(
         .filter_map(|user| {
             let after = state
                 .presence
-                .resolve_public_state(user.id.0, user.iota_id.0);
+                .resolve_public_state(user.id.0, user.iota_id.map(|id| id.0).unwrap_or_default());
             (before.get(&user.id.0) != Some(&after)).then_some((user.id.0, after))
         })
         .collect::<Vec<_>>();
@@ -275,7 +275,7 @@ pub async fn user_connected(
         Ok(preferences) => preferences,
         Err(error) => return Err(error.into()),
     };
-    if user.iota_id.0 != iota_id || !state.presence.has_iota_route(iota_id) {
+    if user.iota_id.map(|id| id.0) != Some(iota_id) || !state.presence.has_iota_route(iota_id) {
         return connection
             .send_error_response(value.get_id(), CommunicationType::ErrorNoIota)
             .await;
@@ -497,6 +497,7 @@ pub async fn iota_connected(
         .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
         .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
     connection.clone().send(&response).await?;
+    crate::transport::omikron_manager::deliver_pending_erasures(iota_id).await;
     publish_changed_states(&state, &before, &users).await;
     connection
         .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))

src/transport/handlers/states.rs

@@ -110,7 +110,7 @@ pub async fn get(
         };
         let status = state
             .presence
-            .resolve_public_state(user_id, user.iota_id.0)
+            .resolve_public_state(user_id, user.iota_id.map(|id| id.0).unwrap_or_default())
             .to_string();
         let mut map = Vec::new();
         if let Some(kind) = DataType::UserId.try_to_id(&tm) {

src/transport/handlers/user_data.rs

@@ -43,7 +43,7 @@ pub async fn get_user(
             .await;
     };
     let id = user.id.0;
-    let iota_id = user.iota_id.0;
+    let iota_id = user.iota_id.map(|id| id.0);
     let username = user.username.clone();
     let display = user
         .display
@@ -57,7 +57,6 @@ pub async fn get_user(
             DataValue::Str(user.public_key.to_base64()),
         )
         .add_typed_default(DataType::UserId, DataValue::SignedNumber(id.into()))
-        .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
         .add_typed_default(DataType::Display, DataValue::Str(display))
         .add_typed_default(
             DataType::SubLevel,
@@ -92,7 +91,7 @@ pub async fn get_user(
         }
         state.presence.resolve_private_state(id)
     } else {
-        state.presence.resolve_public_state(id, iota_id)
+        iota_id.map(|iota_id| state.presence.resolve_public_state(id, iota_id)).unwrap_or(crate::sql::connection_status::UserStatus::user_offline)
     };
     response = response
         .add_typed_default(
@@ -101,8 +100,11 @@ pub async fn get_user(
         )
         .add_typed_default(
             DataType::OmikronConnections,
-            connections(&connection, iota_id),
+            iota_id.map(|iota_id| connections(&connection, iota_id)).unwrap_or_else(|| DataValue::Array(Vec::new())),
         );
+    if let Some(iota_id) = iota_id {
+        response = response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
+    }
     if let Some(route) = route {
         response = response.add_typed_default(
             DataType::OmikronId,
@@ -123,26 +125,25 @@ pub async fn get_iota(
             .map(|iota| (iota.id.0, iota.public_key, None, None))
     } else if let Some(id) = value.get_data(DataType::UserId).as_number() {
         if let Ok(user) = user_repo::get_by_user_id(UserId::from(id as i64)).await {
-            iota_repo::get_iota_by_id(user.iota_id)
-                .await
-                .ok()
-                .map(|iota| (iota.id.0, iota.public_key, Some(user.id.0), None))
+            match user.iota_id {
+                Some(iota_id) => iota_repo::get_iota_by_id(iota_id)
+                    .await
+                    .ok()
+                    .map(|iota| (iota.id.0, iota.public_key, Some(user.id.0), None)),
+                None => None,
+            }
         } else {
             None
         }
     } else if let Some(name) = value.get_data(DataType::Username).as_str() {
         if let Ok(user) = user_repo::get_by_username(name).await {
-            iota_repo::get_iota_by_id(user.iota_id)
-                .await
-                .ok()
-                .map(|iota| {
-                    (
-                        iota.id.0,
-                        iota.public_key,
-                        Some(user.id.0),
-                        Some(name.to_owned()),
-                    )
-                })
+            match user.iota_id {
+                Some(iota_id) => iota_repo::get_iota_by_id(iota_id)
+                    .await
+                    .ok()
+                    .map(|iota| (iota.id.0, iota.public_key, Some(user.id.0), Some(name.to_owned()))),
+                None => None,
+            }
         } else {
             None
         }
@@ -269,7 +270,7 @@ pub async fn change_iota(
             .await;
     }
     let result =
-        match user_repo::change_iota_id(user_id, IotaId::from(value.get_sender() as i64)).await {
+        match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64))).await {
             Ok(()) => user_repo::change_token(user_id, new_token.to_owned()).await,
             Err(error) => Err(error),
         };

src/transport/omikron_connection.rs

@@ -87,6 +87,7 @@ impl OmikronConnection {
         &self.peer_capabilities
     }
 
+
     pub async fn handle(self: Arc<Self>, receiver: &mut WebMtpReceiver) {
         log_in!(
             self.id as i64,
@@ -215,6 +216,24 @@ impl OmikronConnection {
             Some(CommunicationType::DeleteUser) => {
                 crate::transport::handlers::account::user(self, value).await
             }
+            Some(CommunicationType::AttachUserBegin) => {
+                crate::transport::handlers::account::attach_begin(self, value).await
+            }
+            Some(CommunicationType::AttachUserComplete) => {
+                crate::transport::handlers::account::attach_complete(self, value).await
+            }
+            Some(CommunicationType::DeleteUserCredentialBegin) => {
+                crate::transport::handlers::account::delete_credential_begin(self, value).await
+            }
+            Some(CommunicationType::DeleteUserCredentialComplete) => {
+                crate::transport::handlers::account::delete_credential_complete(self, value).await
+            }
+            Some(CommunicationType::EraseHostedUserDataAck) => {
+                crate::transport::handlers::account::erase_hosted_user_data_ack(self, value).await
+            }
+            Some(CommunicationType::ReleaseUserFromIota) => {
+                crate::transport::handlers::account::release_from_iota(self, value).await
+            }
             Some(CommunicationType::DeleteIota) => {
                 crate::transport::handlers::account::iota(self, value).await
             }

src/transport/omikron_manager.rs

@@ -3,7 +3,7 @@ use crate::state::OmegaState;
 use crate::transport::connection::OmikronConnection;
 use crate::transport::omikron_connection::OmikronResult;
 use dashmap::DashMap;
-use mtp::codec::CommunicationValue;
+use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use once_cell::sync::Lazy;
 use rand::prelude::IteratorRandom;
 use std::sync::Arc;
@@ -65,7 +65,8 @@ pub async fn get_all_connections()
             for user in users {
                 for route in state.presence.routes_for_user(user.id.0) {
                     if let Some(iotas) = result.get_mut(&route.omikron_id) {
-                        if let Some(users) = iotas.get_mut(&user.iota_id.0) {
+                        if let Some(iota_id) = user.iota_id
+                            && let Some(users) = iotas.get_mut(&iota_id.0) {
                             users.push(user.id.0);
                         }
                     }
@@ -113,3 +114,28 @@ pub async fn send_to_user(user_id: i64, cv: &CommunicationValue) {
         }
     }
 }
+
+/// Publish the authoritative membership list after an attach, migration, or
+/// release. Omikron replaces its full local index from this snapshot.
+pub async fn publish_iota_user_snapshot(iota_id: i64) {
+    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else { return; };
+    let Some(connection) = get_connected_omikron(omikron_id) else { return; };
+    let Ok(users) = user_repo::get_users_by_iota_id(crate::models::IotaId::from(iota_id)).await else { return; };
+    let user_ids = users.into_iter().map(|user| DataValue::SignedNumber(user.id.0.into())).collect();
+    let snapshot = CommunicationValue::new(CommunicationType::IotaUserData)
+        .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
+        .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
+    let _ = connection.send(&snapshot).await;
+}
+
+pub async fn deliver_pending_erasures(iota_id: i64) {
+    let Ok(users) = user_repo::pending_erasures_for_iota(crate::models::IotaId::from(iota_id)).await else { return; };
+    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else { return; };
+    let Some(connection) = get_connected_omikron(omikron_id) else { return; };
+    for user_id in users {
+        let request = CommunicationValue::new(CommunicationType::EraseHostedUserData)
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.0.into()))
+            .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
+        let _ = connection.clone().send(&request).await;
+    }
+}