Properly handle start sync thread issue

matrixgw_backend/src/controllers/matrix_sync_thread_controller.rs

@@ -20,3 +20,40 @@ pub async fn start_sync(
         }
     }
 }
+
+/// Stop sync thread
+pub async fn stop_sync(
+    client: MatrixClientExtractor,
+    manager: web::Data<ActorRef<MatrixManagerMsg>>,
+) -> HttpResult {
+    match ractor::cast!(
+        manager,
+        MatrixManagerMsg::StopSyncThread(client.auth.user.email.clone())
+    ) {
+        Ok(_) => Ok(HttpResponse::Accepted().finish()),
+        Err(e) => {
+            log::error!("Failed to stop sync thread: {e}");
+            Ok(HttpResponse::InternalServerError().finish())
+        }
+    }
+}
+
+#[derive(serde::Serialize)]
+struct GetSyncStatusResponse {
+    started: bool,
+}
+
+/// Get sync thread status
+pub async fn status(
+    client: MatrixClientExtractor,
+    manager: web::Data<ActorRef<MatrixManagerMsg>>,
+) -> HttpResult {
+    let started = ractor::call!(
+        manager.as_ref(),
+        MatrixManagerMsg::SyncThreadGetStatus,
+        client.auth.user.email
+    )
+    .expect("RPC to Matrix Manager failed");
+
+    Ok(HttpResponse::Ok().json(GetSyncStatusResponse { started }))
+}

matrixgw_backend/src/extractors/matrix_client_extractor.rs

@@ -39,9 +39,13 @@ impl FromRequest for MatrixClientExtractor {
                 matrix_manager_actor,
                 MatrixManagerMsg::GetClient,
                 auth.user.email.clone()
-            )
+            );
-            .expect("Failed to query manager actor!")
+
-            .expect("Failed to get client!");
+            let client = match client {
+                Ok(Ok(client)) => client,
+                Ok(Err(err)) => panic!("Failed to get client! {err:?}"),
+                Err(err) => panic!("Failed to query manager actor! {err:#?}"),
+            };
 
             Ok(Self { auth, client })
         })

matrixgw_backend/src/main.rs

@@ -125,6 +125,14 @@ async fn main() -> std::io::Result<()> {
                 "/api/matrix_sync/start",
                 web::post().to(matrix_sync_thread_controller::start_sync),
             )
+            .route(
+                "/api/matrix_sync/stop",
+                web::post().to(matrix_sync_thread_controller::stop_sync),
+            )
+            .route(
+                "/api/matrix_sync/status",
+                web::get().to(matrix_sync_thread_controller::status),
+            )
     })
     .workers(4)
     .bind(&AppConfig::get().listen_address)?

matrixgw_backend/src/matrix_connection/matrix_manager.rs

@@ -15,6 +15,8 @@ pub enum MatrixManagerMsg {
     GetClient(UserEmail, RpcReplyPort<anyhow::Result<MatrixClient>>),
     DisconnectClient(UserEmail),
     StartSyncThread(UserEmail),
+    StopSyncThread(UserEmail),
+    SyncThreadGetStatus(UserEmail, RpcReplyPort<bool>),
     SyncThreadTerminated(UserEmail, MatrixSyncTaskID),
 }
 
@@ -37,6 +39,15 @@ impl Actor for MatrixManagerActor {
         })
     }
 
+    async fn post_stop(
+        &self,
+        _myself: ActorRef<Self::Msg>,
+        _state: &mut Self::State,
+    ) -> Result<(), ActorProcessingErr> {
+        log::error!("[!] [!] Matrix Manager Actor stopped!");
+        Ok(())
+    }
+
     async fn handle(
         &self,
         myself: ActorRef<Self::Msg>,
@@ -113,10 +124,32 @@ impl Actor for MatrixManagerActor {
                 // Start thread
                 log::debug!("Starting sync thread for {email:?}");
                 let thread_id =
-                    start_sync_thread(client.clone(), state.broadcast_sender.clone(), myself)
+                    match start_sync_thread(client.clone(), state.broadcast_sender.clone(), myself)
-                        .await?;
+                        .await
+                    {
+                        Ok(thread_id) => thread_id,
+                        Err(e) => {
+                            log::error!("Failed to start sync thread! {e}");
+                            return Ok(());
+                        }
+                    };
                 state.running_sync_threads.insert(email, thread_id);
             }
+            MatrixManagerMsg::StopSyncThread(email) => {
+                if let Some(thread_id) = state.running_sync_threads.get(&email)
+                    && let Err(e) = state
+                        .broadcast_sender
+                        .send(BroadcastMessage::StopSyncThread(thread_id.clone()))
+                {
+                    log::error!("Failed to request sync thread stop: {e}");
+                }
+            }
+            MatrixManagerMsg::SyncThreadGetStatus(email, reply) => {
+                let started = state.running_sync_threads.contains_key(&email);
+                if let Err(e) = reply.send(started) {
+                    log::error!("Failed to send sync thread status! {e}");
+                }
+            }
             MatrixManagerMsg::SyncThreadTerminated(email, task_id) => {
                 if state.running_sync_threads.get(&email) == Some(&task_id) {
                     log::info!(