feat(chat): sync group rosters on device and search them locally

.claude/docs/hard-rules.md

@@ -73,7 +73,11 @@ app can fix itself belongs in the service files that wrap the generated code.
 migrations — when the version number increases, the database is deleted and rebuilt from server data
 on next login (`SessionAuthenticator.initializeDatabase`). This means:
 
-- Adding/removing tables or columns → bump version
+- Removing a table, or adding/removing/changing columns on an existing table → bump version
+- **Adding a new table does not bump**, as long as it starts empty and fills from the server.
+  `createTablesIfNeeded()` runs on every open with `create(ifNotExists: true)`, so an existing store
+  gains the table in place and keeps its data. Code reading the new table must treat "empty" as
+  "never synced". Any change to an existing table's columns still bumps.
 - Changing which table a query reads from → bump version if the old schema can't satisfy the new query
 - **Changing the encoding of a value already stored in a column → bump version.** The column's SQL
   type is unchanged, so nothing about `Schema.swift` looks different, but rows written by an earlier

CLAUDE.md

@@ -105,7 +105,7 @@ it before touching the relevant area.
 - **Exhaustive switches** — Prefer `switch` over `if case` for enums so the compiler flags new cases.
 - **Modernize incrementally** — Use modern Swift/SwiftUI APIs in net-new/isolated code; don't refactor working code just to modernize. One observation system per class.
 - **Generated protos** — `FlipcashAPI` only re-exports the published contract packages; there is no generated code to edit here. Change the wrapping service files instead.
-- **Database schema** — Bump `Database.schemaVersion` on every change to what the store *writes*, including the encoding of a persisted blob (no migrations; DB is rebuilt from server).
+- **Database schema** — Bump `Database.schemaVersion` on every change to what the store *writes*, including the encoding of a persisted blob (no migrations; DB is rebuilt from server). A change that only adds new tables, which start empty and fill from the server, does not bump.
 - **Logging** — Message string is a constant; every variable goes in structured `metadata`. Never log proto blobs whole.
 - **Error reporting** — Call `ErrorReporting.captureError(...)` unconditionally; classify via `ServerError.reportingLevel`, never gate at the call site. Best-effort chatter never reports.
 - **Form validation** — Validate free-form input through the `Validator` family and submit the validator's `Output`, never inline regex/trim; keypad amounts parse only via `AmountValidator`.

Flipcash/Core/Controllers/ConversationController.swift

@@ -261,6 +261,10 @@ final class ConversationController {
     /// The emoji reactions concern: the user's taps, their calls, and the stream's updates.
     let reactions: ConversationReactions
 
+    /// Keeps opened groups' full rosters on device from the stream's roster updates. Set by the
+    /// session after init; `nil` leaves roster updates unpersisted.
+    @ObservationIgnored var roster: RosterSync?
+
     init(
         fetching: any ConversationFetching,
         membership: any ConversationMembership,
@@ -408,6 +412,7 @@ final class ConversationController {
                 self.logCounterpartRead(event)
                 self.applyTyping(event)
                 self.applyReactions(event)
+                self.roster?.apply(event)
                 if case .needsCatchUp(let conversationID, _) = gap {
                     self.scheduleGapCatchUp(conversationID)
                 }

Flipcash/Core/Controllers/FlipClient+Protocols.swift

@@ -67,6 +67,12 @@ protocol ConversationFetching: AnyObject, Sendable {
     func getChat(owner: KeyPair, conversationID: ConversationID) async throws -> Conversation
 }
 
+/// Roster read surface used by `RosterSync`. Maps 1:1 to `flipcash.chat.v1.Chat.GetRoster`.
+protocol RosterFetching: AnyObject, Sendable {
+    /// One page of a chat's roster, most recently joined first; see ``FlipClient/RosterPage``.
+    func getRosterPage(owner: KeyPair, conversationID: ConversationID, pagingToken: Data?) async throws -> FlipClient.RosterPage
+}
+
 /// Group membership surface used by `ConversationController`. Separate from ``ConversationFetching``
 /// because these write: they are the only chat RPCs that change what the caller is a member of.
 protocol ConversationMembership: AnyObject, Sendable {
@@ -166,5 +172,5 @@ extension FlipClient {
 }
 
 extension FlipClient: ContactVerifying, OnrampAuthorizing, ContactSyncing,
-                      ConversationFetching, ConversationMembership,
+                      ConversationFetching, ConversationMembership, RosterFetching,
                       ConversationViewerSettings, ConversationEventStreaming {}

Flipcash/Core/Controllers/RosterSearch.swift

@@ -0,0 +1,115 @@
+//
+//  RosterSearch.swift
+//  Flipcash
+//
+
+import Foundation
+import FlipcashCore
+import FlipcashStore
+
+/// A group member a roster search matched.
+nonisolated struct MemberMatch: Identifiable, Hashable, Sendable {
+    let member: ConversationMember
+    let id: UserID
+}
+
+/// Searches a group's members by name or username, for the mention picker.
+///
+/// Callers depend on this rather than on ``LocalRosterSearch`` so a server-side search can replace
+/// or back the local one without them changing.
+protocol RosterSearchSource: AnyObject {
+    /// Readies the source to answer for a group; called when the picker opens.
+    func prepare(chatID: ConversationID) async
+
+    /// Returns up to `limit` members of a group matching `query`, best match first, never the
+    /// signed-in user or a blocked one. An empty query returns the members who spoke most recently.
+    func search(chatID: ConversationID, query: String, limit: Int) async throws -> [MemberMatch]
+}
+
+extension RosterSearchSource {
+    /// Returns up to 20 members of a group matching `query`; see ``search(chatID:query:limit:)``.
+    func search(chatID: ConversationID, query: String) async throws -> [MemberMatch] {
+        try await search(chatID: chatID, query: query, limit: 20)
+    }
+}
+
+/// Answers roster searches from the roster ``RosterSync`` holds on device.
+///
+/// A member matches when every word of the query is a prefix of one of their tokens, the words of
+/// their display name and their username, compared after ``RosterSearchText/normalize(_:)``. Matches
+/// rank in three tiers:
+/// 1. members who sent one of the chat's newest held messages, most recent first;
+/// 2. the member whose username is exactly the query;
+/// 3. everyone else, by normalized display name, then raw display name, both in code point order.
+/// Ties fall to the lowercase user id. The signed-in user and blocked users never match.
+final class LocalRosterSearch: RosterSearchSource {
+
+    /// How many of the chat's newest held messages count toward "spoke recently".
+    static let recentMessageWindow = 50
+
+    private let database: Database
+    private let roster: RosterSync
+    private let selfUserID: UserID
+    private let blockedUserIDs: () -> Set<UserID>
+
+    init(database: Database, roster: RosterSync, selfUserID: UserID, blockedUserIDs: @escaping () -> Set<UserID> = { [] }) {
+        self.database = database
+        self.roster = roster
+        self.selfUserID = selfUserID
+        self.blockedUserIDs = blockedUserIDs
+    }
+
+    func prepare(chatID: ConversationID) async {
+        await roster.refreshFirstPage(chatID)
+    }
+
+    func search(chatID: ConversationID, query: String, limit: Int) async throws -> [MemberMatch] {
+        let words = RosterSearchText.queryWords(query)
+        let recent = try database.recentSenders(conversationID: chatID, window: Self.recentMessageWindow)
+
+        let candidates: Set<UserID>
+        if words.isEmpty {
+            candidates = Set(recent)
+        } else {
+            candidates = try database.rosterMemberIDs(matchingPrefixes: words, conversationID: chatID)
+        }
+        guard !candidates.isEmpty, limit > 0 else { return [] }
+
+        let recency = Dictionary(recent.enumerated().map { ($1, $0) }, uniquingKeysWith: { first, _ in first })
+        let exactUsername = words.count == 1 ? words[0] : nil
+        let blocked = blockedUserIDs()
+
+        let ranked = try database.rosterEntries(conversationID: chatID, userIDs: candidates)
+            .compactMap { entry -> (Rank, RosterEntry, UserID)? in
+                guard let userID = entry.member.userID, userID != selfUserID, !blocked.contains(userID) else { return nil }
+                let username = entry.member.username.map { RosterSearchText.normalize($0.value) }
+                let rank: Rank
+                if let position = recency[userID] {
+                    rank = .recentSpeaker(position)
+                } else if let exactUsername, username == exactUsername {
+                    rank = .exactUsername
+                } else {
+                    rank = .alphabetical
+                }
+                return (rank, entry, userID)
+            }
+            .sorted { lhs, rhs in
+                if lhs.0 != rhs.0 { return lhs.0 < rhs.0 }
+                if lhs.1.sortKey != rhs.1.sortKey {
+                    return lhs.1.sortKey.unicodeScalars.lexicographicallyPrecedes(rhs.1.sortKey.unicodeScalars)
+                }
+                if lhs.1.member.displayName != rhs.1.member.displayName {
+                    return lhs.1.member.displayName.unicodeScalars.lexicographicallyPrecedes(rhs.1.member.displayName.unicodeScalars)
+                }
+                return lhs.2.uuidString.lowercased() < rhs.2.uuidString.lowercased()
+            }
+
+        return ranked.prefix(limit).map { MemberMatch(member: $0.1.member, id: $0.2) }
+    }
+
+    private enum Rank: Comparable {
+        case recentSpeaker(Int)
+        case exactUsername
+        case alphabetical
+    }
+}

Flipcash/Core/Controllers/RosterSync.swift

@@ -0,0 +1,196 @@
+//
+//  RosterSync.swift
+//  Flipcash
+//
+
+import Foundation
+import FlipcashCore
+import FlipcashStore
+
+nonisolated private let logger = Logger(label: "flipcash.roster-sync")
+
+/// Keeps the full roster of each group the user opens on device, for member search.
+///
+/// A group is tracked from its first open. The first open reads `Chat.GetRoster` to the end. Later
+/// opens catch up only when the roster moved past the last version a read applied: they re-read
+/// from the first page, newest joins first, and stop at a member already covered. ``apply(_:)``
+/// keeps a tracked group current from the stream in between. Groups never opened are not fetched.
+/// Every path merges per member by roster version, greater winning.
+actor RosterSync {
+
+    /// Pages a read fetches before it stops, `has_more` or not; at 100 per page, 2,000 members.
+    static let defaultPageCap = 20
+
+    private let fetching: any RosterFetching
+    private let database: Database
+    private let owner: KeyPair
+    private let pageCap: Int
+
+    /// Reads in flight, so a second trigger for the same group joins the first rather than reading twice.
+    private var inFlight: [ConversationID: Task<Void, Never>] = [:]
+
+    init(fetching: any RosterFetching, database: Database, owner: KeyPair, pageCap: Int = RosterSync.defaultPageCap) {
+        self.fetching = fetching
+        self.database = database
+        self.owner = owner
+        self.pageCap = pageCap
+    }
+
+    /// Starts tracking a group and brings its held roster up to date: a full read when none has
+    /// finished or a reconcile is owed, a catch-up when the roster moved past the watermark, nothing
+    /// otherwise. Returns once any read is done.
+    func syncIfNeeded(_ conversationID: ConversationID) async {
+        await run(conversationID, refreshFirstPage: false)
+    }
+
+    /// As ``syncIfNeeded(_:)``, but always re-reads at least the first page, so the newest members'
+    /// names and pictures are current; profile changes don't move the roster version.
+    func refreshFirstPage(_ conversationID: ConversationID) async {
+        await run(conversationID, refreshFirstPage: true)
+    }
+
+    /// Writes a stream roster update into the held roster of a tracked group; other events are ignored.
+    @MainActor
+    func apply(_ event: ConversationStreamEvent) {
+        guard case .rosterChanged(let conversationID, let updates) = event else { return }
+        do {
+            try database.applyRosterUpdates(updates, conversationID: conversationID)
+        } catch {
+            logger.error("Failed to apply roster updates", metadata: [
+                "conversationID": "\(conversationID)",
+                "error": "\(error)",
+            ])
+            ErrorReporting.captureError(error, reason: "Failed to apply roster updates")
+        }
+    }
+
+    // MARK: - Private -
+
+    private func run(_ conversationID: ConversationID, refreshFirstPage: Bool) async {
+        if let running = inFlight[conversationID] {
+            await running.value
+            return
+        }
+        let task = Task { await self.bringUpToDate(conversationID, refreshFirstPage: refreshFirstPage) }
+        inFlight[conversationID] = task
+        await task.value
+        inFlight[conversationID] = nil
+    }
+
+    private func bringUpToDate(_ conversationID: ConversationID, refreshFirstPage: Bool) async {
+        let state: RosterSyncState
+        do {
+            try database.beginTrackingRoster(conversationID: conversationID)
+            guard let tracked = try database.rosterSyncState(conversationID: conversationID) else { return }
+            state = tracked
+        } catch {
+            await report(error, reason: "Failed to read roster sync state", conversationID: conversationID)
+            return
+        }
+
+        if state.needsFullRead {
+            await fullRead(conversationID, reconciling: state.reconcilePending)
+            return
+        }
+        guard state.needsCatchUp || refreshFirstPage else { return }
+
+        do {
+            switch try await catchUp(conversationID, watermark: state.watermark) {
+            case .caughtUp:
+                break
+            case .reconcileNeeded:
+                await fullRead(conversationID, reconciling: true)
+            }
+        } catch {
+            await report(error, reason: "Failed to catch up roster", conversationID: conversationID)
+        }
+    }
+
+    /// Reads from the first page until it reaches a member the watermark already covers.
+    private func catchUp(_ conversationID: ConversationID, watermark: UInt64) async throws -> RosterCatchUpOutcome {
+        // Pages list the most recently joined first, and a member's version is the roster version of
+        // their join, so everyone after the first member at or below the watermark is already held.
+        // `Member.version` is documented to move on future member changes too (a role change, say);
+        // once one does, a member can sit above the watermark out of join order, and this stop rule
+        // needs revisiting.
+        let read = try await readPages(conversationID) { page in
+            page.members.contains { $0.version <= watermark }
+        }
+
+        guard read.stoppedEarly else {
+            // The read ran to the end or the cap without reaching the watermark: it's a full read.
+            try database.applyFullRosterRead(read.members, summaries: read.summaries, isComplete: read.reachedEnd, conversationID: conversationID)
+            logRead("Roster catch-up became a full read", read, conversationID: conversationID)
+            return .caughtUp
+        }
+
+        guard let summary = read.summaries.min(by: { $0.version < $1.version }) else { return .caughtUp }
+        let outcome = try database.applyRosterCatchUp(read.members, summary: summary, conversationID: conversationID)
+        logRead("Caught up roster", read, conversationID: conversationID)
+        return outcome
+    }
+
+    /// Reads the whole roster, up to the page cap. A reconcile runs at background priority, since it
+    /// only removes members who left; a failure leaves it pending for the next open.
+    private func fullRead(_ conversationID: ConversationID, reconciling: Bool) async {
+        let task = Task(priority: reconciling ? .background : nil) {
+            let read = try await self.readPages(conversationID) { _ in false }
+            try self.database.applyFullRosterRead(read.members, summaries: read.summaries, isComplete: read.reachedEnd, conversationID: conversationID)
+            self.logRead("Read full roster", read, conversationID: conversationID)
+        }
+        do {
+            try await task.value
+        } catch {
+            await report(error, reason: "Failed to read full roster", conversationID: conversationID)
+        }
+    }
+
+    private struct Read {
+        var members: [ConversationMember] = []
+        var summaries: [ConversationRosterSummary] = []
+        var pages = 0
+        var reachedEnd = false
+        var stoppedEarly = false
+    }
+
+    /// Pages from the first page until `stop` returns `true` for a page, the last page, or the cap.
+    /// Throws without returning anything read when a page fails, so a partial read records nothing.
+    private func readPages(_ conversationID: ConversationID, stop: (FlipClient.RosterPage) -> Bool) async throws -> Read {
+        var read = Read()
+        var pagingToken: Data?
+        while read.pages < pageCap {
+            let page = try await fetching.getRosterPage(owner: owner, conversationID: conversationID, pagingToken: pagingToken)
+            read.pages += 1
+            read.members.append(contentsOf: page.members)
+            read.summaries.append(page.rosterSummary)
+            if stop(page) {
+                read.stoppedEarly = true
+                break
+            }
+            guard let next = page.nextPagingToken else {
+                read.reachedEnd = true
+                break
+            }
+            pagingToken = next
+        }
+        return read
+    }
+
+    private func logRead(_ message: Logger.Message, _ read: Read, conversationID: ConversationID) {
+        logger.info(message, metadata: [
+            "conversationID": "\(conversationID)",
+            "members": "\(read.members.count)",
+            "pages": "\(read.pages)",
+            "reachedEnd": "\(read.reachedEnd)",
+        ])
+    }
+
+    private func report(_ error: Error, reason: String, conversationID: ConversationID) async {
+        logger.error("Roster sync failed", metadata: [
+            "reason": "\(reason)",
+            "conversationID": "\(conversationID)",
+            "error": "\(error)",
+        ])
+        await ErrorReporting.captureError(error, reason: reason)
+    }
+}

Flipcash/Core/Screens/Conversation/ConversationScreen.swift

@@ -680,6 +680,12 @@ struct ConversationScreen: View {
             guard groupConversation != nil else { return }
             await sessionContainer.knownAuthors.reload()
         }
+        // Page the group's full roster into the store for member search. Detached from this task
+        // inside `RosterSync`, so leaving the chat mid-sync doesn't discard the pages fetched.
+        .task(id: groupConversation?.id) {
+            guard let groupID = groupConversation?.id else { return }
+            await conversationController.roster?.syncIfNeeded(groupID)
+        }
         // Name the senders the chat's roster and the local cache both leave out. Keyed on that set,
         // so it runs when a page of older messages reveals a sender nothing here can name — and not
         // again once the fetch has landed them.