session-ios/SessionUtilitiesKit/JobRunner/JobRunner.swift

// Copyright © 2022 Rangeproof Pty Ltd. All rights reserved.
//
// stringlint:disable

import Foundation
import GRDB

public protocol JobRunnerType {
    // MARK: - Configuration
    
    func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant)
    func canStart(queue: JobQueue?) -> Bool
    func afterBlockingQueue(callback: @escaping () -> ())
        
    // MARK: - State Management
    
    func jobInfoFor(jobs: [Job]?, state: JobRunner.JobState, variant: Job.Variant?) -> [Int64: JobRunner.JobInfo]
    
    func appDidFinishLaunching(using dependencies: Dependencies)
    func appDidBecomeActive(using dependencies: Dependencies)
    func startNonBlockingQueues(using dependencies: Dependencies)
    func stopAndClearPendingJobs(exceptForVariant: Job.Variant?, onComplete: (() -> ())?)
    
    // MARK: - Job Scheduling
    
    @discardableResult func add(_ db: Database, job: Job?, canStartJob: Bool, using dependencies: Dependencies) -> Job?
    func upsert(_ db: Database, job: Job?, canStartJob: Bool, using dependencies: Dependencies)
    @discardableResult func insert(_ db: Database, job: Job?, before otherJob: Job) -> (Int64, Job)?
}

// MARK: - JobRunnerType Convenience

public extension JobRunnerType {
    func allJobInfo() -> [Int64: JobRunner.JobInfo] { return jobInfoFor(jobs: nil, state: .any, variant: nil) }
    
    func jobInfoFor(jobs: [Job]) -> [Int64: JobRunner.JobInfo] {
        return jobInfoFor(jobs: jobs, state: .any, variant: nil)
    }

    func jobInfoFor(jobs: [Job], state: JobRunner.JobState) -> [Int64: JobRunner.JobInfo] {
        return jobInfoFor(jobs: jobs, state: state, variant: nil)
    }

    func jobInfoFor(state: JobRunner.JobState) -> [Int64: JobRunner.JobInfo] {
        return jobInfoFor(jobs: nil, state: state, variant: nil)
    }

    func jobInfoFor(state: JobRunner.JobState, variant: Job.Variant) -> [Int64: JobRunner.JobInfo] {
        return jobInfoFor(jobs: nil, state: state, variant: variant)
    }

    func jobInfoFor(variant: Job.Variant) -> [Int64: JobRunner.JobInfo] {
        return jobInfoFor(jobs: nil, state: .any, variant: variant)
    }
    
    func isCurrentlyRunning(_ job: Job?) -> Bool {
        guard let job: Job = job else { return false }
        
        return !jobInfoFor(jobs: [job], state: .running).isEmpty
    }
    
    func hasJob<T: Encodable>(
        of variant: Job.Variant? = nil,
        inState state: JobRunner.JobState = .any,
        with jobDetails: T
    ) -> Bool {
        guard
            let detailsData: Data = try? JSONEncoder()
                .with(outputFormatting: .sortedKeys)    // Needed for deterministic comparison
                .encode(jobDetails)
        else { return false }
        
        return jobInfoFor(jobs: nil, state: state, variant: variant)
            .values
            .contains(where: { $0.detailsData == detailsData })
    }
    
    func stopAndClearPendingJobs(exceptForVariant: Job.Variant? = nil, onComplete: (() -> ())? = nil) {
        stopAndClearPendingJobs(exceptForVariant: exceptForVariant, onComplete: onComplete)
    }
}

// MARK: - JobExecutor

public protocol JobExecutor {
    /// The maximum number of times the job can fail before it fails permanently
    ///
    /// **Note:** A value of `-1` means it will retry indefinitely
    static var maxFailureCount: Int { get }
    static var requiresThreadId: Bool { get }
    static var requiresInteractionId: Bool { get }

    /// This method contains the logic needed to complete a job
    ///
    /// **Note:** The code in this method should run synchronously and the various
    /// "result" blocks should not be called within a database closure
    ///
    /// - Parameters:
    ///   - job: The job which is being run
    ///   - success: The closure which is called when the job succeeds (with an
    ///   updated `job` and a flag indicating whether the job should forcibly stop running)
    ///   - failure: The closure which is called when the job fails (with an updated
    ///   `job`, an `Error` (if applicable) and a flag indicating whether it was a permanent
    ///   failure)
    ///   - deferred: The closure which is called when the job is deferred (with an
    ///   updated `job`)
    static func run(
        _ job: Job,
        queue: DispatchQueue,
        success: @escaping (Job, Bool, Dependencies) -> (),
        failure: @escaping (Job, Error?, Bool, Dependencies) -> (),
        deferred: @escaping (Job, Dependencies) -> (),
        using dependencies: Dependencies
    )
}

// MARK: - JobRunner

public final class JobRunner: JobRunnerType {
    public struct JobState: OptionSet, Hashable {
        public let rawValue: UInt8
        
        public init(rawValue: UInt8) {
            self.rawValue = rawValue
        }
        
        public static let pending: JobState = JobState(rawValue: 1 << 0)
        public static let running: JobState = JobState(rawValue: 1 << 1)
        
        public static let any: JobState = [ .pending, .running ]
    }
    
    public enum JobResult {
        case succeeded
        case failed
        case deferred
        case notFound
    }

    public struct JobInfo: Equatable, CustomDebugStringConvertible {
        public let variant: Job.Variant
        public let threadId: String?
        public let interactionId: Int64?
        public let detailsData: Data?
        
        public var debugDescription: String {
            let dataDescription: String = detailsData
                .map { data in "Data(hex: \(data.toHexString()), \(data.bytes.count) bytes" }
                .defaulting(to: "nil")
            
            return [
                "JobRunner.JobInfo(",
                "variant: \(variant),",
                " threadId: \(threadId ?? "nil"),",
                " interactionId: \(interactionId.map { "\($0)" } ?? "nil"),",
                " detailsData: \(dataDescription)",
                ")"
            ].joined()
        }
    }
    
    // MARK: - Variables
    
    private let allowToExecuteJobs: Bool
    private let blockingQueue: Atomic<JobQueue?>
    private let queues: Atomic<[Job.Variant: JobQueue]>
    private var blockingQueueDrainCallback: Atomic<[() -> ()]> = Atomic([])
    
    internal var appReadyToStartQueues: Atomic<Bool> = Atomic(false)
    internal var appHasBecomeActive: Atomic<Bool> = Atomic(false)
    internal var perSessionJobsCompleted: Atomic<Set<Int64>> = Atomic([])
    internal var hasCompletedInitialBecomeActive: Atomic<Bool> = Atomic(false)
    internal var shutdownBackgroundTask: Atomic<OWSBackgroundTask?> = Atomic(nil)
    
    private var canStartNonBlockingQueue: Bool {
        blockingQueue.wrappedValue?.hasStartedAtLeastOnce.wrappedValue == true &&
        blockingQueue.wrappedValue?.isRunning.wrappedValue != true &&
        appHasBecomeActive.wrappedValue
    }
    
    // MARK: - Initialization
    
    init(
        isTestingJobRunner: Bool = false,
        variantsToExclude: [Job.Variant] = [],
        using dependencies: Dependencies = Dependencies()
    ) {
        var jobVariants: Set<Job.Variant> = Job.Variant.allCases
            .filter { !variantsToExclude.contains($0) }
            .asSet()
        
        self.allowToExecuteJobs = (
            isTestingJobRunner || (
                HasAppContext() &&
                CurrentAppContext().isMainApp &&
                !SNUtilitiesKit.isRunningTests
            )
        )
        self.blockingQueue = Atomic(
            JobQueue(
                type: .blocking,
                executionType: .serial,
                qos: .default,
                isTestingJobRunner: isTestingJobRunner,
                jobVariants: []
            )
        )
        self.queues = Atomic([
            // MARK: -- Message Send Queue
            
            JobQueue(
                type: .messageSend,
                executionType: .concurrent, // Allow as many jobs to run at once as supported by the device
                qos: .default,
                isTestingJobRunner: isTestingJobRunner,
                jobVariants: [
                    jobVariants.remove(.attachmentUpload),
                    jobVariants.remove(.messageSend),
                    jobVariants.remove(.notifyPushServer),
                    jobVariants.remove(.sendReadReceipts),
                    jobVariants.remove(.groupLeaving),
                    jobVariants.remove(.configurationSync)
                ].compactMap { $0 }
            ),
            
            // MARK: -- Message Receive Queue
            
            JobQueue(
                type: .messageReceive,
                // Explicitly serial as executing concurrently means message receives getting processed at
                // different speeds which can result in:
                // • Small batches of messages appearing in the UI before larger batches
                // • Closed group messages encrypted with updated keys could start parsing before it's key
                //   update message has been processed (ie. guaranteed to fail)
                executionType: .serial,
                qos: .default,
                isTestingJobRunner: isTestingJobRunner,
                jobVariants: [
                    jobVariants.remove(.messageReceive),
                    jobVariants.remove(.configMessageReceive)
                ].compactMap { $0 }
            ),
            
            // MARK: -- Attachment Download Queue
            
            JobQueue(
                type: .attachmentDownload,
                executionType: .serial,
                qos: .utility,
                isTestingJobRunner: isTestingJobRunner,
                jobVariants: [
                    jobVariants.remove(.attachmentDownload)
                ].compactMap { $0 }
            ),
            
            // MARK: -- Expiration Update Queue
            
            JobQueue(
                type: .expirationUpdate,
                executionType: .concurrent, // Allow as many jobs to run at once as supported by the device
                qos: .default,
                isTestingJobRunner: isTestingJobRunner,
                jobVariants: [
                    jobVariants.remove(.expirationUpdate),
                    jobVariants.remove(.getExpiration)
                ].compactMap { $0 }
            ),
            
            // MARK: -- General Queue
            
            JobQueue(
                type: .general(number: 0),
                executionType: .serial,
                qos: .utility,
                isTestingJobRunner: isTestingJobRunner,
                jobVariants: Array(jobVariants)
            )
        ].reduce(into: [:]) { prev, next in
            next.jobVariants.forEach { variant in
                prev[variant] = next
            }
        })
        
        // Now that we've finished setting up the JobRunner, update the queue closures
        self.blockingQueue.mutate {
            $0?.canStart = { [weak self] queue -> Bool in (self?.canStart(queue: queue) == true) }
            $0?.onQueueDrained = { [weak self] in
                // Once all blocking jobs have been completed we want to start running
                // the remaining job queues
                self?.startNonBlockingQueues(using: dependencies)
                
                self?.blockingQueueDrainCallback.mutate {
                    $0.forEach { $0() }
                    $0 = []
                }
            }
        }
        
        self.queues.mutate {
            $0.values.forEach { queue in
                queue.canStart = { [weak self] targetQueue -> Bool in (self?.canStart(queue: targetQueue) == true) }
            }
        }
    }
    
    // MARK: - Configuration
    
    public func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant) {
        blockingQueue.wrappedValue?.setExecutor(executor, for: variant) // The blocking queue can run any job
        queues.wrappedValue[variant]?.setExecutor(executor, for: variant)
    }
    
    public func canStart(queue: JobQueue?) -> Bool {
        return (
            allowToExecuteJobs &&
            appReadyToStartQueues.wrappedValue && (
                queue?.type == .blocking ||
                canStartNonBlockingQueue
            )
        )
    }

    public func afterBlockingQueue(callback: @escaping () -> ()) {
        guard
            (blockingQueue.wrappedValue?.hasStartedAtLeastOnce.wrappedValue != true) ||
            (blockingQueue.wrappedValue?.isRunning.wrappedValue == true)
        else { return callback() }
    
        blockingQueueDrainCallback.mutate { $0.append(callback) }
    }

    // MARK: - State Management

    public func jobInfoFor(
        jobs: [Job]?,
        state: JobRunner.JobState,
        variant: Job.Variant?
    ) -> [Int64: JobRunner.JobInfo] {
        var result: [(Int64, JobRunner.JobInfo)] = []
        let targetKeys: [JobQueue.JobKey] = (jobs?.compactMap { JobQueue.JobKey($0) } ?? [])
        let targetVariants: [Job.Variant] = (variant.map { [$0] } ?? jobs?.map { $0.variant })
            .defaulting(to: [])
        
        // Insert the state of any pending jobs
        if state.contains(.pending) {
            func infoFor(queue: JobQueue?, variants: [Job.Variant]) -> [(Int64, JobRunner.JobInfo)] {
                return (queue?.pendingJobsQueue.wrappedValue
                    .filter { variants.isEmpty || variants.contains($0.variant) }
                    .compactMap { job -> (Int64, JobRunner.JobInfo)? in
                        guard let jobKey: JobQueue.JobKey = JobQueue.JobKey(job) else { return nil }
                        guard
                            targetKeys.isEmpty ||
                            targetKeys.contains(jobKey)
                        else { return nil }
                        
                        return (
                            jobKey.id,
                            JobRunner.JobInfo(
                                variant: job.variant,
                                threadId: job.threadId,
                                interactionId: job.interactionId,
                                detailsData: job.details
                            )
                        )
                    })
                    .defaulting(to: [])
            }
            
            result.append(contentsOf: infoFor(queue: blockingQueue.wrappedValue, variants: targetVariants))
            queues.wrappedValue
                .filter { key, _ -> Bool in targetVariants.isEmpty || targetVariants.contains(key) }
                .map { _, queue in queue }
                .asSet()
                .forEach { queue in result.append(contentsOf: infoFor(queue: queue, variants: targetVariants)) }
        }
        
        // Insert the state of any running jobs
        if state.contains(.running) {
            func infoFor(queue: JobQueue?, variants: [Job.Variant]) -> [(Int64, JobRunner.JobInfo)] {
                return (queue?.infoForAllCurrentlyRunningJobs()
                    .filter { variants.isEmpty || variants.contains($0.value.variant) }
                    .compactMap { jobId, info -> (Int64, JobRunner.JobInfo)? in
                        guard
                            targetKeys.isEmpty ||
                            targetKeys.contains(JobQueue.JobKey(id: jobId, variant: info.variant))
                        else { return nil }
                        
                        return (jobId, info)
                    })
                    .defaulting(to: [])
            }
            
            result.append(contentsOf: infoFor(queue: blockingQueue.wrappedValue, variants: targetVariants))
            queues.wrappedValue
                .filter { key, _ -> Bool in targetVariants.isEmpty || targetVariants.contains(key) }
                .map { _, queue in queue }
                .asSet()
                .forEach { queue in result.append(contentsOf: infoFor(queue: queue, variants: targetVariants)) }
        }
        
        return result
            .reduce(into: [:]) { result, next in
                result[next.0] = next.1
            }
    }
    
    public func appDidFinishLaunching(using dependencies: Dependencies) {
        // Flag that the JobRunner can start it's queues
        appReadyToStartQueues.mutate { $0 = true }
        
        // Note: 'appDidBecomeActive' will run on first launch anyway so we can
        // leave those jobs out and can wait until then to start the JobRunner
        let jobsToRun: (blocking: [Job], nonBlocking: [Job]) = dependencies.storage
            .read { db in
                let blockingJobs: [Job] = try Job
                    .filter(
                        [
                            Job.Behaviour.recurringOnLaunch,
                            Job.Behaviour.runOnceNextLaunch
                        ].contains(Job.Columns.behaviour)
                    )
                    .filter(Job.Columns.shouldBlock == true)
                    .order(
                        Job.Columns.priority.desc,
                        Job.Columns.id
                    )
                    .fetchAll(db)
                let nonblockingJobs: [Job] = try Job
                    .filter(
                        [
                            Job.Behaviour.recurringOnLaunch,
                            Job.Behaviour.runOnceNextLaunch
                        ].contains(Job.Columns.behaviour)
                    )
                    .filter(Job.Columns.shouldBlock == false)
                    .order(
                        Job.Columns.priority.desc,
                        Job.Columns.id
                    )
                    .fetchAll(db)
                
                return (blockingJobs, nonblockingJobs)
            }
            .defaulting(to: ([], []))
        
        // Add and start any blocking jobs
        blockingQueue.wrappedValue?.appDidFinishLaunching(
            with: jobsToRun.blocking,
            canStart: true,
            using: dependencies
        )
        
        // Add any non-blocking jobs (we don't start these incase there are blocking "on active"
        // jobs as well)
        let jobsByVariant: [Job.Variant: [Job]] = jobsToRun.nonBlocking.grouped(by: \.variant)
        let jobQueues: [Job.Variant: JobQueue] = queues.wrappedValue
        
        jobsByVariant.forEach { variant, jobs in
            jobQueues[variant]?.appDidFinishLaunching(
                with: jobs,
                canStart: false,
                using: dependencies
            )
        }
    }
    
    public func appDidBecomeActive(using dependencies: Dependencies) {
        // Flag that the JobRunner can start it's queues and start queueing non-launch jobs
        appReadyToStartQueues.mutate { $0 = true }
        appHasBecomeActive.mutate { $0 = true }
        
        // If we have a running "sutdownBackgroundTask" then we want to cancel it as otherwise it
        // can result in the database being suspended and us being unable to interact with it at all
        shutdownBackgroundTask.mutate {
            $0?.cancel()
            $0 = nil
        }
        
        // Retrieve any jobs which should run when becoming active
        let hasCompletedInitialBecomeActive: Bool = self.hasCompletedInitialBecomeActive.wrappedValue
        let jobsToRun: [Job] = dependencies.storage
            .read { db in
                return try Job
                    .filter(Job.Columns.behaviour == Job.Behaviour.recurringOnActive)
                    .order(
                        Job.Columns.priority.desc,
                        Job.Columns.id
                    )
                    .fetchAll(db)
            }
            .defaulting(to: [])
            .filter { hasCompletedInitialBecomeActive || !$0.shouldSkipLaunchBecomeActive }
        
        // Store the current queue state locally to avoid multiple atomic retrievals
        let jobQueues: [Job.Variant: JobQueue] = queues.wrappedValue
        let blockingQueueIsRunning: Bool = (blockingQueue.wrappedValue?.isRunning.wrappedValue == true)
        
        guard !jobsToRun.isEmpty else {
            if !blockingQueueIsRunning {
                jobQueues.map { _, queue in queue }.asSet().forEach { $0.start(using: dependencies) }
            }
            return
        }
        
        // Add and start any non-blocking jobs (if there are no blocking jobs)
        //
        // We only want to trigger the queue to start once so we need to consolidate the
        // queues to list of jobs (as queues can handle multiple job variants), this means
        // that 'onActive' jobs will be queued before any standard jobs
        let jobsByVariant: [Job.Variant: [Job]] = jobsToRun.grouped(by: \.variant)
        
        jobQueues
            .reduce(into: [:]) { result, variantAndQueue in
                result[variantAndQueue.value] = (result[variantAndQueue.value] ?? [])
                    .appending(contentsOf: (jobsByVariant[variantAndQueue.key] ?? []))
            }
            .forEach { queue, jobs in
                queue.appDidBecomeActive(
                    with: jobs,
                    canStart: !blockingQueueIsRunning,
                    using: dependencies
                )
            }
        
        self.hasCompletedInitialBecomeActive.mutate { $0 = true }
    }
    
    public func startNonBlockingQueues(using dependencies: Dependencies) {
        queues.wrappedValue.map { _, queue in queue }.asSet().forEach { queue in
            queue.start(using: dependencies)
        }
    }
    
    public func stopAndClearPendingJobs(
        exceptForVariant: Job.Variant?,
        onComplete: (() -> ())?
    ) {
        // Inform the JobRunner that it can't start any queues (this is to prevent queues from
        // rescheduling themselves while in the background, when the app restarts or becomes active
        // the JobRunenr will update this flag)
        appReadyToStartQueues.mutate { $0 = false }
        appHasBecomeActive.mutate { $0 = false }
        
        // Stop all queues except for the one containing the `exceptForVariant`
        queues.wrappedValue
            .map { _, queue in queue }
            .asSet()
            .filter { queue -> Bool in
                guard let exceptForVariant: Job.Variant = exceptForVariant else { return true }
                
                return !queue.jobVariants.contains(exceptForVariant)
            }
            .forEach { $0.stopAndClearPendingJobs() }
        
        // Ensure the queue is actually running (if not the trigger the callback immediately)
        guard
            let exceptForVariant: Job.Variant = exceptForVariant,
            let queue: JobQueue = queues.wrappedValue[exceptForVariant],
            queue.isRunning.wrappedValue == true
        else {
            onComplete?()
            return
        }
        
        let oldQueueDrained: (() -> ())? = queue.onQueueDrained
        
        // Create a backgroundTask to give the queue the chance to properly be drained
        shutdownBackgroundTask.mutate {
            $0 = OWSBackgroundTask(labelStr: #function) { [weak queue] state in
                // If the background task didn't succeed then trigger the onComplete (and hope we have
                // enough time to complete it's logic)
                guard state != .cancelled else {
                    queue?.onQueueDrained = oldQueueDrained
                    return
                }
                guard state != .success else { return }
                
                onComplete?()
                queue?.onQueueDrained = oldQueueDrained
                queue?.stopAndClearPendingJobs()
            }
        }
        
        // Add a callback to be triggered once the queue is drained
        queue.onQueueDrained = { [weak self, weak queue] in
            oldQueueDrained?()
            queue?.onQueueDrained = oldQueueDrained
            onComplete?()
            
            self?.shutdownBackgroundTask.mutate { $0 = nil }
        }
    }
    
    // MARK: - Execution
    
    @discardableResult public func add(
        _ db: Database,
        job: Job?,
        canStartJob: Bool,
        using dependencies: Dependencies
    ) -> Job? {
        // Store the job into the database (getting an id for it)
        guard let updatedJob: Job = try? job?.inserted(db) else {
            SNLog("[JobRunner] Unable to add \(job.map { "\($0.variant)" } ?? "unknown") job")
            return nil
        }
        guard !canStartJob || updatedJob.id != nil else {
            SNLog("[JobRunner] Not starting \(job.map { "\($0.variant)" } ?? "unknown") job due to missing id")
            return nil
        }
        
        // Don't add to the queue if the JobRunner isn't ready (it's been saved to the db so it'll be loaded
        // once the queue actually get started later)
        guard canAddToQueue(updatedJob) else { return updatedJob }
        
        queues.wrappedValue[updatedJob.variant]?.add(db, job: updatedJob, canStartJob: canStartJob, using: dependencies)
        
        // Don't start the queue if the job can't be started
        guard canStartJob else { return updatedJob }
        
        // Start the job runner if needed
        db.afterNextTransactionNestedOnce(dedupeId: "JobRunner-Start: \(updatedJob.variant)") { [weak self] _ in
            self?.queues.wrappedValue[updatedJob.variant]?.start(using: dependencies)
        }
        
        return updatedJob
    }
    
    public func upsert(
        _ db: Database,
        job: Job?,
        canStartJob: Bool,
        using dependencies: Dependencies
    ) {
        guard let job: Job = job else { return }    // Ignore null jobs
        guard job.id != nil else {
            add(db, job: job, canStartJob: canStartJob, using: dependencies)
            return
        }
        
        // Don't add to the queue if the JobRunner isn't ready (it's been saved to the db so it'll be loaded
        // once the queue actually get started later)
        guard canAddToQueue(job) else { return }
        
        queues.wrappedValue[job.variant]?.upsert(db, job: job, canStartJob: canStartJob, using: dependencies)
        
        // Don't start the queue if the job can't be started
        guard canStartJob else { return }
        
        // Start the job runner if needed
        
        db.afterNextTransactionNestedOnce(dedupeId: "JobRunner-Start: \(job.variant)") { [weak self] _ in
            self?.queues.wrappedValue[job.variant]?.start(using: dependencies)
        }
    }
    
    @discardableResult public func insert(
        _ db: Database,
        job: Job?,
        before otherJob: Job
    ) -> (Int64, Job)? {
        switch job?.behaviour {
            case .recurringOnActive, .recurringOnLaunch, .runOnceNextLaunch:
                SNLog("[JobRunner] Attempted to insert \(job.map { "\($0.variant)" } ?? "unknown") job before the current one even though it's behaviour is \(job.map { "\($0.behaviour)" } ?? "unknown")")
                return nil
                
            default: break
        }
        
        // Store the job into the database (getting an id for it)
        guard let updatedJob: Job = try? job?.inserted(db) else {
            SNLog("[JobRunner] Unable to add \(job.map { "\($0.variant)" } ?? "unknown") job")
            return nil
        }
        guard let jobId: Int64 = updatedJob.id else {
            SNLog("[JobRunner] Unable to add \(job.map { "\($0.variant)" } ?? "unknown") job due to missing id")
            return nil
        }
        
        queues.wrappedValue[updatedJob.variant]?.insert(updatedJob, before: otherJob)
        
        return (jobId, updatedJob)
    }
    
    internal func afterCurrentlyRunningJob(_ job: Job?, callback: @escaping (JobResult) -> ()) {
        guard let job: Job = job, let jobId: Int64 = job.id, let queue: JobQueue = queues.wrappedValue[job.variant] else {
            callback(.notFound)
            return
        }
        
        queue.afterCurrentlyRunningJob(jobId, callback: callback)
    }
    
    internal func removePendingJob(_ job: Job?) {
        guard let job: Job = job, let jobId: Int64 = job.id else { return }
        
        queues.wrappedValue[job.variant]?.removePendingJob(jobId)
    }
    
    // MARK: - Convenience

    fileprivate static func getRetryInterval(for job: Job) -> TimeInterval {
        // Arbitrary backoff factor...
        // try  1 delay: 0.5s
        // try  2 delay: 1s
        // ...
        // try  5 delay: 16s
        // ...
        // try 11 delay: 512s
        let maxBackoff: Double = 10 * 60 // 10 minutes
        return 0.25 * min(maxBackoff, pow(2, Double(job.failureCount)))
    }
    
    fileprivate func canAddToQueue(_ job: Job) -> Bool {
        // We can only start the job if it's an "on launch" job or the app has become active
        return (
            job.behaviour == .runOnceNextLaunch ||
            job.behaviour == .recurringOnLaunch ||
            appHasBecomeActive.wrappedValue
        )
    }
}

// MARK: - JobQueue

public final class JobQueue: Hashable {
    fileprivate enum QueueType: Hashable {
        case blocking
        case general(number: Int)
        case messageSend
        case messageReceive
        case attachmentDownload
        case expirationUpdate
        
        var name: String {
            switch self {
                case .blocking: return "Blocking"
                case .general(let number): return "General-\(number)"
                case .messageSend: return "MessageSend"
                case .messageReceive: return "MessageReceive"
                case .attachmentDownload: return "AttachmentDownload"
                case .expirationUpdate: return "ExpirationUpdate"
            }
        }
    }
    
    fileprivate enum ExecutionType {
        /// A serial queue will execute one job at a time until the queue is empty, then will load any new/deferred
        /// jobs and run those one at a time
        case serial
        
        /// A concurrent queue will execute as many jobs as the device supports at once until the queue is empty,
        /// then will load any new/deferred jobs and try to start them all
        case concurrent
    }
    
    private class Trigger {
        private var timer: Timer?
        fileprivate var fireTimestamp: TimeInterval = 0
        
        static func create(
            queue: JobQueue,
            timestamp: TimeInterval,
            using dependencies: Dependencies
        ) -> Trigger? {
            /// Setup the trigger (wait at least 1 second before triggering)
            ///
            /// **Note:** We use the `Timer.scheduledTimerOnMainThread` method because running a timer
            /// on our random queue threads results in the timer never firing, the `start` method will redirect itself to
            /// the correct thread
            let trigger: Trigger = Trigger()
            trigger.fireTimestamp = max(1, (timestamp - dependencies.dateNow.timeIntervalSince1970))
            trigger.timer = Timer.scheduledTimerOnMainThread(
                withTimeInterval: trigger.fireTimestamp,
                repeats: false,
                using: dependencies,
                block: { [weak queue] _ in
                    queue?.start(using: dependencies)
                }
            )
            return trigger
        }
        
        func invalidate() {
            // Need to do this to prevent a strong reference cycle
            timer?.invalidate()
            timer = nil
        }
    }
    
    fileprivate struct JobKey: Equatable, Hashable {
        fileprivate let id: Int64
        fileprivate let variant: Job.Variant
        
        fileprivate init(id: Int64, variant: Job.Variant) {
            self.id = id
            self.variant = variant
        }
        
        fileprivate init?(_ job: Job?) {
            guard let id: Int64 = job?.id, let variant: Job.Variant = job?.variant else { return nil }
            
            self.id = id
            self.variant = variant
        }
    }
    
    private static let deferralLoopThreshold: Int = 3
    
    private let id: UUID = UUID()
    fileprivate let type: QueueType
    private let executionType: ExecutionType
    private let qosClass: DispatchQoS
    private let queueKey: DispatchSpecificKey = DispatchSpecificKey<String>()
    private let queueContext: String
    fileprivate let jobVariants: [Job.Variant]
    
    private lazy var internalQueue: DispatchQueue = {
        let result: DispatchQueue = DispatchQueue(
            label: self.queueContext,
            qos: self.qosClass,
            attributes: (self.executionType == .concurrent ? [.concurrent] : []),
            autoreleaseFrequency: .inherit,
            target: nil
        )
        result.setSpecific(key: queueKey, value: queueContext)
        
        return result
    }()
    
    private var executorMap: Atomic<[Job.Variant: JobExecutor.Type]> = Atomic([:])
    fileprivate var canStart: ((JobQueue?) -> Bool)?
    fileprivate var onQueueDrained: (() -> ())?
    fileprivate var hasStartedAtLeastOnce: Atomic<Bool> = Atomic(false)
    fileprivate var isRunning: Atomic<Bool> = Atomic(false)
    fileprivate var pendingJobsQueue: Atomic<[Job]> = Atomic([])
    
    private var nextTrigger: Atomic<Trigger?> = Atomic(nil)
    private var jobCallbacks: Atomic<[Int64: [(JobRunner.JobResult) -> ()]]> = Atomic([:])
    private var currentlyRunningJobIds: Atomic<Set<Int64>> = Atomic([])
    private var currentlyRunningJobInfo: Atomic<[Int64: JobRunner.JobInfo]> = Atomic([:])
    private var deferLoopTracker: Atomic<[Int64: (count: Int, times: [TimeInterval])]> = Atomic([:])
    private let maxDeferralsPerSecond: Int
    
    fileprivate var hasPendingJobs: Bool { !pendingJobsQueue.wrappedValue.isEmpty }
    
    // MARK: - Initialization
    
    fileprivate init(
        type: QueueType,
        executionType: ExecutionType,
        qos: DispatchQoS,
        isTestingJobRunner: Bool,
        jobVariants: [Job.Variant]
    ) {
        self.type = type
        self.executionType = executionType
        self.queueContext = "JobQueue-\(type.name)"
        self.qosClass = qos
        self.maxDeferralsPerSecond = (isTestingJobRunner ? 10 : 1)  // Allow for tripping the defer loop in tests
        self.jobVariants = jobVariants
    }
    
    // MARK: - Hashable
    
    public func hash(into hasher: inout Hasher) {
        id.hash(into: &hasher)
    }
    
    public static func == (lhs: JobQueue, rhs: JobQueue) -> Bool {
        return (lhs.id == rhs.id)
    }
    
    // MARK: - Configuration
    
    fileprivate func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant) {
        executorMap.mutate { $0[variant] = executor }
    }
    
    // MARK: - Execution
    
    fileprivate func add(
        _ db: Database,
        job: Job,
        canStartJob: Bool,
        using dependencies: Dependencies
    ) {
        // Check if the job should be added to the queue
        guard
            canStartJob,
            job.behaviour != .runOnceNextLaunch,
            job.nextRunTimestamp <= dependencies.dateNow.timeIntervalSince1970
        else { return }
        guard job.id != nil else {
            SNLog("[JobRunner] Prevented attempt to add \(job.variant) job without id to queue")
            return
        }
        
        pendingJobsQueue.mutate { $0.append(job) }
        
        // If this is a concurrent queue then we should immediately start the next job
        guard executionType == .concurrent else { return }
        
        // Ensure that the database commit has completed and then trigger the next job to run (need
        // to ensure any interactions have been correctly inserted first)
        db.afterNextTransactionNestedOnce(dedupeId: "JobRunner-Add: \(job.variant)") { [weak self] _ in
            self?.runNextJob(using: dependencies)
        }
    }
    
    /// Upsert a job onto the queue, if the queue isn't currently running and 'canStartJob' is true then this will start
    /// the JobRunner
    ///
    /// **Note:** If the job has a `behaviour` of `runOnceNextLaunch` or the `nextRunTimestamp`
    /// is in the future then the job won't be started
    fileprivate func upsert(
        _ db: Database,
        job: Job,
        canStartJob: Bool,
        using dependencies: Dependencies
    ) {
        guard let jobId: Int64 = job.id else {
            SNLog("[JobRunner] Prevented attempt to upsert \(job.variant) job without id to queue")
            return
        }
        
        // Lock the pendingJobsQueue while checking the index and inserting to ensure we don't run into
        // any multi-threading shenanigans
        //
        // Note: currently running jobs are removed from the pendingJobsQueue so we don't need to check
        // the 'jobsCurrentlyRunning' set
        var didUpdateExistingJob: Bool = false
        
        pendingJobsQueue.mutate { queue in
            if let jobIndex: Array<Job>.Index = queue.firstIndex(where: { $0.id == jobId }) {
                queue[jobIndex] = job
                didUpdateExistingJob = true
            }
        }
        
        // If we didn't update an existing job then we need to add it to the pendingJobsQueue
        guard !didUpdateExistingJob else { return }
        
        add(db, job: job, canStartJob: canStartJob, using: dependencies)
    }
    
    fileprivate func insert(_ job: Job, before otherJob: Job) {
        guard job.id != nil else {
            SNLog("[JobRunner] Prevented attempt to insert \(job.variant) job without id to queue")
            return
        }
        
        // Insert the job before the current job (re-adding the current job to
        // the start of the pendingJobsQueue if it's not in there) - this will mean the new
        // job will run and then the otherJob will run (or run again) once it's
        // done
        pendingJobsQueue.mutate {
            guard let otherJobIndex: Int = $0.firstIndex(of: otherJob) else {
                $0.insert(contentsOf: [job, otherJob], at: 0)
                return
            }
            
            $0.insert(job, at: otherJobIndex)
        }
    }
    
    fileprivate func appDidFinishLaunching(
        with jobs: [Job],
        canStart: Bool,
        using dependencies: Dependencies
    ) {
        pendingJobsQueue.mutate { $0.append(contentsOf: jobs) }
        
        // Start the job runner if needed
        if canStart && !isRunning.wrappedValue {
            start(using: dependencies)
        }
    }
    
    fileprivate func appDidBecomeActive(
        with jobs: [Job],
        canStart: Bool,
        using dependencies: Dependencies
    ) {
        let currentlyRunningJobIds: Set<Int64> = currentlyRunningJobIds.wrappedValue
        
        pendingJobsQueue.mutate { queue in
            // Avoid re-adding jobs to the queue that are already in it (this can
            // happen if the user sends the app to the background before the 'onActive'
            // jobs and then brings it back to the foreground)
            let jobsNotAlreadyInQueue: [Job] = jobs
                .filter { job in
                    !currentlyRunningJobIds.contains(job.id ?? -1) &&
                    !queue.contains(where: { $0.id == job.id })
                }
            
            queue.append(contentsOf: jobsNotAlreadyInQueue)
        }
        
        // Start the job runner if needed
        if canStart && !isRunning.wrappedValue {
            start(using: dependencies)
        }
    }
    
    fileprivate func infoForAllCurrentlyRunningJobs() -> [Int64: JobRunner.JobInfo] {
        return currentlyRunningJobInfo.wrappedValue
    }
    
    fileprivate func afterCurrentlyRunningJob(_ jobId: Int64, callback: @escaping (JobRunner.JobResult) -> ()) {
        guard currentlyRunningJobIds.wrappedValue.contains(jobId) else { return callback(.notFound) }
        
        jobCallbacks.mutate { jobCallbacks in
            jobCallbacks[jobId] = (jobCallbacks[jobId] ?? []).appending(callback)
        }
    }
    
    fileprivate func hasPendingOrRunningJobWith(
        threadId: String? = nil,
        interactionId: Int64? = nil,
        detailsData: Data? = nil
    ) -> Bool {
        let pendingJobs: [Job] = pendingJobsQueue.wrappedValue
        let currentlyRunningJobInfo: [Int64: JobRunner.JobInfo] = currentlyRunningJobInfo.wrappedValue
        var possibleJobIds: Set<Int64> = Set(currentlyRunningJobInfo.keys)
            .inserting(contentsOf: pendingJobs.compactMap { $0.id }.asSet())
        
        // Remove any which don't have the matching threadId (if provided)
        if let targetThreadId: String = threadId {
            let pendingJobIdsWithWrongThreadId: Set<Int64> = pendingJobs
                .filter { $0.threadId != targetThreadId }
                .compactMap { $0.id }
                .asSet()
            let runningJobIdsWithWrongThreadId: Set<Int64> = currentlyRunningJobInfo
                .filter { _, info -> Bool in info.threadId != targetThreadId }
                .map { key, _ in key }
                .asSet()
            
            possibleJobIds = possibleJobIds
                .subtracting(pendingJobIdsWithWrongThreadId)
                .subtracting(runningJobIdsWithWrongThreadId)
        }
        
        // Remove any which don't have the matching interactionId (if provided)
        if let targetInteractionId: Int64 = interactionId {
            let pendingJobIdsWithWrongInteractionId: Set<Int64> = pendingJobs
                .filter { $0.interactionId != targetInteractionId }
                .compactMap { $0.id }
                .asSet()
            let runningJobIdsWithWrongInteractionId: Set<Int64> = currentlyRunningJobInfo
                .filter { _, info -> Bool in info.interactionId != targetInteractionId }
                .map { key, _ in key }
                .asSet()
            
            possibleJobIds = possibleJobIds
                .subtracting(pendingJobIdsWithWrongInteractionId)
                .subtracting(runningJobIdsWithWrongInteractionId)
        }
        
        // Remove any which don't have the matching details (if provided)
        if let targetDetailsData: Data = detailsData {
            let pendingJobIdsWithWrongDetailsData: Set<Int64> = pendingJobs
                .filter { $0.details != targetDetailsData }
                .compactMap { $0.id }
                .asSet()
            let runningJobIdsWithWrongDetailsData: Set<Int64> = currentlyRunningJobInfo
                .filter { _, info -> Bool in info.detailsData != detailsData }
                .map { key, _ in key }
                .asSet()
            
            possibleJobIds = possibleJobIds
                .subtracting(pendingJobIdsWithWrongDetailsData)
                .subtracting(runningJobIdsWithWrongDetailsData)
        }
        
        return !possibleJobIds.isEmpty
    }
    
    fileprivate func removePendingJob(_ jobId: Int64) {
        pendingJobsQueue.mutate { queue in
            queue = queue.filter { $0.id != jobId }
        }
    }
    
    // MARK: - Job Running
    
    fileprivate func start(
        forceWhenAlreadyRunning: Bool = false,
        using dependencies: Dependencies
    ) {
        // Only start if the JobRunner is allowed to start the queue
        guard canStart?(self) == true else { return }
        guard forceWhenAlreadyRunning || !isRunning.wrappedValue else { return }
        
        // The JobRunner runs synchronously we need to ensure this doesn't start
        // on the main thread (if it is on the main thread then swap to a different thread)
        guard DispatchQueue.with(key: queueKey, matches: queueContext, using: dependencies) else {
            internalQueue.async(using: dependencies) { [weak self] in
                self?.start(using: dependencies)
            }
            return
        }
        
        // Flag the JobRunner as running (to prevent something else from trying to start it
        // and messing with the execution behaviour)
        var wasAlreadyRunning: Bool = false
        isRunning.mutate { isRunning in
            wasAlreadyRunning = isRunning
            isRunning = true
        }
        hasStartedAtLeastOnce.mutate { $0 = true }
        
        // Get any pending jobs
        let jobVariants: [Job.Variant] = self.jobVariants
        let jobIdsAlreadyRunning: Set<Int64> = currentlyRunningJobIds.wrappedValue
        let jobsAlreadyInQueue: Set<Int64> = pendingJobsQueue.wrappedValue.compactMap { $0.id }.asSet()
        let jobsToRun: [Job] = dependencies.storage.read(using: dependencies) { db in
            try Job
                .filterPendingJobs(
                    variants: jobVariants,
                    excludeFutureJobs: true,
                    includeJobsWithDependencies: false
                )
                .filter(!jobIdsAlreadyRunning.contains(Job.Columns.id)) // Exclude jobs already running
                .filter(!jobsAlreadyInQueue.contains(Job.Columns.id))   // Exclude jobs already in the queue
                .fetchAll(db)
        }
        .defaulting(to: [])
        
        // Determine the number of jobs to run
        var jobCount: Int = 0
        
        pendingJobsQueue.mutate { queue in
            queue.append(contentsOf: jobsToRun)
            jobCount = queue.count
        }
        
        // If there are no pending jobs and nothing in the queue then schedule the JobRunner
        // to start again when the next scheduled job should start
        guard jobCount > 0 else {
            if jobIdsAlreadyRunning.isEmpty {
                isRunning.mutate { $0 = false }
                scheduleNextSoonestJob(using: dependencies)
            }
            return
        }
        
        // Run the first job in the pendingJobsQueue
        if !wasAlreadyRunning {
            SNLogNotTests("[JobRunner] Starting \(queueContext) with (\(jobCount) job\(jobCount != 1 ? "s" : ""))")
        }
        runNextJob(using: dependencies)
    }
    
    fileprivate func stopAndClearPendingJobs() {
        isRunning.mutate { $0 = false }
        pendingJobsQueue.mutate { $0 = [] }
        deferLoopTracker.mutate { $0 = [:] }
    }
    
    private func runNextJob(using dependencies: Dependencies) {
        // Ensure the queue is running (if we've stopped the queue then we shouldn't start the next job)
        guard isRunning.wrappedValue else { return }
        
        // Ensure this is running on the correct queue
        guard DispatchQueue.with(key: queueKey, matches: queueContext, using: dependencies) else {
            internalQueue.async(using: dependencies) { [weak self] in
                self?.runNextJob(using: dependencies)
            }
            return
        }
        guard let (nextJob, numJobsRemaining): (Job, Int) = pendingJobsQueue.mutate({ queue in queue.popFirst().map { ($0, queue.count) } }) else {
            // If it's a serial queue, or there are no more jobs running then update the 'isRunning' flag
            if executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty {
                isRunning.mutate { $0 = false }
            }
            
            // Always attempt to schedule the next soonest job (otherwise if enough jobs get started in rapid
            // succession then pending/failed jobs in the database may never get re-started in a concurrent queue)
            scheduleNextSoonestJob(using: dependencies)
            return
        }
        guard let jobExecutor: JobExecutor.Type = executorMap.wrappedValue[nextJob.variant] else {
            SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing executor")
            handleJobFailed(
                nextJob,
                error: JobRunnerError.executorMissing,
                permanentFailure: true,
                using: dependencies
            )
            return
        }
        guard !jobExecutor.requiresThreadId || nextJob.threadId != nil else {
            SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing required threadId")
            handleJobFailed(
                nextJob,
                error: JobRunnerError.requiredThreadIdMissing,
                permanentFailure: true,
                using: dependencies
            )
            return
        }
        guard !jobExecutor.requiresInteractionId || nextJob.interactionId != nil else {
            SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing required interactionId")
            handleJobFailed(
                nextJob,
                error: JobRunnerError.requiredInteractionIdMissing,
                permanentFailure: true,
                using: dependencies
            )
            return
        }
        guard nextJob.id != nil else {
            SNLog("[JobRunner] \(queueContext) Unable to run \(nextJob.variant) job due to missing id")
            handleJobFailed(
                nextJob,
                error: JobRunnerError.jobIdMissing,
                permanentFailure: false,
                using: dependencies
            )
            return
        }
        
        // If the 'nextRunTimestamp' for the job is in the future then don't run it yet
        guard nextJob.nextRunTimestamp <= dependencies.dateNow.timeIntervalSince1970 else {
            handleJobDeferred(nextJob, using: dependencies)
            return
        }
        
        // Check if the next job has any dependencies
        let dependencyInfo: (expectedCount: Int, jobs: Set<Job>) = dependencies.storage.read(using: dependencies) { db in
            let expectedDependencies: Set<JobDependencies> = try JobDependencies
                .filter(JobDependencies.Columns.jobId == nextJob.id)
                .fetchSet(db)
            let jobDependencies: Set<Job> = try Job
                .filter(ids: expectedDependencies.compactMap { $0.dependantId })
                .fetchSet(db)
            
            return (expectedDependencies.count, jobDependencies)
        }
        .defaulting(to: (0, []))
        
        guard dependencyInfo.jobs.count == dependencyInfo.expectedCount else {
            SNLog("[JobRunner] \(queueContext) found job with missing dependencies, removing the job")
            handleJobFailed(
                nextJob,
                error: JobRunnerError.missingDependencies,
                permanentFailure: true,
                using: dependencies
            )
            return
        }
        guard dependencyInfo.jobs.isEmpty else {
            SNLog("[JobRunner] \(queueContext) found job with \(dependencyInfo.jobs.count) dependencies, running those first")
            
            /// Remove all jobs this one is dependant on that aren't currently running from the queue and re-insert them at the start
            /// of the queue
            ///
            /// **Note:** We don't add the current job back the the queue because it should only be re-added if it's dependencies
            /// are successfully completed
            let currentlyRunningJobIds: [Int64] = Array(currentlyRunningJobIds.wrappedValue)
            let dependencyJobsNotCurrentlyRunning: [Job] = dependencyInfo.jobs
                .filter { job in !currentlyRunningJobIds.contains(job.id ?? -1) }
                .sorted { lhs, rhs in (lhs.id ?? -1) < (rhs.id ?? -1) }
            
            pendingJobsQueue.mutate { queue in
                queue = queue
                    .filter { !dependencyJobsNotCurrentlyRunning.contains($0) }
                    .inserting(contentsOf: dependencyJobsNotCurrentlyRunning, at: 0)
            }
            handleJobDeferred(nextJob, using: dependencies)
            return
        }
        
        // Update the state to indicate the particular job is running
        //
        // Note: We need to store 'numJobsRemaining' in it's own variable because
        // the 'SNLog' seems to dispatch to it's own queue which ends up getting
        // blocked by the JobRunner's queue becuase 'jobQueue' is Atomic
        var numJobsRunning: Int = 0
        nextTrigger.mutate { trigger in
            trigger?.invalidate()   // Need to invalidate to prevent a memory leak
            trigger = nil
        }
        currentlyRunningJobIds.mutate { currentlyRunningJobIds in
            currentlyRunningJobIds = currentlyRunningJobIds.inserting(nextJob.id)
            numJobsRunning = currentlyRunningJobIds.count
        }
        currentlyRunningJobInfo.mutate { currentlyRunningJobInfo in
            currentlyRunningJobInfo = currentlyRunningJobInfo.setting(
                nextJob.id,
                JobRunner.JobInfo(
                    variant: nextJob.variant,
                    threadId: nextJob.threadId,
                    interactionId: nextJob.interactionId,
                    detailsData: nextJob.details
                )
            )
        }
        SNLog("[JobRunner] \(queueContext) started \(nextJob.variant) job (\(executionType == .concurrent ? "\(numJobsRunning) currently running, " : "")\(numJobsRemaining) remaining)")
        
        /// As it turns out Combine doesn't plat too nicely with concurrent Dispatch Queues, in Combine events are dispatched asynchronously to
        /// the queue which means an odd situation can occasionally occur where the `finished` event can actually run before the `output`
        /// event - this can result in unexpected behaviours (for more information see https://github.com/groue/GRDB.swift/issues/1334)
        ///
        /// Due to this if a job is meant to run on a concurrent queue then we actually want to create a temporary serial queue just for the execution
        /// of that job
        let targetQueue: DispatchQueue = {
            guard executionType == .concurrent else { return internalQueue }
            
            return DispatchQueue(
                label: "\(self.queueContext)-serial",
                qos: self.qosClass,
                attributes: [],
                autoreleaseFrequency: .inherit,
                target: nil
            )
        }()
        
        jobExecutor.run(
            nextJob,
            queue: targetQueue,
            success: handleJobSucceeded,
            failure: handleJobFailed,
            deferred: handleJobDeferred,
            using: dependencies
        )
        
        // If this queue executes concurrently and there are still jobs remaining then immediately attempt
        // to start the next job
        if executionType == .concurrent && numJobsRemaining > 0 {
            internalQueue.async(using: dependencies) { [weak self] in
                self?.runNextJob(using: dependencies)
            }
        }
    }
    
    private func scheduleNextSoonestJob(using dependencies: Dependencies) {
        let jobVariants: [Job.Variant] = self.jobVariants
        let jobIdsAlreadyRunning: Set<Int64> = currentlyRunningJobIds.wrappedValue
        let nextJobTimestamp: TimeInterval? = dependencies.storage.read(using: dependencies) { db in
            try Job
                .filterPendingJobs(
                    variants: jobVariants,
                    excludeFutureJobs: false,
                    includeJobsWithDependencies: false
                )
                .select(.nextRunTimestamp)
                .filter(!jobIdsAlreadyRunning.contains(Job.Columns.id)) // Exclude jobs already running
                .asRequest(of: TimeInterval.self)
                .fetchOne(db)
        }
        
        // If there are no remaining jobs or the JobRunner isn't allowed to start any queues then trigger
        // the 'onQueueDrained' callback and stop
        guard let nextJobTimestamp: TimeInterval = nextJobTimestamp, canStart?(self) == true else {
            if executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty {
                self.onQueueDrained?()
            }
            return
        }
        
        // If the next job isn't scheduled in the future then just restart the JobRunner immediately
        let secondsUntilNextJob: TimeInterval = (nextJobTimestamp - dependencies.dateNow.timeIntervalSince1970)
        
        guard secondsUntilNextJob > 0 else {
            // Only log that the queue is getting restarted if this queue had actually been about to stop
            if executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty {
                let timingString: String = (nextJobTimestamp == 0 ?
                    "that should be in the queue" :
                    "scheduled \(Int(ceil(abs(secondsUntilNextJob)))) second\(Int(ceil(abs(secondsUntilNextJob))) == 1 ? "" : "s") ago"
                )
                SNLog("[JobRunner] Restarting \(queueContext) immediately for job \(timingString)")
            }
            
            // Trigger the 'start' function to load in any pending jobs that aren't already in the
            // queue (for concurrent queues we want to force them to load in pending jobs and add
            // them to the queue regardless of whether the queue is already running)
            internalQueue.async(using: dependencies) { [weak self] in
                self?.start(forceWhenAlreadyRunning: (self?.executionType == .concurrent), using: dependencies)
            }
            return
        }
        
        // Only schedule a trigger if this queue has actually completed
        guard executionType != .concurrent || currentlyRunningJobIds.wrappedValue.isEmpty else { return }
        
        // Setup a trigger
        SNLog("[JobRunner] Stopping \(queueContext) until next job in \(Int(ceil(abs(secondsUntilNextJob)))) second\(Int(ceil(abs(secondsUntilNextJob))) == 1 ? "" : "s")")
        nextTrigger.mutate { trigger in
            trigger?.invalidate()   // Need to invalidate the old trigger to prevent a memory leak
            trigger = Trigger.create(queue: self, timestamp: nextJobTimestamp, using: dependencies)
        }
    }
    
    // MARK: - Handling Results

    /// This function is called when a job succeeds
    private func handleJobSucceeded(
        _ job: Job,
        shouldStop: Bool,
        using dependencies: Dependencies
    ) {
        /// Retrieve the dependant jobs first (the `JobDependecies` table has cascading deletion when the original `Job` is
        /// removed so we need to retrieve these records before that happens)
        let dependantJobs: [Job] = dependencies.storage
            .read(using: dependencies) { db in try job.dependantJobs.fetchAll(db) }
            .defaulting(to: [])
        
        switch job.behaviour {
            case .runOnce, .runOnceNextLaunch:
                dependencies.storage.write(using: dependencies) { db in
                    /// Since this job has been completed we can update the dependencies so other job that were dependant
                    /// on this one can be run
                    _ = try JobDependencies
                        .filter(JobDependencies.Columns.dependantId == job.id)
                        .deleteAll(db)
                    
                    _ = try job.delete(db)
                }
                
            case .recurring where shouldStop == true:
                dependencies.storage.write(using: dependencies) { db in
                    /// Since this job has been completed we can update the dependencies so other job that were dependant
                    /// on this one can be run
                    _ = try JobDependencies
                        .filter(JobDependencies.Columns.dependantId == job.id)
                        .deleteAll(db)
                    
                    _ = try job.delete(db)
                }
                
            /// For `recurring` jobs which have already run, they should automatically run again but we want at least 1 second
            /// to pass before doing so - the job itself should really update it's own `nextRunTimestamp` (this is just a safety net)
            case .recurring where job.nextRunTimestamp <= dependencies.dateNow.timeIntervalSince1970:
                guard let jobId: Int64 = job.id else { break }
                
                dependencies.storage.write(using: dependencies) { db in
                    _ = try Job
                        .filter(id: jobId)
                        .updateAll(
                            db,
                            Job.Columns.failureCount.set(to: 0),
                            Job.Columns.nextRunTimestamp.set(to: (dependencies.dateNow.timeIntervalSince1970 + 1))
                        )
                }
                
            /// For `recurringOnLaunch/Active` jobs which have already run but failed once, we need to clear their
            /// `failureCount` and `nextRunTimestamp` to prevent them from endlessly running over and over again
            case .recurringOnLaunch, .recurringOnActive:
                guard
                    let jobId: Int64 = job.id,
                    job.failureCount != 0 &&
                    job.nextRunTimestamp > TimeInterval.leastNonzeroMagnitude
                else { break }
                
                dependencies.storage.write(using: dependencies) { db in
                    _ = try Job
                        .filter(id: jobId)
                        .updateAll(
                            db,
                            Job.Columns.failureCount.set(to: 0),
                            Job.Columns.nextRunTimestamp.set(to: 0)
                        )
                }
            
            default: break
        }
        
        /// Now that the job has been completed we want to insert any jobs that were dependant on it, that aren't already running
        /// to the start of the queue (the most likely case is that we want an entire job chain to be completed at the same time rather
        /// than being blocked by other unrelated jobs)
        ///
        /// **Note:** If any of these `dependantJobs` have other dependencies then when they attempt to start they will be
        /// removed from the queue, replaced by their dependencies
        if !dependantJobs.isEmpty {
            let currentlyRunningJobIds: [Int64] = Array(currentlyRunningJobIds.wrappedValue)
            let dependantJobsNotCurrentlyRunning: [Job] = dependantJobs
                .filter { job in !currentlyRunningJobIds.contains(job.id ?? -1) }
                .sorted { lhs, rhs in (lhs.id ?? -1) < (rhs.id ?? -1) }
            
            pendingJobsQueue.mutate { queue in
                queue = queue
                    .filter { !dependantJobsNotCurrentlyRunning.contains($0) }
                    .inserting(contentsOf: dependantJobsNotCurrentlyRunning, at: 0)
            }
        }
        
        // Perform job cleanup and start the next job
        performCleanUp(for: job, result: .succeeded, using: dependencies)
        internalQueue.async(using: dependencies) { [weak self] in
            self?.runNextJob(using: dependencies)
        }
    }

    /// This function is called when a job fails, if it's wasn't a permanent failure then the 'failureCount' for the job will be incremented and it'll
    /// be re-run after a retry interval has passed
    private func handleJobFailed(
        _ job: Job,
        error: Error?,
        permanentFailure: Bool,
        using dependencies: Dependencies
    ) {
        guard dependencies.storage.read(using: dependencies, { db in try Job.exists(db, id: job.id ?? -1) }) == true else {
            SNLog("[JobRunner] \(queueContext) \(job.variant) job canceled")
            performCleanUp(for: job, result: .failed, using: dependencies)
            
            internalQueue.async(using: dependencies) { [weak self] in
                self?.runNextJob(using: dependencies)
            }
            return
        }
        
        // If this is the blocking queue and a "blocking" job failed then rerun it
        // immediately (in this case we don't trigger any job callbacks because the
        // job isn't actually done, it's going to try again immediately)
        if self.type == .blocking && job.shouldBlock {
            SNLog("[JobRunner] \(queueContext) \(job.variant) job failed; retrying immediately")
            
            // If it was a possible deferral loop then we don't actually want to
            // retry the job (even if it's a blocking one, this gives a small chance
            // that the app could continue to function)
            let wasPossibleDeferralLoop: Bool = {
                if let error = error, case JobRunnerError.possibleDeferralLoop = error { return true }
                
                return false
            }()
            performCleanUp(
                for: job,
                result: .failed,
                shouldTriggerCallbacks: wasPossibleDeferralLoop,
                using: dependencies
            )
            
            // Only add it back to the queue if it wasn't a deferral loop
            if !wasPossibleDeferralLoop {
                pendingJobsQueue.mutate { $0.insert(job, at: 0) }
            }
            
            internalQueue.async(using: dependencies) { [weak self] in
                self?.runNextJob(using: dependencies)
            }
            return
        }
        
        // Get the max failure count for the job (a value of '-1' means it will retry indefinitely)
        let maxFailureCount: Int = (executorMap.wrappedValue[job.variant]?.maxFailureCount ?? 0)
        let nextRunTimestamp: TimeInterval = (dependencies.dateNow.timeIntervalSince1970 + JobRunner.getRetryInterval(for: job))
        var dependantJobIds: [Int64] = []
        var failureText: String = "failed"
        
        dependencies.storage.write(using: dependencies) { db in
            /// Retrieve a list of dependant jobs so we can clear them from the queue
            dependantJobIds = try job.dependantJobs
                .select(.id)
                .asRequest(of: Int64.self)
                .fetchAll(db)

            /// Delete/update the failed jobs and any dependencies
            let updatedFailureCount: UInt = (job.failureCount + 1)
        
            guard
                !permanentFailure && (
                    maxFailureCount < 0 ||
                    updatedFailureCount <= maxFailureCount
                )
            else {
                failureText = (maxFailureCount >= 0 && updatedFailureCount > maxFailureCount ?
                    "failed permanently; too many retries" :
                    "failed permanently"
                )
                
                // If the job permanently failed or we have performed all of our retry attempts
                // then delete the job and all of it's dependant jobs (it'll probably never succeed)
                _ = try job.dependantJobs
                    .deleteAll(db)

                _ = try job.delete(db)
                return
            }
            
            failureText = "failed; scheduling retry (failure count is \(updatedFailureCount))"
            
            _ = try job
                .with(
                    failureCount: updatedFailureCount,
                    nextRunTimestamp: nextRunTimestamp
                )
                .saved(db)
            
            // Update the failureCount and nextRunTimestamp on dependant jobs as well (update the
            // 'nextRunTimestamp' value to be 1ms later so when the queue gets regenerated they'll
            // come after the dependency)
            try job.dependantJobs
                .updateAll(
                    db,
                    Job.Columns.failureCount.set(to: updatedFailureCount),
                    Job.Columns.nextRunTimestamp.set(to: (nextRunTimestamp + (1 / 1000)))
                )
        }
        
        /// Remove any dependant jobs from the queue (shouldn't be in there but filter the queue just in case so we don't try
        /// to run a deleted job or get stuck in a loop of trying to run dependencies indefinitely)
        if !dependantJobIds.isEmpty {
            pendingJobsQueue.mutate { queue in
                queue = queue.filter { !dependantJobIds.contains($0.id ?? -1) }
            }
        }
        
        SNLog("[JobRunner] \(queueContext) \(job.variant) job \(failureText)")
        performCleanUp(for: job, result: .failed, using: dependencies)
        internalQueue.async(using: dependencies) { [weak self] in
            self?.runNextJob(using: dependencies)
        }
    }
    
    /// This function is called when a job neither succeeds or fails (this should only occur if the job has specific logic that makes it dependant
    /// on other jobs, and it should automatically manage those dependencies)
    public func handleJobDeferred(
        _ job: Job,
        using dependencies: Dependencies
    ) {
        var stuckInDeferLoop: Bool = false
        
        deferLoopTracker.mutate {
            guard let lastRecord: (count: Int, times: [TimeInterval]) = $0[job.id] else {
                $0 = $0.setting(
                    job.id,
                    (1, [dependencies.dateNow.timeIntervalSince1970])
                )
                return
            }
            
            let timeNow: TimeInterval = dependencies.dateNow.timeIntervalSince1970
            stuckInDeferLoop = (
                lastRecord.count >= JobQueue.deferralLoopThreshold &&
                (timeNow - lastRecord.times[0]) < CGFloat(lastRecord.count * maxDeferralsPerSecond)
            )
            
            $0 = $0.setting(
                job.id,
                (
                    lastRecord.count + 1,
                    // Only store the last 'deferralLoopThreshold' times to ensure we aren't running faster
                    // than one loop per second
                    lastRecord.times.suffix(JobQueue.deferralLoopThreshold - 1) + [timeNow]
                )
            )
        }
        
        // It's possible (by introducing bugs) to create a loop where a Job tries to run and immediately
        // defers itself but then attempts to run again (resulting in an infinite loop); this won't block
        // the app since it's on a background thread but can result in 100% of a CPU being used (and a
        // battery drain)
        //
        // This code will maintain an in-memory store for any jobs which are deferred too quickly (ie.
        // more than 'deferralLoopThreshold' times within 'deferralLoopThreshold' seconds)
        guard !stuckInDeferLoop else {
            deferLoopTracker.mutate { $0 = $0.removingValue(forKey: job.id) }
            handleJobFailed(
                job,
                error: JobRunnerError.possibleDeferralLoop,
                permanentFailure: false,
                using: dependencies
            )
            return
        }
        
        performCleanUp(for: job, result: .deferred, using: dependencies)
        internalQueue.async(using: dependencies) { [weak self] in
            self?.runNextJob(using: dependencies)
        }
    }
    
    private func performCleanUp(
        for job: Job,
        result: JobRunner.JobResult,
        shouldTriggerCallbacks: Bool = true,
        using dependencies: Dependencies
    ) {
        // The job is removed from the queue before it runs so all we need to to is remove it
        // from the 'currentlyRunning' set
        currentlyRunningJobIds.mutate { $0 = $0.removing(job.id) }
        currentlyRunningJobInfo.mutate { $0 = $0.removingValue(forKey: job.id) }
        
        guard shouldTriggerCallbacks else { return }
        
        // Run any job callbacks now that it's done
        var jobCallbacksToRun: [(JobRunner.JobResult) -> ()] = []
        jobCallbacks.mutate { jobCallbacks in
            jobCallbacksToRun = (jobCallbacks[job.id] ?? [])
            jobCallbacks = jobCallbacks.removingValue(forKey: job.id)
        }
        
        DispatchQueue.global(qos: .default).async(using: dependencies) {
            jobCallbacksToRun.forEach { $0(result) }
        }
    }
}

// MARK: - JobRunner Singleton
// FIXME: Remove this once the jobRunner is dependency injected everywhere correctly
public extension JobRunner {
    internal static let instance: JobRunner = JobRunner()
    
    // MARK: - Static Access
    
    static func setExecutor(_ executor: JobExecutor.Type, for variant: Job.Variant) {
        instance.setExecutor(executor, for: variant)
    }
    
    static func appDidFinishLaunching(using dependencies: Dependencies = Dependencies()) {
        instance.appDidFinishLaunching(using: dependencies)
    }
    
    static func appDidBecomeActive(using dependencies: Dependencies = Dependencies()) {
        instance.appDidBecomeActive(using: dependencies)
    }
    
    static func afterBlockingQueue(callback: @escaping () -> ()) {
        instance.afterBlockingQueue(callback: callback)
    }
    
    /// Add a job onto the queue, if the queue isn't currently running and 'canStartJob' is true then this will start
    /// the JobRunner
    ///
    /// **Note:** If the job has a `behaviour` of `runOnceNextLaunch` or the `nextRunTimestamp`
    /// is in the future then the job won't be started
    static func add(
        _ db: Database,
        job: Job?,
        canStartJob: Bool = true,
        using dependencies: Dependencies = Dependencies()
    ) { instance.add(db, job: job, canStartJob: canStartJob, using: dependencies) }
    
    /// Upsert a job onto the queue, if the queue isn't currently running and 'canStartJob' is true then this will start
    /// the JobRunner
    ///
    /// **Note:** If the job has a `behaviour` of `runOnceNextLaunch` or the `nextRunTimestamp`
    /// is in the future then the job won't be started
    static func upsert(
        _ db: Database,
        job: Job?,
        canStartJob: Bool = true,
        using dependencies: Dependencies = Dependencies()
    ) { instance.upsert(db, job: job, canStartJob: canStartJob, using: dependencies) }
    
    @discardableResult static func insert(
        _ db: Database,
        job: Job?,
        before otherJob: Job
    ) -> (Int64, Job)? { instance.insert(db, job: job, before: otherJob) }
    
    /// Calling this will clear the JobRunner queues and stop it from running new jobs, any currently executing jobs will continue to run
    /// though (this means if we suspend the database it's likely that any currently running jobs will fail to complete and fail to record their
    /// failure - they _should_ be picked up again the next time the app is launched)
    static func stopAndClearPendingJobs(
        exceptForVariant: Job.Variant? = nil,
        onComplete: (() -> ())? = nil
    ) { instance.stopAndClearPendingJobs(exceptForVariant: exceptForVariant, onComplete: onComplete) }
    
    static func isCurrentlyRunning(_ job: Job?) -> Bool {
        return instance.isCurrentlyRunning(job)
    }
    
    static func afterCurrentlyRunningJob(_ job: Job?, callback: @escaping (JobResult) -> ()) {
        instance.afterCurrentlyRunningJob(job, callback: callback)
    }
    
    static func removePendingJob(_ job: Job?) {
        instance.removePendingJob(job)
    }
}