Initial Commit
This commit is contained in:
Mark Gregory
@@ -0,0 +1,245 @@
|
||||
import Foundation
|
||||
import SwiftData
|
||||
|
||||
@MainActor
|
||||
final class BackgroundSyncManager: ObservableObject {
|
||||
enum Activity: Equatable {
|
||||
case idle
|
||||
case waiting(String)
|
||||
case running(String)
|
||||
|
||||
var title: String {
|
||||
switch self {
|
||||
case .idle: return "Idle"
|
||||
case .waiting(let name): return "Waiting · \(name)"
|
||||
case .running(let name): return "Running · \(name)"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Published private(set) var activity: Activity = .idle
|
||||
@Published private(set) var watchedJobCount = 0
|
||||
@Published private(set) var scheduledJobCount = 0
|
||||
@Published private(set) var isSyncAllRunning = false
|
||||
@Published private(set) var syncAllCompleted = 0
|
||||
@Published private(set) var syncAllTotal = 0
|
||||
|
||||
var activityDisplayTitle: String {
|
||||
if isSyncAllRunning {
|
||||
return "Sync All · \(min(syncAllCompleted + 1, syncAllTotal)) of \(syncAllTotal)"
|
||||
}
|
||||
return activity.title
|
||||
}
|
||||
|
||||
private let modelContext: ModelContext
|
||||
private let runner: RsyncRunner
|
||||
private let watcher = FileWatchService()
|
||||
private var timer: Timer?
|
||||
private var pending: [(UUID, SyncRunTrigger)] = []
|
||||
private var pendingIDs = Set<UUID>()
|
||||
private var isProcessingQueue = false
|
||||
private var watchDebounceTasks: [UUID: Task<Void, Never>] = [:]
|
||||
private var started = false
|
||||
private var notificationToken: NSObjectProtocol?
|
||||
|
||||
init(modelContext: ModelContext, runner: RsyncRunner) {
|
||||
self.modelContext = modelContext
|
||||
self.runner = runner
|
||||
watcher.onChange = { [weak self] jobID, path in
|
||||
Task { @MainActor [weak self] in
|
||||
self?.fileChanged(jobID: jobID, path: path)
|
||||
}
|
||||
}
|
||||
notificationToken = NotificationCenter.default.addObserver(
|
||||
forName: .syncJobsChanged,
|
||||
object: nil,
|
||||
queue: .main
|
||||
) { [weak self] _ in
|
||||
Task { @MainActor [weak self] in self?.refreshConfiguration() }
|
||||
}
|
||||
}
|
||||
|
||||
deinit {
|
||||
if let notificationToken { NotificationCenter.default.removeObserver(notificationToken) }
|
||||
}
|
||||
|
||||
func start() {
|
||||
guard !started else { return }
|
||||
started = true
|
||||
refreshConfiguration()
|
||||
checkSchedules()
|
||||
timer = Timer.scheduledTimer(withTimeInterval: 20, repeats: true) { [weak self] _ in
|
||||
Task { @MainActor [weak self] in
|
||||
self?.refreshConfiguration()
|
||||
self?.checkSchedules()
|
||||
self?.processQueue()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func stop() {
|
||||
timer?.invalidate()
|
||||
timer = nil
|
||||
watcher.stop()
|
||||
watchDebounceTasks.values.forEach { $0.cancel() }
|
||||
watchDebounceTasks.removeAll()
|
||||
started = false
|
||||
}
|
||||
|
||||
func syncAllEnabledJobs() {
|
||||
guard !isSyncAllRunning else { return }
|
||||
|
||||
let enabledJobs = fetchJobs().filter(\.isEnabled)
|
||||
syncAllCompleted = 0
|
||||
syncAllTotal = 0
|
||||
|
||||
for job in enabledJobs {
|
||||
if enqueue(jobID: job.id, trigger: .syncAll) {
|
||||
syncAllTotal += 1
|
||||
}
|
||||
}
|
||||
|
||||
isSyncAllRunning = syncAllTotal > 0
|
||||
if !isSyncAllRunning {
|
||||
syncAllCompleted = 0
|
||||
}
|
||||
}
|
||||
|
||||
func refreshConfiguration() {
|
||||
let jobs = fetchJobs()
|
||||
scheduledJobCount = jobs.filter { $0.isEnabled && $0.scheduleEnabled && $0.scheduleKind != .manual }.count
|
||||
watchedJobCount = jobs.filter { $0.isEnabled && $0.watchEnabled }.count
|
||||
|
||||
if UserDefaults.standard.object(forKey: "fileWatchingEnabled") as? Bool ?? true {
|
||||
watcher.update(jobs: jobs)
|
||||
} else {
|
||||
watcher.update(jobs: [])
|
||||
}
|
||||
|
||||
for job in jobs where job.scheduleEnabled && job.scheduleKind != .manual && job.nextRunAt == nil {
|
||||
job.nextRunAt = job.calculateNextRun(after: Date().addingTimeInterval(-1))
|
||||
}
|
||||
try? modelContext.save()
|
||||
}
|
||||
|
||||
private func checkSchedules() {
|
||||
guard UserDefaults.standard.object(forKey: "backgroundSchedulesEnabled") as? Bool ?? true else { return }
|
||||
let now = Date()
|
||||
for job in fetchJobs() where job.isEnabled && job.scheduleEnabled && job.scheduleKind != .manual {
|
||||
guard let next = job.nextRunAt, next <= now else { continue }
|
||||
enqueue(jobID: job.id, trigger: .scheduled)
|
||||
|
||||
if job.scheduleKind == .once {
|
||||
job.scheduleEnabled = false
|
||||
job.nextRunAt = nil
|
||||
} else {
|
||||
job.nextRunAt = job.calculateNextRun(after: now)
|
||||
}
|
||||
}
|
||||
try? modelContext.save()
|
||||
}
|
||||
|
||||
private func fileChanged(jobID: UUID, path: String) {
|
||||
guard UserDefaults.standard.object(forKey: "fileWatchingEnabled") as? Bool ?? true else { return }
|
||||
guard let job = fetchJob(id: jobID), job.isEnabled, job.watchEnabled else { return }
|
||||
job.lastWatchEventAt = Date()
|
||||
watchDebounceTasks[jobID]?.cancel()
|
||||
let delay = max(1, job.watchDebounceSeconds)
|
||||
watchDebounceTasks[jobID] = Task { [weak self] in
|
||||
try? await Task.sleep(for: .seconds(delay))
|
||||
guard !Task.isCancelled, let self else { return }
|
||||
_ = self.enqueue(jobID: jobID, trigger: .watched)
|
||||
}
|
||||
try? modelContext.save()
|
||||
}
|
||||
|
||||
@discardableResult
|
||||
private func enqueue(jobID: UUID, trigger: SyncRunTrigger) -> Bool {
|
||||
guard !pendingIDs.contains(jobID), runner.currentJobID != jobID else { return false }
|
||||
pending.append((jobID, trigger))
|
||||
pendingIDs.insert(jobID)
|
||||
processQueue()
|
||||
return true
|
||||
}
|
||||
|
||||
private func processQueue() {
|
||||
guard !isProcessingQueue, !runner.isRunning, !pending.isEmpty else {
|
||||
if (isProcessingQueue || runner.isRunning), let first = pending.first, let job = fetchJob(id: first.0) {
|
||||
activity = .waiting(job.name)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
let item = pending.removeFirst()
|
||||
pendingIDs.remove(item.0)
|
||||
guard let job = fetchJob(id: item.0), job.isEnabled else {
|
||||
finishSyncAllItemIfNeeded(item.1)
|
||||
processQueue()
|
||||
return
|
||||
}
|
||||
|
||||
isProcessingQueue = true
|
||||
activity = .running(job.name)
|
||||
Task { [weak self] in
|
||||
await self?.execute(job: job, trigger: item.1)
|
||||
await MainActor.run {
|
||||
guard let self else { return }
|
||||
self.finishSyncAllItemIfNeeded(item.1)
|
||||
self.isProcessingQueue = false
|
||||
self.activity = .idle
|
||||
self.processQueue()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func finishSyncAllItemIfNeeded(_ trigger: SyncRunTrigger) {
|
||||
switch trigger {
|
||||
case .syncAll:
|
||||
syncAllCompleted += 1
|
||||
if syncAllCompleted >= syncAllTotal {
|
||||
isSyncAllRunning = false
|
||||
}
|
||||
case .manual, .scheduled, .watched:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
private func execute(job: SyncJob, trigger: SyncRunTrigger) async {
|
||||
let executable = RsyncLocator.locate()
|
||||
let command = RsyncCommandBuilder.displayCommand(job: job, dryRun: false, executable: executable)
|
||||
let run = SyncRun(job: job, command: command, wasDryRun: false, trigger: trigger)
|
||||
modelContext.insert(run)
|
||||
try? modelContext.save()
|
||||
|
||||
do {
|
||||
let result = try await runner.run(job: job, dryRun: false)
|
||||
run.output = result.stdout
|
||||
run.errorOutput = result.stderr
|
||||
run.exitCode = Int(result.exitCode)
|
||||
run.wasCancelled = result.wasCancelled
|
||||
run.finishedAt = Date()
|
||||
job.lastRunAt = run.finishedAt
|
||||
job.lastExitCode = Int(result.exitCode)
|
||||
job.updatedAt = Date()
|
||||
} catch {
|
||||
run.errorOutput = error.localizedDescription
|
||||
run.exitCode = -1
|
||||
run.finishedAt = Date()
|
||||
job.lastRunAt = run.finishedAt
|
||||
job.lastExitCode = -1
|
||||
}
|
||||
try? modelContext.save()
|
||||
}
|
||||
|
||||
private func fetchJobs() -> [SyncJob] {
|
||||
(try? modelContext.fetch(FetchDescriptor<SyncJob>(sortBy: [SortDescriptor(\SyncJob.name)]))) ?? []
|
||||
}
|
||||
|
||||
private func fetchJob(id: UUID) -> SyncJob? {
|
||||
fetchJobs().first { $0.id == id }
|
||||
}
|
||||
}
|
||||
|
||||
extension Notification.Name {
|
||||
static let syncJobsChanged = Notification.Name("RsyncManager.syncJobsChanged")
|
||||
}
|
||||
Reference in New Issue
Block a user