<commit_before>//
//  CloudKitSendStatusOperation.swift
//  Account
//
//  Created by <NAME> on 5/2/20.
//  Copyright © 2020 Ranchero Software, LLC. All rights reserved.
//

import Foundation
import Articles
import os.log
import RSCore
import RSWeb
import SyncDatabase

class CloudKitSendStatusOperation: MainThreadOperation {
	
	private var log = OSLog(subsystem: Bundle.main.bundleIdentifier!, category: "CloudKit")
	private let blockSize = 150
	
	// MainThreadOperation
	public var isCanceled = false
	public var id: Int?
	public weak var operationDelegate: MainThreadOperationDelegate?
	public var name: String? = "CloudKitSendStatusOperation"
	public var completionBlock: MainThreadOperation.MainThreadOperationCompletionBlock?

	private weak var account: Account?
	private weak var articlesZone: CloudKitArticlesZone?
	private weak var refreshProgress: DownloadProgress?
	private var showProgress: Bool
	private var database: SyncDatabase

	init(account: Account, articlesZone: CloudKitArticlesZone, refreshProgress: DownloadProgress, showProgress: Bool, database: SyncDatabase) {
		self.account = account
		self.articlesZone = articlesZone
		self.refreshProgress = refreshProgress
		self.showProgress = showProgress
		self.database = database
	}
	
	func run() {
		os_log(.debug, log: log, "Sending article statuses...")
		
		if showProgress {
			
			database.selectPendingCount() { result in
				switch result {
				case .success(let count):
					let ticks = count / self.blockSize
					self.refreshProgress?.addToNumberOfTasksAndRemaining(ticks)
					self.selectForProcessing()
				case .failure(let databaseError):
					os_log(.error, log: self.log, "Send status count pending error: %@.", databaseError.localizedDescription)
					self.operationDelegate?.cancelOperation(self)
				}
			}
			
		} else {
			
			selectForProcessing()
			
		}
		
	}
	
}

private extension CloudKitSendStatusOperation {
	
	func selectForProcessing() {
		database.selectForProcessing(limit: blockSize) { result in
			switch result {
			case .success(let syncStatuses):
				
				func stopProcessing() {
					if self.showProgress {
						self.refreshProgress?.completeTask()
					}
					os_log(.debug, log: self.log, "Done sending article statuses.")
					self.operationDelegate?.operationDidComplete(self)
				}
				
				guard syncStatuses.count > 0 else {
					stopProcessing()
					return
				}
				
				self.processStatuses(syncStatuses) { stop in
					if stop {
						stopProcessing()
					} else {
						self.selectForProcessing()
					}
				}
				
			case .failure(let databaseError):
				os_log(.error, log: self.log, "Send status error: %@.", databaseError.localizedDescription)
				self.operationDelegate?.cancelOperation(self)
			}
		}
	}
	
	func processStatuses(_ syncStatuses: [SyncStatus], completion: @escaping (Bool) -> Void) {
		guard let account = account, let articlesZone = articlesZone else {
			completion(true)
			return
		}
		
		let articleIDs = syncStatuses.map({ $0.articleID })
		account.fetchArticlesAsync(.articleIDs(Set(articleIDs))) { result in
			
			func processWithArticles(_ articles: Set<Article>) {
				
				let syncStatusesDict = Dictionary(grouping: syncStatuses, by: { $0.articleID })
				let articlesDict = articles.reduce(into: [String: Article]()) { result, article in
					result[article.articleID] = article
				}
				let statusUpdates = syncStatusesDict.compactMap { (key, value) in
					return CloudKitArticleStatusUpdate(articleID: key, statuses: value, article: articlesDict[key])
				}
				
				func done(_ stop: Bool) {
					// Don't clear the last one since we might have had additional ticks added
					if self.showProgress && self.refreshProgress?.numberRemaining ?? 0 > 1 {
						self.refreshProgress?.completeTask()
					}
					os_log(.debug, log: self.log, "Done sending article status block...")
					completion(stop)
				}
				
				// If this happens, we have somehow gotten into a state where we have new status records
				// but the articles didn't come back in the fetch.  Rather than crashing, we stop processing
				// and hope that it gets cleared up later.
				guard !statusUpdates.isEmpty else {
					done(true)
					return
				}
				
				articlesZone.modifyArticles(statusUpdates) { result in
					switch result {
					case .success:
						self.database.deleteSelectedForProcessing(statusUpdates.map({ $0.articleID })) { _ in
							done(false)
						}
					case .failure(let error):
						self.database.resetSelectedForProcessing(syncStatuses.map({ $0.articleID })) { _ in
							self.processAccountError(account, error)
							os_log(.error, log: self.log, "Send article status modify articles error: %@.", error.localizedDescription)
							completion(true)
						}
					}
				}
				
			}

			switch result {
			case .success(let articles):
				processWithArticles(articles)
			case .failure(let databaseError):
				self.database.resetSelectedForProcessing(syncStatuses.map({ $0.articleID })) { _ in
					os_log(.error, log: self.log, "Send article status fetch articles error: %@.", databaseError.localizedDescription)
					completion(true)
				}
			}

		}<commit_msg>Continue processing if we don't find any articles. Fixes #3213<commit_after>//
//  CloudKitSendStatusOperation.swift
//  Account
//
//  Created by <NAME> on 5/2/20.
//  Copyright © 2020 Ranchero Software, LLC. All rights reserved.
//

import Foundation
import Articles
import os.log
import RSCore
import RSWeb
import SyncDatabase

class CloudKitSendStatusOperation: MainThreadOperation {
	
	private var log = OSLog(subsystem: Bundle.main.bundleIdentifier!, category: "CloudKit")
	private let blockSize = 150
	
	// MainThreadOperation
	public var isCanceled = false
	public var id: Int?
	public weak var operationDelegate: MainThreadOperationDelegate?
	public var name: String? = "CloudKitSendStatusOperation"
	public var completionBlock: MainThreadOperation.MainThreadOperationCompletionBlock?

	private weak var account: Account?
	private weak var articlesZone: CloudKitArticlesZone?
	private weak var refreshProgress: DownloadProgress?
	private var showProgress: Bool
	private var database: SyncDatabase

	init(account: Account, articlesZone: CloudKitArticlesZone, refreshProgress: DownloadProgress, showProgress: Bool, database: SyncDatabase) {
		self.account = account
		self.articlesZone = articlesZone
		self.refreshProgress = refreshProgress
		self.showProgress = showProgress
		self.database = database
	}
	
	func run() {
		os_log(.debug, log: log, "Sending article statuses...")
		
		if showProgress {
			
			database.selectPendingCount() { result in
				switch result {
				case .success(let count):
					let ticks = count / self.blockSize
					self.refreshProgress?.addToNumberOfTasksAndRemaining(ticks)
					self.selectForProcessing()
				case .failure(let databaseError):
					os_log(.error, log: self.log, "Send status count pending error: %@.", databaseError.localizedDescription)
					self.operationDelegate?.cancelOperation(self)
				}
			}
			
		} else {
			
			selectForProcessing()
			
		}
		
	}
	
}

private extension CloudKitSendStatusOperation {
	
	func selectForProcessing() {
		database.selectForProcessing(limit: blockSize) { result in
			switch result {
			case .success(let syncStatuses):
				
				func stopProcessing() {
					self.database.resetAllSelectedForProcessing { _ in
						if self.showProgress {
							self.refreshProgress?.completeTask()
						}
						os_log(.debug, log: self.log, "Done sending article statuses.")
						self.operationDelegate?.operationDidComplete(self)
					}
				}
				
				guard syncStatuses.count > 0 else {
					stopProcessing()
					return
				}
				
				self.processStatuses(syncStatuses) { stop in
					if stop {
						stopProcessing()
					} else {
						self.selectForProcessing()
					}
				}
				
			case .failure(let databaseError):
				os_log(.error, log: self.log, "Send status error: %@.", databaseError.localizedDescription)
				self.operationDelegate?.cancelOperation(self)
			}
		}
	}
	
	func processStatuses(_ syncStatuses: [SyncStatus], completion: @escaping (Bool) -> Void) {
		guard let account = account, let articlesZone = articlesZone else {
			completion(true)
			return
		}
		
		let articleIDs = syncStatuses.map({ $0.articleID })
		account.fetchArticlesAsync(.articleIDs(Set(articleIDs))) { result in
			
			func processWithArticles(_ articles: Set<Article>) {
				
				let syncStatusesDict = Dictionary(grouping: syncStatuses, by: { $0.articleID })
				let articlesDict = articles.reduce(into: [String: Article]()) { result, article in
					result[article.articleID] = article
				}
				let statusUpdates = syncStatusesDict.compactMap { (key, value) in
					return CloudKitArticleStatusUpdate(articleID: key, statuses: value, article: articlesDict[key])
				}
				
				func done(_ stop: Bool) {
					// Don't clear the last one since we might have had additional ticks added
					if self.showProgress && self.refreshProgress?.numberRemaining ?? 0 > 1 {
						self.refreshProgress?.completeTask()
					}
					os_log(.debug, log: self.log, "Done sending article status block...")
					completion(stop)
				}
				
				// If this happens, we have somehow gotten into a state where we have new status records
				// but the articles didn't come back in the fetch.  Rather than crashing, we continue processing
				// and hope that it gets cleared up later.
				guard !statusUpdates.isEmpty else {
					done(false)
					return
				}
				
				articlesZone.modifyArticles(statusUpdates) { result in
					switch result {
					case .success:
						self.database.deleteSelectedForProcessing(statusUpdates.map({ $0.articleID })) { _ in
							done(false)
						}
					case .failure(let error):
						self.database.resetSelectedForProcessing(syncStatuses.map({ $0.articleID })) { _ in
							self.processAccountError(account, error)
							os_log(.error, log: self.log, "Send article status modify articles error: %@.", error.localizedDescription)
							completion(true)
						}
					}
				}
				
			}

			switch result {
			case .success(let articles):
				processWithArticles(articles)
			case .failure(let databaseError):
				self.database.resetSelectedForProcessing(syncStatuses.map({ $0.articleID })) { _ in
					os_log(.error, log: self.log, "Send article status fetch articles error: %@.", databaseError.localizedDescription)
					completion(true)
				}
			}

		}
{'fraction_non_alphanumeric': 0.09533449007133217, 'fraction_numerical': 0.004048582995951417, 'mean_word_length': 3.4394522892597346, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1792412', 'n_tokens_mistral': 3751, 'n_tokens_neox': 2872, 'n_words': 785}