ios: deliver notifications even if iOS fails to fire expiration notice, prevent repeat delivery of stale notifications (#5861)

* ios: deliver notification when iOS fails to fire expiration notice for NSE

* update core api

* update ui

* sha256map.nix

* do not enable background processes in maintenance mode

* fix ios

* fix parser

* ios: fix command

* compatible parser for connection ID

* log

* pass DB queue ID

* simplexmq

* query plans

* fix broadcast bot test

apps/ios/SimpleX NSE/NotificationService.swift

@@ -143,8 +143,7 @@ class NSEThreads {
 }
 
 struct ExpectedMessage {
-    var ntfConn: UserNtfConn
+    var ntfConn: NtfConn
-    var receiveConnId: String?
     var expectedMsgId: String?
     var allowedGetNextAttempts: Int
     var msgBestAttemptNtf: NSENotificationData?
@@ -152,6 +151,14 @@ struct ExpectedMessage {
     var shouldProcessNtf: Bool
     var startedProcessingNewMsgs: Bool
     var semaphore: DispatchSemaphore
+    
+    var connMsgReq: ConnMsgReq? {
+        if let expectedMsg_ = ntfConn.expectedMsg_ {
+            ConnMsgReq(msgConnId: ntfConn.agentConnId, msgDbQueueId: ntfConn.agentDbQueueId, msgTs: expectedMsg_.msgTs)
+        } else {
+            nil
+        }
+    }
 }
 
 // Notification service extension creates a new instance of the class and calls didReceive for each notification.
@@ -172,20 +179,27 @@ class NotificationService: UNNotificationServiceExtension {
 
     override func didReceive(_ request: UNNotificationRequest, withContentHandler contentHandler: @escaping (UNNotificationContent) -> Void) {
         logger.debug("DEBUGGING: NotificationService.didReceive")
-        let ntf = if let ntf_ = request.content.mutableCopy() as? UNMutableNotificationContent { ntf_ } else { UNMutableNotificationContent() }
+        let receivedNtf = if let ntf_ = request.content.mutableCopy() as? UNMutableNotificationContent { ntf_ } else { UNMutableNotificationContent() }
-        setServiceBestAttemptNtf(ntf)
+        setServiceBestAttemptNtf(receivedNtf)
         self.contentHandler = contentHandler
         registerGroupDefaults()
         let appState = appStateGroupDefault.get()
         logger.debug("NotificationService: app is \(appState.rawValue)")
         switch appState {
         case .stopped:
+//            Use this block to debug notificaitons delivery in CLI, with "ejected" database and stopped chat
+//            if let nrData = ntfRequestData(request) {
+//                logger.debug("NotificationService get notification connections: /_ntf conns \(nrData.nonce) \(nrData.encNtfInfo)")
+//                contentHandler(receivedNtf)
+//                return;
+//            }
             setBadgeCount()
-            setServiceBestAttemptNtf(createAppStoppedNtf(badgeCount))
+            contentHandler(createAppStoppedNtf(badgeCount))
-            deliverBestAttemptNtf()
         case .suspended:
-            receiveNtfMessages(request, contentHandler)
+            setExpirationTimer()
+            receiveNtfMessages(request)
         case .suspending:
+            setExpirationTimer()
             Task {
                 let state: AppState = await withCheckedContinuation { cont in
                     appSubscriber = appStateSubscriber { s in
@@ -206,42 +220,55 @@ class NotificationService: UNNotificationServiceExtension {
                     }
                 }
                 logger.debug("NotificationService: app state is now \(state.rawValue)")
-                if state.inactive {
+                if state.inactive && self.contentHandler != nil {
-                    receiveNtfMessages(request, contentHandler)
+                    receiveNtfMessages(request)
                 } else {
-                    deliverBestAttemptNtf()
+                    contentHandler(receivedNtf)
                 }
             }
-        case .active: contentHandler(UNMutableNotificationContent())
+        case .active: contentHandler(receivedNtf)
-        case .activating: contentHandler(UNMutableNotificationContent())
+        case .activating: contentHandler(receivedNtf)
-        case .bgRefresh: contentHandler(UNMutableNotificationContent())
+        case .bgRefresh: contentHandler(receivedNtf)
         }
     }
 
-    func receiveNtfMessages(_ request: UNNotificationRequest, _ contentHandler: @escaping (UNNotificationContent) -> Void) {
+    private func setExpirationTimer() -> Void {
+        DispatchQueue.main.asyncAfter(deadline: .now() + 30) {
+            self.deliverBestAttemptNtf(urgent: true)
+        }
+    }
+
+    private func ntfRequestData(_ request: UNNotificationRequest) -> (nonce: String, encNtfInfo: String)? {
+        if let ntfData = request.content.userInfo["notificationData"] as? [AnyHashable : Any],
+           let nonce = ntfData["nonce"] as? String,
+           let encNtfInfo = ntfData["message"] as? String {
+            (nonce, encNtfInfo)
+        } else {
+            nil
+        }
+    }
+    
+    func receiveNtfMessages(_ request: UNNotificationRequest) {
         logger.debug("NotificationService: receiveNtfMessages")
         if case .documents = dbContainerGroupDefault.get() {
             deliverBestAttemptNtf()
             return
         }
-        let userInfo = request.content.userInfo
+        if let nrData = ntfRequestData(request),
-        if let ntfData = userInfo["notificationData"] as? [AnyHashable : Any],
-           let nonce = ntfData["nonce"] as? String,
-           let encNtfInfo = ntfData["message"] as? String,
            // check it here again
            appStateGroupDefault.get().inactive {
             // thread is added to activeThreads tracking set here - if thread started chat it needs to be suspended
             if let t = threadId { NSEThreads.shared.startThread(t, self) }
             let dbStatus = startChat()
             if case .ok = dbStatus,
-               let ntfConns = apiGetNtfConns(nonce: nonce, encNtfInfo: encNtfInfo) {
+               let ntfConns = apiGetNtfConns(nonce: nrData.nonce, encNtfInfo: nrData.encNtfInfo) {
                 logger.debug("NotificationService: receiveNtfMessages: apiGetNtfConns ntfConns count = \(ntfConns.count)")
-
+//                logger.debug("NotificationService: receiveNtfMessages: apiGetNtfConns ntfConns \(String(describing: ntfConns.map { $0.connEntity.id }))")
                 for ntfConn in ntfConns {
                     addExpectedMessage(ntfConn: ntfConn)
                 }
 
-                let connIdsToGet = expectedMessages.compactMap { (id, _) in
+                let connMsgReqs = expectedMessages.compactMap { (id, _) in
                     let started = NSEThreads.queue.sync {
                         let canStart = checkCanStart(id)
                         if let t = threadId { logger.debug("NotificationService thread \(t, privacy: .private): receiveNtfMessages: can start: \(canStart)") }
@@ -253,7 +280,7 @@ class NotificationService: UNNotificationServiceExtension {
                         return canStart
                     }
                     if started {
-                        return expectedMessages[id]?.receiveConnId
+                        return expectedMessages[id]?.connMsgReq
                     } else {
                         if let t = threadId { logger.debug("NotificationService thread \(t, privacy: .private): receiveNtfMessages: entity \(id, privacy: .private) waiting on semaphore") }
                         expectedMessages[id]?.semaphore.wait()
@@ -264,17 +291,17 @@ class NotificationService: UNNotificationServiceExtension {
                                 expectedMessages[id]?.startedProcessingNewMsgs = true
                                 expectedMessages[id]?.shouldProcessNtf = true
                             }
-                            if let connId = expectedMessages[id]?.receiveConnId {
+                            if let connMsgReq = expectedMessages[id]?.connMsgReq {
-                                let _ = getConnNtfMessage(connId: connId)
+                                let _ = getConnNtfMessage(connMsgReq: connMsgReq)
                             }
                         }
                         return nil
                     }
                 }
 
-                if !connIdsToGet.isEmpty {
+                if !connMsgReqs.isEmpty {
-                    if let r = apiGetConnNtfMessages(connIds: connIdsToGet) {
+                    if let r = apiGetConnNtfMessages(connMsgReqs: connMsgReqs) {
-                        logger.debug("NotificationService: receiveNtfMessages: apiGetConnNtfMessages count = \(r.count)")
+                        logger.debug("NotificationService: receiveNtfMessages: apiGetConnNtfMessages count = \(r.count), expecting messages \(r.count { $0 != nil })")
                     }
                     return
                 }
@@ -285,18 +312,16 @@ class NotificationService: UNNotificationServiceExtension {
         deliverBestAttemptNtf()
     }
 
-    func addExpectedMessage(ntfConn: UserNtfConn) {
+    func addExpectedMessage(ntfConn: NtfConn) {
-        if let connEntity = ntfConn.connEntity_,
+        let expectedMsgId = ntfConn.expectedMsg_?.msgId
-           let receiveEntityId = connEntity.id, ntfConn.expectedMsg_ != nil {
+        if let receiveEntityId = ntfConn.connEntity.id {
-            let expectedMsgId = ntfConn.expectedMsg_?.msgId
             logger.debug("NotificationService: addExpectedMessage: expectedMsgId = \(expectedMsgId ?? "nil", privacy: .private)")
             expectedMessages[receiveEntityId] = ExpectedMessage(
                 ntfConn: ntfConn,
-                receiveConnId: connEntity.conn.agentConnId,
                 expectedMsgId: expectedMsgId,
                 allowedGetNextAttempts: 3,
-                msgBestAttemptNtf: ntfConn.defaultBestAttemptNtf,
+                msgBestAttemptNtf: defaultBestAttemptNtf(ntfConn),
-                ready: false,
+                ready: ntfConn.expectedMsg_ == nil, // show defaultBestAttemptNtf(ntfConn) if there is no expected message
                 shouldProcessNtf: false,
                 startedProcessingNewMsgs: false,
                 semaphore: DispatchSemaphore(value: 0)
@@ -350,10 +375,10 @@ class NotificationService: UNNotificationServiceExtension {
                 NSEThreads.shared.droppedNotifications.append((id, ntf))
                 if signalReady { entityReady(id) }
                 self.deliverBestAttemptNtf()
-            } else if (expectedMessages[id]?.allowedGetNextAttempts ?? 0) > 0, let receiveConnId = expectedMessages[id]?.receiveConnId {
+            } else if (expectedMessages[id]?.allowedGetNextAttempts ?? 0) > 0, let connMsgReq = expectedMessages[id]?.connMsgReq {
                 logger.debug("NotificationService processNtf: msgInfo msgId = \(info.msgId, privacy: .private): unexpected msgInfo, get next message")
                 expectedMessages[id]?.allowedGetNextAttempts -= 1
-                if let receivedMsg = getConnNtfMessage(connId: receiveConnId) {
+                if let receivedMsg = getConnNtfMessage(connMsgReq: connMsgReq) {
                     logger.debug("NotificationService processNtf, on getConnNtfMessage: msgInfo msgId = \(info.msgId, privacy: .private), receivedMsg msgId = \(receivedMsg.msgId, privacy: .private)")
                 } else {
                     logger.debug("NotificationService processNtf, on getConnNtfMessage: msgInfo msgId = \(info.msgId, privacy: .private): no next message, deliver best attempt")
@@ -373,13 +398,9 @@ class NotificationService: UNNotificationServiceExtension {
                 setBadgeCount()
             }
             let prevBestAttempt = expectedMessages[id]?.msgBestAttemptNtf
-            if prevBestAttempt?.callInvitation != nil {
+            if prevBestAttempt?.callInvitation == nil || ntf.callInvitation != nil {
-                if ntf.callInvitation != nil { // replace with newer call
-                    expectedMessages[id]?.msgBestAttemptNtf = ntf
-                } // otherwise keep call as best attempt
-            } else {
                 expectedMessages[id]?.msgBestAttemptNtf = ntf
-            }
+            } // otherwise keep call as best attempt
         } else {
             NSEThreads.shared.droppedNotifications.append((id, ntf))
             if signalReady { entityReady(id) }
@@ -406,7 +427,11 @@ class NotificationService: UNNotificationServiceExtension {
     }
 
     private func deliverBestAttemptNtf(urgent: Bool = false) {
-        if (urgent || !expectingMoreMessages) {
+        logger.debug("NotificationService.deliverBestAttemptNtf urgent: \(urgent) expectingMoreMessages: \(self.expectingMoreMessages)")
+        if let handler = contentHandler, urgent || !expectingMoreMessages {
+            if urgent {
+                contentHandler = nil
+            }
             logger.debug("NotificationService.deliverBestAttemptNtf")
             // stop processing other messages
             for (key, _) in expectedMessages {
@@ -420,18 +445,18 @@ class NotificationService: UNNotificationServiceExtension {
             } else {
                 suspend = false
             }
-            deliverCallkitOrNotification(urgent: urgent, suspend: suspend)
+            deliverCallkitOrNotification(urgent: urgent, suspend: suspend, handler: handler)
         }
     }
 
-    private func deliverCallkitOrNotification(urgent: Bool, suspend: Bool = false) {
+    private func deliverCallkitOrNotification(urgent: Bool, suspend: Bool = false, handler: @escaping (UNNotificationContent) -> Void) {
         if useCallKit() && expectedMessages.contains(where: { $0.value.msgBestAttemptNtf?.callInvitation != nil }) {
             logger.debug("NotificationService.deliverCallkitOrNotification: will suspend, callkit")
             if urgent {
                 // suspending NSE even though there may be other notifications
                 // to allow the app to process callkit call
                 suspendChat(0)
-                deliverNotification()
+                deliverNotification(handler: handler)
             } else {
                 // suspending NSE with delay and delivering after the suspension
                 // because pushkit notification must be processed without delay
@@ -439,7 +464,7 @@ class NotificationService: UNNotificationServiceExtension {
                 DispatchQueue.global().asyncAfter(deadline: .now() + fastNSESuspendSchedule.delay) {
                     suspendChat(fastNSESuspendSchedule.timeout)
                     DispatchQueue.global().asyncAfter(deadline: .now() + Double(fastNSESuspendSchedule.timeout)) {
-                        self.deliverNotification()
+                        self.deliverNotification(handler: handler)
                     }
                 }
             }
@@ -458,12 +483,12 @@ class NotificationService: UNNotificationServiceExtension {
                     }
                 }
             }
-            deliverNotification()
+            deliverNotification(handler: handler)
         }
     }
 
-    private func deliverNotification() {
+    private func deliverNotification(handler: @escaping (UNNotificationContent) -> Void) {
-        if let handler = contentHandler, let ntf = prepareNotification() {
+        if serviceBestAttemptNtf != nil, let ntf = prepareNotification() {
             contentHandler = nil
             serviceBestAttemptNtf = nil
             switch ntf {
@@ -496,7 +521,9 @@ class NotificationService: UNNotificationServiceExtension {
                   let callNtf = callNtfKV.value.msgBestAttemptNtf {
             return useCallKit() ? .callkit(callInv) : .nse(callNtf.notificationContent(badgeCount))
         } else {
+            logger.debug("NotificationService prepareNotification \(String(describing: self.expectedMessages.map { $0.key }))")
             let ntfEvents = expectedMessages.compactMap { $0.value.msgBestAttemptNtf?.notificationEvent }
+            logger.debug("NotificationService prepareNotification \(ntfEvents.count)")
             if ntfEvents.isEmpty {
                 return .empty
             } else if let ntfEvent = ntfEvents.count == 1 ? ntfEvents.first : nil {
@@ -654,7 +681,7 @@ func doStartChat() -> DBMigrationResult? {
     let state = NSEChatState.shared.value
     NSEChatState.shared.set(.starting)
     if let user = apiGetActiveUser() {
-        logger.debug("NotificationService active user \(String(describing: user))")
+        logger.debug("NotificationService active user \(user.displayName)")
         do {
             try setNetworkConfig(networkConfig)
             try apiSetAppFilePaths(filesFolder: getAppFilesDirectory().path, tempFolder: getTempFilesDirectory().path, assetsFolder: getWallpaperDirectory().deletingLastPathComponent().path)
@@ -893,7 +920,7 @@ func apiSetEncryptLocalFiles(_ enable: Bool) throws {
     throw r
 }
 
-func apiGetNtfConns(nonce: String, encNtfInfo: String) -> [UserNtfConn]? {
+func apiGetNtfConns(nonce: String, encNtfInfo: String) -> [NtfConn]? {
     guard apiGetActiveUser() != nil else {
         logger.debug("no active user")
         return nil
@@ -901,7 +928,7 @@ func apiGetNtfConns(nonce: String, encNtfInfo: String) -> [UserNtfConn]? {
     let r = sendSimpleXCmd(.apiGetNtfConns(nonce: nonce, encNtfInfo: encNtfInfo))
     if case let .ntfConns(ntfConns) = r {
         logger.debug("apiGetNtfConns response ntfConns: \(ntfConns.count)")
-        return ntfConns.compactMap { toUserNtfConn($0) }
+        return ntfConns
     } else if case let .chatCmdError(_, error) = r {
         logger.debug("apiGetNtfMessage error response: \(String.init(describing: error))")
     } else {
@@ -910,30 +937,23 @@ func apiGetNtfConns(nonce: String, encNtfInfo: String) -> [UserNtfConn]? {
     return nil
 }
 
-func toUserNtfConn(_ ntfConn: NtfConn) -> UserNtfConn? {
+func apiGetConnNtfMessages(connMsgReqs: [ConnMsgReq]) -> [NtfMsgInfo?]? {
-    if let user = ntfConn.user_ {
-        return UserNtfConn(user: user, connEntity_: ntfConn.connEntity_, expectedMsg_: ntfConn.expectedMsg_)
-    } else {
-        return nil
-    }
-}
-
-func apiGetConnNtfMessages(connIds: [String]) -> [NtfMsgInfo?]? {
     guard apiGetActiveUser() != nil else {
         logger.debug("no active user")
         return nil
     }
-    let r = sendSimpleXCmd(.apiGetConnNtfMessages(connIds: connIds))
+    logger.debug("apiGetConnNtfMessages command: \(ChatCommand.apiGetConnNtfMessages(connMsgReqs: connMsgReqs).cmdString)")
+    let r = sendSimpleXCmd(.apiGetConnNtfMessages(connMsgReqs: connMsgReqs))
     if case let .connNtfMessages(receivedMsgs) = r {
-        logger.debug("apiGetConnNtfMessages response receivedMsgs: \(receivedMsgs.count)")
+        logger.debug("apiGetConnNtfMessages response receivedMsgs: total \(receivedMsgs.count), expecting messages \(receivedMsgs.count { $0 != nil })")
         return receivedMsgs
     }
     logger.debug("apiGetConnNtfMessages error: \(responseError(r))")
     return nil
 }
 
-func getConnNtfMessage(connId: String) -> NtfMsgInfo? {
+func getConnNtfMessage(connMsgReq: ConnMsgReq) -> NtfMsgInfo? {
-    let r_ = apiGetConnNtfMessages(connIds: [connId])
+    let r_ = apiGetConnNtfMessages(connMsgReqs: [connMsgReq])
     if let r = r_, let receivedMsg = r.count == 1 ? r.first : nil {
         return receivedMsg
     }
@@ -974,33 +994,28 @@ func setNetworkConfig(_ cfg: NetCfg) throws {
     throw r
 }
 
-struct UserNtfConn {
+func defaultBestAttemptNtf(_ ntfConn: NtfConn) -> NSENotificationData {
-    var user: User
+    let user = ntfConn.user
-    var connEntity_: ConnectionEntity?
+    let connEntity = ntfConn.connEntity
-    var expectedMsg_: NtfMsgInfo?
+    return if !user.showNotifications {
-
+        .noNtf
-    var defaultBestAttemptNtf: NSENotificationData {
+    } else {
-        return if !user.showNotifications {
+        switch ntfConn.connEntity {
-            .noNtf
+        case let .rcvDirectMsgConnection(_, contact):
-        } else if let connEntity = connEntity_ {
+            contact?.chatSettings.enableNtfs == .all
-            switch connEntity {
+            ? .connectionEvent(user, connEntity)
-            case let .rcvDirectMsgConnection(_, contact):
+            : .noNtf
-                contact?.chatSettings.enableNtfs == .all
+        case let .rcvGroupMsgConnection(_, groupInfo, _):
-                ? .connectionEvent(user, connEntity)
+            groupInfo.chatSettings.enableNtfs == .all
-                : .noNtf
+            ? .connectionEvent(user, connEntity)
-            case let .rcvGroupMsgConnection(_, groupInfo, _):
+            : .noNtf
-                groupInfo.chatSettings.enableNtfs == .all
+        case .sndFileConnection: .noNtf
-                ? .connectionEvent(user, connEntity)
+        case .rcvFileConnection: .noNtf
-                : .noNtf
+        case let .userContactConnection(_, userContact):
-            case .sndFileConnection: .noNtf
+            userContact.groupId == nil
-            case .rcvFileConnection: .noNtf
+            ? .connectionEvent(user, connEntity)
-            case let .userContactConnection(_, userContact):
+            : .noNtf
-                userContact.groupId == nil
-                ? .connectionEvent(user, connEntity)
-                : .noNtf
-            }
-        } else {
-            .noNtf
         }
     }
 }
+

apps/ios/SimpleX SE/ShareModel.swift

@@ -179,7 +179,7 @@ class ShareModel: ObservableObject {
                 resetChatCtrl() // Clears retained migration result
                 registerGroupDefaults()
                 haskell_init_se()
-                let (_, result) = chatMigrateInit(dbKey, confirmMigrations: defaultMigrationConfirmation())
+                let (_, result) = chatMigrateInit(dbKey, confirmMigrations: defaultMigrationConfirmation(), backgroundMode: false)
                 if let e = migrationError(result) { return e }
                 try apiSetAppFilePaths(
                     filesFolder: getAppFilesDirectory().path,

apps/ios/SimpleXChat/APITypes.swift

@@ -67,7 +67,7 @@ public enum ChatCommand {
     case apiCheckToken(token: DeviceToken)
     case apiDeleteToken(token: DeviceToken)
     case apiGetNtfConns(nonce: String, encNtfInfo: String)
-    case apiGetConnNtfMessages(connIds: [String])
+    case apiGetConnNtfMessages(connMsgReqs: [ConnMsgReq])
     case apiNewGroup(userId: Int64, incognito: Bool, groupProfile: GroupProfile)
     case apiAddMember(groupId: Int64, contactId: Int64, memberRole: GroupMemberRole)
     case apiJoinGroup(groupId: Int64)
@@ -246,7 +246,7 @@ public enum ChatCommand {
             case let .apiCheckToken(token): return "/_ntf check \(token.cmdString)"
             case let .apiDeleteToken(token): return "/_ntf delete \(token.cmdString)"
             case let .apiGetNtfConns(nonce, encNtfInfo): return "/_ntf conns \(nonce) \(encNtfInfo)"
-            case let .apiGetConnNtfMessages(connIds): return "/_ntf conn messages \(connIds.joined(separator: ","))"
+            case let .apiGetConnNtfMessages(connMsgReqs): return "/_ntf conn messages \(connMsgReqs.map { $0.cmdString }.joined(separator: ","))"
             case let .apiNewGroup(userId, incognito, groupProfile): return "/_group \(userId) incognito=\(onOff(incognito)) \(encodeJSON(groupProfile))"
             case let .apiAddMember(groupId, contactId, memberRole): return "/_add #\(groupId) \(contactId) \(memberRole)"
             case let .apiJoinGroup(groupId): return "/_join #\(groupId)"

apps/ios/SimpleXChat/ChatTypes.swift

@@ -2399,14 +2399,14 @@ public enum ConnectionEntity: Decodable, Hashable {
 
     public var id: String? {
         switch self {
-        case let .rcvDirectMsgConnection(_, contact):
+        case let .rcvDirectMsgConnection(conn, contact):
-            return contact?.id
+            contact?.id ?? conn.id
         case let .rcvGroupMsgConnection(_, _, groupMember):
-            return groupMember.id
+            groupMember.id
         case let .userContactConnection(_, userContact):
-            return userContact.id
+            userContact.id
         default:
-            return nil
+            nil
         }
     }
 
@@ -2422,10 +2422,11 @@ public enum ConnectionEntity: Decodable, Hashable {
 }
 
 public struct NtfConn: Decodable, Hashable {
-    public var user_: User?
+    public var user: User
-    public var connEntity_: ConnectionEntity?
+    public var agentConnId: String
+    public var agentDbQueueId: Int64
+    public var connEntity: ConnectionEntity
     public var expectedMsg_: NtfMsgInfo?
-
 }
 
 public struct NtfMsgInfo: Decodable, Hashable {
@@ -2433,6 +2434,29 @@ public struct NtfMsgInfo: Decodable, Hashable {
     public var msgTs: Date
 }
 
+let iso8601DateFormatter = {
+    let f = ISO8601DateFormatter()
+    f.formatOptions = [.withInternetDateTime]
+    return f
+}()
+
+// used in apiGetConnNtfMessages
+public struct ConnMsgReq {
+    public var msgConnId: String
+    public var msgDbQueueId: Int64
+    public var msgTs: Date // SystemTime encodes as a number, should be taken from NtfMsgInfo
+
+    public init(msgConnId: String, msgDbQueueId: Int64, msgTs: Date) {
+        self.msgConnId = msgConnId
+        self.msgDbQueueId = msgDbQueueId
+        self.msgTs = msgTs
+    }
+
+    public var cmdString: String {
+        "\(msgConnId):\(msgDbQueueId):\(iso8601DateFormatter.string(from: msgTs))"
+    }
+}
+
 public struct NtfMsgAckInfo: Decodable, Hashable {
     public var msgId: String
     public var msgTs_: Date?

cabal.project

@@ -12,7 +12,7 @@ constraints: zip +disable-bzip2 +disable-zstd
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/simplexmq.git
-    tag: 08b84deba458407ae97d55debd98b872cb6c4d79
+    tag: 3d10c9bf9e4d8196d39162ff8712f6b729b8c247
 
 source-repository-package
     type: git

scripts/nix/sha256map.nix

@@ -1,5 +1,5 @@
 {
-  "https://github.com/simplex-chat/simplexmq.git"."08b84deba458407ae97d55debd98b872cb6c4d79" = "0b4n7d81spl1r7zppr0lc40ls9m1i93g4l3hzg2996pi3bxmafrr";
+  "https://github.com/simplex-chat/simplexmq.git"."3d10c9bf9e4d8196d39162ff8712f6b729b8c247" = "1nnr6klv240da97qmrzlh8jywpimcnlrxnxnjrm2rd0w0w7gvra1";
   "https://github.com/simplex-chat/hs-socks.git"."a30cc7a79a08d8108316094f8f2f82a0c5e1ac51" = "0yasvnr7g91k76mjkamvzab2kvlb1g5pspjyjn2fr6v83swjhj38";
   "https://github.com/simplex-chat/direct-sqlcipher.git"."f814ee68b16a9447fbb467ccc8f29bdd3546bfd9" = "1ql13f4kfwkbaq7nygkxgw84213i0zm7c1a8hwvramayxl38dq5d";
   "https://github.com/simplex-chat/sqlcipher-simple.git"."a46bd361a19376c5211f1058908fc0ae6bf42446" = "1z0r78d8f0812kxbgsm735qf6xx8lvaz27k1a0b4a2m0sshpd5gl";

src/Simplex/Chat/Controller.hs

@@ -357,7 +357,7 @@ data ChatCommand
   | APICheckToken DeviceToken
   | APIDeleteToken DeviceToken
   | APIGetNtfConns {nonce :: C.CbNonce, encNtfInfo :: ByteString}
-  | ApiGetConnNtfMessages {connIds :: NonEmpty AgentConnId}
+  | APIGetConnNtfMessages (NonEmpty ConnMsgReq)
   | APIAddMember GroupId ContactId GroupMemberRole
   | APIJoinGroup {groupId :: GroupId, enableNtfs :: MsgFilter}
   | APIAcceptMember GroupId GroupMemberId GroupMemberRole
@@ -1131,12 +1131,18 @@ instance FromJSON ChatTagData where
   parseJSON invalid = JT.prependFailure "bad ChatTagData, " (JT.typeMismatch "Object" invalid)
 
 data NtfConn = NtfConn
-  { user_ :: Maybe User,
+  { user :: User,
-    connEntity_ :: Maybe ConnectionEntity,
+    agentConnId :: AgentConnId,
+    agentDbQueueId :: Int64,
+    connEntity :: ConnectionEntity,
+    -- Decrypted ntf meta of the expected message (the one notification was sent for).
+    -- Nothing means it failed to decrypt or to decode, we can still show event for entity
     expectedMsg_ :: Maybe NtfMsgInfo
   }
   deriving (Show)
 
+-- brokerTs is the same msgTs, it is used in ConnMsgReq / APIGetConnNtfMessages
+-- to set it as last connection message in case queue is empty
 data NtfMsgInfo = NtfMsgInfo {msgId :: Text, msgTs :: UTCTime}
   deriving (Show)
 

src/Simplex/Chat/Core.hs

@@ -34,7 +34,7 @@ import Text.Read (readMaybe)
 import UnliftIO.Async
 
 simplexChatCore :: ChatConfig -> ChatOpts -> (User -> ChatController -> IO ()) -> IO ()
-simplexChatCore cfg@ChatConfig {confirmMigrations, testView} opts@ChatOpts {coreOptions = CoreChatOpts {dbOptions, logAgent, yesToUpMigrations}} chat =
+simplexChatCore cfg@ChatConfig {confirmMigrations, testView} opts@ChatOpts {coreOptions = CoreChatOpts {dbOptions, logAgent, yesToUpMigrations}, maintenance} chat =
   case logAgent of
     Just level -> do
       setLogLevel level
@@ -48,7 +48,8 @@ simplexChatCore cfg@ChatConfig {confirmMigrations, testView} opts@ChatOpts {core
       exitFailure
     run db@ChatDatabase {chatStore} = do
       u_ <- getSelectActiveUser chatStore
-      cc <- newChatController db u_ cfg opts False
+      let backgroundMode = not maintenance
+      cc <- newChatController db u_ cfg opts backgroundMode
       u <- maybe (createActiveUser cc) pure u_
       unless testView $ putStrLn $ "Current user: " <> userStr u
       runSimplexChat opts u cc chat

src/Simplex/Chat/Library/Commands.hs

@@ -1294,26 +1294,17 @@ processChatCommand' vr = \case
     ntfInfos <- withAgent $ \a -> getNotificationConns a nonce encNtfInfo
     (errs, ntfMsgs) <- lift $ partitionEithers <$> withStoreBatch' (\db -> map (getMsgConn db) (L.toList ntfInfos))
     unless (null errs) $ toView $ CRChatErrors (Just user) errs
-    pure $ CRNtfConns ntfMsgs
+    pure $ CRNtfConns $ catMaybes ntfMsgs
     where
-      getMsgConn :: DB.Connection -> NotificationInfo -> IO NtfConn
+      getMsgConn :: DB.Connection -> NotificationInfo -> IO (Maybe NtfConn)
-      getMsgConn db NotificationInfo {ntfConnId, ntfMsgMeta = nMsgMeta} = do
+      getMsgConn db NotificationInfo {ntfConnId, ntfDbQueueId, ntfMsgMeta = nMsgMeta} = do
         let agentConnId = AgentConnId ntfConnId
-        user_ <- getUserByAConnId db agentConnId
+            mkNtfConn user connEntity = NtfConn {user, agentConnId, agentDbQueueId = ntfDbQueueId, connEntity, expectedMsg_ = expectedMsgInfo <$> nMsgMeta}
-        connEntity_ <-
+        getUserByAConnId db agentConnId
-          pure user_ $>>= \user ->
+          $>>= \user -> fmap (mkNtfConn user) . eitherToMaybe <$> runExceptT (getConnectionEntity db vr user agentConnId)
-            eitherToMaybe <$> runExceptT (getConnectionEntity db vr user agentConnId)
+  APIGetConnNtfMessages connMsgs -> withUser $ \_ -> do
-        pure $
+    msgs <- lift $ withAgent' (`getConnectionMessages` connMsgs)
-          NtfConn
+    let ntfMsgs = L.map (receivedMsgInfo <$>) msgs
-            { user_,
-              connEntity_,
-              -- Decrypted ntf meta of the expected message (the one notification was sent for)
-              expectedMsg_ = expectedMsgInfo <$> nMsgMeta
-            }
-  ApiGetConnNtfMessages connIds -> withUser $ \_ -> do
-    let acIds = L.map (\(AgentConnId acId) -> acId) connIds
-    msgs <- lift $ withAgent' $ \a -> getConnectionMessages a acIds
-    let ntfMsgs = L.map (\msg -> receivedMsgInfo <$> msg) msgs
     pure $ CRConnNtfMessages ntfMsgs
   GetUserProtoServers (AProtocolType p) -> withUser $ \user -> withServerProtocol p $ do
     srvs <- withFastStore (`getUserServers` user)
@@ -4005,7 +3996,7 @@ chatCommandP =
       "/_ntf check " *> (APICheckToken <$> strP),
       "/_ntf delete " *> (APIDeleteToken <$> strP),
       "/_ntf conns " *> (APIGetNtfConns <$> strP <* A.space <*> strP),
-      "/_ntf conn messages " *> (ApiGetConnNtfMessages <$> strP),
+      "/_ntf conn messages " *> (APIGetConnNtfMessages <$> connMsgsP),
       "/_add #" *> (APIAddMember <$> A.decimal <* A.space <*> A.decimal <*> memberRole),
       "/_join #" *> (APIJoinGroup <$> A.decimal <*> pure MFAll), -- needs to be changed to support in UI
       "/_accept member #" *> (APIAcceptMember <$> A.decimal <* A.space <*> A.decimal <*> memberRole),
@@ -4320,6 +4311,12 @@ chatCommandP =
       cfArgs <- optional $ CFArgs <$> (" key=" *> strP <* A.space) <*> (" nonce=" *> strP)
       path <- filePath
       pure $ CryptoFile path cfArgs
+    connMsgsP = L.fromList <$> connMsgP `A.sepBy1'` A.char ','
+    connMsgP = do
+      AgentConnId msgConnId <- strP <* A.char ':'
+      msgDbQueueId <- strP <* A.char ':'
+      ts <- strP
+      pure ConnMsgReq {msgConnId, msgDbQueueId, msgTs = Just ts}
     memberRole =
       A.choice
         [ " owner" $> GROwner,

src/Simplex/Chat/Store/SQLite/Migrations/agent_query_plans.txt

@@ -1119,9 +1119,9 @@ Query: UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id =
 Plan:
 SEARCH rcv_messages USING COVERING INDEX idx_rcv_messages_conn_id_internal_id (conn_id=? AND internal_id=?)
 
-Query: UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ?
+Query: UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ? AND last_broker_ts < ?
 Plan:
-SEARCH rcv_queues USING COVERING INDEX idx_rcv_queue_id (conn_id=? AND rcv_queue_id=?)
+SEARCH rcv_queues USING INDEX idx_rcv_queue_id (conn_id=? AND rcv_queue_id=?)
 
 Query: UPDATE rcv_queues SET rcv_primary = ? WHERE conn_id = ?
 Plan:

src/Simplex/Chat/View.hs

@@ -342,8 +342,8 @@ responseToView hu@(currentRH, user_) cfg@ChatConfig {logLevel, showReactions, sh
   CRContactConnectionDeleted u PendingContactConnection {pccConnId} -> ttyUser u ["connection :" <> sShow pccConnId <> " deleted"]
   CRNtfTokenStatus status -> ["device token status: " <> plain (smpEncode status)]
   CRNtfToken _ status mode srv -> ["device token status: " <> plain (smpEncode status) <> ", notifications mode: " <> plain (strEncode mode) <> ", server: " <> sShow srv]
-  CRNtfConns {} -> []
+  CRNtfConns {ntfConns} -> map (\NtfConn {agentConnId, expectedMsg_} -> plain $ show agentConnId <> " " <> show expectedMsg_) ntfConns
-  CRConnNtfMessages {} -> []
+  CRConnNtfMessages ntfMsgs -> [sShow ntfMsgs]
   CRNtfMessage {} -> []
   CRCurrentRemoteHost rhi_ ->
     [ maybe

tests/Bots/BroadcastTests.hs

@@ -78,10 +78,11 @@ testBroadcastMessages ps = do
           bob <## "I broadcast messages to all connected users from @alice."
           cath `connectVia` botLink
           alice #> "@broadcast_bot hello all!"
+          alice <# "broadcast_bot> hello all!" -- we broadcast to the sender too, /feed is used by bot
           bob <# "broadcast_bot> hello all!"
           cath <# "broadcast_bot> hello all!"
           alice <# "broadcast_bot> > hello all!"
-          alice <## "      Forwarded to 2 contact(s)"
+          alice <## "      Forwarded to 3 contact(s), 0 errors"
   where
     cc `connectVia` botLink = do
       cc ##> ("/c " <> botLink)