synctable.coffee

###
CoCalc, Copyright (C) 2016, Sagemath Inc.

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

---

SYNCHRONIZED TABLE -- defined by an object query

    - Do a query against a PostgreSQL table using our object query description.
    - Synchronization with the backend database is done automatically.

   Methods:
      - constructor(query): query = the name of a table (or a more complicated object)

      - set(map):  Set the given keys of map to their values; one key must be
                   the primary key for the table.  NOTE: Computed primary keys will
                   get automatically filled in; these are keys in schema.coffee,
                   where the set query looks like this say:
                      (obj, db) -> db.sha1(obj.project_id, obj.path)
      - get():     Current value of the query, as an immutable.js Map from
                   the primary key to the records, which are also immutable.js Maps.
      - get(key):  The record with given key, as an immutable Map.
      - get(keys): Immutable Map from given keys to the corresponding records.
      - get_one(): Returns one record as an immutable Map (useful if there
                   is only one record)

      - close():   Frees up resources, stops syncing, don't use object further

   Events:
      - 'before-change': fired right before (and in the same event loop) actually
                  applying remote incoming changes
      - 'change', [array of string primary keys] : fired any time the value of the query result
                 changes, *including* if changed by calling set on this object.
                 Also, called with empty list on first connection if there happens
                 to be nothing in this table.   If the primary key is not a string it is
                 converted to a JSON string.
      - 'disconnected': fired when table is disconnected from the server for some reason
      - 'connected': fired when table has successfully connected and finished initializing
                     and is ready to use
      - 'saved', [array of saved objects]: fired after confirmed successful save of objects to backend

STATES:

A SyncTable is a finite state machine as follows:

                          -------------------<------------------
                         \|/                                   |
    [connecting] --> [connected]  -->  [disconnected]  --> [reconnecting]

Also, there is a final state called 'closed', that the SyncTable moves to when
it will not be used further; this frees up all connections and used memory.
The table can't be used after it is closed.   The only way to get to the
closed state is to explicitly call close() on the table; otherwise, the
table will keep attempting to connect and work, until it works.

    (anything)  --> [closed]



- connecting   -- connecting to the backend, and have never connected before.

- connected    -- successfully connected to the backend, initialized, and receiving updates.

- disconnected -- table was successfully initialized, but the network connection
                  died. Can still takes writes, but they will never try to save to
                  the backend.  Waiting to reconnect when user connects back to the backend.

- reconnecting -- client just reconnected to the backend, so this table is now trying
                  to get the full current state of the table and initialize a changefeed.

- closed       -- table is closed, and memory/connections used by the table is freed.


WORRY: what if the user does a set and connecting (or reconnecting) takes a long time, e.g., suspend
a laptop, then resume?  The changes may get saved... a month later.  For some things, e.g., logs,
this could be fine.  However, on reconnect, the first thing is that complete upstream state of
table is set on server version of table, so reconnecting user only sends its changes if upstream
hasn't changed anything in that same record.
###

# if true, will log to the console a huge amount of info about every get/set
DEBUG = false

exports.set_debug = (x) ->
    DEBUG = !!x

{EventEmitter} = require('events')
immutable      = require('immutable')
async          = require('async')
underscore     = require('underscore')

misc           = require('./misc')
schema         = require('./schema')

{defaults, required} = misc

# We represent synchronized tables by an immutable.js mapping from the primary
# key to the object.  Since PostgresQL primary keys can be compound (more than
# just strings), e.g., they can be arrays, so we convert complicated keys to their
# JSON representation.  A binary object doesn't make sense here in pure javascript,
# but these do:
#       string, number, time, boolean, or array
# Everything automatically converts fine to a string except array, which is the
# main thing this function deals with below.
# NOTE (1)  RIGHT NOW:  This should be safe to change at
# any time, since the keys aren't stored longterm.
# If we do something with localStorage, this will no longer be safe
# without a version number.
# NOTE (2) Of course you could use both a string and an array as primary keys
# in the same table.  You could evily make the string equal the json of an array,
# and this *would* break things.  We are thus assuming that such mixing
# doesn't happen.  An alternative would be to just *always* use a *stable* version of stringify.
# NOTE (3) we use a stable version, since otherwise things will randomly break if the
# key is an object.

json_stable_stringify = require('json-stable-stringify')

to_key = (x) ->
    if typeof(x) == 'object'
        return json_stable_stringify(x)
    else
        return x

# Plug: Class to ensure that the SyncTable stays "plugged" into the hub, if at all possible.
# NOTE: I implemented this outside of SyncTable so that it would be much easier
#       to reason about, and be sure the code is right.
class Plug
    constructor : (opts) ->
        @_opts = defaults opts,
            name       : 'plug'     # Used only for debug logging
            no_sign_in : required   # True if sign is isn't required before connecting, e.g., anonymous synctable and project.
            client     : required   # The client object, which provides:
                                    #   'connected' and 'signed_in' events, and
                                    #   is_connected() and is_signed_in() functions.
            connect    : required   # A function to call to create a connection; it should run as
                                    # quickly as it can and call it's callback with an error if
                                    # and only if it fails.  It will definitely only be called
                                    # once at a time, so no need to put in any sort of block.
        @connect()

    dbg: (f) =>
        #return @_opts.client.dbg("Plug('#{@_opts.name}').#{f}")
        return =>

    # Keep trying until we connect - always succeeds if it terminates
    connect: (cb) =>
        dbg = @dbg('connect')
        if @_is_connecting
            dbg("already connecting")
            return
        @_is_connecting = true
        dbg('')
        misc.retry_until_success
            f           : @__try_to_connect_once
            log         : dbg
            start_delay : 4000
            max_delay   : 20000
            cb          : =>
                delete @_is_connecting
                dbg("success!")
                cb?()

    # Try to connect exactly once.  cb gets error if and only if fails to connect.
    __try_to_connect_once: (cb) =>
        # timer for giving up on waiting to try to connect
        give_up_timer = undefined

        # actually try to connect
        do_connect = =>
            if give_up_timer
                clearInterval(give_up_timer)
            @_opts.connect(cb)

        # Which event/condition has too be true before we even try to connect.
        if @_opts.no_sign_in
            event = 'connected'
        else
            event = 'signed_in'

        if @_opts.client["is_#{event}"]()
            # The condition is satisfied, so try once to connect.
            do_connect()
        else
            # Wait until condition is satisfied...
            @_opts.client.once(event, do_connect)
            # ... but don't wait forever, in case for some reason we miss
            # the event (this can maybe rarely happen).
            give_up = =>
                @_opts.client.removeListener(event, do_connect)
                cb("timeout")
            timer = setTimeout(give_up, 5000+Math.random()*10000)

class SyncTable extends EventEmitter
    constructor: (@_query, @_options, @_client, @_debounce_interval, @_throttle_changes, @_cache_key) ->
        @_init_query()
        # The value of this query locally.
        @_value_local = undefined

        # Our best guess as to the value of this query on the server,
        # according to queries and updates the server pushes to us.
        @_value_server = undefined

        # The changefeed id, when set by doing a change-feed aware query.
        @_id = undefined

        # Not connected yet
        @_state = 'disconnected'   # disconnected <--> connected --> closed
        @_created = new Date()

        @_plug = new Plug
            name       : @_table
            client     : @_client
            connect    : @_connect
            no_sign_in : @_schema.anonymous or @_client.is_project()  # note: projects don't have to authenticate

        @_client.on 'disconnected', =>
            #console.log("synctable: DISCONNECTED")
            # When the connection is dropped, the backend hub notices that it was dropped
            # and immediately cancels all changefeeds.  Thus we set @_id to undefined
            # below, so that we don't redundantly cancel them again, which leads to an error
            # and wastes resources (which can pile up).
            @_id = undefined
            @_disconnected('client disconnect')

        # No throttling of change events unless explicitly requested *or* part of the schema.
        @_throttle_changes ?= schema.SCHEMA[@_table]?.user_query?.get?.throttle_changes

        if not @_throttle_changes
            @emit_change = (changed_keys) => @emit('change', changed_keys)
        else
            # throttle emitting of change events
            all_changed_keys = {}
            do_emit_changes = =>
                #console.log("#{@_table} -- emitting changes", misc.keys(all_changed_keys))
                # CRITICAL: some code depends on emitting change even for the *empty* list of keys!
                # E.g., projects page won't load for new users.  This is the *change* from not
                # loaded to being loaded, which does make sense.
                @emit('change', misc.keys(all_changed_keys))
                all_changed_keys = {}
            do_emit_changes = underscore.throttle(do_emit_changes, @_throttle_changes)
            @emit_change = (changed_keys) =>
                #console.log("#{@_table} -- queue changes", changed_keys)
                for key in changed_keys
                    all_changed_keys[key] = true
                do_emit_changes()

    dbg: (f) =>
        #return @_client.dbg("SyncTable('#{@_table}').#{f}")
        return =>

    _connect: (cb) =>
        dbg = @dbg("connect")
        dbg()
        if @_state == 'closed'
            cb?('closed')
            return
        if @_state == 'connected'
            cb?()
            return
        if @_id?
            @_client.query_cancel(id:@_id)
            @_id = undefined

        async.series([
            (cb) =>
                # 1. save, in case we have any local unsaved changes, then sync with upstream.
                if @_value_local? and @_value_server?
                    @_save(cb)
                else
                    cb()
            (cb) =>
                # 2. Now actually do the changefeed query.
                @_reconnect(cb)
        ], cb)

    _reconnect: (cb) =>
        dbg = @dbg("_run")
        if @_state == 'closed'
            dbg("closed so don't do anything ever again")
            cb?()
            return
        first_resp = true
        this_query_id = undefined
        dbg("do the query")
        @_client.query
            query   : @_query
            changes : true
            timeout : 30
            options : @_options
            cb      : (err, resp) =>

                if @_state == 'closed'
                    # already closed so ignore anything else.
                    return

                if first_resp
                    dbg("query got ", err, resp)
                    first_resp = false
                    if @_state == 'closed'
                        cb?("closed")
                    else if resp?.event == 'query_cancel'
                        cb?("query-cancel")
                    else if err
                        cb?(err)
                    else if not resp?.query?[@_table]?
                        cb?("got no data")
                    else
                        # Successfully completed query
                        this_query_id = @_id = resp.id
                        @_state = 'connected'
                        @_update_all(resp.query[@_table])
                        @emit("connected", resp.query[@_table])  # ready to use!
                        cb?()
                        # Do any pending saves
                        for cb in @_connected_save_cbs ? []
                            @save(cb)
                        delete @_connected_save_cbs
                else
                    if @_state != 'connected'
                        dbg("nothing to do -- ignore these, and make sure they stop")
                        if this_query_id?
                            @_client.query_cancel(id:this_query_id)
                        return
                    if err or resp?.event == 'query_cancel'
                        @_disconnected("err=#{err}, resp?.event=#{resp?.event}")
                    else
                        # Handle the update
                        @_update_change(resp)

    _disconnected: (why) =>
        dbg = @dbg("_disconnected")
        dbg("why=#{why}")
        if @_state == 'disconnected'
            dbg("already disconnected")
            return
        if @_id
            @_client.query_cancel(id:@_id)
        @_state = 'disconnected'
        @_plug.connect()  # start trying to connect again

    # Return string key used in the immutable map in which this table is stored.
    key: (obj) =>
        return @_key(obj)

    # Return true if there are changes to this synctable that
    # have NOT been confirmed as saved to the backend database.
    has_uncommitted_changes: () =>
        if not @_value_server? and not @_value_local?
            return false
        if @_value_local? and not @_value_server?
            return true
        return not @_value_server.equals(@_value_local)

    get: (arg) =>
        if not @_value_local?
            return
        if arg?
            if misc.is_array(arg)
                x = {}
                for k in arg
                    x[to_key(k)] = @_value_local.get(to_key(k))
                return immutable.fromJS(x)
            else
                return @_value_local.get(to_key(arg))
        else
            return @_value_local

    get_one: =>
        return @_value_local?.toSeq().first()

    _parse_query: (query) =>
        if typeof(query) == 'string'
            # name of a table -- get all fields
            v = misc.copy(schema.SCHEMA[query].user_query.get.fields)
            for k, _ of v
                v[k] = null
            return {"#{query}": [v]}
        else
            keys = misc.keys(query)
            if keys.length != 1
                throw Error("must specify exactly one table")
            table = keys[0]
            x = {}
            if not misc.is_array(query[table])
                return {"#{table}": [query[table]]}
            else
                return {"#{table}": query[table]}

    _init_query: =>
        # first parse the query to allow for some convenient shortcuts
        @_query = @_parse_query(@_query)

        # Check that the query is probably valid, and record the table and schema
        if misc.is_array(@_query)
            throw Error("must be a single query")
        tables = misc.keys(@_query)
        if misc.len(tables) != 1
            throw Error("must query only a single table")
        @_table = tables[0]
        if @_client.is_project()
            @_client_query = schema.SCHEMA[@_table].project_query
        else
            @_client_query = schema.SCHEMA[@_table].user_query
        if not misc.is_array(@_query[@_table])
            throw Error("must be a multi-document queries")
        @_schema = schema.SCHEMA[@_table]
        if not @_schema?
            throw Error("unknown schema for table #{@_table}")
        @_primary_keys = schema.client_db.primary_keys(@_table)
        # TODO: could put in more checks on validity of query here, using schema...
        for primary_key in @_primary_keys
            if not @_query[@_table][0][primary_key]?
                # must include each primary key in query
                @_query[@_table][0][primary_key] = null
        # Function @_to_key to extract primary key from object
        if @_primary_keys.length == 1
            # very common case
            pk = @_primary_keys[0]
            @_key = (obj) =>
                if not obj?
                    return
                if immutable.Map.isMap(obj)
                    return to_key(obj.get(pk))
                else
                    return to_key(obj[pk])
        else
            # compound primary key
            @_key = (obj) =>
                if not obj?
                    return
                v = []
                if immutable.Map.isMap(obj)
                    for pk in @_primary_keys
                        a = obj.get(pk)
                        if not a?
                            return
                        v.push(a)
                else
                    for pk in @_primary_keys
                        a = obj[pk]
                        if not a?
                            return
                        v.push(a)
                return to_key(v)

        # Which fields the user is allowed to set.
        @_set_fields = []
        # Which fields *must* be included in any set query
        @_required_set_fields = {}
        for field in misc.keys(@_query[@_table][0])
            if @_client_query?.set?.fields?[field]?
                @_set_fields.push(field)
            if @_client_query?.set?.required_fields?[field]?
                @_required_set_fields[field] = true

        # Is anonymous access to this table allowed?
        @_anonymous = !!@_schema.anonymous

    # Return map from keys that have changed along with how they changed, or undefined
    # if the value of local or the server hasn't been initialized
    _changes: =>
        if not @_value_server? or not @_value_local?
            return
        changed = {}
        @_value_local.map (new_val, key) =>
            old_val = @_value_server.get(key)
            if not new_val.equals(old_val)
                changed[key] = {new_val:new_val, old_val:old_val}
        return changed

    _save: (cb) =>
        if @__is_saving
            cb?("already saving")
        else
            @__is_saving = true
            @__save (err) =>
                @__is_saving = false
                cb?(err)

    __save: (cb) =>
        if @_state == 'closed'
            cb?("closed")
            return
        # console.log("_save('#{@_table}')")
        # Determine which records have changed and what their new values are.
        if not @_value_server?
            cb?("don't know server yet")
            return
        if not @_value_local?
            cb?("don't know local yet")
            return

        if not @_client_query.set?
            # Nothing to do -- can never set anything for this table.
            # There are some tables (e.g., stats) where the remote values
            # could change while user is offline, and the code below would
            # result in warnings.
            cb?()
            return

        changed = @_changes()
        at_start = @_value_local

        # Send our changes to the server.
        query = []
        saved_objs = []
        # sort so that behavior is more predictable = faster (e.g., sync patches are in
        # order); the keys are strings so default sort is fine
        for key in misc.keys(changed).sort()
            c = changed[key]
            obj = {}
            # NOTE: this may get replaced below with proper javascript, e.g., for compound primary key
            if @_primary_keys.length == 1
                obj[@_primary_keys[0]] = key
            else
                # unwrap compound primary key
                v = JSON.parse(key)
                i = 0
                for primary_key in @_primary_keys
                    obj[primary_key] = v[i]
                    i += 1

            for k in @_set_fields
                v = c.new_val.get(k)
                if v?
                    if @_required_set_fields[k] or not immutable.is(v, c.old_val?.get(k))
                        if immutable.Iterable.isIterable(v)
                            obj[k] = v.toJS()
                        else
                            obj[k] = v
            query.push({"#{@_table}":obj})
            saved_objs.push(obj)

        # console.log("sending #{query.length} changes: #{misc.to_json(query)}")
        if query.length == 0
            cb?()
            return
        #console.log("query=#{misc.to_json(query)}")
        #Use this to test fix_if_no_update_soon:
        #    if Math.random() <= .5
        #        query = []
        #@_fix_if_no_update_soon() # -disabled -- instead use "checking changefeed ids".
        @_client.query
            query   : query
            options : [{set:true}]  # force it to be a set query
            timeout : 30
            cb      : (err) =>
                if err
                    console.warn("_save('#{@_table}') error:", err)
                    if err == 'clock'
                        @_client.alert_message(type:'error', timeout:9999,  message:"Your computer's clock is or was off!  Fix it and **refresh your browser**.")
                    cb?(err)
                else
                    if @_state == 'closed'
                        # this can happen in case synctable is closed after _save is called but before returning from this query.
                        cb?("closed")
                        return
                    if not @_value_server? or not @_value_local?
                        # There is absolutely no possible way this can happen, since it was
                        # checked for above before the call, and these can only get set by
                        # the close method to undefined, which also sets the @_state to closed,
                        # so would get caught above.  However, evidently this **does happen**:
                        #   https://github.com/sagemathinc/cocalc/issues/1870
                        cb?("value_server and value_local must be set")
                        return
                    @emit('saved', saved_objs)
                    # success: each change in the query what committed successfully to the database; we can
                    # safely set @_value_server (for each value) as long as it didn't change in the meantime.
                    for k, v of changed
                        if immutable.is(@_value_server.get(k), v.old_val)  # immutable.is since either could be undefined
                            #console.log "setting @_value_server[#{k}] =", v.new_val?.toJS()
                            @_value_server = @_value_server.set(k, v.new_val)
                    if not at_start.equals(@_value_local)
                        # keep saving until @_value_local doesn't change *during* the save -- this means
                        # when saving stops that we guarantee there are no unsaved changes.
                        @_save(cb)
                    else
                        cb?()

    save: (cb) =>
        if @_state == 'closed'
            cb?("closed")
            return

        if @_state != 'connected'
            cb?("not connected")    # do not change this error message; it is assumed elsewhere.
            return

        @_save_debounce ?= {}

        if not @_value_server? or not @_value_local?
            @_connected_save_cbs ?= []
            @_connected_save_cbs.push(cb)
            return

        misc.async_debounce
            f        : (cb) =>
                misc.retry_until_success
                    f         : @_save
                    max_delay : 5000
                    max_time  : 30000
                    cb        : cb
            interval : @_debounce_interval
            state    : @_save_debounce
            cb       : cb

    # Handle an update of all records from the database.  This happens on
    # initialization, and also if we disconnect and reconnect.
    _update_all: (v) =>
        dbg = @dbg("_update_all")

        if @_state == 'closed'
            # nothing to do -- just ignore updates from db
            return

        if not v?
            console.warn("_update_all('#{@_table}') called with v=undefined")
            return

        @emit('before-change')
        # Restructure the array of records in v as a mapping from the primary key
        # to the corresponding record.
        x = {}
        for y in v
            x[@_key(y)] = y

        conflict = false

        # Figure out what to change in our local view of the database query result.
        if not @_value_local? or not @_value_server?
            dbg("easy case -- nothing has been initialized yet, so just set everything.")
            @_value_local = @_value_server = immutable.fromJS(x)
            first_connect = true
            changed_keys = misc.keys(x)  # of course all keys have been changed.
        else
            dbg("harder case -- everything has already been initialized.")
            changed_keys = []

            # DELETE or CHANGED:
            # First check through each key in our local view of the query
            # and if the value differs from what is in the database (i.e.,
            # what we just got from DB), make that change.
            # (Later we will possibly merge in the change
            # using the last known upstream database state.)
            @_value_local.map (local, key) =>
                if x[key]?
                    # update value we have locally
                    if @_handle_new_val(x[key], changed_keys)
                        conflict = true
                else
                    # This is a value defined locally that does not exist
                    # on the remote serve.   It could be that the value
                    # was deleted when we weren't connected, in which case
                    # we should delete the value we have locally.  On the
                    # other hand, maybe the local value was newly set
                    # while we weren't connected, so we know it but the
                    # backend server doesn't, which case we should keep it,
                    # and set conflict=true, so it gets saved to the backend.

                    if @_value_local.get(key).equals(@_value_server.get(key))
                        # The local value for this key was saved to the backend before
                        # we got disconnected, so there's definitely no need to try
                        # keep it around, given that the backend no longer has it
                        # as part of the query.  CRITICAL: This doesn't necessarily mean
                        # the value was deleted from the database, but instead that
                        # it doesn't satisfy the synctable query, e.g., it isn't one
                        # of the 150 most recent file_use notifications, or it isn't
                        # a patch that is at least as new as the newest snapshot.
                        #console.log("removing local value: #{key}")
                        @_value_local = @_value_local.delete(key)
                        changed_keys.push(key)
                    else
                        conflict = true

            # NEWLY ADDED:
            # Next check through each key in what's on the remote database,
            # and if the corresponding local key isn't defined, set its value.
            # Here we are simply checking for newly added records.
            for key, val of x
                if not @_value_local.get(key)?
                    @_value_local = @_value_local.set(key, immutable.fromJS(val))
                    changed_keys.push(key)

        # It's possibly that nothing changed (e.g., typical case on reconnect!) so we check.
        # If something really did change, we set the server state to what we just got, and
        # also inform listeners of which records changed (by giving keys).
        #console.log("update_all: changed_keys=", changed_keys)
        if changed_keys.length != 0
            @_value_server = immutable.fromJS(x)
            @emit_change(changed_keys)
        else if first_connect
            # First connection and table is empty.
            @emit_change(changed_keys)
        if conflict
            @save()

    # Apply one incoming change from the database to the in-memory
    # local synchronized table
    _update_change: (change) =>
        #console.log("_update_change", change)
        if @_state == 'closed'
            # We might get a few more updates even after
            # canceling the changefeed, so we just ignore them.
            return
        if not @_value_local?
            console.warn("_update_change(#{@_table}): tried to call _update_change even though local not yet defined (ignoring)")
            return
        if not @_value_server?
            console.warn("_update_change(#{@_table}): tried to call _update_change even though set not yet defined (ignoring)")
            return
        if DEBUG
            console.log("_update_change('#{@_table}'): #{misc.to_json(change)}")
        @emit('before-change')
        changed_keys = []
        conflict = false
        if change.new_val?
            conflict = @_handle_new_val(change.new_val, changed_keys)

        if change.old_val? and @_key(change.old_val) != @_key(change.new_val)
            # Delete a record (TODO: untested)
            key = @_key(change.old_val)
            @_value_local = @_value_local.delete(key)
            @_value_server = @_value_server.delete(key)
            changed_keys.push(key)

        #console.log("update_change: changed_keys=", changed_keys)
        if changed_keys.length > 0
            #console.log("_update_change: change")
            @emit_change(changed_keys)
            if conflict
                @save()

    _handle_new_val: (val, changed_keys) =>
        key       = @_key(val)
        new_val   = immutable.fromJS(val)
        local_val = @_value_local.get(key)
        conflict  = false
        if not new_val.equals(local_val)
            #console.log("change table='#{@_table}': #{misc.to_json(local_val?.toJS())} --> #{misc.to_json(new_val.toJS())}") if @_table == 'patches'
            if not local_val?
                @_value_local = @_value_local.set(key, new_val)
                changed_keys.push(key)
            else
                server = @_value_server.get(key)
                # Set in @_value_local every key whose value changed between new_val and server; basically, we're
                # determining and applying the "patch" from upstream, even though it was sent as a complete record.
                # We can compute the patch, since we know the last server value.
                new_val.map (v, k) =>
                    if not immutable.is(v, server?.get(k))
                        local_val = local_val.set(k, v)
                        #console.log("#{@_table}: set #{k} to #{v}")
                server?.map (v, k) =>
                    if not new_val.has(k)
                        local_val = local_val.delete(k)
                if not local_val.equals(@_value_local.get(key))
                    @_value_local = @_value_local.set(key, local_val)
                    changed_keys.push(key)
                if not local_val.equals(new_val)
                    #console.log("#{@_table}: conflict! ", local_val, new_val) if @_table == 'patches'
                    @emit('conflict', {new_val:new_val, old_val:local_val})
                    conflict = true
        @_value_server = @_value_server.set(key, new_val)
        return conflict

    # obj is an immutable.js Map without the primary key
    # set.  If the database schema defines a way to compute
    # the primary key from other keys, try to use it here.
    # This function returns the computed primary key if it works,
    # and returns undefined otherwise.
    _computed_primary_key: (obj) =>
        if @_primary_keys.length == 1
            f = @_client_query.set.fields[@_primary_keys[0]]
            if typeof(f) == 'function'
                return f(obj.toJS(), schema.client_db)
            else
                return
        else
            v = []
            for pk in @_primary_keys
                f = @_client_query.set.fields[pk]
                if typeof(f) == 'function'
                    v.push(f(obj.toJS(), schema.client_db))
                else
                    return
            return v

    # Changes (or creates) one entry in the table.
    # The input field changes is either an Immutable.js Map or a JS Object map.
    # If changes does not have the primary key then a random record is updated,
    # and there *must* be at least one record.  Exception: computed primary
    # keys will be computed (see stuff about computed primary keys above).
    # The second parameter 'merge' can be one of three values:
    #   'deep'   : (DEFAULT) deep merges the changes into the record, keep as much info as possible.
    #   'shallow': shallow merges, replacing keys by corresponding values
    #   'none'   : do no merging at all -- just replace record completely
    # The cb is called with cb(err) if something goes wrong.
    # Returns the updated value.
    set: (changes, merge, cb) =>
        if @_state == 'closed'
            # Attempting to set on a closed table is dangerous since any data set *will* be
            # silently lost.  So spit out a visible warning.
            console.warn("WARNING: attempt to do a set on a closed table: '#{@_table}', #{misc.to_json(@_query)}")
            cb?("closed")
            return

        if not immutable.Map.isMap(changes)
            changes = immutable.fromJS(changes)
        if not @_value_local?
            @_value_local = immutable.Map({})

        if not merge?
            merge = 'deep'
        else if typeof(merge) == 'function'
            cb = merge
            merge = 'deep'

        if not immutable.Map.isMap(changes)
            cb?("type error -- changes must be an immutable.js Map or JS map")
            return

        if DEBUG
            console.log("set('#{@_table}'): #{misc.to_json(changes.toJS())}")

        # Ensure that each key is allowed to be set.
        if not @_client_query.set?
            cb?("users may not set #{@_table}")
            return
        can_set = @_client_query.set.fields
        try
            changes.map (v, k) => if (can_set[k] == undefined) then throw Error("users may not set #{@_table}.#{k}")
        catch e
            cb?(e)
            return

        # Determine the primary key's value
        id = @_key(changes)
        if not id?
            # attempt to compute primary key if it is a computed primary key
            id0 = @_computed_primary_key(changes)
            id = to_key(id0)
            if not id? and @_primary_keys.length == 1
                # use a "random" primary key from existing data
                id0 = id = @_value_local.keySeq().first()
            if not id?
                cb?("must specify primary key #{@_primary_keys.join(',')}, have at least one record, or have a computed primary key")
                return
            # Now id is defined
            if @_primary_keys.length == 1
                changes = changes.set(@_primary_keys[0], id0)
            else
                i = 0
                for pk in @_primary_keys
                    changes = changes.set(pk, id0[i])
                    i += 1

        # Get the current value
        cur  = @_value_local.get(id)
        if not cur?
            # No record with the given primary key.  Require that all the @_required_set_fields
            # are specified, or it will become impossible to sync this table to the backend.
            for k,_ of @_required_set_fields
                if not changes.get(k)?
                    cb?("must specify field '#{k}' for new records")
                    return
            # If no current value, then next value is easy -- it equals the current value in all cases.
            new_val = changes
        else
            # Use the appropriate merge strategy to get the next val.  Fortunately these are all built
            # into immutable.js!
            switch merge
                when 'deep'
                    new_val = cur.mergeDeep(changes)
                when 'shallow'
                    new_val = cur.merge(changes)
                when 'none'
                    new_val = changes
                else
                    cb?("merge must be one of 'deep', 'shallow', 'none'")
                    return
        # If something changed, then change in our local store, and also kick off a save to the backend.
        if not immutable.is(new_val, cur)
            @_value_local = @_value_local.set(id, new_val)
            @save(cb)
            @emit_change([id])  # CRITICAL: other code assumes the key is *NOT* sent with this change event!
        else
            cb?()

        return new_val

    close: =>
        if @_state == 'closed'
            # already closed
            return
        # decrement the reference to this synctable
        if global_cache_decref(@)
            # close: not zero -- so don't close it yet -- still in use by multiple clients
            return
        @_client.removeListener('disconnected', @_disconnected)
        # do a last attempt at a save (so we don't lose data), then really close.
        @_save()  # this will synchronously construct the last save and send it
        # The moment the sync part of @_save is done, we remove listeners and clear
        # everything up.  It's critical that as soon as @close is called that there
        # be no possible way any further connect events (etc) can make this SyncTable
        # do anything!!  That finality assumption is made elsewhere (e.g in smc-project/client.coffee)
        @removeAllListeners()
        if @_id?
            @_client.query_cancel(id:@_id)
            delete @_id
        @_state = 'closed'
        delete @_value_local
        delete @_value_server

    # wait until some function of this synctable is truthy
    # (this might be exactly the same code as in the postgres-synctable.coffee SyncTable....)
    wait: (opts) =>
        opts = defaults opts,
            until   : required     # waits until "until(@)" evaluates to something truthy
            timeout : 30           # in *seconds* -- set to 0 to disable (sort of DANGEROUS, obviously.)
            cb      : required     # cb(undefined, until(@)) on success and cb('timeout') on failure due to timeout; cb('closed') if closed
        if @_state == 'closed'
            # instantly fail -- table is closed so can't wait for anything
            opts.cb("closed")
            return
        x = opts.until(@)
        if x
            opts.cb(undefined, x)  # already true
            return
        fail_timer = undefined
        f = =>
            x = opts.until(@)
            if x
                @removeListener('change', f)
                if fail_timer? then clearTimeout(fail_timer)
                opts.cb(undefined, x)
        @on('change', f)
        if opts.timeout
            fail = =>
                @removeListener('change', f)
                opts.cb('timeout')
            fail_timer = setTimeout(fail, 1000*opts.timeout)
        return

synctables = {}

# for debugging; in particular, verify that synctables are freed.
# Do not leave in production; could be slight security risk.
## window?.synctables = synctables

exports.sync_table = (query, options, client, debounce_interval=2000, throttle_changes=undefined, use_cache=true) ->
    cache_key = json_stable_stringify(query:query, options:options, debounce_interval:debounce_interval, throttle_changes:throttle_changes)
    if not use_cache
        return new SyncTable(query, options, client, debounce_interval, throttle_changes, cache_key)

    S = synctables[cache_key]
    if S?
        if S._state == 'connected'
            # same behavior as newly created synctable
            async.nextTick () ->
                if S._state == 'connected'
                    S.emit('connected')
        S._reference_count += 1
        return S
    else
        S = synctables[cache_key] = new SyncTable(query, options, client, debounce_interval, throttle_changes, cache_key)
        S._reference_count = 1
        return S

global_cache_decref = (S) ->
    if S._reference_count?
        S._reference_count -= 1
        if S._reference_count <= 0
            delete synctables[S._cache_key]
            return false  # not in use
        else
            return true   # still in use

#if window?
#    window.synctables = synctables



###
Various mock clients for unit testing

Events:
    - disconnected
    - connected
###

class exports.TestBrowserClient1 extends EventEmitter

    is_project: =>
        return false

    is_connected: =>
        return true

    is_signed_in: =>
        return true

    dbg: =>
        return =>

    query_cancel: =>

    query: (opts) =>
        opts = defaults opts,
            query   : required
            changes : undefined
            options : undefined    # if given must be an array of objects, e.g., [{limit:5}]
            timeout : 30
            cb      : undefined
        @emit 'query', opts