Replication with GraphQL

With RxDB you can do a two-way replication with a GraphQL endpoint. This allows you to replicate data from the server into the client-side database and then query and modify it in realtime.

When the user is offline, you still can use the data and later sync it with the server when the client is online again like in other Offline-First systems.

Comparison to Couchdb-Sync

Pros:

The GraphQL-replication is faster and needs less resources

You do not need a couchdb-compliant endpoint, only a GraphQL-endpoint

Cons:

You can not replicate multiple databases with each other

It is assumed that the GraphQL-server is the single source of truth

You have to setup things at the server side while with couchdb-sync you only have to start a server

NOTICE: To play around, check out the full example of the RxDB GraphQL replication with server and client

Usage

Data Design

To use the GraphQL-replication you first have to ensure that your data is sortable by update time and your documents never get deleted, only have a deleted-flag set.

For example if your documents look like this,

{ "id" : "foobar" , "name" : "Alice" , "lastName" : "Wilson" , "updatedAt" : 1564783474 , "deleted" : false }

Then your data is always sortable by updatedAt . This ensures that when RxDB fetches 'new' changes, it can send the latest updatedAt to the GraphQL-endpoint and then recieve all newer documents.

Deleted documents still exist but have deleted: true set. This ensures that when RxDB fetches new documents, even the deleted documents are send back and can be known at the client-side.

GraphQL Server

At the server-side, there must exist an endpoint which returns newer rows when the last replicated document is used as input. For example lets say you create a Query feedForRxDBReplication which returns a list of newer documents, related to the given one, sorted by updatedAt .

For the push-replication, you also need a modifier which lets RxDB update data with a changed document as input.

input HumanInput { id: ID!, name: String!, lastName: String!, updatedAt: Int!, deleted: Boolean! } type Human { id: ID!, name: String!, lastName: String!, updatedAt: Int!, deleted: Boolean! } type Query { feedForRxDBReplication(lastId: String!, minUpdatedAt: Int!, limit: Int!): [Human!]! } type Mutation { setHuman(human: HumanInput): Human }

The resolver would then look like:

const rootValue = { feedForRxDBReplication: args => { const sortedDocuments = documents.sort((a, b) => { if (a.updatedAt > b.updatedAt) return 1 ; if (a.updatedAt < b.updatedAt) return -1 ; if (a.updatedAt === b.updatedAt) { if (a.id > b.id) return 1 ; if (a.id < b.id) return -1 ; else return 0 ; } }); const filterForMinUpdatedAtAndId = sortedDocuments.filter(doc => { if (doc.updatedAt < args.minUpdatedAt) return false ; if (doc.updatedAt > args.minUpdatedAt) return true ; if (doc.updatedAt === args.minUpdatedAt) { if (doc.id > args.lastId) return true ; else return false ; } }); const limited = filterForMinUpdatedAtAndId.slice( 0 , args.limit); return limited; }, setHuman: args => { const doc = args.human; documents = documents.filter(d => d.id !== doc.id); doc.updatedAt = Math .round( new Date ().getTime() / 1000 ); documents.push(doc); return doc; }, }

RxDB Client

Import the plugin

The graphql-replication is not part of the default-build of RxDB. You have to import the plugin before you can use it.

import { RxDBReplicationGraphQLPlugin } from 'rxdb/plugins/replication-graphql' ; addRxPlugin(RxDBReplicationGraphQLPlugin);

Pull replication

For the pull-replication, you first need a pullQueryBuilder . This is a function that gets the last replicated document as input and returns an object with a GraphQL-query and its variables (or a promise that resolves to the same object). RxDB will use the query builder to construct what is later send to the GraphQL endpoint.

const pullQueryBuilder = doc => { if (!doc) { doc = { id: '' , updatedAt: 0 }; } const query = `{ feedForRxDBReplication(lastId: " ${doc.name} ", minUpdatedAt: ${doc.updatedAt} , limit: 5) { id, name, lastName, updatedAt deleted } }` ; return { query, variables: {} }; };

With the queryBuilder, you can then setup the pull-replication.

const replicationState = myCollection.syncGraphQL({ url: 'http://example.com/graphql' , pull: { pullQueryBuilder, modifier: doc => doc }, deletedFlag: 'deleted' , live: true });

Push replication

For the push-replication, you also need a queryBuilder . Here, the builder recieves a changed document as input which has to be send to the server. It also returns a GraphQL-Query and its data.

const pushQueryBuilder = doc => { const query = ` mutation CreateHuman($human: HumanInput) { setHuman(human: $human) { id, updatedAt } } ` ; const variables = { human: doc }; return { query, variables }; };

With the queryBuilder, you can then setup the push-replication.

const replicationState = myCollection.syncGraphQL({ url: 'http://example.com/graphql' , push: { pushQueryBuilder, batchSize: 5 , modifier: d => d }, deletedFlag: 'deleted' , live: true });

Of course you can start the push- and the pull-replication in a single call to myCollection.syncGraphQL() .

Using subscriptions

For the pull-replication, RxDB will run the pull-function every 10 seconds to fetch new documents from the server. This means that when a change happens on the server, RxDB will, in the worst case, take 10 seconds until the changes is replicated to the client.

To improve this, it is recommended to setup GraphQL Subscriptions which will trigger the replication cycle when a change happens on the server.

import { SubscriptionClient } from 'subscriptions-transport-ws' ; const replicationState = myCollection.syncGraphQL({ url: 'http://example.com/graphql' , pull: { pullQueryBuilder, }, deletedFlag: 'deleted' , live: true , liveInterval: 60 * 1000 }); const wsClient = new SubscriptionClient( 'ws://example.com/subscriptions' , { reconnect: true , } ); const query = `subscription onHumanChanged { humanChanged { id } }` ; const changeObservable = wsClient.request({ query }); changeObservable.subscribe({ next(data) { replicationState.run(); } });

Helper Functions (beta)

RxDB provides the helper functions graphQLSchemaFromRxSchema() , pullQueryBuilderFromRxSchema() and pushQueryBuilderFromRxSchema() that can be used to generate the GraphQL Schema from the RxJsonSchema . To learn how to use them, please inspect the (GraphQL Example)[https://github.com/pubkey/rxdb/tree/master/examples/graphql]

Conflict Resolution

RxDB assumes that the Conflict Resolution will happen on the server side. When the clients sends a document to the server which causes a conflict, this has to be resolved there and then the resulting document can be synced down to RxDB. While CouchDB uses revision-flags for conflicts, you can use any logic like relying on the updatedAt date or other flags.

RxGraphQLReplicationState

When you call myCollection.syncGraphQL() it returns a RxGraphQLReplicationState which can be used to subscribe to events, for debugging or other functions.

.isStopped()

Returns true if the replication is stopped. This can be if a non-live replication is finished or a replication got canceled.

replicationState.isStopped();

.setHeaders()

Changes the headers for the replication after it has been set up.

replicationState.setHeaders({ Authorization: `...` });

.awaitInitialReplication()

Returns a Promise that is resolved as soon as the initial replication is done.

await replicationState.awaitInitialReplication(); console .log( 'initial sync done, client data is equal to server data' );

.run()

Triggers a replication cycle with the server. This is done automatically if the data changes on the client side or the pull-interval is called. This returns a Promise which is resolved when the run-cycle is done. Calling run() many times is no problem because it is queued internally.

await replicationState.run();

.cancel()

Cancels the replication. This is done autmatically if the RxCollection or it's RxDatabase is destroyed.

await replicationState.cancel();

.recieved$

An Observable that emits each document that is recieved from the endpoint.

.send$

An Observable that emits each document that is send to the endpoint.

.error$

An Observable that emits each error that happens during the replication. Use this if something does not work for debugging. RxDB will handle network errors automatically, other errors must be solved by the developer.

replicationState.error$.subscribe(error => { console .log( 'something was wrong' ); console .dir(error); });

.canceled$

An Observable that emits true when the replication is canceled, false if not.

.active$

An Observable that emits true when the replication is doing something, false when not.

NOTICE: To play around, check out the full example of the RxDB GraphQL replication with server and client

If you are new to RxDB, you should continue here