Sync All Conversations Consent Filtering (#449)

* add consent sync filtering

* update async methods

* update async methods

* more fixes

* add missing async modifier

* bump podspec to 3.0.18

---------

Co-authored-by: cameronvoell <[email protected]>

Author: nplasterer

Package.resolved

@@ -21,7 +21,7 @@ let package = Package(
 		.package(url: "https://github.com/bufbuild/connect-swift", exact: "1.0.0"),
 		.package(url: "https://github.com/apple/swift-docc-plugin.git", from: "1.4.3"),
 		.package(url: "https://github.com/krzyzanowskim/CryptoSwift.git", exact: "1.8.3"),
-		.package(url: "https://github.com/xmtp/libxmtp-swift.git", exact: "3.0.12")
+		.package(url: "https://github.com/xmtp/libxmtp-swift.git", exact: "3.0.13")
 	],
 	targets: [
 		.target(

Package.swift

@@ -204,8 +204,9 @@ public final class Client {
 		let dbURL = directoryURL.appendingPathComponent(alias).path
 
 		let ffiClient = try await LibXMTP.createClient(
-			host: options.api.env.url,
-			isSecure: options.api.env.isSecure == true,
+			api: connectToBackend(
+				host: options.api.env.url,
+				isSecure: options.api.env.isSecure == true),
 			db: dbURL,
 			encryptionKey: options.dbEncryptionKey,
 			inboxId: inboxId,
@@ -297,8 +298,9 @@ public final class Client {
 		let dbURL = directoryURL.appendingPathComponent(alias).path
 
 		let ffiClient = try await LibXMTP.createClient(
-			host: api.env.url,
-			isSecure: api.env.isSecure == true,
+			api: connectToBackend(
+				host: api.env.url,
+				isSecure: api.env.isSecure == true),
 			db: dbURL,
 			encryptionKey: nil,
 			inboxId: inboxId,
@@ -448,18 +450,18 @@ public final class Client {
 		}
 	}
 
-	public func findConversation(conversationId: String) throws -> Conversation?
+	public func findConversation(conversationId: String) async throws -> Conversation?
 	{
 		do {
 			let conversation = try ffiClient.conversation(
 				conversationId: conversationId.hexToData)
-			return try conversation.toConversation(client: self)
+			return try await conversation.toConversation(client: self)
 		} catch {
 			return nil
 		}
 	}
 
-	public func findConversationByTopic(topic: String) throws -> Conversation? {
+	public func findConversationByTopic(topic: String) async throws -> Conversation? {
 		do {
 			let regexPattern = #"/xmtp/mls/1/g-(.*?)/proto"#
 			if let regex = try? NSRegularExpression(pattern: regexPattern) {
@@ -471,7 +473,7 @@ public final class Client {
 						with: match.range(at: 1))
 					let conversation = try ffiClient.conversation(
 						conversationId: conversationId.hexToData)
-					return try conversation.toConversation(client: self)
+					return try await conversation.toConversation(client: self)
 				}
 			}
 		} catch {

Sources/XMTPiOS/Client.swift

@@ -29,9 +29,9 @@ public enum Conversation: Identifiable, Equatable, Hashable {
 	public func isCreator() async throws -> Bool {
 		switch self {
 		case let .group(group):
-			return try group.isCreator()
+			return try await group.isCreator()
 		case let .dm(dm):
-			return try dm.isCreator()
+			return try await dm.isCreator()
 		}
 	}
 

Sources/XMTPiOS/Conversation.swift

@@ -79,9 +79,11 @@ public actor Conversations {
 	public func sync() async throws {
 		try await ffiConversations.sync()
 	}
-	public func syncAllConversations() async throws -> UInt32 {
-        // TODO: add consent state here
-        return try await ffiConversations.syncAllConversations(consentState: nil)
+	public func syncAllConversations(consentState: ConsentState? = nil)
+		async throws -> UInt32
+	{
+		return try await ffiConversations.syncAllConversations(
+			consentState: consentState?.toFFI)
 	}
 
 	public func listGroups(
@@ -105,7 +107,7 @@ public actor Conversations {
 		let conversations = try await ffiConversations.listGroups(
 			opts: options)
 
-		let sortedConversations = try sortConversations(
+		let sortedConversations = try await sortConversations(
 			conversations, order: order)
 
 		return sortedConversations.map {
@@ -134,7 +136,7 @@ public actor Conversations {
 		let conversations = try await ffiConversations.listDms(
 			opts: options)
 
-		let sortedConversations = try sortConversations(
+		let sortedConversations = try await sortConversations(
 			conversations, order: order)
 
 		return sortedConversations.map {
@@ -160,36 +162,42 @@ public actor Conversations {
 		if let limit {
 			options.limit = Int64(limit)
 		}
-		let conversations = try await ffiConversations.list(
+		let ffiConversations = try await ffiConversations.list(
 			opts: options)
 
-		let sortedConversations = try sortConversations(
-			conversations, order: order)
+		let sortedConversations = try await sortConversations(
+            ffiConversations, order: order)
+        
+        var conversations: [Conversation] = []
+        for sortedConversation in sortedConversations {
+            let conversation = try await sortedConversation.toConversation(client: client)
+            conversations.append(conversation)
+        }
 
-		return try sortedConversations.map {
-			try $0.toConversation(client: client)
-		}
+		return conversations
 	}
 
 	private func sortConversations(
 		_ conversations: [FfiConversation],
 		order: ConversationOrder
-	) throws -> [FfiConversation] {
+	) async throws -> [FfiConversation] {
 		switch order {
 		case .lastMessage:
-			let conversationWithTimestamp: [(FfiConversation, Int64?)] =
-				try conversations.map { conversation in
-					let message = try conversation.findMessages(
-						opts: FfiListMessagesOptions(
-							sentBeforeNs: nil,
-							sentAfterNs: nil,
-							limit: 1,
-							deliveryStatus: nil,
-							direction: .descending
-						)
-					).first
-					return (conversation, message?.sentAtNs)
-				}
+			var conversationWithTimestamp: [(FfiConversation, Int64?)] = []
+
+			for conversation in conversations {
+				let message = try await conversation.findMessages(
+					opts: FfiListMessagesOptions(
+						sentBeforeNs: nil,
+						sentAfterNs: nil,
+						limit: 1,
+						deliveryStatus: nil,
+						direction: .descending
+					)
+				).first
+				conversationWithTimestamp.append(
+					(conversation, message?.sentAtNs))
+			}
 
 			let sortedTuples = conversationWithTimestamp.sorted { (lhs, rhs) in
 				(lhs.1 ?? 0) > (rhs.1 ?? 0)
@@ -207,41 +215,45 @@ public actor Conversations {
 			let ffiStreamActor = FfiStreamActor()
 			let conversationCallback = ConversationStreamCallback {
 				conversation in
-				guard !Task.isCancelled else {
-					continuation.finish()
-					return
-				}
-				do {
-					let conversationType = try conversation.conversationType()
-					if conversationType == .dm {
-						continuation.yield(
-							Conversation.dm(
-								conversation.dmFromFFI(client: self.client))
-						)
-					} else if conversationType == .group {
-						continuation.yield(
-							Conversation.group(
-								conversation.groupFromFFI(client: self.client))
-						)
+				Task {
+					guard !Task.isCancelled else {
+						continuation.finish()
+						return
+					}
+					do {
+						let conversationType =
+							try await conversation.conversationType()
+						if conversationType == .dm {
+							continuation.yield(
+								Conversation.dm(
+									conversation.dmFromFFI(client: self.client))
+							)
+						} else if conversationType == .group {
+							continuation.yield(
+								Conversation.group(
+									conversation.groupFromFFI(
+										client: self.client))
+							)
+						}
+					} catch {
+						print("Error processing conversation type: \(error)")
 					}
-				} catch {
-					// Do nothing if the conversation type is neither a group or dm
 				}
 			}
 
 			let task = Task {
 				let stream: FfiStreamCloser
-					switch type {
-					case .groups:
-						stream = await ffiConversations.streamGroups(
-							callback: conversationCallback)
-					case .all:
-						stream = await ffiConversations.stream(
-							callback: conversationCallback)
-					case .dms:
-						stream = await ffiConversations.streamDms(
-							callback: conversationCallback)
-					}
+				switch type {
+				case .groups:
+					stream = await ffiConversations.streamGroups(
+						callback: conversationCallback)
+				case .all:
+					stream = await ffiConversations.stream(
+						callback: conversationCallback)
+				case .dms:
+					stream = await ffiConversations.streamDms(
+						callback: conversationCallback)
+				}
 				await ffiStreamActor.setFfiStream(stream)
 				continuation.onTermination = { @Sendable reason in
 					Task {
@@ -268,7 +280,9 @@ public actor Conversations {
 		if !canMessage {
 			throw ConversationError.memberNotRegistered([peerAddress])
 		}
-		if let existingDm = try await client.findDmByAddress(address: peerAddress) {
+		if let existingDm = try await client.findDmByAddress(
+			address: peerAddress)
+		{
 			return existingDm
 		}
 
@@ -385,20 +399,20 @@ public actor Conversations {
 
 			let task = Task {
 				let stream: FfiStreamCloser
-					switch type {
-					case .groups:
-						stream = await ffiConversations.streamAllGroupMessages(
-							messageCallback: messageCallback
-						)
-					case .dms:
-						stream = await ffiConversations.streamAllDmMessages(
-							messageCallback: messageCallback
-						)
-					case .all:
-						stream = await ffiConversations.streamAllMessages(
-							messageCallback: messageCallback
-						)
-					}
+				switch type {
+				case .groups:
+					stream = await ffiConversations.streamAllGroupMessages(
+						messageCallback: messageCallback
+					)
+				case .dms:
+					stream = await ffiConversations.streamAllDmMessages(
+						messageCallback: messageCallback
+					)
+				case .all:
+					stream = await ffiConversations.streamAllMessages(
+						messageCallback: messageCallback
+					)
+				}
 				await ffiStreamActor.setFfiStream(stream)
 			}
 
@@ -417,7 +431,7 @@ public actor Conversations {
 		let conversation =
 			try await ffiConversations
 			.processStreamedWelcomeMessage(envelopeBytes: envelopeBytes)
-		return try conversation.toConversation(client: client)
+		return try await conversation.toConversation(client: client)
 	}
 
 	public func newConversation(

Sources/XMTPiOS/Conversations.swift

@@ -14,8 +14,8 @@ public struct Dm: Identifiable, Equatable, Hashable {
 		Topic.groupMessage(id).description
 	}
 
-	func metadata() throws -> FfiConversationMetadata {
-		return try ffiConversation.groupMetadata()
+	func metadata() async throws -> FfiConversationMetadata {
+		return try await ffiConversation.groupMetadata()
 	}
 
 	public func sync() async throws {
@@ -30,12 +30,12 @@ public struct Dm: Identifiable, Equatable, Hashable {
 		id.hash(into: &hasher)
 	}
 
-	public func isCreator() throws -> Bool {
-		return try metadata().creatorInboxId() == client.inboxID
+	public func isCreator() async throws -> Bool {
+		return try await metadata().creatorInboxId() == client.inboxID
 	}
 
-	public func creatorInboxId() throws -> String {
-		return try metadata().creatorInboxId()
+	public func creatorInboxId() async throws -> String {
+		return try await metadata().creatorInboxId()
 	}
 
 	public func addedByInboxId() throws -> String {
@@ -253,7 +253,7 @@ public struct Dm: Identifiable, Equatable, Hashable {
 
 		options.direction = direction
 
-		return try ffiConversation.findMessages(opts: options).compactMap {
+		return try await ffiConversation.findMessages(opts: options).compactMap {
 			ffiMessage in
 			return Message(client: self.client, ffiMessage: ffiMessage)
 				.decodeOrNull()

Sources/XMTPiOS/Dm.swift

@@ -10,8 +10,8 @@ extension FfiConversation {
 		Dm(ffiConversation: self, client: client)
 	}
 
-	func toConversation(client: Client) throws -> Conversation {
-		if try conversationType() == .dm {
+	func toConversation(client: Client) async throws -> Conversation {
+		if try await conversationType() == .dm {
 			return Conversation.dm(self.dmFromFFI(client: client))
 		} else {
 			return Conversation.group(self.groupFromFFI(client: client))