Event Streaming

Event Streaming is a feature that notifies a client application of database changes as they occur. A subscribing application can use streaming to track changes without running ongoing polling queries. With streaming, client applications subscribe to changes on a target set of documents in the database identified by an FQL query.

Some use cases where event streaming is recommended:

  • Maintain "live" application state: Client apps or browsers are made aware immediately as data changes. Users can track live e-commerce inventory, auction status, and productivity workflow state.

  • Brokered communications, documents, and collections can be used as message channels between multiple parties. A chat application is a good example of such a case.

  • Publish-subscribe (pub-sub) use cases, including updating external services and data stores, and asynchronous work queuing. An application client can integrate with a message queue, inserting event notifications from Fauna into a Kafka topic, consumed by several downstream consumers, or data stores, such as a full-text search engine.

Supported drivers

Event Streaming is supported in the following Fauna drivers:

Create a stream

Define a stream

To create a stream for a set of documents, append the toStream() method to a query that returns the set.

For example, an e-commerce application manages a catalog of products in a collection named Product. The definition of this collection, which is used in this guide, is as follows:

collection Product {
    index byType {
        terms [ .type ]
        values [ .name, .price ]
    }
}

A business requirement of the application may be to take some administrative action when a change occurs in the Product collection. In other words, we want to get a notification when a document is added, removed, or changed in the Product collection. Fauna streaming is based on FQL. So, creating a stream that captures these notifications is simply a matter of initializing the stream with a query that captures the target set.

Product.all().toStream()

Notice that the base query Product.all() is a full collection scan, which when issued as a normal query returns all documents in a collection. That same query may be used to define the set of documents to be subscribed to. To create a stream, append the toStream() function to the query that defines the set to be streamed. The toStream() function returns a stream token, which is used to start the stream subscription.

Filter

Some use cases only require notifications for changes to a specific subset of documents in a collection. Because Fauna streaming is based on FQL queries, it is easy to filter, server-side, for that set of desired documents. Targeting for your desired set of documents with a stream query reduces data on the wire, and eliminates the need to filter our undesired event in your client application.

Consider the example requirement of our e-commerce application. It’s necessary to receive notifications on products of the category book that are less than 100 in price. The following stream query satisfies that requirement:

Product.where( .type == 'book' && .price < 100).toStream()

The stream query specifies the exact set of documents that will be subscribed to. Event notifications will be sent on the stream only when a change occurs in the set of documents defined by the stream query, such as when a document is added, deleted, or changed. For example, each of the following database operations would trigger a notification on the stream defined above.

Product.create( {
  name: "Music as History",
  type: "book",
  price: 16.89,
  quantity: 20
})
let p = Product.byType('book')
 .where(.name == "Music as History" ).first()!
p.update({ quantity: p.quantity + 10 })
Product.byType('book')
 .where(.name == "Music as History" ).first()!
 .update({ price: 110 })

Notice that the last example write operation, which increased the price of the book titled Music as History to 110, still triggers a notification event on the stream. This is because the book was changed in a way that left the set of documents the stream subscribed to. Although a notification was sent when the document left the set, any changes that occur to that document will no longer trigger an event notification because it no longer meets the criteria of the stream query.

Track changes on specific fields

In addition to specifying what set of documents that a stream is subscribed to, users may also define the set of fields in a document that will trigger a notification. Consider the case where the application needs to receive a notification whenever the price of a product in a target category changes. An example scenario is that a user is browsing for products of type electronics. The UI of the web application needs to be updated when the price of a product the user is browsing changes. The following stream query serves the requirements.

Product.byType('electronics').changesOn(.price)

With this stream query, the subscribing client will only receive notifications whenever the price of a Product of type electronics changes.

Stream changes on a single document

In some use cases, streaming on a single document is appropriate. For instance, consider that a user has placed a given product in their shopping cart. During the time the product is in the cart, the quantity may drop as other users purchase the same product. A requirement of the application is the user be notified that the quantity of a product is changing while it is in their cart. The following query would meet this requirement but is inefficient.

Product.where(.id == 390896421523423744).toStream()

The above stream is highly inefficient as it watches for changes in the entire Product collection but only produces events for one of them. Users can optimize this use case by creating a singleton set of a single product.

Set.single(Product.byId("390896421523423744")).toStream()

Optimize stream performance with indexes

Just as we strongly recommend the use of indexed queries for production workloads, we also recommend the use of indexes to increase the performance of stream queries. While an unindexed streaming query is allowed, your stream queries will be more performant and cost-effective when an index is used to serve them.

Notice that many of the preceding stream query examples used the byType index in our Product collection definition at the top of this page. This index uses the type field as the index term, and the price field as the index value. This index has been designed to suit the query patterns we’re streaming on.

Therefore, the stream is defined as:

Product.where( .type == 'book' && .price < 100).toStream()

It would be far more efficient in cost and time if it were modified, as follows, to use the byType index:

Product.byType('book').where( .price < 100 ).toStream()

There is a trade-off between streams on a collection versus streams on an index: index streams only produce events when fields in their configured terms or values change. So, while the use of an index increases the efficiency of a stream query, users need to design their indexes such that they fit the change events they need to be subscribed to. For example, given the stream query above, the following write operation would not trigger a notification.

let p = Product.byType('book')
 .where(.name == "Music as History" ).first()!

p.update({ quantity: p.quantity - 1 })

In this example, even though the book Music as History meets the query criteria of the indexed stream query, none of the fields covered by the index changed as a result of the write operation. Yes, the quantity field did change, but that is not a field that is covered by the index.

The need to have fields covered by the index extends to the use of changesOn() as well. Any field users wish to trigger a notification on in an index stream query must be covered by the index. The following example query would not be accepted by Fauna as a legal stream request, because the quantity field is neither a term nor a value of the byType index.

Product.byType('book').where(.price < 100 ).changesOn(.quantity)

Projecting only the data you need into the stream

By default, add and update events project the entire document that was modified, even when an index was used. In cases where it’s unnecessary to stream the entire document, users may add a projection to the stream query to suit their needs. Consider this example stream query, which projects only the name and price fields.

Product.byType('book').where( .price < 100 ).toStream(){name, price}

Only the document’s name and price fields will be returned in events on this stream.

{
  type: 'add',
  data: {
    name: 'The Fraud',
    price: 80
  },
  txn_ts: 1710456655930000,
  stats: {
    ...
  }
}

Consume streams using Fauna’s client drivers

Applications consume streams with the use of a Fauna client driver. Use the Client.stream() function to initialize the stream. Take the following example, where our earlier stream query is used to subscribe to the stream.

import { Client, fql } from "fauna";

const client = new Client();

const booksQuery = fql`Product.where(.type == 'book' && .price < 100).toStream()`;
const stream = client.stream(booksQuery);

try {
  for await (const event of stream) {
    switch (event.type) {
      case "add":
      case "update":
        // Do something on update
        console.log(event.data);
        break;
      case "remove":
        // Do something on remove
        console.log(event.data);
        break;
    }
  }
} catch (error) {
  console.log(error);
}

Event types

Write events that occur in the database will generate a set of notification types in the stream, based on what type of write occurred. You can see in the preceding JavaScript example that the various streaming events are handled with a switch statement. The set of event types are:

  • Add: Occurs whenever a new document has been created that matches the stream query’s criteria. Or, whenever a new document has been updated that matches the stream query’s criteria. Essentially, add events occur whenever a document enters the set defined by the stream query.

  • Remove: Occurs when a document leaves the set defined by the stream query. This can happen when a document that meets the stream query’s criteria is deleted. This can also happen when a document is updated to no longer meet the stream query’s criteria.

  • Update: Occurs when a field in a document has been updated that meets the stream query criteria.

  • Status: Occurs periodically, and serves to update the subscribing client, similar to a keep-alive, during periods when events are either not occurring on the database, or when events are filtered out of the stream server-side.

Your application code will need to handle the occurrence of any of these events.

Event shapes

Events sent through the stream include a field named type, which identifies the type of event that has occurred. This will be a string with a value of status, add, remove, or update, as described above. The data field will include the entire document by default. In the case that there is a projection in the stream query, then the data field will include only the fields defined in the projection.

For example, for the stream query Product.byType('book').where(.price < 100).toStream(), creating the following document.

Product.createData({
  type: 'book',
  name: 'The Fraud',
  quantity: "55",
  price: 23.0
})

Will generate the following notification:

{
  type: 'add',
  data: Document {
    coll: Module { name: 'Product' },
    id: '392372948586987553',
    ts: TimeStub { isoString: '2024-03-14T22:20:53.520Z' },
    type: 'book',
    name: 'The Fraud',
    quantity: '55',
    price: 23
  },
  txn_ts: 1710454853520000,
  stats: {
    read_ops: 1,
    storage_bytes_read: 109,
    compute_ops: 1,
    processing_time_ms: 1,
    rate_limits_hit: []
  }
}

Had the streaming query had the following projection of the name and price fields; Product.byType('book').where(.price < 100).toStream(){ name, price }, the update event would have had the following shape:

{
  type: 'add',
  data: { name: 'The Fraud', price: 23 },
  txn_ts: 1710454853520000,
  stats: {
    read_ops: 1,
    storage_bytes_read: 109,
    compute_ops: 1,
    processing_time_ms: 0,
    rate_limits_hit: []
  }
}

Stream snapshots

In some use cases, it’s necessary to first load a set of documents into the application that you then wish to keep updated with changes occurring in the database. Take the use case of presenting a leader board of a multiplayer gaming application as an example. As each player logs on to our multiplayer game, the leader board should be loaded into each player’s UI and kept up to date with player rankings and positions as each player’s score changes.

To serve such use cases, Fauna streaming allows you to both query for a set of documents, and open a stream on that same set of documents, all in the same operation. Consider the following code snippet, which builds on our earlier examples. The following code queries for the set of products of type book that are less than 100 in price. It starts an event stream on that same set.

import { Client, fql } from "fauna";

const client = new Client();

const booksQuery = fql`
  let products = Product.where( .type == 'book' && .price < 100 )
  {
    products: products,
    streamToken: products.toStream()
  }`;
const response = await client.query(booksQuery);
const { products, streamToken } = response.data;

// Paginate through all matching products
for await (const product of client.paginate(products)) {
  console.log(product);
}

const stream = client.stream(streamToken);
try {
  for await (const event of stream) {
    switch (event.type) {
      case "add":
      case "update":
        // Do something on update
        console.log(event.data);
        break;
      case "remove":
        // Do something on remove
        console.log(event.data);
        break;
    }
  }
} catch (error) {
  console.log(error);
}

Notice in the FQL query string, that the set of Product documents returned by the query is assigned to the variable products. That set is returned to the client using a projection to the products field of the object returned in the query operation. The toStream() function is called on the same products set, and that stream is projected into the streamToken field of the query result object. The application code then iterates on the initial seed set of documents returned by the query and then awaits for new events on that set.

Is this article helpful? 

Tell Fauna how the article can be improved:
 Visit Fauna's forums or email docs@fauna.com

Thank you for your feedback!