[WIP] 0.3.0 mtp update

iota-storage/src/users/user_manager.rs

@@ -19,7 +19,10 @@ pub struct UserResidency {
 }
 
 fn now_millis() -> i64 {
-    SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as i64
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64
 }
 
 pub fn add_user(user: UserProfile) {
@@ -235,7 +238,10 @@ pub fn release_user(user_id: i64) -> Result<(), crate::storage_error::StorageErr
     })?;
     db::with_db(|conn| {
         let tx = conn.unchecked_transaction()?;
-        tx.execute("DELETE FROM trusted_apps WHERE user_id = ?1", params![user_id])?;
+        tx.execute(
+            "DELETE FROM trusted_apps WHERE user_id = ?1",
+            params![user_id],
+        )?;
         tx.execute("DELETE FROM users WHERE user_id = ?1", params![user_id])?;
         tx.execute(
             r#"INSERT INTO user_residency (user_id, username, lifecycle_state, data_state, updated_at)
@@ -246,7 +252,8 @@ pub fn release_user(user_id: i64) -> Result<(), crate::storage_error::StorageErr
         tx.commit()?;
         Ok(())
     })?;
-    remove_user_credential(user_id).map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
+    remove_user_credential(user_id)
+        .map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
 }
 
 /// Authoritative hosted-data erasure used by local purge and future Omega
@@ -256,14 +263,35 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage
         let tx = conn.unchecked_transaction()?;
         tx.execute("DELETE FROM message_edits WHERE message_id IN (SELECT id FROM messages WHERE storage_owner = ?1)", params![user_id])?;
         tx.execute("DELETE FROM reactions WHERE message_id IN (SELECT id FROM messages WHERE storage_owner = ?1)", params![user_id])?;
-        tx.execute("DELETE FROM messages WHERE storage_owner = ?1", params![user_id])?;
-        tx.execute("DELETE FROM contacts WHERE storage_owner = ?1", params![user_id])?;
-        tx.execute("DELETE FROM communities WHERE storage_owner = ?1", params![user_id])?;
+        tx.execute(
+            "DELETE FROM messages WHERE storage_owner = ?1",
+            params![user_id],
+        )?;
+        tx.execute(
+            "DELETE FROM contacts WHERE storage_owner = ?1",
+            params![user_id],
+        )?;
+        tx.execute(
+            "DELETE FROM communities WHERE storage_owner = ?1",
+            params![user_id],
+        )?;
         tx.execute("DELETE FROM settings WHERE user_id = ?1", params![user_id])?;
-        tx.execute("DELETE FROM sync_events WHERE user_id = ?1", params![user_id])?;
-        tx.execute("DELETE FROM sync_heads WHERE user_id = ?1", params![user_id])?;
-        tx.execute("DELETE FROM client_sync_state WHERE user_id = ?1", params![user_id])?;
-        tx.execute("DELETE FROM trusted_apps WHERE user_id = ?1", params![user_id])?;
+        tx.execute(
+            "DELETE FROM sync_events WHERE user_id = ?1",
+            params![user_id],
+        )?;
+        tx.execute(
+            "DELETE FROM sync_heads WHERE user_id = ?1",
+            params![user_id],
+        )?;
+        tx.execute(
+            "DELETE FROM client_sync_state WHERE user_id = ?1",
+            params![user_id],
+        )?;
+        tx.execute(
+            "DELETE FROM trusted_apps WHERE user_id = ?1",
+            params![user_id],
+        )?;
         tx.execute(
             "UPDATE user_residency SET data_state = 'empty', updated_at = ?2 WHERE user_id = ?1",
             params![user_id, now_millis()],
@@ -273,7 +301,8 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage
     })?;
     crate::util::e2ee_storage::purge_user(user_id)
         .map_err(crate::storage_error::StorageError::Other)?;
-    delete_user_directory(user_id).map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
+    delete_user_directory(user_id)
+        .map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
 }
 
 /// Complete local erasure is idempotent and is the target for a durable
@@ -281,12 +310,19 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage
 pub fn erase_user_locally(user_id: i64) -> Result<(), crate::storage_error::StorageError> {
     purge_user_data(user_id)?;
     db::with_db(|conn| {
-        conn.execute("DELETE FROM trusted_apps WHERE user_id = ?1", params![user_id])?;
+        conn.execute(
+            "DELETE FROM trusted_apps WHERE user_id = ?1",
+            params![user_id],
+        )?;
         conn.execute("DELETE FROM users WHERE user_id = ?1", params![user_id])?;
-        conn.execute("DELETE FROM user_residency WHERE user_id = ?1", params![user_id])?;
+        conn.execute(
+            "DELETE FROM user_residency WHERE user_id = ?1",
+            params![user_id],
+        )?;
         Ok(())
     })?;
-    remove_user_credential(user_id).map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
+    remove_user_credential(user_id)
+        .map_err(|error| crate::storage_error::StorageError::Other(error.to_string()))
 }
 
 pub fn get_residency() -> Vec<UserResidency> {
@@ -306,14 +342,15 @@ pub fn get_residency() -> Vec<UserResidency> {
 
 pub fn clear() {
     if let Err(e) = db::with_db(|conn| {
-        conn.execute_batch("DELETE FROM trusted_apps; DELETE FROM users; DELETE FROM user_residency;")?;
+        conn.execute_batch(
+            "DELETE FROM trusted_apps; DELETE FROM users; DELETE FROM user_residency;",
+        )?;
         Ok(())
     }) {
         eprintln!("Failed to clear users: {}", e);
     }
 }
 
-
 pub fn save_users() {
     // No-op: users are auto-saved via SQLite.
 }

iota-storage/src/util/db.rs

@@ -77,12 +77,22 @@ fn add_column_if_missing(
     column: &str,
     definition: &str,
 ) -> Result<(), StorageError> {
-    let mut statement =
-        conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
+    add_table_column_if_missing(conn, "messages", column, definition)
+}
+
+fn add_table_column_if_missing(
+    conn: &Connection,
+    table: &str,
+    column: &str,
+    definition: &str,
+) -> Result<(), StorageError> {
+    let mut statement = conn.prepare(&format!(
+        "SELECT 1 FROM pragma_table_info('{table}') WHERE name = ?1"
+    ))?;
     let exists = statement.exists([column])?;
 
     if !exists {
-        conn.execute_batch(&format!("ALTER TABLE messages ADD COLUMN {definition};"))?;
+        conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {definition};"))?;
     }
 
     Ok(())
@@ -266,6 +276,72 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
         )?;
     }
 
+    if current_version < 8 {
+        conn.execute_batch(
+            r#"
+            CREATE TABLE IF NOT EXISTS relay_replay (
+                signer_id INTEGER NOT NULL,
+                message_id TEXT NOT NULL,
+                created_at INTEGER NOT NULL,
+                PRIMARY KEY (signer_id, message_id)
+            );
+            CREATE INDEX IF NOT EXISTS idx_relay_replay_created_at
+                ON relay_replay (created_at);
+            CREATE TABLE IF NOT EXISTS pending_relays (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                destination_id INTEGER NOT NULL,
+                target_kind INTEGER NOT NULL DEFAULT 0,
+                frame BLOB NOT NULL,
+                created_at INTEGER NOT NULL,
+                frame_id INTEGER NOT NULL DEFAULT 0,
+                UNIQUE(destination_id, frame)
+            );
+            CREATE INDEX IF NOT EXISTS idx_pending_relays_destination
+                ON pending_relays (destination_id, id);
+            PRAGMA user_version = 8;
+            "#,
+        )?;
+    }
+
+    if current_version < 9 {
+        add_table_column_if_missing(
+            conn,
+            "pending_relays",
+            "target_kind",
+            "target_kind INTEGER NOT NULL DEFAULT 0",
+        )?;
+        add_table_column_if_missing(
+            conn,
+            "pending_relays",
+            "type_map_version",
+            "type_map_version TEXT NOT NULL DEFAULT '1.0'",
+        )?;
+        add_table_column_if_missing(
+            conn,
+            "pending_relays",
+            "frame_id",
+            "frame_id INTEGER NOT NULL DEFAULT 0",
+        )?;
+        conn.execute_batch(
+            r#"
+            CREATE TABLE IF NOT EXISTS relay_inbox (
+                signer_id INTEGER NOT NULL,
+                message_id TEXT NOT NULL,
+                created_at INTEGER NOT NULL,
+                destination_id INTEGER NOT NULL,
+                frame BLOB NOT NULL,
+                type_map_version TEXT NOT NULL,
+                frame_id INTEGER NOT NULL,
+                state TEXT NOT NULL CHECK (state IN ('received', 'applied', 'queued', 'delivered', 'rejected')),
+                PRIMARY KEY (signer_id, message_id)
+            );
+            CREATE INDEX IF NOT EXISTS idx_relay_inbox_state
+                ON relay_inbox (state, created_at);
+            PRAGMA user_version = 9;
+            "#,
+        )?;
+    }
+
     Ok(())
 }
 
@@ -335,7 +411,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, 7);
+        assert_eq!(version, 9);
         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")?;
@@ -352,8 +428,16 @@ 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, 7);
-        for table in ["sync_heads", "sync_events", "client_sync_state", "user_residency"] {
+        assert_eq!(version, 9);
+        for table in [
+            "sync_heads",
+            "sync_events",
+            "client_sync_state",
+            "user_residency",
+            "relay_replay",
+            "pending_relays",
+            "relay_inbox",
+        ] {
             let exists: i64 = conn.query_row(
                 "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
                 [table],
@@ -361,6 +445,11 @@ mod tests {
             )?;
             assert_eq!(exists, 1);
         }
+        for column in ["frame_id", "target_kind", "type_map_version"] {
+            let mut statement =
+                conn.prepare("SELECT 1 FROM pragma_table_info('pending_relays') WHERE name = ?1")?;
+            assert!(statement.exists([column])?);
+        }
         Ok(())
     }
 }

iota-storage/src/util/e2ee_storage.rs

@@ -24,19 +24,6 @@ pub struct ChatSecretQuery {
     pub secret_id: Option<String>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct PendingChatSecretForward {
-    pub recipient_user_id: String,
-    pub chat_id: String,
-    pub sender_user_id: String,
-    pub secret_id: String,
-    pub version: i64,
-    pub encrypted_secret: Vec<u8>,
-    pub kem_ciphertext: Vec<u8>,
-    pub wrapping_scheme: String,
-    pub created_at: i64,
-}
-
 static E2EE_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
     db::create_shared_connection(
         "e2ee",
@@ -63,21 +50,6 @@ static E2EE_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
         CREATE INDEX IF NOT EXISTS idx_chat_secrets_owner
             ON chat_secrets (user_id, chat_id, secret_id);
 
-        CREATE TABLE IF NOT EXISTS pending_chat_secret_forwards (
-            recipient_user_id TEXT NOT NULL,
-            chat_id TEXT NOT NULL,
-            sender_user_id TEXT NOT NULL,
-            secret_id TEXT NOT NULL,
-            version INTEGER NOT NULL,
-            encrypted_secret BLOB NOT NULL,
-            kem_ciphertext BLOB NOT NULL,
-            wrapping_scheme TEXT NOT NULL,
-            created_at INTEGER NOT NULL,
-            PRIMARY KEY (recipient_user_id, chat_id, secret_id)
-        );
-
-        CREATE INDEX IF NOT EXISTS idx_pending_chat_secret_forwards_recipient
-            ON pending_chat_secret_forwards (recipient_user_id, created_at);
     "#,
     )
     .expect("Failed to create or initialize E2EE DB")
@@ -115,89 +87,15 @@ pub fn put_chat_secret(record: StoredChatSecret) -> Result<(), StorageError> {
     })
 }
 
-pub fn put_pending_chat_secret_forward(
-    record: PendingChatSecretForward,
-) -> Result<(), StorageError> {
-    db::with_conn(&E2EE_DB, |conn| {
-        conn.execute(
-            r#"
-            INSERT INTO pending_chat_secret_forwards (
-                recipient_user_id, chat_id, sender_user_id, secret_id, version,
-                encrypted_secret, kem_ciphertext, wrapping_scheme, created_at
-            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
-            ON CONFLICT(recipient_user_id, chat_id, secret_id) DO UPDATE SET
-                sender_user_id = excluded.sender_user_id,
-                version = excluded.version,
-                encrypted_secret = excluded.encrypted_secret,
-                kem_ciphertext = excluded.kem_ciphertext,
-                wrapping_scheme = excluded.wrapping_scheme,
-                created_at = excluded.created_at
-            "#,
-            params![
-                record.recipient_user_id,
-                record.chat_id,
-                record.sender_user_id,
-                record.secret_id,
-                record.version,
-                record.encrypted_secret,
-                record.kem_ciphertext,
-                record.wrapping_scheme,
-                record.created_at,
-            ],
-        )?;
-        Ok(())
-    })
-}
-
-pub fn get_pending_chat_secret_forwards(
-    limit: i64,
-) -> Result<Vec<PendingChatSecretForward>, StorageError> {
-    db::with_conn(&E2EE_DB, |conn| {
-        let mut stmt = conn.prepare(
-            r#"
-            SELECT recipient_user_id, chat_id, sender_user_id, secret_id, version,
-                   encrypted_secret, kem_ciphertext, wrapping_scheme, created_at
-            FROM pending_chat_secret_forwards
-            ORDER BY created_at ASC
-            LIMIT ?1
-            "#,
-        )?;
-        let rows = stmt.query_map(params![limit.clamp(1, 500)], pending_forward_from_row)?;
-        let mut out = Vec::new();
-        for row in rows {
-            out.push(row?);
-        }
-        Ok(out)
-    })
-}
-
-pub fn delete_pending_chat_secret_forward(
-    recipient_user_id: &str,
-    chat_id: &str,
-    secret_id: &str,
-) -> Result<(), StorageError> {
-    db::with_conn(&E2EE_DB, |conn| {
-        conn.execute(
-            r#"
-            DELETE FROM pending_chat_secret_forwards
-            WHERE recipient_user_id = ?1 AND chat_id = ?2 AND secret_id = ?3
-            "#,
-            params![recipient_user_id, chat_id, secret_id],
-        )?;
-        Ok(())
-    })
-}
-
-/// Erase every E2EE record owned by, or queued for, a user.  The operation is
+/// Erase every E2EE record owned by a user. The operation is
 /// intentionally idempotent so it can be retried after an interrupted remote
 /// erasure request.
 pub fn purge_user(user_id: i64) -> Result<(), StorageError> {
     let user_id = user_id.to_string();
     db::with_conn(&E2EE_DB, |conn| {
         let tx = conn.unchecked_transaction()?;
-        tx.execute("DELETE FROM chat_secrets WHERE user_id = ?1", params![user_id])?;
         tx.execute(
-            "DELETE FROM pending_chat_secret_forwards WHERE recipient_user_id = ?1 OR sender_user_id = ?1",
+            "DELETE FROM chat_secrets WHERE user_id = ?1",
             params![user_id],
         )?;
         tx.commit()?;
@@ -242,17 +140,3 @@ fn chat_secret_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<StoredChatS
         updated_at: row.get(8)?,
     })
 }
-
-fn pending_forward_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<PendingChatSecretForward> {
-    Ok(PendingChatSecretForward {
-        recipient_user_id: row.get(0)?,
-        chat_id: row.get(1)?,
-        sender_user_id: row.get(2)?,
-        secret_id: row.get(3)?,
-        version: row.get(4)?,
-        encrypted_secret: row.get(5)?,
-        kem_ciphertext: row.get(6)?,
-        wrapping_scheme: row.get(7)?,
-        created_at: row.get(8)?,
-    })
-}

iota-storage/src/util/mod.rs

@@ -4,5 +4,7 @@ pub mod communities_util;
 pub mod config_util;
 pub mod db;
 pub mod e2ee_storage;
+pub mod relay_queue;
+pub mod relay_replay;
 pub mod settings;
 pub mod sync;

iota-storage/src/util/relay_queue.rs

@@ -0,0 +1,135 @@
+use crate::storage_error::StorageError;
+use crate::util::db;
+use iota_util::route_target::RouteTarget;
+use rusqlite::params;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PendingRelay {
+    pub id: i64,
+    pub target: RouteTarget,
+    pub frame: Vec<u8>,
+    pub created_at: i64,
+    pub frame_id: u32,
+    pub type_map_version: String,
+}
+
+pub fn enqueue(
+    target: RouteTarget,
+    frame: &[u8],
+    created_at: i64,
+    frame_id: u32,
+    type_map_version: &str,
+) -> Result<(), StorageError> {
+    let destination_id = i64::try_from(target.id())
+        .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
+    let target_kind = match target {
+        RouteTarget::User(_) => 0_i64,
+        RouteTarget::Iota(_) => 1_i64,
+    };
+    db::with_db(|connection| {
+        connection.execute(
+            "INSERT OR IGNORE INTO pending_relays (destination_id, target_kind, frame, created_at, frame_id, type_map_version) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+            params![
+                destination_id,
+                target_kind,
+                frame,
+                created_at,
+                i64::from(frame_id),
+                type_map_version
+            ],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn list(limit: i64) -> Result<Vec<PendingRelay>, StorageError> {
+    db::with_db(|connection| {
+        let mut statement = connection.prepare(
+            "SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version FROM pending_relays ORDER BY id LIMIT ?1",
+        )?;
+        let rows = statement.query_map(params![limit.clamp(1, 500)], |row| {
+            let destination_id = row.get::<_, i64>(1)?;
+            let target_kind = row.get::<_, i64>(2)?;
+            let destination_id = u64::try_from(destination_id).map_err(|_| {
+                rusqlite::Error::FromSqlConversionFailure(
+                    1,
+                    rusqlite::types::Type::Integer,
+                    "negative relay destination ID".into(),
+                )
+            })?;
+            let target = match target_kind {
+                0 => RouteTarget::User(destination_id),
+                1 => RouteTarget::Iota(destination_id),
+                _ => {
+                    return Err(rusqlite::Error::FromSqlConversionFailure(
+                        2,
+                        rusqlite::types::Type::Integer,
+                        "invalid relay target kind".into(),
+                    ));
+                }
+            };
+            Ok(PendingRelay {
+                id: row.get(0)?,
+                target,
+                frame: row.get(3)?,
+                created_at: row.get(4)?,
+                frame_id: u32::try_from(row.get::<_, i64>(5)?).map_err(|_| {
+                    rusqlite::Error::FromSqlConversionFailure(
+                        5,
+                        rusqlite::types::Type::Integer,
+                        "negative relay frame ID".into(),
+                    )
+                })?,
+                type_map_version: row.get(6)?,
+            })
+        })?;
+        rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
+    })
+}
+
+pub fn acknowledge(destination_id: u64, frame_id: u32) -> Result<bool, StorageError> {
+    let destination_id = i64::try_from(destination_id)
+        .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
+    db::with_db(|connection| {
+        let changed = connection.execute(
+            "DELETE FROM pending_relays WHERE destination_id = ?1 AND target_kind = 0 AND frame_id = ?2",
+            params![destination_id, i64::from(frame_id)],
+        )?;
+        Ok(changed == 1)
+    })
+}
+
+pub fn acknowledge_iota(destination_id: u64, frame_id: u32) -> Result<bool, StorageError> {
+    let destination_id = i64::try_from(destination_id)
+        .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
+    db::with_db(|connection| {
+        let changed = connection.execute(
+            "DELETE FROM pending_relays WHERE destination_id = ?1 AND target_kind = 1 AND frame_id = ?2",
+            params![destination_id, i64::from(frame_id)],
+        )?;
+        Ok(changed == 1)
+    })
+}
+
+pub fn remove_for_frame(target: RouteTarget, frame_id: u32) -> Result<bool, StorageError> {
+    let destination_id = i64::try_from(target.id())
+        .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
+    let target_kind = match target {
+        RouteTarget::User(_) => 0_i64,
+        RouteTarget::Iota(_) => 1_i64,
+    };
+    db::with_db(|connection| {
+        let changed = connection.execute(
+            "DELETE FROM pending_relays WHERE destination_id = ?1 AND target_kind = ?2 AND frame_id = ?3",
+            params![destination_id, target_kind, i64::from(frame_id)],
+        )?;
+        Ok(changed == 1)
+    })
+}
+
+pub fn delete(id: i64) -> Result<(), StorageError> {
+    db::with_db(|connection| {
+        connection.execute("DELETE FROM pending_relays WHERE id = ?1", params![id])?;
+        Ok(())
+    })
+}

iota-storage/src/util/relay_replay.rs

@@ -0,0 +1,152 @@
+use crate::storage_error::StorageError;
+use crate::util::db;
+use rusqlite::params;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RelayReservation {
+    New,
+    Existing { state: String, frame_matches: bool },
+}
+
+pub fn reserve(
+    signer_id: u64,
+    message_id: &str,
+    created_at: u64,
+    destination_id: u64,
+    frame: &[u8],
+    frame_id: u32,
+    type_map_version: &str,
+) -> Result<RelayReservation, StorageError> {
+    let signer_id = i64::try_from(signer_id)
+        .map_err(|_| StorageError::Other("relay signer ID exceeds SQLite range".into()))?;
+    let created_at = i64::try_from(created_at)
+        .map_err(|_| StorageError::Other("relay creation time exceeds SQLite range".into()))?;
+    let destination_id = i64::try_from(destination_id)
+        .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
+
+    db::with_db(|connection| {
+        let inserted = connection.execute(
+            "INSERT OR IGNORE INTO relay_inbox (signer_id, message_id, created_at, destination_id, frame, frame_id, type_map_version, state) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'received')",
+            params![
+                signer_id,
+                message_id,
+                created_at,
+                destination_id,
+                frame,
+                i64::from(frame_id),
+                type_map_version
+            ],
+        )?;
+        if inserted == 1 {
+            return Ok(RelayReservation::New);
+        }
+
+        let (state, existing_destination_id, existing_frame, existing_type_map_version):
+            (String, i64, Vec<u8>, String) = connection.query_row(
+            "SELECT state, destination_id, frame, type_map_version FROM relay_inbox WHERE signer_id = ?1 AND message_id = ?2",
+            params![signer_id, message_id],
+            |row| {
+                Ok((
+                    row.get(0)?,
+                    row.get(1)?,
+                    row.get::<_, Vec<u8>>(2)?,
+                    row.get(3)?,
+                ))
+            },
+        )?;
+        Ok(RelayReservation::Existing {
+            state,
+            frame_matches: existing_destination_id == destination_id
+                && existing_frame == frame
+                && existing_type_map_version == type_map_version,
+        })
+    })
+}
+
+pub fn mark_delivered_for_frame(destination_id: u64, frame_id: u32) -> Result<(), StorageError> {
+    let destination_id = i64::try_from(destination_id)
+        .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
+    db::with_db(|connection| {
+        connection.execute(
+            "UPDATE relay_inbox SET state = 'delivered' WHERE destination_id = ?1 AND frame_id = ?2",
+            params![destination_id, i64::from(frame_id)],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), StorageError> {
+    if !matches!(
+        state,
+        "received" | "applied" | "queued" | "delivered" | "rejected"
+    ) {
+        return Err(StorageError::Other("invalid relay inbox state".into()));
+    }
+    let signer_id = i64::try_from(signer_id)
+        .map_err(|_| StorageError::Other("relay signer ID exceeds SQLite range".into()))?;
+    db::with_db(|connection| {
+        connection.execute(
+            "UPDATE relay_inbox SET state = ?3 WHERE signer_id = ?1 AND message_id = ?2",
+            params![signer_id, message_id, state],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn prune_completed(before_created_at: i64) -> Result<(), StorageError> {
+    db::with_db(|connection| {
+        connection.execute(
+            "DELETE FROM relay_inbox WHERE created_at < ?1 AND state IN ('delivered', 'rejected')",
+            params![before_created_at],
+        )?;
+        connection.execute(
+            "DELETE FROM relay_replay WHERE created_at < ?1",
+            params![before_created_at],
+        )?;
+        Ok(())
+    })
+}
+
+pub fn accept(signer_id: u64, message_id: &str, created_at: u64) -> Result<bool, StorageError> {
+    let signer_id = i64::try_from(signer_id)
+        .map_err(|_| StorageError::Other("relay signer ID exceeds SQLite range".into()))?;
+    let created_at = i64::try_from(created_at)
+        .map_err(|_| StorageError::Other("relay creation time exceeds SQLite range".into()))?;
+
+    db::with_db(|connection| {
+        let inserted = connection.execute(
+            "INSERT OR IGNORE INTO relay_replay (signer_id, message_id, created_at) VALUES (?1, ?2, ?3)",
+            params![signer_id, message_id, created_at],
+        )?;
+        Ok(inserted == 1)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use rusqlite::{Connection, params};
+
+    #[test]
+    fn replay_identity_uses_signer_and_message_id() -> Result<(), rusqlite::Error> {
+        let connection = Connection::open_in_memory()?;
+        connection.execute_batch(
+            "CREATE TABLE relay_replay (signer_id INTEGER NOT NULL, message_id TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (signer_id, message_id));",
+        )?;
+
+        let first = connection.execute(
+            "INSERT OR IGNORE INTO relay_replay (signer_id, message_id, created_at) VALUES (?1, ?2, ?3)",
+            params![7_i64, "message", 1_i64],
+        )?;
+        let duplicate = connection.execute(
+            "INSERT OR IGNORE INTO relay_replay (signer_id, message_id, created_at) VALUES (?1, ?2, ?3)",
+            params![7_i64, "message", 2_i64],
+        )?;
+        let other_signer = connection.execute(
+            "INSERT OR IGNORE INTO relay_replay (signer_id, message_id, created_at) VALUES (?1, ?2, ?3)",
+            params![8_i64, "message", 2_i64],
+        )?;
+
+        assert_eq!((first, duplicate, other_signer), (1, 0, 1));
+        Ok(())
+    }
+}