[Fix] Durability

client/src/client_connection.rs

@@ -226,11 +226,6 @@ impl ClientConnection {
         // Direct messages                                  //
         // ************************************************ //
 
-        if cv.is_type(CommunicationType::MessageState) {
-            message_handlers::handle_message_state(&cv);
-            return;
-        }
-
         if cv.is_type(CommunicationType::MessageEdit) {
             self.send_message(&message_handlers::handle_message_edit(&cv))
                 .await;

iota-connection/src/message_handlers.rs

@@ -72,7 +72,6 @@ fn add_conversation_for_user(
     if let Some(name) = name {
         contact.user_name = Some(name.to_string());
     }
-    contact.set_last_message_at(now_millis_i64());
     mod_user(user_id, &contact)
 }
 
@@ -228,7 +227,11 @@ pub fn apply_verified_relay_content(
                 reply_to,
                 origin_iota_received_at: sent_by_self.then_some(accepted_at),
                 destination_iota_received_at: (!sent_by_self).then_some(accepted_at),
-                initial_state: MessageState::Sent,
+                initial_state: if sent_by_self {
+                    MessageState::Sending
+                } else {
+                    MessageState::Sent
+                },
             })
             .map_err(|error| error.to_string())?;
             Ok(())
@@ -430,6 +433,12 @@ fn stored_message_fields(
             DataValue::Str(relay_message_id.clone()),
         ));
     }
+    if let Some(relay_signer_id) = message.relay_signer_id {
+        fields.push((
+            DataType::SenderId,
+            DataValue::SignedNumber(relay_signer_id.into()),
+        ));
+    }
     for (data_type, timestamp) in [
         (DataType::AuthoredAt, message.authored_at),
         (
@@ -452,6 +461,15 @@ fn stored_message_fields(
             fields.push((data_type, DataValue::SignedNumber(timestamp.into())));
         }
     }
+    if let Some(failed_at) = message.delivery_failed_at {
+        fields.push((
+            DataType::UpdatedAt,
+            DataValue::SignedNumber(failed_at.into()),
+        ));
+    }
+    if let Some(failure) = &message.delivery_failure {
+        fields.push((DataType::ErrorType, DataValue::Str(failure.clone())));
+    }
     if message.edited {
         fields.push((DataType::Edited, DataValue::Bool(true)));
     }
@@ -644,6 +662,12 @@ fn contact_value(
     if let Some(name) = &contact.user_name {
         fields.push((DataType::Username, DataValue::Str(name.clone())));
     }
+    if contact.created_at > 0 {
+        fields.push((
+            DataType::CreatedAt,
+            DataValue::SignedNumber(contact.created_at.into()),
+        ));
+    }
     if let Some(last_message_at) = contact.last_message_at {
         fields.push((
             DataType::LastMessageAt,
@@ -918,26 +942,6 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue {
         )
 }
 
-pub fn handle_message_state(cv: &CommunicationValue) {
-    let sender_id = match required_sender_id(cv) {
-        Ok(sender_id) => sender_id,
-        Err(_) => return,
-    };
-    let receiver_id = match data_i64(cv, DataType::ChatPartnerId) {
-        Some(id) if id > 0 => id,
-        _ => return,
-    };
-
-    let timestamp_i64 = data_i64(cv, DataType::SendTime).unwrap_or_else(now_millis_i64);
-
-    let _ = chat_files::change_message_state(
-        timestamp_i64,
-        receiver_id,
-        sender_id,
-        MessageState::from_str(cv.get_data(DataType::MessageState).as_str().unwrap_or("")),
-    );
-}
-
 pub fn handle_messages_get(cv: &CommunicationValue) -> CommunicationValue {
     let my_id = match cv.require_sender() {
         Ok(my_id) => my_id,
@@ -1026,6 +1030,12 @@ pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue {
         if let Some(name) = user.user_name {
             container.push((DataType::Username, DataValue::Str(name)));
         }
+        if user.created_at > 0 {
+            container.push((
+                DataType::CreatedAt,
+                DataValue::SignedNumber(user.created_at.into()),
+            ));
+        }
         if let Some(ts) = user.last_message_at {
             container.push((DataType::LastMessageAt, DataValue::SignedNumber(ts as i128)));
         }

iota-storage/src/users/contact.rs

@@ -1,31 +1,32 @@
-use std::time::{SystemTime, UNIX_EPOCH};
-
 #[derive(Debug, Clone)]
 pub struct Contact {
     pub user_id: i64,
     pub user_name: Option<String>,
+    pub created_at: i64,
     pub last_message_at: Option<i64>,
 }
 
 impl Default for Contact {
     fn default() -> Self {
-        let now = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_millis() as i64;
         Contact {
             user_id: 0,
             user_name: None,
-            last_message_at: Some(now),
+            created_at: 0,
+            last_message_at: None,
         }
     }
 }
 
 impl Contact {
     pub fn new(user_id: i64) -> Self {
+        let created_at = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis() as i64;
         Contact {
             user_id: user_id,
             user_name: None,
+            created_at,
             last_message_at: None,
         }
     }

iota-storage/src/users/user_profile.rs

@@ -27,6 +27,29 @@ impl UserProfile {
         public_key: String,
         private_key_hash: String,
         reset_token: String,
+    ) -> Self {
+        Self::new_with_created_at(
+            user_id,
+            username,
+            display_name,
+            public_key,
+            private_key_hash,
+            reset_token,
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_millis() as i64,
+        )
+    }
+
+    pub fn new_with_created_at(
+        user_id: i64,
+        username: String,
+        display_name: Option<String>,
+        public_key: String,
+        private_key_hash: String,
+        reset_token: String,
+        created_at: i64,
     ) -> Self {
         Self {
             user_id,
@@ -34,10 +57,7 @@ impl UserProfile {
             display_name,
             public_key,
             private_key_hash,
-            created_at: SystemTime::now()
+            created_at,
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_millis() as i64,
             reset_token,
             trusted_apps: std::collections::HashMap::new(),
         }

iota-storage/src/util/chat_files.rs

@@ -60,6 +60,8 @@ pub struct StoredMessage {
     pub client_received_recorded_at: Option<i64>,
     pub read_at: Option<i64>,
     pub read_recorded_at: Option<i64>,
+    pub delivery_failed_at: Option<i64>,
+    pub delivery_failure: Option<String>,
     pub content: String,
     pub edited: bool,
     pub sent_by_self: bool,
@@ -483,7 +485,7 @@ pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
         destination_iota_received_at,
         initial_state,
     } = message;
-    let msg_id = db::with_db(|conn| {
+    db::with_db(|conn| {
         let tx = conn.unchecked_transaction()?;
         tx.execute(
             r#"
@@ -517,14 +519,16 @@ pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
             msg_id,
             Operation::Upsert,
         )?;
+        let mut contact = crate::users::contact::Contact::new(external_user);
+        contact.set_last_message_at(
+            destination_iota_received_at
+                .or(origin_iota_received_at)
+                .unwrap_or(authored_at),
+        );
+        crate::util::chats_util::upsert_contact(&tx, storage_owner, &contact)?;
         tx.commit()?;
         Ok(msg_id)
-    })?;
+    })
-
-    let mut contact = crate::users::contact::Contact::new(external_user);
-    contact.set_last_message_at(send_time);
-    crate::util::chats_util::mod_user(storage_owner, &contact)?;
-    Ok(msg_id)
 }
 
 pub fn change_message_state_by_relay_id(
@@ -579,11 +583,11 @@ pub fn record_message_receipt(
     };
     db::with_db(|conn| {
         let tx = conn.unchecked_transaction()?;
-        let Some((message_id, external_user)) = tx
+        let Some((message_id, external_user, authored_at)) = tx
             .query_row(
-                "SELECT id, external_user FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
+                "SELECT id, external_user, authored_at FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
                 params![storage_owner, target_signer_id, target_message_id],
-                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
+                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?, row.get::<_, Option<i64>>(2)?)),
             )
             .optional()?
         else {
@@ -594,6 +598,13 @@ pub fn record_message_receipt(
                 "message receipt signer is not the chat partner".into(),
             ));
         }
+        if event_at > recorded_at.saturating_add(5 * 60 * 1000)
+            || authored_at.is_some_and(|authored_at| event_at < authored_at)
+        {
+            return Err(StorageError::Other(
+                "message receipt event time is outside the accepted clock range".into(),
+            ));
+        }
         tx.execute(
             "INSERT OR IGNORE INTO message_receipts (storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
             params![storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at],
@@ -646,7 +657,7 @@ pub fn record_destination_iota_received(
             return Ok(());
         };
         tx.execute(
-            "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2",
+            "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), delivery_failed_at = NULL, delivery_failure = NULL, message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2",
             params![accepted_at, message_id],
         )?;
         sync::record_event(
@@ -661,6 +672,41 @@ pub fn record_destination_iota_received(
     })
 }
 
+pub fn record_delivery_failure(
+    storage_owner: i64,
+    relay_signer_id: i64,
+    relay_message_id: &str,
+    failure: &str,
+    failed_at: i64,
+) -> Result<(), StorageError> {
+    db::with_db(|conn| {
+        let tx = conn.unchecked_transaction()?;
+        let Some(message_id) = tx
+            .query_row(
+                "SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
+                params![storage_owner, relay_signer_id, relay_message_id],
+                |row| row.get::<_, i64>(0),
+            )
+            .optional()?
+        else {
+            return Ok(());
+        };
+        tx.execute(
+            "UPDATE messages SET delivery_failed_at = ?1, delivery_failure = ?2 WHERE id = ?3 AND destination_iota_received_at IS NULL",
+            params![failed_at, failure, message_id],
+        )?;
+        sync::record_event(
+            &tx,
+            storage_owner,
+            EntityType::Message,
+            message_id,
+            Operation::Upsert,
+        )?;
+        tx.commit()?;
+        Ok(())
+    })
+}
+
 pub fn change_message_state(
     timestamp: i64,
     storage_owner: i64,
@@ -783,11 +829,11 @@ pub fn get_messages(
             SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
                    origin_iota_received_at, destination_iota_received_at,
                    client_received_at, client_received_recorded_at, read_at,
-                   read_recorded_at, content, sent_by_self, message_state, height,
+                   read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height,
                    reply_to, edited_count
             FROM messages
             WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0
-            ORDER BY message_time DESC, id DESC
+            ORDER BY COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) DESC, id DESC
             LIMIT ?3 OFFSET ?4
             "#,
         )?;
@@ -808,12 +854,14 @@ pub fn get_messages(
                     client_received_recorded_at: row.get(8)?,
                     read_at: row.get(9)?,
                     read_recorded_at: row.get(10)?,
-                    content: row.get(11)?,
+                    delivery_failed_at: row.get(11)?,
-                    sent_by_self: row.get::<_, i64>(12)? != 0,
+                    delivery_failure: row.get(12)?,
-                    message_state: row.get(13)?,
+                    content: row.get(13)?,
-                    height: row.get(14).unwrap_or(0),
+                    sent_by_self: row.get::<_, i64>(14)? != 0,
-                    reply_to: row.get(15).ok().flatten(),
+                    message_state: row.get(15)?,
-                    edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
+                    height: row.get(16).unwrap_or(0),
+                    reply_to: row.get(17).ok().flatten(),
+                    edited: row.get::<_, i64>(18).unwrap_or(0) > 0,
                     reactions: Vec::new(),
                 })
             },
@@ -854,7 +902,7 @@ pub fn get_message(
             SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
                    origin_iota_received_at, destination_iota_received_at,
                    client_received_at, client_received_recorded_at, read_at,
-                   read_recorded_at, content, sent_by_self, message_state, height,
+                   read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height,
                    reply_to, edited_count, external_user
             FROM messages
             WHERE storage_owner = ?1
@@ -878,13 +926,15 @@ pub fn get_message(
                 client_received_recorded_at: row.get(8)?,
                 read_at: row.get(9)?,
                 read_recorded_at: row.get(10)?,
-                content: row.get(11)?,
+                delivery_failed_at: row.get(11)?,
-                sent_by_self: row.get::<_, i64>(12)? != 0,
+                delivery_failure: row.get(12)?,
-                message_state: row.get(13)?,
+                content: row.get(13)?,
-                height: row.get(14).unwrap_or(0),
+                sent_by_self: row.get::<_, i64>(14)? != 0,
-                reply_to: row.get(15).ok().flatten(),
+                message_state: row.get(15)?,
-                edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
+                height: row.get(16).unwrap_or(0),
-                external_user: row.get(17)?,
+                reply_to: row.get(17).ok().flatten(),
+                edited: row.get::<_, i64>(18).unwrap_or(0) > 0,
+                external_user: row.get(19)?,
                 reactions: Vec::new(),
             })
         })?;
@@ -928,14 +978,16 @@ pub fn get_message_with_offset(
               AND external_user = ?2
               AND deleted_by_external = 0
               AND (
-                   message_time > ?3
+                   COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) >
-                   OR (message_time = ?3 AND id > ?4)
+                       COALESCE(?3, ?4)
+                   OR (COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) =
+                       COALESCE(?3, ?4) AND id > ?4)
               )
             "#,
             params![
                 storage_owner,
                 external_user,
-                message.message_time,
+                message.destination_iota_received_at.or(message.origin_iota_received_at).or(message.authored_at),
                 message.id
             ],
             |row| row.get(0),
@@ -953,9 +1005,9 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<StoredMessage
     // A journal id uniquely identifies a row. Load all messages for this owner and retain only
     // those ids; this keeps reaction hydration identical to normal message loading.
     match db::with_db(|conn| {
-        let mut stmt = conn.prepare("SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, read_recorded_at, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0")?;
+        let mut stmt = conn.prepare("SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0")?;
         let rows = stmt.query_map([storage_owner], |row| {
-            let external_user: i64 = row.get(17)?;
+            let external_user: i64 = row.get(19)?;
             Ok(StoredMessage {
                 id: row.get(0)?,
                 external_user,
@@ -969,12 +1021,14 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<StoredMessage
                 client_received_recorded_at: row.get(8)?,
                 read_at: row.get(9)?,
                 read_recorded_at: row.get(10)?,
-                content: row.get(11)?,
+                delivery_failed_at: row.get(11)?,
-                sent_by_self: row.get::<_, i64>(12)? != 0,
+                delivery_failure: row.get(12)?,
-                message_state: row.get(13)?,
+                content: row.get(13)?,
-                height: row.get(14).unwrap_or(0),
+                sent_by_self: row.get::<_, i64>(14)? != 0,
-                reply_to: row.get(15).ok().flatten(),
+                message_state: row.get(15)?,
-                edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
+                height: row.get(16).unwrap_or(0),
+                reply_to: row.get(17).ok().flatten(),
+                edited: row.get::<_, i64>(18).unwrap_or(0) > 0,
                 reactions: Vec::new(),
             })
         })?;

iota-storage/src/util/chats_util.rs

@@ -4,6 +4,42 @@ use crate::util::db;
 use crate::util::sync::{self, EntityType, Operation};
 use rusqlite::params;
 
+pub(crate) fn upsert_contact(
+    tx: &rusqlite::Transaction<'_>,
+    storage_owner: i64,
+    contact: &Contact,
+) -> Result<(), StorageError> {
+    tx.execute(
+        r#"
+        INSERT INTO contacts (storage_owner, user_id, user_name, created_at, last_message_at)
+        VALUES (?1, ?2, ?3, ?4, ?5)
+        ON CONFLICT(storage_owner, user_id) DO UPDATE SET
+            user_name = COALESCE(excluded.user_name, contacts.user_name),
+            created_at = MIN(contacts.created_at, excluded.created_at),
+            last_message_at = CASE
+                WHEN excluded.last_message_at IS NULL THEN contacts.last_message_at
+                WHEN contacts.last_message_at IS NULL THEN excluded.last_message_at
+                ELSE MAX(contacts.last_message_at, excluded.last_message_at)
+            END
+        "#,
+        params![
+            storage_owner,
+            contact.user_id,
+            contact.user_name,
+            contact.created_at,
+            contact.last_message_at,
+        ],
+    )?;
+    sync::record_event(
+        tx,
+        storage_owner,
+        EntityType::Contact,
+        contact.user_id,
+        Operation::Upsert,
+    )?;
+    Ok(())
+}
+
 pub fn has_user(storage_owner: i64, user_id: i64) -> Result<bool, StorageError> {
     db::with_db(|conn| {
         Ok(conn.query_row(
@@ -15,31 +51,7 @@ pub fn has_user(storage_owner: i64, user_id: i64) -> Result<bool, StorageError>
 }
 
 pub fn mod_user(storage_owner: i64, contact: &Contact) -> Result<(), StorageError> {
-    db::with_immediate_transaction(|tx| {
+    db::with_immediate_transaction(|tx| upsert_contact(tx, storage_owner, contact))
-        tx.execute(
-            r#"
-            INSERT INTO contacts (storage_owner, user_id, user_name, last_message_at)
-            VALUES (?1, ?2, ?3, ?4)
-            ON CONFLICT(storage_owner, user_id) DO UPDATE SET
-                user_name = excluded.user_name,
-                last_message_at = excluded.last_message_at
-            "#,
-            params![
-                storage_owner,
-                contact.user_id,
-                contact.user_name,
-                contact.last_message_at,
-            ],
-        )?;
-        sync::record_event(
-            tx,
-            storage_owner,
-            EntityType::Contact,
-            contact.user_id,
-            Operation::Upsert,
-        )?;
-        Ok(())
-    })
 }
 
 pub fn get_users_by_ids(storage_owner: i64, ids: &[i64]) -> Result<Vec<Contact>, StorageError> {
@@ -57,7 +69,7 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Result<Option<Contact>, Sto
     db::with_db(|conn| {
         match conn.query_row(
             r#"
-            SELECT user_id, user_name, last_message_at
+            SELECT user_id, user_name, created_at, last_message_at
             FROM contacts
             WHERE storage_owner = ?1 AND user_id = ?2
             LIMIT 1
@@ -67,7 +79,8 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Result<Option<Contact>, Sto
                 Ok(Contact {
                     user_id: r.get(0)?,
                     user_name: r.get(1)?,
-                    last_message_at: r.get(2)?,
+                    created_at: r.get(2)?,
+                    last_message_at: r.get(3)?,
                 })
             },
         ) {
@@ -82,7 +95,7 @@ pub fn get_users(storage_owner: i64) -> Result<Vec<Contact>, StorageError> {
     db::with_db(|conn| {
         let mut stmt = conn.prepare(
             r#"
-            SELECT user_id, user_name, last_message_at
+            SELECT user_id, user_name, created_at, last_message_at
             FROM contacts
             WHERE storage_owner = ?1
             ORDER BY
@@ -96,7 +109,8 @@ pub fn get_users(storage_owner: i64) -> Result<Vec<Contact>, StorageError> {
             Ok(Contact {
                 user_id: r.get(0)?,
                 user_name: r.get(1)?,
-                last_message_at: r.get(2)?,
+                created_at: r.get(2)?,
+                last_message_at: r.get(3)?,
             })
         })?;
 

iota-storage/src/util/db.rs

@@ -180,6 +180,7 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
                 storage_owner INTEGER NOT NULL,
                 user_id INTEGER NOT NULL,
                 user_name TEXT,
+                created_at INTEGER NOT NULL,
                 last_message_at INTEGER,
                 UNIQUE(storage_owner, user_id)
             );
@@ -527,6 +528,57 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
         )?;
     }
 
+    if current_version < 15 {
+        let messages_exist: bool = conn.query_row(
+            "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')",
+            [],
+            |row| row.get(0),
+        )?;
+        if messages_exist {
+            conn.execute_batch(
+                r#"
+                DROP INDEX IF EXISTS idx_messages_history;
+                CREATE INDEX IF NOT EXISTS idx_messages_history_accepted
+                    ON messages (
+                        storage_owner,
+                        external_user,
+                        deleted_by_external,
+                        destination_iota_received_at DESC,
+                        origin_iota_received_at DESC,
+                        authored_at DESC,
+                        id DESC
+                    );
+                "#,
+            )?;
+        }
+        conn.pragma_update(None, "user_version", 15)?;
+    }
+
+    if current_version < 16 {
+        let messages_exist: bool = conn.query_row(
+            "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')",
+            [],
+            |row| row.get(0),
+        )?;
+        if messages_exist {
+            add_column_if_missing(conn, "delivery_failed_at", "delivery_failed_at INTEGER")?;
+            add_column_if_missing(conn, "delivery_failure", "delivery_failure TEXT")?;
+        }
+        let contacts_exist: bool = conn.query_row(
+            "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'contacts')",
+            [],
+            |row| row.get(0),
+        )?;
+        if contacts_exist {
+            add_table_column_if_missing(conn, "contacts", "created_at", "created_at INTEGER")?;
+            conn.execute(
+                "UPDATE contacts SET created_at = COALESCE(created_at, last_message_at, 0)",
+                [],
+            )?;
+        }
+        conn.pragma_update(None, "user_version", 16)?;
+    }
+
     Ok(())
 }
 
@@ -596,7 +648,7 @@ mod tests {
         run_migrations_on_connection(&conn)?;
 
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 13);
+        assert_eq!(version, 16);
         for column in ["height", "reply_to", "edited_count", "deleted_by_external"] {
             let mut statement =
                 conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
@@ -615,7 +667,7 @@ mod tests {
         run_migrations_on_connection(&conn)?;
         run_migrations_on_connection(&conn)?;
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 13);
+        assert_eq!(version, 16);
         for table in [
             "sync_heads",
             "sync_events",
@@ -650,7 +702,7 @@ mod tests {
         run_migrations_on_connection(&conn)?;
 
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 13);
+        assert_eq!(version, 16);
         for column in [
             "id",
             "user_id",

omikron-connector/src/omikron_connection.rs

@@ -40,6 +40,21 @@ const IOTA_KEYRING_PATH: &str = "iota.mk";
 static IDENTITY_PATH: std::sync::OnceLock<PathBuf> = std::sync::OnceLock::new();
 static OMIKRON_PUBLIC_KEY_PATH: std::sync::OnceLock<PathBuf> = std::sync::OnceLock::new();
 
+fn record_origin_delivery_failure(signer_id: u64, relay_message_id: &str, failure: &str) {
+    let Ok(storage_owner) = i64::try_from(signer_id) else {
+        return;
+    };
+    if let Err(error) = chat_files::record_delivery_failure(
+        storage_owner,
+        storage_owner,
+        relay_message_id,
+        failure,
+        now_millis_i64(),
+    ) {
+        log!("Relay delivery failure storage failed: {error}");
+    }
+}
+
 /*
  * Keeps identity and pinned Omikron key files independent from the process
  * working directory, so restarts use the same trusted material.
@@ -1136,6 +1151,11 @@ impl OmikronConnection {
                 Ok(destination_iota) => destination_iota,
                 Err(error) => {
                     log!("Relay origin route lookup failed: {}", error);
+                    record_origin_delivery_failure(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                        "destination_iota_not_found",
+                    );
                     self.send_relay_response(frame.id(), CommunicationType::ErrorNoIota)
                         .await;
                     return;
@@ -1145,6 +1165,11 @@ impl OmikronConnection {
                 Ok(value) => value,
                 Err(error) => {
                     log!("Relay origin forwarding validation failed: {}", error);
+                    record_origin_delivery_failure(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                        "forwarding_validation_failed",
+                    );
                     self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
                         .await;
                     return;
@@ -1154,6 +1179,11 @@ impl OmikronConnection {
                 Ok(bytes) => bytes,
                 Err(error) => {
                     log!("Relay origin retry could not be serialized: {}", error);
+                    record_origin_delivery_failure(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                        "serialization_failed",
+                    );
                     self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
                         .await;
                     return;
@@ -1167,6 +1197,11 @@ impl OmikronConnection {
                 &type_map_version,
             ) {
                 log!("Relay origin retry queue failed: {}", error);
+                record_origin_delivery_failure(
+                    verified.context.signer_id,
+                    &verified.context.message_id,
+                    "queue_failed",
+                );
                 self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
                     .await;
                 return;
@@ -1242,6 +1277,11 @@ impl OmikronConnection {
                 }
                 Ok(response) => {
                     log!("Relay origin route returned {}", response.get_type());
+                    record_origin_delivery_failure(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                        "destination_rejected",
+                    );
                     self.send_relay_response(
                         frame.id(),
                         response
@@ -1252,6 +1292,11 @@ impl OmikronConnection {
                 }
                 Err(error) => {
                     log!("Relay origin forwarding failed: {}", error);
+                    record_origin_delivery_failure(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                        "destination_unreachable",
+                    );
                     self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
                         .await;
                 }
@@ -1521,7 +1566,6 @@ impl OmikronConnection {
         dispatch!(DeleteApp, handle_delete_app);
         dispatch!(ClientConnected, handle_client_connected);
         dispatch!(ClientStateAck, handle_client_state_ack);
-        dispatch!(MessageState, handle_message_state);
         dispatch!(MessageEdit, handle_message_edit);
         dispatch!(MessageEditLive, handle_message_edit_live);
         dispatch!(MessageReactionAdd, handle_message_reaction_add);
@@ -1771,10 +1815,6 @@ impl OmikronConnection {
             .await;
     }
 
-    async fn handle_message_state(self: Arc<Self>, cv: &CommunicationValue) {
-        message_handlers::handle_message_state(cv);
-    }
-
     fn mutation_live_message(
         ty: CommunicationType,
         request: &CommunicationValue,

omikron-connector/src/user_ops.rs

@@ -84,7 +84,7 @@ fn sign_lifecycle_payload(
 async fn inspect_credential_account(
     connection: &dyn OmikronClient,
     credential: &TuCredential,
-) -> Result<(String, String), LifecycleUserError> {
+) -> Result<(String, String, i64), LifecycleUserError> {
     if credential.omega_host != omega_discovery::omega_host() {
         return Err(LifecycleUserError::OmegaHostMismatch);
     }
@@ -108,10 +108,16 @@ async fn inspect_credential_account(
         .as_str()
         .map(str::to_owned)
         .ok_or(LifecycleUserError::RemoteRejected)?;
+    let created_at = response
+        .get_data(DataType::CreatedAt)
+        .as_signed_number()
+        .and_then(|value| i64::try_from(value).ok())
+        .filter(|value| *value > 0)
+        .ok_or(LifecycleUserError::RemoteRejected)?;
     if public_key != public_key_bundle_to_base64(&credential.public_key_bundle()) {
         return Err(LifecycleUserError::RemoteRejected);
     }
-    Ok((username, public_key))
+    Ok((username, public_key, created_at))
 }
 
 async fn credential_proof(
@@ -168,17 +174,17 @@ pub async fn attach_user_from_tu(
 ) -> Result<UserProfile, LifecycleUserError> {
     let credential = TuCredential::parse(contents)
         .map_err(|error| LifecycleUserError::InvalidCredential(error.to_string()))?;
-    let (username, public_key) = inspect_credential_account(connection, &credential).await?;
+    let (username, public_key, created_at) =
-    let profile = UserProfile::new(
+        inspect_credential_account(connection, &credential).await?;
+    let profile = UserProfile::new_with_created_at(
         credential.user_id,
         username,
         None,
         public_key,
         hex_hash(contents),
         String::new(),
+        created_at,
     );
-    write_user_credential(&profile.username, &credential.to_canonical_string())
-        .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
     pending_operations::upsert(&PendingUserOperation {
         user_id: profile.user_id,
         operation: PendingUserOperationKind::Attach,
@@ -191,6 +197,13 @@ pub async fn attach_user_from_tu(
         created_at: now_millis(),
     })
     .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
+    write_user_credential(&profile.username, &credential.to_canonical_string())
+        .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
+    pending_operations::update_phase(
+        profile.user_id,
+        PendingUserOperationPhase::CredentialWritten,
+    )
+    .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?;
     if let Err(error) = credential_proof(
         connection,
         &credential,