| 1 | import Foundation |
| 2 | import RastrilloNative |
| 3 | |
| 4 | /// Editor captures and headless sends both commit here. Merging with the latest |
| 5 | /// durable state prevents either writer from erasing an acknowledged batch. |
| 6 | @MainActor final class DocsReplicaCache { |
| 7 | let store: BlobStore |
| 8 | private var generations: [String: Int] = [:] |
| 9 | init(store: BlobStore) { self.store = store } |
| 10 | func generation(_ key: String) -> Int { generations[key, default: 0] } |
| 11 | func invalidate(_ key: String) { generations[key, default: 0] += 1 } |
| 12 | @discardableResult func commit(_ candidate: PersistedReplica, key: String, generation: Int, valid: () -> Bool = { true }) async throws -> Bool { |
| 13 | let path = "cache/docs/\(key)/replica" |
| 14 | while self.generation(key) == generation { |
| 15 | let before = try store.read(path) |
| 16 | let previous = try before.map { try JSONDecoder().decode(PersistedReplica.self, from: $0) } |
| 17 | let merger = CRDTReplica(guid: candidate.guid) |
| 18 | do { |
| 19 | _ = try await merger.open(update: previous?.state ?? Data()) |
| 20 | _ = try await merger.apply(candidate.state) |
| 21 | let state = try await merger.persistentState().state |
| 22 | await merger.close() |
| 23 | guard self.generation(key) == generation else { throw CancellationError() } |
| 24 | // No suspension between the compare and write. A concurrent |
| 25 | // editor or headless writer forces a fresh merge, never a loss. |
| 26 | guard try store.read(path) == before else { continue } |
| 27 | let merged = PersistedReplica(format: 1, sequence: max(previous?.sequence ?? 0, candidate.sequence), guid: candidate.guid, savedAt: Date(), state: state) |
| 28 | guard valid() else { return false } |
| 29 | try store.write(JSONEncoder().encode(merged), to: path) |
| 30 | return true |
| 31 | } catch { |
| 32 | await merger.close() |
| 33 | throw error |
| 34 | } |
| 35 | } |
| 36 | throw CancellationError() |
| 37 | } |
| 38 | } |
| 39 | |
amadan