fix: schedule commit pending proposals - WPB-22062

WireDomain/Sources/WireDomain/Components/ClientSessionComponent.swift

@@ -807,8 +807,8 @@
     )
 
     public lazy var workAgent: WorkAgent = .init(scheduler: PriorityOrderWorkItemScheduler())
 
-    public lazy var conversationUpdatesGenerator: ConversationUpdatesGeneratorProtocol = ConversationUpdatesGenerator(
+    public lazy var conversationUpdatesGenerator: IncrementalGeneratorProtocol = ConversationUpdatesGenerator(
         repository: conversationRepository,
         context: syncContext,
         onConversationUpdated: { [weak self] workItem in
@@ -816,5 +816,30 @@
             self?.workAgent.submitItem(workItem)
         }
     )
 
+    public lazy var commitPendingProposalsGenerator: LiveGeneratorProtocol = CommitPendingProposalsGenerator(
+        repository: conversationRepository,
+        mlsService: mlsService,
+        context: syncContext,
+        isMLSGroupBroken: { [weak self] groupID in
+            self?.isMLSGroupBroken(groupID: groupID) == true
+        },
+        onCommitPendingProposals: { [weak self] workItem in
+
+            self?.workAgent.submitItem(workItem)
+        }
+    )
+
+    public lazy var generatorsDirectory = GeneratorsDirectory(
+        generators: [
+            conversationUpdatesGenerator,
+            commitPendingProposalsGenerator
+        ],
+        syncStatePublisher: syncStateSubject.eraseToAnyPublisher()
+    )
+
+    private func isMLSGroupBroken(groupID: MLSGroupID) -> Bool {
+        let brokenGroupIds = journal[.brokenMLSGroupIDs]
+        return brokenGroupIds.contains(groupID.description)
+    }
 }

WireDomain/Sources/WireDomain/Event Decryption/UpdateEventDecryptor.swift

@@ -80,7 +80,6 @@ struct UpdateEventDecryptor: UpdateEventDecryptorProtocol {
 
         var decryptedEvents = [UpdateEvent]()
         var brokenMLSGroupIDs = Set<String>()
-        var shouldCommitPendingProposals = false
 
         for event in eventEnvelope.events {
             logAttributes[.messageType] = event.name
@@ -120,7 +119,6 @@ struct UpdateEventDecryptor: UpdateEventDecryptorProtocol {
                     "decrypting MLS add message event...",
                     attributes: logAttributes
                 )
-                shouldCommitPendingProposals = true
 
                 do {
                     let decryptedEventData = try await mlsMessageDecryptor.decryptedMessageAddEventData(
@@ -170,23 +168,9 @@ struct UpdateEventDecryptor: UpdateEventDecryptorProtocol {
             }
         }
 
-        if shouldCommitPendingProposals {
-            Task.detached {
-                // we don't need to wait for this, as it can take a while to finish
-                // it should not block decryption
-                await commitPendingProposalsIfNeeded()
-            }
-        }
-
         return EventDecryptorResult(events: decryptedEvents, brokenMLSGroupIDs: brokenMLSGroupIDs)
     }
 
-    private func commitPendingProposalsIfNeeded() async {
-        // MLSService will be nil when called from push notification service.
-        // As we don't need to commit pending proposals in that case.
-        await mlsService?.commitPendingProposalsIfNeeded()
-    }
-
     private func appendFailedToDecryptProteusMessage(
         eventData: ConversationProteusMessageAddEvent,
         error: ProteusError

WireDomain/Sources/WireDomain/Repositories/Conversations/LocalStore/ConversationLocalStore.swift

@@ -57,7 +57,7 @@
     // MARK: - Public
 
     public func qualifiedID(for conversation: ZMConversation) async -> QualifiedID? {
         await context.perform {
             conversation.qualifiedID
         }
     }
@@ -66,7 +66,7 @@
         _ lastReadMessage: LastRead,
         in conversation: ZMConversation
     ) async {
         await context.perform { [context] in
             guard conversation.isSelfConversation else {
                 return
             }
@@ -82,7 +82,7 @@
         _ clearedMessage: Cleared,
         in conversation: ZMConversation
     ) async {
         await context.perform { [context] in
             guard conversation.isSelfConversation else {
                 return
             }
@@ -98,7 +98,7 @@
         conversation: ZMConversation
     ) async -> (mlsGroupID: MLSGroupID, isMLSReady: Bool)? {
 
         await context.perform {
             guard let mlsGroupID = conversation.mlsGroupID else {
                 return nil
             }
@@ -112,7 +112,7 @@
         mlsGroupID: MLSGroupID,
         conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.mlsStatus = .ready
             conversation.mlsGroupID = mlsGroupID
         }
@@ -122,7 +122,7 @@
         newMLSGroupID: MLSGroupID,
         conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.mlsStatus = .pendingJoinAfterReset
             conversation.mlsGroupID = newMLSGroupID
         }
@@ -145,7 +145,7 @@
     public func fetchOtherUserIDInOneOnOneConversation(
         conversation: ZMConversation
     ) async -> WireDataModel.QualifiedID? {
         await context.perform {
             guard conversation.conversationType == .oneOnOne else {
                 WireLogger.conversation.info(
                     "conversation type is not expected 'oneOnOne', aborting."
@@ -156,7 +156,7 @@
 
             guard
                 let otherUser = conversation.localParticipantsExcludingSelf.first,
                 let otherUserID = otherUser.remoteIdentifier,
                 let otherUserDomain = otherUser.domain ?? self.localDomain
             else {
                 WireLogger.conversation.warn(
@@ -182,7 +182,7 @@
             let mutedStatus = mutedStatusInfo.status
             let mutedReference = mutedStatusInfo.referenceDate
 
             if let mutedStatus, let mutedReference {
                 localConversation.updateMutedStatus(
                     status: Int32(mutedStatus),
                     referenceDate: mutedReference
@@ -236,9 +236,9 @@
 
         await context.perform {
             // If user is already part of the conversation, its role will be updated.
             // If not, user will be added to the conversation.
             conversation.addParticipantAndUpdateConversationState(
                 user: user,
                 role: role
             )
         }
@@ -247,7 +247,7 @@
     public func increaseUnreadCount(
         for conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.internalEstimatedUnreadCount += 1
         }
     }
@@ -255,7 +255,7 @@
     public func decreaseUnreadCount(
         for conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.internalEstimatedUnreadCount -= 1
         }
     }
@@ -263,7 +263,7 @@
     public func increaseUnreadSelfMentionCount(
         for conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.internalEstimatedUnreadSelfMentionCount += 1
         }
     }
@@ -271,7 +271,7 @@
     public func increaseUnreadSelfReplyCount(
         for conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.internalEstimatedUnreadSelfReplyCount += 1
         }
     }
@@ -286,7 +286,7 @@
         permission: Conversation.ChannelPermission,
         conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.privateChannelPermission = PrivateChannelPermission(permission)
         }
     }
@@ -301,7 +301,7 @@
             domain: conversationDomain
         )
 
         await context.perform {
             conversation?.channelHistoryDepth = historyDepth
         }
     }
@@ -333,9 +333,9 @@
                 ]
             )
         }
 
         let usersAndRoles = await withTaskGroup(of: UserAndRole?.self) { taskGroup in
             for newParticipant in participants {
                 taskGroup.addTask { [self] in
                     let user = await context.perform { [context] in
                         ZMUser.fetchOrCreate(
@@ -359,7 +359,7 @@
             }
 
             var usersAndRoles: [UserAndRole?] = []
 
             for await userAndRole in taskGroup {
                 usersAndRoles.append(userAndRole)
             }
@@ -388,8 +388,8 @@
             )
         }
 
         await context.perform {
             localConversation.addParticipantsAndUpdateConversationState(
                 usersAndRoles: usersAndRoles
             )
         }
@@ -399,7 +399,7 @@
         user: ZMUser,
         conversation: ZMConversation
     ) async {
         await context.perform { [context] in
             if user.objectID.isTemporaryID || conversation.objectID.isTemporaryID {
                 do {
                     try context.obtainPermanentIDs(for: [user, conversation])
@@ -417,7 +417,7 @@
         to conversation: ZMConversation
     ) async {
         await context.perform { [context] in
             let systemMessage = ZMSystemMessage(nonce: UUID(), managedObjectContext: context)
             systemMessage.systemMessageType = message.type
             systemMessage.sender = message.sender
             systemMessage.users = message.users ?? Set()
@@ -436,7 +436,7 @@
             systemMessage.relevantForConversationStatus = message.relevantForStatus
             systemMessage.participantsRemovedReason = message.removedReason
             systemMessage.domains = message.domains
 
             conversation.append(systemMessage)
         }
     }
@@ -534,7 +534,7 @@
             domain: conversationDomain
         )
 
         await context.perform {
             conversation.isPendingMetadataRefresh = true
             conversation.needsToBeUpdatedFromBackend = true
         }
@@ -575,7 +575,7 @@
     public func conversationNeedsBackendUpdate(
         _ conversation: ZMConversation
     ) async -> Bool {
         await context.perform {
             conversation.needsToBeUpdatedFromBackend
         }
     }
@@ -583,7 +583,7 @@
     public func isConversationArchived(
         _ conversation: ZMConversation
     ) async -> Bool {
         await context.perform {
             conversation.isArchived
         }
     }
@@ -591,7 +591,7 @@
     public func isConversationForcedReadOnly(
         _ conversation: ZMConversation
     ) async -> Bool {
         await context.perform {
             conversation.isForcedReadOnly
         }
     }
@@ -601,7 +601,7 @@
         senderID: UUID?,
         conversation: ZMConversation
     ) async -> Bool {
         await context.perform {
             conversation.isMessageSilenced(message, senderID: senderID)
         }
     }
@@ -624,7 +624,7 @@
     ) async {
         let scheduledDate = date + TimeInterval(commitDelay)
 
         await context.perform {
             conversation.commitPendingProposalDate = scheduledDate
         }
     }
@@ -635,7 +635,7 @@
         date: Date
     ) async {
         // Update the legal hold state in the conversation
         await context.perform {
             conversation.updateSecurityLevelIfNeededAfterReceiving(
                 message: genericMessage,
                 timestamp: date
@@ -660,8 +660,8 @@
             return
         }
 
         await context.perform {
             conversation.addParticipantAndSystemMessageIfMissing(
                 participant,
                 date: date
             )
@@ -671,7 +671,7 @@
     public func conversationMutedMessageTypes(
         _ conversation: ZMConversation
     ) async -> MutedMessageTypes {
         await context.perform {
             conversation.mutedMessageTypes
         }
     }
@@ -680,7 +680,7 @@
         _ conversation: ZMConversation
     ) async -> MutedMessageTypes {
         await context.perform { [context] in
             let selfUser = ZMUser.selfUser(in: context)
             return selfUser.mutedMessagesTypes.union(conversation.mutedMessageTypes)
         }
     }
@@ -688,7 +688,7 @@
     public func lastReadServerTimestamp(
         _ conversation: ZMConversation
     ) async -> Date? {
         await context.perform {
             conversation.lastReadServerTimeStamp
         }
     }
@@ -696,7 +696,7 @@
     public func conversationMessageDestructionTimeout(
         _ conversation: ZMConversation
     ) async -> MessageDestructionTimeoutValue {
         await context.perform {
             conversation.activeMessageDestructionTimeoutValue ?? .init(rawValue: 0)
         }
     }
@@ -705,7 +705,7 @@
         timeoutValue: Double,
         for conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.setMessageDestructionTimeoutValue(
                 .init(rawValue: timeoutValue),
                 for: .groupConversation
@@ -717,7 +717,7 @@
         hasReadReceiptsEnabled: Bool,
         for conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.hasReadReceiptsEnabled = hasReadReceiptsEnabled
         }
     }
@@ -725,7 +725,7 @@
     public func messageProtocol(
         for conversation: ZMConversation
     ) async -> WireDataModel.MessageProtocol {
         await context.perform {
             conversation.messageProtocol
         }
     }
@@ -733,13 +733,13 @@
     public func isGroupConversation(
         _ conversation: ZMConversation
     ) async -> Bool {
         await context.perform {
             conversation.conversationType == .group
         }
     }
 
     public func isSelfConversation(_ conversation: ZMConversation) async -> Bool {
         await context.perform {
             conversation.conversationType == .self
         }
     }
@@ -747,7 +747,7 @@
     public func name(
         for conversation: ZMConversation
     ) async -> String? {
         await context.perform {
             conversation.displayName
         }
     }
@@ -767,7 +767,7 @@
         }
 
         let allGroupConversations = await context.perform {
             // swiftformat:disable:next redundantProperty
             let allGroupConversations: [ZMConversation] = user.participantRoles.compactMap {
                 guard $0.conversation?.conversationType == .group else {
                     return nil
@@ -781,9 +781,9 @@
 
         for conversation in allGroupConversations {
             let (userTeam, isTeamMember, conversationTeam, conversationID, conversationDomain) = await context.perform {
                 (
                     user.team,
                     user.isTeamMember,
                     conversation.team,
                     conversation.remoteIdentifier as UUID,
                     conversation.domain
@@ -818,8 +818,8 @@
                 )
             }
 
             await context.perform {
                 conversation.removeParticipantAndUpdateConversationState(
                     user: user,
                     initiatingUser: user
                 )
@@ -833,7 +833,7 @@
     ) async -> Role {
         await context.perform { [context] in
             Role.fetchOrCreateRole(
                 with: role,
                 teamOrConversation: TeamOrConversation.matching(conversation),
                 in: context
             )
@@ -845,7 +845,7 @@
         accessModes: [String],
         accessRoles: [String]
     ) async {
         await context.perform { [context] in
             conversation.accessModeStrings = accessModes
             conversation.accessRoleStringsV2 = accessRoles
 
@@ -868,7 +868,7 @@
         isDeletedRemotely: Bool,
         conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.isDeletedRemotely = isDeletedRemotely
         }
     }
@@ -878,9 +878,9 @@
         users: Set<ZMUser>,
         initiatingUser: ZMUser
     ) async {
         await context.perform {
             conversation.removeParticipantsAndUpdateConversationState(
                 users: users,
                 initiatingUser: initiatingUser
             )
         }
@@ -889,7 +889,7 @@
     public func localParticipants(
         in conversation: ZMConversation
     ) async -> Set<ZMUser> {
         await context.perform {
             conversation.localParticipants
         }
     }
@@ -897,7 +897,7 @@
     public func conversationName(
         conversation: ZMConversation
     ) async -> String? {
         await context.perform {
             conversation.userDefinedName
         }
     }
@@ -906,7 +906,7 @@
         newName: String,
         conversation: ZMConversation
     ) async {
         await context.perform {
             conversation.userDefinedName = newName
         }
     }
@@ -946,10 +946,10 @@
             domain: conversation.qualifiedID?.domain
         )
 
         await context.perform { [self] in
             localConversation.conversationType = .connection
 
             commonUpdate(
                 from: conversation,
                 for: localConversation,
                 serverTimestamp: serverTimestamp,
@@ -1003,15 +1003,15 @@
             domain: conversation.qualifiedID?.domain
         )
 
         let mlsGroupID = await context.perform {
             conversation.mlsGroupID
         }
 
         await context.perform { [self] in
             localConversation.conversationType = .`self`
             localConversation.isPendingMetadataRefresh = false
 
             commonUpdate(
                 from: conversation,
                 for: localConversation,
                 serverTimestamp: serverTimestamp,
@@ -1060,13 +1060,13 @@
             domain: conversation.qualifiedID?.domain
         )
 
         await context.perform { [self] in
             isInitialFetch = localConversation.isPendingInitialFetch
 
             localConversation.conversationType = .group
             localConversation.remoteIdentifier = id
             localConversation.isPendingMetadataRefresh = false
             localConversation.isPendingInitialFetch = false
             localConversation.groupType = conversation.groupType.map { groupType in
                 switch groupType {
                 case .group:
@@ -1089,7 +1089,7 @@
                     .disabled
                 }
             } ?? .disabled
 
             commonUpdate(
                 from: conversation,
                 for: localConversation,
@@ -1102,7 +1102,7 @@
                 from: conversation,
                 for: localConversation
             )
 
             if isInitialFetch {
                 assignMessageProtocol(
                     from: conversation,
@@ -1137,9 +1137,9 @@
             isMLSEnabled: isMLSEnabled
         )
 
         await context.perform { [self] in
             if isInitialFetch {
                 // we just got a new conversation, we display new conversation header
                 localConversation.appendNewConversationSystemMessage(
                     at: .distantPast,
                     users: localConversation.localParticipants
@@ -1154,7 +1154,7 @@
             }
 
             // If we discover this group is actually a fake one on one,
             // then we should link the one on one user.
             linkOneOnOneUserIfNeeded(for: localConversation)
 
             // All metadata has been updated, object does not need to be updated from backend
@@ -1189,14 +1189,14 @@
             let conversationType = BackendConversationType.clientConversationType(
                 rawValue: conversationTypeRawValue
             )
 
             if localConversation.oneOnOneUser?.connection?.status == .sent {
                 localConversation.conversationType = .connection
             } else {
                 localConversation.conversationType = conversationType
             }
 
             assignMessageProtocol(
                 from: conversation,
                 for: localConversation
             )
@@ -1297,11 +1297,20 @@
             domain: conversationDomain
         )
 
         return await context.perform {
             handler(conversation)
         }
     }
 
+    public func execute(
+        identifier: MLSGroupID,
+        block: @escaping @Sendable (ZMConversation?, NSManagedObjectContext) -> Void
+    ) async {
+        await context.perform { [context] in
+            let conversation = ZMConversation.fetch(with: identifier, in: context)
+            block(conversation, context)
+        }
+    }
 }
 
 // MARK: - Private helpers

WireDomain/Sources/WireDomain/Repositories/Conversations/Protocols/ConversationLocalStoreProtocol.swift

@@ -469,4 +469,8 @@ public protocol ConversationLocalStoreProtocol {
 
     func fetchServerTimeDelta() async -> TimeInterval
 
+    func execute(
+        identifier: MLSGroupID,
+        block: @escaping @Sendable (ZMConversation?, NSManagedObjectContext) -> Void
+    ) async
 }

WireDomain/Sources/WireDomain/Repositories/Conversations/Protocols/ConversationRepositoryProtocol.swift

@@ -192,4 +192,10 @@ public protocol ConversationRepositoryProtocol: Sendable {
         conversationID: String
     ) async throws -> String?
 
+    /// Checks if selfUser is still in a given conversation
+    /// - Parameter groupID: mlsGroupID of the conversation
+    /// - Returns: true if selfUser belongs to the conversation, false otherwise
+    func isSelfAnActiveMember(
+        in groupID: WireDataModel.MLSGroupID
+    ) async -> Bool
 }

WireDomain/Sources/WireDomain/Repositories/Conversations/Repository/ConversationRepository.swift

@@ -18,21 +18,21 @@
 
 import Foundation
 import WireDataModel
 import WireLogging
 import WireNetwork
 
 public final class ConversationRepository: ConversationRepositoryProtocol {
 
     // MARK: - Properties
 
     private let conversationsAPI: any ConversationsAPI
     private let conversationsLocalStore: any ConversationLocalStoreProtocol
     private let userLocalStore: any UserLocalStoreProtocol
     private let teamRepository: any TeamRepositoryProtocol
     private let messageRepository: any MessageRepositoryProtocol
     private let localDomain: String
     private let isFederationEnabled: Bool
     private let isMLSEnabled: Bool
     private let mlsProvider: MLSProvider
 
     // MARK: - Object lifecycle
@@ -377,6 +377,16 @@
         await deleteMembership(for: removedUserIDs, time: date)
     }
 
+    public func isSelfAnActiveMember(
+        in groupID: MLSGroupID
+    ) async -> Bool {
+        nonisolated(unsafe) var isSelfAnActiveMember = false
+        await conversationsLocalStore.execute(identifier: groupID) { conversation, _ in
+            isSelfAnActiveMember = conversation?.isSelfAnActiveMember ?? false
+        }
+        return isSelfAnActiveMember
+    }
+
     // MARK: - Private
 
     private func addSystemMessage(
@@ -411,9 +421,9 @@
 
     private func getRemovedUsers(
         from userIDs: Set<UserID>
     ) async -> [WireDataModel.ZMUser] {
         await withTaskGroup(of: WireDataModel.ZMUser.self) { taskGroup in
             for userID in userIDs {
                 taskGroup.addTask { [self] in
                     await userLocalStore.fetchOrCreateUser(
                         id: userID.id,
@@ -423,7 +433,7 @@
             }
 
             var users: [WireDataModel.ZMUser] = []
 
             for await user in taskGroup {
                 users.append(user)
             }

WireDomain/Sources/WireDomain/Synchronization/CommitPendingProsalsGenerator.swift

@@ -0,0 +1,163 @@
+//
+// Wire
+// Copyright (C) 2025 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 WireDataModel
+import WireLogging
+
+public final class CommitPendingProposalsGenerator: NSObject, LiveGeneratorProtocol {
+
+    private let context: NSManagedObjectContext
+    private let fetchedResultsController: NSFetchedResultsController<ZMConversation>
+    private let repository: ConversationRepositoryProtocol
+    private let mlsService: MLSServiceInterface
+    private let isMLSGroupBroken: (MLSGroupID) -> Bool
+    private var onCommitPendingProposals: (CommitPendingProposalItem) -> Void
+
+    init(
+        repository: ConversationRepositoryProtocol,
+        mlsService: MLSServiceInterface,
+        context: NSManagedObjectContext,
+        isMLSGroupBroken: @escaping (MLSGroupID) -> Bool,
+        onCommitPendingProposals: @escaping (CommitPendingProposalItem) -> Void
+    ) {
+        let request = NSFetchRequest<ZMConversation>(entityName: ZMConversation.entityName())
+        request.predicate = ZMConversation.commitPendingProposalDatePredicate()
+        request.sortDescriptors = [ZMConversation.sortCommitPendingProsalsByDateAscending()]
+        self.fetchedResultsController = NSFetchedResultsController(
+            fetchRequest: request,
+            managedObjectContext: context,
+            sectionNameKeyPath: nil,
+            cacheName: nil
+        )
+        self.context = context
+        self.onCommitPendingProposals = onCommitPendingProposals
+        self.repository = repository
+        self.mlsService = mlsService
+        self.isMLSGroupBroken = isMLSGroupBroken
+        super.init()
+    }
+
+    /// Starts monitoring and triggers work for any conversations with commitPendingProposalDate not nil.
+    public func start() async {
+        fetchedResultsController.delegate = self
+        do {
+            try fetchedResultsController.performFetch()
+        } catch {
+            WireLogger.conversation
+                .error("error fetching conversations with pending commit proposals: \(String(describing: error))")
+        }
+
+        let conversations = fetchedResultsController.fetchedObjects ?? []
+        for conversation in conversations {
+            await context.perform {
+                self.commitPendingProposalItem(for: conversation)
+            }
+        }
+    }
+
+    private func commitPendingProposalItem(for conversation: ZMConversation) {
+        if let id = conversation.qualifiedID,
+           let timestamp = conversation.commitPendingProposalDate,
+           let mlsGroupID = conversation.mlsGroupID,
+           conversation.isSelfAnActiveMember,
+           !isMLSGroupBroken(mlsGroupID) {
+
+            Task {
+                await generateItemForSubconversation(
+                    parentID: mlsGroupID,
+                    timestamp: timestamp,
+                    conversationID: id
+                )
+            }
+
+            WireLogger.workAgent.debug(
+                "generate commit pending proposal work-item",
+                attributes: [.mlsGroupID: mlsGroupID.safeForLoggingDescription]
+            )
+            onCommitPendingProposals(
+                CommitPendingProposalItem(
+                    repository: repository,
+                    conversationID: id,
+                    groupID: mlsGroupID,
+                    timestamp: timestamp,
+                    mlsService: mlsService
+                )
+            )
+        }
+    }
+
+    func generateItemForSubconversation(
+        parentID: MLSGroupID,
+        timestamp: Date,
+        conversationID: QualifiedID
+    ) async {
+
+        if let subgroupID = await mlsService.subConferenceConversation(
+            parentGroupID: parentID
+        ) {
+            WireLogger.workAgent.debug(
+                "generate subconversation commit pending proposal work-item for \(parentID.safeForLoggingDescription)",
+                attributes: [.mlsGroupID: subgroupID.safeForLoggingDescription]
+            )
+            onCommitPendingProposals(
+                CommitPendingProposalItem(
+                    repository: repository,
+                    conversationID: conversationID,
+                    groupID: subgroupID,
+                    timestamp: timestamp,
+                    mlsService: mlsService
+                )
+            )
+        }
+    }
+
+    public func stop() {
+        fetchedResultsController.delegate = nil
+    }
+}
+
+// MARK: - NSFetchedResultsControllerDelegate
+
+extension CommitPendingProposalsGenerator: NSFetchedResultsControllerDelegate {
+
+    public func controller(
+        _ controller: NSFetchedResultsController<NSFetchRequestResult>,
+        didChange anObject: Any,
+        at indexPath: IndexPath?,
+        for type: NSFetchedResultsChangeType,
+        newIndexPath: IndexPath?
+    ) {
+        guard let conversation = anObject as? ZMConversation else {
+            fatal("unexpected object, expected ZMConversation")
+        }
+
+        switch type {
+        case .insert:
+            commitPendingProposalItem(for: conversation)
+
+        case .update:
+            break
+
+        case .move, .delete:
+            break
+
+        @unknown default:
+            break
+        }
+    }
+}

WireDomain/Sources/WireDomain/Synchronization/ConversationUpdatesGenerator.swift

@@ -19,13 +19,7 @@
 import WireDataModel
 import WireLogging
 
-/// sourcery: AutoMockable
-public protocol ConversationUpdatesGeneratorProtocol {
-    func start() async
-    func stop()
-}
-
-public final class ConversationUpdatesGenerator: NSObject, ConversationUpdatesGeneratorProtocol {
+public final class ConversationUpdatesGenerator: NSObject, IncrementalGeneratorProtocol {
 
     private let context: NSManagedObjectContext
     private var fetchedResultsController: NSFetchedResultsController<ZMConversation>?
@@ -58,8 +52,8 @@
 
         let conversations = fetchedResultsController?.fetchedObjects ?? []
         for conversation in conversations {
             await context.perform {
                 if let id = conversation.qualifiedID {
                     self.onConversationUpdated(UpdateConversationItem(
                         repository: self.repository,
                         conversationID: id.toAPIModel()