DO cross the streams! Introducing Combine Support in Ditto - build complex workflows from multiple reactive streams
Tags: swift, ios
Published: 2022-03-10 19:10:39

Historically, iOS developers had to use delegates, polling timers, notification centers, and callbacks to build reactive architectures. While these techniques are useful for simple callbacks, they falter when dealing with multiple events in robust event-driven apps. Creating complex chains, combinations, and permutations of multiple reactive streams is incredibly difficult. Thus, reactive libraries like the open source Rx suite [^1] and Apple's [Combine](https://developer.apple.com/documentation/combine) were created to solve this exact issue.

Since version 1.1.1 we have several `Combine.Publisher` extension methods on our long-running callback APIs. Notably:

- [LiveQueryPublisher](https://software.ditto.live/cocoa/DittoSwift/1.1.1/api-reference/Classes/DittoPendingCursorOperation/LiveQueryPublisher.html)
- [SingleDocumentLiveQueryPublisher](https://software.ditto.live/cocoa/DittoSwift/1.1.1/api-reference/Classes/DittoPendingIDSpecificOperation/SingleDocumentLiveQueryPublisher.html)
- [RemotePeersPublisher](https://software.ditto.live/cocoa/DittoSwift/1.1.1/api-reference/Classes/Ditto/RemotePeersPublisher.html)
- [FetchAttachmentPublisher](https://software.ditto.live/cocoa/DittoSwift/1.1.1/api-reference/Classes/DittoCollection/FetchAttachmentPublisher.html)

These new extension methods make it easy to use reactive techniques in your iOS or macOS apps without an additional library. We consider these APIs stable: they are available for iOS versions 13.0+ and macOS 10.15+. Of all the extension methods, you'll probably use `LiveQueryPublisher` the most. This Publisher is a method on the `DittoPendingCursorOperation` and facilitates most of the synchronization behavior of your app. Here's an example:

```swift
let cancellables = Set<AnyCancellables>()

// observe all documents in a collection
ditto.store["cars"]
  .findAll()
  .liveQueryPublisher()
  .sink { (documents, event)
    // do something with documents and events
  }
  .store(in: &cancellables)

// or you can observe documents matching a query
ditto.store["cars"]
  .find("color == $args.color && mileage > $args.mileage", args: [ "color": "red", "mileage": 5000])
  .sort("mileage")
  .limit(50)
  .liveQueryPublisher()
  .sink { (documents, event)
    // do something with documents and events
  }
  .store(in: &cancellables)
```

Stopping the live query is identical to stopping the publisher:

```swift
cancellable = nil // if you have a single cancellable, stop it by setting it to nil

cancellables.removeAll() // removeAll will stop all attached publishers and their respective live queries
```

Often, our users love to use [`Codable`](https://developer.apple.com/documentation/swift/codable) with Ditto Documents. Since `Codable`s can fail to decode for some reason (type mismatch is the most common example), you may want to handle each decode error individually as they're streamed out. The `liveQueryPublisher` gives each emission as an `Array<DittoDocument>` each time the query results change.

```swift
struct Car: Codable {
  var _id: String
  var name: String
  var mileage: Float
  var isSold: Bool
}

ditto.store["cars"]
  .findAll()
  .liveQueryPublisher()
  .flatMap({ (docs, _) in docs.publisher })
  .tryMap({ try $0.typed(as: Car.self).value })
  .sink(receiveCompletion: { (error) in
    print("Decoding a document failed: \(String(describing: error))")
  }, receiveValue: { (car: Car) in
    print("Successfully decoded a car \(Car)")
  })
```

There may be times where your application would like to collect the final values of only _successfully decoded_ `Car` objects. You may want this functionality if you're not interested in stale or malformed legacy documents. Here we use `flatMap` to turn the `Publisher<[DittoDocument]>` into a `Publisher<DittoDocument>` that will emit for each item in the array. Now for each of the single emitted `DittoDocument` we will use [`compactMap`](<https://developer.apple.com/documentation/combine/publisher/compactmap(_:)>) on the decoding function. Notice the `try?` here. If any `Car` fails to decode, this function will return a `nil`. `compactMap` will skip any `nil` values. At the end we will use `collect` to gather all of the emitted `Car` objects and repackage them into an `Array`.

```swift
let cancellable = ditto.store["cars"]
  .findAll()
  .liveQueryPublisher()
  .flatMap({ $0.documents.publisher })
  .compactMap({ try? $0.typed(as: Car.self).value })
  .collect()
  .sink { cars in
    print("Sucessfully decoded cars: \(cars). Failed to decode cars were removed from the array.")
  }
```

## SQL `JOIN`-like behavior with Ditto using Combine

A question that we get all the time is "How do I perform SQL-like `JOIN`s with Ditto"? While Ditto's current interface can't handle relationships like a traditional SQL database, our Combine support can help us achieve the same effect. [^2].

Let's say you're trying to build a menu view in SwiftUI like the first image shown below.

![Combine menu](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7ajsf3ct7ohvrowx92ij.jpg)

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wj9usui5pia7ns49zrhj.png)

It's likely that you'll use a SwiftUI [`List`](https://developer.apple.com/documentation/swiftui/list) with multiple [`Section`s](https://developer.apple.com/documentation/swiftui/section) with the help of [`ForEach`](https://developer.apple.com/documentation/swiftui/foreach). Assume each document in their respective collection looks like the following:

<div className="grid gap-4 lg:grid-cols-2">

```jsonc
// products
{
  "_id": "chicken-sandwich",
  "name": "Chicken Sandwich",
  "detail": "Grilled chicken, tomatoes, lettuce, and mustard",
  "categoryId": "entree"
}
```

```jsonc
// categories
{
  "_id": "entrees",
  "name": "Main Courses and Entrees"
}
```

</div>

We can create representable `Codable`s for each Document type. Notice that we've added `Identifiable` to help `ForEach` iteration:

<div className="grid gap-4 lg:grid-cols-2">

```swift
struct Category: Codable {
    var _id: String
    var name: String
    var isOnSale: Bool
}

extension Category: Identifiable {
    var id: String {
        return self._id
    }
}
```

```swift
struct Product: Codable {
    var _id: String
    var name: String
    var detail: String
    var categoryId: String
}

extension Product: Identifiable {
    var id: String {
        return self._id
    }
}
```

</div>

Notice that the `product` has a `categoryId`, this is our foreign key. Linking these foreign keys with our earlier APIs wasn't very straightforward. However, with our new Combine extensions, we can use the [`combineLatest`](<https://developer.apple.com/documentation/combine/just/combinelatest(_:)>) function to emit a single callback for both the `products` and `categories` collections.

First we will need to create a `JOIN`-ed struct that will house our nested values:

```swift
struct CategorizedProducts {
  // This is the category
  var category: Category
  // This is the products filtered by the category above
  var products: [Product]
}

// We add `Identifiable` to help `ForEach` iteration.
// Since this is unique by inner category._id property, we ensure
// to return its value
extension CategorizedProducts: Identifiable {
  var id: String {
      return self.category._id
  }
}
```

To populate `CategorizedProducts` we initially need to create our `combineLatest` implementation. First we will need to get access to both `categories` and `products` publishers. We use `Codable` to map documents into concrete data types in the `.tryMap` operator.

```swift
let categoriesPublisher = categoriesCollection.findAll().liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })

let productsPublisher = productsCollection.findAll().liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Product.self).value }) })
```

Finally, we can combine the latest values of each publisher using `.combineLatest` and `.map`. In the `.map` function, we iterate over each category and use it to create a `CategorizedProducts` object and filter all products by the `categoryId`.

```swift
let cancellable = categoriesPublisher.combineLatest(productsPublisher)
    .map { (categories, products) in
        return categories.map({ category -> CategorizedProducts in
            let filteredProducts = products.filter { product in product.categoryId == category._id }
            return CategorizedProducts(category: category, products: filteredProducts)
        })
    }
    .sink { categorizedProducts in
      print("categories with their products", categorizedProducts)
    }
```

If any update, insert, or deletions occur to the `products` _or_ `categories` collection, you'll always get a new set of `categorizedProducts`. To show this menu we can iterate over each `categorizedProducts` in SwiftUI like so:

```swift
List {
  ForEach(viewModel.categorizedProducts) { categorizedProducts in
    Section(categorizedProducts.category.name) {
      ForEach(categorizedProducts.products) { product in
        VStack(alignment: .leading) {
          Text(product.name)
              .bold()
          Text(product.detail)
              .font(.caption)
        }
      }
    }
  }
}
```

For an example project using this technique, [checkout the source code on GitHub here.](https://github.com/getditto/samples/tree/master/combine-menu)

### Making SQL `JOIN`-like behavior more efficient

Using `.combineLatest` as a way to get a SQL JOIN-like can achieve both a reactive API as well as better management of relational models. However, it's important to know that this is just an approximation. Remember, Ditto live queries sync exactly what you tell it to sync from the mesh. The example above will sync _all `categories` and all `products`_. This behavior may be desirable for many use cases but let's take at some ways we can reduce what is synced over the mesh.

You can use the queries to limit what documents are of interest by specifying a more selective query. Let's say you're only interested in getting `CategorizedProducts` where the category is one of `"appetizers"`, `"entrees"`, or `"desserts"`. We could now filter using a [query](https://docs.ditto.live/concepts/querying#compound---and--or--not--contains-contains) with `.find` instead of `.findAll` on the `categories` collection.

```swift
let categoriesPublisher = categoriesCollection
    .find("contains([$args.categoryIds], _id)", args: ["categoryId": ["appetizers", "entrees", "desserts"]])
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })

let productsPublisher = productsCollection
    .find("contains([$args.categoryIds], categoryIds)", args: ["categoryId": ["appetizers", "entrees", "desserts"]])
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Product.self).value }) })

let cancellable = categoriesPublisher.combineLatest(productsPublisher)
    .map { (categories, products) in
        return categories.map({ category -> CategorizedProducts in
            let filteredProducts = products.filter { product in product.categoryId == category._id }
            return CategorizedProducts(category: category, products: filteredProducts)
        })
    }
    .sink { categorizedProducts in
      print("categories with their products where categoryId are appetizers, entrees, desserts", categorizedProducts)
    }
```

Now the device will only sync relevant `CategorizedProducts` by the specified category \_ids.

### Advanced Scenarios for SQL `JOIN`-like behavior

Let's say we have a scenario where we only want to show `CategorizedProducts` where the `Category.isOnSale == true`. To do a SQL JOIN-like behavior with Ditto and Combine is not as straightforward as it was in the example above. This is because we are querying on a property that _only exists_ on the `categories` collection. Previously we were querying on both the primary key `Category._id` and the foreign key `Product.categoryId`. To do this is a bit harder than most and requires a decent amount of understanding of all the Combine operators.

First we will need to create a live query of `categories` where the `isOnSale == true`.

```swift
let categoriesPublisher = categoriesCollection
    .find("isOnSale == true")
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })
```

Since we don't know the categories that are passed ahead of time to filter the products, we need to filter them _after_ the values have returned from the categories live query publisher. Once we've received the categories that fit the `isOnSale == true` query, we can _then create_ map it's result into an `AnyPublisher<[CategorizedProducts], Never>`. We use `AnyPublisher<[CategorizedProducts], Never>` for brevity so that the entire chain isn't convoluted with complex generics.

Once the category's publisher emits data, we retrieve an array of `categoryIds: [String]` to feed it to the `products` live query publisher by filtering on the `categoryId` foreign key property. Next, _inside of the first categories publisher's map_, we will use our `.combineLatest` technique to map and filter the `CategorizedProducts`.

The most important function in this chain is the [`switchToLatest`](<https://developer.apple.com/documentation/combine/publisher/switchtolatest()-453ht>) right before the `.sink`. While it can be tricky for your eyes to follow, the `switchToLatest` will _dispose of any publishers_ if the top level publisher changes. This is a critical operator because we absolutely want to dispose the product live queries if the categories change. Categories may change if a new `isOnSale` is added, an existing category that has `isOnSale` becomes `false`, or is deleted. Without `switchToLatest` we will get mismatched `products` from previous categories.

```swift
let categoriesPublisher = categoriesCollection
    .find("isOnSale == true")
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })


categoriesPublisher
    .map({ categories -> AnyPublisher<[CategorizedProducts], Never> in
        // retrieve the categoryIds for all that were on sale.
        let categoryIds: [String] = categories.map{ $0._id }

        let productsPublisher = self.productsCollection
            .find("contains($args.categoryIds, categoryId)", args: ["categoryIds": categoryIds])
            .liveQueryPublisher()
            .tryMap({ try $0.documents.map({ try $0.typed(as: Product.self).value }) })
            .catch({ _ in Just([]) })
            .eraseToAnyPublisher()

        // we now create CategorizedProducts from the filtered categories and filtered products
        return Just(categories).combineLatest(productsPublisher)
            .map { (categories, products) -> [CategorizedProducts] in
                return categories.map({ category -> CategorizedProducts in
                    let filteredProducts = products.filter { product in product.categoryId == category._id }
                    return CategorizedProducts(category: category, products: filteredProducts)
                })
            }
            .eraseToAnyPublisher()
    })
    .switchToLatest() // extremely important so that we dispose of the products publisher if the categories change
    .catch({ _ in Just([]) })
    .sink { categorizedProducts in
      // completely filtered categorizedProducts
    }
```

Now if any category is added, removed, or updated to match `isOnSale == true`, we will instantly retrieve a new set of `CategorizedProducts` with live queries specifically limited to the matched `Category._id`
As you can see this is _very complex_ and really shows off the power of Combine with Ditto. We understand that the last example is a very complex and verbose code sample to achieve SQL `JOIN`-like behavior and we are working hard on adding native support directly within Ditto.

We're extremely excited to see all the new iOS and macOS applications that leverage Combine with Ditto. Building reactive applications has always been a core tenet of Ditto's design philosophy. Now with Combine, you can have incredible control over Ditto's live query system to build complex and robust event-driven apps.

## Footnotes

[^1]: Rx is suite of libraries that really gained a tremendous foothold for designing Reactive streams. There are libraries for JavaScript - [RxJS](https://rxjs.dev/), Swift - [RxSwift](https://github.com/ReactiveX/RxSwift), Java - [RxJava](https://github.com/ReactiveX/RxJava), Rx .NET - [Reactive Extensions](https://github.com/dotnet/reactive) and more. If you're building Ditto applications beyond Swift, you should take a look at these libraries. Most of the operators available in Apple's Combine framework have equivalent functions.
[^2]: We get so many requests for relationships, aggregates, syncing specific fields and more that we're planning on adding this functionality directly in an upcoming iteration of our Ditto query language. This will likely _look_ like SQL but will have some changes to accommodate working with Ditto's distributed database semantics.