cmux-imessage: review follow-ups (date-ordered previews, NULL rows, stderr drain, terminal error code)

cmux-tui/crates/cmux-imessage/src/reader.rs

@@ -19,9 +19,12 @@ const ATTACHMENT_TABLES: [&str; 2] = ["attachment", "message_attachment_join"];
 /// The most rows one call returns, whatever the caller asks for.
 pub const MAX_LIMIT: usize = 500;
 
-/// A message the conversation view shows: not a tapback (`associated_message_type` 2000-3005),
-/// and not a group event such as a rename or a member change (`item_type` > 0).
-const VISIBLE: &str = "m.associated_message_type = 0 AND m.item_type = 0";
+/// A message the conversation view shows: not a tapback or other associated message
+/// (`associated_message_type` 2000-2007, removals 3000-3007), and not a group event such as a
+/// rename or a member change (`item_type` > 0). Old rows may hold NULL in either column; NULL
+/// means a plain message.
+const VISIBLE: &str =
+    "IFNULL(m.associated_message_type, 0) = 0 AND IFNULL(m.item_type, 0) = 0";
 
 /// `~/Library/Messages/chat.db` for the current user, when `HOME` is set.
 pub fn default_database_path() -> Option<PathBuf> {
@@ -64,16 +67,19 @@ impl Reader {
     pub fn threads(&self, limit: usize) -> Result<Vec<Thread>, MirrorError> {
         let limit = clamp(limit);
         let sql = format!(
-            "WITH last AS (
-                 SELECT cmj.chat_id AS chat_id, MAX(cmj.message_id) AS last_id
+            // The newest visible message per chat by date (rowid breaks ties): iCloud backfill
+            // gives old messages high rowids, so the highest rowid is not the newest message.
+            "WITH ranked AS (
+                 SELECT cmj.chat_id AS chat_id, m.ROWID AS last_id, IFNULL(m.date, 0) AS last_date,
+                        ROW_NUMBER() OVER (
+                            PARTITION BY cmj.chat_id ORDER BY IFNULL(m.date, 0) DESC, m.ROWID DESC
+                        ) AS rank
                  FROM chat_message_join cmj JOIN message m ON m.ROWID = cmj.message_id
-                 WHERE {VISIBLE}
-                 GROUP BY cmj.chat_id)
+                 WHERE {VISIBLE})
              SELECT c.ROWID, c.guid, c.display_name, c.style, c.service_name, c.chat_identifier,
-                    last.last_id
-             FROM chat c JOIN last ON last.chat_id = c.ROWID
-             JOIN message m ON m.ROWID = last.last_id
-             ORDER BY m.date DESC, c.ROWID DESC
+                    ranked.last_id
+             FROM chat c JOIN ranked ON ranked.chat_id = c.ROWID AND ranked.rank = 1
+             ORDER BY ranked.last_date DESC, c.ROWID DESC
              LIMIT ?1"
         );
         let mut statement = self.connection.prepare(&sql)?;
@@ -142,7 +148,9 @@ impl Reader {
         let sql = format!(
             "SELECT m.ROWID, m.guid, cmj.chat_id, h.id, m.is_from_me, m.date, m.text,
                     m.attributedBody
-             FROM message m JOIN chat_message_join cmj ON cmj.message_id = m.ROWID
+             FROM message m
+             JOIN (SELECT message_id, MIN(chat_id) AS chat_id FROM chat_message_join
+                   WHERE message_id > ?1 GROUP BY message_id) cmj ON cmj.message_id = m.ROWID
              LEFT JOIN handle h ON h.ROWID = m.handle_id
              WHERE m.ROWID > ?1 AND {VISIBLE}
              ORDER BY m.ROWID ASC

cmux-tui/crates/cmux-imessage/src/send.rs

@@ -118,20 +118,30 @@ impl CommandRunner for SystemRunner {
             .stdout(Stdio::null())
             .stderr(Stdio::piped())
             .spawn()?;
+        // Drain stderr while the child runs: a child that fills the pipe buffer would otherwise
+        // block before it exits. The reader thread ends when the pipe closes (the child exits
+        // or is killed below).
+        let reader = child.stderr.take().map(|mut pipe| {
+            std::thread::spawn(move || {
+                use std::io::Read;
+                let mut bytes = Vec::new();
+                let _ = pipe.read_to_end(&mut bytes);
+                String::from_utf8_lossy(&bytes).into_owned()
+            })
+        });
+        let collect = |reader: Option<std::thread::JoinHandle<String>>| {
+            reader.and_then(|handle| handle.join().ok()).unwrap_or_default()
+        };
         let deadline = Instant::now() + SEND_TIMEOUT;
         loop {
             if let Some(status) = child.try_wait()? {
-                let mut stderr = String::new();
-                if let Some(mut pipe) = child.stderr.take() {
-                    use std::io::Read;
-                    let _ = pipe.read_to_string(&mut stderr);
-                }
-                return Ok(CommandOutput { success: status.success(), stderr });
+                return Ok(CommandOutput { success: status.success(), stderr: collect(reader) });
             }
             if Instant::now() >= deadline {
                 // Our own child: stop it and reap it so nothing is left behind.
                 let _ = child.kill();
                 let _ = child.wait();
+                let _ = collect(reader);
                 return Err(std::io::Error::new(ErrorKind::TimedOut, "osascript timed out"));
             }
             std::thread::sleep(Duration::from_millis(50));
@@ -182,15 +192,27 @@ pub fn send(request: &SendRequest, runner: &dyn CommandRunner) -> Result<(), Sen
         return Ok(());
     }
     let detail = output.stderr.trim().to_string();
-    Err(if detail.contains(ERR_NOT_AUTHORIZED) {
-        SendError::NotPermitted(detail)
-    } else if detail.contains(ERR_NO_SUCH_OBJECT) {
-        SendError::InvalidChat(detail)
-    } else {
-        SendError::Failed(detail)
+    let code = terminal_error_code(&detail).map(str::to_owned);
+    Err(match code.as_deref() {
+        Some(ERR_NOT_AUTHORIZED) => SendError::NotPermitted(detail),
+        Some(ERR_NO_SUCH_OBJECT) => SendError::InvalidChat(detail),
+        _ => SendError::Failed(detail),
     })
 }
 
+/// The AppleScript error code osascript ends its diagnostic with (`... (-1743)`), so a code
+/// quoted earlier in the message (inside a chat id, say) never decides the mapping.
+fn terminal_error_code(detail: &str) -> Option<&str> {
+    let trimmed = detail.trim_end();
+    let inner = trimmed.strip_suffix(')')?;
+    let open = inner.rfind('(')?;
+    let digits = inner.get(open..)?.strip_prefix("(-")?;
+    if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
+        return None;
+    }
+    trimmed.get(open..=inner.len())
+}
+
 /// `service;-;handle` (1:1) or `service;+;room` (group), with no control characters, and not
 /// starting with `-` (so `osascript` never reads it as an option).
 fn is_chat_guid(guid: &str) -> bool {

cmux-tui/crates/cmux-imessage/tests/reader.rs

@@ -357,3 +357,44 @@ fn messages_serialize_with_snake_case_keys() {
     assert_eq!(wire[0]["text"], "hey");
     assert_eq!(wire[0]["is_from_me"], false);
 }
+
+#[test]
+fn thread_preview_is_the_newest_by_date_not_by_rowid() {
+    let fx = Fixture::new();
+    let alice = fx.handle("+15550001111");
+    let chat = fx.chat("iMessage;-;+15550001111", 45, None, &[alice]);
+    fx.message(chat, alice, false, 10, Some("newest"));
+    // iCloud backfill: an older message arrives later and gets a higher rowid.
+    fx.message(chat, alice, false, 1, Some("backfilled"));
+    let threads = fx.reader().threads(10).unwrap();
+    assert_eq!(threads[0].last_message.as_ref().unwrap().text.as_deref(), Some("newest"));
+}
+
+#[test]
+fn null_item_and_association_columns_count_as_visible() {
+    let fx = Fixture::new();
+    let alice = fx.handle("+15550001111");
+    let chat = fx.chat("iMessage;-;+15550001111", 45, None, &[alice]);
+    let rowid = fx.message(chat, alice, false, 1, Some("old row"));
+    fx.db
+        .execute("UPDATE message SET item_type = NULL, associated_message_type = NULL WHERE ROWID = ?1", params![rowid])
+        .unwrap();
+    let reader = fx.reader();
+    assert_eq!(reader.messages(chat, None, 10).unwrap().len(), 1);
+    assert_eq!(reader.threads(10).unwrap().len(), 1);
+    assert_eq!(reader.since(0, 10).unwrap().len(), 1);
+}
+
+#[test]
+fn since_returns_a_message_once_even_when_joined_to_two_chats() {
+    let fx = Fixture::new();
+    let alice = fx.handle("+15550001111");
+    let a = fx.chat("iMessage;-;+15550001111", 45, None, &[alice]);
+    let b = fx.chat("SMS;-;+15550001111", 45, None, &[alice]);
+    let rowid = fx.message(a, alice, false, 1, Some("both"));
+    fx.db
+        .execute("INSERT INTO chat_message_join (chat_id, message_id) VALUES (?1, ?2)", params![b, rowid])
+        .unwrap();
+    let fresh = fx.reader().since(0, 10).unwrap();
+    assert_eq!(fresh.iter().map(|m| m.rowid).collect::<Vec<_>>(), vec![rowid]);
+}

cmux-tui/crates/cmux-imessage/tests/send.rs

@@ -110,3 +110,30 @@ fn osascript_failures_map_to_codes() {
     let slow = FakeRunner::with(Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "slow")));
     assert_eq!(send(&request("iMessage;-;+1", "hi"), &slow), Err(SendError::TimedOut));
 }
+
+#[test]
+fn the_terminal_error_code_decides_not_codes_quoted_earlier() {
+    let quoted = "execution error: Messages got an error: Can’t get chat id \"iMessage;-;(-1743)\". (-1728)";
+    assert_eq!(
+        send(&request("iMessage;-;(-1743)", "hi"), &FakeRunner::failing(quoted)).map_err(|e| e.code()),
+        Err("invalid_chat")
+    );
+    let denied_then_newline = "execution error: Not authorized to send Apple events to Messages. (-1743)\n";
+    assert_eq!(
+        send(&request("iMessage;-;+1", "hi"), &FakeRunner::failing(denied_then_newline)).map_err(|e| e.code()),
+        Err("not_permitted")
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn system_runner_reads_stderr_past_the_pipe_buffer_without_stalling() {
+    use cmux_imessage::send::SystemRunner;
+    let started = std::time::Instant::now();
+    let args =
+        vec!["-c".to_string(), "head -c 300000 /dev/zero | tr '\\0' x >&2; exit 3".to_string()];
+    let output = SystemRunner.run("sh", &args).expect("sh runs");
+    assert!(!output.success);
+    assert_eq!(output.stderr.len(), 300_000);
+    assert!(started.elapsed() < std::time::Duration::from_secs(30), "took {:?}", started.elapsed());
+}