various small cleanups

ntfy-daemon/src/message_repo/mod.rs

@@ -26,11 +26,9 @@ impl Db {
         Ok(this)
     }
     fn migrate(&mut self) -> Result<()> {
-        {
+        self.conn
-            self.conn
+            .borrow()
-                .borrow()
+            .execute_batch(include_str!("./migrations/00.sql"))?;
-                .execute_batch(include_str!("./migrations/00.sql"))?
-        };
         Ok(())
     }
     fn get_or_insert_server(&mut self, server: &str) -> Result<i64> {
@@ -90,9 +88,9 @@ impl Db {
         ",
         )?;
         let msgs: Result<Vec<String>, _> = stmt
-            .query_map(params![server, topic, since], |row| Ok(row.get(0)?))?
+            .query_map(params![server, topic, since], |row| row.get(0))?
             .collect();
-        Ok(msgs?)
+        msgs
     }
     pub fn insert_subscription(&mut self, sub: models::Subscription) -> Result<(), Error> {
         let server_id = self.get_or_insert_server(&sub.server)?;
@@ -116,7 +114,7 @@ impl Db {
             WHERE server = ?1 AND topic = ?2",
             params![server_id, topic],
         )?;
-        if res <= 0 {
+        if res == 0 {
             return Err(Error::SubscriptionNotFound("removing subscription".into()));
         }
         Ok(())
@@ -162,7 +160,7 @@ impl Db {
                 sub.topic,
             ],
         )?;
-        if res <= 0 {
+        if res == 0 {
             return Err(Error::SubscriptionNotFound("updating subscription".into()));
         }
         info!(info = ?sub, "stored subscription info");
@@ -184,7 +182,7 @@ impl Db {
             ",
             params![server_id, topic, value],
         )?;
-        if res <= 0 {
+        if res == 0 {
             return Err(Error::SubscriptionNotFound("updating read_until".into()));
         }
         Ok(())
@@ -198,7 +196,7 @@ impl Db {
             ",
             params![server_id, topic],
         )?;
-        if res <= 0 {
+        if res == 0 {
             return Err(Error::SubscriptionNotFound("deleting messages".into()));
         }
         Ok(())

ntfy-daemon/src/models.rs

@@ -148,7 +148,7 @@ impl Subscription {
         let mut url = url::Url::parse(server)?;
         url.path_segments_mut()
             .map_err(|_| url::ParseError::RelativeUrlWithCannotBeABaseBase)?
-            .push(&topic)
+            .push(topic)
             .push("json");
         url.query_pairs_mut()
             .append_pair("since", &since.to_string());
@@ -162,7 +162,7 @@ impl Subscription {
         if let Err(e) = Self::build_url(&self.server, &self.topic, 0) {
             errs.push(e);
         };
-        if errs.len() > 0 {
+        if !errs.is_empty() {
             return Err(errs);
         }
         Ok(self)

ntfy-daemon/src/system_client.rs

@@ -63,7 +63,7 @@ impl output_channel::Server for NotifyForwarder {
         let already_stored: bool = {
             // If this fails parsing, the message is not valid at all.
             // The server is probably misbehaving.
-            let min_message: MinMessage = pry!(serde_json::from_str(&message)
+            let min_message: MinMessage = pry!(serde_json::from_str(message)
                 .map_err(|e| Error::InvalidMinMessage(message.to_string(), e)));
             let model = self.model.borrow();
             match self.env.db.insert_message(&model.server, message) {
@@ -83,20 +83,15 @@ impl output_channel::Server for NotifyForwarder {
             // Show notification
             // Our priority is to show notifications. If anything fails, panic.
             if !{ self.model.borrow().muted } {
-                let msg: Message = pry!(serde_json::from_str(&message)
+                let msg: Message = pry!(serde_json::from_str(message)
                     .map_err(|e| Error::InvalidMessage(message.to_string(), e)));
                 let np = self.env.proxy.clone();
 
-                let title = { msg.notification_title(&*self.model.borrow()) };
+                let title = { msg.notification_title(&self.model.borrow()) };
 
                 let n = models::Notification {
-                    title: title.to_string(),
+                    title,
-                    body: msg
+                    body: msg.display_message().as_deref().unwrap_or("").to_string(),
-                        .display_message()
-                        .as_ref()
-                        .map(|x| x.as_str())
-                        .unwrap_or("")
-                        .to_string(),
                     actions: msg.actions,
                 };
 
@@ -380,7 +375,7 @@ impl SystemNotifier {
     pub fn watch_subscribed(&mut self) -> Promise<(), capnp::Error> {
         let f: Vec<_> = pry!(self.env.db.list_subscriptions())
             .into_iter()
-            .map(|m| self.watch(m.clone()))
+            .map(|m| self.watch(m))
             .collect();
         Promise::from_future(async move {
             join_all(f.into_iter().map(|x| async move {
@@ -434,7 +429,7 @@ impl system_notifier::Server for SystemNotifier {
             pry!(self
                 .env
                 .db
-                .remove_subscription(&server, &topic)
+                .remove_subscription(server, topic)
                 .map_err(|e| capnp::Error::failed(e.to_string())));
             info!(server, topic, "Unsubscribed");
         }

src/subscription.rs

@@ -9,7 +9,7 @@ use glib::Properties;
 use gtk::{gio, glib};
 use ntfy_daemon::models;
 use ntfy_daemon::ntfy_capnp::{output_channel, subscription, watch_handle, Status};
-use tracing::{debug, debug_span, error, instrument};
+use tracing::{debug, error, instrument};
 
 struct TopicWatcher {
     sub: glib::WeakRef<Subscription>,
@@ -217,8 +217,7 @@ impl Subscription {
         val.set_display_name(&*imp.display_name.borrow());
         val.set_read_until(imp.read_until.get());
         Promise::from_future(async move {
-            let _span = debug_span!("send_updated_info").entered();
+            debug!("sending update_info");
-            debug!("sending");
             req.send().promise.await?;
             Ok(())
         })
@@ -233,12 +232,8 @@ impl Subscription {
     }
     fn update_unread_count(&self) {
         let imp = self.imp();
-        if let Some(last) = Self::last_message(&imp.messages) {
+        if Self::last_message(&imp.messages).map(|last| last.time) > Some(imp.read_until.get()) {
-            if last.time > imp.read_until.get() {
+            imp.unread_count.set(1);
-                imp.unread_count.set(1);
-            } else {
-                imp.unread_count.set(0);
-            }
         } else {
             imp.unread_count.set(0);
         }
@@ -256,13 +251,12 @@ impl Subscription {
     }
     pub fn flag_all_as_read(&self) -> Promise<(), capnp::Error> {
         let imp = self.imp();
-        let Some(last) = Self::last_message(&imp.messages) else {
+        let Some(value) = Self::last_message(&imp.messages)
+            .map(|last| last.time)
+            .filter(|time| *time > self.imp().read_until.get())
+        else {
             return Promise::ok(());
         };
-        let value = last.time;
-        if self.imp().read_until.get() == value {
-            return Promise::ok(());
-        }
 
         let this = self.clone();
         Promise::from_future(async move {
@@ -276,14 +270,15 @@ impl Subscription {
     }
     pub fn publish_msg(&self, mut msg: models::Message) -> Promise<(), capnp::Error> {
         let imp = self.imp();
+        let json = {
+            msg.topic = self.topic();
+            serde_json::to_string(&msg).map_err(|e| capnp::Error::failed(e.to_string()))
+        };
         let mut req = imp.client.get().unwrap().publish_request();
-        msg.topic = self.topic();
-        let json = serde_json::to_string(&msg).map_err(|e| capnp::Error::failed(e.to_string()));
         req.get().set_message(&pry!(json));
 
         Promise::from_future(async move {
-            let _span = debug_span!("publish").entered();
+            debug!("sending publish");
-            debug!("sending");
             req.send().promise.await?;
             Ok(())
         })
@@ -294,8 +289,7 @@ impl Subscription {
         let req = imp.client.get().unwrap().clear_notifications_request();
         let this = self.clone();
         Promise::from_future(async move {
-            let _span = debug_span!("clear_notifications").entered();
+            debug!("sending clear_notifications");
-            debug!("sending");
             req.send().promise.await?;
             this.imp().messages.remove_all();
             Ok(())

src/widgets/message_row.rs

@@ -145,39 +145,29 @@ impl MessageRow {
             self.attach(&tags, 0, row, 3, 1);
         }
     }
+    fn fetch_image_bytes(url: &str) -> anyhow::Result<Vec<u8>> {
+        let path = glib::user_cache_dir().join("com.ranfdev.Notify").join(&url);
+        let bytes = if path.exists() {
+            std::fs::read(&path)?
+        } else {
+            let mut bytes = vec![];
+            ureq::get(&url)
+                .call()?
+                .into_reader()
+                .take(5 * 1_000_000) // 5 MB
+                .read_to_end(&mut bytes)?;
+            bytes
+        };
+        Ok(bytes)
+    }
     fn build_image(&self, url: String) -> gtk::Picture {
         let (tx, rx) = glib::MainContext::channel(Default::default());
         gio::spawn_blocking(move || {
-            let path = glib::user_cache_dir().join("com.ranfdev.Notify").join(&url);
+            if let Err(e) =
-            let bytes = if path.exists() {
+                Self::fetch_image_bytes(&url).map(|bytes| tx.send(glib::Bytes::from_owned(bytes)))
-                match std::fs::read(&path) {
+            {
-                    Ok(v) => v,
+                error!(error = %e)
-                    Err(e) => {
+            }
-                        error!(error = %e, path = %path.display(), "reading image from disk");
-                        return glib::ControlFlow::Break;
-                    }
-                }
-            } else {
-                let res = match ureq::get(&url).call() {
-                    Ok(res) => res,
-                    Err(e) => {
-                        error!(error = %e, "fetching image");
-                        return glib::ControlFlow::Break;
-                    }
-                };
-                let mut bytes = vec![];
-                if let Err(e) = res
-                    .into_reader()
-                    .take(5 * 1_000_000) // 5 MB
-                    .read_to_end(&mut bytes)
-                {
-                    error!(error = %e, "reading image data");
-                    return glib::ControlFlow::Break;
-                }
-                bytes
-            };
-
-            tx.send(glib::Bytes::from_owned(bytes)).unwrap();
             glib::ControlFlow::Break
         });
         let picture = gtk::Picture::new();

src/widgets/window.rs

@@ -346,7 +346,7 @@ impl NotifyWindow {
             .bind_model(Some(&imp.subscription_list_model), |obj| {
                 let sub = obj.downcast_ref::<Subscription>().unwrap();
 
-                Self::build_subscription_ui(&sub).upcast()
+                Self::build_subscription_row(&sub).upcast()
             });
 
         let this = self.clone();
@@ -445,7 +445,7 @@ impl NotifyWindow {
         chip
     }
 
-    fn build_subscription_ui(sub: &Subscription) -> impl glib::IsA<gtk::Widget> {
+    fn build_subscription_row(sub: &Subscription) -> impl glib::IsA<gtk::Widget> {
         let b = gtk::Box::builder().spacing(4).build();
 
         let label = gtk::Label::builder()