277 lines
9.7 KiB
Swift
277 lines
9.7 KiB
Swift
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
|
|
@Published private(set) var lastError: String?
|
|
|
|
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()
|
|
pruneHistoryIfNeeded(force: true)
|
|
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))
|
|
}
|
|
saveContext("refreshing automation configuration")
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
saveContext("updating scheduled jobs")
|
|
}
|
|
|
|
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)
|
|
}
|
|
saveContext("recording file-watch activity")
|
|
}
|
|
|
|
@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)
|
|
saveContext("creating run history")
|
|
|
|
do {
|
|
let result = try await runner.run(job: job, dryRun: false, trigger: trigger)
|
|
run.output = SyncRun.sanitizedLog(result.stdout)
|
|
run.errorOutput = SyncRun.sanitizedLog(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 = SyncRun.sanitizedLog(error.localizedDescription)
|
|
run.exitCode = -1
|
|
run.finishedAt = Date()
|
|
job.lastRunAt = run.finishedAt
|
|
job.lastExitCode = -1
|
|
}
|
|
saveContext("saving run history")
|
|
pruneHistoryIfNeeded()
|
|
}
|
|
|
|
private func pruneHistoryIfNeeded(force: Bool = false) {
|
|
let now = Date()
|
|
if !force, let last = UserDefaults.standard.object(forKey: "lastHistoryPruneAt") as? Date, now.timeIntervalSince(last) < 3600 { return }
|
|
|
|
let configuredDays = UserDefaults.standard.integer(forKey: "historyRetentionDays")
|
|
let days = configuredDays == 0 ? 90 : max(7, min(configuredDays, 3650))
|
|
let cutoff = now.addingTimeInterval(-TimeInterval(days) * 86_400)
|
|
|
|
do {
|
|
let runs = try modelContext.fetch(FetchDescriptor<SyncRun>())
|
|
for run in runs where run.finishedAt != nil && run.startedAt < cutoff {
|
|
modelContext.delete(run)
|
|
}
|
|
try modelContext.save()
|
|
UserDefaults.standard.set(now, forKey: "lastHistoryPruneAt")
|
|
} catch {
|
|
lastError = "Could not prune run history: \(error.localizedDescription)"
|
|
}
|
|
}
|
|
|
|
private func saveContext(_ operation: String) {
|
|
do {
|
|
try modelContext.save()
|
|
} catch {
|
|
lastError = "Database error while \(operation): \(error.localizedDescription)"
|
|
}
|
|
}
|
|
|
|
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")
|
|
}
|