const fp = require('lodash/fp')
const http = require('http')
const socketio = require('socket.io')

const r = require('./rethink')

console.log('starting server...')
let server = http.createServer()
let io = socketio(server)

const getInitialMessagesPromise = channelName => (
  r.table('messages')
  .orderBy({ index: r.desc('timestamp') })
  .filter({ to: channelName })
  .limit(100)
)

const getMessagesBeforePromise = (channelName, timestamp) => (
  r.table('messages')
    .orderBy({ index: r.desc('timestamp') })
    .filter({ to: channelName })
    .filter(r.row('timestamp').le(r.ISO8601(timestamp)))
    .limit(101)
)

const getMessagesAfterPromise = (channelName, timestamp) => (
  r.table('messages')
    .orderBy({ index: r.desc('timestamp') })
    .filter({ to: channelName })
    .filter(r.row('timestamp').ge(r.ISO8601(timestamp)))
)

const subscribeToChannels = () => client => {
  r.table('channels').changes({ includeInitial: true }).run()
    .then(feed => {
      feed.each((err, change) => {
        if (err) {
          console.error('error from channels feed', err)
          return
        }

        client.emit('action', {
          type: 'client/CHANNEL_CHANGE',
          payload: change,
        })
      })
    })
}

const loadMessages = ({ channelName = null, before = null, after = null }) => client => {
  if (!channelName) {
    return
  }

  let messagePromise
  if (before) {
    messagePromise = getMessagesBeforePromise(channelName, before)
  } else if (after) {
    messagePromise = getMessagesAfterPromise(channelName, after)
  } else {
    messagePromise = getInitialMessagesPromise(channelName)
  }

  messagePromise.then(messages => {
    client.emit('action', {
      type: 'client/LOADED_MESSAGES',
      payload: {
        channelName,
        before,
        after,
        messages: fp.reverse(messages),
      }
    })
  })
}

const subscribeToMessages = ({ channelName = null, timestamp = null }) => client => {
  if (!channelName || !timestamp) {
    return
  }

  r.table('messages')
    .orderBy({ index: r.desc('timestamp') })
    .filter({ to: channelName })
    .filter(r.row('timestamp').gt(r.ISO8601(timestamp)))
    .changes({ includeInitial: true })
    .run()
    .then(feed => {
      feed.each((err, change) => {
        client.emit('action', {
          type: 'client/MESSAGE_CHANGE',
          payload: change,
        })
      })
    })
}

const ACTIONS = {
  'server/SUBSCRIBE_TO_CHANNELS': subscribeToChannels,
  'server/LOAD_MESSAGES': loadMessages,
  'server/SUBSCRIBE_TO_MESSAGES': subscribeToMessages,
}

io.on('connection', client => {
  client.on('action', ({ type, payload = null }) => {
    if (!ACTIONS[type]) {
      console.error(`Unknown action: ${type}`)
      return
    }

    ACTIONS[type](payload || {})(client)
  })
})

server.listen(3001)

@@ -298,35 +298,85 @@
 )%0A  
-.orderBy(%7B index: r.desc('t
+  .between(%5BchannelName, r.minval%5D, %5BchannelName, r.maxval%5D, %7B index: 'toAndT
 imes
@@ -384,44 +384,64 @@
 amp'
-)
  %7D)%0A  
-.filter(%7B to: channelName
+  .orderBy(%7B index: r.desc('toAndTimestamp')
  %7D)%0A
+  
   .l
@@ -530,32 +530,101 @@
 ble('messages')%0A
+    .between(r.minval, r.ISO8601(timestamp), %7B index: 'timestamp' %7D)%0A
     .orderBy(%7B i
@@ -689,65 +689,8 @@
  %7D)%0A
-    .filter(r.row('timestamp').le(r.ISO8601(timestamp)))%0A
     
@@ -768,34 +768,154 @@
  =%3E (%0A  
-r.table('messages'
+// TODO: Add limit and retry requests on client.%0A  r.table('messages')%0A    .between(r.ISO8601(timestamp), r.maxval, %7B index: 'timestamp' %7D
 )%0A    .o
@@ -990,65 +990,8 @@
  %7D)%0A
-    .filter(r.row('timestamp').ge(r.ISO8601(timestamp)))%0A
 )%0A%0Ac
@@ -2171,24 +2171,93 @@
 'messages')%0A
+    .between(r.ISO8601(timestamp), r.maxval, %7B index: 'timestamp' %7D)%0A
     .orderBy
@@ -2326,65 +2326,8 @@
  %7D)%0A
-    .filter(r.row('timestamp').gt(r.ISO8601(timestamp)))%0A
     

Use RethinkDB indexes because RAM isn't infinite