fix: send a flag change to where the server still has the message

crates/flectar-mail-core/src/db/repo/actions.rs

@@ -366,6 +366,33 @@ pub fn has_active_move_from(conn: &Connection, message_id: i64, folder_id: i64)
 }
 
 /// Most recent undoable action (pending or just done, within the window).
+/// Where the first move of `message_id` queued after action `after` takes it
+/// from: its folder, and its UID there when known. Until that move runs, the
+/// server still has the message there, though the local row shows it moved.
+pub fn next_move_source(
+    conn: &Connection,
+    message_id: i64,
+    after: i64,
+) -> Result<Option<(i64, Option<i64>)>> {
+    let payload: Option<String> = conn
+        .query_row(
+            "SELECT payload FROM pending_actions
+             WHERE message_id = ?1 AND id > ?2 AND state IN ('pending', 'inflight')
+               AND kind IN ('archive', 'unarchive', 'trash', 'spam', 'not_spam', 'move')
+             ORDER BY id LIMIT 1",
+            params![message_id, after],
+            |row| row.get(0),
+        )
+        .optional()?;
+    let Some(payload) = payload else {
+        return Ok(None);
+    };
+    let payload: serde_json::Value = parse_json_column(&payload, 0)?;
+    Ok(payload["srcFolderId"]
+        .as_i64()
+        .map(|folder_id| (folder_id, payload["srcUid"].as_i64())))
+}
+
 pub fn last_undoable(conn: &Connection, since_ms: i64) -> Result<Option<PendingAction>> {
     let mut stmt = conn.prepare(
         "SELECT id, account_id, kind, message_id, thread_id, payload, state,

crates/flectar-mail-core/src/queue/mod.rs

@@ -524,21 +524,32 @@ async fn execute_on_imap(
     }
 }
 
-/// Resolve the message's current remote (folder_name, uid); selects the folder.
+/// Resolve where the server has the message now, for `action` on it, and
+/// select that folder. Actions run in order, so a move queued after this one
+/// hasn't run yet: the server still has the message where that move takes it
+/// from, though the local row already shows it moved.
 async fn resolve_remote(
     ctx: &SyncCtx,
     session: &mut Session,
+    action: &repo::actions::PendingAction,
     message_id: i64,
 ) -> Result<Option<(repo::folders::Folder, u32)>> {
-    let row = ctx
+    let after = action.id;
+    let (row, waiting_move) = ctx
         .db
-        .read(move |conn| repo::messages::get_row(conn, message_id))
+        .read(move |conn| {
+            Ok((
+                repo::messages::get_row(conn, message_id)?,
+                repo::actions::next_move_source(conn, message_id, after)?,
+            ))
+        })
         .await?;
     let Some(row) = row else { return Ok(None) };
-
-    // The payload's remote location: where the message lived when the action
-    // was enqueued (optimistic mutation may have already moved it locally).
-    let Some(folder_id) = row.folder_id else {
+    let (folder_id, known_uid) = match waiting_move {
+        Some((folder_id, uid)) => (Some(folder_id), uid),
+        None => (row.folder_id, row.uid),
+    };
+    let Some(folder_id) = folder_id else {
         return Ok(None);
     };
     let folder = ctx
@@ -549,7 +560,7 @@ async fn resolve_remote(
         return Ok(None);
     };
     imap::select(session, &folder.path).await?;
-    let uid = match valid_uid(row.uid) {
+    let uid = match valid_uid(known_uid) {
         Some(uid) => Some(uid),
         None => uid_by_message_id(session, row.message_id.as_deref()).await?,
     };
@@ -641,7 +652,7 @@ async fn flag_action(
     let Some(message_id) = action.message_id else {
         return Ok(());
     };
-    match resolve_remote(ctx, session, message_id).await? {
+    match resolve_remote(ctx, session, action, message_id).await? {
         Some((_folder, uid)) => imap::store_flag(session, uid, flag, add).await,
         None => Ok(()), // deleted remotely; flag intent is moot
     }
@@ -660,7 +671,7 @@ async fn keyword_action(
     let Some(keyword) = action.payload["keyword"].as_str() else {
         return Ok(());
     };
-    match resolve_remote(ctx, session, message_id).await? {
+    match resolve_remote(ctx, session, action, message_id).await? {
         Some((_folder, uid)) => imap::store_flag(session, uid, keyword, add).await,
         None => Ok(()), // deleted remotely; label intent is moot
     }

crates/flectar-mail-core/src/queue/move_tests.rs

@@ -272,3 +272,59 @@ async fn undo_and_star_find_a_moved_message_before_its_mailbox_syncs() {
         .unwrap();
     assert_eq!(row.folder_id, Some(1));
 }
+
+/// A star given just before an archive, with both still waiting, reaches the
+/// message where the server still has it: the archive's local move must not
+/// send the star to the Archive mailbox, where the message isn't yet.
+#[tokio::test]
+async fn a_star_waiting_before_an_archive_reaches_the_message() {
+    let temp = tempfile::tempdir().unwrap();
+    let credentials = Arc::new(DevelopmentFileCredentialStore::new(
+        temp.path().join("credentials.json"),
+    ));
+    let core = Core::start_mail_ui_with_credentials(Paths::for_tests(temp.path()), credentials)
+        .await
+        .unwrap();
+    let (config, thread_id) = core
+        .db
+        .write(|conn| {
+            testutil::seed_account(conn);
+            conn.execute_batch(
+                "UPDATE folders SET uidvalidity = 7;
+                 INSERT INTO folders (id, account_id, path, role, uidvalidity)
+                   VALUES (2, 1, 'Archive', 'archive', 7);",
+            )?;
+            let (thread_id, message_id) =
+                testutil::seed_message(conn, "alice@test.dev", "Plans", false);
+            conn.execute(
+                "UPDATE messages SET uid = 5, message_id = 'plans@test.dev' WHERE id = ?1",
+                [message_id],
+            )?;
+            Ok((repo::accounts::get_config(conn, 1)?.unwrap(), thread_id))
+        })
+        .await
+        .unwrap();
+    let server = Server::start(BTreeMap::from([
+        ("INBOX".into(), (6, vec![stored(5, "plans@test.dev", &[])])),
+        ("Archive".into(), (10, Vec::new())),
+    ]))
+    .await;
+    let mut session = server.connect().await;
+    let ctx = core.sync_ctx();
+    let perform = |kind| {
+        core.perform_action(PerformActionArgs {
+            kind,
+            target: ActionTarget::Threads(vec![thread_id]),
+            params: None,
+        })
+    };
+
+    perform(ActionKind::Star).await.unwrap();
+    perform(ActionKind::Archive).await.unwrap();
+    execute_due(&ctx, &config, &mut session).await.unwrap();
+    assert!(server.mailbox("INBOX").is_empty());
+    assert_eq!(
+        server.mailbox("Archive"),
+        [stored(10, "plans@test.dev", &["\\Flagged"])]
+    );
+}