Wrap Firestore's addSnapshotListner method for use with AsyncThrowingStream.

Query+AsyncTrowingStream.swift

import FirebaseFirestore

// MARK: - async

extension Query {
    func addSnapshotListener<T>(
        includeMetadataChanges: Bool = false
    ) -> AsyncThrowingStream<[T], Error> where T: Decodable{
        .init { continuation in
            let listener = addSnapshotListener(includeMetadataChanges: includeMetadataChanges) { result in
                do {
                    let snapshot = try result.get()
                    continuation.yield(try snapshot.documents.map { try $0.data(as: T.self) })
                } catch {
                    continuation.finish(throwing: error)
                }
            }

            continuation.onTermination = { @Sendable _ in
                listener.remove()
            }
        }
    }
}

// MARK: - Result

extension Query {
    func addSnapshotListener(
        includeMetadataChanges: Bool = false,
        listener: @escaping (Result<QuerySnapshot, Error>) -> ()
    ) -> some ListenerRegistration {
        addSnapshotListener(includeMetadataChanges: includeMetadataChanges) { snapshot, error in
            if let error {
                listener(.failure(error))
            } else {
                listener(.success(snapshot!))
            }
        }
    }
}