Add capability for reset all sessions of a given channel.

Author: jheretic

crates/bitpart-cli/src/main.rs

@@ -111,6 +111,18 @@ enum Commands {
         device_name: String,
     },
 
+    /// reset all active chat sessions on a channel
+    #[command(arg_required_else_help = true)]
+    ChannelReset {
+        /// Channel ID
+        #[arg(short, long)]
+        id: String,
+
+        /// Bot ID
+        #[arg(short, long)]
+        bot_id: String,
+    },
+
     /// delete a bot
     #[command(arg_required_else_help = true)]
     Delete {
@@ -296,6 +308,17 @@ async fn main() -> Result<()> {
             send(&mut sender, &req).await?;
             hangup(&mut sender).await?;
         }
+        Commands::ChannelReset { id, bot_id } => {
+            let req = json!({"message_type": "ResetChannel",
+                "data" : {
+                "id": id,
+                "bot_id": bot_id,
+            }});
+            debug!("Request: {:?}", req.to_string());
+
+            send(&mut sender, &req).await?;
+            hangup(&mut sender).await?;
+        }
         Commands::Delete { id } => {
             let req = json!({"message_type": "DeleteBot",
                 "data" : {

crates/bitpart-common/src/error.rs

@@ -20,12 +20,15 @@ use figment;
 use futures;
 use hex;
 use opentelemetry_otlp;
-use presage;
+use presage::{
+    self,
+    libsignal_service::{prelude::InvalidDeviceId, protocol::SignalProtocolError},
+};
 use presage_store_bitpart::BitpartStoreError;
 use prost;
 use sea_orm::DbErr;
 use serde_json::Error as SerdeError;
-use std::{array, io};
+use std::{array, io, num::ParseIntError};
 use thiserror::Error;
 use thiserror_ext::Box;
 use tokio;
@@ -78,6 +81,12 @@ pub enum BitpartErrorKind {
     ProtocolBuffers(#[from] prost::UnknownEnumValue),
     #[error("Bincode error: `{0}`")]
     Bincode(#[from] bincode::Error),
+    #[error("Parse Int error: `{0}`")]
+    ParseInt(#[from] ParseIntError),
+    #[error("Invalid DeviceID error: `{0}`")]
+    InvalidDeviceId(#[from] InvalidDeviceId),
+    #[error("Signal Protocol error: `{0}`")]
+    SignalProtocol(#[from] SignalProtocolError),
 }
 
 impl<S: std::error::Error> From<presage::Error<S>> for BitpartErrorKind {

crates/bitpart-common/src/socket.rs

@@ -56,6 +56,10 @@ pub enum SocketMessage<S: Serialize> {
         bot_id: String,
         device_name: String,
     },
+    ResetChannel {
+        id: String,
+        bot_id: String,
+    },
     ChatRequest(Box<Request>),
     Response(Response<S>),
     Error(Response<S>),

crates/bitpart/src/api.rs

@@ -448,6 +448,26 @@ pub async fn start_channel(channel_id: &str, bot_id: &str, state: &mut ApiState)
     Ok(recv.await?)
 }
 
+pub async fn reset_channel(channel_id: &str, bot_id: &str, state: &mut ApiState) -> Result<String> {
+    let (send, recv) = oneshot::channel();
+    let contents = signal::ChannelMessageContents::ResetSessions {
+        id: channel_id.to_owned(),
+    };
+    let mut data = state.tokens.lock().await;
+    let token = data
+        .entry((bot_id.to_owned(), channel_id.to_owned()))
+        .or_insert(state.parent_token.child_token());
+    let msg = signal::ChannelMessage {
+        msg: contents,
+        db: state.db.clone(),
+        token: token.clone(),
+        tracker: state.tracker.clone(),
+        sender: send,
+    };
+    state.manager.send(msg);
+    Ok(recv.await?)
+}
+
 pub async fn read_channel(
     id: &str,
     bot_id: &str,

crates/bitpart/src/channels/signal.rs

@@ -81,6 +81,9 @@ pub enum ChannelMessageContents {
         id: String,
         attachments_dir: PathBuf,
     },
+    ResetSessions {
+        id: String,
+    },
 }
 
 pub struct ChannelMessage {
@@ -235,6 +238,31 @@ async fn process_channel_message(msg: ChannelMessage) -> Result<()> {
                     .map_err(BitpartErrorKind::Signal)?)
             }
         }
+        ChannelMessageContents::ResetSessions { id } => {
+            let store = BitpartStore::open(&id, &db, OnNewIdentity::Trust).await?;
+            if let Ok(mut manager) = Manager::load_registered(store).await {
+                let sessions: Vec<(String, Vec<u8>)> = manager.store().get_all("sessions").await?;
+                for (address, _) in sessions {
+                    let timestamp = std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .expect("Time went backwards")
+                        .as_millis() as u64;
+                    let addr: Vec<&str> = address.split('.').collect();
+                    let uuid = uuid::Uuid::parse_str(addr[0])?;
+                    manager
+                        .send_session_reset(&ServiceId::Aci(uuid.into()), timestamp)
+                        .await?
+                }
+                Ok(sender
+                    .send("".to_owned())
+                    .map_err(BitpartErrorKind::Signal)?)
+            } else {
+                warn!("Skipping startup of unregistered channel");
+                Ok(sender
+                    .send("".to_owned())
+                    .map_err(BitpartErrorKind::Signal)?)
+            }
+        }
     }
 }
 

crates/bitpart/src/socket.rs

@@ -159,6 +159,12 @@ async fn process_message(
                         Err(err) => wrap_error("ReadChannel", &err.to_string()),
                     }
                 }
+                SocketMessage::ResetChannel { id, bot_id } => {
+                    match api::reset_channel(&id, &bot_id, state).await {
+                        Ok(res) => wrap_response("ReadChannel", &res),
+                        Err(err) => wrap_error("ReadChannel", &err.to_string()),
+                    }
+                }
                 SocketMessage::ListChannels(options) => {
                     if let Some(paginate) = options {
                         match api::list_channels(paginate.limit, paginate.offset, state).await {