| 1 | import Foundation |
| 2 | |
| 3 | /// One queued change, durable on disk until the server has it. |
| 4 | public struct OutboxEntry: Codable, Sendable, Identifiable, Equatable { |
| 5 | /// Stable for the entry's life; apps may send it as an idempotency key. |
| 6 | public let id: UUID |
| 7 | /// Per account and monotonic, recovered as max + 1 at load. It orders a lane. |
| 8 | public let seq: UInt64 |
| 9 | /// The FIFO scope, such as "doc:<key>". It never changes. |
| 10 | public let lane: String |
| 11 | /// App-defined. |
| 12 | public let kind: String |
| 13 | /// Opaque app bytes. |
| 14 | public let payload: Data |
| 15 | public let createdAt: Date |
| 16 | /// Failed sends so far. Counted in memory; written with the next rewrite. |
| 17 | public var attempts: Int |
| 18 | /// nil means sendable. |
| 19 | public var hold: Hold? |
| 20 | |
| 21 | public init(id: UUID, seq: UInt64, lane: String, kind: String, payload: Data, createdAt: Date, attempts: Int = 0, hold: Hold? = nil) { |
| 22 | self.id = id; self.seq = seq; self.lane = lane; self.kind = kind; self.payload = payload |
| 23 | self.createdAt = createdAt; self.attempts = attempts; self.hold = hold |
| 24 | } |
| 25 | } |
| 26 | |
| 27 | /// Why an entry stopped: the person has to decide what happens to it. |
| 28 | public struct Hold: Codable, Sendable, Equatable { |
| 29 | public enum Reason: Codable, Sendable, Equatable { |
| 30 | case conflict(Data) |
| 31 | case rejected(String) |
| 32 | case accessLost |
| 33 | case tooLarge |
| 34 | } |
| 35 | public let reason: Reason |
| 36 | /// Every v1 hold stops its lane. A hold that lets later entries pass |
| 37 | /// arrives with Sheets, whose writes to different cells do not depend on |
| 38 | /// each other; Docs' updates do, so it never needs one. |
| 39 | public var blocksLane: Bool { true } |
| 40 | public init(_ reason: Reason) { self.reason = reason } |
| 41 | } |
| 42 | |
| 43 | /// Per-lane counts, for status and for the app's own caps. |
| 44 | public struct OutboxSummary: Sendable, Equatable { |
| 45 | public struct Lane: Sendable, Equatable { |
| 46 | /// Entries not held, including any in flight. |
| 47 | public var pending = 0 |
| 48 | public var held = 0 |
| 49 | public var inFlight = 0 |
| 50 | /// Payload bytes of every entry in the lane, held or not. |
| 51 | public var bytes = 0 |
| 52 | public var firstHold: Hold.Reason? |
| 53 | public init() {} |
| 54 | } |
| 55 | public var lanes: [String: Lane] = [:] |
| 56 | /// The store could not be read. Counts are the last ones known, which is |
| 57 | /// never the same thing as "nothing pending". |
| 58 | public var unavailable = false |
| 59 | /// Unsent entries were lost to a lost key or a corrupt file. Report it |
| 60 | /// once, then call `Outbox.acknowledgeLoss()`. |
| 61 | public var lostUnsent = false |
| 62 | |
| 63 | public init() {} |
| 64 | |
| 65 | public var pending: Int { lanes.values.reduce(0) { $0 + $1.pending } } |
| 66 | public var held: Int { lanes.values.reduce(0) { $0 + $1.held } } |
| 67 | public var inFlight: Int { lanes.values.reduce(0) { $0 + $1.inFlight } } |
| 68 | /// Lanes with anything in them: the default "N" of "Offline · N pending". |
| 69 | public var pendingLanes: Int { lanes.values.filter { $0.pending + $0.held > 0 }.count } |
| 70 | public var isEmpty: Bool { lanes.values.allSatisfy { $0.pending + $0.held == 0 } } |
| 71 | |
| 72 | /// The same summary, narrowed to one lane. |
| 73 | public func only(_ lane: String) -> OutboxSummary { |
| 74 | var copy = self |
| 75 | copy.lanes = lanes[lane].map { [lane: $0] } ?? [:] |
| 76 | return copy |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | public enum OutboxError: Error, Sendable, Equatable { |
| 81 | /// `replace` was asked to merge an entry that is being sent. The batch's |
| 82 | /// `ack` would otherwise delete the merged file, and with it content whose |
| 83 | /// own files the merge had already removed. |
| 84 | case inFlight |
| 85 | /// `replace` needs a run of entries that are next to each other in one lane. |
| 86 | case notContiguous |
| 87 | /// `replace` never merges a held entry: it is waiting for a person. |
| 88 | case held |
| 89 | case unknownEntry(UUID) |
| 90 | case duplicateEntry(UUID) |
| 91 | } |
| 92 | |
| 93 | /// Serializes disk mutations and send reservations for one account. Memory is |
| 94 | /// changed after each successful disk operation, including partial batch acks. |
| 95 | public actor Outbox { |
| 96 | public nonisolated let account: String |
| 97 | public static let directory = "outbox" |
| 98 | private let store: BlobStore |
| 99 | private var values: [OutboxEntry] = [] |
| 100 | private var flying: Set<UUID> = [] |
| 101 | private var loaded = false |
| 102 | private var unavailable = false |
| 103 | private var lost = false |
| 104 | private var next: UInt64 = 1 |
| 105 | private var subscribers: [UUID: AsyncStream<OutboxSummary>.Continuation] = [:] |
| 106 | public init(store: BlobStore, account: String) { self.store = store; self.account = account } |
| 107 | private func path(_ seq: UInt64) -> String { "outbox/" + String(format: "%016llx", seq) + ".entry" } |
| 108 | public func load() throws { |
| 109 | guard !loaded || unavailable else { return } |
| 110 | var recovered: [OutboxEntry] = [] |
| 111 | do { |
| 112 | let names = try store.list(Self.directory) |
| 113 | for name in names where name.hasSuffix(".entry") { |
| 114 | let file = Self.directory + "/" + name |
| 115 | do { |
| 116 | guard let data = try store.read(file) else { continue } |
| 117 | guard let entry = try? JSONDecoder().decode(OutboxEntry.self, from: data), path(entry.seq) == file, |
| 118 | !recovered.contains(where: { $0.id == entry.id }) else { |
| 119 | try store.remove(file); lost = true; continue |
| 120 | } |
| 121 | recovered.append(entry) |
| 122 | } catch BlobStoreError.corrupt { lost = true } |
| 123 | } |
| 124 | values = recovered.sorted { $0.seq < $1.seq } |
| 125 | next = max(next, (values.last?.seq ?? 0) + 1) |
| 126 | loaded = true; unavailable = false; notify() |
| 127 | } catch BlobStoreError.keyLost { |
| 128 | values = []; flying = []; loaded = true; unavailable = false; lost = true; notify() |
| 129 | } catch { unavailable = true; notify(); throw error } |
| 130 | } |
| 131 | private func mutate<T>(_ body: () throws -> T) throws -> T { |
| 132 | do { let result = try body(); unavailable = false; notify(); return result } |
| 133 | catch { unavailable = true; notify(); throw error } |
| 134 | } |
| 135 | private func persist(_ entry: OutboxEntry) throws { try store.write(JSONEncoder().encode(entry), to: path(entry.seq)) } |
| 136 | @discardableResult |
| 137 | public func enqueue(lane: String, kind: String, payload: Data, id: UUID = UUID()) throws -> OutboxEntry { |
| 138 | try load() |
| 139 | guard !values.contains(where: { $0.id == id }) else { throw OutboxError.duplicateEntry(id) } |
| 140 | return try mutate { |
| 141 | let entry = OutboxEntry(id: id, seq: next, lane: lane, kind: kind, payload: payload, createdAt: Date()) |
| 142 | try persist(entry); values.append(entry); next += 1; return entry |
| 143 | } |
| 144 | } |
| 145 | /// A last-known snapshot, including held entries. On an unavailable store |
| 146 | /// this may be stale, just like summary (whose unavailable flag is set). |
| 147 | /// Use it for display/deduplication hints only; replay must first call load() |
| 148 | /// successfully. Every mutation and beginSend independently require load(). |
| 149 | public func entries(lane: String, excludingInFlight: Bool = false) -> [OutboxEntry] { |
| 150 | try? load() |
| 151 | return values.filter { $0.lane == lane && (!excludingInFlight || !flying.contains($0.id)) } |
| 152 | } |
| 153 | /// Fails closed to an empty batch when loading is unavailable. |
| 154 | public func sendable(lane: String, maxBytes: Int? = nil) -> [OutboxEntry] { |
| 155 | try? load() |
| 156 | guard !unavailable else { return [] } |
| 157 | var result: [OutboxEntry] = []; var bytes = 0 |
| 158 | for entry in values where entry.lane == lane { |
| 159 | if entry.hold != nil { break } |
| 160 | // Compaction may use the suffix after a reserved batch, but beginSend |
| 161 | // below never sends it before the earlier batch finishes. |
| 162 | if flying.contains(entry.id) { continue } |
| 163 | if let maxBytes, !result.isEmpty, bytes + entry.payload.count > maxBytes { break } |
| 164 | result.append(entry); bytes += entry.payload.count |
| 165 | } |
| 166 | return result |
| 167 | } |
| 168 | public func beginSend(lane: String, maxBytes: Int? = nil) throws -> [OutboxEntry] { |
| 169 | try load() |
| 170 | guard flying.isEmpty else { return [] } |
| 171 | let batch = sendable(lane: lane, maxBytes: maxBytes) |
| 172 | if !batch.isEmpty { flying.formUnion(batch.map(\.id)); notify() } |
| 173 | return batch |
| 174 | } |
| 175 | public func endSend(_ ids: [UUID], failed: Bool = true) { |
| 176 | for id in ids { |
| 177 | flying.remove(id) |
| 178 | if failed, let index = values.firstIndex(where: { $0.id == id }) { values[index].attempts += 1 } |
| 179 | } |
| 180 | notify() |
| 181 | } |
| 182 | @discardableResult |
| 183 | public func replace(_ ids: [UUID], with payload: Data) throws -> OutboxEntry { |
| 184 | try load() |
| 185 | guard !ids.isEmpty, Set(ids).count == ids.count else { throw OutboxError.notContiguous } |
| 186 | if ids.contains(where: { flying.contains($0) }) { throw OutboxError.inFlight } |
| 187 | let run = try ids.map { id in |
| 188 | guard let entry = values.first(where: { $0.id == id }) else { throw OutboxError.unknownEntry(id) } |
| 189 | return entry |
| 190 | }.sorted { $0.seq < $1.seq } |
| 191 | let first = run[0] |
| 192 | let lane = values.filter { $0.lane == first.lane } |
| 193 | guard run.allSatisfy({ $0.lane == first.lane && $0.kind == first.kind }), |
| 194 | let start = lane.firstIndex(where: { $0.id == first.id }), |
| 195 | Array(lane.dropFirst(start).prefix(run.count)).map(\.id) == run.map(\.id) else { throw OutboxError.notContiguous } |
| 196 | let sendableIDs = Set(sendable(lane: first.lane).map(\.id)) |
| 197 | guard run.allSatisfy({ $0.hold == nil && sendableIDs.contains($0.id) }) else { throw OutboxError.held } |
| 198 | return try mutate { |
| 199 | let merged = OutboxEntry(id: UUID(), seq: first.seq, lane: first.lane, kind: first.kind, payload: payload, createdAt: first.createdAt) |
| 200 | try persist(merged) |
| 201 | values[values.firstIndex(where: { $0.id == first.id })!] = merged |
| 202 | for entry in run.dropFirst() { |
| 203 | try store.remove(path(entry.seq)); values.removeAll { $0.id == entry.id } |
| 204 | } |
| 205 | return merged |
| 206 | } |
| 207 | } |
| 208 | public func ack(_ ids: [UUID]) throws { |
| 209 | try load() |
| 210 | try mutate { |
| 211 | for id in ids { |
| 212 | guard let entry = values.first(where: { $0.id == id }) else { continue } |
| 213 | try store.remove(path(entry.seq)); values.removeAll { $0.id == id }; flying.remove(id) |
| 214 | } |
| 215 | } |
| 216 | } |
| 217 | public func hold(_ ids: [UUID], _ hold: Hold) throws { |
| 218 | try load() |
| 219 | try mutate { |
| 220 | for id in ids { |
| 221 | guard let index = values.firstIndex(where: { $0.id == id }) else { continue } |
| 222 | var entry = values[index]; entry.hold = hold |
| 223 | try persist(entry); values[index] = entry; flying.remove(id) |
| 224 | } |
| 225 | } |
| 226 | } |
| 227 | public func release(_ id: UUID) throws { |
| 228 | try load() |
| 229 | guard let index = values.firstIndex(where: { $0.id == id }) else { throw OutboxError.unknownEntry(id) } |
| 230 | try mutate { var entry = values[index]; entry.hold = nil; try persist(entry); values[index] = entry } |
| 231 | } |
| 232 | public func discard(lane: String) throws { try load(); try ack(values.filter { $0.lane == lane }.map(\.id)) } |
| 233 | public func purge() throws { |
| 234 | try load() |
| 235 | try mutate { try store.remove(Self.directory); values = []; flying = [] } |
| 236 | } |
| 237 | public func acknowledgeLoss() { lost = false; notify() } |
| 238 | private var currentSummary: OutboxSummary { |
| 239 | var result = OutboxSummary(); result.unavailable = unavailable; result.lostUnsent = lost |
| 240 | for entry in values { |
| 241 | var lane = result.lanes[entry.lane] ?? OutboxSummary.Lane() |
| 242 | if let hold = entry.hold { lane.held += 1; if lane.firstHold == nil { lane.firstHold = hold.reason } } |
| 243 | else { lane.pending += 1 } |
| 244 | if flying.contains(entry.id) { lane.inFlight += 1 } |
| 245 | lane.bytes += entry.payload.count; result.lanes[entry.lane] = lane |
| 246 | } |
| 247 | return result |
| 248 | } |
| 249 | /// Last-known counts; unavailable marks retained counts as non-authoritative. |
| 250 | public var summary: OutboxSummary { try? load(); return currentSummary } |
| 251 | private func notify() { for subscriber in subscribers.values { subscriber.yield(currentSummary) } } |
| 252 | private func unsubscribe(_ id: UUID) { subscribers[id] = nil } |
| 253 | public var changes: AsyncStream<OutboxSummary> { |
| 254 | try? load() |
| 255 | let (stream, continuation) = AsyncStream<OutboxSummary>.makeStream(bufferingPolicy: .bufferingNewest(1)) |
| 256 | let id = UUID(); subscribers[id] = continuation; continuation.yield(currentSummary) |
| 257 | continuation.onTermination = { [weak self] _ in Task { await self?.unsubscribe(id) } } |
| 258 | return stream |
| 259 | } |
| 260 | } |
| 261 | |
amadan