Merge pull request #50 from Simperium/revisions/increase-count

Revisions: Increase count

Author: roundhill

package.json

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

src/simperium/channel.js

@@ -450,6 +450,11 @@ Channel.prototype.onChangeVersion = function( data ) {
 };
 
 Channel.prototype.onVersion = function( data ) {
+	// invalid version, give up without emitting
+	if ( data.slice( -2 ) === '\n?' ) {
+		return;
+	}
+
 	var ghost = parseVersionMessage( data );
 
 	this.emit( 'version', ghost.id, ghost.version, ghost.data );
@@ -647,35 +652,141 @@ LocalQueue.prototype.resendSentChanges = function() {
 	}
 }
 
+/**
+ * Since revision data is basically immutable we can prevent the
+ * need to refetch it after it has been loaded once.
+ *
+ * E.g. key could be `${ entityId }.${ versionNumber }`
+ *
+ * @type {Map<String,Object>} stores specific revisions as a cache
+ */
+export const revisionCache = new Map();
+
+/**
+ * Attempts to fetch an entity's revisions
+ *
+ * By default, a bucket stores two kinds of history:
+ * 	- revisions: the most-recent changes to an entity (60 of these)
+ * 	- archive: a "snapshot" of every ten revisions (100 of these)
+ *
+ * Together the revisions and archive span changes over the
+ * 1,060 most-recent changes to an entity, but of course once
+ * we hit the archive we lose save granularity.
+ *
+ * Individual buckets can override the defaults as well and also
+ * completely eliminate them.
+ *
+ * We don't have a listing of which revisions exist for a given entity.
+ *
+ * @param {Object} channel used to send messages to the Simperium server
+ * @param {String} id entity id for which to fetch revisions
+ * @param {Function} callback called on error or when finished
+ */
 function collectionRevisions( channel, id, callback ) {
-	var expectedVersions = -1;
-	var onGhostRetrieved = function( ghost ) {
-		var version = Math.min( ghost.version, 30 );
-		var i;
-		expectedVersions = version;
-
-		// Loop through requested revision count and request each version
-		for ( i = 0; i < version; i++ ) {
-			channel.send( 'e:' + id + '.' + ( ghost.version - i ) );
+	/** @type {Number} ms delay arbitrarily chosen to give up on fetch */
+	const TIMEOUT = 200;
+
+	/** @type {Set} tracks requested revisions */
+	const requestedVersions = new Set();
+
+	/** @type {Array<Object>} contains the revisions and associated data */
+	const versions = [];
+
+	/** @type {Number} remembers newest version of an entity */
+	let latestVersion;
+
+	/** @type {Number} handle for "start finishing" timeout */
+	let timeout;
+
+	/**
+	 * Receive a version update from the server and
+	 * dispatch the next fetch or finish the fetching
+	 *
+	 * @param {String} id entity id
+	 * @param {Number} version version of returned entity
+	 * @param {Object} data value of entity at revision
+	 */
+	function onVersion( id, version, data ) {
+		revisionCache.set( `${ id }.${ version }`, data );
+		versions.push( { id, version, data } );
+
+		// if we have every possible revision already, finish it!
+		// this bypasses any mandatory delay
+		if ( versions.length === latestVersion ) {
+			return finish();
 		}
-	};
 
-	var versions = [];
-	var onVersion = function( id, version, data ) {
-		versions.push( {id: id, version: version, data: data} );
+		fetchNextVersion( version );
+
+		// defer the final response to the application
+		clearTimeout( timeout );
+		timeout = setTimeout( finish, TIMEOUT );
+	}
+
+	/**
+	 * Stop listening for versions and stop fetching them
+	 * and pass accumulated data back to application
+	 */
+	function finish() {
+		clearTimeout( timeout );
+		channel.removeListener( `version.${ id }`, onVersion );
+
+		// sort newest first
+		callback( null, versions.sort( ( a, b ) => b.version - a.version ) );
+	}
 
-		// Check if all versions have been collected
-		if ( expectedVersions === versions.length ) {
-			channel.removeListener( 'version.' + id, onVersion );
-			callback( null, versions.sort( function( a, b ) {
-				return a.version > b.version ? -1 : 1;
-			} ) );
+	/**
+	 * Find the next version which isn't around and issue
+	 * a fetch if possible
+	 *
+	 * @param {Number} prevVersion starting point for finding next version
+	 */
+	function fetchNextVersion( prevVersion ) {
+		let version = prevVersion;
+
+		// find the next version to request
+		// some could have come back already
+		// or been requested already
+		while ( version > 0 && requestedVersions.has( version ) ) {
+			version -= 1;
 		}
-	};
 
-	channel.on( 'version.' + id, onVersion );
+		// we have them all
+		if ( ! version ) {
+			return;
+		}
 
-	channel.store.get( id ).then( onGhostRetrieved, function( e ) {
-		callback( e );
-	} );
+		requestedVersions.add( version );
+
+		// fetch from server or local cache
+		if ( revisionCache.has( `${ id }.${ version }` ) ) {
+			onVersion( id, version, revisionCache.get( `${ id }.${ version }` ) );
+		} else {
+			channel.send( `e:${ id }.${ version }` );
+		}
+	}
+
+	// start listening for the responses
+	channel.on( `version.${ id }`, onVersion );
+
+	// request the first revision and start the sequence
+	// pre-emptively fetch as many as could exist by default
+	channel.store.get( id ).then( ( { version } ) => {
+		latestVersion = version;
+
+		// grab latest change revisions
+		for ( let i = 0; i < 60 && ( version - i ) > 0; i++ ) {
+			fetchNextVersion( version - i );
+		}
+
+		// grab archive revisions
+		// these are like 1, 11, 21, 31, …, 41, normal revisions [42, 43, 44, 45, …]
+		const firstArchive = Math.round( ( version - 60 ) / 10 ) * 10 + 1; // 127 -> 67 -> 6 -> 60 -> 61
+		for ( let i = 0; i < 100 && ( firstArchive - 10 * i ) > 0; i++ ) {
+			fetchNextVersion( firstArchive - 10 * i );
+		}
+	}, callback );
+
+	// and set an initial timeout for failed connections
+	timeout = setTimeout( finish, TIMEOUT * 4 );
 }