feat(rpc): add rowid-forward message pagination

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.14.0 - Unreleased
+
+### JSON-RPC
+- feat: add bounded `messages.after` pagination with authoritative database-instance-scoped ROWID cursors, cross-chat catchup, and optional standalone reaction events (#200, #201, thanks @vincentkoc).
+
 ## 0.13.5 - Unreleased
 
 ### JSON-RPC

README.md

@@ -205,7 +205,7 @@ specific outgoing Apple ID phone number or inline reply target.
 It is intended for agents and long-running integrations that want a single
 process for chats, history, send, and watch.
 
-Read methods: `chats.list`, `messages.history`, `messages.stats`, `messages.scheduled`, `watch.subscribe`,
+Read methods: `chats.list`, `messages.history`, `messages.after`, `messages.stats`, `messages.scheduled`, `watch.subscribe`,
 `watch.unsubscribe`, `message.send_status`. Mutating: `send`, `poll.send`.
 Bridge introspection: `handles.check`. See [docs/rpc.md](docs/rpc.md) for
 request and response shapes.

Sources/IMsgCore/MessageStore+Messages.swift

@@ -121,12 +121,90 @@ extension MessageStore {
     }
   }
 
+  public func messagesAfterPage(
+    afterRowID: Int64,
+    chatID: Int64?,
+    limit: Int,
+    includeReactions: Bool = false
+  ) throws -> MessagesAfterPage {
+    guard limit > 0 else {
+      return MessagesAfterPage(messages: [], nextRowID: afterRowID, hasMore: false)
+    }
+
+    return try withConnection { db in
+      var physicalLimit = limit == Int.max ? limit : limit + 1
+
+      while true {
+        let query = MessagesAfterQuery(
+          store: self,
+          afterRowID: MessageID(rawValue: afterRowID),
+          chatID: chatID.map { ChatID(rawValue: $0) },
+          limit: physicalLimit,
+          includeReactions: includeReactions
+        )
+        var physicalMessages: [Message] = []
+        var parentCache: ReplyParentCache = [:]
+        var pollOptionCache = PollOptionTextCache()
+        let rows = try db.prepareRowIterator(query.sql, bindings: query.bindings)
+        while let row = try rows.failableNext() {
+          let decoded = try decodeMessageRow(
+            row,
+            columns: query.selection.columns,
+            fallbackChatID: query.fallbackChatID
+          )
+          physicalMessages.append(
+            try message(
+              from: decoded,
+              db,
+              parentCache: &parentCache,
+              pollOptionCache: &pollOptionCache
+            ))
+        }
+
+        let visibleMessages = try pageVisibleMessages(physicalMessages, db: db)
+        if visibleMessages.count > limit {
+          let overflowRowID = visibleMessages[limit].rowID
+          let consumed = physicalMessages.prefix { $0.rowID < overflowRowID }
+          let pageMessages = try pageVisibleMessages(Array(consumed), db: db)
+          let nextRowID = consumed.last?.rowID ?? afterRowID
+          return MessagesAfterPage(
+            messages: try enrichMessagesWithTrailingURLPreviews(
+              pageMessages,
+              afterRowID: nextRowID,
+              db: db
+            ),
+            nextRowID: nextRowID,
+            hasMore: true
+          )
+        }
+        if physicalMessages.count < physicalLimit || physicalLimit == Int.max {
+          return MessagesAfterPage(
+            messages: visibleMessages,
+            nextRowID: physicalMessages.last?.rowID ?? afterRowID,
+            hasMore: false
+          )
+        }
+        guard let nextLimit = nextHistoryPhysicalLimit(after: physicalLimit) else {
+          return MessagesAfterPage(
+            messages: visibleMessages,
+            nextRowID: physicalMessages.last?.rowID ?? afterRowID,
+            hasMore: false
+          )
+        }
+        physicalLimit = nextLimit
+      }
+    }
+  }
+
   func messagesAfterBatch(
     afterRowID: Int64,
     chatID: Int64?,
     limit: Int,
     includeReactions: Bool
   ) throws -> MessagesAfterBatch {
+    guard limit > 0 else {
+      return MessagesAfterBatch(messages: [], maxScannedRowID: afterRowID)
+    }
     let query = MessagesAfterQuery(
       store: self,
       afterRowID: MessageID(rawValue: afterRowID),

Sources/IMsgCore/MessageStore+URLPreviews.swift

@@ -1,4 +1,5 @@
 import Foundation
+import SQLite
 
 enum URLPreviewCoalescingFallback {
   case suppress
@@ -96,6 +97,127 @@ extension MessageStore {
     message.balloonBundleID == MessageStore.urlPreviewBalloonBundleID
   }
 
+  func pageVisibleMessages(_ messages: [Message], db: Connection) throws -> [Message] {
+    try coalesceURLPreviewMessages(
+      messages,
+      validateExistingCoalescence: { text, preview in
+        try self.precedingTextMessageForURLPreview(preview, db: db)?.rowID == text.rowID
+      },
+      fallbackForUnmatchedPreview: { preview in
+        guard try self.precedingTextMessageForURLPreview(preview, db: db) != nil else {
+          return nil
+        }
+        return .suppress
+      }
+    )
+  }
+
+  func enrichMessagesWithTrailingURLPreviews(
+    _ messages: [Message],
+    afterRowID: Int64,
+    db: Connection
+  ) throws -> [Message] {
+    guard schema.hasBalloonBundleIDColumn, !messages.isEmpty else { return messages }
+
+    var enriched = messages
+    let indexByRowID = Dictionary(
+      uniqueKeysWithValues: messages.enumerated().map { ($0.element.rowID, $0.offset) }
+    )
+    var lastBaseByChatID: [Int64: Message] = [:]
+    for message in messages where !isURLPreviewBalloon(message) && !message.isReaction {
+      lastBaseByChatID[message.chatID] = message
+    }
+    let pageBases = lastBaseByChatID.values.sorted { $0.rowID < $1.rowID }
+    guard !pageBases.isEmpty else { return messages }
+
+    let reactionFilter =
+      schema.hasReactionColumns
+      ? """
+       AND (
+         next.associated_message_type IS NULL
+         OR next.associated_message_type < 2000
+         OR next.associated_message_type > 3006
+       )
+      """
+      : ""
+    let selection = MessageRowSelection(store: self, includeChatID: true)
+
+    // Keep each VALUES block below SQLite's historical 999-variable limit.
+    for start in stride(from: 0, to: pageBases.count, by: 400) {
+      let end = min(start + 400, pageBases.count)
+      let batch = pageBases[start..<end]
+      let values = Array(repeating: "(?, ?)", count: batch.count).joined(separator: ", ")
+      let sql = """
+        WITH page_base(parent_rowid, chat_id) AS (VALUES \(values)),
+        preview_window AS (
+          SELECT page_base.*,
+                 (
+                   SELECT next.ROWID
+                   FROM message next
+                   JOIN chat_message_join next_cmj ON next.ROWID = next_cmj.message_id
+                   WHERE next.ROWID > ?
+                     AND next_cmj.chat_id = page_base.chat_id
+                     AND COALESCE(next.balloon_bundle_id, '') <> ?
+                     \(reactionFilter)
+                   ORDER BY next_cmj.message_id ASC
+                   LIMIT 1
+                 ) AS boundary_rowid
+          FROM page_base
+        )
+        SELECT \(selection.selectList),
+               preview_window.parent_rowid AS preview_parent_rowid
+        FROM preview_window
+        JOIN chat_message_join cmj ON cmj.chat_id = preview_window.chat_id
+        JOIN message m ON m.ROWID = cmj.message_id
+        LEFT JOIN handle h ON m.handle_id = h.ROWID
+        WHERE m.ROWID > ?
+          AND (preview_window.boundary_rowid IS NULL OR m.ROWID < preview_window.boundary_rowid)
+          AND m.balloon_bundle_id = ?
+        ORDER BY m.ROWID ASC
+        """
+      var bindings: [Binding?] = []
+      for message in batch {
+        bindings.append(message.rowID)
+        bindings.append(message.chatID)
+      }
+      bindings.append(afterRowID)
+      bindings.append(MessageStore.urlPreviewBalloonBundleID)
+      bindings.append(afterRowID)
+      bindings.append(MessageStore.urlPreviewBalloonBundleID)
+
+      var parentCache: ReplyParentCache = [:]
+      var pollOptionCache = PollOptionTextCache()
+      let rows = try db.prepareRowIterator(sql, bindings: bindings)
+      while let row = try rows.failableNext() {
+        let parentRowID = try int64Value(row, "preview_parent_rowid")
+        guard
+          let parentRowID,
+          let index = indexByRowID[parentRowID]
+        else {
+          continue
+        }
+        let decoded = try decodeMessageRow(
+          row,
+          columns: selection.columns,
+          fallbackChatID: enriched[index].chatID
+        )
+        let preview = try message(
+          from: decoded,
+          db,
+          parentCache: &parentCache,
+          pollOptionCache: &pollOptionCache
+        )
+        guard
+          try precedingTextMessageForURLPreview(preview, db: db)?.rowID == parentRowID
+        else {
+          continue
+        }
+        enriched[index] = enriched[index].withURLPreview(urlPreviewMetadata(from: preview))
+      }
+    }
+    return enriched
+  }
+
   private func previousMessageInSameChat(
     _ chronological: [(offset: Int, element: Message)],
     before position: Int,

Sources/IMsgCore/MessagesAfterPage.swift

@@ -0,0 +1,13 @@
+public struct MessagesAfterPage: Sendable, Equatable {
+  public let messages: [Message]
+  /// Physical scan cursor scoped to the same Messages database instance.
+  /// Discard it after that database is replaced, restored, or recreated.
+  public let nextRowID: Int64
+  public let hasMore: Bool
+
+  public init(messages: [Message], nextRowID: Int64, hasMore: Bool) {
+    self.messages = messages
+    self.nextRowID = nextRowID
+    self.hasMore = hasMore
+  }
+}

Sources/imsg/RPCServer+MessagesAfterHandlers.swift

@@ -0,0 +1,107 @@
+import CoreFoundation
+import Foundation
+import IMsgCore
+
+extension RPCServer {
+  func handleMessagesAfter(id: Any?, params: [String: Any]) async throws {
+    let supportedParams: Set<String> = [
+      "since_rowid",
+      "chat_id",
+      "limit",
+      "attachments",
+      "convert_attachments",
+      "include_reactions",
+    ]
+    if let unknown = params.keys.filter({ !supportedParams.contains($0) }).sorted().first {
+      throw RPCError.invalidParams("unknown messages.after param: \(unknown)")
+    }
+
+    guard let sinceRowID = strictMessagesAfterInt64(params["since_rowid"]), sinceRowID >= 0 else {
+      throw RPCError.invalidParams("since_rowid must be a non-negative integer")
+    }
+
+    let chatID: Int64?
+    if let rawChatID = params["chat_id"] {
+      guard let parsed = strictMessagesAfterInt64(rawChatID), parsed > 0 else {
+        throw RPCError.invalidParams("chat_id must be a positive integer")
+      }
+      chatID = parsed
+    } else {
+      chatID = nil
+    }
+
+    let limit: Int
+    if let rawLimit = params["limit"] {
+      guard let parsed = strictMessagesAfterInt(rawLimit), (1...500).contains(parsed) else {
+        throw RPCError.invalidParams("limit must be an integer between 1 and 500")
+      }
+      limit = parsed
+    } else {
+      limit = 100
+    }
+
+    let includeAttachments = try strictMessagesAfterBool(
+      params["attachments"],
+      name: "attachments"
+    )
+    let attachmentOptions = AttachmentQueryOptions(
+      convertUnsupported: try strictMessagesAfterBool(
+        params["convert_attachments"],
+        name: "convert_attachments"
+      ))
+    let page = try store.messagesAfterPage(
+      afterRowID: sinceRowID,
+      chatID: chatID,
+      limit: limit,
+      includeReactions: try strictMessagesAfterBool(
+        params["include_reactions"],
+        name: "include_reactions"
+      )
+    )
+    let reactionsByMessageID = try store.reactions(for: page.messages)
+    var payloads: [[String: Any]] = []
+    payloads.reserveCapacity(page.messages.count)
+    for message in page.messages {
+      payloads.append(
+        try await buildMessagePayload(
+          store: store,
+          cache: cache,
+          message: message,
+          includeAttachments: includeAttachments,
+          includeReactions: true,
+          prefetchedReactions: reactionsByMessageID[message.rowID] ?? [],
+          attachmentOptions: attachmentOptions,
+          contactResolver: contactResolver
+        ))
+    }
+
+    respond(
+      id: id,
+      result: [
+        "messages": payloads,
+        "next_rowid": page.nextRowID,
+        "has_more": page.hasMore,
+      ]
+    )
+  }
+}
+
+private func strictMessagesAfterInt64(_ value: Any?) -> Int64? {
+  guard let number = value as? NSNumber else { return nil }
+  guard CFGetTypeID(number) != CFBooleanGetTypeID() else { return nil }
+  return Int64(number.stringValue)
+}
+
+private func strictMessagesAfterInt(_ value: Any?) -> Int? {
+  guard let number = value as? NSNumber else { return nil }
+  guard CFGetTypeID(number) != CFBooleanGetTypeID() else { return nil }
+  return Int(number.stringValue)
+}
+
+private func strictMessagesAfterBool(_ value: Any?, name: String) throws -> Bool {
+  guard let value else { return false }
+  guard let number = value as? NSNumber, CFGetTypeID(number) == CFBooleanGetTypeID() else {
+    throw RPCError.invalidParams("\(name) must be a boolean")
+  }
+  return number.boolValue
+}

Sources/imsg/RPCServer.swift

@@ -34,6 +34,7 @@ let kSupportedRPCMethods: [String] = [
   "chats.markUnread",
   "messages.stats",
   "messages.history",
+  "messages.after",
   "watch.subscribe",
   "watch.unsubscribe",
   "send",
@@ -168,6 +169,11 @@ final class RPCServer {
         try await handleMessagesStats(id: id, params: params)
       case "messages.history":
         try await handleMessagesHistory(id: id, params: params)
+      case "messages.after":
+        guard request.paramsAreNamed else {
+          throw RPCError.invalidParams("messages.after params must be an object")
+        }
+        try await handleMessagesAfter(id: id, params: params)
       case "watch.subscribe":
         try await handleWatchSubscribe(id: id, params: params)
       case "watch.unsubscribe":