Overwritten Ghost (#61)

* Adds a unit test to simulate a change queue being emptied when application restarts

Local modifications end up overwriting the network changes instead of merging the changes

* Refactor the updateObjectVersion function and document what it does

* Stubs out a method to use during a network change

- Fixes test that changes due to additional async call, the sent change will be the rebased patch

* Adds subscribe method to bucket and channel to support handling network changes

* Adds bucket implementation for subscriber interface

- adds unit tests for bucket

* Update jsdoc for NetworkChangeSubscriber return value

* Remove console log

* typos

* Fixing things

* Update test so there is no sent change pending

* subscribe is now beforeNetworkChange

Instance names, arguments, methods have been changed to more clearly reflect the purpose of the API

* Add release notes and version bump

* Fix Channel change processor to always use the latest Ghost (#84)

* Always produce diffs against the most up to date ghost

* Flow check on the CI

* Don't track sent change unless it is actually sent

* Test: Crossed Wires scenario (#86)

* Integration test for 'crossed wires' problem

* Fix: stop skipping updates when local changes queued

Author: beaucollins

.travis.yml

@@ -5,6 +5,8 @@ script:
   - npm run lint
   # unit tests with coverage report
   - npm run test:coverage
+  # Flow type check
+  - npx flow
 node_js:
   - "10"
   - "9"

RELEASE-NOTES.txt

@@ -2,4 +2,7 @@
 
 ## Future Release
 
-- Fixed data corruption when remote updates arrive while waiting on confirmation of local changes [#78](https://github.com/Simperium/node-simperium/pull/78)
+### Fixes
+
+- Prevent data corruption when remote updates arrive while waiting on confirmation of local changes [#78](https://github.com/Simperium/node-simperium/pull/78)
+- Prevent data loss when receiving remote updates while local changes exist [#61](https://github.com/Simperium/node-simperium/pull/61)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "simperium",
-  "version": "0.3.3",
+  "version": "1.0.0",
   "description": "A simperium client for node.js",
   "main": "./lib/simperium/index.js",
   "repository": {

src/simperium/auth.js

@@ -10,7 +10,7 @@ type User = {
 };
 
 const fromJSON = ( json: string ): User => {
-	const data: {} = JSON.parse( json );
+	const data = JSON.parse( json );
 	if ( ! data.access_token && typeof data.access_token !== 'string' ) {
 		throw new Error( 'access_token not present' );
 	}
@@ -74,9 +74,10 @@ export class Auth extends EventEmitter {
 	}
 
 	getUrlOptions( path: string ) {
-		const options = url.parse( `${URL}/${ this.appId }/${ path}` );
+		const { port, ...options } = url.parse( `${URL}/${ this.appId }/${ path}` );
 		return {
 			... options,
+			port: port ? Number( port ) : undefined,
 			method: 'POST',
 			headers: {'X-Simperium-API-Key': this.appSecret }
 		};

src/simperium/bucket.js

@@ -196,8 +196,64 @@ Bucket.prototype.setChannel = function( channel ) {
 		.on( 'update', this.onChannelUpdate )
 		.on( 'indexingStateChange', this.onChannelIndexingStateChange )
 		.on( 'remove', this.onChannelRemove );
+
+	/**
+	 * Fetch the current state for a given id in the datastore
+	 *
+	 * @param {string} id - object to fetch current state for
+	 * @returns {Promise<Object|null>} local state or null if no state to provide
+	 */
+	const localStateForKey = ( id ) => {
+		return this.get( id ).then( ( object ) => {
+			if ( object ) {
+				return object.data;
+			}
+		} );
+	}
+
+	// Sets up a default network change subscriber that triggers a channel
+	// change operation. This will queue up any changes that exist that
+	// may have not been added to the local queue yet
+	channel.beforeNetworkChange( ( id, ... args ) => {
+		const changeResolver = this.changeResolver
+			// if there is a subcriber, let it handle the network change
+			? Promise.resolve( this.changeResolver( id, ... args ) )
+			// if no subscriber, resolving null will mean we should handle it
+			: Promise.resolve( null );
+
+		return changeResolver
+			.then( localState => {
+				// if subscriber provided local state of any truthy type use that
+				// as the object's local state
+				if ( localState ) {
+					return localState;
+				}
+				// Subscriber did not provide local state, fetch it from the datastore
+				return localStateForKey( id );
+			} )
+	} );
 };
 
+/**
+ * @callback NetworkChangeResolver
+ * @param { string } key - bucket object being changed
+ * @param { Object } data - the new object data
+ * @param { Object } base - the object data before the patch is applied
+ * @param { Object } patch - the patch used to bring base to data
+ * @returns { Object | null | Promise<Object | null> } - resolve when network change should be applied
+ */
+
+/**
+ * Subscribe to changes to this bucket that are coming from simperium that
+ * did not originate on this client. Can be used to save an object before
+ * network changes are applied.
+ *
+ * @param {NetworkChangeResolver} changeResolver - callback executed when network changes for this bucket are going to be applied
+ */
+Bucket.prototype.beforeNetworkChange = function( changeResolver ) {
+	this.changeResolver = changeResolver;
+}
+
 /**
  * Reloads all the data from the currently cached set of ghost data
  */
@@ -258,9 +314,11 @@ Bucket.prototype.update = function( id, data, remoteUpdateInfo, options, callbac
 	}
 
 	const task = this.storeAPI.update( id, data, this.isIndexing )
+		.then( bucketObject => {
+			return this.channel.update( bucketObject, options.sync );
+		} )
 		.then( bucketObject => {
 			this.emit( 'update', id, bucketObject.data, remoteUpdateInfo );
-			this.channel.update( bucketObject, options.sync );
 			return bucketObject;
 		} );
 	return deprecateCallback( callback, task );
@@ -307,11 +365,15 @@ Bucket.prototype.getVersion = function( id, callback ) {
  *
  * @param {String} id - object to sync
  * @param {?bucketStoreGetCallback} callback - optional callback
- * @returns {Promise<Object>} - object id, data
+ * @returns {Promise<BucketObject>} - object id, data
  */
 Bucket.prototype.touch = function( id, callback ) {
 	const task = this.storeAPI.get( id )
-		.then( object => this.update( object.id, object.data ) );
+		.then( object => {
+			if ( object ) {
+				return this.update( object.id, object.data );
+			}
+		} );
 
 	return deprecateCallback( callback, task );
 };