ios, core: better notifications processing to avoid contention for database (#3485)

* core: forward notifications about message processing (for iOS notifications)

* simplexmq

* the option to keep database key, to allow re-opening the database

* export new init with keepKey and reopen DB api

* stop remote ctrl when suspending chat

* ios: close/re-open db on suspend/activate

* allow activating chat without restoring (for NSE)

* update NSE to suspend/activate (does not work)

* simplexmq

* suspend chat and close database when last notification in the process is processed

* stop reading notifications on message markers

* replace async stream with cancellable concurrent queue

* better synchronization of app and NSE

* remove outside of task

* remove unused var

* whitespace

* more debug logging, handle cancelled read after dequeue

* comments

* more comments

apps/ios/SimpleX NSE/ConcurrentQueue.swift

@@ -0,0 +1,64 @@
+//
+//  ConcurrentQueue.swift
+//  SimpleX NSE
+//
+//  Created by Evgeny on 08/12/2023.
+//  Copyright © 2023 SimpleX Chat. All rights reserved.
+//
+
+import Foundation
+
+struct DequeueElement<T> {
+    var elementId: UUID?
+    var task: Task<T?, Never>
+}
+
+class ConcurrentQueue<T> {
+    private var queue: [T] = []
+    private var queueLock = DispatchQueue(label: "chat.simplex.app.SimpleX-NSE.concurrent-queue.lock.\(UUID())")
+    private var continuations = [(elementId: UUID, continuation: CheckedContinuation<T?, Never>)]()
+
+    func enqueue(_ el: T) {
+        resumeContinuation(el) { self.queue.append(el) }
+    }
+
+    func frontEnqueue(_ el: T) {
+        resumeContinuation(el) { self.queue.insert(el, at: 0) }
+    }
+
+    private func resumeContinuation(_ el: T, add: @escaping () -> Void) {
+        queueLock.sync {
+            if let (_, cont) = continuations.first {
+                continuations.remove(at: 0)
+                cont.resume(returning: el)
+            } else {
+                add()
+            }
+        }
+    }
+
+    func dequeue() -> DequeueElement<T> {
+        queueLock.sync {
+            if queue.isEmpty {
+                let elementId = UUID()
+                let task = Task {
+                    await withCheckedContinuation { cont in
+                        continuations.append((elementId, cont))
+                    }
+                }
+                return DequeueElement(elementId: elementId, task: task)
+            } else {
+                let el = queue.remove(at: 0)
+                return DequeueElement(task: Task { el })
+            }
+        }
+    }
+
+    func cancelDequeue(_ elementId: UUID) {
+        queueLock.sync {
+            let cancelled = continuations.filter { $0.elementId == elementId }
+            continuations.removeAll { $0.elementId == elementId }
+            cancelled.forEach { $0.continuation.resume(returning: nil) }
+        }
+    }
+}

apps/ios/SimpleX NSE/NotificationService.swift

@@ -14,68 +14,167 @@ import SimpleXChat
 
 let logger = Logger()
 
-let suspendingDelay: UInt64 = 2_000_000_000
+let suspendingDelay: UInt64 = 2_500_000_000
 
-typealias NtfStream = AsyncStream<NSENotification>
+let nseSuspendTimeout: Int = 10
 
+typealias NtfStream = ConcurrentQueue<NSENotification>
+
+// Notifications are delivered via concurrent queues, as they are all received from chat controller in a single loop that
+// writes to ConcurrentQueue and when notification is processed, the instance of Notification service extension reads from the queue.
+// One queue per connection (entity) is used.
+// The concurrent queues allow for read cancellation, to ensure that notifications are not lost in case the next the current thread completes
+// before expected notification is read (multiple notifications can be expected, because one notification can be delivered for several messages.
 actor PendingNtfs {
     static let shared = PendingNtfs()
     private var ntfStreams: [String: NtfStream] = [:]
-    private var ntfConts: [String: NtfStream.Continuation] = [:]
 
-    func createStream(_ id: String) {
-        logger.debug("PendingNtfs.createStream: \(id, privacy: .public)")
-        if ntfStreams.index(forKey: id) == nil {
-            ntfStreams[id] = AsyncStream { cont in
-                ntfConts[id] = cont
-                logger.debug("PendingNtfs.createStream: store continuation")
-            }
+    func createStream(_ id: String) async {
+        logger.debug("NotificationService PendingNtfs.createStream: \(id, privacy: .public)")
+        if ntfStreams[id] == nil {
+            ntfStreams[id] = ConcurrentQueue()
+            logger.debug("NotificationService PendingNtfs.createStream: created ConcurrentQueue")
         }
     }
 
-    func readStream(_ id: String, for nse: NotificationService, msgCount: Int = 1, showNotifications: Bool) async {
-        logger.debug("PendingNtfs.readStream: \(id, privacy: .public) \(msgCount, privacy: .public)")
+    func readStream(_ id: String, for nse: NotificationService, ntfInfo: NtfMessages) async {
+        logger.debug("NotificationService PendingNtfs.readStream: \(id, privacy: .public) \(ntfInfo.ntfMessages.count, privacy: .public)")
+        if !ntfInfo.user.showNotifications {
+            nse.setBestAttemptNtf(.empty)
+        }
         if let s = ntfStreams[id] {
-            logger.debug("PendingNtfs.readStream: has stream")
-            var rcvCount = max(1, msgCount)
-            for await ntf in s {
-                nse.setBestAttemptNtf(showNotifications ? ntf : .empty)
-                rcvCount -= 1
-                if rcvCount == 0 || ntf.categoryIdentifier == ntfCategoryCallInvitation { break }
+            logger.debug("NotificationService PendingNtfs.readStream: has stream")
+            var expected = Set(ntfInfo.ntfMessages.map { $0.msgId })
+            logger.debug("NotificationService PendingNtfs.readStream: expecting: \(expected, privacy: .public)")
+            var readCancelled = false
+            var dequeued: DequeueElement<NSENotification>?
+            nse.cancelRead = {
+                readCancelled = true
+                if let elementId = dequeued?.elementId {
+                    s.cancelDequeue(elementId)
+                }
             }
-            logger.debug("PendingNtfs.readStream: exiting")
+            while !readCancelled {
+                dequeued = s.dequeue()
+                if let ntf = await dequeued?.task.value {
+                    if readCancelled {
+                        logger.debug("NotificationService PendingNtfs.readStream: read cancelled, put ntf to queue front")
+                        s.frontEnqueue(ntf)
+                        break
+                    } else if case let .msgInfo(info) = ntf {
+                        let found = expected.remove(info.msgId)
+                        if found != nil {
+                            logger.debug("NotificationService PendingNtfs.readStream: msgInfo, last: \(expected.isEmpty, privacy: .public)")
+                            if expected.isEmpty { break }
+                        } else if let msgTs = ntfInfo.msgTs, info.msgTs > msgTs {
+                            logger.debug("NotificationService PendingNtfs.readStream: unexpected msgInfo")
+                            s.frontEnqueue(ntf)
+                            break
+                        }
+                    } else if ntfInfo.user.showNotifications {
+                        logger.debug("NotificationService PendingNtfs.readStream: setting best attempt")
+                        nse.setBestAttemptNtf(ntf)
+                        if ntf.isCallInvitation { break }
+                    }
+                } else {
+                    break
+                }
+            }
+            nse.cancelRead = nil
+            logger.debug("NotificationService PendingNtfs.readStream: exiting")
         }
     }
 
-    func writeStream(_ id: String, _ ntf: NSENotification) {
-        logger.debug("PendingNtfs.writeStream: \(id, privacy: .public)")
-        if let cont = ntfConts[id] {
-            logger.debug("PendingNtfs.writeStream: writing ntf")
-            cont.yield(ntf)
+    func writeStream(_ id: String, _ ntf: NSENotification) async {
+        logger.debug("NotificationService PendingNtfs.writeStream: \(id, privacy: .public)")
+        if let s = ntfStreams[id] {
+            logger.debug("NotificationService PendingNtfs.writeStream: writing ntf")
+            s.enqueue(ntf)
+        }
+    }
+}
+
+// The current implementation assumes concurrent notification delivery and uses semaphores
+// to process only one notification per connection (entity) at a time.
+class NtfStreamSemaphores {
+    static let shared = NtfStreamSemaphores()
+    private static let queue = DispatchQueue(label: "chat.simplex.app.SimpleX-NSE.notification-semaphores.lock")
+    private var semaphores: [String: DispatchSemaphore] = [:]
+
+    func waitForStream(_ id: String) {
+        streamSemaphore(id, value: 0)?.wait()
+    }
+
+    func signalStreamReady(_ id: String) {
+        streamSemaphore(id, value: 1)?.signal()
+    }
+
+    // this function returns nil if semaphore is just created, so passed value shoud be coordinated with the desired end value of the semaphore
+    private func streamSemaphore(_ id: String, value: Int) -> DispatchSemaphore? {
+        NtfStreamSemaphores.queue.sync {
+            if let s = semaphores[id] {
+                return s
+            } else {
+                semaphores[id] = DispatchSemaphore(value: value)
+                return nil
+            }
         }
     }
 }
 
 enum NSENotification {
-    case nse(notification: UNMutableNotificationContent)
-    case callkit(invitation: RcvCallInvitation)
+    case nse(UNMutableNotificationContent)
+    case callkit(RcvCallInvitation)
     case empty
+    case msgInfo(NtfMsgInfo)
 
-    var categoryIdentifier: String? {
+    var isCallInvitation: Bool {
         switch self {
-        case let .nse(ntf): return ntf.categoryIdentifier
-        case .callkit: return ntfCategoryCallInvitation
-        case .empty: return nil
+        case let .nse(ntf): ntf.categoryIdentifier == ntfCategoryCallInvitation
+        case .callkit: true
+        case .empty: false
+        case .msgInfo: false
         }
     }
 }
 
+// Once the last thread in the process completes processing chat controller is suspended, and the database is closed, to avoid
+// background crashes and contention for database with the application (both UI and background fetch triggered either on schedule
+// or when background notification is received.
+class NSEThreads {
+    static let shared = NSEThreads()
+    private static let queue = DispatchQueue(label: "chat.simplex.app.SimpleX-NSE.notification-threads.lock")
+    private var threads: Set<UUID> = []
+
+    func startThread() -> UUID {
+        NSEThreads.queue.sync {
+            let (_, t) = threads.insert(UUID())
+            return t
+        }
+    }
+
+    func endThread(_ t: UUID) -> Bool {
+        NSEThreads.queue.sync {
+            let t_ = threads.remove(t)
+            return t_ != nil && threads.isEmpty
+        }
+    }
+}
+
+// Notification service extension creates a new instance of the class and calls didReceive for each notification.
+// Each didReceive is called in its own thread, but multiple calls can be made in one process, and, empirically, there is never
+// more than one process for notification service extension.
+// Soon after notification service delivers the last notification it is either suspended or terminated.
 class NotificationService: UNNotificationServiceExtension {
     var contentHandler: ((UNNotificationContent) -> Void)?
     var bestAttemptNtf: NSENotification?
     var badgeCount: Int = 0
+    var threadId: UUID?
+    var receiveEntityId: String?
+    var cancelRead: (() -> Void)?
 
     override func didReceive(_ request: UNNotificationRequest, withContentHandler contentHandler: @escaping (UNNotificationContent) -> Void) {
+        threadId = NSEThreads.shared.startThread()
         logger.debug("DEBUGGING: NotificationService.didReceive")
         if let ntf = request.content.mutableCopy() as? UNMutableNotificationContent {
             setBestAttemptNtf(ntf)
@@ -93,7 +192,7 @@ class NotificationService: UNNotificationServiceExtension {
             setBadgeCount()
             Task {
                 var state = appState
-                for _ in 1...5 {
+                for _ in 1...6 {
                     _ = try await Task.sleep(nanoseconds: suspendingDelay)
                     state = appStateGroupDefault.get()
                     if state == .suspended || state != .suspending { break }
@@ -123,24 +222,28 @@ class NotificationService: UNNotificationServiceExtension {
            let encNtfInfo = ntfData["message"] as? String,
            let dbStatus = startChat() {
             if case .ok = dbStatus,
-               let ntfMsgInfo = apiGetNtfMessage(nonce: nonce, encNtfInfo: encNtfInfo) {
-                logger.debug("NotificationService: receiveNtfMessages: apiGetNtfMessage \(String(describing: ntfMsgInfo), privacy: .public)")
-                if let connEntity = ntfMsgInfo.connEntity {
+               let ntfInfo = apiGetNtfMessage(nonce: nonce, encNtfInfo: encNtfInfo) {
+                logger.debug("NotificationService: receiveNtfMessages: apiGetNtfMessage \(String(describing: ntfInfo), privacy: .public)")
+                if let connEntity = ntfInfo.connEntity_ {
                     setBestAttemptNtf(
-                        ntfMsgInfo.ntfsEnabled
-                        ? .nse(notification: createConnectionEventNtf(ntfMsgInfo.user, connEntity))
+                        ntfInfo.ntfsEnabled
+                        ? .nse(createConnectionEventNtf(ntfInfo.user, connEntity))
                         : .empty
                     )
                     if let id = connEntity.id {
-                        Task {
-                            logger.debug("NotificationService: receiveNtfMessages: in Task, connEntity id \(id, privacy: .public)")
-                            await PendingNtfs.shared.createStream(id)
-                            await PendingNtfs.shared.readStream(id, for: self, msgCount: ntfMsgInfo.ntfMessages.count, showNotifications: ntfMsgInfo.user.showNotifications)
-                            deliverBestAttemptNtf()
+                        receiveEntityId = id
+                        NtfStreamSemaphores.shared.waitForStream(id)
+                        if receiveEntityId != nil {
+                            Task {
+                                logger.debug("NotificationService: receiveNtfMessages: in Task, connEntity id \(id, privacy: .public)")
+                                await PendingNtfs.shared.createStream(id)
+                                await PendingNtfs.shared.readStream(id, for: self, ntfInfo: ntfInfo)
+                                deliverBestAttemptNtf()
+                            }
                         }
+                        return
                     }
                 }
-                return
             } else {
                 setBestAttemptNtf(createErrorNtf(dbStatus))
             }
@@ -159,14 +262,14 @@ class NotificationService: UNNotificationServiceExtension {
     }
 
     func setBestAttemptNtf(_ ntf: UNMutableNotificationContent) {
-        setBestAttemptNtf(.nse(notification: ntf))
+        setBestAttemptNtf(.nse(ntf))
     }
 
     func setBestAttemptNtf(_ ntf: NSENotification) {
         logger.debug("NotificationService.setBestAttemptNtf")
         if case let .nse(notification) = ntf {
             notification.badge = badgeCount as NSNumber
-            bestAttemptNtf = .nse(notification: notification)
+            bestAttemptNtf = .nse(notification)
         } else {
             bestAttemptNtf = ntf
         }
@@ -174,9 +277,33 @@ class NotificationService: UNNotificationServiceExtension {
 
     private func deliverBestAttemptNtf() {
         logger.debug("NotificationService.deliverBestAttemptNtf")
+        if let cancel = cancelRead {
+            cancelRead = nil
+            cancel()
+        }
+        if let id = receiveEntityId {
+            receiveEntityId = nil
+            NtfStreamSemaphores.shared.signalStreamReady(id)
+        }
+        if let t = threadId {
+            threadId = nil
+            if NSEThreads.shared.endThread(t) {
+                suspendChat(nseSuspendTimeout)
+            }
+        }
         if let handler = contentHandler, let ntf = bestAttemptNtf {
+            contentHandler = nil
+            bestAttemptNtf = nil
+            let deliver: (UNMutableNotificationContent?) -> Void = { ntf in
+                let useNtf = if let ntf = ntf {
+                    appStateGroupDefault.get().running ? UNMutableNotificationContent() : ntf
+                } else {
+                    UNMutableNotificationContent()
+                }
+                handler(useNtf)
+            }
             switch ntf {
-            case let .nse(content): handler(content)
+            case let .nse(content): deliver(content)
             case let .callkit(invitation):
                 CXProvider.reportNewIncomingVoIPPushPayload([
                     "displayName": invitation.contact.displayName,
@@ -184,33 +311,71 @@ class NotificationService: UNNotificationServiceExtension {
                     "media": invitation.callType.media.rawValue
                 ]) { error in
                     if error == nil {
-                        handler(UNMutableNotificationContent())
+                        deliver(nil)
                     } else {
-                        logger.debug("reportNewIncomingVoIPPushPayload success to CallController for \(invitation.contact.id)")
-                        handler(createCallInvitationNtf(invitation))
+                        logger.debug("NotificationService reportNewIncomingVoIPPushPayload success to CallController for \(invitation.contact.id)")
+                        deliver(createCallInvitationNtf(invitation))
                     }
                 }
-            case .empty: handler(UNMutableNotificationContent())
+            case .empty: deliver(nil) // used to mute notifications that did not unsubscribe yet
+            case .msgInfo: deliver(nil) // unreachable, the best attempt is never set to msgInfo
             }
-            bestAttemptNtf = nil
         }
     }
 }
 
-var chatStarted = false
-var networkConfig: NetCfg = getNetCfg()
-var xftpConfig: XFTPFileConfig? = getXFTPCfg()
+class NSEChatState {
+    static let shared = NSEChatState()
+    private var value_ = NSEState.created
 
+    var value: NSEState {
+        value_
+    }
+
+    func set(_ state: NSEState) {
+        nseStateGroupDefault.set(state)
+        value_ = state
+    }
+
+    init() {
+        set(.created)
+    }
+}
+
+var receiverStarted = false
+let startLock = DispatchSemaphore(value: 1)
+let suspendLock = DispatchSemaphore(value: 1)
+var networkConfig: NetCfg = getNetCfg()
+let xftpConfig: XFTPFileConfig? = getXFTPCfg()
+
+// startChat uses semaphore startLock to ensure that only one didReceive thread can start chat controller
+// Subsequent calls to didReceive will be waiting on semaphore and won't start chat again, as it will be .active
 func startChat() -> DBMigrationResult? {
+    logger.debug("NotificationService: startChat")
+    if case .active = NSEChatState.shared.value { return .ok }
+
+    startLock.wait()
+    defer { startLock.signal() }
+    
+    return switch NSEChatState.shared.value {
+    case .created: doStartChat()
+    case .active: .ok
+    case .suspending: activateChat()
+    case .suspended: activateChat()
+    }
+}
+
+func doStartChat() -> DBMigrationResult? {
+    logger.debug("NotificationService: doStartChat")
     hs_init(0, nil)
-    if chatStarted { return .ok }
     let (_, dbStatus) = chatMigrateInit(confirmMigrations: defaultMigrationConfirmation())
     if dbStatus != .ok {
         resetChatCtrl()
+        NSEChatState.shared.set(.created)
         return dbStatus
     }
     if let user = apiGetActiveUser() {
-        logger.debug("active user \(String(describing: user))")
+        logger.debug("NotificationService active user \(String(describing: user))")
         do {
             try setNetworkConfig(networkConfig)
             try apiSetTempFolder(tempFolder: getTempFilesDirectory().path)
@@ -218,32 +383,102 @@ func startChat() -> DBMigrationResult? {
             try setXFTPConfig(xftpConfig)
             try apiSetEncryptLocalFiles(privacyEncryptLocalFilesGroupDefault.get())
             let justStarted = try apiStartChat()
-            chatStarted = true
+            NSEChatState.shared.set(.active)
             if justStarted {
                 chatLastStartGroupDefault.set(Date.now)
-                Task { await receiveMessages() }
+                Task {
+                    if !receiverStarted {
+                        receiverStarted = true
+                        await receiveMessages()
+                    }
+                }
             }
             return .ok
         } catch {
             logger.error("NotificationService startChat error: \(responseError(error), privacy: .public)")
         }
     } else {
-        logger.debug("no active user")
+        logger.debug("NotificationService: no active user")
     }
     return nil
 }
 
+func activateChat() -> DBMigrationResult? {
+    logger.debug("NotificationService: activateChat")
+    let state = NSEChatState.shared.value
+    NSEChatState.shared.set(.active)
+    if apiActivateChat() {
+        logger.debug("NotificationService: activateChat: after apiActivateChat")
+        return .ok
+    } else {
+        NSEChatState.shared.set(state)
+        return nil
+    }
+}
+
+// suspendChat uses semaphore suspendLock to ensure that only one suspension can happen.
+func suspendChat(_ timeout: Int) {
+    logger.debug("NotificationService: suspendChat")
+    let state = NSEChatState.shared.value
+    if !state.canSuspend {
+        logger.error("NotificationService suspendChat called, current state: \(state.rawValue, privacy: .public)")
+    } else {
+        suspendLock.wait()
+        defer { suspendLock.signal() }
+
+        NSEChatState.shared.set(.suspending)
+        if apiSuspendChat(timeoutMicroseconds: timeout * 1000000) {
+            logger.debug("NotificationService: activateChat: after apiActivateChat")
+            DispatchQueue.global().asyncAfter(deadline: .now() + Double(timeout) + 1, execute: chatSuspended)
+        } else {
+            NSEChatState.shared.set(state)
+        }
+    }
+}
+
+func chatSuspended() {
+    logger.debug("NotificationService chatSuspended")
+    if case .suspending = NSEChatState.shared.value {
+        NSEChatState.shared.set(.suspended)
+        chatCloseStore()
+    }
+}
+
+// A single loop is used per Notification service extension process to receive and process all messages depending on the NSE state
+// If the extension is not active yet, or suspended/suspending, or the app is running, the notifications will no be received.
 func receiveMessages() async {
     logger.debug("NotificationService receiveMessages")
     while true {
-        updateNetCfg()
+        switch NSEChatState.shared.value {
+        case .created: await delayWhenInactive()
+        case .active:
+            if appStateGroupDefault.get().running {
+                suspendChat(nseSuspendTimeout)
+                await delayWhenInactive()
+            } else {
+                updateNetCfg()
+                await receiveMsg()
+            }
+        case .suspending: await receiveMsg()
+        case .suspended: await delayWhenInactive()
+        }
+    }
+
+    func receiveMsg() async {
         if let msg = await chatRecvMsg() {
+            logger.debug("NotificationService receiveMsg: message")
             if let (id, ntf) = await receivedMsgNtf(msg) {
+                logger.debug("NotificationService receiveMsg: notification")
                 await PendingNtfs.shared.createStream(id)
                 await PendingNtfs.shared.writeStream(id, ntf)
             }
         }
     }
+
+    func delayWhenInactive() async {
+        logger.debug("NotificationService delayWhenInactive")
+        _ = try? await Task.sleep(nanoseconds: 1000_000000)
+    }
 }
 
 func chatRecvMsg() async -> ChatResponse? {
@@ -257,14 +492,14 @@ private let isInChina = SKStorefront().countryCode == "CHN"
 private func useCallKit() -> Bool { !isInChina && callKitEnabledGroupDefault.get() }
 
 func receivedMsgNtf(_ res: ChatResponse) async -> (String, NSENotification)? {
-    logger.debug("NotificationService processReceivedMsg: \(res.responseType)")
+    logger.debug("NotificationService receivedMsgNtf: \(res.responseType, privacy: .public)")
     switch res {
     case let .contactConnected(user, contact, _):
-        return (contact.id, .nse(notification: createContactConnectedNtf(user, contact)))
+        return (contact.id, .nse(createContactConnectedNtf(user, contact)))
 //        case let .contactConnecting(contact):
 //            TODO profile update
     case let .receivedContactRequest(user, contactRequest):
-        return (UserContact(contactRequest: contactRequest).id, .nse(notification: createContactRequestNtf(user, contactRequest)))
+        return (UserContact(contactRequest: contactRequest).id, .nse(createContactRequestNtf(user, contactRequest)))
     case let .newChatItem(user, aChatItem):
         let cInfo = aChatItem.chatInfo
         var cItem = aChatItem.chatItem
@@ -274,7 +509,7 @@ func receivedMsgNtf(_ res: ChatResponse) async -> (String, NSENotification)? {
         if let file = cItem.autoReceiveFile() {
             cItem = autoReceiveFile(file, encrypted: cItem.encryptLocalFile) ?? cItem
         }
-        let ntf: NSENotification = cInfo.ntfsEnabled ? .nse(notification: createMessageReceivedNtf(user, cInfo, cItem)) : .empty
+        let ntf: NSENotification = cInfo.ntfsEnabled ? .nse(createMessageReceivedNtf(user, cInfo, cItem)) : .empty
         return cItem.showNotification ? (aChatItem.chatId, ntf) : nil
     case let .rcvFileSndCancelled(_, aChatItem, _):
         cleanupFile(aChatItem)
@@ -292,10 +527,15 @@ func receivedMsgNtf(_ res: ChatResponse) async -> (String, NSENotification)? {
         // Do not post it without CallKit support, iOS will stop launching the app without showing CallKit
         return (
             invitation.contact.id,
-            useCallKit() ? .callkit(invitation: invitation) : .nse(notification: createCallInvitationNtf(invitation))
+            useCallKit() ? .callkit(invitation) : .nse(createCallInvitationNtf(invitation))
         )
+    case let .ntfMessage(_, connEntity, ntfMessage):
+        return if let id = connEntity.id { (id, .msgInfo(ntfMessage)) } else { nil }
+    case .chatSuspended:
+        chatSuspended()
+        return nil
     default:
-        logger.debug("NotificationService processReceivedMsg ignored event: \(res.responseType)")
+        logger.debug("NotificationService receivedMsgNtf ignored event: \(res.responseType)")
         return nil
     }
 }
@@ -334,6 +574,21 @@ func apiStartChat() throws -> Bool {
     }
 }
 
+func apiActivateChat() -> Bool {
+    chatReopenStore()
+    let r = sendSimpleXCmd(.apiActivateChat(restoreChat: false))
+    if case .cmdOk = r { return true }
+    logger.error("NotificationService apiActivateChat error: \(String(describing: r))")
+    return false
+}
+
+func apiSuspendChat(timeoutMicroseconds: Int) -> Bool {
+    let r = sendSimpleXCmd(.apiSuspendChat(timeoutMicroseconds: timeoutMicroseconds))
+    if case .cmdOk = r { return true }
+    logger.error("NotificationService apiSuspendChat error: \(String(describing: r))")
+    return false
+}
+
 func apiSetTempFolder(tempFolder: String) throws {
     let r = sendSimpleXCmd(.setTempFolder(tempFolder: tempFolder))
     if case .cmdOk = r { return }
@@ -364,8 +619,8 @@ func apiGetNtfMessage(nonce: String, encNtfInfo: String) -> NtfMessages? {
         return nil
     }
     let r = sendSimpleXCmd(.apiGetNtfMessage(nonce: nonce, encNtfInfo: encNtfInfo))
-    if case let .ntfMessages(user, connEntity, msgTs, ntfMessages) = r, let user = user {
-        return NtfMessages(user: user, connEntity: connEntity, msgTs: msgTs, ntfMessages: ntfMessages)
+    if case let .ntfMessages(user, connEntity_, msgTs, ntfMessages) = r, let user = user {
+        return NtfMessages(user: user, connEntity_: connEntity_, msgTs: msgTs, ntfMessages: ntfMessages)
     } else if case let .chatCmdError(_, error) = r {
         logger.debug("apiGetNtfMessage error response: \(String.init(describing: error))")
     } else {
@@ -405,11 +660,11 @@ func setNetworkConfig(_ cfg: NetCfg) throws {
 
 struct NtfMessages {
     var user: User
-    var connEntity: ConnectionEntity?
+    var connEntity_: ConnectionEntity?
     var msgTs: Date?
     var ntfMessages: [NtfMsgInfo]
 
     var ntfsEnabled: Bool {
-        user.showNotifications && (connEntity?.ntfsEnabled ?? false)
+        user.showNotifications && (connEntity_?.ntfsEnabled ?? false)
     }
 }