2022-05-06 04:44:26 +02:00
|
|
|
// Copyright © 2022 Rangeproof Pty Ltd. All rights reserved.
|
|
|
|
|
|
|
|
import Foundation
|
|
|
|
import GRDB
|
2020-11-11 00:58:56 +01:00
|
|
|
import PromiseKit
|
2022-05-06 04:44:26 +02:00
|
|
|
import SessionSnodeKit
|
2020-11-11 00:58:56 +01:00
|
|
|
|
|
|
|
@objc(LKClosedGroupPoller)
|
|
|
|
public final class ClosedGroupPoller : NSObject {
|
2021-05-12 02:33:29 +02:00
|
|
|
private var isPolling: [String:Bool] = [:]
|
|
|
|
private var timers: [String:Timer] = [:]
|
2021-08-24 06:58:27 +02:00
|
|
|
private let internalQueue: DispatchQueue = DispatchQueue(label:"isPollingQueue")
|
2020-11-11 00:58:56 +01:00
|
|
|
|
|
|
|
// MARK: Settings
|
2021-07-30 01:23:56 +02:00
|
|
|
private static let minPollInterval: Double = 2
|
|
|
|
private static let maxPollInterval: Double = 30
|
2020-11-11 00:58:56 +01:00
|
|
|
|
|
|
|
// MARK: Error
|
|
|
|
private enum Error : LocalizedError {
|
|
|
|
case insufficientSnodes
|
|
|
|
case pollingCanceled
|
|
|
|
|
|
|
|
internal var errorDescription: String? {
|
|
|
|
switch self {
|
|
|
|
case .insufficientSnodes: return "No snodes left to poll."
|
|
|
|
case .pollingCanceled: return "Polling canceled."
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-05-12 02:33:29 +02:00
|
|
|
// MARK: Initialization
|
|
|
|
public static let shared = ClosedGroupPoller()
|
|
|
|
|
|
|
|
private override init() { }
|
|
|
|
|
2020-11-11 00:58:56 +01:00
|
|
|
// MARK: Public API
|
2021-05-12 02:33:29 +02:00
|
|
|
@objc public func start() {
|
2020-11-25 06:15:16 +01:00
|
|
|
#if DEBUG
|
|
|
|
assert(Thread.current.isMainThread) // Timers don't do well on background queues
|
|
|
|
#endif
|
2021-05-12 02:33:29 +02:00
|
|
|
let storage = SNMessagingKitConfiguration.shared.storage
|
|
|
|
let allGroupPublicKeys = storage.getUserClosedGroupPublicKeys()
|
|
|
|
allGroupPublicKeys.forEach { startPolling(for: $0) }
|
2020-11-11 00:58:56 +01:00
|
|
|
}
|
|
|
|
|
2021-05-12 02:33:29 +02:00
|
|
|
public func startPolling(for groupPublicKey: String) {
|
|
|
|
guard !isPolling(for: groupPublicKey) else { return }
|
2021-08-20 02:01:25 +02:00
|
|
|
// Might be a race condition that the setUpPolling finishes too soon,
|
|
|
|
// and the timer is not created, if we mark the group as is polling
|
|
|
|
// after setUpPolling. So the poller may not work, thus misses messages.
|
2022-01-31 04:09:56 +01:00
|
|
|
internalQueue.sync{ isPolling[groupPublicKey] = true }
|
2021-08-20 02:01:25 +02:00
|
|
|
setUpPolling(for: groupPublicKey)
|
2020-11-11 00:58:56 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@objc public func stop() {
|
2021-05-12 02:33:29 +02:00
|
|
|
let storage = SNMessagingKitConfiguration.shared.storage
|
|
|
|
let allGroupPublicKeys = storage.getUserClosedGroupPublicKeys()
|
|
|
|
allGroupPublicKeys.forEach { stopPolling(for: $0) }
|
|
|
|
}
|
|
|
|
|
|
|
|
public func stopPolling(for groupPublicKey: String) {
|
2022-01-31 04:09:56 +01:00
|
|
|
internalQueue.sync{ isPolling[groupPublicKey] = false }
|
2021-08-20 02:01:25 +02:00
|
|
|
timers[groupPublicKey]?.invalidate()
|
2020-11-11 00:58:56 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// MARK: Private API
|
2021-05-12 02:33:29 +02:00
|
|
|
private func setUpPolling(for groupPublicKey: String) {
|
2022-02-24 06:47:08 +01:00
|
|
|
Threading.pollerQueue.async {
|
|
|
|
self.poll(groupPublicKey).done(on: Threading.pollerQueue) { [weak self] _ in
|
2021-05-12 02:40:42 +02:00
|
|
|
self?.pollRecursively(groupPublicKey)
|
2022-02-24 06:47:08 +01:00
|
|
|
}.catch(on: Threading.pollerQueue) { [weak self] error in
|
2022-02-22 04:01:48 +01:00
|
|
|
// The error is logged in poll(_:)
|
2021-05-12 02:40:42 +02:00
|
|
|
self?.pollRecursively(groupPublicKey)
|
|
|
|
}
|
2021-05-12 02:33:29 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private func pollRecursively(_ groupPublicKey: String) {
|
|
|
|
let groupID = LKGroupUtilities.getEncodedClosedGroupIDAsData(groupPublicKey)
|
|
|
|
guard isPolling(for: groupPublicKey),
|
|
|
|
let thread = TSGroupThread.fetch(uniqueId: TSGroupThread.threadId(fromGroupId: groupID)) else { return }
|
|
|
|
// Get the received date of the last message in the thread. If we don't have any messages yet, pick some
|
|
|
|
// reasonable fake time interval to use instead.
|
|
|
|
let lastMessageDate =
|
|
|
|
(thread.numberOfInteractions() > 0) ? thread.lastInteraction.receivedAtDate() : Date().addingTimeInterval(-5 * 60)
|
|
|
|
let timeSinceLastMessage = Date().timeIntervalSince(lastMessageDate)
|
|
|
|
let minPollInterval = ClosedGroupPoller.minPollInterval
|
|
|
|
let limit: Double = 12 * 60 * 60
|
|
|
|
let a = (ClosedGroupPoller.maxPollInterval - minPollInterval) / limit
|
|
|
|
let nextPollInterval = a * min(timeSinceLastMessage, limit) + minPollInterval
|
|
|
|
SNLog("Next poll interval for closed group with public key: \(groupPublicKey) is \(nextPollInterval) s.")
|
2022-02-22 04:01:48 +01:00
|
|
|
timers[groupPublicKey] = Timer.scheduledTimerOnMainThread(withTimeInterval: nextPollInterval, repeats: false) { [weak self] timer in
|
2021-05-12 02:33:29 +02:00
|
|
|
timer.invalidate()
|
2022-02-24 06:47:08 +01:00
|
|
|
Threading.pollerQueue.async {
|
|
|
|
self?.poll(groupPublicKey).done(on: Threading.pollerQueue) { _ in
|
2022-02-22 04:01:48 +01:00
|
|
|
self?.pollRecursively(groupPublicKey)
|
2022-02-24 06:47:08 +01:00
|
|
|
}.catch(on: Threading.pollerQueue) { error in
|
2022-02-21 03:27:44 +01:00
|
|
|
// The error is logged in poll(_:)
|
2022-02-22 04:01:48 +01:00
|
|
|
self?.pollRecursively(groupPublicKey)
|
2021-05-12 02:40:42 +02:00
|
|
|
}
|
2020-11-11 00:58:56 +01:00
|
|
|
}
|
2021-05-12 02:33:29 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private func poll(_ groupPublicKey: String) -> Promise<Void> {
|
|
|
|
guard isPolling(for: groupPublicKey) else { return Promise.value(()) }
|
2022-04-21 08:42:35 +02:00
|
|
|
|
|
|
|
let promise = SnodeAPI.getSwarm(for: groupPublicKey)
|
|
|
|
.then2 { [weak self] swarm -> Promise<(Snode, [SnodeReceivedMessage])> in
|
|
|
|
// randomElement() uses the system's default random generator, which is cryptographically secure
|
|
|
|
guard let snode = swarm.randomElement() else { return Promise(error: Error.insufficientSnodes) }
|
|
|
|
guard let self = self, self.isPolling(for: groupPublicKey) else {
|
|
|
|
return Promise(error: Error.pollingCanceled)
|
|
|
|
}
|
2022-03-22 04:09:47 +01:00
|
|
|
|
2022-05-06 04:44:26 +02:00
|
|
|
return SnodeAPI.getMessages(from: snode, associatedWith: groupPublicKey)
|
|
|
|
.map2 { messages in (snode, messages) }
|
2021-05-12 02:33:29 +02:00
|
|
|
}
|
2022-04-21 08:42:35 +02:00
|
|
|
|
2022-04-01 08:22:45 +02:00
|
|
|
promise.done2 { [weak self] snode, messages in
|
2022-05-06 04:44:26 +02:00
|
|
|
guard self?.isPolling(for: groupPublicKey) == true else { return }
|
2022-03-22 04:09:47 +01:00
|
|
|
|
2022-04-01 08:22:45 +02:00
|
|
|
if !messages.isEmpty {
|
2022-05-06 04:44:26 +02:00
|
|
|
SNLog("Received \(messages.count) message(s) in closed group with public key: \(groupPublicKey).")
|
2022-04-21 08:42:35 +02:00
|
|
|
|
2022-04-01 08:22:45 +02:00
|
|
|
GRDBStorage.shared.write { db in
|
2022-04-27 02:48:54 +02:00
|
|
|
var jobDetailMessages: [MessageReceiveJob.Details.MessageInfo] = []
|
|
|
|
|
2022-04-21 08:42:35 +02:00
|
|
|
messages.forEach { message in
|
|
|
|
guard let envelope = SNProtoEnvelope.from(message) else { return }
|
|
|
|
|
|
|
|
do {
|
2022-04-27 02:48:54 +02:00
|
|
|
jobDetailMessages.append(
|
|
|
|
MessageReceiveJob.Details.MessageInfo(
|
|
|
|
data: try envelope.serializedData(),
|
|
|
|
serverHash: message.info.hash
|
2022-04-21 08:42:35 +02:00
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
// Persist the received message after the MessageReceiveJob is created
|
2022-04-27 02:48:54 +02:00
|
|
|
_ = try message.info.saved(db)
|
2022-04-21 08:42:35 +02:00
|
|
|
}
|
|
|
|
catch {
|
2022-05-06 04:44:26 +02:00
|
|
|
switch error {
|
|
|
|
// Ignore unique constraint violations here (they will be hanled in the MessageReceiveJob)
|
|
|
|
case .SQLITE_CONSTRAINT_UNIQUE: break
|
|
|
|
|
|
|
|
default:
|
|
|
|
SNLog("Failed to deserialize envelope due to error: \(error).")
|
|
|
|
}
|
2022-04-21 08:42:35 +02:00
|
|
|
}
|
|
|
|
}
|
2022-04-27 02:48:54 +02:00
|
|
|
|
|
|
|
JobRunner.add(
|
|
|
|
db,
|
|
|
|
job: Job(
|
|
|
|
variant: .messageReceive,
|
|
|
|
behaviour: .runOnce,
|
|
|
|
threadId: groupPublicKey,
|
|
|
|
details: MessageReceiveJob.Details(
|
|
|
|
messages: jobDetailMessages,
|
|
|
|
isBackgroundPoll: false
|
|
|
|
)
|
|
|
|
)
|
|
|
|
)
|
2022-04-01 08:22:45 +02:00
|
|
|
}
|
|
|
|
}
|
2020-11-11 00:58:56 +01:00
|
|
|
}
|
2021-05-12 02:33:29 +02:00
|
|
|
promise.catch2 { error in
|
|
|
|
SNLog("Polling failed for closed group with public key: \(groupPublicKey) due to error: \(error).")
|
|
|
|
}
|
|
|
|
return promise.map { _ in }
|
|
|
|
}
|
|
|
|
|
|
|
|
// MARK: Convenience
|
|
|
|
private func isPolling(for groupPublicKey: String) -> Bool {
|
2021-08-24 06:58:27 +02:00
|
|
|
return internalQueue.sync{ isPolling[groupPublicKey] ?? false }
|
2020-11-11 00:58:56 +01:00
|
|
|
}
|
|
|
|
}
|