feat: discover team members in large teams - WPB-24947

WireDomain/Sources/WireDomain/Components/ClientSessionComponent.swift

@@ -398,6 +398,12 @@
         mlsService: mlsService
     )
 
+    private lazy var teamMemberDiscoveryAgent = TeamMemberDiscoveryAgent(
+        api: teamsAPI,
+        store: teamLocalStore,
+        journal: journal
+    )
+
     public private(set) lazy var incrementalSync = IncrementalSync(
         selfClientID: selfClientID,
         pushChannelAPI: pushChannelAPI,
@@ -411,7 +417,8 @@
         liveBrokenGroupSubject: liveBrokenGroupSubject,
         journal: journal,
         mlsGroupRepairAgent: mlsGroupRepairAgent,
-        earService: earService
+        earService: earService,
+        teamMemberDiscoveryAgent: teamMemberDiscoveryAgent
     )
 
     public lazy var incrementalSyncV2: IncrementalSyncV2 = if let sharedContainerURL {
@@ -430,6 +437,7 @@
             journal: journal,
             mlsGroupRepairAgent: mlsGroupRepairAgent,
             earService: earService,
+            teamMemberDiscoveryAgent: teamMemberDiscoveryAgent,
             createPushChannelState: { [selfClientID] in
                 PushChannelState(sharedContainerURL: sharedContainerURL, clientID: selfClientID)
             }
@@ -844,7 +852,7 @@
     )
 
     public lazy var workAgent: WorkAgent = .init(scheduler: PriorityOrderWorkItemScheduler())
 
     public lazy var conversationUpdatesGenerator: IncrementalGeneratorProtocol = ConversationUpdatesGenerator(
         repository: conversationRepository,
         context: syncContext,
@@ -853,7 +861,7 @@
             self?.workAgent.submitItem(workItem)
         }
     )
 
     public lazy var commitPendingProposalsGenerator: LiveGeneratorProtocol = CommitPendingProposalsGenerator(
         repository: conversationRepository,
         mlsService: mlsService,
@@ -866,7 +874,7 @@
             self?.workAgent.submitItem(workItem)
         }
     )
 
     public lazy var invalidMLSGroupGenerator: IncrementalGeneratorProtocol = InvalidMLSGroupGenerator(
         context: syncContext,
         mlsService: mlsService,
@@ -901,7 +909,7 @@
         liveBrokenGroupSubject.sink { [weak self]  liveMLSBrokenGroups in
             guard let self else { return }
             WireLogger.mls.debug("detected during live sync \(liveMLSBrokenGroups.count) broken MLS groups")
             let item = RepairBrokenMLSGroupsItem(repairAgent: mlsGroupRepairAgent)
             Task {
                 await self.workAgent.submitItem(item)
             }

WireDomain/Sources/WireDomain/Synchronization/IncrementalSync.swift

@@ -45,6 +45,7 @@
     private let journal: Journal
     private let mlsGroupRepairAgent: MLSGroupRepairAgentProtocol
     private let earService: EARServiceInterface
+    private let teamMemberDiscoveryAgent: any TeamMemberDiscoveryAgentProtocol
 
     public init(
         selfClientID: String,
@@ -59,7 +60,8 @@
         liveBrokenGroupSubject: PassthroughSubject<Set<String>, Never>,
         journal: Journal,
         mlsGroupRepairAgent: MLSGroupRepairAgentProtocol,
-        earService: EARServiceInterface
+        earService: EARServiceInterface,
+        teamMemberDiscoveryAgent: any TeamMemberDiscoveryAgentProtocol
     ) {
         self.selfClientID = selfClientID
         self.pushChannelAPI = pushChannelAPI
@@ -74,6 +76,7 @@
         self.journal = journal
         self.mlsGroupRepairAgent = mlsGroupRepairAgent
         self.earService = earService
+        self.teamMemberDiscoveryAgent = teamMemberDiscoveryAgent
     }
 
     public func perform() async throws -> Token {
@@ -161,14 +164,18 @@
 
             await mlsGroupRepairAgent.repairConversations()
 
+            if !backgroundAccessibleOnly {
+                await teamMemberDiscoveryAgent.discoverMembers()
+            }
+
             let liveEventTask = Task { @Sendable [self] in
                 logger.info("handling live event stream", attributes: .incrementalSyncV2, .safePublic)
                 syncStateSubject.send(.liveSyncing(.ongoing))
 
                 do {
                     // because we might be interrupted when in background, we wrap the sync in an expiringActivity that
                     // will cancel the task - not keeping any db operation (sqlite file opened) in suspend mode
                     try await withExpiringActivity(reason: "processLiveStream IncrementalSync") {
                         await processLiveEvents(
                             liveEventStream: liveEventStream,
                             processedEnvelopeIDs: processedEnvelopeIDs,
@@ -349,7 +356,7 @@
                 privateKeys: privateKeys,
                 backgroundAccessibleOnly: backgroundAccessibleOnly
             )
             let envelopes = envelopesWithObjectIDs.map(\.envelope)
             let envelopesObjectIDs = envelopesWithObjectIDs.map(\.objectID)
 
             guard !envelopes.isEmpty else {

WireDomain/Sources/WireDomain/Synchronization/IncrementalSyncV2.swift

@@ -20,7 +20,7 @@
 import Foundation
 import WireCoreCrypto
 import WireDataModel
 import WireLogging
 import WireNetwork
 
 public typealias CreatePushChannelStateClosure = () -> PushChannelStateProtocol
@@ -52,6 +52,7 @@
     private let createPushChannelState: CreatePushChannelStateClosure
     private let mlsGroupRepairAgent: MLSGroupRepairAgentProtocol
     private let earService: EARServiceInterface
+    private let teamMemberDiscoveryAgent: any TeamMemberDiscoveryAgentProtocol
 
     weak var delegate: (any LiveSyncDelegate)?
 
@@ -70,6 +71,7 @@
         journal: Journal,
         mlsGroupRepairAgent: MLSGroupRepairAgentProtocol,
         earService: EARServiceInterface,
+        teamMemberDiscoveryAgent: any TeamMemberDiscoveryAgentProtocol,
         createPushChannelState: @escaping CreatePushChannelStateClosure,
         syncMarkerGenerator: @escaping SyncMarkerGenerator = { UUID().uuidString }
     ) {
@@ -87,6 +89,7 @@
         self.journal = journal
         self.mlsGroupRepairAgent = mlsGroupRepairAgent
         self.earService = earService
+        self.teamMemberDiscoveryAgent = teamMemberDiscoveryAgent
         self.syncMarkerGenerator = syncMarkerGenerator
         self.createPushChannelState = createPushChannelState
     }
@@ -152,6 +155,8 @@
 
         await mlsGroupRepairAgent.repairConversations()
 
+        await teamMemberDiscoveryAgent.discoverMembers()
+
         let task = Task { @Sendable [self] in
             logger.debug("handling live event stream", attributes: logAttributes)
             syncStateSubject.send(.liveSyncing(.ongoing))
@@ -160,14 +165,14 @@
                 // because we might be interrupted when in background, we wrap the sync in an expiringActivity that will
                 // cancel the task (not keeping any file lock in suspend mode)
                 try await withExpiringActivity(reason: "processLiveStream IncrementalSyncV2") {
                     await processLiveStream(
                         liveEventStream,
                         pushChannel: pushChannel,
                         syncMarker: syncMarker
                     )
 
                     WireLogger.sync.debug("Live stream ended, close push channel", attributes: logAttributes)
                     await pushChannel.close()
                     await pushChannelState.markAsClosed()
                 }
             } catch {

WireDomain/Sources/WireDomain/Synchronization/Protocols/TeamMemberDiscoveryAgentProtocol.swift

@@ -0,0 +1,30 @@
+//
+// Wire
+// Copyright (C) 2026 Wire Swiss GmbH
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see http://www.gnu.org/licenses/.
+//
+
+import Foundation
+
+// sourcery: AutoMockable
+/// Discovers self-team members by replaying recent team notifications,
+/// working around the 2000-member cap on the legacy bulk team-members endpoint.
+public protocol TeamMemberDiscoveryAgentProtocol {
+
+    /// Fetch recent team notifications and apply any discovered
+    /// `team.member-join` events to local storage.
+    func discoverMembers() async
+
+}

WireDomain/Sources/WireDomain/Synchronization/TeamMemberDiscoveryAgent.swift

@@ -0,0 +1,105 @@
+//
+// Wire
+// Copyright (C) 2026 Wire Swiss GmbH
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see http://www.gnu.org/licenses/.
+//
+
+import Foundation
+import WireLogging
+import WireNetwork
+
+struct TeamMemberDiscoveryAgent: TeamMemberDiscoveryAgentProtocol {
+
+    private let api: any TeamsAPI
+    private let store: any TeamLocalStoreProtocol
+    private let journal: Journal
+
+    // MARK: - Life cycle
+
+    init(
+        api: any TeamsAPI,
+        store: any TeamLocalStoreProtocol,
+        journal: Journal
+    ) {
+        self.api = api
+        self.store = store
+        self.journal = journal
+    }
+
+    // MARK: - TeamMemberDiscoveryAgentProtocol
+
+    func discoverMembers() async {
+        guard let selfTeamID = await store.selfTeamID() else {
+            WireLogger.sync.debug("team member discovery: self user is not in a team")
+            return
+        }
+
+        do {
+            try await runDiscovery(selfTeamID: selfTeamID)
+        } catch TeamsAPIError.missedEvents {
+            WireLogger.sync.warn(
+                "team member discovery: server lost cursor history, resetting and re-running",
+                attributes: .safePublic
+            )
+            journal[.lastTeamNotificationID] = nil
+            do {
+                try await runDiscovery(selfTeamID: selfTeamID)
+            } catch {
+                WireLogger.sync.error(
+                    "team member discovery failed after cursor reset: \(String(describing: error))"
+                )
+            }
+        } catch {
+            WireLogger.sync.error("team member discovery failed: \(String(describing: error))")
+        }
+    }
+
+    private func runDiscovery(selfTeamID: UUID) async throws {
+        let sinceNotificationID = journal[.lastTeamNotificationID].flatMap(UUID.init(uuidString:))
+        let pager = try api.getNotifications(sinceNotificationID: sinceNotificationID)
+
+        var discoveredCount = 0
+        for try await notifications in pager {
+            let teamMembersInfo = notifications.map { notification -> TeamMemberInfo in
+                switch notification.kind {
+                case let .memberJoin(event):
+                    TeamMemberInfo(
+                        id: event.userID,
+                        selfPermission: nil,
+                        creatorID: nil,
+                        creationDate: event.time
+                    )
+                }
+            }
+
+            guard !teamMembersInfo.isEmpty else { continue }
+
+            try await store.storeTeamMembers(
+                selfTeamID: selfTeamID,
+                teamMembersInfo: teamMembersInfo
+            )
+            discoveredCount += teamMembersInfo.count
+
+            // Persist forward progress per page so an interruption
+            // doesn't force a full re-walk on the next run.
+            if let lastID = notifications.last?.id {
+                journal[.lastTeamNotificationID] = lastID.uuidString
+            }
+        }
+
+        WireLogger.sync.debug("team member discovery: stored \(discoveredCount) member(s)")
+    }
+
+}

WireDomain/Sources/WireDomain/Utilities/Journal/JournalKey.swift

@@ -127,6 +127,18 @@ public extension JournalKey where Value == Bool {
 
 }
 
+public extension JournalKey where Value == String? {
+
+    /// The id of the last team notification processed by
+    /// `TeamMemberDiscoveryAgent`, used as a cursor on subsequent runs.
+
+    static let lastTeamNotificationID = Self(
+        "lastTeamNotificationID",
+        defaultValue: nil
+    )
+
+}
+
 public extension JournalKey where Value == Set<String> {
 
     /// The set of MLS group IDs to be repaired.

WireDomain/Tests/WireDomainTests/Synchronization/IncrementalSyncTests.swift

@@ -20,8 +20,9 @@
 import CoreData
 import GenericMessageProtocol
 import XCTest
+
 @testable import WireDataModel
 @testable import WireDataModelSupport
 @testable import WireDomain
 @testable import WireDomainSupport
 @testable import WireNetwork
@@ -45,6 +46,7 @@
     var mlsGroupRepairAgent: MockMLSGroupRepairAgentProtocol!
     var cancellables: Set<AnyCancellable>!
     var earService: MockEARServiceInterface!
+    private var teamMemberDiscoveryAgentMock: MockTeamMemberDiscoveryAgentProtocol!
     fileprivate var notificationContext: MockNotificationContext!
 
     override func setUp() {
@@ -64,6 +66,8 @@
         mlsGroupRepairAgent = MockMLSGroupRepairAgentProtocol()
         cancellables = Set<AnyCancellable>()
         earService = MockEARServiceInterface()
+        teamMemberDiscoveryAgentMock = MockTeamMemberDiscoveryAgentProtocol()
+        teamMemberDiscoveryAgentMock.discoverMembers_MockMethod = {}
         notificationContext = MockNotificationContext()
 
         earService.underlyingIsLocked = false
@@ -83,12 +87,14 @@
             liveBrokenGroupSubject: liveBrokenGroupSubject,
             journal: journal,
             mlsGroupRepairAgent: mlsGroupRepairAgent,
-            earService: earService
+            earService: earService,
+            teamMemberDiscoveryAgent: teamMemberDiscoveryAgentMock
         )
     }
 
     override func tearDown() {
         sut = nil
+        teamMemberDiscoveryAgentMock = nil
         journal = nil
         pushChannelAPI = nil
         updateEventsSync = nil
@@ -309,7 +315,7 @@
         }.store(in: &cancellables)
 
         let token = try await sut.perform()
         await token.task.value
         wait(for: [expectation], timeout: 5)
 
         // Then live events were decrypted (duplicates skipped).
@@ -387,13 +393,13 @@
         // Repair broken MLS conversations
         mlsGroupRepairAgent.repairConversations_MockMethod = {}
 
         // When
         let task = Task {
             try await sut.perform()
         }
 
         // Then
         do {
             _ = try await task.value
         } catch {
             XCTAssertEqual(pushChannel.close_Invocations.count, 1)

WireDomain/Tests/WireDomainTests/Synchronization/IncrementalSyncV2Tests.swift

@@ -19,6 +19,7 @@
 import Combine
 import CoreData
 import XCTest
+
 @testable import WireDataModel
 @testable import WireDataModelSupport
 @testable import WireDomain
@@ -47,6 +48,7 @@
     var journal: Journal!
     var cancellables: Set<AnyCancellable>!
     var earService: MockEARServiceInterface!
+    private var teamMemberDiscoveryAgentMock: MockTeamMemberDiscoveryAgentProtocol!
 
     override func setUp() {
         pushChannelAPI = MockPushChannelV2API()
@@ -75,6 +77,8 @@
         liveBrokenGroupSubject = .init()
         cancellables = .init()
         earService = MockEARServiceInterface()
+        teamMemberDiscoveryAgentMock = MockTeamMemberDiscoveryAgentProtocol()
+        teamMemberDiscoveryAgentMock.discoverMembers_MockMethod = {}
 
         sut = IncrementalSyncV2(
             selfClientID: Scaffolding.selfClientID,
@@ -91,6 +95,7 @@
             journal: journal,
             mlsGroupRepairAgent: mlsGroupRepairAgent,
             earService: earService,
+            teamMemberDiscoveryAgent: teamMemberDiscoveryAgentMock,
             createPushChannelState: {
                 self.pushChannelState
             },
@@ -110,6 +115,7 @@
 
     override func tearDown() {
         sut = nil
+        teamMemberDiscoveryAgentMock = nil
         pushChannelAPI = nil
         mlsGroupRepairAgent = nil
         pullServerTimeSync = nil