   projection
 
 # TODO: Should we add retry?
-catchErrors = (f) ->
-  return (args...) ->
+observerCallback = (f) ->
+  return (id, args...) ->
     try
-      f args...
+      # We call f only if the first character of id is in PREFIX.
+      # By that we allow each instance to operate only on a subset
+      # of documents, allowing simple coordination while scaling.
+      f id, args... if id[0] in PREFIX
     catch e
       Log.error "PeerDB exception: #{ e }: #{ util.inspect args, depth: 10 }"
       Log.debug e.stack

    initializing = true

    observers =
      added: catchErrors (id, fields) =>
        @_sourceUpdated id, fields if updateAll or not initializing

    unless updateAll
      observers.changed = catchErrors (id, fields) =>
        @_sourceUpdated id, fields

    handle = @Meta.collection.find({}, fields: sourceFields).observeChanges observers

      currentSchema = newSchema
      currentName = newName

    # We do not check for not migrated documents if we are skipping migrations
    unless process.env.PEERDB_SKIP_MIGRATIONS
      # For all those documents which lack schema information we assume they have the last schema
      @Meta.collection.update
        _schema:
          $exists: false
      ,
        $set:
          _schema: currentSchema
      ,
        multi: true

      notMigrated = _.pluck @Meta.collection.find(
        _schema:
          $ne: currentSchema
      ,
        fields:
          _id: 1
      ).fetch(), '_id'

      throw new Error "Not all documents migrated to the latest schema version (#{ currentSchema }): #{ notMigrated }" if notMigrated.length

    @Meta.schema = currentSchema

