| 1 | import Foundation |
| 2 | |
| 3 | // 🤖 Background message sync: pull what's new for every joined instance and |
| 4 | // file it — decrypted — into the local store (MessageStore), entirely off the |
| 5 | // main thread. Driven by iOS Background App Refresh (BGAppRefreshTask in the |
| 6 | // app target); the Mac doesn't need it (its sockets are live while it runs). |
| 7 | // |
| 8 | // This is the same ingest path the feed and the NSE use: fetch ciphertext, |
| 9 | // unwrap the thread key natively, decrypt once, upsert by (instance, thread, |
| 10 | // id). Idempotent by construction, so overlapping with a live app session or |
| 11 | // an NSE ingest is harmless. |
| 12 | public enum BackgroundSync { |
| 13 | /// Threads synced per instance per pass, newest activity first — bounds |
| 14 | /// the pass to the BG-task budget (~30s); anything missed catches up on |
| 15 | /// the next pass or the next app open. |
| 16 | public static let maxThreadsPerInstance = 20 |
| 17 | |
| 18 | /// Sync every instance. Returns the number of new messages stored. |
| 19 | /// Cooperatively cancellable (the BG task's expiration handler cancels). |
| 20 | @discardableResult |
| 21 | public static func run(instances: [SavedInstance]) async -> Int { |
| 22 | var stored = 0 |
| 23 | for inst in instances { |
| 24 | if Task.isCancelled { break } |
| 25 | stored += await sync(inst) |
| 26 | } |
| 27 | return stored |
| 28 | } |
| 29 | |
| 30 | private static func sync(_ inst: SavedInstance) async -> Int { |
| 31 | guard let base = inst.baseURL else { return 0 } |
| 32 | let api = LChatAPI(baseURL: base, token: inst.deviceToken) |
| 33 | let threads = await api.threads() |
| 34 | .sorted { ($0.last_at ?? 0) > ($1.last_at ?? 0) } |
| 35 | .prefix(maxThreadsPerInstance) |
| 36 | var stored = 0 |
| 37 | for t in threads { |
| 38 | if Task.isCancelled { break } |
| 39 | guard let wrapped = t.wrapped, !wrapped.isEmpty, |
| 40 | let raw = try? ECIES.unwrap(wrappedB64: wrapped, privPkcs8B64: inst.userPriv), |
| 41 | let key = try? GroupKey(rawKey: raw) else { continue } |
| 42 | let local = await MessageStore.shared.latest(instance: inst.id, thread: t.id) |
| 43 | // Nothing newer than what we hold — skip the messages round-trip. |
| 44 | if local.at > 0, (t.last_at ?? 0) <= local.at { continue } |
| 45 | let rows = await api.messages(thread: t.id, after: local.id) |
| 46 | guard !rows.isEmpty else { continue } |
| 47 | // Deletes that happened while the app was suspended arrive as |
| 48 | // tombstone rows (#582) — drop the targets from the store (and |
| 49 | // its FTS index) before filing the page's real messages. |
| 50 | for id in ChatFeedModel.tombstoneEvictions(rows) { |
| 51 | await MessageStore.shared.delete(instance: inst.id, thread: t.id, id: id) |
| 52 | } |
| 53 | let batch: [MessageStore.StoredMessage] = rows.compactMap { row in |
| 54 | guard (row.tombstone_of ?? 0) <= 0 else { return nil } |
| 55 | guard let p = ChatFeedModel.renderable(row.blob, key: key, |
| 56 | group: t.kind == "group") else { return nil } |
| 57 | // Reactions are encrypted events now — rows carry no tally; |
| 58 | // the feed derives it by folding the events on load. |
| 59 | return .init(id: row.id, at: row.created_at, |
| 60 | edited: (row.edited_at ?? 0) > 0, |
| 61 | reactions: [], payload: p) |
| 62 | } |
| 63 | guard !batch.isEmpty else { continue } |
| 64 | await MessageStore.shared.upsert(instance: inst.id, thread: t.id, batch) |
| 65 | stored += batch.count |
| 66 | } |
| 67 | return stored |
| 68 | } |
| 69 | } |
| 70 | |
amadan