 
 sync = (connection) ->
   loop
-    syncToken = SyncToken.findOne().syncToken
+    syncToken = SyncState.findOne(SYNC_TOKEN_ID).syncToken
     result = connection.call 'syncNewPackageData', syncToken
 
     if result.resetData

        Versions.find(packageName: oldPackage.packageName).forEach (document, index, cursor) ->
          insertLatestPackage document

    initializing = false

    console.log "Latest packages observe initialized"

    connection = DDP.connect 'packages.meteor.com'

    Defaults = new Mongo.Collection 'defaults', connection
    Changes = new Mongo.Collection 'changes', connection

    connection.subscribe 'defaults', ->
      try
        SyncToken.insert
          _id: 'syncToken'
          syncToken: Defaults.findOne().syncToken
      catch error
        throw error unless /E11000 duplicate key error index:.*SyncToken\.\$_id/.test error.err

      connection.subscribe 'changes', ->
        Changes.find().observe
          added: (document) ->
            sync connection
          changed: (document, oldDocument) ->
            sync connection

  .run()

  # TODO: Slight race condition here. There might be another document inserted between removal and this insertion.
  LatestPackages.insert document

latestPackagesObserve = ->
  console.log "Starting latest packages observe"

  try
    # We try to create the initial document.
    SyncState.insert
      _id: LAST_UPDATED_ID
      lastUpdated: null
  catch error
    throw error unless /E11000 duplicate key error index:.*SyncState\.\$_id/.test error.err

  timeoutHandle = null
  newestLastUpdated = null

  # Update sync state after 30 seconds of no updates. This assures that if there was a series of observe
  # callbacks called, we really processed them all. Otherwise we might set state but program might
  # terminate before we had a chance to process all observe callbacks. Which will mean that those
  # packages from pending observe callbacks will not be processed the next time the program runs.
  updateSyncState = (newLastUpdated) ->
    # We allow that in a series of observe callbacks the order of last updated timestamps is
    # not monotonic. In the case that last updated timestamps are not monotonic between
    # series of observe callbacks, we will have to (and do) restart the observe.
    if not newestLastUpdated or newestLastUpdated < newLastUpdated
      newestLastUpdated = newLastUpdated

    Meteor.clearTimeout timeoutHandle
    timeoutHandle = Meteor.setTimeout ->
      lastUpdated = newestLastUpdated
      newestLastUpdated = null

      SyncState.update
        _id: LAST_UPDATED_ID
      ,
        $set:
          lastUpdated: lastUpdated
    ,
      30 * 1000 # ms

  observeHandle = null
  currentLastUpdated = null

  startObserve = ->
    observeHandle?.stop()
    observeHandle = null


    if currentLastUpdated
      query =
        lastUpdated:
          $gte: currentLastUpdated
    else
      query = {}

    observeHandle = Versions.find(query).observeChanges
      added: (id, fields) ->
        insertLatestPackage _.extend {_id: id}, fields

        updateSyncState fields.lastUpdated.valueOf()

      changed: (id, fields) ->
        # Will possibly not update anything, if the change is for an older package.
        LatestPackages.update id, fieldsToModifier fields

        updateSyncState fields.lastUpdated.valueOf() if 'lastUpdated' of fields

      removed: (id) ->
        oldPackage = LatestPackages.findOne id

