browser/syncbase_dispatcher.js - release.projects.todos - Git at Google

 // Copyright 2015 The Vanadium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 // Syncbase-based implementation of Dispatcher.
 //
 // Schema design doc (a bit outdated):
 // https://docs.google.com/document/d/1GtBk75QmjSorUW6T6BATCoiS_LTqOrGksgqjqJ1Hiow/edit#
 //
 // TODO(sadovsky): Support arbitrary item reordering using a Dewey-Decimal-like
 // order-tracking scheme, with suffix randomization to prevent conflicts.
 //
 // NOTE: For now, when an item is deleted, any sub-items that were added
 // concurrently (on some other device) are orphaned. Eventually, we'll GC
 // orphaned records; for now, we don't bother. This orphaning-based approach
 // enables us to use simple last-one-wins conflict resolution for all records
 // stored in Syncbase.
 // NOTE: For now, we actually always orphan tags when deleting todo entries.
 //
 // TODO(sadovsky): Orphaning degrades performance, because scan responses (e.g.
 // scan to get all todos for a given list) include orphaned records.

 'use strict';

 var _ = require('lodash');
 var async = require('async');
 var inherits = require('inherits');
 var syncbase = require('syncbase');
 var nosql = syncbase.nosql;
 var vanadium = require('vanadium');
 var verror = vanadium.verror, vtrace = vanadium.vtrace;

 var Dispatcher = require('./dispatcher');
 var util = require('./util');
 var wn = util.wn, wt = util.wt;

 inherits(SyncbaseDispatcher, Dispatcher);
 module.exports = SyncbaseDispatcher;

 function SyncbaseDispatcher(rt, db) {
   var that = this;
   Dispatcher.call(this);
   this.rt_ = rt;
   this.ctx_ = rt.getContext();
   this.db_ = db;
   this.tb_ = db.table('tb');

   // Initialize Syncbase watch.
   var errCb = function(err) {
     if (err) {
       that.emit('watchError', err);
     }
   };
   this.db_.getResumeMarker(that.ctx_, function(err, marker) {
     if (err) {
       errCb(err);
       return;
     }
     var stream = that.db_.watch(that.ctx_, that.tb_.name, '', marker, errCb);

     // TODO(aghassemi): Ideally the app would update its in-memory data
     // structures directly from the watch stream, but since syncgroup changes do
     // not yet show up in the watch stream, it can't. See v.io/i/764.
     stream.on('data', function(change) {
       // TODO(aghassemi): fromSync ought to be change.fromSync, but due to issue
       // v.io/i/685 we must set it to true for now.
       var fromSync = true;
       that.emit('change', fromSync);
     });

     stream.on('error', errCb);
   });
 }

 ////////////////////////////////////////
 // Helpers

 function noop() {}

 var SEP = '/';  // separator for key parts

 function join() {
   var args = Array.prototype.slice.call(arguments);
   return args.join(SEP);
 }

 // Similar to Go's strings.SplitN.
 function splitN(str, n) {
   console.assert(n >= 0);
   var parts = str.split(SEP);
   var res = parts.slice(0, n-1);
   if (parts.length >= n) {
     res.push(parts.slice(n-1).join(SEP));
   }
   return res;
 }

 function newListKey() {
   return util.uid();
 }

 function newTodoKey(listId) {
   return join(listId, 'todos', util.uid());
 }

 function tagKey(todoId, tag) {
   return join(todoId, 'tags', tag);
 }

 // TODO(sadovsky): Maybe switch from JSON to VOM/VDL.
 function marshal(x) {
   return JSON.stringify(x);
 }
 function unmarshal(x) {
   return JSON.parse(x);
 }

 var SILENT = new vanadium.context.ContextKey();

 // Defines a SyncbaseDispatcher method. If the first argument to fn is not a
 // context, creates a new context with a timeout.
 function define(name, fn) {
   SyncbaseDispatcher.prototype[name] = function() {
     var args = Array.prototype.slice.call(arguments);
     var ctx = args[0];
     if (ctx instanceof vanadium.context.Context) {
       args.shift();
     } else {
       ctx = wt(this.ctx_);
     }
     args = [wn(ctx, name)].concat(args);
     // The callback argument is optional, and clients don't always pass one.
     if (typeof args[args.length - 1] !== 'function') {
       args.push(noop);
     }
     if (!ctx.value(SILENT)) {
       // Build args string for logging the invocation. Drop ctx and cb, convert
       // to JSON, drop square brackets.
       var cb = args[args.length - 1];
       var argsStr = JSON.stringify(args.slice(1, -1)).slice(1, -1);
       args[args.length - 1] = util.logFn(name + '(' + argsStr + ')', cb);
     }
     return fn.apply(this, args);
   };
 }

 ////////////////////////////////////////
 // SyncbaseDispatcher impl

 // Returns list objects without 'sg' fields.
 define('getListsOnly_', function(ctx, cb) {
   this.getRowsQuery_(ctx, 'k not like "%' + SEP + '%"', function(err, rows) {
     if (err) return cb(err);
     var lists = _.map(rows, function(row) {
       console.assert(!row.key.includes(SEP));
       return _.assign(unmarshal(row.value), {_id: row.key});
     });
     return cb(null, lists);
   });
 });

 // Returns a list of objects describing syncgroups.
 // TODO(sadovsky): Would be nice if this could be done with a single RPC.
 define('getSyncgroups_', function(ctx, cb) {
   var that = this;
   this.db_.getSyncgroupNames(ctx, function(err, sgNames) {
     if (err) return cb(err);
     async.map(sgNames, function(sgName, cb) {
       that.getSyncgroup_(ctx, sgName, cb);
     }, cb);
   });
 });

 // Returns list objects with 'sg' fields.
 define('getLists', function(ctx, cb) {
   var that = this;
   async.parallel([
     function(cb) {
       that.getListsOnly_(ctx, cb);
     },
     function(cb) {
       that.getSyncgroups_(ctx, cb);
     }
   ], function(err, results) {
     if (err) return cb(err);
     var lists = results[0], sgs = results[1];
     // Left join: add 'sg' field to each list for which there's an SG.
     _.forEach(lists, function(list) {
       var listId = list._id;
       _.forEach(sgs, function(sg) {
         console.assert(sg.spec.prefixes.length === 1);
         if (listId === sg.spec.prefixes[0].row) {
           list.sg = sg;
         }
       });
     });
     return cb(null, lists);
   });
 });

 define('getTodos', function(ctx, listId, cb) {
   this.getRows_(ctx, join(listId, 'todos'), function(err, rows) {
     if (err) return cb(err);
     var todos = [];
     var todo = {};
     _.forEach(rows, function(row) {
       var parts = splitN(row.key, 5);
       console.assert(parts.length >= 3 && parts[0] === listId);
       if (parts.length === 3) {  // todo entry
         todo = _.assign(unmarshal(row.value), {_id: row.key, tags: []});
         todos.push(todo);
       } else if (parts.length === 5) {  // tag for todo entry
         var tagName = parts[4];
         if (tagKey(todo._id, tagName) !== row.key) {
           // Orphaned tag (from a deleted todo entry); skip.
           // TODO(ivanpi): Garbage-collect orphaned tags.
           return;
         }
         todo.tags.push(tagName);
       } else {
         throw new Error('bad key: ' + row.key);
       }
     });
     return cb(null, todos);
   });
 });

 define('addList', function(ctx, list, cb) {
   console.assert(!list._id);
   var listId = newListKey();
   var v = marshal(list);
   this.tb_.put(ctx, listId, v, function(err) {
     if (err) return cb(err);
     return cb(null, listId);
   });
 });

 define('editListName', function(ctx, listId, name, cb) {
   this.update_(ctx, listId, function(list) {
     return _.assign(list, {name: name});
   }, cb);
 });

 define('addTodo', function(ctx, listId, todo, cb) {
   console.assert(!todo._id);
   var that = this;
   var tags = todo.tags;
   delete todo.tags;
   var todoId = newTodoKey(listId);
   var v = marshal(todo);
   // Write todo and tags in a batch.
   var opts = new nosql.BatchOptions();
   nosql.runInBatch(ctx, this.db_, opts, function(db, cb) {
     // NOTE: Dealing with tables is awkward given that batches and syncgroups
     // are database-level. Maybe we should just get rid of tables. Doing so
     // would solve other problems as well, e.g. the API inconsistency for
     // creating databases vs. tables.
     var tb = db.table('tb');
     tb.put(wn(ctx, 'put:' + todoId), todoId, v, function(err) {
       if (err) return cb(err);
       async.each(tags, function(tag, cb) {
         that.addTagImpl_(ctx, tb, todoId, tag, cb);
       }, cb);
     });
   }, cb);
 });

 define('removeTodo', function(ctx, todoId, cb) {
   // TODO(ivanpi): Also delete corresponding tags.
   this.tb_.delete(ctx, todoId, cb);
 });

 define('editTodoText', function(ctx, todoId, text, cb) {
   this.update_(ctx, todoId, function(todo) {
     return _.assign(todo, {text: text});
   }, cb);
 });

 define('markTodoDone', function(ctx, todoId, done, cb) {
   this.update_(ctx, todoId, function(todo) {
     return _.assign(todo, {done: done});
   }, cb);
 });

 define('addTag', function(ctx, todoId, tag, cb) {
   this.addTagImpl_(ctx, this.tb_, todoId, tag, cb);
 });

 define('removeTag', function(ctx, todoId, tag, cb) {
   this.tb_.delete(ctx, tagKey(todoId, tag), cb);
 });

 ////////////////////////////////////////
 // Syncgroup methods

 // TODO(sadovsky): It's not clear from the Syncbase API how SG names should be
 // constructed, and it's also weird that db.Syncgroup(name) expects an absolute
 // name. We should probably allow clients to specify DB-relative SG names.

 // Currently, SG names must be of the form <syncbaseName>/%%sync/<suffix>.
 // We use <app>/<db>/<table>/<listId> for the suffix part.

 SyncbaseDispatcher.prototype.sgNameToListId = function(sgName) {
   return sgName.replace(new RegExp('.*/%%sync/todos/db/'), '');
 };

 SyncbaseDispatcher.prototype.listIdToSgName = function(listId) {
   // TODO(sadovsky): fullName doesn't include the mount table name, i.e. the
   // part corresponding to namespaceRoots. Our workaround is to specify a
   // fully-qualified Syncbase name.
   var prefix = this.tb_.fullName.replace('/todos/db/tb', '/%%sync/todos/db');
   return prefix + '/' + listId;
 };

 // Returns an object describing the syncgroup with the given name.
 // Currently, this object will have two fields: 'name' and 'spec'.
 define('getSyncgroup_', function(ctx, sgName, cb) {
   var sg = this.db_.syncgroup(sgName);
   async.parallel([
     function(cb) {
       sg.getSpec(ctx, function(err, spec, version) {
         if (err) return cb(err);
         cb(null, spec);
       });
     },
     function(cb) {
       // TODO(sadovsky): For now, in the UI we just want to show who's on the
       // ACL for a given list, so we don't bother with getMembers. On top of
       // that, currently getMembers returns a map of random Syncbase instance
       // ids to SyncgroupMemberInfo structs, neither of which tell us anything
       // useful.
       if (true) {
         process.nextTick(cb);
       } else {
         sg.getMembers(ctx, cb);
       }
     }
   ], function(err, results) {
     if (err) return cb(err);
     cb(null, {
       name: sgName,
       spec: results[0]
     });
   });
 });

 // TODO(sadovsky): Copied from test-syncgroup.js. I have no idea whether this
 // value is appropriate.
 var MEMBER_INFO = new nosql.SyncgroupMemberInfo({
   syncPriority: 8
 });

 function mkSgPerms(blessings) {
   // TODO(sadovsky): Maybe make perms more restrictive.
   return new Map([
     ['Admin',   {'in': blessings}],
     ['Read',    {'in': blessings}],
     ['Write',   {'in': blessings}],
     ['Resolve', {'in': blessings}],
     ['Debug',   {'in': blessings}]
   ]);
 }

 define('createSyncgroup', function(ctx, sgName, blessings, mtName, cb) {
   var sg = this.db_.syncgroup(sgName);
   var spec = new nosql.SyncgroupSpec({
     perms: mkSgPerms(blessings),
     prefixes: [new nosql.TableRow({
       tableName: 'tb',
       row: this.sgNameToListId(sgName)
     })],
     mountTables: [vanadium.naming.join(mtName, 'rendezvous')]
   });
   sg.create(ctx, spec, MEMBER_INFO, this.maybeEmit_(cb));
 });

 // TODO(sadovsky): The update to the syncgroup won't show up immediately in the
 // UIs of remote peers, since the watch stream does not include changes to
 // syncgroups. See v.io/i/764.
 define('addToSyncgroupPerms', function(ctx, sgName, blessing, cb) {
   var that = this;
   var sg = this.db_.syncgroup(sgName);
   sg.getSpec(ctx, function(err, spec, version) {
     var blessings = spec.perms.get('Admin')['in'];
     if (_.includes(blessings, blessing)) {
       return cb();  // addToSyncgroupPerms is idempotent
     }
     blessings.push(blessing);
     spec.perms = mkSgPerms(blessings);
     sg.setSpec(ctx, spec, version, that.maybeEmit_(cb));
   });
 });

 define('joinSyncgroup', function(ctx, sgName, cb) {
   var sg = this.db_.syncgroup(sgName);
   sg.join(ctx, MEMBER_INFO, this.maybeEmit_(cb));
 });

 ////////////////////////////////////////
 // vtrace methods

 // DO NOT USE THIS. vtrace RPCs are extremely slow in JavaScript because VOM
 // decoding is slow for trace records, which are deeply nested. E.g. 100 puts
 // can take 20+ seconds with vtrace vs. 2 seconds without.
 // EDIT: It might not be quite that bad - the "20+ seconds" cited above might
 // also include the latency added by having the Chrome dev console open.
 SyncbaseDispatcher.prototype.resetTraceRecords = function() {
   this.ctx_ = vtrace.withNewStore(this.rt_.getContext());
   vtrace.getStore(this.ctx_).setCollectRegexp('.*');
 };

 SyncbaseDispatcher.prototype.getTraceRecords = function() {
   return vtrace.getStore(this.ctx_).traceRecords();
 };

 SyncbaseDispatcher.prototype.logTraceRecords = function() {
   console.log(vtrace.formatTraces(this.getTraceRecords()));
 };

 ////////////////////////////////////////
 // Internal helpers

 // TODO(aghassemi): Remove this once changes to syncgroups are included in the
 // watch stream.
 SyncbaseDispatcher.prototype.maybeEmit_ = function(cb) {
   var that = this;
   cb = cb || noop;
   return function(err) {
     cb.apply(null, arguments);
     if (err) return;
     that.emit('change');
   };
 };

 // Writes the given tag into the given table.
 define('addTagImpl_', function(ctx, tb, todoId, tag, cb) {
   if (tag === '') {
     return process.nextTick(cb);
   }
   tb.put(ctx, tagKey(todoId, tag), null, cb);
 });

 // Returns all rows in the table with the given key prefix.
 define('getRows_', function(ctx, prefix, cb) {
   var rows = [], streamErr = null;
   var range = nosql.rowrange.prefix(prefix || '');
   this.tb_.scan(ctx, range, function(err) {
     if (err) return cb(err);
     if (streamErr) return cb(streamErr);
     cb(null, rows);
   }).on('data', function(row) {
     rows.push(row);
   }).on('error', function(err) {
     streamErr = streamErr || err.error;
   });
 });

 // Returns all rows in the table matching the where clause.
 define('getRowsQuery_', function(ctx, where, cb) {
   var rows = [], streamErr = null;
   var q = 'select k, v from ' + this.tb_.name + ' where ' + where;
   this.db_.exec(ctx, q, function(err) {
     if (err) return cb(err);
     if (streamErr) return cb(streamErr);
     rows = rows.slice(1);  // remove header
     cb(null, rows);
   }).on('data', function(row) {
     rows.push({key: row[0], value: row[1]});
   }).on('error', function(err) {
     streamErr = streamErr || err.error;
   });
 });

 // Performs a read-modify-write on key, applying updateFn to the value.
 // Takes care of value marshalling and unmarshalling.
 // TODO(sadovsky): Atomic read-modify-write requires 4 RPCs. Query-based update
 // mechanism would bring it down to 1.
 define('update_', function(ctx, key, updateFn, cb) {
   var opts = new nosql.BatchOptions();
   nosql.runInBatch(ctx, this.db_, opts, function(db, cb) {
     var tb = db.table('tb');
     tb.get(wn(ctx, 'get:' + key), key, function(err, value) {
       if (err) {
         if (err instanceof verror.NoExistError) {
           // Concurrent delete, likely from a remote peer. Pretend our update
           // never happened.
           // TODO(sadovsky): Maybe make it so client transactions take priority
           // over sync transactions, so that app developers don't have to worry
           // about this concurrency scenario.
           return cb();
         }
         return cb(err);
       }
       var newValue = marshal(updateFn(unmarshal(value)));
       tb.put(wn(ctx, 'put:' + key), key, newValue, cb);
     });
   }, cb);
 });
	// Copyright 2015 The Vanadium Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	// Syncbase-based implementation of Dispatcher.
	//
	// Schema design doc (a bit outdated):
	// https://docs.google.com/document/d/1GtBk75QmjSorUW6T6BATCoiS_LTqOrGksgqjqJ1Hiow/edit#
	//
	// TODO(sadovsky): Support arbitrary item reordering using a Dewey-Decimal-like
	// order-tracking scheme, with suffix randomization to prevent conflicts.
	//
	// NOTE: For now, when an item is deleted, any sub-items that were added
	// concurrently (on some other device) are orphaned. Eventually, we'll GC
	// orphaned records; for now, we don't bother. This orphaning-based approach
	// enables us to use simple last-one-wins conflict resolution for all records
	// stored in Syncbase.
	// NOTE: For now, we actually always orphan tags when deleting todo entries.
	//
	// TODO(sadovsky): Orphaning degrades performance, because scan responses (e.g.
	// scan to get all todos for a given list) include orphaned records.

	'use strict';

	var _ = require('lodash');
	var async = require('async');
	var inherits = require('inherits');
	var syncbase = require('syncbase');
	var nosql = syncbase.nosql;
	var vanadium = require('vanadium');
	var verror = vanadium.verror, vtrace = vanadium.vtrace;

	var Dispatcher = require('./dispatcher');
	var util = require('./util');
	var wn = util.wn, wt = util.wt;

	inherits(SyncbaseDispatcher, Dispatcher);
	module.exports = SyncbaseDispatcher;

	function SyncbaseDispatcher(rt, db) {
	var that = this;
	Dispatcher.call(this);
	this.rt_ = rt;
	this.ctx_ = rt.getContext();
	this.db_ = db;
	this.tb_ = db.table('tb');

	// Initialize Syncbase watch.
	var errCb = function(err) {
	if (err) {
	that.emit('watchError', err);
	}
	};
	this.db_.getResumeMarker(that.ctx_, function(err, marker) {
	if (err) {
	errCb(err);
	return;
	}
	var stream = that.db_.watch(that.ctx_, that.tb_.name, '', marker, errCb);

	// TODO(aghassemi): Ideally the app would update its in-memory data
	// structures directly from the watch stream, but since syncgroup changes do
	// not yet show up in the watch stream, it can't. See v.io/i/764.
	stream.on('data', function(change) {
	// TODO(aghassemi): fromSync ought to be change.fromSync, but due to issue
	// v.io/i/685 we must set it to true for now.
	var fromSync = true;
	that.emit('change', fromSync);
	});

	stream.on('error', errCb);
	});
	}

	////////////////////////////////////////
	// Helpers

	function noop() {}

	var SEP = '/'; // separator for key parts

	function join() {
	var args = Array.prototype.slice.call(arguments);
	return args.join(SEP);
	}

	// Similar to Go's strings.SplitN.
	function splitN(str, n) {
	console.assert(n >= 0);
	var parts = str.split(SEP);
	var res = parts.slice(0, n-1);
	if (parts.length >= n) {
	res.push(parts.slice(n-1).join(SEP));
	}
	return res;
	}

	function newListKey() {
	return util.uid();
	}

	function newTodoKey(listId) {
	return join(listId, 'todos', util.uid());
	}

	function tagKey(todoId, tag) {
	return join(todoId, 'tags', tag);
	}

	// TODO(sadovsky): Maybe switch from JSON to VOM/VDL.
	function marshal(x) {
	return JSON.stringify(x);
	}
	function unmarshal(x) {
	return JSON.parse(x);
	}

	var SILENT = new vanadium.context.ContextKey();

	// Defines a SyncbaseDispatcher method. If the first argument to fn is not a
	// context, creates a new context with a timeout.
	function define(name, fn) {
	SyncbaseDispatcher.prototype[name] = function() {
	var args = Array.prototype.slice.call(arguments);
	var ctx = args[0];
	if (ctx instanceof vanadium.context.Context) {
	args.shift();
	} else {
	ctx = wt(this.ctx_);
	}
	args = [wn(ctx, name)].concat(args);
	// The callback argument is optional, and clients don't always pass one.
	if (typeof args[args.length - 1] !== 'function') {
	args.push(noop);
	}
	if (!ctx.value(SILENT)) {
	// Build args string for logging the invocation. Drop ctx and cb, convert
	// to JSON, drop square brackets.
	var cb = args[args.length - 1];
	var argsStr = JSON.stringify(args.slice(1, -1)).slice(1, -1);
	args[args.length - 1] = util.logFn(name + '(' + argsStr + ')', cb);
	}
	return fn.apply(this, args);
	};
	}

	////////////////////////////////////////
	// SyncbaseDispatcher impl

	// Returns list objects without 'sg' fields.
	define('getListsOnly_', function(ctx, cb) {
	this.getRowsQuery_(ctx, 'k not like "%' + SEP + '%"', function(err, rows) {
	if (err) return cb(err);
	var lists = _.map(rows, function(row) {
	console.assert(!row.key.includes(SEP));
	return _.assign(unmarshal(row.value), {_id: row.key});
	});
	return cb(null, lists);
	});
	});

	// Returns a list of objects describing syncgroups.
	// TODO(sadovsky): Would be nice if this could be done with a single RPC.
	define('getSyncgroups_', function(ctx, cb) {
	var that = this;
	this.db_.getSyncgroupNames(ctx, function(err, sgNames) {
	if (err) return cb(err);
	async.map(sgNames, function(sgName, cb) {
	that.getSyncgroup_(ctx, sgName, cb);
	}, cb);
	});
	});

	// Returns list objects with 'sg' fields.
	define('getLists', function(ctx, cb) {
	var that = this;
	async.parallel([
	function(cb) {
	that.getListsOnly_(ctx, cb);
	},
	function(cb) {
	that.getSyncgroups_(ctx, cb);
	}
	], function(err, results) {
	if (err) return cb(err);
	var lists = results[0], sgs = results[1];
	// Left join: add 'sg' field to each list for which there's an SG.
	_.forEach(lists, function(list) {
	var listId = list._id;
	_.forEach(sgs, function(sg) {
	console.assert(sg.spec.prefixes.length === 1);
	if (listId === sg.spec.prefixes[0].row) {
	list.sg = sg;
	}
	});
	});
	return cb(null, lists);
	});
	});

	define('getTodos', function(ctx, listId, cb) {
	this.getRows_(ctx, join(listId, 'todos'), function(err, rows) {
	if (err) return cb(err);
	var todos = [];
	var todo = {};
	_.forEach(rows, function(row) {
	var parts = splitN(row.key, 5);
	console.assert(parts.length >= 3 && parts[0] === listId);
	if (parts.length === 3) { // todo entry
	todo = _.assign(unmarshal(row.value), {_id: row.key, tags: []});
	todos.push(todo);
	} else if (parts.length === 5) { // tag for todo entry
	var tagName = parts[4];
	if (tagKey(todo._id, tagName) !== row.key) {
	// Orphaned tag (from a deleted todo entry); skip.
	// TODO(ivanpi): Garbage-collect orphaned tags.
	return;
	}
	todo.tags.push(tagName);
	} else {
	throw new Error('bad key: ' + row.key);
	}
	});
	return cb(null, todos);
	});
	});

	define('addList', function(ctx, list, cb) {
	console.assert(!list._id);
	var listId = newListKey();
	var v = marshal(list);
	this.tb_.put(ctx, listId, v, function(err) {
	if (err) return cb(err);
	return cb(null, listId);
	});
	});

	define('editListName', function(ctx, listId, name, cb) {
	this.update_(ctx, listId, function(list) {
	return _.assign(list, {name: name});
	}, cb);
	});

	define('addTodo', function(ctx, listId, todo, cb) {
	console.assert(!todo._id);
	var that = this;
	var tags = todo.tags;
	delete todo.tags;
	var todoId = newTodoKey(listId);
	var v = marshal(todo);
	// Write todo and tags in a batch.
	var opts = new nosql.BatchOptions();
	nosql.runInBatch(ctx, this.db_, opts, function(db, cb) {
	// NOTE: Dealing with tables is awkward given that batches and syncgroups
	// are database-level. Maybe we should just get rid of tables. Doing so
	// would solve other problems as well, e.g. the API inconsistency for
	// creating databases vs. tables.
	var tb = db.table('tb');
	tb.put(wn(ctx, 'put:' + todoId), todoId, v, function(err) {
	if (err) return cb(err);
	async.each(tags, function(tag, cb) {
	that.addTagImpl_(ctx, tb, todoId, tag, cb);
	}, cb);
	});
	}, cb);
	});

	define('removeTodo', function(ctx, todoId, cb) {
	// TODO(ivanpi): Also delete corresponding tags.
	this.tb_.delete(ctx, todoId, cb);
	});

	define('editTodoText', function(ctx, todoId, text, cb) {
	this.update_(ctx, todoId, function(todo) {
	return _.assign(todo, {text: text});
	}, cb);
	});

	define('markTodoDone', function(ctx, todoId, done, cb) {
	this.update_(ctx, todoId, function(todo) {
	return _.assign(todo, {done: done});
	}, cb);
	});

	define('addTag', function(ctx, todoId, tag, cb) {
	this.addTagImpl_(ctx, this.tb_, todoId, tag, cb);
	});

	define('removeTag', function(ctx, todoId, tag, cb) {
	this.tb_.delete(ctx, tagKey(todoId, tag), cb);
	});

	////////////////////////////////////////
	// Syncgroup methods

	// TODO(sadovsky): It's not clear from the Syncbase API how SG names should be
	// constructed, and it's also weird that db.Syncgroup(name) expects an absolute
	// name. We should probably allow clients to specify DB-relative SG names.

	// Currently, SG names must be of the form <syncbaseName>/%%sync/<suffix>.
	// We use <app>/<db>/<table>/<listId> for the suffix part.

	SyncbaseDispatcher.prototype.sgNameToListId = function(sgName) {
	return sgName.replace(new RegExp('.*/%%sync/todos/db/'), '');
	};

	SyncbaseDispatcher.prototype.listIdToSgName = function(listId) {
	// TODO(sadovsky): fullName doesn't include the mount table name, i.e. the
	// part corresponding to namespaceRoots. Our workaround is to specify a
	// fully-qualified Syncbase name.
	var prefix = this.tb_.fullName.replace('/todos/db/tb', '/%%sync/todos/db');
	return prefix + '/' + listId;
	};

	// Returns an object describing the syncgroup with the given name.
	// Currently, this object will have two fields: 'name' and 'spec'.
	define('getSyncgroup_', function(ctx, sgName, cb) {
	var sg = this.db_.syncgroup(sgName);
	async.parallel([
	function(cb) {
	sg.getSpec(ctx, function(err, spec, version) {
	if (err) return cb(err);
	cb(null, spec);
	});
	},
	function(cb) {
	// TODO(sadovsky): For now, in the UI we just want to show who's on the
	// ACL for a given list, so we don't bother with getMembers. On top of
	// that, currently getMembers returns a map of random Syncbase instance
	// ids to SyncgroupMemberInfo structs, neither of which tell us anything
	// useful.
	if (true) {
	process.nextTick(cb);
	} else {
	sg.getMembers(ctx, cb);
	}
	}
	], function(err, results) {
	if (err) return cb(err);
	cb(null, {
	name: sgName,
	spec: results[0]
	});
	});
	});

	// TODO(sadovsky): Copied from test-syncgroup.js. I have no idea whether this
	// value is appropriate.
	var MEMBER_INFO = new nosql.SyncgroupMemberInfo({
	syncPriority: 8
	});

	function mkSgPerms(blessings) {
	// TODO(sadovsky): Maybe make perms more restrictive.
	return new Map([
	['Admin', {'in': blessings}],
	['Read', {'in': blessings}],
	['Write', {'in': blessings}],
	['Resolve', {'in': blessings}],
	['Debug', {'in': blessings}]
	]);
	}

	define('createSyncgroup', function(ctx, sgName, blessings, mtName, cb) {
	var sg = this.db_.syncgroup(sgName);
	var spec = new nosql.SyncgroupSpec({
	perms: mkSgPerms(blessings),
	prefixes: [new nosql.TableRow({
	tableName: 'tb',
	row: this.sgNameToListId(sgName)
	})],
	mountTables: [vanadium.naming.join(mtName, 'rendezvous')]
	});
	sg.create(ctx, spec, MEMBER_INFO, this.maybeEmit_(cb));
	});

	// TODO(sadovsky): The update to the syncgroup won't show up immediately in the
	// UIs of remote peers, since the watch stream does not include changes to
	// syncgroups. See v.io/i/764.
	define('addToSyncgroupPerms', function(ctx, sgName, blessing, cb) {
	var that = this;
	var sg = this.db_.syncgroup(sgName);
	sg.getSpec(ctx, function(err, spec, version) {
	var blessings = spec.perms.get('Admin')['in'];
	if (_.includes(blessings, blessing)) {
	return cb(); // addToSyncgroupPerms is idempotent
	}
	blessings.push(blessing);
	spec.perms = mkSgPerms(blessings);
	sg.setSpec(ctx, spec, version, that.maybeEmit_(cb));
	});
	});

	define('joinSyncgroup', function(ctx, sgName, cb) {
	var sg = this.db_.syncgroup(sgName);
	sg.join(ctx, MEMBER_INFO, this.maybeEmit_(cb));
	});

	////////////////////////////////////////
	// vtrace methods

	// DO NOT USE THIS. vtrace RPCs are extremely slow in JavaScript because VOM
	// decoding is slow for trace records, which are deeply nested. E.g. 100 puts
	// can take 20+ seconds with vtrace vs. 2 seconds without.
	// EDIT: It might not be quite that bad - the "20+ seconds" cited above might
	// also include the latency added by having the Chrome dev console open.
	SyncbaseDispatcher.prototype.resetTraceRecords = function() {
	this.ctx_ = vtrace.withNewStore(this.rt_.getContext());
	vtrace.getStore(this.ctx_).setCollectRegexp('.*');
	};

	SyncbaseDispatcher.prototype.getTraceRecords = function() {
	return vtrace.getStore(this.ctx_).traceRecords();
	};

	SyncbaseDispatcher.prototype.logTraceRecords = function() {
	console.log(vtrace.formatTraces(this.getTraceRecords()));
	};

	////////////////////////////////////////
	// Internal helpers

	// TODO(aghassemi): Remove this once changes to syncgroups are included in the
	// watch stream.
	SyncbaseDispatcher.prototype.maybeEmit_ = function(cb) {
	var that = this;
	cb = cb \|\| noop;
	return function(err) {
	cb.apply(null, arguments);
	if (err) return;
	that.emit('change');
	};
	};

	// Writes the given tag into the given table.
	define('addTagImpl_', function(ctx, tb, todoId, tag, cb) {
	if (tag === '') {
	return process.nextTick(cb);
	}
	tb.put(ctx, tagKey(todoId, tag), null, cb);
	});

	// Returns all rows in the table with the given key prefix.
	define('getRows_', function(ctx, prefix, cb) {
	var rows = [], streamErr = null;
	var range = nosql.rowrange.prefix(prefix \|\| '');
	this.tb_.scan(ctx, range, function(err) {
	if (err) return cb(err);
	if (streamErr) return cb(streamErr);
	cb(null, rows);
	}).on('data', function(row) {
	rows.push(row);
	}).on('error', function(err) {
	streamErr = streamErr \|\| err.error;
	});
	});

	// Returns all rows in the table matching the where clause.
	define('getRowsQuery_', function(ctx, where, cb) {
	var rows = [], streamErr = null;
	var q = 'select k, v from ' + this.tb_.name + ' where ' + where;
	this.db_.exec(ctx, q, function(err) {
	if (err) return cb(err);
	if (streamErr) return cb(streamErr);
	rows = rows.slice(1); // remove header
	cb(null, rows);
	}).on('data', function(row) {
	rows.push({key: row[0], value: row[1]});
	}).on('error', function(err) {
	streamErr = streamErr \|\| err.error;
	});
	});

	// Performs a read-modify-write on key, applying updateFn to the value.
	// Takes care of value marshalling and unmarshalling.
	// TODO(sadovsky): Atomic read-modify-write requires 4 RPCs. Query-based update
	// mechanism would bring it down to 1.
	define('update_', function(ctx, key, updateFn, cb) {
	var opts = new nosql.BatchOptions();
	nosql.runInBatch(ctx, this.db_, opts, function(db, cb) {
	var tb = db.table('tb');
	tb.get(wn(ctx, 'get:' + key), key, function(err, value) {
	if (err) {
	if (err instanceof verror.NoExistError) {
	// Concurrent delete, likely from a remote peer. Pretend our update
	// never happened.
	// TODO(sadovsky): Maybe make it so client transactions take priority
	// over sync transactions, so that app developers don't have to worry
	// about this concurrency scenario.
	return cb();
	}
	return cb(err);
	}
	var newValue = marshal(updateFn(unmarshal(value)));
	tb.put(wn(ctx, 'put:' + key), key, newValue, cb);
	});
	}, cb);
	});