<commit_before>//
//  EventQueueService.swift
//  RoverData
//
//  Created by <NAME> on 2017-09-01.
//  Copyright © 2017 Rover Labs Inc. All rights reserved.
//

import os.log
import UIKit

public class EventQueue {
    let client: EventsClient
    let flushAt: Int
    let flushInterval: Double
    let maxBatchSize: Int
    let maxQueueSize: Int
    
    // Use an implicitly unwrapped optional to avoid a circular dependency during injection
    var contextProvider: ContextProvider!
    
    let serialQueue: Foundation.OperationQueue = {
        let q = Foundation.OperationQueue()
        q.maxConcurrentOperationCount = 1
        return q
    }()
    
    struct WeakObserver {
        private(set) public weak var value: EventQueueObserver?
        
        public init(_ value: EventQueueObserver) {
            self.value = value
        }
    }
    
    var observers = [WeakObserver]()
    
    // The following variables comprise the state of the EventQueueService and should only be modified from within an operation on the serial queue
    
    var eventQueue = [Event]()
    var uploadTask: URLSessionTask?
    var timer: Timer?
    
    #if swift(>=4.2)
    var backgroundTask = UIBackgroundTaskIdentifier.invalid
    #else
    var backgroundTask = UIBackgroundTaskInvalid
    #endif
    
    var cache: URL? {
        return FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask).first?.appendingPathComponent("events").appendingPathExtension("plist")
    }
    
    init(client: EventsClient, flushAt: Int, flushInterval: Double, maxBatchSize: Int, maxQueueSize: Int) {
        self.client = client
        self.flushAt = flushAt
        self.flushInterval = flushInterval
        self.maxBatchSize = maxBatchSize
        self.maxQueueSize = maxQueueSize
    }
    
    public func restore() {
        restoreEvents()
        
        #if swift(>=4.2)
        NotificationCenter.default.addObserver(forName: UIApplication.didBecomeActiveNotification, object: nil, queue: nil) { _ in
            self.startTimer()
        }
        
        NotificationCenter.default.addObserver(forName: UIApplication.willResignActiveNotification, object: nil, queue: nil) { _ in
            self.stopTimer()
        }
        
        NotificationCenter.default.addObserver(forName: UIApplication.didEnterBackgroundNotification, object: nil, queue: nil) { _ in
            self.beginBackgroundTask()
            self.flushEvents()
        }
        #else
        NotificationCenter.default.addObserver(forName: .UIApplicationDidBecomeActive, object: nil, queue: nil) { _ in
            self.startTimer()
        }
        
        NotificationCenter.default.addObserver(forName: .UIApplicationWillResignActive, object: nil, queue: nil) { _ in
            self.stopTimer()
        }
        
        NotificationCenter.default.addObserver(forName: .UIApplicationDidEnterBackground, object: nil, queue: nil) { _ in
            self.beginBackgroundTask()
            self.flushEvents()
        }
        #endif
        
        if (UIApplication.shared.applicationState == .active) {
            startTimer()
        }
    }
    
    func restoreEvents() {
        serialQueue.addOperation {
            os_log("Restoring events from cache...", log: .events, type: .debug)
            
            guard let cache = self.cache else {
                os_log("Cache not found", log: .events, type: .error)
                return
            }
            
            if !FileManager.default.fileExists(atPath: cache.path) {
                os_log("Cache is empty, no events to restore", log: .events, type: .debug)
                return
            }
            
            do {
                let data = try Data(contentsOf: cache)
                self.eventQueue = try PropertyListDecoder().decode([Event].self, from: data)
                os_log("%d event(s) restored from cache", log: .events, type: .debug, self.eventQueue.count)
            } catch {
                os_log("Failed to restore events from cache: %@", log: .events, type: .error, error.localizedDescription)
            }
        }
    }
    
    public func addEvent(_ info: EventInfo) {
        let context = self.contextProvider.context
        
        serialQueue.addOperation {
            if self.eventQueue.count == self.maxQueueSize {
                os_log("Event queue is at capacity (%d) – removing oldest event", log: .events, type: .debug, self.maxQueueSize)
                self.eventQueue.remove(at: 0)
            }
            
            let event = Event(
                name: info.name,
                context: context,
                namespace: info.namespace,
                attributes: info.attributes,
                timestamp: info.timestamp ?? Date()
            )
            
            self.eventQueue.append(event)
            os_log("Added event to queue: %@", log: .events, type: .debug, event.name)
            os_log("Queue now contains %d event(s)", log: .events, type: .debug, self.eventQueue.count)
        }
        
        persistEvents()
        
        if UIApplication.shared.applicationState == .active {
            flushEvents(minBatchSize: flushAt)
        } else {
            flushEvents()
        }
        
        observers.compactMap({ $0.value }).forEach { $0.eventQueue(self, didAddEvent: info) }
    }
    
    func persistEvents() {
        serialQueue.addOperation {
            os_log("Persisting events to cache...", log: .events, type: .debug)
            
            guard let cache = self.cache else {
                os_log("Cache not found", log: .events, type: .error)
                return
            }
            
            do {<commit_msg>Checking app state must be on main thread<commit_after>//
//  EventQueueService.swift
//  RoverData
//
//  Created by <NAME> on 2017-09-01.
//  Copyright © 2017 Rover Labs Inc. All rights reserved.
//

import os.log
import UIKit

public class EventQueue {
    let client: EventsClient
    let flushAt: Int
    let flushInterval: Double
    let maxBatchSize: Int
    let maxQueueSize: Int
    
    // Use an implicitly unwrapped optional to avoid a circular dependency during injection
    var contextProvider: ContextProvider!
    
    let serialQueue: Foundation.OperationQueue = {
        let q = Foundation.OperationQueue()
        q.maxConcurrentOperationCount = 1
        return q
    }()
    
    struct WeakObserver {
        private(set) public weak var value: EventQueueObserver?
        
        public init(_ value: EventQueueObserver) {
            self.value = value
        }
    }
    
    var observers = [WeakObserver]()
    
    // The following variables comprise the state of the EventQueueService and should only be modified from within an operation on the serial queue
    
    var eventQueue = [Event]()
    var uploadTask: URLSessionTask?
    var timer: Timer?
    
    #if swift(>=4.2)
    var backgroundTask = UIBackgroundTaskIdentifier.invalid
    #else
    var backgroundTask = UIBackgroundTaskInvalid
    #endif
    
    var cache: URL? {
        return FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask).first?.appendingPathComponent("events").appendingPathExtension("plist")
    }
    
    init(client: EventsClient, flushAt: Int, flushInterval: Double, maxBatchSize: Int, maxQueueSize: Int) {
        self.client = client
        self.flushAt = flushAt
        self.flushInterval = flushInterval
        self.maxBatchSize = maxBatchSize
        self.maxQueueSize = maxQueueSize
    }
    
    public func restore() {
        restoreEvents()
        
        #if swift(>=4.2)
        NotificationCenter.default.addObserver(forName: UIApplication.didBecomeActiveNotification, object: nil, queue: nil) { _ in
            self.startTimer()
        }
        
        NotificationCenter.default.addObserver(forName: UIApplication.willResignActiveNotification, object: nil, queue: nil) { _ in
            self.stopTimer()
        }
        
        NotificationCenter.default.addObserver(forName: UIApplication.didEnterBackgroundNotification, object: nil, queue: nil) { _ in
            self.beginBackgroundTask()
            self.flushEvents()
        }
        #else
        NotificationCenter.default.addObserver(forName: .UIApplicationDidBecomeActive, object: nil, queue: nil) { _ in
            self.startTimer()
        }
        
        NotificationCenter.default.addObserver(forName: .UIApplicationWillResignActive, object: nil, queue: nil) { _ in
            self.stopTimer()
        }
        
        NotificationCenter.default.addObserver(forName: .UIApplicationDidEnterBackground, object: nil, queue: nil) { _ in
            self.beginBackgroundTask()
            self.flushEvents()
        }
        #endif
        
        if (UIApplication.shared.applicationState == .active) {
            startTimer()
        }
    }
    
    func restoreEvents() {
        serialQueue.addOperation {
            os_log("Restoring events from cache...", log: .events, type: .debug)
            
            guard let cache = self.cache else {
                os_log("Cache not found", log: .events, type: .error)
                return
            }
            
            if !FileManager.default.fileExists(atPath: cache.path) {
                os_log("Cache is empty, no events to restore", log: .events, type: .debug)
                return
            }
            
            do {
                let data = try Data(contentsOf: cache)
                self.eventQueue = try PropertyListDecoder().decode([Event].self, from: data)
                os_log("%d event(s) restored from cache", log: .events, type: .debug, self.eventQueue.count)
            } catch {
                os_log("Failed to restore events from cache: %@", log: .events, type: .error, error.localizedDescription)
            }
        }
    }
    
    public func addEvent(_ info: EventInfo) {
        let context = self.contextProvider.context
        
        serialQueue.addOperation {
            if self.eventQueue.count == self.maxQueueSize {
                os_log("Event queue is at capacity (%d) – removing oldest event", log: .events, type: .debug, self.maxQueueSize)
                self.eventQueue.remove(at: 0)
            }
            
            let event = Event(
                name: info.name,
                context: context,
                namespace: info.namespace,
                attributes: info.attributes,
                timestamp: info.timestamp ?? Date()
            )
            
            self.eventQueue.append(event)
            os_log("Added event to queue: %@", log: .events, type: .debug, event.name)
            os_log("Queue now contains %d event(s)", log: .events, type: .debug, self.eventQueue.count)
        }
        
        persistEvents()
        
        let onMainThread: (() -> Void) -> Void = { block in
            if Thread.isMainThread {
                block()
            } else {
                DispatchQueue.main.sync {
                    block()
                }
            }
        }
        
        onMainThread {
            if UIApplication.shared.applicationState == .active {
                flushEvents(minBatchSize: flushAt)
            } else {
                flushEvents()
            }
        }
        
        observers.compactMap({ $0.value }).forEach { $0.eventQueue(self, didAddEvent: info) }
    }
    
    func persistEvents() {
        serialQueue.addOperation {
            os_log("Persisting events to cache...", log: .events, type: .debug)
            
            guard let cache = self.cache else {
                os_log("Cache not found", log: .events, type: .error)
                return
            }
            
            do {
{'fraction_non_alphanumeric': 0.09117946974847042, 'fraction_numerical': 0.003399048266485384, 'mean_word_length': 1.8270478020658178, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1774404', 'n_tokens_mistral': 3131, 'n_tokens_neox': 3039, 'n_words': 792}