| 1 | import Foundation |
| 2 | |
| 3 | // 🤖 Full-history backfill into the local store (#584) — the native twin of the |
| 4 | // web's runBackfill (#457): page each chat's history backwards with the |
| 5 | // server's `before` cursor, decrypting rows once and filing them into |
| 6 | // MessageStore, persisting an oldest-fetched watermark per thread so an |
| 7 | // interrupted walk RESUMES rather than restarts. A thread that reaches its |
| 8 | // beginning (an empty page) is marked complete and never re-walked — the |
| 9 | // ordinary forward sync keeps coverage total from then on. |
| 10 | // |
| 11 | // User-triggered from search ("Search earlier messages…"), never automatic: |
| 12 | // full history is a deliberate download, not a background surprise. One |
| 13 | // runner instance drives both the iOS thread-list search and the Mac header |
| 14 | // search; views observe `progress` for the #228 time machine ("Searching back |
| 15 | // to March 2025…") and call cancel() for Stop — the watermark survives, so |
| 16 | // stopping costs nothing. |
| 17 | @MainActor |
| 18 | public final class HistoryBackfill: ObservableObject { |
| 19 | @Published public private(set) var running = false |
| 20 | @Published public private(set) var progress: String? |
| 21 | /// Bumps once per filed page — search UIs re-run their query on it, so |
| 22 | /// hits stream in while the walk digs. |
| 23 | @Published public private(set) var pageTick = 0 |
| 24 | /// Instances whose every chat completed — the trigger hides for them. |
| 25 | @Published public private(set) var completed: Set<String> = [] |
| 26 | |
| 27 | private var task: Task<Void, Never>? |
| 28 | |
| 29 | public init() {} |
| 30 | |
| 31 | /// Pure cursor rule (mirrors the web's backfillCursor; unit-tested): |
| 32 | /// nil = the thread is complete, nothing to do; 0 = nothing stored yet, |
| 33 | /// seed the recent page first; otherwise the `before` cursor to resume at. |
| 34 | nonisolated public static func cursor(markOldest: Int64, markComplete: Bool, |
| 35 | oldestStored: Int64) -> Int64? { |
| 36 | if markComplete { return nil } |
| 37 | if markOldest > 0 { return markOldest } |
| 38 | return oldestStored |
| 39 | } |
| 40 | |
| 41 | public func start(instance: SavedInstance) { |
| 42 | guard !running, task == nil else { return } // one walk at a time |
| 43 | running = true |
| 44 | progress = "Searching…" |
| 45 | task = Task { [weak self] in |
| 46 | await self?.walkInstance(instance) |
| 47 | await MainActor.run { |
| 48 | self?.running = false |
| 49 | self?.progress = nil |
| 50 | self?.task = nil |
| 51 | } |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | public func cancel() { |
| 56 | task?.cancel() |
| 57 | } |
| 58 | |
| 59 | private func walkInstance(_ inst: SavedInstance) async { |
| 60 | guard let base = inst.baseURL else { return } |
| 61 | let api = LChatAPI(baseURL: base, token: inst.deviceToken) |
| 62 | var sawIncomplete = false |
| 63 | for t in await api.threads() { |
| 64 | if Task.isCancelled { break } |
| 65 | // Chats only — app-threads' logs are bounded and load whole already. |
| 66 | guard t.kind == "group" || t.kind == "dm", |
| 67 | let wrapped = t.wrapped, !wrapped.isEmpty, |
| 68 | let raw = try? ECIES.unwrap(wrappedB64: wrapped, privPkcs8B64: inst.userPriv), |
| 69 | let key = try? GroupKey(rawKey: raw) else { continue } |
| 70 | let done = await walkThread(inst: inst, api: api, thread: t, key: key) |
| 71 | if !done { sawIncomplete = true } |
| 72 | } |
| 73 | if !Task.isCancelled, !sawIncomplete { |
| 74 | _ = await MainActor.run { completed.insert(inst.id) } |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | /// Walk one thread to its beginning (or cancellation). Returns whether the |
| 79 | /// thread is complete. |
| 80 | private func walkThread(inst: SavedInstance, api: LChatAPI, |
| 81 | thread t: ServerThread, key: GroupKey) async -> Bool { |
| 82 | let mark = await MessageStore.shared.backfillMark(instance: inst.id, thread: t.id) |
| 83 | let oldestStored = await MessageStore.shared.oldestStoredId(instance: inst.id, thread: t.id) |
| 84 | guard var cursor = Self.cursor(markOldest: mark.oldest, markComplete: mark.complete, |
| 85 | oldestStored: oldestStored) else { return true } |
| 86 | if cursor == 0 { |
| 87 | // Empty store for this thread: seed the recent window for a cursor. |
| 88 | let rows = await api.messages(thread: t.id, after: 0) |
| 89 | guard let first = rows.first else { |
| 90 | // No history at all — trivially complete. |
| 91 | await MessageStore.shared.setBackfillMark(instance: inst.id, thread: t.id, |
| 92 | oldest: 0, complete: true) |
| 93 | return true |
| 94 | } |
| 95 | await file(rows, inst: inst, thread: t, key: key) |
| 96 | cursor = first.id |
| 97 | await MessageStore.shared.setBackfillMark(instance: inst.id, thread: t.id, |
| 98 | oldest: cursor, complete: false) |
| 99 | } |
| 100 | while !Task.isCancelled { |
| 101 | let rows = await api.messagesBefore(thread: t.id, before: cursor) |
| 102 | guard let first = rows.first else { |
| 103 | await MessageStore.shared.setBackfillMark(instance: inst.id, thread: t.id, |
| 104 | oldest: cursor, complete: true) |
| 105 | return true |
| 106 | } |
| 107 | await file(rows, inst: inst, thread: t, key: key) |
| 108 | cursor = first.id |
| 109 | await MessageStore.shared.setBackfillMark(instance: inst.id, thread: t.id, |
| 110 | oldest: cursor, complete: false) |
| 111 | await MainActor.run { |
| 112 | self.progress = Self.progressLabel(first.created_at) |
| 113 | self.pageTick += 1 |
| 114 | } |
| 115 | } |
| 116 | return false |
| 117 | } |
| 118 | |
| 119 | /// Decrypt + file one page — BackgroundSync's recipe: tombstones evict |
| 120 | /// their targets (#582), renderable rows upsert (decrypted once, here). |
| 121 | private func file(_ rows: [ServerMessage], inst: SavedInstance, |
| 122 | thread t: ServerThread, key: GroupKey) async { |
| 123 | for id in ChatFeedModel.tombstoneEvictions(rows) { |
| 124 | await MessageStore.shared.delete(instance: inst.id, thread: t.id, id: id) |
| 125 | } |
| 126 | let batch: [MessageStore.StoredMessage] = rows.compactMap { row -> MessageStore.StoredMessage? in |
| 127 | guard (row.tombstone_of ?? 0) <= 0 else { return nil } |
| 128 | guard let p = ChatFeedModel.renderable(row.blob, key: key, |
| 129 | group: t.kind == "group") else { return nil } |
| 130 | return .init(id: row.id, at: row.created_at, |
| 131 | edited: (row.edited_at ?? 0) > 0, |
| 132 | reactions: [], payload: p) |
| 133 | } |
| 134 | guard !batch.isEmpty else { return } |
| 135 | await MessageStore.shared.upsert(instance: inst.id, thread: t.id, batch) |
| 136 | } |
| 137 | |
| 138 | /// "Searching back to March 2025…" — how deep the time machine has dug. |
| 139 | nonisolated static func progressLabel(_ unixSecs: Int64) -> String { |
| 140 | guard unixSecs > 0 else { return "Searching…" } |
| 141 | let f = DateFormatter() |
| 142 | f.dateFormat = "MMMM yyyy" |
| 143 | let month = f.string(from: Date(timeIntervalSince1970: TimeInterval(unixSecs))) |
| 144 | return "Searching back to \(month)…" |
| 145 | } |
| 146 | } |
| 147 | |
amadan