fix: adapt to grammers API changes (Session::Error, fallible refs)

Recent grammers changes introduced an associated Error type on the
Session trait and made several methods fallible:

- SqliteSession::open now returns a SqliteSessionError.
- SenderPool::new requires S::Error: std::error::Error + Send + Sync + 'static.
- peer_ref, sender_ref, and Peer::to_ref now return Result<Option<PeerRef>, Box<dyn Error>>.
- Client::stream_updates and UpdateStream::sync_update_state now return Result.

Update the codebase accordingly and fix the from-env example.

Verified: cargo check --all-targets, cargo test, cargo clippy --all-targets, cargo fmt --check.

Author: alissonlauffer

ferogram/examples/from-env.rs

@@ -15,6 +15,8 @@ use ferogram::prelude::ConnectionExt;
 use grammers::{Client, SenderPool, client::UpdatesConfiguration, update::Update};
 use tokio::{task::JoinSet, time::sleep};
 
+type Result = std::result::Result<(), Box<dyn Error + Send + Sync>>;
+
 async fn handle_update(client: Client, update: Update) {
     match update {
         Update::NewMessage(message) if !message.outgoing() => {
@@ -27,7 +29,7 @@ async fn handle_update(client: Client, update: Update) {
                 sleep(Duration::from_secs(5)).await;
             }
             if let Err(e) = client
-                .send_message(peer.to_ref().await.unwrap(), message.text())
+                .send_message(peer.to_ref().await.unwrap().unwrap(), message.text())
                 .await
             {
                 println!("Failed to respond! {e}");
@@ -38,7 +40,7 @@ async fn handle_update(client: Client, update: Update) {
 }
 
 #[tokio::main(flavor = "multi_thread")]
-async fn main() -> Result<(), Box<dyn Error>> {
+async fn main() -> Result {
     println!("Connecting...");
 
     // Connect the client from environment variables.
@@ -65,7 +67,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 ..Default::default()
             },
         )
-        .await;
+        .await?;
 
     loop {
         // Empty finished handlers (you could look at their return value here too.)
@@ -87,7 +89,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     }
 
     println!("Saving session file...");
-    updates.sync_update_state().await;
+    updates.sync_update_state().await?;
 
     // Pool's `run()` won't finish until all handles are dropped or quit is called.
     // Here there are at least three handles alive: `handle`, `client` and `updates`

ferogram/src/client.rs

@@ -46,7 +46,9 @@ pub trait ConnectionExt {
         api_id: i32,
         api_hash: &str,
         session: Arc<S>,
-    ) -> impl Future<Output = Result<(SenderPool, Client), ClientError>> + Send;
+    ) -> impl Future<Output = Result<(SenderPool, Client), ClientError>> + Send
+    where
+        S::Error: std::error::Error + Send + Sync + 'static;
 
     /// Like [`Self::connect`] but with a custom [`ClientConfiguration`].
     fn connect_with_configuration<S: Session + 'static>(
@@ -55,7 +57,9 @@ pub trait ConnectionExt {
         api_hash: &str,
         session: Arc<S>,
         configuration: ClientConfiguration,
-    ) -> impl Future<Output = Result<(SenderPool, Client), ClientError>> + Send;
+    ) -> impl Future<Output = Result<(SenderPool, Client), ClientError>> + Send
+    where
+        S::Error: std::error::Error + Send + Sync + 'static;
 }
 
 impl ConnectionExt for Client {
@@ -86,7 +90,11 @@ impl ConnectionExt for Client {
 
         let session_path =
             std::env::var("SESSION_FILE").unwrap_or_else(|_| "grammers.session".to_string());
-        let session = Arc::new(SqliteSession::open(session_path).await?);
+        let session = Arc::new(
+            SqliteSession::open(session_path)
+                .await
+                .map_err(|e| ClientError::SessionError(e.into()))?,
+        );
 
         let account = bot_token.unwrap_or_else(|_| phone_number.unwrap());
         Self::connect_with_configuration(&account, api_id, &api_hash, session, configuration).await
@@ -97,7 +105,10 @@ impl ConnectionExt for Client {
         api_id: i32,
         api_hash: &str,
         session: Arc<S>,
-    ) -> Result<(SenderPool, Self), ClientError> {
+    ) -> Result<(SenderPool, Self), ClientError>
+    where
+        S::Error: std::error::Error + Send + Sync + 'static,
+    {
         Self::connect_with_configuration(account, api_id, api_hash, session, Default::default())
             .await
     }
@@ -108,7 +119,10 @@ impl ConnectionExt for Client {
         api_hash: &str,
         session: Arc<S>,
         configuration: ClientConfiguration,
-    ) -> Result<(SenderPool, Self), ClientError> {
+    ) -> Result<(SenderPool, Self), ClientError>
+    where
+        S::Error: std::error::Error + Send + Sync + 'static,
+    {
         // Test session validity.
         {
             let SenderPool { runner, handle, .. } = SenderPool::new(Arc::clone(&session), api_id);

ferogram/src/context.rs

@@ -51,13 +51,15 @@ impl Context {
     }
 
     /// Cached reference to [`Self::peer`], if it is in cache.
-    pub async fn peer_ref(&self) -> Option<PeerRef> {
+    pub async fn peer_ref(
+        &self,
+    ) -> Result<Option<PeerRef>, Box<dyn std::error::Error + Send + Sync>> {
         match &self.update {
             Update::NewMessage(message) | Update::MessageEdited(message) => {
                 message.peer_ref().await
             }
             Update::CallbackQuery(query) => query.peer_ref().await,
-            _ => None,
+            _ => Ok(None),
         }
     }
 
@@ -75,15 +77,17 @@ impl Context {
     }
 
     /// Cached reference to [`Self::sender`], if it is in cache.
-    pub async fn sender_ref(&self) -> Option<PeerRef> {
+    pub async fn sender_ref(
+        &self,
+    ) -> Result<Option<PeerRef>, Box<dyn std::error::Error + Send + Sync>> {
         match &self.update {
             Update::NewMessage(message) | Update::MessageEdited(message) => {
                 message.sender_ref().await
             }
             Update::CallbackQuery(query) => query.sender_ref().await,
             Update::InlineQuery(query) => query.sender_ref().await,
             Update::InlineSend(send) => send.sender_ref().await,
-            _ => None,
+            _ => Ok(None),
         }
     }
 
@@ -188,9 +192,9 @@ impl Context {
     }
 
     /// Chat action sender.
-    pub async fn action(&self) -> ActionSender {
-        let peer = self.peer_ref().await.unwrap();
-        self.client.action(peer)
+    pub async fn action(&self) -> Result<ActionSender, InvocationError> {
+        let peer = self.peer_ref().await?.unwrap();
+        Ok(self.client.action(peer))
     }
 
     /// Check if the peer is a group.
@@ -317,7 +321,7 @@ impl Context {
     ///
     /// If the update is a [`CallbackQuery`], it will load the message first.
     pub async fn forward_to_self(&self) -> Result<Message, InvocationError> {
-        let chat = self.client.get_me().await?.to_ref().await.unwrap();
+        let chat = self.client.get_me().await?.to_ref().await?.unwrap();
 
         self.forward_to(chat).await
     }

ferogram/src/dispatcher.rs

@@ -88,7 +88,15 @@ impl Dispatcher {
             let _guard = DispatcherExitGuard;
 
             let mut handler_tasks = JoinSet::new();
-            let mut updates = client.stream_updates(updates, configuration).await;
+            let mut updates = match client.stream_updates(updates, configuration).await {
+                Ok(updates) => updates,
+                Err(e) => {
+                    tracing::error!("Failed to start update stream: {e}");
+                    handle.quit();
+                    let _ = pool_task.await;
+                    return;
+                }
+            };
 
             loop {
                 tokio::select! {
@@ -181,7 +189,9 @@ impl Dispatcher {
             }
 
             tracing::info!("Saving session...");
-            updates.sync_update_state().await;
+            if let Err(e) = updates.sync_update_state().await {
+                tracing::error!("Failed to save session: {e}");
+            }
 
             tracing::info!("Exiting...");
             handle.quit();

ferogram/src/error.rs

@@ -19,6 +19,8 @@ pub enum ClientError {
     DatabaseError(#[from] libsql::Error),
     #[error("failed to establish connection to telegram: {0:?}")]
     ConnectionError(#[from] InvocationError),
+    #[error("failed to use session: {0:?}")]
+    SessionError(Box<dyn std::error::Error + Send + Sync>),
     #[error("variable `{0}` were expected, but none was found")]
     ExpectedVariable(String),
 }