fix: coalesce GUID-linked URL previews

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 0.13.2 - Unreleased
 
+### Read Commands
+- fix: coalesce GUID-linked Apple URL previews even when the originating text row omits the URL, preserving one logical message across history, search, stats, unread counts, and at-least-once live watch with a no-skip physical `cursor` (#181, thanks @omarshahine).
+
 ## 0.13.1 - 2026-07-17
 
 ### Highlights

README.md

@@ -233,8 +233,18 @@ Before handing the file to Messages, `imsg` stages it under
 ## Watch Behavior
 
 `imsg watch` starts at the newest message by default and streams messages
-written after it starts. Use `--since-rowid <id>` to resume from a stored
-cursor.
+written after it starts. Use `--since-rowid <cursor>` to resume from a stored
+`cursor` field. The cursor normally matches `id`, but advances past a folded
+URL-preview row only after applicable interleaved rows have been emitted, so
+resuming does not skip part of the stream. Until that frontier is complete, an
+interleaved event can repeat the prior cursor and may be replayed after a
+reconnect; deduplicate at-least-once delivery by stable `chat_id` plus `id` or
+`guid`.
+
+New live text rows are held for up to two seconds so a delayed, GUID-linked
+Apple URL preview can be folded into the same logical event. Backlog and resume
+reads are not delayed; when both physical rows already exist, they are
+coalesced immediately.
 
 The watcher listens for filesystem events on `chat.db`, `chat.db-wal`,
 `chat.db-shm`, and the containing Messages directory, then backs that up with

Sources/IMsgCore/Message+URLPreview.swift

@@ -1,6 +1,10 @@
 import Foundation
 
 extension Message {
+  var physicalCompletionRowID: Int64 {
+    max(rowID, urlPreview?.rowID ?? rowID)
+  }
+
   public struct URLPreviewMetadata: Sendable, Equatable {
     public let rowID: Int64
     public let guid: String
@@ -15,12 +19,14 @@ extension Message {
     }
   }
 
-  public func withURLPreview(_ preview: URLPreviewMetadata) -> Message {
+  public func withURLPreview(_ preview: URLPreviewMetadata, text replacementText: String? = nil)
+    -> Message
+  {
     Message(
       rowID: rowID,
       chatID: chatID,
       sender: sender,
-      text: text,
+      text: replacementText ?? text,
       date: date,
       isFromMe: isFromMe,
       service: service,
@@ -37,6 +43,42 @@ extension Message {
       ),
       balloonBundleID: balloonBundleID,
       urlPreview: preview,
+      cursorRowID: cursorRowID,
+      reaction: ReactionMetadata(
+        isReaction: isReaction,
+        reactionType: reactionType,
+        isReactionAdd: isReactionAdd,
+        reactedToGUID: reactedToGUID
+      ),
+      poll: poll,
+      isRead: isRead,
+      dateRead: dateRead
+    )
+  }
+
+  func withCursorRowID(_ cursorRowID: Int64) -> Message {
+    Message(
+      rowID: rowID,
+      chatID: chatID,
+      sender: sender,
+      text: text,
+      date: date,
+      isFromMe: isFromMe,
+      service: service,
+      handleID: handleID,
+      attachmentsCount: attachmentsCount,
+      guid: guid,
+      routing: RoutingMetadata(
+        replyToGUID: replyToGUID,
+        threadOriginatorGUID: threadOriginatorGUID,
+        threadOriginatorPart: threadOriginatorPart,
+        destinationCallerID: destinationCallerID,
+        replyToText: replyToText,
+        replyToSender: replyToSender
+      ),
+      balloonBundleID: balloonBundleID,
+      urlPreview: urlPreview,
+      cursorRowID: cursorRowID,
       reaction: ReactionMetadata(
         isReaction: isReaction,
         reactionType: reactionType,

Sources/IMsgCore/MessageStatsQueries.swift

@@ -110,6 +110,18 @@ struct StatsPreviewQuery {
       ? "previous.balloon_bundle_id" : "NULL"
     let previewBody = schema.hasAttributedBody ? "preview.attributedBody" : "NULL"
     let previousBody = schema.hasAttributedBody ? "previous.attributedBody" : "NULL"
+    let previewGUID = schema.hasGUIDColumn ? "preview.guid" : "NULL"
+    let previousGUID = schema.hasGUIDColumn ? "previous.guid" : "NULL"
+    let previewReplyToGUID = schema.hasReplyToGUIDColumn ? "preview.reply_to_guid" : "NULL"
+    let previousReplyToGUID = schema.hasReplyToGUIDColumn ? "previous.reply_to_guid" : "NULL"
+    let previewAssociatedGUID =
+      schema.hasReactionColumns ? "preview.associated_message_guid" : "NULL"
+    let previousAssociatedGUID =
+      schema.hasReactionColumns ? "previous.associated_message_guid" : "NULL"
+    let previewAssociatedType =
+      schema.hasReactionColumns ? "preview.associated_message_type" : "NULL"
+    let previousAssociatedType =
+      schema.hasReactionColumns ? "previous.associated_message_type" : "NULL"
     self.sql = """
       WITH base AS (
         SELECT cmj.chat_id, m.ROWID AS message_id
@@ -124,6 +136,10 @@ struct StatsPreviewQuery {
       )
       SELECT s.chat_id AS chat_id,
              preview.ROWID AS preview_rowid,
+             IFNULL(\(previewGUID), '') AS preview_guid,
+             IFNULL(\(previewReplyToGUID), '') AS preview_reply_to_guid,
+             IFNULL(\(previewAssociatedGUID), '') AS preview_associated_guid,
+             \(previewAssociatedType) AS preview_associated_type,
              preview.handle_id AS preview_handle_id,
              IFNULL(preview_handle.id, '') AS preview_sender,
              IFNULL(preview.text, '') AS preview_text,
@@ -134,6 +150,10 @@ struct StatsPreviewQuery {
              IFNULL(\(previewDestination), '') AS preview_destination_caller_id,
              IFNULL(\(previewBalloon), '') AS preview_balloon_bundle_id,
              previous.ROWID AS previous_rowid,
+             IFNULL(\(previousGUID), '') AS previous_guid,
+             IFNULL(\(previousReplyToGUID), '') AS previous_reply_to_guid,
+             IFNULL(\(previousAssociatedGUID), '') AS previous_associated_guid,
+             \(previousAssociatedType) AS previous_associated_type,
              previous.handle_id AS previous_handle_id,
              IFNULL(previous_handle.id, '') AS previous_sender,
              IFNULL(previous.text, '') AS previous_text,

Sources/IMsgCore/MessageStore+Chats.swift

@@ -263,6 +263,9 @@ extension MessageStore {
           handleID: decoded.handleID,
           attachmentsCount: decoded.attachments,
           guid: decoded.guid,
+          routing: Message.RoutingMetadata(
+            replyToGUID: routedReplyToGUID(decoded)
+          ),
           balloonBundleID: decoded.balloonBundleID.nilIfEmpty,
           reaction: Message.ReactionMetadata(
             isReaction: reaction.isReaction,

Sources/IMsgCore/MessageStore+MessageConstruction.swift

@@ -113,4 +113,84 @@ extension MessageStore {
     }
     return nil
   }
+
+  func linkedURLPreviewLookahead(
+    afterRowID: Int64,
+    candidates: [Message],
+    db: Connection
+  ) throws -> [Message] {
+    guard schema.hasBalloonBundleIDColumn else { return [] }
+
+    let textCandidates = candidates.filter { !isURLPreviewBalloon($0) }
+    guard !textCandidates.isEmpty else { return [] }
+
+    let candidateChatIDsByRowID = Dictionary(grouping: textCandidates, by: \.rowID)
+      .mapValues { Set($0.map(\.chatID)) }
+    let selection = MessageRowSelection(store: self, includeChatID: true)
+    var previews: [Message] = []
+    var seenPreviewChatIDsByRowID: [Int64: Set<Int64>] = [:]
+    var parentCache: ReplyParentCache = [:]
+    var pollOptionCache = PollOptionTextCache()
+    let maximumDateWindowsPerQuery = 200
+
+    for startIndex in stride(
+      from: 0,
+      to: textCandidates.count,
+      by: maximumDateWindowsPerQuery
+    ) {
+      let endIndex = min(startIndex + maximumDateWindowsPerQuery, textCandidates.count)
+      let chunk = textCandidates[startIndex..<endIndex]
+      let datePredicate = Array(
+        repeating: "(m.date >= ? AND m.date <= ?)",
+        count: chunk.count
+      ).joined(separator: " OR ")
+      let sql = """
+        SELECT \(selection.selectList)
+        FROM message m
+        LEFT JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
+        LEFT JOIN handle h ON m.handle_id = h.ROWID
+        WHERE m.ROWID > ?
+          AND m.balloon_bundle_id = ?
+          AND (\(datePredicate))
+        ORDER BY m.ROWID ASC
+        """
+      var bindings: [Binding?] = [
+        afterRowID,
+        MessageStore.urlPreviewBalloonBundleID,
+      ]
+      for candidate in chunk {
+        bindings.append(MessageStore.appleEpoch(candidate.date))
+        bindings.append(
+          MessageStore.appleEpoch(
+            candidate.date.addingTimeInterval(MessageStore.urlPreviewCoalescingWindow)
+          ))
+      }
+
+      let rows = try db.prepareRowIterator(sql, bindings: bindings)
+      while let row = try rows.failableNext() {
+        let decoded = try decodeMessageRow(
+          row,
+          columns: selection.columns,
+          fallbackChatID: nil
+        )
+        let preview = try message(
+          from: decoded,
+          db,
+          parentCache: &parentCache,
+          pollOptionCache: &pollOptionCache
+        )
+        guard
+          let preceding = try precedingTextMessageForURLPreview(preview, db: db),
+          candidateChatIDsByRowID[preceding.rowID]?.contains(preview.chatID) == true
+        else {
+          continue
+        }
+        guard
+          seenPreviewChatIDsByRowID[decoded.rowID, default: []].insert(preview.chatID).inserted
+        else { continue }
+        previews.append(preview)
+      }
+    }
+    return previews
+  }
 }

Sources/IMsgCore/MessageStore+MessageRows.swift

@@ -99,7 +99,7 @@ struct MessageRowSelection {
     let columns = MessageRowColumns.message(chatID: includeChatID ? "chat_id" : nil)
     let schema = store.schema
     let bodyColumn = schema.hasAttributedBody ? "m.attributedBody" : "NULL"
-    let guidColumn = schema.hasReactionColumns ? "m.guid" : "NULL"
+    let guidColumn = schema.hasGUIDColumn ? "m.guid" : "NULL"
     let associatedGuidColumn = schema.hasReactionColumns ? "m.associated_message_guid" : "NULL"
     let associatedTypeColumn = schema.hasReactionColumns ? "m.associated_message_type" : "NULL"
     let destinationCallerColumn =

Sources/IMsgCore/MessageStore+Messages.swift

@@ -9,6 +9,32 @@ extension MessageStore {
     }
   }
 
+  func maxRowID(chatID: Int64?, includeReactions: Bool) throws -> Int64 {
+    return try withConnection { db in
+      let reactionFilter =
+        includeReactions || !schema.hasReactionColumns
+        ? ""
+        : " AND (m.associated_message_type IS NULL OR m.associated_message_type < 2000 OR m.associated_message_type > 3006)"
+      let value: Binding?
+      if let chatID {
+        value = try db.scalar(
+          """
+          SELECT MAX(m.ROWID)
+          FROM message m
+          JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
+          WHERE cmj.chat_id = ?\(reactionFilter)
+          """,
+          chatID
+        )
+      } else {
+        value = try db.scalar(
+          "SELECT MAX(m.ROWID) FROM message m WHERE 1 = 1\(reactionFilter)"
+        )
+      }
+      return int64Value(value) ?? 0
+    }
+  }
+
   public func messages(chatID: Int64, limit: Int) throws -> [Message] {
     return try messages(chatID: chatID, limit: limit, filter: nil)
   }
@@ -112,7 +138,7 @@ extension MessageStore {
         includeReactions: includeReactions
       )
       if !batch.messages.isEmpty {
-        return batch.messages
+        return Array(batch.messages.prefix(limit))
       }
       guard batch.maxScannedRowID > cursor else {
         return []
@@ -125,7 +151,9 @@ extension MessageStore {
     afterRowID: Int64,
     chatID: Int64?,
     limit: Int,
-    includeReactions: Bool
+    includeReactions: Bool,
+    suppressLateURLPreviews: Bool = true,
+    deduplicateURLBalloons: Bool = true
   ) throws -> MessagesAfterBatch {
     let query = MessagesAfterQuery(
       store: self,
@@ -157,6 +185,37 @@ extension MessageStore {
             pollOptionCache: &pollOptionCache
           ))
       }
+      let previewLookahead = try linkedURLPreviewLookahead(
+        afterRowID: maxScannedRowID,
+        candidates: messages,
+        db: db
+      )
+      if let lookaheadMaxRowID = previewLookahead.map(\.rowID).max() {
+        let gapQuery = MessagesAfterQuery(
+          store: self,
+          afterRowID: MessageID(rawValue: maxScannedRowID),
+          chatID: chatID.map { ChatID(rawValue: $0) },
+          limit: Int.max,
+          includeReactions: includeReactions,
+          throughRowID: MessageID(rawValue: lookaheadMaxRowID)
+        )
+        let gapRows = try db.prepareRowIterator(gapQuery.sql, bindings: gapQuery.bindings)
+        while let row = try gapRows.failableNext() {
+          let decoded = try decodeMessageRow(
+            row,
+            columns: gapQuery.selection.columns,
+            fallbackChatID: gapQuery.fallbackChatID
+          )
+          maxScannedRowID = max(maxScannedRowID, decoded.rowID)
+          messages.append(
+            try message(
+              from: decoded,
+              db,
+              parentCache: &parentCache,
+              pollOptionCache: &pollOptionCache
+            ))
+        }
+      }
       let coalesced = try coalesceURLPreviewMessages(
         messages,
         validateExistingCoalescence: { text, preview in
@@ -166,10 +225,11 @@ extension MessageStore {
           guard try self.precedingTextMessageForURLPreview(preview, db: db) != nil else {
             return nil
           }
-          return .suppress
+          return suppressLateURLPreviews ? .suppress : nil
         }
       )
       let visibleMessages = coalesced.filter { message in
+        guard deduplicateURLBalloons else { return true }
         guard isURLPreviewBalloon(message) else { return true }
         return !shouldSkipURLBalloonDuplicate(
           chatID: message.chatID,
@@ -396,16 +456,28 @@ extension MessageStore {
   }
 
   func routedReplyToGUID(_ row: DecodedMessageRow) -> String? {
-    if let associatedType = row.associatedType, ReactionType.isReaction(associatedType) {
+    routedReplyToGUID(
+      databaseReplyToGUID: row.databaseReplyToGUID,
+      associatedGUID: row.associatedGUID,
+      associatedType: row.associatedType
+    )
+  }
+
+  func routedReplyToGUID(
+    databaseReplyToGUID: String,
+    associatedGUID: String,
+    associatedType: Int?
+  ) -> String? {
+    if let associatedType, ReactionType.isReaction(associatedType) {
       return nil
     }
-    let databaseReplyToGUID = normalizeAssociatedGUID(row.databaseReplyToGUID)
-    if !databaseReplyToGUID.isEmpty {
-      return databaseReplyToGUID
+    let normalizedDatabaseGUID = normalizeAssociatedGUID(databaseReplyToGUID)
+    if !normalizedDatabaseGUID.isEmpty {
+      return normalizedDatabaseGUID
     }
     return replyToGUID(
-      associatedGuid: row.associatedGUID,
-      associatedType: row.associatedType
+      associatedGuid: associatedGUID,
+      associatedType: associatedType
     )
   }
 }