import Foundation import Postbox import SwiftSignalKit import TelegramApi import MtProtoKit import SyncCore public enum CallListViewType { case all case missed } public enum CallListViewEntry { case message(Message, [Message]) case hole(MessageIndex) } public final class CallListView { public let entries: [CallListViewEntry] public let earlier: MessageIndex? public let later: MessageIndex? init(entries: [CallListViewEntry], earlier: MessageIndex?, later: MessageIndex?) { self.entries = entries self.earlier = earlier self.later = later } } private func pendingWebpages(entries: [MessageHistoryEntry]) -> (Set, [MessageId: (MediaId, String)]) { var messageIds = Set() var localWebpages: [MessageId: (MediaId, String)] = [:] for entry in entries { for media in entry.message.media { if let media = media as? TelegramMediaWebpage { if case let .Pending(_, url) = media.content { messageIds.insert(entry.message.id) if let url = url, media.webpageId.namespace == Namespaces.Media.LocalWebpage { localWebpages[entry.message.id] = (media.webpageId, url) } } break } } } return (messageIds, localWebpages) } private func pollMessages(entries: [MessageHistoryEntry]) -> (Set, [MessageId: Message]) { var messageIds = Set() var messages: [MessageId: Message] = [:] for entry in entries { for media in entry.message.media { if let poll = media as? TelegramMediaPoll, poll.pollId.namespace == Namespaces.Media.CloudPoll, entry.message.id.namespace == Namespaces.Message.Cloud, !poll.isClosed { messageIds.insert(entry.message.id) messages[entry.message.id] = entry.message break } } } return (messageIds, messages) } private func fetchWebpage(account: Account, messageId: MessageId) -> Signal { return account.postbox.loadedPeerWithId(messageId.peerId) |> take(1) |> mapToSignal { peer in if let inputPeer = apiInputPeer(peer) { let isScheduledMessage = Namespaces.Message.allScheduled.contains(messageId.namespace) let messages: Signal if isScheduledMessage { messages = account.network.request(Api.functions.messages.getScheduledMessages(peer: inputPeer, id: [messageId.id])) } else { switch inputPeer { case let .inputPeerChannel(channelId, accessHash): messages = account.network.request(Api.functions.channels.getMessages(channel: Api.InputChannel.inputChannel(channelId: channelId, accessHash: accessHash), id: [Api.InputMessage.inputMessageID(id: messageId.id)])) default: messages = account.network.request(Api.functions.messages.getMessages(id: [Api.InputMessage.inputMessageID(id: messageId.id)])) } } return messages |> retryRequest |> mapToSignal { result in let messages: [Api.Message] let chats: [Api.Chat] let users: [Api.User] switch result { case let .messages(messages: apiMessages, chats: apiChats, users: apiUsers): messages = apiMessages chats = apiChats users = apiUsers case let .messagesSlice(_, _, _, _, messages: apiMessages, chats: apiChats, users: apiUsers): messages = apiMessages chats = apiChats users = apiUsers case let .channelMessages(_, _, _, _, apiMessages, apiChats, apiUsers): messages = apiMessages chats = apiChats users = apiUsers case .messagesNotModified: messages = [] chats = [] users = [] } return account.postbox.transaction { transaction -> Void in var peers: [Peer] = [] var peerPresences: [PeerId: PeerPresence] = [:] for chat in chats { if let groupOrChannel = mergeGroupOrChannel(lhs: transaction.getPeer(chat.peerId), rhs: chat) { peers.append(groupOrChannel) } } for apiUser in users { if let user = TelegramUser.merge(transaction.getPeer(apiUser.peerId) as? TelegramUser, rhs: apiUser) { peers.append(user) if let presence = TelegramUserPresence(apiUser: apiUser) { peerPresences[user.id] = presence } } } for message in messages { if let storeMessage = StoreMessage(apiMessage: message, namespace: isScheduledMessage ? Namespaces.Message.ScheduledCloud : Namespaces.Message.Cloud) { var webpage: TelegramMediaWebpage? for media in storeMessage.media { if let media = media as? TelegramMediaWebpage { webpage = media } } if let webpage = webpage { updateMessageMedia(transaction: transaction, id: webpage.webpageId, media: webpage) } else { if let previousMessage = transaction.getMessage(messageId) { for media in previousMessage.media { if let media = media as? TelegramMediaWebpage { updateMessageMedia(transaction: transaction, id: media.webpageId, media: nil) break } } } } break } } updatePeers(transaction: transaction, peers: peers, update: { _, updated -> Peer in return updated }) updatePeerPresences(transaction: transaction, accountPeerId: account.peerId, peerPresences: peerPresences) } } } else { return .complete() } } } private func fetchPoll(account: Account, messageId: MessageId) -> Signal { return account.postbox.loadedPeerWithId(messageId.peerId) |> take(1) |> mapToSignal { peer -> Signal in guard let inputPeer = apiInputPeer(peer) else { return .complete() } return account.network.request(Api.functions.messages.getPollResults(peer: inputPeer, msgId: messageId.id)) |> map(Optional.init) |> `catch` { _ -> Signal in return .single(nil) } |> mapToSignal { updates -> Signal in if let updates = updates { account.stateManager.addUpdates(updates) } return .complete() } } } private func wrappedHistoryViewAdditionalData(chatLocation: ChatLocationInput, additionalData: [AdditionalMessageHistoryViewData]) -> [AdditionalMessageHistoryViewData] { var result = additionalData switch chatLocation { case let .peer(peerId): if peerId.namespace == Namespaces.Peer.CloudChannel { if result.firstIndex(where: { if case .peerChatState = $0 { return true } else { return false } }) == nil { result.append(.peerChatState(peerId)) } } case let .external(peerId, _, _): if peerId.namespace == Namespaces.Peer.CloudChannel { if result.firstIndex(where: { if case .peerChatState = $0 { return true } else { return false } }) == nil { result.append(.peerChatState(peerId)) } } } return result } private final class PeerCachedDataContext { var viewIds = Set() var timestamp: Double? var hasCachedData: Bool = false let disposable = MetaDisposable() deinit { self.disposable.dispose() } } private final class CachedChannelParticipantsContext { var subscribers = Bag() var timestamp: Double? let disposable = MetaDisposable() deinit { self.disposable.dispose() } } private final class ChannelPollingContext { var subscribers = Bag() let disposable = MetaDisposable() deinit { self.disposable.dispose() } } private final class FeaturedStickerPacksContext { var subscribers = Bag() let disposable = MetaDisposable() var timestamp: Double? deinit { self.disposable.dispose() } } private struct ViewCountContextState { struct ReplyInfo { var commentsPeerId: PeerId? var maxReadIncomingMessageId: MessageId? var maxMessageId: MessageId? } var timestamp: Int32 var clientId: Int32 var result: ReplyInfo? func isStillValidFor(_ other: ViewCountContextState) -> Bool { if other.timestamp > self.timestamp + 30 { return false } if other.clientId > self.clientId { return false } return true } } public final class AccountViewTracker { weak var account: Account? private let queue = Queue() private var nextViewId: Int32 = 0 private var viewPendingWebpageMessageIds: [Int32: Set] = [:] private var viewPollMessageIds: [Int32: Set] = [:] private var pendingWebpageMessageIds: [MessageId: Int] = [:] private var pollMessageIds: [MessageId: Int] = [:] private var webpageDisposables: [MessageId: Disposable] = [:] private var pollDisposables: [MessageId: Disposable] = [:] private var viewVisibleCallListHoleIds: [Int32: Set] = [:] private var visibleCallListHoleIds: [MessageIndex: Int] = [:] private var visibleCallListHoleDisposables: [MessageIndex: Disposable] = [:] private var updatedViewCountMessageIdsAndTimestamps: [MessageId: ViewCountContextState] = [:] private var nextUpdatedViewCountDisposableId: Int32 = 0 private var updatedViewCountDisposables = DisposableDict() private var updatedReactionsMessageIdsAndTimestamps: [MessageId: Int32] = [:] private var nextUpdatedReactionsDisposableId: Int32 = 0 private var updatedReactionsDisposables = DisposableDict() private var updatedSeenLiveLocationMessageIdsAndTimestamps: [MessageId: Int32] = [:] private var nextSeenLiveLocationDisposableId: Int32 = 0 private var seenLiveLocationDisposables = DisposableDict() private var updatedUnsupportedMediaMessageIdsAndTimestamps: [MessageId: Int32] = [:] private var refreshSecretChatMediaMessageIdsAndTimestamps: [MessageId: Int32] = [:] private var nextUpdatedUnsupportedMediaDisposableId: Int32 = 0 private var updatedUnsupportedMediaDisposables = DisposableDict() private var updatedSeenPersonalMessageIds = Set() private var cachedDataContexts: [PeerId: PeerCachedDataContext] = [:] private var cachedChannelParticipantsContexts: [PeerId: CachedChannelParticipantsContext] = [:] private var channelPollingContexts: [PeerId: ChannelPollingContext] = [:] private var featuredStickerPacksContext: FeaturedStickerPacksContext? let chatHistoryPreloadManager: ChatHistoryPreloadManager private let historyViewStateValidationContexts: HistoryViewStateValidationContexts public var orderedPreloadMedia: Signal<[ChatHistoryPreloadMediaItem], NoError> { return self.chatHistoryPreloadManager.orderedMedia } private let externallyUpdatedPeerIdDisposable = MetaDisposable() public let chatListPreloadItems = Promise<[ChatHistoryPreloadItem]>([]) init(account: Account) { self.account = account self.historyViewStateValidationContexts = HistoryViewStateValidationContexts(queue: self.queue, postbox: account.postbox, network: account.network, accountPeerId: account.peerId) self.chatHistoryPreloadManager = ChatHistoryPreloadManager(postbox: account.postbox, network: account.network, accountPeerId: account.peerId, networkState: account.networkState, preloadItemsSignal: self.chatListPreloadItems.get() |> distinctUntilChanged) self.externallyUpdatedPeerIdDisposable.set((account.stateManager.externallyUpdatedPeerIds |> deliverOn(self.queue)).start(next: { [weak self] peerIds in guard let strongSelf = self else { return } for (peerId, _) in strongSelf.cachedDataContexts { if peerIds.contains(peerId) { strongSelf.forceUpdateCachedPeerData(peerId: peerId) } } })) } deinit { self.updatedViewCountDisposables.dispose() self.updatedReactionsDisposables.dispose() self.externallyUpdatedPeerIdDisposable.dispose() } func reset() { self.queue.async { self.cachedDataContexts.removeAll() } } private func updatePendingWebpages(viewId: Int32, messageIds: Set, localWebpages: [MessageId: (MediaId, String)]) { self.queue.async { var addedMessageIds: [MessageId] = [] var removedMessageIds: [MessageId] = [] let viewMessageIds: Set = self.viewPendingWebpageMessageIds[viewId] ?? Set() let viewAddedMessageIds = messageIds.subtracting(viewMessageIds) let viewRemovedMessageIds = viewMessageIds.subtracting(messageIds) for messageId in viewAddedMessageIds { if let count = self.pendingWebpageMessageIds[messageId] { self.pendingWebpageMessageIds[messageId] = count + 1 } else { self.pendingWebpageMessageIds[messageId] = 1 addedMessageIds.append(messageId) } } for messageId in viewRemovedMessageIds { if let count = self.pendingWebpageMessageIds[messageId] { if count == 1 { self.pendingWebpageMessageIds.removeValue(forKey: messageId) removedMessageIds.append(messageId) } else { self.pendingWebpageMessageIds[messageId] = count - 1 } } else { assertionFailure() } } if messageIds.isEmpty { self.viewPendingWebpageMessageIds.removeValue(forKey: viewId) } else { self.viewPendingWebpageMessageIds[viewId] = messageIds } for messageId in removedMessageIds { if let disposable = self.webpageDisposables.removeValue(forKey: messageId) { disposable.dispose() } } if let account = self.account { for messageId in addedMessageIds { if self.webpageDisposables[messageId] == nil { if let (_, url) = localWebpages[messageId] { self.webpageDisposables[messageId] = (webpagePreview(account: account, url: url) |> mapToSignal { webpage -> Signal in return account.postbox.transaction { transaction -> Void in if let webpage = webpage { transaction.updateMessage(messageId, update: { currentMessage in let storeForwardInfo = currentMessage.forwardInfo.flatMap(StoreMessageForwardInfo.init) var media = currentMessage.media for i in 0 ..< media.count { if let _ = media[i] as? TelegramMediaWebpage { media[i] = webpage break } } return .update(StoreMessage(id: currentMessage.id, globallyUniqueId: currentMessage.globallyUniqueId, groupingKey: currentMessage.groupingKey, threadId: currentMessage.threadId, timestamp: currentMessage.timestamp, flags: StoreMessageFlags(currentMessage.flags), tags: currentMessage.tags, globalTags: currentMessage.globalTags, localTags: currentMessage.localTags, forwardInfo: storeForwardInfo, authorId: currentMessage.author?.id, text: currentMessage.text, attributes: currentMessage.attributes, media: media)) }) } } }).start(completed: { [weak self] in if let strongSelf = self { strongSelf.queue.async { strongSelf.webpageDisposables.removeValue(forKey: messageId) } } }) } else if messageId.namespace == Namespaces.Message.Cloud { self.webpageDisposables[messageId] = fetchWebpage(account: account, messageId: messageId).start(completed: { [weak self] in if let strongSelf = self { strongSelf.queue.async { strongSelf.webpageDisposables.removeValue(forKey: messageId) } } }) } } else { assertionFailure() } } } } } private func updatePolls(viewId: Int32, messageIds: Set, messages: [MessageId: Message]) { let queue = self.queue self.queue.async { var addedMessageIds: [MessageId] = [] var removedMessageIds: [MessageId] = [] let viewMessageIds: Set = self.viewPollMessageIds[viewId] ?? Set() let viewAddedMessageIds = messageIds.subtracting(viewMessageIds) let viewRemovedMessageIds = viewMessageIds.subtracting(messageIds) for messageId in viewAddedMessageIds { if let count = self.pollMessageIds[messageId] { self.pollMessageIds[messageId] = count + 1 } else { self.pollMessageIds[messageId] = 1 addedMessageIds.append(messageId) } } for messageId in viewRemovedMessageIds { if let count = self.pollMessageIds[messageId] { if count == 1 { self.pollMessageIds.removeValue(forKey: messageId) removedMessageIds.append(messageId) } else { self.pollMessageIds[messageId] = count - 1 } } else { assertionFailure() } } if messageIds.isEmpty { self.viewPollMessageIds.removeValue(forKey: viewId) } else { self.viewPollMessageIds[viewId] = messageIds } for messageId in removedMessageIds { if let disposable = self.pollDisposables.removeValue(forKey: messageId) { disposable.dispose() } } if let account = self.account { for messageId in addedMessageIds { if self.pollDisposables[messageId] == nil { var deadlineTimer: Signal = .single(false) if let message = messages[messageId] { for media in message.media { if let poll = media as? TelegramMediaPoll { if let _ = poll.deadlineTimeout, message.id.namespace == Namespaces.Message.Cloud { let startDate: Int32 if let forwardInfo = message.forwardInfo { startDate = forwardInfo.date } else { startDate = message.timestamp } let timestamp = Int32(CFAbsoluteTimeGetCurrent() + NSTimeIntervalSince1970) let remainingTime = timestamp - startDate - 1 if remainingTime > 0 { deadlineTimer = .single(false) |> then( .single(true) |> suspendAwareDelay(Double(remainingTime), queue: queue) ) } else { deadlineTimer = .single(true) } } } } } let pollSignal: Signal = deadlineTimer |> distinctUntilChanged |> mapToSignal { reachedDeadline -> Signal in if reachedDeadline { var signal = fetchPoll(account: account, messageId: messageId) |> ignoreValues signal = (signal |> then( .complete() |> delay(0.5, queue: Queue.concurrentDefaultQueue()) )) |> restart return signal } else { var signal = fetchPoll(account: account, messageId: messageId) |> ignoreValues signal = (signal |> then( .complete() |> delay(30.0, queue: Queue.concurrentDefaultQueue()) )) |> restart return signal } } self.pollDisposables[messageId] = pollSignal.start() } else { assertionFailure() } } } } } private func updateVisibleCallListHoles(viewId: Int32, holeIds: Set) { self.queue.async { var addedHoleIds: [MessageIndex] = [] var removedHoleIds: [MessageIndex] = [] let viewHoleIds: Set = self.viewVisibleCallListHoleIds[viewId] ?? Set() let viewAddedHoleIds = holeIds.subtracting(viewHoleIds) let viewRemovedHoleIds = viewHoleIds.subtracting(holeIds) for holeId in viewAddedHoleIds { if let count = self.visibleCallListHoleIds[holeId] { self.visibleCallListHoleIds[holeId] = count + 1 } else { self.visibleCallListHoleIds[holeId] = 1 addedHoleIds.append(holeId) } } for holeId in viewRemovedHoleIds { if let count = self.visibleCallListHoleIds[holeId] { if count == 1 { self.visibleCallListHoleIds.removeValue(forKey: holeId) removedHoleIds.append(holeId) } else { self.visibleCallListHoleIds[holeId] = count - 1 } } else { assertionFailure() } } if holeIds.isEmpty { self.viewVisibleCallListHoleIds.removeValue(forKey: viewId) } else { self.viewVisibleCallListHoleIds[viewId] = holeIds } for holeId in removedHoleIds { if let disposable = self.visibleCallListHoleDisposables.removeValue(forKey: holeId) { disposable.dispose() } } if let account = self.account { for holeId in addedHoleIds { if self.visibleCallListHoleDisposables[holeId] == nil { self.visibleCallListHoleDisposables[holeId] = fetchCallListHole(network: account.network, postbox: account.postbox, accountPeerId: account.peerId, holeIndex: holeId).start(completed: { [weak self] in if let strongSelf = self { strongSelf.queue.async { strongSelf.visibleCallListHoleDisposables.removeValue(forKey: holeId) } } }) } else { assertionFailure() } } } } } public struct UpdatedMessageReplyInfo { var timestamp: Int32 var commentsPeerId: PeerId var maxReadIncomingMessageId: MessageId? var maxMessageId: MessageId? } func applyMaxReadIncomingMessageIdForReplyInfo(id: MessageId, maxReadIncomingMessageId: MessageId) { self.queue.async { if var state = self.updatedViewCountMessageIdsAndTimestamps[id], var result = state.result { result.maxReadIncomingMessageId = maxReadIncomingMessageId state.result = result self.updatedViewCountMessageIdsAndTimestamps[id] = state } } } public func replyInfoForMessageId(_ id: MessageId) -> Signal { return Signal { [weak self] subscriber in let state = self?.updatedViewCountMessageIdsAndTimestamps[id] let result = state?.result if let state = state, let result = result, let commentsPeerId = result.commentsPeerId { subscriber.putNext(UpdatedMessageReplyInfo(timestamp: state.timestamp, commentsPeerId: commentsPeerId, maxReadIncomingMessageId: result.maxReadIncomingMessageId, maxMessageId: result.maxMessageId)) } else { subscriber.putNext(nil) } subscriber.putCompletion() return EmptyDisposable } |> runOn(self.queue) } public func updateReplyInfoForMessageId(_ id: MessageId, info: UpdatedMessageReplyInfo) { self.queue.async { [weak self] in guard let strongSelf = self else { return } guard let current = strongSelf.updatedViewCountMessageIdsAndTimestamps[id] else { return } strongSelf.updatedViewCountMessageIdsAndTimestamps[id] = ViewCountContextState(timestamp: Int32(CFAbsoluteTimeGetCurrent()), clientId: current.clientId, result: ViewCountContextState.ReplyInfo(commentsPeerId: info.commentsPeerId, maxReadIncomingMessageId: info.maxReadIncomingMessageId, maxMessageId: info.maxMessageId)) } } public func updateViewCountForMessageIds(messageIds: Set, clientId: Int32) { self.queue.async { var addedMessageIds: [MessageId] = [] let updatedState = ViewCountContextState(timestamp: Int32(CFAbsoluteTimeGetCurrent()), clientId: clientId, result: nil) for messageId in messageIds { let messageTimestamp = self.updatedViewCountMessageIdsAndTimestamps[messageId] if messageTimestamp == nil || !messageTimestamp!.isStillValidFor(updatedState) { self.updatedViewCountMessageIdsAndTimestamps[messageId] = updatedState addedMessageIds.append(messageId) } } if !addedMessageIds.isEmpty { for (peerId, messageIds) in messagesIdsGroupedByPeerId(Set(addedMessageIds)) { let disposableId = self.nextUpdatedViewCountDisposableId self.nextUpdatedViewCountDisposableId += 1 if let account = self.account { let signal: Signal<[MessageId: ViewCountContextState], NoError> = (account.postbox.transaction { transaction -> Signal<[MessageId: ViewCountContextState], NoError> in guard let peer = transaction.getPeer(peerId), let inputPeer = apiInputPeer(peer) else { return .complete() } return account.network.request(Api.functions.messages.getMessagesViews(peer: inputPeer, id: messageIds.map { $0.id }, increment: .boolTrue)) |> map(Optional.init) |> `catch` { _ -> Signal in return .single(nil) } |> mapToSignal { result -> Signal<[MessageId: ViewCountContextState], NoError> in guard case let .messageViews(viewCounts, chats, users)? = result else { return .complete() } return account.postbox.transaction { transaction -> [MessageId: ViewCountContextState] in var peers: [Peer] = [] var peerPresences: [PeerId: PeerPresence] = [:] var resultStates: [MessageId: ViewCountContextState] = [:] for apiUser in users { if let user = TelegramUser.merge(transaction.getPeer(apiUser.peerId) as? TelegramUser, rhs: apiUser) { peers.append(user) if let presence = TelegramUserPresence(apiUser: apiUser) { peerPresences[user.id] = presence } } } for chat in chats { if let groupOrChannel = mergeGroupOrChannel(lhs: transaction.getPeer(chat.peerId), rhs: chat) { peers.append(groupOrChannel) } } updatePeers(transaction: transaction, peers: peers, update: { _, updated -> Peer in return updated }) updatePeerPresences(transaction: transaction, accountPeerId: account.peerId, peerPresences: peerPresences) for i in 0 ..< messageIds.count { if i < viewCounts.count { if case let .messageViews(_, views, forwards, replies) = viewCounts[i] { transaction.updateMessage(messageIds[i], update: { currentMessage in let storeForwardInfo = currentMessage.forwardInfo.flatMap(StoreMessageForwardInfo.init) var attributes = currentMessage.attributes var foundReplies = false var commentsChannelId: PeerId? var recentRepliersPeerIds: [PeerId]? var repliesCount: Int32? var repliesMaxId: Int32? var repliesReadMaxId: Int32? if let replies = replies { switch replies { case let .messageReplies(_, repliesCountValue, _, recentRepliers, channelId, maxId, readMaxId): if let channelId = channelId { commentsChannelId = PeerId(namespace: Namespaces.Peer.CloudChannel, id: channelId) } repliesCount = repliesCountValue if let recentRepliers = recentRepliers { recentRepliersPeerIds = recentRepliers.map { $0.peerId } } else { recentRepliersPeerIds = nil } repliesMaxId = maxId repliesReadMaxId = readMaxId } } var maxMessageId: MessageId? if let commentsChannelId = commentsChannelId { if let repliesMaxId = repliesMaxId { maxMessageId = MessageId(peerId: commentsChannelId, namespace: Namespaces.Message.Cloud, id: repliesMaxId) } } loop: for j in 0 ..< attributes.count { if let attribute = attributes[j] as? ViewCountMessageAttribute { if let views = views { attributes[j] = ViewCountMessageAttribute(count: max(attribute.count, Int(views))) } } else if let _ = attributes[j] as? ForwardCountMessageAttribute { if let forwards = forwards { attributes[j] = ForwardCountMessageAttribute(count: Int(forwards)) } } else if let attribute = attributes[j] as? ReplyThreadMessageAttribute { foundReplies = true if let repliesCount = repliesCount { var resolvedMaxReadMessageId: MessageId.Id? if let previousMaxReadMessageId = attribute.maxReadMessageId, let repliesReadMaxIdValue = repliesReadMaxId { resolvedMaxReadMessageId = max(previousMaxReadMessageId, repliesReadMaxIdValue) repliesReadMaxId = resolvedMaxReadMessageId } else if let repliesReadMaxIdValue = repliesReadMaxId { resolvedMaxReadMessageId = repliesReadMaxIdValue repliesReadMaxId = resolvedMaxReadMessageId } else { resolvedMaxReadMessageId = attribute.maxReadMessageId } attributes[j] = ReplyThreadMessageAttribute(count: repliesCount, latestUsers: recentRepliersPeerIds ?? [], commentsPeerId: commentsChannelId, maxMessageId: repliesMaxId, maxReadMessageId: resolvedMaxReadMessageId) } } } var maxReadIncomingMessageId: MessageId? if let commentsChannelId = commentsChannelId { if let repliesReadMaxId = repliesReadMaxId { maxReadIncomingMessageId = MessageId(peerId: commentsChannelId, namespace: Namespaces.Message.Cloud, id: repliesReadMaxId) } } resultStates[messageIds[i]] = ViewCountContextState(timestamp: Int32(CFAbsoluteTimeGetCurrent()), clientId: clientId, result: ViewCountContextState.ReplyInfo(commentsPeerId: commentsChannelId, maxReadIncomingMessageId: maxReadIncomingMessageId, maxMessageId: maxMessageId)) if !foundReplies, let repliesCount = repliesCount { attributes.append(ReplyThreadMessageAttribute(count: repliesCount, latestUsers: recentRepliersPeerIds ?? [], commentsPeerId: commentsChannelId, maxMessageId: repliesMaxId, maxReadMessageId: repliesReadMaxId)) } return .update(StoreMessage(id: currentMessage.id, globallyUniqueId: currentMessage.globallyUniqueId, groupingKey: currentMessage.groupingKey, threadId: currentMessage.threadId, timestamp: currentMessage.timestamp, flags: StoreMessageFlags(currentMessage.flags), tags: currentMessage.tags, globalTags: currentMessage.globalTags, localTags: currentMessage.localTags, forwardInfo: storeForwardInfo, authorId: currentMessage.author?.id, text: currentMessage.text, attributes: attributes, media: currentMessage.media)) }) } } } return resultStates } } } |> switchToLatest) |> afterDisposed { [weak self] in self?.queue.async { self?.updatedViewCountDisposables.set(nil, forKey: disposableId) } } |> deliverOn(self.queue) self.updatedViewCountDisposables.set(signal.start(next: { [weak self] updatedStates in guard let strongSelf = self else { return } for (id, state) in updatedStates { strongSelf.updatedViewCountMessageIdsAndTimestamps[id] = state } }), forKey: disposableId) } } } } } public func updateReactionsForMessageIds(messageIds: Set) { /*self.queue.async { var addedMessageIds: [MessageId] = [] let timestamp = Int32(CFAbsoluteTimeGetCurrent()) for messageId in messageIds { let messageTimestamp = self.updatedReactionsMessageIdsAndTimestamps[messageId] if messageTimestamp == nil || messageTimestamp! < timestamp - 5 * 60 { self.updatedReactionsMessageIdsAndTimestamps[messageId] = timestamp addedMessageIds.append(messageId) } } if !addedMessageIds.isEmpty { for (peerId, messageIds) in messagesIdsGroupedByPeerId(Set(addedMessageIds)) { let disposableId = self.nextUpdatedReactionsDisposableId self.nextUpdatedReactionsDisposableId += 1 if let account = self.account { let signal = (account.postbox.transaction { transaction -> Signal in if let peer = transaction.getPeer(peerId), let inputPeer = apiInputPeer(peer) { return account.network.request(Api.functions.messages.getMessagesReactions(peer: inputPeer, id: messageIds.map { $0.id })) |> map(Optional.init) |> `catch` { _ -> Signal in return .single(nil) } |> mapToSignal { updates -> Signal in guard let updates = updates else { return .complete() } return account.postbox.transaction { transaction -> Void in let updateList: [Api.Update] switch updates { case let .updates(updates, _, _, _, _): updateList = updates case let .updatesCombined(updates, _, _, _, _, _): updateList = updates case let .updateShort(update, _): updateList = [update] default: updateList = [] } for update in updateList { switch update { case let .updateMessageReactions(peer, msgId, reactions): transaction.updateMessage(MessageId(peerId: peer.peerId, namespace: Namespaces.Message.Cloud, id: msgId), update: { currentMessage in let updatedReactions = ReactionsMessageAttribute(apiReactions: reactions) let storeForwardInfo = currentMessage.forwardInfo.flatMap(StoreMessageForwardInfo.init) var attributes = currentMessage.attributes loop: for j in 0 ..< attributes.count { if let attribute = attributes[j] as? ReactionsMessageAttribute { if updatedReactions.reactions == attribute.reactions { return .skip } attributes[j] = updatedReactions break loop } } return .update(StoreMessage(id: currentMessage.id, globallyUniqueId: currentMessage.globallyUniqueId, groupingKey: currentMessage.groupingKey, timestamp: currentMessage.timestamp, flags: StoreMessageFlags(currentMessage.flags), tags: currentMessage.tags, globalTags: currentMessage.globalTags, localTags: currentMessage.localTags, forwardInfo: storeForwardInfo, authorId: currentMessage.author?.id, text: currentMessage.text, attributes: attributes, media: currentMessage.media)) }) default: break } } } } } else { return .complete() } } |> switchToLatest) |> afterDisposed { [weak self] in self?.queue.async { self?.updatedReactionsDisposables.set(nil, forKey: disposableId) } } self.updatedReactionsDisposables.set(signal.start(), forKey: disposableId) } } } }*/ } public func updateSeenLiveLocationForMessageIds(messageIds: Set) { self.queue.async { var addedMessageIds: [MessageId] = [] let timestamp = Int32(CFAbsoluteTimeGetCurrent()) for messageId in messageIds { let messageTimestamp = self.updatedSeenLiveLocationMessageIdsAndTimestamps[messageId] if messageTimestamp == nil || messageTimestamp! < timestamp - 1 * 60 { self.updatedSeenLiveLocationMessageIdsAndTimestamps[messageId] = timestamp addedMessageIds.append(messageId) } } if !addedMessageIds.isEmpty { for (peerId, messageIds) in messagesIdsGroupedByPeerId(Set(addedMessageIds)) { let disposableId = self.nextSeenLiveLocationDisposableId self.nextSeenLiveLocationDisposableId += 1 if let account = self.account { let signal = (account.postbox.transaction { transaction -> Signal in if let peer = transaction.getPeer(peerId), let inputPeer = apiInputPeer(peer) { let request: Signal switch inputPeer { case .inputPeerChat, .inputPeerSelf, .inputPeerUser: request = account.network.request(Api.functions.messages.readMessageContents(id: messageIds.map { $0.id })) |> map { _ in true } case let .inputPeerChannel(channelId, accessHash): request = account.network.request(Api.functions.channels.readMessageContents(channel: .inputChannel(channelId: channelId, accessHash: accessHash), id: messageIds.map { $0.id })) |> map { _ in true } default: return .complete() } return request |> `catch` { _ -> Signal in return .single(false) } |> mapToSignal { _ -> Signal in return .complete() } } else { return .complete() } } |> switchToLatest) |> afterDisposed { [weak self] in self?.queue.async { self?.seenLiveLocationDisposables.set(nil, forKey: disposableId) } } self.seenLiveLocationDisposables.set(signal.start(), forKey: disposableId) } } } } } public func updateUnsupportedMediaForMessageIds(messageIds: Set) { self.queue.async { var addedMessageIds: [MessageId] = [] let timestamp = Int32(CFAbsoluteTimeGetCurrent()) for messageId in messageIds { let messageTimestamp = self.updatedUnsupportedMediaMessageIdsAndTimestamps[messageId] if messageTimestamp == nil || messageTimestamp! < timestamp - 10 * 60 * 60 { self.updatedUnsupportedMediaMessageIdsAndTimestamps[messageId] = timestamp addedMessageIds.append(messageId) } } if !addedMessageIds.isEmpty { for (peerId, messageIds) in messagesIdsGroupedByPeerId(Set(addedMessageIds)) { let disposableId = self.nextUpdatedUnsupportedMediaDisposableId self.nextUpdatedUnsupportedMediaDisposableId += 1 if let account = self.account { let signal = account.postbox.transaction { transaction -> Peer? in if let peer = transaction.getPeer(peerId) { return peer } else { return nil } } |> mapToSignal { peer -> Signal in guard let peer = peer else { return .complete() } var fetchSignal: Signal? if let messageId = messageIds.first, messageId.namespace == Namespaces.Message.ScheduledCloud { if let inputPeer = apiInputPeer(peer) { fetchSignal = account.network.request(Api.functions.messages.getScheduledMessages(peer: inputPeer, id: messageIds.map { $0.id })) } } else if peerId.namespace == Namespaces.Peer.CloudUser || peerId.namespace == Namespaces.Peer.CloudGroup { fetchSignal = account.network.request(Api.functions.messages.getMessages(id: messageIds.map { Api.InputMessage.inputMessageID(id: $0.id) })) } else if peerId.namespace == Namespaces.Peer.CloudChannel { if let inputChannel = apiInputChannel(peer) { fetchSignal = account.network.request(Api.functions.channels.getMessages(channel: inputChannel, id: messageIds.map { Api.InputMessage.inputMessageID(id: $0.id) })) } } guard let signal = fetchSignal else { return .complete() } return signal |> map { result -> ([Api.Message], [Api.Chat], [Api.User]) in switch result { case let .messages(messages, chats, users): return (messages, chats, users) case let .messagesSlice(_, _, _, _, messages, chats, users): return (messages, chats, users) case let .channelMessages(_, _, _, _, messages, chats, users): return (messages, chats, users) case .messagesNotModified: return ([], [], []) } } |> `catch` { _ in return Signal<([Api.Message], [Api.Chat], [Api.User]), NoError>.single(([], [], [])) } |> mapToSignal { messages, chats, users -> Signal in return account.postbox.transaction { transaction -> Void in var peers: [Peer] = [] var peerPresences: [PeerId: PeerPresence] = [:] for chat in chats { if let groupOrChannel = mergeGroupOrChannel(lhs: transaction.getPeer(chat.peerId), rhs: chat) { peers.append(groupOrChannel) } } for apiUser in users { if let user = TelegramUser.merge(transaction.getPeer(apiUser.peerId) as? TelegramUser, rhs: apiUser) { peers.append(user) if let presence = TelegramUserPresence(apiUser: apiUser) { peerPresences[user.id] = presence } } } updatePeers(transaction: transaction, peers: peers, update: { _, updated -> Peer in return updated }) updatePeerPresences(transaction: transaction, accountPeerId: account.peerId, peerPresences: peerPresences) for message in messages { guard let storeMessage = StoreMessage(apiMessage: message) else { continue } guard case let .Id(id) = storeMessage.id else { continue } transaction.updateMessage(id, update: { _ in return .update(storeMessage) }) } } } } |> afterDisposed { [weak self] in self?.queue.async { self?.updatedUnsupportedMediaDisposables.set(nil, forKey: disposableId) } } self.updatedUnsupportedMediaDisposables.set(signal.start(), forKey: disposableId) } } } } } public func refreshSecretMediaMediaForMessageIds(messageIds: Set) { self.queue.async { var addedMessageIds: [MessageId] = [] let timestamp = Int32(CFAbsoluteTimeGetCurrent()) for messageId in messageIds { let messageTimestamp = self.refreshSecretChatMediaMessageIdsAndTimestamps[messageId] if messageTimestamp == nil { self.refreshSecretChatMediaMessageIdsAndTimestamps[messageId] = timestamp addedMessageIds.append(messageId) } } if !addedMessageIds.isEmpty { for (_, messageIds) in messagesIdsGroupedByPeerId(Set(addedMessageIds)) { let disposableId = self.nextUpdatedUnsupportedMediaDisposableId self.nextUpdatedUnsupportedMediaDisposableId += 1 if let account = self.account { let signal = account.postbox.transaction { transaction -> [TelegramMediaFile] in var result: [TelegramMediaFile] = [] for id in messageIds { if let message = transaction.getMessage(id) { for media in message.media { if let file = media as? TelegramMediaFile, file.isAnimatedSticker { result.append(file) } } } } return result } |> mapToSignal { files -> Signal in guard !files.isEmpty else { return .complete() } var stickerPacks = Set() for file in files { for attribute in file.attributes { if case let .Sticker(_, packReferenceValue, _) = attribute, let packReference = packReferenceValue { if case .id = packReference { stickerPacks.insert(packReference) } } } } var requests: [Signal] = [] for reference in stickerPacks { if case let .id(id, accessHash) = reference { requests.append(account.network.request(Api.functions.messages.getStickerSet(stickerset: .inputStickerSetID(id: id, accessHash: accessHash))) |> map(Optional.init) |> `catch` { _ -> Signal in return .single(nil) }) } } if requests.isEmpty { return .complete() } return combineLatest(requests) |> mapToSignal { results -> Signal in return account.postbox.transaction { transaction -> Void in for result in results { switch result { case let .stickerSet(_, _, documents)?: for document in documents { if let file = telegramMediaFileFromApiDocument(document) { if transaction.getMedia(file.fileId) != nil { let _ = transaction.updateMedia(file.fileId, update: file) } } } default: break } } } } } |> afterDisposed { [weak self] in self?.queue.async { self?.updatedUnsupportedMediaDisposables.set(nil, forKey: disposableId) } } self.updatedUnsupportedMediaDisposables.set(signal.start(), forKey: disposableId) } } } } } public func updateMarkAllMentionsSeen(peerId: PeerId) { self.queue.async { guard let account = self.account else { return } let _ = (account.postbox.transaction { transaction -> Set in let ids = Set(transaction.getMessageIndicesWithTag(peerId: peerId, namespace: Namespaces.Message.Cloud, tag: .unseenPersonalMessage).map({ $0.id })) if let summary = transaction.getMessageTagSummary(peerId: peerId, tagMask: .unseenPersonalMessage, namespace: Namespaces.Message.Cloud), summary.count > 0 { var maxId: Int32 = summary.range.maxId if let index = transaction.getTopPeerMessageIndex(peerId: peerId, namespace: Namespaces.Message.Cloud) { maxId = index.id.id } transaction.replaceMessageTagSummary(peerId: peerId, tagMask: .unseenPersonalMessage, namespace: Namespaces.Message.Cloud, count: 0, maxId: maxId) addSynchronizeMarkAllUnseenPersonalMessagesOperation(transaction: transaction, peerId: peerId, maxId: summary.range.maxId) } return ids } |> deliverOn(self.queue)).start(next: { [weak self] messageIds in //self?.updateMarkMentionsSeenForMessageIds(messageIds: messageIds) }) } } public func updateMarkMentionsSeenForMessageIds(messageIds: Set) { self.queue.async { var addedMessageIds: [MessageId] = [] for messageId in messageIds { if !self.updatedSeenPersonalMessageIds.contains(messageId) { self.updatedSeenPersonalMessageIds.insert(messageId) addedMessageIds.append(messageId) } } if !addedMessageIds.isEmpty { if let account = self.account { let _ = (account.postbox.transaction { transaction -> Void in for id in addedMessageIds { if let message = transaction.getMessage(id) { var consume = false inner: for attribute in message.attributes { if let attribute = attribute as? ConsumablePersonalMentionMessageAttribute, !attribute.consumed, !attribute.pending { consume = true break inner } } if consume { transaction.updateMessage(id, update: { currentMessage in var attributes = currentMessage.attributes loop: for j in 0 ..< attributes.count { if let attribute = attributes[j] as? ConsumablePersonalMentionMessageAttribute { attributes[j] = ConsumablePersonalMentionMessageAttribute(consumed: attribute.consumed, pending: true) break loop } } return .update(StoreMessage(id: currentMessage.id, globallyUniqueId: currentMessage.globallyUniqueId, groupingKey: currentMessage.groupingKey, threadId: currentMessage.threadId, timestamp: currentMessage.timestamp, flags: StoreMessageFlags(currentMessage.flags), tags: currentMessage.tags, globalTags: currentMessage.globalTags, localTags: currentMessage.localTags, forwardInfo: currentMessage.forwardInfo.flatMap(StoreMessageForwardInfo.init), authorId: currentMessage.author?.id, text: currentMessage.text, attributes: attributes, media: currentMessage.media)) }) transaction.setPendingMessageAction(type: .consumeUnseenPersonalMessage, id: id, action: ConsumePersonalMessageAction()) } } } }).start() } } } } public func forceUpdateCachedPeerData(peerId: PeerId) { self.queue.async { let context: PeerCachedDataContext if let existingContext = self.cachedDataContexts[peerId] { context = existingContext } else { context = PeerCachedDataContext() self.cachedDataContexts[peerId] = context } context.timestamp = CFAbsoluteTimeGetCurrent() guard let account = self.account else { return } let queue = self.queue context.disposable.set(combineLatest(fetchAndUpdateSupplementalCachedPeerData(peerId: peerId, network: account.network, postbox: account.postbox), fetchAndUpdateCachedPeerData(accountPeerId: account.peerId, peerId: peerId, network: account.network, postbox: account.postbox)).start(next: { [weak self] supplementalStatus, cachedStatus in queue.async { guard let strongSelf = self else { return } if !supplementalStatus || !cachedStatus { if let existingContext = strongSelf.cachedDataContexts[peerId] { existingContext.timestamp = nil } } } })) } } private func updateCachedPeerData(peerId: PeerId, viewId: Int32, hasCachedData: Bool) { self.queue.async { let context: PeerCachedDataContext var dataUpdated = false if let existingContext = self.cachedDataContexts[peerId] { context = existingContext context.hasCachedData = hasCachedData if context.timestamp == nil || abs(CFAbsoluteTimeGetCurrent() - context.timestamp!) > 60.0 * 5 { context.timestamp = CFAbsoluteTimeGetCurrent() dataUpdated = true } } else { context = PeerCachedDataContext() context.hasCachedData = hasCachedData self.cachedDataContexts[peerId] = context if !context.hasCachedData || context.timestamp == nil || abs(CFAbsoluteTimeGetCurrent() - context.timestamp!) > 60.0 * 5 { context.timestamp = CFAbsoluteTimeGetCurrent() dataUpdated = true } } context.viewIds.insert(viewId) if dataUpdated { guard let account = self.account else { return } let queue = self.queue context.disposable.set(combineLatest(fetchAndUpdateSupplementalCachedPeerData(peerId: peerId, network: account.network, postbox: account.postbox), fetchAndUpdateCachedPeerData(accountPeerId: account.peerId, peerId: peerId, network: account.network, postbox: account.postbox)).start(next: { [weak self] supplementalStatus, cachedStatus in queue.async { guard let strongSelf = self else { return } if !supplementalStatus || !cachedStatus { if let existingContext = strongSelf.cachedDataContexts[peerId] { existingContext.timestamp = nil } } } })) } } } private func removePeerView(peerId: PeerId, id: Int32) { self.queue.async { if let context = self.cachedDataContexts[peerId] { context.viewIds.remove(id) if context.viewIds.isEmpty { context.disposable.set(nil) context.hasCachedData = false } } } } public func polledChannel(peerId: PeerId) -> Signal { return Signal { subscriber in let disposable = MetaDisposable() self.queue.async { let context: ChannelPollingContext if let current = self.channelPollingContexts[peerId] { context = current } else { context = ChannelPollingContext() self.channelPollingContexts[peerId] = context } if context.subscribers.isEmpty { if let account = self.account { context.disposable.set(keepPollingChannel(postbox: account.postbox, network: account.network, peerId: peerId, stateManager: account.stateManager).start()) } } let index = context.subscribers.add(Void()) disposable.set(ActionDisposable { self.queue.async { if let context = self.channelPollingContexts[peerId] { context.subscribers.remove(index) if context.subscribers.isEmpty { context.disposable.set(nil) } } } }) } return disposable } } func wrappedMessageHistorySignal(chatLocation: ChatLocationInput, signal: Signal<(MessageHistoryView, ViewUpdateType, InitialMessageHistoryData?), NoError>, addHoleIfNeeded: Bool) -> Signal<(MessageHistoryView, ViewUpdateType, InitialMessageHistoryData?), NoError> { let history = withState(signal, { [weak self] () -> Int32 in if let strongSelf = self { return OSAtomicIncrement32(&strongSelf.nextViewId) } else { return -1 } }, next: { [weak self] next, viewId in if let strongSelf = self { strongSelf.queue.async { let (messageIds, localWebpages) = pendingWebpages(entries: next.0.entries) strongSelf.updatePendingWebpages(viewId: viewId, messageIds: messageIds, localWebpages: localWebpages) let (pollMessageIds, pollMessageDict) = pollMessages(entries: next.0.entries) strongSelf.updatePolls(viewId: viewId, messageIds: pollMessageIds, messages: pollMessageDict) if case let .peer(peerId) = chatLocation, peerId.namespace == Namespaces.Peer.CloudChannel { strongSelf.historyViewStateValidationContexts.updateView(id: viewId, view: next.0) } else if case let .external(peerId, _, _) = chatLocation, peerId.namespace == Namespaces.Peer.CloudChannel { strongSelf.historyViewStateValidationContexts.updateView(id: viewId, view: next.0, location: chatLocation) } } } }, disposed: { [weak self] viewId in if let strongSelf = self { strongSelf.queue.async { strongSelf.updatePendingWebpages(viewId: viewId, messageIds: [], localWebpages: [:]) strongSelf.updatePolls(viewId: viewId, messageIds: [], messages: [:]) switch chatLocation { case let .peer(peerId): if peerId.namespace == Namespaces.Peer.CloudChannel { strongSelf.historyViewStateValidationContexts.updateView(id: viewId, view: nil) } case let .external(peerId, _, _): if peerId.namespace == Namespaces.Peer.CloudChannel { strongSelf.historyViewStateValidationContexts.updateView(id: viewId, view: nil, location: chatLocation) } } } } }) let peerId: PeerId switch chatLocation { case let .peer(peerIdValue): peerId = peerIdValue case let .external(peerIdValue, _, _): peerId = peerIdValue } if peerId.namespace == Namespaces.Peer.CloudChannel { return Signal { subscriber in let combinedDisposable = MetaDisposable() self.queue.async { var addHole = false if let context = self.channelPollingContexts[peerId] { if context.subscribers.isEmpty { addHole = true } } else { addHole = true } if addHole { let _ = self.account?.postbox.transaction({ transaction -> Void in if transaction.getPeerChatListIndex(peerId) == nil { if let message = transaction.getTopPeerMessageId(peerId: peerId, namespace: Namespaces.Message.Cloud) { //transaction.addHole(peerId: peerId, namespace: Namespaces.Message.Cloud, space: .everywhere, range: message.id + 1 ... (Int32.max - 1)) } } }).start() } let disposable = history.start(next: { next in subscriber.putNext(next) }, error: { error in subscriber.putError(error) }, completed: { subscriber.putCompletion() }) let polled = self.polledChannel(peerId: peerId).start() combinedDisposable.set(ActionDisposable { disposable.dispose() polled.dispose() }) } return combinedDisposable } } else { return history } } public func scheduledMessagesViewForLocation(_ chatLocation: ChatLocationInput, additionalData: [AdditionalMessageHistoryViewData] = []) -> Signal<(MessageHistoryView, ViewUpdateType, InitialMessageHistoryData?), NoError> { if let account = self.account { let signal = account.postbox.aroundMessageHistoryViewForLocation(chatLocation, anchor: .upperBound, count: 200, fixedCombinedReadStates: nil, topTaggedMessageIdNamespaces: [], tagMask: nil, namespaces: .just(Namespaces.Message.allScheduled), orderStatistics: [], additionalData: additionalData) return withState(signal, { [weak self] () -> Int32 in if let strongSelf = self { return OSAtomicIncrement32(&strongSelf.nextViewId) } else { return -1 } }, next: { [weak self] next, viewId in if let strongSelf = self { strongSelf.queue.async { let (messageIds, localWebpages) = pendingWebpages(entries: next.0.entries) strongSelf.updatePendingWebpages(viewId: viewId, messageIds: messageIds, localWebpages: localWebpages) strongSelf.historyViewStateValidationContexts.updateView(id: viewId, view: next.0, location: chatLocation) } } }, disposed: { [weak self] viewId in if let strongSelf = self { strongSelf.queue.async { strongSelf.updatePendingWebpages(viewId: viewId, messageIds: [], localWebpages: [:]) strongSelf.historyViewStateValidationContexts.updateView(id: viewId, view: nil) } } }) } else { return .never() } } public func aroundMessageOfInterestHistoryViewForLocation(_ chatLocation: ChatLocationInput, count: Int, tagMask: MessageTags? = nil, orderStatistics: MessageHistoryViewOrderStatistics = [], additionalData: [AdditionalMessageHistoryViewData] = []) -> Signal<(MessageHistoryView, ViewUpdateType, InitialMessageHistoryData?), NoError> { if let account = self.account { let signal = account.postbox.aroundMessageOfInterestHistoryViewForChatLocation(chatLocation, count: count, topTaggedMessageIdNamespaces: [Namespaces.Message.Cloud], tagMask: tagMask, namespaces: .not(Namespaces.Message.allScheduled), orderStatistics: orderStatistics, additionalData: wrappedHistoryViewAdditionalData(chatLocation: chatLocation, additionalData: additionalData)) return wrappedMessageHistorySignal(chatLocation: chatLocation, signal: signal, addHoleIfNeeded: true) } else { return .never() } } public func aroundIdMessageHistoryViewForLocation(_ chatLocation: ChatLocationInput, count: Int, ignoreRelatedChats: Bool, messageId: MessageId, tagMask: MessageTags? = nil, orderStatistics: MessageHistoryViewOrderStatistics = [], additionalData: [AdditionalMessageHistoryViewData] = []) -> Signal<(MessageHistoryView, ViewUpdateType, InitialMessageHistoryData?), NoError> { if let account = self.account { let signal = account.postbox.aroundIdMessageHistoryViewForLocation(chatLocation, count: count, ignoreRelatedChats: ignoreRelatedChats, messageId: messageId, topTaggedMessageIdNamespaces: [Namespaces.Message.Cloud], tagMask: tagMask, namespaces: .not(Namespaces.Message.allScheduled), orderStatistics: orderStatistics, additionalData: wrappedHistoryViewAdditionalData(chatLocation: chatLocation, additionalData: additionalData)) return wrappedMessageHistorySignal(chatLocation: chatLocation, signal: signal, addHoleIfNeeded: false) } else { return .never() } } public func aroundMessageHistoryViewForLocation(_ chatLocation: ChatLocationInput, index: MessageHistoryAnchorIndex, anchorIndex: MessageHistoryAnchorIndex, count: Int, clipHoles: Bool = true, ignoreRelatedChats: Bool = false, fixedCombinedReadStates: MessageHistoryViewReadState?, tagMask: MessageTags? = nil, orderStatistics: MessageHistoryViewOrderStatistics = [], additionalData: [AdditionalMessageHistoryViewData] = []) -> Signal<(MessageHistoryView, ViewUpdateType, InitialMessageHistoryData?), NoError> { if let account = self.account { let inputAnchor: HistoryViewInputAnchor switch index { case .upperBound: inputAnchor = .upperBound case .lowerBound: inputAnchor = .lowerBound case let .message(index): inputAnchor = .index(index) } let signal = account.postbox.aroundMessageHistoryViewForLocation(chatLocation, anchor: inputAnchor, count: count, clipHoles: clipHoles, ignoreRelatedChats: ignoreRelatedChats, fixedCombinedReadStates: fixedCombinedReadStates, topTaggedMessageIdNamespaces: [Namespaces.Message.Cloud], tagMask: tagMask, namespaces: .not(Namespaces.Message.allScheduled), orderStatistics: orderStatistics, additionalData: wrappedHistoryViewAdditionalData(chatLocation: chatLocation, additionalData: additionalData)) return wrappedMessageHistorySignal(chatLocation: chatLocation, signal: signal, addHoleIfNeeded: false) } else { return .never() } } func wrappedPeerViewSignal(peerId: PeerId, signal: Signal, updateData: Bool) -> Signal { if updateData { self.queue.async { if let existingContext = self.cachedDataContexts[peerId] { existingContext.timestamp = nil } } } return withState(signal, { [weak self] () -> Int32 in if let strongSelf = self { return OSAtomicIncrement32(&strongSelf.nextViewId) } else { return -1 } }, next: { [weak self] next, viewId in if let strongSelf = self { strongSelf.updateCachedPeerData(peerId: peerId, viewId: viewId, hasCachedData: next.cachedData != nil) } }, disposed: { [weak self] viewId in if let strongSelf = self { strongSelf.removePeerView(peerId: peerId, id: viewId) } }) } public func peerView(_ peerId: PeerId, updateData: Bool = false) -> Signal { if let account = self.account { return wrappedPeerViewSignal(peerId: peerId, signal: account.postbox.peerView(id: peerId), updateData: updateData) } else { return .never() } } public func featuredStickerPacks() -> Signal<[FeaturedStickerPackItem], NoError> { return Signal { subscriber in if let account = self.account { let view = account.postbox.combinedView(keys: [.orderedItemList(id: Namespaces.OrderedItemList.CloudFeaturedStickerPacks)]).start(next: { next in if let view = next.views[.orderedItemList(id: Namespaces.OrderedItemList.CloudFeaturedStickerPacks)] as? OrderedItemListView { subscriber.putNext(view.items.map { $0.contents as! FeaturedStickerPackItem }) } else { subscriber.putNext([]) } }, error: { error in subscriber.putError(error) }, completed: { subscriber.putCompletion() }) let disposable = MetaDisposable() self.queue.async { let context: FeaturedStickerPacksContext if let current = self.featuredStickerPacksContext { context = current } else { context = FeaturedStickerPacksContext() self.featuredStickerPacksContext = context } let timestamp = CFAbsoluteTimeGetCurrent() if context.timestamp == nil || abs(context.timestamp! - timestamp) > 60.0 * 60.0 { context.timestamp = timestamp context.disposable.set(updatedFeaturedStickerPacks(network: account.network, postbox: account.postbox).start()) } let index = context.subscribers.add(Void()) disposable.set(ActionDisposable { self.queue.async { if let context = self.featuredStickerPacksContext { context.subscribers.remove(index) } } }) } return ActionDisposable { view.dispose() disposable.dispose() } } else { subscriber.putNext([]) subscriber.putCompletion() return EmptyDisposable } } } public func callListView(type: CallListViewType, index: MessageIndex, count: Int) -> Signal { if let account = self.account { let granularity: Int32 = 60 * 60 * 24 let timezoneOffset: Int32 = { let nowTimestamp = Int32(CFAbsoluteTimeGetCurrent() + NSTimeIntervalSince1970) var now: time_t = time_t(nowTimestamp) var timeinfoNow: tm = tm() localtime_r(&now, &timeinfoNow) return Int32(timeinfoNow.tm_gmtoff) }() let groupingPredicate: (Message, Message) -> Bool = { lhs, rhs in if lhs.id.peerId != rhs.id.peerId { return false } let lhsTimestamp = ((lhs.timestamp + timezoneOffset) / (granularity)) * (granularity) let rhsTimestamp = ((rhs.timestamp + timezoneOffset) / (granularity)) * (granularity) if lhsTimestamp != rhsTimestamp { return false } var lhsVideo = false var lhsMissed = false var lhsOther = false inner: for media in lhs.media { if let action = media as? TelegramMediaAction { if case let .phoneCall(_, discardReason, _, video) = action.action { lhsVideo = video if lhs.flags.contains(.Incoming), let discardReason = discardReason, case .missed = discardReason { lhsMissed = true } else { lhsOther = true } break inner } } } var rhsVideo = false var rhsMissed = false var rhsOther = false inner: for media in rhs.media { if let action = media as? TelegramMediaAction { if case let .phoneCall(_, discardReason, _, video) = action.action { rhsVideo = video if rhs.flags.contains(.Incoming), let discardReason = discardReason, case .missed = discardReason { rhsMissed = true } else { rhsOther = true } break inner } } } if lhsMissed != rhsMissed || lhsOther != rhsOther || lhsVideo != rhsVideo { return false } return true } let key = PostboxViewKey.globalMessageTags(globalTag: type == .all ? GlobalMessageTags.Calls : GlobalMessageTags.MissedCalls, position: index, count: count, groupingPredicate: groupingPredicate) let signal = account.postbox.combinedView(keys: [key]) |> map { view -> GlobalMessageTagsView in let messageView = view.views[key] as! GlobalMessageTagsView return messageView } let managed = withState(signal, { [weak self] () -> Int32 in if let strongSelf = self { return OSAtomicIncrement32(&strongSelf.nextViewId) } else { return -1 } }, next: { [weak self] next, viewId in if let strongSelf = self { var holes = Set() for entry in next.entries { if case let .hole(index) = entry { holes.insert(index) } } strongSelf.updateVisibleCallListHoles(viewId: viewId, holeIds: holes) } }, disposed: { [weak self] viewId in if let strongSelf = self { strongSelf.updateVisibleCallListHoles(viewId: viewId, holeIds: Set()) } }) return managed |> map { view -> CallListView in var entries: [CallListViewEntry] = [] if !view.entries.isEmpty { var currentMessages: [Message] = [] for entry in view.entries { switch entry { case let .hole(index): if !currentMessages.isEmpty { entries.append(.message(currentMessages[currentMessages.count - 1], currentMessages)) currentMessages.removeAll() } //entries.append(.hole(index)) case let .message(message): if currentMessages.isEmpty || groupingPredicate(message, currentMessages[currentMessages.count - 1]) { currentMessages.append(message) } else { if !currentMessages.isEmpty { entries.append(.message(currentMessages[currentMessages.count - 1], currentMessages)) currentMessages.removeAll() } currentMessages.append(message) } } } if !currentMessages.isEmpty { entries.append(.message(currentMessages[currentMessages.count - 1], currentMessages)) currentMessages.removeAll() } } return CallListView(entries: entries, earlier: view.earlier, later: view.later) } } else { return .never() } } public func unseenPersonalMessagesCount(peerId: PeerId) -> Signal { if let account = self.account { let pendingKey: PostboxViewKey = .pendingMessageActionsSummary(type: .consumeUnseenPersonalMessage, peerId: peerId, namespace: Namespaces.Message.Cloud) let summaryKey: PostboxViewKey = .historyTagSummaryView(tag: .unseenPersonalMessage, peerId: peerId, namespace: Namespaces.Message.Cloud) return account.postbox.combinedView(keys: [pendingKey, summaryKey]) |> map { views -> Int32 in var count: Int32 = 0 if let view = views.views[pendingKey] as? PendingMessageActionsSummaryView { count -= view.count } if let view = views.views[summaryKey] as? MessageHistoryTagSummaryView { if let unseenCount = view.count { count += unseenCount } } return max(0, count) } |> distinctUntilChanged } else { return .never() } } private func wrappedChatListView(signal: Signal<(ChatListView, ViewUpdateType), NoError>) -> Signal<(ChatListView, ViewUpdateType), NoError> { return withState(signal, { [weak self] () -> Int32 in if let strongSelf = self { return OSAtomicIncrement32(&strongSelf.nextViewId) } else { return -1 } }, next: { [weak self] next, viewId in if let strongSelf = self { strongSelf.queue.async { } } }, disposed: { [weak self] viewId in if let strongSelf = self { strongSelf.queue.async { } } }) } public func tailChatListView(groupId: PeerGroupId, filterPredicate: ChatListFilterPredicate? = nil, count: Int) -> Signal<(ChatListView, ViewUpdateType), NoError> { if let account = self.account { return self.wrappedChatListView(signal: account.postbox.tailChatListView(groupId: groupId, filterPredicate: filterPredicate, count: count, summaryComponents: ChatListEntrySummaryComponents(tagSummary: ChatListEntryMessageTagSummaryComponent(tag: .unseenPersonalMessage, namespace: Namespaces.Message.Cloud), actionsSummary: ChatListEntryPendingMessageActionsSummaryComponent(type: PendingMessageActionType.consumeUnseenPersonalMessage, namespace: Namespaces.Message.Cloud)))) } else { return .never() } } public func aroundChatListView(groupId: PeerGroupId, filterPredicate: ChatListFilterPredicate? = nil, index: ChatListIndex, count: Int) -> Signal<(ChatListView, ViewUpdateType), NoError> { if let account = self.account { return self.wrappedChatListView(signal: account.postbox.aroundChatListView(groupId: groupId, filterPredicate: filterPredicate, index: index, count: count, summaryComponents: ChatListEntrySummaryComponents(tagSummary: ChatListEntryMessageTagSummaryComponent(tag: .unseenPersonalMessage, namespace: Namespaces.Message.Cloud), actionsSummary: ChatListEntryPendingMessageActionsSummaryComponent(type: PendingMessageActionType.consumeUnseenPersonalMessage, namespace: Namespaces.Message.Cloud)))) } else { return .never() } } }