| 1 | import Foundation |
| 2 | import SwiftUI |
| 3 | |
| 4 | // 🤖 The native runtime for one schema-described app thread (To-do, Reminders): |
| 5 | // owns the folded state and the transport, so SchemaAppView can be a pure |
| 6 | // renderer. The data path mirrors the web framework exactly: |
| 7 | // |
| 8 | // 1. seed from the encrypted "apps" snapshot (thread_state — the web writes |
| 9 | // it; snapshot + newer events ≡ folding all of history), |
| 10 | // 2. fold the thread's decrypted app events newer than the snapshot horizon, |
| 11 | // 3. apply live socket frames as they arrive, |
| 12 | // 4. run actions (emit / schedule / unschedule) over the same wire the web |
| 13 | // uses — REST POST /api/messages for events, /api/schedule for timers. |
| 14 | // |
| 15 | // Native READS snapshots but never writes them (the web clients keep doing |
| 16 | // that); everything it sends is encrypted under the thread key first, so the |
| 17 | // server keeps seeing only ciphertext. |
| 18 | @MainActor |
| 19 | public final class SchemaAppStore: ObservableObject { |
| 20 | @Published public private(set) var state: AppJSON |
| 21 | /// Inline feedback from a failed action (a parse hint) — rendered in |
| 22 | /// layout, never as a floating toast. Cleared when the composer changes. |
| 23 | @Published public var hintText: String? |
| 24 | /// Bumped when a grace window lapses so rows re-partition on time. |
| 25 | @Published private var graceBeat = 0 |
| 26 | |
| 27 | public let schema: AppJSON |
| 28 | public let appId: String |
| 29 | let instance: SavedInstance |
| 30 | let thread: ChatThread |
| 31 | let myUserId: String |
| 32 | let key: GroupKey? |
| 33 | let api: LChatAPI? |
| 34 | private var socket: ChatSocket? |
| 35 | private var seededUpTo: Int64 = 0 |
| 36 | private var maxFolded: Int64 = 0 |
| 37 | private var tickerFired = Set<String>() |
| 38 | private var tickerTimer: Timer? |
| 39 | private var graceTask: Task<Void, Never>? |
| 40 | private var started = false |
| 41 | // Live frames arriving mid-load buffer here: load() publishes a state |
| 42 | // computed from the pre-live snapshot, so folding them early would be |
| 43 | // overwritten. All on the main actor, so this is just ordering, not races. |
| 44 | private var loading = true |
| 45 | private var pendingLive: [AppJSON] = [] |
| 46 | // Optimistic sends queue behind each other so events reach the server in |
| 47 | // fold order (the web's ws.send is ordered; parallel POSTs wouldn't be). |
| 48 | private var sendChain: Task<Void, Never>? |
| 49 | |
| 50 | /// `appId` defaults to the thread's own app (a personal app-thread); a |
| 51 | /// GROUP thread hosting shared apps passes the picked app explicitly — |
| 52 | /// same events, same reducer, folded from the group's log instead. |
| 53 | public init?(instance: SavedInstance, thread: ChatThread, myUserId: String, appId: String? = nil) { |
| 54 | guard let appId = appId ?? thread.appId, let schema = AppSchemaRegistry.schema(for: appId) else { return nil } |
| 55 | self.schema = schema |
| 56 | self.appId = appId |
| 57 | self.instance = instance |
| 58 | self.thread = thread |
| 59 | self.myUserId = myUserId |
| 60 | self.key = try? GroupKey(base64url: thread.threadKeyB64) |
| 61 | self.api = instance.baseURL.map { LChatAPI(baseURL: $0, token: instance.deviceToken) } |
| 62 | self.state = SchemaEngine.initialState(schema) |
| 63 | } |
| 64 | |
| 65 | public func start() { |
| 66 | guard !started else { return } |
| 67 | started = true |
| 68 | Task { await load() } |
| 69 | startSocket() |
| 70 | startTicker() |
| 71 | } |
| 72 | |
| 73 | public func stop() { |
| 74 | socket?.stop() |
| 75 | socket = nil |
| 76 | tickerTimer?.invalidate() |
| 77 | tickerTimer = nil |
| 78 | graceTask?.cancel() |
| 79 | started = false |
| 80 | } |
| 81 | |
| 82 | // MARK: load: snapshot seed + history fold |
| 83 | |
| 84 | private func load() async { |
| 85 | guard let api, let key else { return } |
| 86 | // Re-entrant, and a pure rebuild from the server's truth (snapshot + |
| 87 | // newer events ≡ all of history): a failed optimistic send reloads to |
| 88 | // reconcile, and starting clean is what makes the phantom fold — the |
| 89 | // event the server never saw — disappear. Live frames buffer again |
| 90 | // while this pass computes a fresh state. |
| 91 | loading = true |
| 92 | seededUpTo = 0 |
| 93 | // Seed from the encrypted snapshot so the list is complete even when |
| 94 | // old events have fallen out of the message window (1,000 todos work |
| 95 | // as well as 10 — Memory, apps/README.md). |
| 96 | let snap = await api.threadState(thread: thread.id, slot: "apps") |
| 97 | let rows = await api.messages(thread: thread.id, after: 0) |
| 98 | // The snapshot open + a decrypt/parse/fold per history row is the same |
| 99 | // work chat's foldPageDetached moved off the main actor — on it, an |
| 100 | // app-thread with a full window froze the UI on open exactly the way |
| 101 | // chats used to. Only the publish tail below touches main. |
| 102 | let schema = self.schema |
| 103 | let appId = self.appId |
| 104 | let folded = await Task.detached(priority: .userInitiated) { |
| 105 | () -> (state: AppJSON, seededUpTo: Int64, maxFolded: Int64) in |
| 106 | var next = SchemaEngine.initialState(schema) |
| 107 | var seededUpTo: Int64 = 0 |
| 108 | if let snap, !snap.blob.isEmpty, |
| 109 | let data = try? key.decryptData(blob: snap.blob), |
| 110 | let json = AppJSON.parse(data) { |
| 111 | let seeded = json["states"][appId] |
| 112 | if case .object = seeded { |
| 113 | next = seeded |
| 114 | seededUpTo = Int64(SchemaEngine.num(json["upTo"])) |
| 115 | } |
| 116 | } |
| 117 | var maxFolded: Int64 = 0 |
| 118 | for row in rows where row.id > seededUpTo { |
| 119 | if let event = Self.decryptEvent(key: key, blob: row.blob) { |
| 120 | next = SchemaEngine.fold(schema, state: next, event: event) |
| 121 | maxFolded = max(maxFolded, row.id) |
| 122 | } |
| 123 | } |
| 124 | return (next, seededUpTo, maxFolded) |
| 125 | }.value |
| 126 | seededUpTo = folded.seededUpTo |
| 127 | maxFolded = max(maxFolded, folded.maxFolded) |
| 128 | var next = folded.state |
| 129 | // Anything that arrived live while we were loading folds on top (the |
| 130 | // fetch may already contain it — reducers are idempotent by id). |
| 131 | for event in pendingLive { |
| 132 | next = SchemaEngine.fold(schema, state: next, event: event) |
| 133 | } |
| 134 | pendingLive = [] |
| 135 | loading = false |
| 136 | publish(next) |
| 137 | } |
| 138 | |
| 139 | private func decryptEvent(_ blob: String) -> AppJSON? { |
| 140 | guard let key else { return nil } |
| 141 | return Self.decryptEvent(key: key, blob: blob) |
| 142 | } |
| 143 | |
| 144 | private nonisolated static func decryptEvent(key: GroupKey, blob: String) -> AppJSON? { |
| 145 | guard let data = try? key.decryptData(blob: blob), |
| 146 | let json = AppJSON.parse(data), case .object = json else { return nil } |
| 147 | // Chat payloads (and anything without a type this schema owns) are not |
| 148 | // app events; fold() itself ignores unknown types, this just skips the |
| 149 | // obvious non-events cheaply. |
| 150 | guard let type = json["type"].stringValue, type != "chat" else { return nil } |
| 151 | return json |
| 152 | } |
| 153 | |
| 154 | private func publish(_ next: AppJSON) { |
| 155 | guard next != state else { return } |
| 156 | state = next |
| 157 | scheduleGraceRepaint() |
| 158 | } |
| 159 | |
| 160 | // MARK: live socket |
| 161 | |
| 162 | private func startSocket() { |
| 163 | guard let base = instance.baseURL else { return } |
| 164 | let socket = ChatSocket(baseURL: base, token: instance.deviceToken) |
| 165 | socket?.onFrame = { [weak self] frame in |
| 166 | guard let self else { return } |
| 167 | guard frame.thread_id == self.thread.id, frame.type == "msg", |
| 168 | frame.ephemeral != true, let id = frame.id, let blob = frame.blob else { return } |
| 169 | guard id > self.seededUpTo else { return } |
| 170 | if let event = self.decryptEvent(blob) { |
| 171 | self.maxFolded = max(self.maxFolded, id) |
| 172 | if self.loading { |
| 173 | self.pendingLive.append(event) |
| 174 | } else { |
| 175 | self.publish(SchemaEngine.fold(self.schema, state: self.state, event: event)) |
| 176 | } |
| 177 | } |
| 178 | } |
| 179 | socket?.start() |
| 180 | self.socket = socket |
| 181 | } |
| 182 | |
| 183 | // MARK: grace repaint — rows linger then tuck; state never changes, so a |
| 184 | // lapsed grace window needs an explicit repaint at its expiry. |
| 185 | |
| 186 | private func scheduleGraceRepaint() { |
| 187 | graceTask?.cancel() |
| 188 | guard case .object(let p) = schema["partition"], |
| 189 | let flag = p["flag"]?.stringValue, let atField = p["at"]?.stringValue, |
| 190 | let graceMs = p["graceMs"].map(SchemaEngine.num) else { return } |
| 191 | let now = Date().timeIntervalSince1970 * 1000 |
| 192 | let items = state[schema["collection"].stringValue ?? "items"].arrayValue |
| 193 | let expiries = items.compactMap { item -> Double? in |
| 194 | guard SchemaEngine.truthy(item[flag]) else { return nil } |
| 195 | let at = max(SchemaEngine.num(item[atField]), 1) |
| 196 | let expiry = at + graceMs |
| 197 | return expiry > now ? expiry : nil |
| 198 | } |
| 199 | guard let next = expiries.min() else { return } |
| 200 | graceTask = Task { [weak self] in |
| 201 | try? await Task.sleep(nanoseconds: UInt64(max(0, next - now + 40) * 1_000_000)) |
| 202 | guard !Task.isCancelled else { return } |
| 203 | // Animated: this publish is what tucks a lapsed row into its |
| 204 | // collapsible section, and the list should slide, not jump. |
| 205 | withAnimation(.default) { self?.graceBeat += 1 } |
| 206 | self?.scheduleGraceRepaint() |
| 207 | } |
| 208 | } |
| 209 | |
| 210 | // MARK: ticker (per-item timed rules, e.g. Reminders' offline fallback fire) |
| 211 | |
| 212 | private func startTicker() { |
| 213 | guard !schema["ticker"].isNull else { return } |
| 214 | let seconds = max(SchemaEngine.num(schema["ticker"]["seconds"]), 1) |
| 215 | tickerTimer = Timer.scheduledTimer(withTimeInterval: seconds, repeats: true) { [weak self] _ in |
| 216 | Task { @MainActor [weak self] in self?.tick() } |
| 217 | } |
| 218 | } |
| 219 | |
| 220 | private func tick() { |
| 221 | let now = Date().timeIntervalSince1970 * 1000 |
| 222 | let due = SchemaEngine.tickerRules(schema, state: state, now: now, fired: &tickerFired) |
| 223 | guard !due.isEmpty else { return } |
| 224 | Task { |
| 225 | for event in due { _ = await emit(event) } |
| 226 | } |
| 227 | } |
| 228 | |
| 229 | // MARK: actions |
| 230 | |
| 231 | public func run(_ name: String, item: AppJSON = .null, input: String = "", |
| 232 | extra: [String: AppJSON] = [:]) async { |
| 233 | _ = await SchemaEngine.runAction(schema, name: name, item: item, input: input, |
| 234 | extra: extra, io: self) |
| 235 | } |
| 236 | } |
| 237 | |
| 238 | // MARK: - the action io: the same wire the web's ctx uses |
| 239 | |
| 240 | extension SchemaAppStore: SchemaActionIO { |
| 241 | /// Stamp the transport fields (web: apps.js emit injects name/ts/uid) and |
| 242 | /// send the encrypted event as a normal message. Folds OPTIMISTICALLY — |
| 243 | /// the row appears the instant the gesture lands (the web's socket echo is |
| 244 | /// near-instant; a REST round-trip is a visible pause) — then queues the |
| 245 | /// POST. The socket echo re-folds harmlessly (reducers are idempotent by |
| 246 | /// id); a failed send hints inline and reloads from the server, which |
| 247 | /// drops the optimistic fold the server never saw. |
| 248 | public func emit(_ event: AppJSON) async -> Bool { |
| 249 | guard let key, var obj = event.objectValue, let base = instance.baseURL else { return false } |
| 250 | obj["name"] = .string(instance.screenName) |
| 251 | obj["ts"] = .number(Date().timeIntervalSince1970 * 1000) |
| 252 | if !myUserId.isEmpty { obj["uid"] = .string(myUserId) } |
| 253 | let stamped = AppJSON.object(obj) |
| 254 | guard let data = try? JSONSerialization.data(withJSONObject: stamped.anyValue), |
| 255 | let blob = try? key.encryptData(data) else { return false } |
| 256 | var comps = URLComponents(url: base.appending(path: "api/messages"), resolvingAgainstBaseURL: false)! |
| 257 | comps.queryItems = [URLQueryItem(name: "token", value: instance.deviceToken)] |
| 258 | guard let url = comps.url, |
| 259 | let body = try? JSONSerialization.data(withJSONObject: ["thread_id": thread.id, "blob": blob]) else { return false } |
| 260 | publish(SchemaEngine.fold(schema, state: state, event: stamped)) |
| 261 | let prev = sendChain |
| 262 | // Strong self: a queued send must survive cache eviction of the store. |
| 263 | sendChain = Task { |
| 264 | await prev?.value |
| 265 | await self.post(url: url, body: body, event: stamped) |
| 266 | } |
| 267 | return true |
| 268 | } |
| 269 | |
| 270 | private func post(url: URL, body: Data, event: AppJSON) async { |
| 271 | var req = URLRequest(url: url) |
| 272 | req.httpMethod = "POST" |
| 273 | req.setValue("application/json", forHTTPHeaderField: "Content-Type") |
| 274 | req.httpBody = body |
| 275 | if let (_, resp) = try? await URLSession.shared.data(for: req), |
| 276 | ((resp as? HTTPURLResponse)?.statusCode ?? 500) < 300 { return } |
| 277 | let text = event["text"].stringValue ?? "" |
| 278 | hint(text.isEmpty |
| 279 | ? "Couldn't send — check your connection and try again." |
| 280 | : "Couldn't send “\(text)” — check your connection and try again.") |
| 281 | await load() |
| 282 | } |
| 283 | |
| 284 | /// Encrypt a future event now and hand the ciphertext to the server's |
| 285 | /// scheduler (the web's ctx.schedule). The server learns only (thread, |
| 286 | /// time); nil = offline fallback, recorded as sid "". |
| 287 | public func schedule(event: AppJSON, atMs: Double) async -> String? { |
| 288 | guard let key, let api, var obj = event.objectValue else { return nil } |
| 289 | obj["name"] = .string(instance.screenName) |
| 290 | obj["ts"] = .number(Date().timeIntervalSince1970 * 1000) |
| 291 | if !myUserId.isEmpty { obj["uid"] = .string(myUserId) } |
| 292 | guard let data = try? JSONSerialization.data(withJSONObject: AppJSON.object(obj).anyValue), |
| 293 | let blob = try? key.encryptData(data) else { return nil } |
| 294 | let sec = atMs / 1000 |
| 295 | guard sec.isFinite, sec > 0, sec < 9e18 else { return nil } // Int64-conversion safety, not policy |
| 296 | return await api.schedule(thread: thread.id, deliverAt: Int64(sec), blob: blob) |
| 297 | } |
| 298 | |
| 299 | public func unschedule(_ sid: String) async { |
| 300 | guard !sid.isEmpty else { return } |
| 301 | await api?.unschedule(sid) |
| 302 | } |
| 303 | |
| 304 | public func newId() -> String { UUID().uuidString } |
| 305 | |
| 306 | public func parse(kind: String, text: String) -> AppJSON? { |
| 307 | guard kind == "reminder", let p = ReminderParse.parse(text) else { return nil } |
| 308 | return .object(["text": .string(p.text), "when": .number(p.when)]) |
| 309 | } |
| 310 | |
| 311 | public func hint(_ text: String) { |
| 312 | hintText = text |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | // MARK: - cache: keep a thread's store (fold + socket) alive across visits, |
| 317 | // the schema-app mirror of NativeFeedModelCache. |
| 318 | |
| 319 | @MainActor |
| 320 | public final class SchemaAppStoreCache { |
| 321 | public static let shared = SchemaAppStoreCache() |
| 322 | |
| 323 | private struct Entry { |
| 324 | let store: SchemaAppStore |
| 325 | let seedKey: String |
| 326 | var lastUsed = Date() |
| 327 | } |
| 328 | |
| 329 | private var entries: [String: Entry] = [:] |
| 330 | private let maxEntries = 4 |
| 331 | |
| 332 | public func store(for instance: SavedInstance, thread: ChatThread, myUserId: String, |
| 333 | appId: String? = nil) -> SchemaAppStore? { |
| 334 | let key = instance.id + "|" + thread.id + "|" + (appId ?? thread.appId ?? "") |
| 335 | if var entry = entries[key], entry.seedKey == thread.threadKeyB64 { |
| 336 | entry.lastUsed = Date() |
| 337 | entries[key] = entry |
| 338 | return entry.store |
| 339 | } |
| 340 | entries.removeValue(forKey: key)?.store.stop() |
| 341 | guard let store = SchemaAppStore(instance: instance, thread: thread, myUserId: myUserId, appId: appId) else { return nil } |
| 342 | store.start() |
| 343 | entries[key] = Entry(store: store, seedKey: thread.threadKeyB64) |
| 344 | while entries.count > maxEntries, |
| 345 | let oldest = entries.min(by: { $0.value.lastUsed < $1.value.lastUsed }) { |
| 346 | entries.removeValue(forKey: oldest.key)?.store.stop() |
| 347 | } |
| 348 | return store |
| 349 | } |
| 350 | } |
| 351 | |
amadan