Merge branch 'draft-improvements'

submodules/Postbox/Sources/AllTypingDraftsView.swift

@@ -0,0 +1,49 @@
+import Foundation
+
+final class MutableAllTypingDraftsView: MutablePostboxView {
+    fileprivate var keys: Set<PeerAndThreadId>
+
+    init(postbox: PostboxImpl) {
+        self.keys = Set(postbox.currentTypingDrafts.keys)
+    }
+
+    func replay(postbox: PostboxImpl, transaction: PostboxTransaction) -> Bool {
+        if transaction.updatedTypingDrafts.isEmpty {
+            return false
+        }
+        var updated = false
+        for (key, update) in transaction.updatedTypingDrafts {
+            if update.value != nil {
+                if self.keys.insert(key).inserted {
+                    updated = true
+                }
+            } else {
+                if self.keys.remove(key) != nil {
+                    updated = true
+                }
+            }
+        }
+        return updated
+    }
+
+    func refreshDueToExternalTransaction(postbox: PostboxImpl) -> Bool {
+        let new = Set(postbox.currentTypingDrafts.keys)
+        if new == self.keys {
+            return false
+        }
+        self.keys = new
+        return true
+    }
+
+    func immutableView() -> PostboxView {
+        return AllTypingDraftsView(self)
+    }
+}
+
+public final class AllTypingDraftsView: PostboxView {
+    public let keys: Set<PeerAndThreadId>
+
+    init(_ view: MutableAllTypingDraftsView) {
+        self.keys = view.keys
+    }
+}

submodules/Postbox/Sources/MessageHistoryView.swift

@@ -1560,13 +1560,15 @@ public final class MessageHistoryView: PostboxView {
         }
         
         if !self.holeLater, let typingDraft = mutableView.typingDraft {
-            entries.append(MessageHistoryEntry(
+            let newEntry = MessageHistoryEntry(
                 message: typingDraft,
                 isRead: false,
                 location: nil,
                 monthLocation: nil,
                 attributes: MutableMessageHistoryEntryAttributes(authorIsContact: false)
-            ))
+            )
+            entries.append(newEntry)
+            entries.sort()
         }
         
         self.entries = entries

submodules/Postbox/Sources/TypingDraftsView.swift

@@ -0,0 +1,98 @@
+import Foundation
+
+final class MutableTypingDraftsView: MutablePostboxView {
+    fileprivate let peerAndThreadId: PeerAndThreadId
+    fileprivate var typingDraft: Message?
+    
+    init(postbox: PostboxImpl, peerAndThreadId: PeerAndThreadId) {
+        self.peerAndThreadId = peerAndThreadId
+        
+        self.reload(postbox: postbox)
+    }
+    
+    private func reload(postbox: PostboxImpl) {
+        if let typingDraft = postbox.currentTypingDrafts[self.peerAndThreadId] {
+            self.typingDraft = self.renderTypingDraft(postbox: postbox, typingDraft: typingDraft)
+        } else {
+            self.typingDraft = nil
+        }
+    }
+    
+    func replay(postbox: PostboxImpl, transaction: PostboxTransaction) -> Bool {
+        var updated = false
+        
+        if let typingDraftUpdate = transaction.updatedTypingDrafts[self.peerAndThreadId] {
+            if let typingDraft = typingDraftUpdate.value {
+                self.typingDraft = self.renderTypingDraft(postbox: postbox, typingDraft: typingDraft)
+            } else {
+                self.typingDraft = nil
+            }
+            updated = true
+        }
+        
+        return updated
+    }
+
+    private func renderTypingDraft(postbox: PostboxImpl, typingDraft: PostboxImpl.TypingDraft) -> Message? {
+        guard let peer = postbox.peerTable.get(self.peerAndThreadId.peerId), let author = postbox.peerTable.get(typingDraft.authorId) else {
+            return nil
+        }
+        
+        var peers = SimpleDictionary<PeerId, Peer>()
+        peers[peer.id] = peer
+        peers[author.id] = author
+        
+        var associatedThreadInfo: Message.AssociatedThreadInfo?
+        if let threadId = typingDraft.threadId, let data = postbox.messageHistoryThreadIndexTable.get(peerId: self.peerAndThreadId.peerId, threadId: threadId) {
+            associatedThreadInfo = postbox.seedConfiguration.decodeMessageThreadInfo(data.data)
+        }
+        
+        return Message(
+            stableId: typingDraft.stableId,
+            stableVersion: typingDraft.stableVersion,
+            id: MessageId(
+                peerId: self.peerAndThreadId.peerId,
+                namespace: 1,
+                id: Int32.max - 50000),
+            globallyUniqueId: nil,
+            groupingKey: nil,
+            groupInfo: nil,
+            threadId: typingDraft.threadId,
+            timestamp: typingDraft.timestamp,
+            flags: [.Incoming],
+            tags: [],
+            globalTags: [],
+            localTags: [],
+            customTags: [],
+            forwardInfo: nil,
+            author: author,
+            text: typingDraft.text,
+            attributes: typingDraft.attributes,
+            media: [],
+            peers: peers,
+            associatedMessages: SimpleDictionary(),
+            associatedMessageIds: [],
+            associatedMedia: [:],
+            associatedThreadInfo: associatedThreadInfo,
+            associatedStories: [:]
+        )
+    }
+
+    func refreshDueToExternalTransaction(postbox: PostboxImpl) -> Bool {
+        self.reload(postbox: postbox)
+        
+        return true
+    }
+    
+    func immutableView() -> PostboxView {
+        return TypingDraftsView(self)
+    }
+}
+
+public final class TypingDraftsView: PostboxView {
+    public let typingDraft: Message?
+    
+    init(_ view: MutableTypingDraftsView) {
+        self.typingDraft = view.typingDraft
+    }
+}

submodules/Postbox/Sources/Views.swift

@@ -102,6 +102,8 @@ public enum PostboxViewKey: Hashable {
     case savedMessagesStats(peerId: PeerId)
     case chatInterfaceState(peerId: PeerId)
     case historyView(HistoryView)
+    case typingDrafts(PeerAndThreadId)
+    case allTypingDrafts
 
     public func hash(into hasher: inout Hasher) {
         switch self {
@@ -187,7 +189,7 @@ public enum PostboxViewKey: Hashable {
         case let .topChatMessage(peerIds):
             hasher.combine(peerIds)
         case .contacts:
-            hasher.combine(16)
+            hasher.combine(24)
         case let .deletedMessages(peerId):
             hasher.combine(peerId)
         case let .notice(key):
@@ -228,6 +230,11 @@ public enum PostboxViewKey: Hashable {
             hasher.combine(20)
             hasher.combine(historyView.peerId)
             hasher.combine(historyView.threadId)
+        case let .typingDrafts(peerId):
+            hasher.combine(23)
+            hasher.combine(peerId)
+        case .allTypingDrafts:
+            hasher.combine(22)
         }
     }
     
@@ -545,6 +552,18 @@ public enum PostboxViewKey: Hashable {
             } else {
                 return false
             }
+        case let .typingDrafts(peerId):
+            if case .typingDrafts(peerId) = rhs {
+                return true
+            } else {
+                return false
+            }
+        case .allTypingDrafts:
+            if case .allTypingDrafts = rhs {
+                return true
+            } else {
+                return false
+            }
         }
     }
 }
@@ -672,5 +691,9 @@ func postboxViewForKey(postbox: PostboxImpl, key: PostboxViewKey) -> MutablePost
             topTaggedMessages: [:],
             additionalDatas: []
         )
+    case let .typingDrafts(peerId):
+        return MutableTypingDraftsView(postbox: postbox, peerAndThreadId: peerId)
+    case .allTypingDrafts:
+        return MutableAllTypingDraftsView(postbox: postbox)
     }
 }

submodules/TelegramCore/Sources/State/AccountStateManagementUtils.swift

@@ -4018,7 +4018,7 @@ func replayFinalState(
         }
         
         case update(Update)
-        case cancel
+        case cancel(updatedTimestamp: Int32)
     }
     
     var liveTypingDraftUpdates: [PeerAndThreadId: [LiveTypingDraftUpdate]] = [:]
@@ -4236,11 +4236,11 @@ func replayFinalState(
                         let allKey = PeerAndThreadId(peerId: chatPeerId, threadId: nil)
                         
                         if liveTypingDraftUpdates[key] != nil {
-                            liveTypingDraftUpdates[key] = [.cancel]
-                            liveTypingDraftUpdates[allKey] = [.cancel]
+                            liveTypingDraftUpdates[key] = [.cancel(updatedTimestamp: message.timestamp)]
+                            liveTypingDraftUpdates[allKey] = [.cancel(updatedTimestamp: message.timestamp)]
                         } else if let currentDraft = transaction.getCurrentTypingDraft(location: key) {
-                            liveTypingDraftUpdates[key] = [.cancel]
-                            liveTypingDraftUpdates[allKey] = [.cancel]
+                            liveTypingDraftUpdates[key] = [.cancel(updatedTimestamp: message.timestamp)]
+                            liveTypingDraftUpdates[allKey] = [.cancel(updatedTimestamp: message.timestamp)]
                             messages[i] = messages[i].withUpdatedCustomStableId(currentDraft.stableId)
                         }
                     }
@@ -6080,6 +6080,23 @@ func replayFinalState(
     }
     
     if !liveTypingDraftUpdates.isEmpty {
+        for (key, updates) in liveTypingDraftUpdates {
+            if key.threadId == nil {
+                var maxCancelledTimestamp: Int32?
+                for update in updates {
+                    if case let .cancel(updatedTimestamp) = update {
+                        if let current = maxCancelledTimestamp {
+                            maxCancelledTimestamp = max(current, updatedTimestamp)
+                        } else {
+                            maxCancelledTimestamp = updatedTimestamp
+                        }
+                    }
+                }
+                if let maxCancelledTimestamp {
+                    transaction.offsetPendingMessagesTimestamps(lowerBound: MessageId(peerId: key.peerId, namespace: Namespaces.Message.Local, id: 1), excludeIds: Set(), timestamp: maxCancelledTimestamp)
+                }
+            }
+        }
         transaction.combineTypingDrafts(locations: Set(liveTypingDraftUpdates.keys), update: { key, current in
             guard let update = liveTypingDraftUpdates[key]?.max(by: { lhs, rhs in
                 switch lhs {
@@ -6105,6 +6122,11 @@ func replayFinalState(
                 if let current, current.id == update.id {
                     timestamp = current.timestamp
                 }
+                if current == nil {
+                    if let index = transaction.getTopPeerMessageIndex(peerId: key.peerId) {
+                        timestamp = max(timestamp, index.timestamp)
+                    }
+                }
                 return (
                     update.id,
                     Namespaces.Message.Cloud,

submodules/TelegramCore/Sources/State/PendingMessageManager.swift

@@ -31,6 +31,8 @@ private enum PendingMessageState {
     case waitingForUploadToStart(groupId: Int64?, upload: Signal<PendingMessageUploadedContentResult, PendingMessageUploadError>)
     case uploading(groupId: Int64?)
     case waitingToBeSent(groupId: Int64?, content: PendingMessageUploadedContentAndReuploadInfo)
+    case waitingForSendGate(groupId: Int64?, content: PendingMessageUploadedContentAndReuploadInfo)
+    case waitingForForwardSendGate
     case sending(groupId: Int64?)
     case waitingForNewTopic(message: Message)
     
@@ -46,6 +48,10 @@ private enum PendingMessageState {
             return groupId
         case let .waitingToBeSent(groupId, _):
             return groupId
+        case let .waitingForSendGate(groupId, _):
+            return groupId
+        case .waitingForForwardSendGate:
+            return nil
         case let .sending(groupId):
             return groupId
         case let .waitingForNewTopic(message):
@@ -230,6 +236,9 @@ public final class PendingMessageManager {
     
     private var messageContexts: [MessageId: PendingMessageContext] = [:]
     private var pendingMessageIds = Set<MessageId>()
+    private var liveTypingDraftKeys: Set<PeerAndThreadId> = []
+    private let allTypingDraftsDisposable = MetaDisposable()
+    private var forwardSendGateGroups: [PeerAndThreadId: [[(PendingMessageContext, Message, ForwardSourceInfoAttribute)]]] = [:]
     private let beginSendingMessagesDisposables = DisposableSet()
     
     private var newTopicDisposables: [PeerId: Disposable] = [:]
@@ -250,6 +259,14 @@ public final class PendingMessageManager {
         self.localInputActivityManager = localInputActivityManager
         self.messageMediaPreuploadManager = messageMediaPreuploadManager
         self.revalidationContext = revalidationContext
+
+        let queue = self.queue
+        self.allTypingDraftsDisposable.set(
+            (postbox.combinedView(keys: [.allTypingDrafts])
+            |> deliverOn(queue)).start(next: { [weak self] view in
+                self?.handleLiveTypingDraftsUpdate(view)
+            })
+        )
     }
     
     deinit {
@@ -257,8 +274,117 @@ public final class PendingMessageManager {
         for (_, disposable) in self.newTopicDisposables {
             disposable.dispose()
         }
+        self.allTypingDraftsDisposable.dispose()
     }
     
+    private func handleLiveTypingDraftsUpdate(_ combined: CombinedView) {
+        assert(self.queue.isCurrent())
+
+        let new: Set<PeerAndThreadId>
+        if let view = combined.views[.allTypingDrafts] as? AllTypingDraftsView {
+            new = view.keys
+        } else {
+            new = []
+        }
+        let cleared = self.liveTypingDraftKeys.subtracting(new)
+        self.liveTypingDraftKeys = new
+        for key in cleared {
+            self.drainSendGate(key: key)
+        }
+    }
+
+    private func shouldGateSend(messageId: MessageId, threadId: Int64?) -> Bool {
+        if messageId.namespace == Namespaces.Message.ScheduledCloud {
+            return false
+        }
+        if messageId.peerId.namespace == Namespaces.Peer.SecretChat {
+            return false
+        }
+        if messageId.peerId == self.accountPeerId {
+            return false
+        }
+        if threadId == Message.newTopicThreadId {
+            return false
+        }
+        return true
+    }
+
+    private func isSendGateOpen(for key: PeerAndThreadId) -> Bool {
+        return !self.liveTypingDraftKeys.contains(key)
+    }
+
+    private func drainSendGate(key: PeerAndThreadId) {
+        assert(self.queue.isCurrent())
+
+        // (1) Single-message drain: snapshot then commit in messageId.id order.
+        var singleDrains: [(context: PendingMessageContext, messageId: MessageId, content: PendingMessageUploadedContentAndReuploadInfo)] = []
+        for (id, context) in self.messageContexts {
+            if id.peerId != key.peerId {
+                continue
+            }
+            if context.threadId != key.threadId {
+                continue
+            }
+            if case let .waitingForSendGate(groupId, content) = context.state, groupId == nil {
+                singleDrains.append((context, id, content))
+            }
+        }
+        singleDrains.sort(by: { $0.messageId.id < $1.messageId.id })
+        for entry in singleDrains {
+            self.commitSendingSingleMessage(messageContext: entry.context, messageId: entry.messageId, content: entry.content)
+        }
+
+        // (2) Grouped-album drain: collect distinct groupIds whose members match the key,
+        // iterate ascending by min messageId.id, fire commitSendingMessageGroup.
+        var groupKeys: [(groupId: Int64, minMessageId: Int32)] = []
+        var seenGroupIds = Set<Int64>()
+        for (id, context) in self.messageContexts {
+            if id.peerId != key.peerId {
+                continue
+            }
+            if context.threadId != key.threadId {
+                continue
+            }
+            if case let .waitingForSendGate(groupId, _) = context.state, let groupId = groupId {
+                if !seenGroupIds.contains(groupId) {
+                    seenGroupIds.insert(groupId)
+                    groupKeys.append((groupId, id.id))
+                } else {
+                    if let index = groupKeys.firstIndex(where: { $0.groupId == groupId }), id.id < groupKeys[index].minMessageId {
+                        groupKeys[index].minMessageId = id.id
+                    }
+                }
+            }
+        }
+        groupKeys.sort(by: { $0.minMessageId < $1.minMessageId })
+        for (groupId, _) in groupKeys {
+            if let data = self.dataForPendingMessageGroup(groupId) {
+                self.commitSendingMessageGroup(groupId: groupId, messages: data)
+            }
+        }
+
+        // (3) Forward drain: pop parked groups for this key in FIFO order; fire each.
+        if let parkedGroups = self.forwardSendGateGroups.removeValue(forKey: key) {
+            for messages in parkedGroups {
+                for (context, _, _) in messages {
+                    context.state = .sending(groupId: nil)
+                }
+                let sendMessage: Signal<PendingMessageResult, NoError> = self.sendGroupMessagesContent(network: self.network, postbox: self.postbox, stateManager: self.stateManager, accountPeerId: self.accountPeerId, group: messages.map { data in
+                    let (_, message, forwardInfo) = data
+                    return (message.id, PendingMessageUploadedContentAndReuploadInfo(content: .forward(forwardInfo), reuploadInfo: nil, cacheReferenceKey: nil))
+                })
+                |> map { _ -> PendingMessageResult in
+                    return .progress(1.0)
+                }
+                messages[0].0.sendDisposable.set((sendMessage
+                |> deliverOn(self.queue)).start())
+            }
+        }
+
+        self.updateWaitingUploads(peerId: key.peerId)
+        self.updatePendingMediaUploads()
+    }
+
     private func updatePendingMediaUploads() {
         assert(self.queue.isCurrent())
         
@@ -321,7 +447,29 @@ public final class PendingMessageManager {
                     }
                 }
             }
-            
+
+            if !removedMessageIds.isEmpty && !self.forwardSendGateGroups.isEmpty {
+                for key in Array(self.forwardSendGateGroups.keys) {
+                    guard let parkedGroups = self.forwardSendGateGroups[key] else {
+                        continue
+                    }
+                    var rebuilt: [[(PendingMessageContext, Message, ForwardSourceInfoAttribute)]] = []
+                    for group in parkedGroups {
+                        let filtered = group.filter { entry in
+                            return !removedMessageIds.contains(entry.1.id)
+                        }
+                        if !filtered.isEmpty {
+                            rebuilt.append(filtered)
+                        }
+                    }
+                    if rebuilt.isEmpty {
+                        self.forwardSendGateGroups.removeValue(forKey: key)
+                    } else {
+                        self.forwardSendGateGroups[key] = rebuilt
+                    }
+                }
+            }
+
             if !addedMessageIds.isEmpty {
                 Logger.shared.log("PendingMessageManager", "added messages: \(addedMessageIds)")
                 self.beginSendingMessages(Array(addedMessageIds).sorted())
@@ -715,11 +863,24 @@ public final class PendingMessageManager {
                         if messages.isEmpty {
                             continue
                         }
-                        
+
+                        let firstMessage = messages[0].1
+                        let key = PeerAndThreadId(peerId: firstMessage.id.peerId, threadId: firstMessage.threadId)
+                        if strongSelf.shouldGateSend(messageId: firstMessage.id, threadId: firstMessage.threadId) && !strongSelf.isSendGateOpen(for: key) {
+                            for (context, _, _) in messages {
+                                context.state = .waitingForForwardSendGate
+                            }
+                            if strongSelf.forwardSendGateGroups[key] == nil {
+                                strongSelf.forwardSendGateGroups[key] = []
+                            }
+                            strongSelf.forwardSendGateGroups[key]!.append(messages)
+                            continue
+                        }
+
                         for (context, _, _) in messages {
                             context.state = .sending(groupId: nil)
                         }
-                        
+
                         let sendMessage: Signal<PendingMessageResult, NoError> = strongSelf.sendGroupMessagesContent(network: strongSelf.network, postbox: strongSelf.postbox, stateManager: strongSelf.stateManager, accountPeerId: strongSelf.accountPeerId, group: messages.map { data in
                             let (_, message, forwardInfo) = data
                             return (message.id, PendingMessageUploadedContentAndReuploadInfo(content: .forward(forwardInfo), reuploadInfo: nil, cacheReferenceKey: nil))
@@ -739,7 +900,12 @@ public final class PendingMessageManager {
         if let groupId = groupId {
             messageContext.state = .waitingToBeSent(groupId: groupId, content: content)
         } else {
-            self.commitSendingSingleMessage(messageContext: messageContext, messageId: messageId, content: content)
+            let key = PeerAndThreadId(peerId: messageId.peerId, threadId: messageContext.threadId)
+            if self.shouldGateSend(messageId: messageId, threadId: messageContext.threadId) && !self.isSendGateOpen(for: key) {
+                messageContext.state = .waitingForSendGate(groupId: nil, content: content)
+            } else {
+                self.commitSendingSingleMessage(messageContext: messageContext, messageId: messageId, content: content)
+            }
         }
         self.updatePendingMediaUploads()
     }
@@ -757,6 +923,8 @@ public final class PendingMessageManager {
             switch context.state {
             case .none:
                 continue loop
+            case .waitingForForwardSendGate:
+                continue loop
             case let .collectingInfo(message):
                 if message.groupingKey == groupId {
                     return nil
@@ -773,6 +941,10 @@ public final class PendingMessageManager {
                 if contextGroupId == groupId {
                     result.append((context, id, content))
                 }
+            case let .waitingForSendGate(contextGroupId, content):
+                if contextGroupId == groupId {
+                    result.append((context, id, content))
+                }
             case let .sending(contextGroupId):
                 if contextGroupId == groupId {
                     return nil
@@ -792,6 +964,16 @@ public final class PendingMessageManager {
     }
     
     private func commitSendingMessageGroup(groupId: Int64, messages: [(messageContext: PendingMessageContext, messageId: MessageId, content: PendingMessageUploadedContentAndReuploadInfo)]) {
+        let firstMessageId = messages[0].messageId
+        let firstThreadId = messages[0].messageContext.threadId
+        let key = PeerAndThreadId(peerId: firstMessageId.peerId, threadId: firstThreadId)
+        if self.shouldGateSend(messageId: firstMessageId, threadId: firstThreadId) && !self.isSendGateOpen(for: key) {
+            for entry in messages {
+                entry.messageContext.state = .waitingForSendGate(groupId: groupId, content: entry.content)
+            }
+            return
+        }
+
         for (context, _, _) in messages {
             context.state = .sending(groupId: groupId)
         }