| 1 | import DocsCRDT |
| 2 | import Foundation |
| 3 | |
| 4 | /// Handles never leave the actor. Every untrusted decode executes in bounded WASM. |
| 5 | actor CRDTReplica { |
| 6 | private final class Sandbox { |
| 7 | let host: UnsafeMutableRawPointer |
| 8 | init() throws { guard let host = docs_host_new() else { throw Failure.unavailable }; self.host = host } |
| 9 | deinit { docs_host_free(host) } |
| 10 | func invoke(_ data: Data) throws -> Data { |
| 11 | let maximum = 8 * 1024 * 1024 + 1 |
| 12 | guard !data.isEmpty, data.count <= maximum else { throw Failure.unavailable } |
| 13 | var output = Data(count: maximum) |
| 14 | let count = output.withUnsafeMutableBytes { target in data.withUnsafeBytes { source in |
| 15 | docs_host_call(host, source.bindMemory(to: UInt8.self).baseAddress!, data.count, |
| 16 | target.bindMemory(to: UInt8.self).baseAddress!, maximum) |
| 17 | } } |
| 18 | guard count >= 0, count <= maximum else { throw Failure.rejected } |
| 19 | output.count = Int(count); return output |
| 20 | } |
| 21 | func invoke(_ command: [String: JSONValue]) throws -> Data { try invoke(JSONEncoder().encode(command)) } |
| 22 | func projection() throws -> CRDTProjection { |
| 23 | try JSONDecoder().decode(CRDTProjection.self, from: invoke(["op": .string("project")])) |
| 24 | } |
| 25 | } |
| 26 | private var authority: Sandbox? |
| 27 | private var merger: Sandbox? |
| 28 | private var opened = false |
| 29 | private var sources: [UInt64: Sandbox] = [:] |
| 30 | private var sourceSnapshots: [UInt64: Data] = [:] |
| 31 | private var nextGeneration: UInt64 = 1 |
| 32 | // IDs remain reserved for this session even when a source is retired or traps. |
| 33 | private var reserved: Set<UInt64> = [] |
| 34 | private var lastGood = Data() |
| 35 | private var commandsApplied = 0 |
| 36 | private var durableSequence: Int64 = 0 |
| 37 | private let guid: String |
| 38 | enum Failure: Error { case unavailable, rejected, invalidProjection, generationLimit } |
| 39 | init(guid: String) { self.guid = guid } |
| 40 | struct PersistentState: Sendable { |
| 41 | let state: Data |
| 42 | let sequence: Int64 |
| 43 | let commandsApplied: Int |
| 44 | let guid: String |
| 45 | } |
| 46 | func persistentState() throws -> PersistentState { |
| 47 | _ = try currentAuthority() |
| 48 | return PersistentState(state: lastGood, sequence: durableSequence, commandsApplied: commandsApplied, guid: guid) |
| 49 | } |
| 50 | func setSequence(_ sequence: Int64) { durableSequence = max(durableSequence, sequence) } |
| 51 | func close() { opened = false; authority = nil; merger = nil; sources.removeAll(); sourceSnapshots.removeAll(); lastGood = Data() } |
| 52 | private func fresh(update: Data) throws -> (UInt64, Sandbox) { |
| 53 | let sandbox = try Sandbox() |
| 54 | var client = UInt64(UInt32.random(in: 1...UInt32.max)) |
| 55 | while reserved.contains(client) { client = UInt64(UInt32.random(in: 1...UInt32.max)) } |
| 56 | reserved.insert(client) |
| 57 | _ = try sandbox.invoke(["op": .string("open"), "guid": .string(guid), "client": .number(Double(client))]) |
| 58 | if !update.isEmpty { _ = try sandbox.invoke(Data([255]) + update) } |
| 59 | let used = try JSONDecoder().decode([UInt64].self, from: sandbox.invoke(["op": .string("clients")])) |
| 60 | reserved.formUnion(used) |
| 61 | // Imported state may already contain the proposed ID. Never write it. |
| 62 | if used.contains(client) { return try fresh(update: update) } |
| 63 | return (client, sandbox) |
| 64 | } |
| 65 | func open(update: Data) throws -> CRDTProjection { |
| 66 | sources.removeAll(); sourceSnapshots.removeAll() |
| 67 | commandsApplied = 0; durableSequence = 0 |
| 68 | authority = try fresh(update: update).1 |
| 69 | opened = true |
| 70 | lastGood = try authority!.invoke(["op": .string("snapshot")]) |
| 71 | return try projection() |
| 72 | } |
| 73 | private func currentAuthority() throws -> Sandbox { |
| 74 | guard opened else { throw Failure.unavailable } |
| 75 | if authority == nil { authority = try fresh(update: lastGood).1 } |
| 76 | guard let authority else { throw Failure.unavailable } |
| 77 | return authority |
| 78 | } |
| 79 | func projection() throws -> CRDTProjection { |
| 80 | let authority = try currentAuthority() |
| 81 | return try authority.projection() |
| 82 | } |
| 83 | /// Bound retained WASM memories. The caller defers publication until render |
| 84 | /// acknowledgement and queued-callback drainage allow old sources to retire. |
| 85 | func presentation(source generation: UInt64, selection: NSRange, allowReuse: Bool = false) throws -> (UInt64, CRDTProjection, NSRange)? { |
| 86 | let authority = try currentAuthority() |
| 87 | // Only the caller knows that this source has actually been rendered and |
| 88 | // has no pending candidate. Byte-identical state is additionally required: |
| 89 | // peers, undo, and divergent queued edits must still publish a fresh source. |
| 90 | // Never reconstruct or transfer a client ID to enable this fast path. |
| 91 | let reuse = allowReuse && sources[generation] != nil && sourceSnapshots[generation] == lastGood |
| 92 | guard reuse || sources.count < 3 else { return nil } |
| 93 | var resolved = selection |
| 94 | if let displayed = sources[generation] { |
| 95 | func endpoint(_ position: Int) throws -> Int { |
| 96 | let encoded = try displayed.invoke(["op": .string("anchor"), "at": .number(Double(position))]) |
| 97 | let anchor = try JSONDecoder().decode(JSONValue.self, from: encoded) |
| 98 | return try JSONDecoder().decode(Int.self, from: authority.invoke(["op": .string("resolve"), "anchor": anchor])) |
| 99 | } |
| 100 | do { |
| 101 | let start = try endpoint(selection.location), end = try endpoint(NSMaxRange(selection)) |
| 102 | resolved = NSRange(location: min(start, end), length: abs(end - start)) |
| 103 | } catch { |
| 104 | if let snapshot = sourceSnapshots[generation] { sources[generation] = try? fresh(update: snapshot).1 } |
| 105 | else { sources.removeValue(forKey: generation) } |
| 106 | recover() |
| 107 | throw error |
| 108 | } |
| 109 | } |
| 110 | if reuse { return (generation, try authority.projection(), resolved) } |
| 111 | let (_, source) = try fresh(update: lastGood) |
| 112 | let generation = nextGeneration; nextGeneration += 1 |
| 113 | sources[generation] = source; sourceSnapshots[generation] = lastGood |
| 114 | return (generation, try authority.projection(), resolved) |
| 115 | } |
| 116 | func retainGenerations(_ ids: Set<UInt64>) { |
| 117 | sources = sources.filter { ids.contains($0.key) } |
| 118 | sourceSnapshots = sourceSnapshots.filter { ids.contains($0.key) } |
| 119 | } |
| 120 | private func recover() { |
| 121 | // No invalid interpreter is reused. A new authority ID cannot reuse |
| 122 | // clocks from a partially executed command. UI reports undo loss. |
| 123 | authority = try? fresh(update: lastGood).1 |
| 124 | } |
| 125 | func mergeUpdates(_ updates: [Data]) throws -> Data { |
| 126 | guard let first = updates.first else { throw Failure.rejected } |
| 127 | if updates.count == 1 { return first } |
| 128 | guard opened else { throw Failure.unavailable } |
| 129 | if merger == nil { merger = try Sandbox() } |
| 130 | var envelope = Data([253]) |
| 131 | for update in updates { envelope.append(update) } |
| 132 | do { return try merger!.invoke(envelope) } |
| 133 | catch { merger = nil; throw error } |
| 134 | } |
| 135 | func apply(_ update: Data) throws -> CRDTProjection { |
| 136 | let authority = try currentAuthority() |
| 137 | do { |
| 138 | _ = try authority.invoke(Data([255]) + update) |
| 139 | let projected = try authority.projection() |
| 140 | lastGood = try authority.invoke(["op": .string("snapshot")]) |
| 141 | return projected |
| 142 | } catch { recover(); throw error } |
| 143 | } |
| 144 | func command(_ command: [String: JSONValue]) throws -> (Data, CRDTProjection) { |
| 145 | let authority = try currentAuthority() |
| 146 | let generation: UInt64? = if case .number(let value) = command["generation"] { UInt64(exactly: value) } else { nil } |
| 147 | let isHistory = command["op"] == .string("undo") || command["op"] == .string("redo") |
| 148 | let writer: Sandbox |
| 149 | if isHistory || generation == nil { writer = authority } |
| 150 | else { |
| 151 | let generation = generation! |
| 152 | if sources[generation] == nil, let snapshot = sourceSnapshots[generation] { sources[generation] = try fresh(update: snapshot).1 } |
| 153 | guard let source = sources[generation] else { throw Failure.unavailable } |
| 154 | writer = source |
| 155 | } |
| 156 | let sourceBefore = writer === authority ? lastGood : (generation.flatMap { sourceSnapshots[$0] } ?? lastGood) |
| 157 | var authorityTouched = writer === authority |
| 158 | do { |
| 159 | let update = try writer.invoke(command) |
| 160 | let sourceAfter = try writer.invoke(["op": .string("snapshot")]) |
| 161 | if writer !== authority { authorityTouched = true; _ = try authority.invoke(Data([254]) + update) } |
| 162 | let projected = try authority.projection() |
| 163 | lastGood = try authority.invoke(["op": .string("snapshot")]) |
| 164 | if writer !== authority, let generation { sourceSnapshots[generation] = sourceAfter } |
| 165 | commandsApplied += 1 |
| 166 | return (update, projected) |
| 167 | } catch { |
| 168 | if writer !== authority, let generation { sources[generation] = try? fresh(update: sourceBefore).1 } |
| 169 | if authorityTouched { recover() } |
| 170 | throw error |
| 171 | } |
| 172 | } |
| 173 | } |
| 174 | |
amadan