Realtime pipeline public preview

Firestore/Swift/Source/PipelineResultChange.swift

@@ -20,7 +20,7 @@
 import Foundation
 
 @available(iOS 13, tvOS 13, macOS 10.15, macCatalyst 13, watchOS 7, *)
-struct PipelineResultChange: Sendable {
+public struct PipelineResultChange: Sendable {
   public enum ChangeType {
     case added, modified, removed
   }

Firestore/Swift/Source/SwiftAPI/Firestore+Pipeline.swift

@@ -60,7 +60,7 @@ import Foundation
   ///
   /// - Returns: A `RealtimePipelineSource` for building a realtime pipeline.
   @available(iOS 13, tvOS 13, macOS 10.15, macCatalyst 13, watchOS 7, *)
-  @nonobjc internal func realtimePipeline() -> RealtimePipelineSource {
+  @nonobjc func realtimePipeline() -> RealtimePipelineSource {
     return RealtimePipelineSource(db: self) { stages, db in
       RealtimePipeline(stages: stages, db: db)
     }

Firestore/Swift/Source/SwiftAPI/Pipeline/RealtimePipeline.swift

@@ -20,10 +20,10 @@
 import Foundation
 
 @available(iOS 13, tvOS 13, macOS 10.15, macCatalyst 13, watchOS 7, *)
-struct PipelineListenOptions: Sendable, Equatable, Hashable {
+public struct PipelineListenOptions: Sendable, Equatable, Hashable {
   /// Defines how to handle server-generated timestamps that are not yet known locally
   /// during latency compensation.
-  struct ServerTimestampBehavior: Sendable, Equatable, Hashable {
+  public struct ServerTimestampBehavior: Sendable, Equatable, Hashable {
     /// The raw string value for the behavior, used for implementation and hashability.
     let rawValue: String
     /// Creates a new behavior with a private raw value.
@@ -91,7 +91,7 @@ struct PipelineListenOptions: Sendable, Equatable, Hashable {
 }
 
 @available(iOS 13, tvOS 13, macOS 10.15, macCatalyst 13, watchOS 7, *)
-struct RealtimePipeline: @unchecked Sendable {
+public struct RealtimePipeline: @unchecked Sendable {
   private var stages: [Stage]
 
   let bridge: RealtimePipelineBridge
@@ -103,7 +103,7 @@ struct RealtimePipeline: @unchecked Sendable {
     bridge = RealtimePipelineBridge(stages: stages.map { $0.bridge }, db: db)
   }
 
-  struct Snapshot: Sendable {
+  public struct Snapshot: Sendable {
     /// An array of all the results in the `PipelineSnapshot`.
     let results_cache: [PipelineResult]
 

Firestore/Swift/Source/SwiftAPI/Pipeline/RealtimePipelineSource.swift

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 @available(iOS 13, tvOS 13, macOS 10.15, macCatalyst 13, watchOS 7, *)
-struct RealtimePipelineSource: @unchecked Sendable {
+public struct RealtimePipelineSource: @unchecked Sendable {
   let db: Firestore
   let factory: ([Stage], Firestore) -> RealtimePipeline
 
@@ -22,27 +22,62 @@ struct RealtimePipelineSource: @unchecked Sendable {
     self.factory = factory
   }
 
-  func collection(_ path: String) -> RealtimePipeline {
+  public func collection(_ path: String) -> RealtimePipeline {
     return factory([CollectionSource(collection: db.collection(path), db: db)], db)
   }
 
-  func collection(_ coll: CollectionReference) -> RealtimePipeline {
+  public func collection(_ coll: CollectionReference) -> RealtimePipeline {
     return factory([CollectionSource(collection: coll, db: db)], db)
   }
 
-  func collectionGroup(_ collectionId: String) -> RealtimePipeline {
+  public func collectionGroup(_ collectionId: String) -> RealtimePipeline {
     return factory(
       [CollectionGroupSource(collectionId: collectionId)],
       db
     )
   }
 
-  func documents(_ docs: [DocumentReference]) -> RealtimePipeline {
+  public func documents(_ docs: [DocumentReference]) -> RealtimePipeline {
     return factory([DocumentsSource(docs: docs, db: db)], db)
   }
 
-  func documents(_ paths: [String]) -> RealtimePipeline {
+  public func documents(_ paths: [String]) -> RealtimePipeline {
     let docs = paths.map { db.document($0) }
     return factory([DocumentsSource(docs: docs, db: db)], db)
   }
+
+  /// Creates a `RealtimePipeline` from an existing `Query`.
+  ///
+  /// This allows you to convert a standard Firestore query into a pipeline, which can then be
+  /// further modified with additional pipeline stages.
+  ///
+  /// - Parameter query: The `Query` to convert into a pipeline.
+  /// - Returns: A `RealtimePipeline` that is equivalent to the given query.
+  public func create(from query: Query) -> RealtimePipeline {
+    let stageBridges = PipelineBridge.createStageBridges(from: query)
+    let stages: [Stage] = stageBridges.map { bridge in
+      switch bridge.name {
+      case "collection":
+        return CollectionSource(
+          bridge: bridge as! CollectionSourceStageBridge,
+          db: query.firestore
+        )
+      case "collection_group":
+        return CollectionGroupSource(bridge: bridge as! CollectionGroupSourceStageBridge)
+      case "documents":
+        return DocumentsSource(bridge: bridge as! DocumentsSourceStageBridge, db: query.firestore)
+      case "where":
+        return Where(bridge: bridge as! WhereStageBridge)
+      case "limit":
+        return Limit(bridge: bridge as! LimitStageBridge)
+      case "sort":
+        return Sort(bridge: bridge as! SortStageBridge)
+      case "offset":
+        return Offset(bridge: bridge as! OffsetStageBridge)
+      default:
+        fatalError("Unknown stage type \(bridge.name)")
+      }
+    }
+    return factory(stages, db)
+  }
 }

Firestore/Swift/Tests/Integration/QueryIntegrationTests.swift

@@ -332,3 +332,41 @@ class QueryIntegrationTests: FSTIntegrationTestCase {
     )
   }
 }
+
+class QueryAsPipelineIntegrationTests: QueryIntegrationTests {
+  override class var isRunningPipeline: Bool {
+    return true
+  }
+
+  override func check(_ coll: CollectionReference, query: Query,
+                      matchesResult expectedKeys: [String]) async throws {
+    let collPipeline = coll.firestore.realtimePipeline().create(from: coll)
+    var collIterator = collPipeline.snapshotStream().makeAsyncIterator()
+    var _ = try await collIterator.next()
+
+    let pipeline = query.firestore.realtimePipeline().create(from: query)
+
+    var cacheIterator = pipeline.snapshotStream(options: .init(source: .cache)).makeAsyncIterator()
+    let cacheSnapshot = try await cacheIterator.next()
+    let cacheResultIds = cacheSnapshot?.results().map { $0.id }
+
+    var serverIterator = pipeline.snapshotStream(options: .init(
+      includeMetadataChanges: true,
+      source: .default
+    )).makeAsyncIterator()
+    var serverSnapshot = try await serverIterator.next()
+    if serverSnapshot?.metadata.isFromCache == true {
+      serverSnapshot = try await serverIterator.next()
+    }
+    let serverResultIds = serverSnapshot?.results().map { $0.id }
+
+    var remoteKeysIterator = pipeline.snapshotStream(options: .init(source: .cache))
+      .makeAsyncIterator()
+    let remoteKeysSnapshot = try await remoteKeysIterator.next()
+    let remoteKeysResultIds = remoteKeysSnapshot?.results().map { $0.id }
+
+    XCTAssertEqual(cacheResultIds, serverResultIds)
+    XCTAssertEqual(serverResultIds, remoteKeysResultIds)
+    XCTAssertEqual(remoteKeysResultIds, expectedKeys)
+  }
+}