diff --git a/services/commtest/tests/tunnelbroker_integration_tests.rs b/services/commtest/tests/tunnelbroker_integration_tests.rs
index de4e3620d..3e01bcbf0 100644
--- a/services/commtest/tests/tunnelbroker_integration_tests.rs
+++ b/services/commtest/tests/tunnelbroker_integration_tests.rs
@@ -1,111 +1,112 @@
 mod proto {
   tonic::include_proto!("tunnelbroker");
 }
 
 use commtest::identity::device::create_device;
 use commtest::identity::olm_account_infos::{
   MOCK_CLIENT_KEYS_1, MOCK_CLIENT_KEYS_2,
 };
 use commtest::tunnelbroker::socket::create_socket;
 use futures_util::{SinkExt, StreamExt};
 use proto::tunnelbroker_service_client::TunnelbrokerServiceClient;
 use proto::MessageToDevice;
 use std::time::Duration;
 use tokio::time::sleep;
 use tokio_tungstenite::tungstenite::Message;
 
-use tunnelbroker_messages::{
-  MessageToDevice as WebsocketMessageToDevice, RefreshKeyRequest,
-};
+use tunnelbroker_messages::{MessageToDeviceRequest, RefreshKeyRequest};
 
 #[tokio::test]
 async fn send_refresh_request() {
   // Create session as a keyserver
   let device_info = create_device(None).await;
   let mut socket = create_socket(&device_info).await;
 
   // Send request for keyserver to refresh keys (identity service)
   let mut tunnelbroker_client =
     TunnelbrokerServiceClient::connect("http://localhost:50051")
       .await
       .unwrap();
 
   let refresh_request = RefreshKeyRequest {
     device_id: device_info.device_id.clone(),
     number_of_keys: 5,
   };
 
   let payload = serde_json::to_string(&refresh_request).unwrap();
   let request = MessageToDevice {
     device_id: device_info.device_id.clone(),
     payload,
   };
   let grpc_message = tonic::Request::new(request);
 
   tunnelbroker_client
     .send_message_to_device(grpc_message)
     .await
     .unwrap();
 
   // Have keyserver receive any websocket messages
   let response = socket.next().await.unwrap().unwrap();
 
   // Check that message received by keyserver matches what identity server
   // issued
   let serialized_response: RefreshKeyRequest =
     serde_json::from_str(&response.to_text().unwrap()).unwrap();
   assert_eq!(serialized_response, refresh_request);
 }
 
 #[tokio::test]
 async fn test_messages_order() {
   let sender = create_device(Some(&MOCK_CLIENT_KEYS_1)).await;
   let receiver = create_device(Some(&MOCK_CLIENT_KEYS_2)).await;
 
   let messages = vec![
-    WebsocketMessageToDevice {
+    MessageToDeviceRequest {
+      client_message_id: "5".to_string(),
       device_id: receiver.device_id.clone(),
       payload: "first message".to_string(),
     },
-    WebsocketMessageToDevice {
+    MessageToDeviceRequest {
+      client_message_id: "2".to_string(),
       device_id: receiver.device_id.clone(),
       payload: "second message".to_string(),
     },
-    WebsocketMessageToDevice {
+    MessageToDeviceRequest {
+      client_message_id: "7".to_string(),
       device_id: receiver.device_id.clone(),
       payload: "third message".to_string(),
     },
   ];
 
   let serialized_messages: Vec<_> = messages
     .iter()
     .map(|message| {
       serde_json::to_string(message)
         .expect("Failed to serialize message to device")
     })
     .map(Message::text)
     .collect();
 
   let (mut sender_socket, _) = create_socket(&sender).await.split();
 
   for msg in serialized_messages.clone() {
     sender_socket
       .send(msg)
       .await
       .expect("Failed to send the message over WebSocket");
   }
 
   // Wait a specified duration to ensure that message had time to persist
   sleep(Duration::from_millis(100)).await;
 
   let mut receiver_socket = create_socket(&receiver).await;
 
   for msg in messages {
     if let Some(Ok(response)) = receiver_socket.next().await {
       let received_payload = response.to_text().unwrap();
       assert_eq!(msg.payload, received_payload);
     } else {
       panic!("Unable to receive message");
     }
   }
 }
diff --git a/services/commtest/tests/tunnelbroker_persist_tests.rs b/services/commtest/tests/tunnelbroker_persist_tests.rs
index 5461b5f44..0d28a1d09 100644
--- a/services/commtest/tests/tunnelbroker_persist_tests.rs
+++ b/services/commtest/tests/tunnelbroker_persist_tests.rs
@@ -1,95 +1,94 @@
 mod proto {
   tonic::include_proto!("tunnelbroker");
 }
 
 use commtest::identity::device::create_device;
 use commtest::identity::olm_account_infos::{
   MOCK_CLIENT_KEYS_1, MOCK_CLIENT_KEYS_2,
 };
 use commtest::tunnelbroker::socket::create_socket;
 use futures_util::{SinkExt, StreamExt};
 use proto::tunnelbroker_service_client::TunnelbrokerServiceClient;
 use proto::MessageToDevice;
 use std::time::Duration;
 use tokio::time::sleep;
 use tokio_tungstenite::tungstenite::Message;
-use tunnelbroker_messages::{
-  MessageToDevice as WebsocketMessageToDevice, RefreshKeyRequest,
-};
+use tunnelbroker_messages::{MessageToDeviceRequest, RefreshKeyRequest};
 
 /// Tests that a message to an offline device gets pushed to dynamodb
 /// then recalled once a device connects
 
 #[tokio::test]
 async fn persist_grpc_messages() {
   let device_info = create_device(None).await;
 
   // Send request for keyserver to refresh keys (identity service)
   let mut tunnelbroker_client =
     TunnelbrokerServiceClient::connect("http://localhost:50051")
       .await
       .unwrap();
 
   let refresh_request = RefreshKeyRequest {
     device_id: device_info.device_id.to_string(),
     number_of_keys: 5,
   };
 
   let payload = serde_json::to_string(&refresh_request).unwrap();
   let request = MessageToDevice {
     device_id: device_info.device_id.to_string(),
     payload,
   };
   let grpc_message = tonic::Request::new(request);
   tunnelbroker_client
     .send_message_to_device(grpc_message)
     .await
     .unwrap();
 
   // Wait a specified duration to ensure that message had time to persist
   sleep(Duration::from_millis(100)).await;
 
   let mut socket = create_socket(&device_info).await;
   // Have keyserver receive any websocket messages
   if let Some(Ok(response)) = socket.next().await {
     // Check that message received by keyserver matches what identity server
     // issued
     let serialized_response: RefreshKeyRequest =
       serde_json::from_str(&response.to_text().unwrap()).unwrap();
     assert_eq!(serialized_response, refresh_request);
   };
 }
 
 #[tokio::test]
 async fn persist_websocket_messages() {
   let sender = create_device(Some(&MOCK_CLIENT_KEYS_1)).await;
   let receiver = create_device(Some(&MOCK_CLIENT_KEYS_2)).await;
 
   // Send message to not connected client
   let payload = "persisted message";
-  let request = WebsocketMessageToDevice {
+  let request = MessageToDeviceRequest {
+    client_message_id: "mockID".to_string(),
     device_id: receiver.device_id.clone(),
     payload: payload.to_string(),
   };
 
   let serialized_request = serde_json::to_string(&request)
     .expect("Failed to serialize message to device");
 
   let mut sender_socket = create_socket(&sender).await;
   sender_socket
     .send(Message::Text(serialized_request))
     .await
     .expect("Failed to send message");
 
   // Wait a specified duration to ensure that message had time to persist
   sleep(Duration::from_millis(100)).await;
 
   // Connect receiver
   let mut receiver_socket = create_socket(&receiver).await;
 
   // Receive message
   if let Some(Ok(response)) = receiver_socket.next().await {
     let received_payload = response.to_text().unwrap();
     assert_eq!(payload, received_payload);
   };
 }
diff --git a/services/tunnelbroker/src/websockets/session.rs b/services/tunnelbroker/src/websockets/session.rs
index 3d59bf0aa..ae027aedd 100644
--- a/services/tunnelbroker/src/websockets/session.rs
+++ b/services/tunnelbroker/src/websockets/session.rs
@@ -1,269 +1,272 @@
 use aws_sdk_dynamodb::error::SdkError;
 use aws_sdk_dynamodb::operation::put_item::PutItemError;
 use derive_more;
 use futures_util::stream::SplitSink;
 use futures_util::SinkExt;
 use futures_util::StreamExt;
 use hyper_tungstenite::{tungstenite::Message, WebSocketStream};
 use lapin::message::Delivery;
 use lapin::options::{
   BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions,
   QueueDeclareOptions, QueueDeleteOptions,
 };
 use lapin::types::FieldTable;
 use lapin::BasicProperties;
 use tokio::io::AsyncRead;
 use tokio::io::AsyncWrite;
 use tracing::{debug, error, info};
 use tunnelbroker_messages::{session::DeviceTypes, Messages};
 
 use crate::database::{self, DatabaseClient, DeviceMessage};
 use crate::error::Error;
 use crate::identity;
 
 pub struct DeviceInfo {
   pub device_id: String,
   pub notify_token: Option<String>,
   pub device_type: DeviceTypes,
   pub device_app_version: Option<String>,
   pub device_os: Option<String>,
 }
 
 pub struct WebsocketSession<S> {
   tx: SplitSink<WebSocketStream<S>, Message>,
   db_client: DatabaseClient,
   pub device_info: DeviceInfo,
   amqp_channel: lapin::Channel,
   // Stream of messages from AMQP endpoint
   amqp_consumer: lapin::Consumer,
 }
 
 #[derive(
   Debug, derive_more::Display, derive_more::From, derive_more::Error,
 )]
 pub enum SessionError {
   InvalidMessage,
   SerializationError(serde_json::Error),
   MessageError(database::MessageErrors),
   AmqpError(lapin::Error),
   InternalError,
   UnauthorizedDevice,
   PersistenceError(SdkError<PutItemError>),
 }
 
 pub fn consume_error<T>(result: Result<T, SessionError>) {
   if let Err(e) = result {
     error!("{}", e)
   }
 }
 
 // Parse a session request and retrieve the device information
 pub async fn handle_first_message_from_device(
   message: &str,
 ) -> Result<DeviceInfo, Error> {
   let serialized_message = serde_json::from_str::<Messages>(message)?;
 
   match serialized_message {
     Messages::ConnectionInitializationMessage(mut session_info) => {
       let device_info = DeviceInfo {
         device_id: session_info.device_id.clone(),
         notify_token: session_info.notify_token.take(),
         device_type: session_info.device_type,
         device_app_version: session_info.device_app_version.take(),
         device_os: session_info.device_os.take(),
       };
 
       // Authenticate device
       debug!("Authenticating device: {}", &session_info.device_id);
       let auth_request = identity::verify_user_access_token(
         &session_info.user_id,
         &device_info.device_id,
         &session_info.access_token,
       )
       .await;
 
       match auth_request {
         Err(e) => {
           error!("Failed to complete request to identity service: {:?}", e);
           return Err(SessionError::InternalError.into());
         }
         Ok(false) => {
           info!("Device failed authentication: {}", &session_info.device_id);
           return Err(SessionError::UnauthorizedDevice.into());
         }
         Ok(true) => {
           debug!(
             "Successfully authenticated device: {}",
             &session_info.device_id
           );
         }
       }
 
       Ok(device_info)
     }
     _ => {
       debug!("Received invalid request");
       Err(SessionError::InvalidMessage.into())
     }
   }
 }
 
 impl<S: AsyncRead + AsyncWrite + Unpin> WebsocketSession<S> {
   pub async fn from_frame(
     tx: SplitSink<WebSocketStream<S>, Message>,
     db_client: DatabaseClient,
     frame: Message,
     amqp_channel: &lapin::Channel,
   ) -> Result<WebsocketSession<S>, Error> {
     let device_info = match frame {
       Message::Text(payload) => {
         handle_first_message_from_device(&payload).await?
       }
       _ => {
         error!("Client sent wrong frame type for establishing connection");
         return Err(SessionError::InvalidMessage.into());
       }
     };
 
     // We don't currently have a use case to interact directly with the queue,
     // however, we need to declare a queue for a given device
     amqp_channel
       .queue_declare(
         &device_info.device_id,
         QueueDeclareOptions::default(),
         FieldTable::default(),
       )
       .await?;
 
     let amqp_consumer = amqp_channel
       .basic_consume(
         &device_info.device_id,
         "tunnelbroker",
         BasicConsumeOptions::default(),
         FieldTable::default(),
       )
       .await?;
 
     Ok(WebsocketSession {
       tx,
       db_client,
       device_info,
       amqp_channel: amqp_channel.clone(),
       amqp_consumer,
     })
   }
 
   pub async fn handle_websocket_frame_from_device(
     &self,
     msg: String,
   ) -> Result<(), SessionError> {
     let serialized_message = serde_json::from_str::<Messages>(&msg)?;
 
     match serialized_message {
-      Messages::MessageToDevice(message_to_device) => {
-        debug!("Received message for {}", message_to_device.device_id);
+      Messages::MessageToDeviceRequest(message_to_device_request) => {
+        debug!(
+          "Received message for {}",
+          message_to_device_request.device_id
+        );
         self
           .db_client
           .persist_message(
-            message_to_device.device_id.as_str(),
-            message_to_device.payload.as_str(),
-            "message_id",
+            &message_to_device_request.device_id,
+            &message_to_device_request.payload,
+            &message_to_device_request.client_message_id,
           )
           .await?;
 
         self
           .amqp_channel
           .basic_publish(
             "",
-            &message_to_device.device_id,
+            &message_to_device_request.device_id,
             BasicPublishOptions::default(),
-            message_to_device.payload.as_bytes(),
+            message_to_device_request.payload.as_bytes(),
             BasicProperties::default(),
           )
           .await?;
       }
       _ => {
         error!("Client sent invalid message type");
         return Err(SessionError::InvalidMessage);
       }
     }
 
     Ok(())
   }
 
   pub async fn next_amqp_message(
     &mut self,
   ) -> Option<Result<Delivery, lapin::Error>> {
     self.amqp_consumer.next().await
   }
 
   pub async fn deliver_persisted_messages(
     &mut self,
   ) -> Result<(), SessionError> {
     // Check for persisted messages
     let messages = self
       .db_client
       .retrieve_messages(&self.device_info.device_id)
       .await
       .unwrap_or_else(|e| {
         error!("Error while retrieving messages: {}", e);
         Vec::new()
       });
 
     for message in messages {
       let device_message = DeviceMessage::from_hashmap(message)?;
       self
         .send_message_to_device(Message::Text(device_message.payload))
         .await;
       if let Err(e) = self
         .db_client
         .delete_message(&self.device_info.device_id, &device_message.message_id)
         .await
       {
         error!("Failed to delete message: {}:", e);
       }
     }
 
     debug!(
       "Flushed messages for device: {}",
       &self.device_info.device_id
     );
 
     Ok(())
   }
 
   pub async fn send_message_to_device(&mut self, message: Message) {
     if let Err(e) = self.tx.send(message).await {
       error!("Failed to send message to device: {}", e);
     }
   }
 
   // Release WebSocket and remove from active connections
   pub async fn close(&mut self) {
     if let Err(e) = self.tx.close().await {
       debug!("Failed to close WebSocket session: {}", e);
     }
 
     if let Err(e) = self
       .amqp_channel
       .basic_cancel(
         self.amqp_consumer.tag().as_str(),
         BasicCancelOptions::default(),
       )
       .await
     {
       error!("Failed to cancel consumer: {}", e);
     }
 
     if let Err(e) = self
       .amqp_channel
       .queue_delete(
         self.device_info.device_id.as_str(),
         QueueDeleteOptions::default(),
       )
       .await
     {
       error!("Failed to delete queue: {}", e);
     }
   }
 }
diff --git a/shared/tunnelbroker_messages/src/messages/message_to_device_request.rs b/shared/tunnelbroker_messages/src/messages/message_to_device_request.rs
new file mode 100644
index 000000000..1138824e6
--- /dev/null
+++ b/shared/tunnelbroker_messages/src/messages/message_to_device_request.rs
@@ -0,0 +1,34 @@
+// Message sent from WebSocket clients to Tunnelbroker
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, PartialEq, Debug)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub struct MessageToDeviceRequest {
+  #[serde(rename = "clientMessageID")]
+  pub client_message_id: String,
+  #[serde(rename = "deviceID")]
+  pub device_id: String,
+  pub payload: String,
+}
+
+#[cfg(test)]
+mod message_to_device_request_tests {
+  use super::*;
+
+  #[test]
+  fn test_message_to_device_request_deserialization() {
+    let example_payload = r#"{
+      "type": "MessageToDeviceRequest",
+      "clientMessageID": "client123",
+      "deviceID": "alice",
+      "payload": "message from Bob"
+    }"#;
+
+    let request =
+      serde_json::from_str::<MessageToDeviceRequest>(example_payload).unwrap();
+    assert_eq!(request.client_message_id, "client123");
+    assert_eq!(request.device_id, "alice");
+    assert_eq!(request.payload, "message from Bob");
+  }
+}
diff --git a/shared/tunnelbroker_messages/src/messages/mod.rs b/shared/tunnelbroker_messages/src/messages/mod.rs
index b29844f43..f793ab2ed 100644
--- a/shared/tunnelbroker_messages/src/messages/mod.rs
+++ b/shared/tunnelbroker_messages/src/messages/mod.rs
@@ -1,18 +1,24 @@
 // Messages sent between Tunnelbroker and a device
 pub mod keys;
 pub mod message_to_device;
+pub mod message_to_device_request;
 pub mod session;
 
 pub use keys::*;
 pub use message_to_device::*;
+pub use message_to_device_request::*;
 pub use session::*;
 
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 #[serde(untagged)]
 pub enum Messages {
   RefreshKeysRequest(RefreshKeyRequest),
   ConnectionInitializationMessage(ConnectionInitializationMessage),
+  // MessageToDeviceRequest must be placed before MessageToDevice.
+  // This is due to serde's pattern matching behavior where it prioritizes
+  // the first matching pattern it encounters.
+  MessageToDeviceRequest(MessageToDeviceRequest),
   MessageToDevice(MessageToDevice),
 }
diff --git a/shared/tunnelbroker_messages/src/messages/session.rs b/shared/tunnelbroker_messages/src/messages/session.rs
index 7093786b3..0f99d6d69 100644
--- a/shared/tunnelbroker_messages/src/messages/session.rs
+++ b/shared/tunnelbroker_messages/src/messages/session.rs
@@ -1,73 +1,73 @@
 // Messages sent between Tunnelbroker and a device
 
 use serde::{Deserialize, Serialize};
 
 /// The workflow when estabilishing a Tunnelbroker connection:
 ///   - Client sends ConnectionInitializationMessage
 ///   - Tunnelbroker validates access_token with identity service
 ///   - Tunnelbroker emits an AMQP message declaring that it has opened a new
 ///     connection with a given device, so that the respective tunnelbroker
 ///     instance can close the existing connection.
 ///   - Tunnelbroker returns a session_id representing that the connection was
 ///     accepted
 ///   - Tunnelbroker will flush all messages related to device from RabbitMQ.
 ///     This must be done first before flushing DynamoDB to prevent duplicated
 ///     messages.
 ///   - Tunnelbroker flushes all messages in DynamoDB
 ///   - Tunnelbroker orders messages by creation date (oldest first), and sends
 ///     messages to device
 ///   - Tunnelbroker then polls for incoming messages from device
 
 #[derive(Serialize, Deserialize, Debug, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub enum DeviceTypes {
   Mobile,
   Web,
   Keyserver,
 }
 
 /// Message sent by a client to Tunnelbroker to initiate a websocket
 /// session. Tunnelbroker will then validate the access token with identity
 /// service before continuing with the request.
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 #[serde(tag = "type", rename_all = "camelCase")]
 pub struct ConnectionInitializationMessage {
   #[serde(rename = "deviceID")]
   pub device_id: String,
   pub access_token: String,
   #[serde(rename = "userID")]
   pub user_id: String,
   pub notify_token: Option<String>,
   pub device_type: DeviceTypes,
   pub device_app_version: Option<String>,
   pub device_os: Option<String>,
 }
 
 #[derive(Serialize, Deserialize)]
 pub struct ConnectionInitializationResponse {
   pub session_id: String,
 }
 
 #[cfg(test)]
 mod session_tests {
   use super::*;
 
   #[test]
   fn test_session_deserialization() {
     let example_payload = r#"{
       "type": "sessionRequest",
       "accessToken": "xkdeifjsld",
       "deviceID": "foo",
       "userID": "alice",
       "deviceType": "keyserver"
     }"#;
 
     let request =
       serde_json::from_str::<ConnectionInitializationMessage>(example_payload)
         .unwrap();
     assert_eq!(request.device_id, "foo");
     assert_eq!(request.access_token, "xkdeifjsld");
     assert_eq!(request.device_os, None);
     assert_eq!(request.device_type, DeviceTypes::Keyserver);
   }
 }