New APIs for message processing loops

This patch adds new C APIs
dc_get_next_msgs() and dc_wait_next_msgs(),
and their JSON-RPC counterparts
get_next_msgs() and wait_next_msgs().

New configuration "last_msg_id"
tracks the last message ID processed by the bot.
get_next_msgs() returns message IDs above
the "last_msg_id".
wait_next_msgs() waits for new message notification
and calls get_next_msgs().
wait_next_msgs() can be used to build
a separate message processing loop
independent of the event loop.

Async Python API get_fresh_messages_in_arrival_order()
is deprecated in favor of get_next_messages().

Introduced Python APIs:
- Account.wait_next_incoming_message()
- Message.is_from_self()
- Message.is_from_device()

Introduced Rust APIs:
- Context.set_config_u32()
- Context.get_config_u32()

src/chat.rs

@@ -1666,6 +1666,7 @@ impl Chat {
                     ],
                 )
                 .await?;
+            context.new_msgs_notify.notify_one();
             msg.id = MsgId::new(u32::try_from(raw_id)?);
 
             maybe_set_logging_xdc(context, msg, self.id).await?;
@@ -3628,6 +3629,7 @@ pub async fn add_device_msg_with_importance(
                 ),
             )
             .await?;
+        context.new_msgs_notify.notify_one();
 
         msg_id = MsgId::new(u32::try_from(row_id)?);
         if !msg.hidden {
@@ -3741,6 +3743,7 @@ pub(crate) async fn add_info_msg_with_cmd(
             parent.map(|msg|msg.rfc724_mid.clone()).unwrap_or_default()
         )
     ).await?;
+    context.new_msgs_notify.notify_one();
 
     let msg_id = MsgId::new(row_id.try_into()?);
     context.emit_msgs_changed(chat_id, msg_id);

src/config.rs

@@ -308,6 +308,9 @@ pub enum Config {
     /// This value is used internally to remember the MsgId of the logging xdc
     #[strum(props(default = "0"))]
     DebugLogging,
+
+    /// Last message processed by the bot.
+    LastMsgId,
 }
 
 impl Context {
@@ -358,6 +361,11 @@ impl Context {
         Ok(self.get_config_parsed(key).await?.unwrap_or_default())
     }
 
+    /// Returns 32-bit unsigned integer configuration value for the given key.
+    pub async fn get_config_u32(&self, key: Config) -> Result<u32> {
+        Ok(self.get_config_parsed(key).await?.unwrap_or_default())
+    }
+
     /// Returns 64-bit signed integer configuration value for the given key.
     pub async fn get_config_i64(&self, key: Config) -> Result<i64> {
         Ok(self.get_config_parsed(key).await?.unwrap_or_default())
@@ -459,6 +467,12 @@ impl Context {
         Ok(())
     }
 
+    /// Set the given config to an unsigned 32-bit integer value.
+    pub async fn set_config_u32(&self, key: Config, value: u32) -> Result<()> {
+        self.set_config(key, Some(&value.to_string())).await?;
+        Ok(())
+    }
+
     /// Set the given config to a boolean value.
     pub async fn set_config_bool(&self, key: Config, value: bool) -> Result<()> {
         self.set_config(key, if value { Some("1") } else { Some("0") })

src/context.rs

@@ -11,7 +11,7 @@ use std::time::{Duration, Instant, SystemTime};
 use anyhow::{bail, ensure, Context as _, Result};
 use async_channel::{self as channel, Receiver, Sender};
 use ratelimit::Ratelimit;
-use tokio::sync::{Mutex, RwLock};
+use tokio::sync::{Mutex, Notify, RwLock};
 use tokio::task;
 
 use crate::chat::{get_chat_cnt, ChatId};
@@ -218,6 +218,11 @@ pub struct InnerContext {
     /// IMAP UID resync request.
     pub(crate) resync_request: AtomicBool,
 
+    /// Notify about new messages.
+    ///
+    /// This causes [`Context::wait_next_msgs`] to wake up.
+    pub(crate) new_msgs_notify: Notify,
+
     /// Server ID response if ID capability is supported
     /// and the server returned non-NIL on the inbox connection.
     /// <https://datatracker.ietf.org/doc/html/rfc2971>
@@ -363,6 +368,11 @@ impl Context {
             blobdir.display()
         );
 
+        let new_msgs_notify = Notify::new();
+        // Notify once immediately to allow processing old messages
+        // without starting I/O.
+        new_msgs_notify.notify_one();
+
         let inner = InnerContext {
             id,
             blobdir,
@@ -379,6 +389,7 @@ impl Context {
             quota: RwLock::new(None),
             quota_update_request: AtomicBool::new(false),
             resync_request: AtomicBool::new(false),
+            new_msgs_notify,
             server_id: RwLock::new(None),
             creation_time: std::time::SystemTime::now(),
             last_full_folder_scan: Mutex::new(None),
@@ -767,6 +778,10 @@ impl Context {
             "debug_logging",
             self.get_config_int(Config::DebugLogging).await?.to_string(),
         );
+        res.insert(
+            "last_msg_id",
+            self.get_config_int(Config::LastMsgId).await?.to_string(),
+        );
 
         let elapsed = self.creation_time.elapsed();
         res.insert("uptime", duration_to_str(elapsed.unwrap_or_default()));
@@ -813,6 +828,66 @@ impl Context {
         Ok(list)
     }
 
+    /// Returns a list of messages with database ID higher than requested.
+    ///
+    /// Blocked contacts and chats are excluded,
+    /// but self-sent messages and contact requests are included in the results.
+    pub async fn get_next_msgs(&self) -> Result<Vec<MsgId>> {
+        let last_msg_id = match self.get_config(Config::LastMsgId).await? {
+            Some(s) => MsgId::new(s.parse()?),
+            None => MsgId::new_unset(),
+        };
+
+        let list = self
+            .sql
+            .query_map(
+                "SELECT m.id
+                     FROM msgs m
+                     LEFT JOIN contacts ct
+                            ON m.from_id=ct.id
+                     LEFT JOIN chats c
+                            ON m.chat_id=c.id
+                     WHERE m.id>?
+                       AND m.hidden=0
+                       AND m.chat_id>9
+                       AND ct.blocked=0
+                       AND c.blocked!=1
+                     ORDER BY m.id ASC",
+                (
+                    last_msg_id.to_u32(), // Explicitly convert to u32 because 0 is allowed.
+                ),
+                |row| {
+                    let msg_id: MsgId = row.get(0)?;
+                    Ok(msg_id)
+                },
+                |rows| {
+                    let mut list = Vec::new();
+                    for row in rows {
+                        list.push(row?);
+                    }
+                    Ok(list)
+                },
+            )
+            .await?;
+        Ok(list)
+    }
+
+    /// Returns a list of messages with database ID higher than last marked as seen.
+    ///
+    /// This function is supposed to be used by bot to request messages
+    /// that are not processed yet.
+    ///
+    /// Waits for notification and returns a result.
+    /// Note that the result may be empty if the message is deleted
+    /// shortly after notification or notification is manually triggered
+    /// to interrupt waiting.
+    /// Notification may be manually triggered by calling [`Self::stop_io`].
+    pub async fn wait_next_msgs(&self) -> Result<Vec<MsgId>> {
+        self.new_msgs_notify.notified().await;
+        let list = self.get_next_msgs().await?;
+        Ok(list)
+    }
+
     /// Searches for messages containing the query string.
     ///
     /// If `chat_id` is provided this searches only for messages in this chat, if `chat_id`
@@ -1444,4 +1519,38 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_get_next_msgs() -> Result<()> {
+        let alice = TestContext::new_alice().await;
+        let bob = TestContext::new_bob().await;
+
+        let alice_chat = alice.create_chat(&bob).await;
+
+        assert!(alice.get_next_msgs().await?.is_empty());
+        assert!(bob.get_next_msgs().await?.is_empty());
+
+        let sent_msg = alice.send_text(alice_chat.id, "Hi Bob").await;
+        let received_msg = bob.recv_msg(&sent_msg).await;
+
+        let bob_next_msg_ids = bob.get_next_msgs().await?;
+        assert_eq!(bob_next_msg_ids.len(), 1);
+        assert_eq!(bob_next_msg_ids.get(0), Some(&received_msg.id));
+
+        bob.set_config_u32(Config::LastMsgId, received_msg.id.to_u32())
+            .await?;
+        assert!(bob.get_next_msgs().await?.is_empty());
+
+        // Next messages include self-sent messages.
+        let alice_next_msg_ids = alice.get_next_msgs().await?;
+        assert_eq!(alice_next_msg_ids.len(), 1);
+        assert_eq!(alice_next_msg_ids.get(0), Some(&sent_msg.sender_msg_id));
+
+        alice
+            .set_config_u32(Config::LastMsgId, sent_msg.sender_msg_id.to_u32())
+            .await?;
+        assert!(alice.get_next_msgs().await?.is_empty());
+
+        Ok(())
+    }
 }

src/message.rs

@@ -1468,6 +1468,12 @@ pub async fn markseen_msgs(context: &Context, msg_ids: Vec<MsgId>) -> Result<()>
         return Ok(());
     }
 
+    let old_last_msg_id = MsgId::new(context.get_config_u32(Config::LastMsgId).await?);
+    let last_msg_id = msg_ids.iter().fold(&old_last_msg_id, std::cmp::max);
+    context
+        .set_config_u32(Config::LastMsgId, last_msg_id.to_u32())
+        .await?;
+
     let msgs = context
         .sql
         .query_map(

src/receive_imf.rs

@@ -363,6 +363,7 @@ pub(crate) async fn receive_imf_inner(
             chat_id.emit_msg_event(context, *msg_id, incoming && fresh);
         }
     }
+    context.new_msgs_notify.notify_one();
 
     mime_parser
         .handle_reports(context, from_id, sent_timestamp, &mime_parser.parts)

src/scheduler.rs

@@ -62,6 +62,11 @@ impl SchedulerState {
     /// Starts the scheduler if it is not yet started.
     async fn do_start(mut inner: RwLockWriteGuard<'_, InnerSchedulerState>, context: Context) {
         info!(context, "starting IO");
+
+        // Notify message processing loop
+        // to allow processing old messages after restart.
+        context.new_msgs_notify.notify_one();
+
         let ctx = context.clone();
         match Scheduler::start(context).await {
             Ok(scheduler) => *inner = InnerSchedulerState::Started(scheduler),
@@ -95,6 +100,11 @@ impl SchedulerState {
         // to terminate on receiving the next event and then call stop_io()
         // which will emit the below event(s)
         info!(context, "stopping IO");
+
+        // Wake up message processing loop even if there are no messages
+        // to allow for clean shutdown.
+        context.new_msgs_notify.notify_one();
+
         if let Some(debug_logging) = context.debug_logging.read().await.as_ref() {
             debug_logging.loop_handle.abort();
         }