40 lines
1.2 KiB
Swift
40 lines
1.2 KiB
Swift
import Domain
|
|
import Store
|
|
|
|
public actor SyncEngine {
|
|
private let library: any Library
|
|
private let batchSize = 500
|
|
|
|
public init(library: any Library) {
|
|
self.library = library
|
|
}
|
|
|
|
public func sync(_ source: any LibrarySource) async throws {
|
|
do {
|
|
try await pull(source, since: library.cursor(for: source.id))
|
|
} catch SourceError.cannotCalculateChanges {
|
|
try await pull(source, since: nil)
|
|
}
|
|
}
|
|
|
|
private func pull(_ source: any LibrarySource, since cursor: Cursor?) async throws {
|
|
var batch: [Change] = []
|
|
for try await change in source.changes(since: cursor) {
|
|
if case let .checkpoint(next) = change {
|
|
await library.apply(batch)
|
|
batch.removeAll(keepingCapacity: true)
|
|
await library.setCursor(next, for: source.id)
|
|
} else {
|
|
batch.append(change)
|
|
if batch.count >= batchSize {
|
|
await library.apply(batch)
|
|
batch.removeAll(keepingCapacity: true)
|
|
}
|
|
}
|
|
}
|
|
if !batch.isEmpty {
|
|
await library.apply(batch)
|
|
}
|
|
}
|
|
}
|