Consent syncing

Package.swift

@@ -25,7 +25,7 @@ let package = Package(
 		.package(url: "https://github.com/1024jp/GzipSwift", from: "5.2.0"),
 		.package(url: "https://github.com/bufbuild/connect-swift", exact: "0.12.0"),
 		.package(url: "https://github.com/apple/swift-docc-plugin.git", from: "1.0.0"),
-		.package(url: "https://github.com/xmtp/libxmtp-swift.git", exact: "0.6.0"),
+		.package(url: "https://github.com/xmtp/libxmtp-swift.git", exact: "3.0.0"),
 	],
 	targets: [
 		// Targets are the basic building blocks of a package. A target can define a module or a test suite.

Sources/XMTPiOS/Client.swift

@@ -140,7 +140,7 @@ public final class Client {
 	{
 		let accountAddress = account.address.lowercased()
 		let inboxId = try await getOrCreateInboxId(
-			options: options, address: accountAddress)
+			api: options.api, address: accountAddress)
 
 		return try await initializeClient(
 			accountAddress: accountAddress,
@@ -155,7 +155,7 @@ public final class Client {
 	{
 		let accountAddress = address.lowercased()
 		let inboxId = try await getOrCreateInboxId(
-			options: options, address: accountAddress)
+			api: options.api, address: accountAddress)
 
 		return try await initializeClient(
 			accountAddress: accountAddress,
@@ -256,19 +256,19 @@ public final class Client {
 	}
 
 	public static func getOrCreateInboxId(
-		options: ClientOptions, address: String
+		api: ClientOptions.Api, address: String
 	) async throws -> String {
 		var inboxId: String
 		do {
 			inboxId =
 				try await getInboxIdForAddress(
 					logger: XMTPLogger(),
-					host: options.api.env.url,
-					isSecure: options.api.env.isSecure == true,
+					host: api.env.url,
+					isSecure: api.env.isSecure == true,
 					accountAddress: address
 				) ?? generateInboxId(accountAddress: address, nonce: 0)
 		} catch {
-			inboxId = generateInboxId(accountAddress: address, nonce: 0)
+			inboxId = try generateInboxId(accountAddress: address, nonce: 0)
 		}
 		return inboxId
 	}
@@ -387,7 +387,11 @@ public final class Client {
 	}
 
 	public func requestMessageHistorySync() async throws {
-		try await ffiClient.requestHistorySync()
+		try await ffiClient.sendSyncRequest(kind: .messages)
+	}
+
+	public func syncConsent() async throws {
+		try await ffiClient.sendSyncRequest(kind: .consent)
 	}
 
 	public func revokeAllOtherInstallations(signingKey: SigningKey) async throws

Sources/XMTPiOS/Conversation.swift

@@ -4,7 +4,7 @@ import LibXMTP
 public enum Conversation: Identifiable, Equatable, Hashable {
 	case group(Group)
 	case dm(Dm)
-	
+
 	public static func == (lhs: Conversation, rhs: Conversation) -> Bool {
 		lhs.topic == rhs.topic
 	}
@@ -44,7 +44,7 @@ public enum Conversation: Identifiable, Equatable, Hashable {
 		}
 	}
 
-	public func consentState() async throws -> ConsentState {
+	public func consentState() throws -> ConsentState {
 		switch self {
 		case let .group(group):
 			return try group.consentState()
@@ -92,6 +92,15 @@ public enum Conversation: Identifiable, Equatable, Hashable {
 				content: content, options: options)
 		}
 	}
+
+	public func publishMessages() async throws {
+		switch self {
+		case let .group(group):
+			return try await group.publishMessages()
+		case let .dm(dm):
+			return try await dm.publishMessages()
+		}
+	}
 
 	public var type: ConversationType {
 		switch self {
@@ -168,19 +177,21 @@ public enum Conversation: Identifiable, Equatable, Hashable {
 	}
 
 	public func messages(
-		limit: Int? = nil, before: Date? = nil, after: Date? = nil,
+		limit: Int? = nil,
+		beforeNs: Int64? = nil,
+		afterNs: Int64? = nil,
 		direction: SortDirection? = .descending,
 		deliveryStatus: MessageDeliveryStatus = .all
 	) async throws -> [DecodedMessage] {
 		switch self {
 		case let .group(group):
 			return try await group.messages(
-				before: before, after: after, limit: limit,
+				beforeNs: beforeNs, afterNs: afterNs, limit: limit,
 				direction: direction, deliveryStatus: deliveryStatus
 			)
 		case let .dm(dm):
 			return try await dm.messages(
-				before: before, after: after, limit: limit,
+				beforeNs: beforeNs, afterNs: afterNs, limit: limit,
 				direction: direction, deliveryStatus: deliveryStatus
 			)
 		}

Sources/XMTPiOS/Conversations.swift

@@ -34,6 +34,10 @@ public enum ConversationOrder {
 	case createdAt, lastMessage
 }
 
+public enum ConversationType {
+	case all, groups, dms
+}
+
 final class ConversationStreamCallback: FfiConversationCallback {
 	func onError(error: LibXMTP.FfiSubscribeError) {
 		print("Error ConversationStreamCallback \(error)")
@@ -80,11 +84,13 @@ public actor Conversations {
 	}
 
 	public func listGroups(
-		createdAfter: Date? = nil, createdBefore: Date? = nil, limit: Int? = nil
+		createdAfter: Date? = nil, createdBefore: Date? = nil,
+		limit: Int? = nil, order: ConversationOrder = .createdAt,
+		consentState: ConsentState? = nil
 	) async throws -> [Group] {
 		var options = FfiListConversationsOptions(
 			createdAfterNs: nil, createdBeforeNs: nil, limit: nil,
-			consentState: nil)
+			consentState: consentState?.toFFI)
 		if let createdAfter {
 			options.createdAfterNs = Int64(createdAfter.millisecondsSinceEpoch)
 		}
@@ -95,17 +101,25 @@ public actor Conversations {
 		if let limit {
 			options.limit = Int64(limit)
 		}
-		return try await ffiConversations.listGroups(opts: options).map {
+		let conversations = try await ffiConversations.listGroups(
+			opts: options)
+
+		let sortedConversations = try sortConversations(
+			conversations, order: order)
+
+		return sortedConversations.map {
 			$0.groupFromFFI(client: client)
 		}
 	}
 
 	public func listDms(
-		createdAfter: Date? = nil, createdBefore: Date? = nil, limit: Int? = nil
+		createdAfter: Date? = nil, createdBefore: Date? = nil,
+		limit: Int? = nil, order: ConversationOrder = .createdAt,
+		consentState: ConsentState? = nil
 	) async throws -> [Dm] {
 		var options = FfiListConversationsOptions(
 			createdAfterNs: nil, createdBeforeNs: nil, limit: nil,
-			consentState: nil)
+			consentState: consentState?.toFFI)
 		if let createdAfter {
 			options.createdAfterNs = Int64(createdAfter.millisecondsSinceEpoch)
 		}
@@ -116,7 +130,13 @@ public actor Conversations {
 		if let limit {
 			options.limit = Int64(limit)
 		}
-		return try await ffiConversations.listDms(opts: options).map {
+		let conversations = try await ffiConversations.listDms(
+			opts: options)
+
+		let sortedConversations = try sortConversations(
+			conversations, order: order)
+
+		return sortedConversations.map {
 			$0.dmFromFFI(client: client)
 		}
 	}
@@ -179,40 +199,49 @@ public actor Conversations {
 		}
 	}
 
-	public func stream() -> AsyncThrowingStream<
+	public func stream(type: ConversationType = .all) -> AsyncThrowingStream<
 		Conversation, Error
 	> {
 		AsyncThrowingStream { continuation in
 			let ffiStreamActor = FfiStreamActor()
+			let conversationCallback = ConversationStreamCallback {
+				conversation in
+				guard !Task.isCancelled else {
+					continuation.finish()
+					return
+				}
+				do {
+					let conversationType = try conversation.groupMetadata()
+						.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 {
+					// Do nothing if the conversation type is neither a group or dm
+				}
+			}
+
 			let task = Task {
-				let stream = await ffiConversations.stream(
-					callback: ConversationStreamCallback { conversation in
-						guard !Task.isCancelled else {
-							continuation.finish()
-							return
-						}
-						do {
-							let conversationType =
-								try conversation.groupMetadata()
-								.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 {
-							// Do nothing if the conversation type is neither a group or dm
-						}
+				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)
 					}
-				)
 				await ffiStreamActor.setFfiStream(stream)
 				continuation.onTermination = { @Sendable reason in
 					Task {
@@ -306,7 +335,8 @@ public actor Conversations {
 			throw ConversationError.memberCannotBeSelf
 		}
 		let addressMap = try await self.client.canMessage(addresses: addresses)
-		let unregisteredAddresses = addressMap
+		let unregisteredAddresses =
+			addressMap
 			.filter { !$0.value }
 			.map { $0.key }
 
@@ -328,34 +358,47 @@ public actor Conversations {
 		return group
 	}
 
-	public func streamAllMessages() -> AsyncThrowingStream<
-		DecodedMessage, Error
-	> {
+	public func streamAllMessages(type: ConversationType = .all)
+		-> AsyncThrowingStream<DecodedMessage, Error>
+	{
 		AsyncThrowingStream { continuation in
 			let ffiStreamActor = FfiStreamActor()
-			let task = Task {
-				let stream =
-					await ffiConversations
-					.streamAllMessages(
-						messageCallback: MessageCallback(client: self.client) {
-							message in
-							guard !Task.isCancelled else {
-								continuation.finish()
-								Task {
-									await ffiStreamActor.endStream()  // End the stream upon cancellation
-								}
-								return
-							}
-							do {
-								continuation.yield(
-									try Message(
-										client: self.client, ffiMessage: message
-									).decode())
-							} catch {
-								print("Error onMessage \(error)")
-							}
-						}
+
+			let messageCallback = MessageCallback(client: self.client) {
+				message in
+				guard !Task.isCancelled else {
+					continuation.finish()
+					Task {
+						await ffiStreamActor.endStream()
+					}
+					return
+				}
+				do {
+					continuation.yield(
+						try Message(client: self.client, ffiMessage: message)
+							.decode()
 					)
+				} catch {
+					print("Error onMessage \(error)")
+				}
+			}
+
+			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
+						)
+					}
 				await ffiStreamActor.setFfiStream(stream)
 			}