eleven / ocho Public

Download this file

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	/// Walk ONE thread's history — the Media/Links/Docs screen's auto-complete
56	/// (spec § Completeness). Same resumable marks as the search walk, so
57	/// either feature finishing the walk finishes it for both. The key is the
58	/// caller's already-unwrapped copy (ChatThread.threadKeyB64), so this
59	/// works from a screen that never saw the wrapped blob.
60	public func start(instance: SavedInstance, threadID: String, threadKind: String,
61	threadKeyB64: String) {
62	guard !running, task == nil else { return }
63	guard let base = instance.baseURL,
64	let raw = Data(base64urlEncoded: threadKeyB64),
65	let key = try? GroupKey(rawKey: raw) else { return }
66	running = true
67	loadingLabel = true
68	progress = "Loading earlier messages…"
69	let api = LChatAPI(baseURL: base, token: instance.deviceToken)
70	task = Task { [weak self] in
71	_ = await self?.walkThread(inst: instance, api: api,
72	threadID: threadID, kind: threadKind, key: key)
73	await MainActor.run {
74	self?.running = false
75	self?.progress = nil
76	self?.task = nil
77	self?.loadingLabel = false
78	}
79	}
80	}
81
82	// The single-thread walk narrates "Loading earlier messages…" (it loads
83	// the chat's history, not a search); the instance walk keeps "Searching".
84	private var loadingLabel = false
85
86	public func cancel() {
87	task?.cancel()
88	}
89
90	private func walkInstance(_ inst: SavedInstance) async {
91	guard let base = inst.baseURL else { return }
92	let api = LChatAPI(baseURL: base, token: inst.deviceToken)
93	var sawIncomplete = false
94	for t in await api.threads() {
95	if Task.isCancelled { break }
96	// Chats only — app-threads' logs are bounded and load whole already.
97	guard t.kind == "group" || t.kind == "dm",
98	let wrapped = t.wrapped, !wrapped.isEmpty,
99	let raw = try? ECIES.unwrap(wrappedB64: wrapped, privPkcs8B64: inst.userPriv),
100	let key = try? GroupKey(rawKey: raw) else { continue }
101	let done = await walkThread(inst: inst, api: api,
102	threadID: t.id, kind: t.kind, key: key)
103	if !done { sawIncomplete = true }
104	}
105	if !Task.isCancelled, !sawIncomplete {
106	_ = await MainActor.run { completed.insert(inst.id) }
107	}
108	}
109
110	/// Walk one thread to its beginning (or cancellation). Returns whether the
111	/// thread is complete.
112	private func walkThread(inst: SavedInstance, api: LChatAPI,
113	threadID: String, kind: String, key: GroupKey) async -> Bool {
114	let mark = await MessageStore.shared.backfillMark(instance: inst.id, thread: threadID)
115	let oldestStored = await MessageStore.shared.oldestStoredId(instance: inst.id, thread: threadID)
116	guard var cursor = Self.cursor(markOldest: mark.oldest, markComplete: mark.complete,
117	oldestStored: oldestStored) else { return true }
118	if cursor == 0 {
119	// Empty store for this thread: seed the recent window for a cursor.
120	let rows = await api.messages(thread: threadID, after: 0)
121	guard let first = rows.first else {
122	// No history at all — trivially complete.
123	await MessageStore.shared.setBackfillMark(instance: inst.id, thread: threadID,
124	oldest: 0, complete: true)
125	return true
126	}
127	await file(rows, inst: inst, threadID: threadID, kind: kind, key: key)
128	cursor = first.id
129	await MessageStore.shared.setBackfillMark(instance: inst.id, thread: threadID,
130	oldest: cursor, complete: false)
131	}
132	while !Task.isCancelled {
133	let rows = await api.messagesBefore(thread: threadID, before: cursor)
134	guard let first = rows.first else {
135	await MessageStore.shared.setBackfillMark(instance: inst.id, thread: threadID,
136	oldest: cursor, complete: true)
137	return true
138	}
139	await file(rows, inst: inst, threadID: threadID, kind: kind, key: key)
140	cursor = first.id
141	await MessageStore.shared.setBackfillMark(instance: inst.id, thread: threadID,
142	oldest: cursor, complete: false)
143	await MainActor.run {
144	self.progress = Self.progressLabel(first.created_at, loading: self.loadingLabel)
145	self.pageTick += 1
146	}
147	}
148	return false
149	}
150
151	/// Decrypt + file one page — BackgroundSync's recipe: tombstones evict
152	/// their targets (#582), renderable rows upsert (decrypted once, here).
153	private func file(_ rows: [ServerMessage], inst: SavedInstance,
154	threadID: String, kind: String, key: GroupKey) async {
155	for id in ChatFeedModel.tombstoneEvictions(rows) {
156	await MessageStore.shared.delete(instance: inst.id, thread: threadID, id: id)
157	}
158	let batch: [MessageStore.StoredMessage] = rows.compactMap { row -> MessageStore.StoredMessage? in
159	guard (row.tombstone_of ?? 0) <= 0 else { return nil }
160	guard let p = ChatFeedModel.renderable(row.blob, key: key,
161	group: kind == "group") else { return nil }
162	return .init(id: row.id, at: row.created_at,
163	edited: (row.edited_at ?? 0) > 0,
164	reactions: [], payload: p)
165	}
166	guard !batch.isEmpty else { return }
167	await MessageStore.shared.upsert(instance: inst.id, thread: threadID, batch)
168	}
169
170	/// "Searching back to March 2025…" — how deep the time machine has dug —
171	/// or the media screen's "Loading earlier messages… back to March 2025".
172	nonisolated static func progressLabel(_ unixSecs: Int64, loading: Bool = false) -> String {
173	guard unixSecs > 0 else { return loading ? "Loading earlier messages…" : "Searching…" }
174	let f = DateFormatter()
175	f.dateFormat = "MMMM yyyy"
176	let month = f.string(from: Date(timeIntervalSince1970: TimeInterval(unixSecs)))
177	return loading ? "Loading earlier messages… back to \(month)" : "Searching back to \(month)…"
178	}
179	}
180