const assign = require('xtend/mutable')
const Observable = require('zen-observable')
const encodeQueryString = require('./encodeQueryString')
const validators = require('../validators')
const pick = require('../util/pick')
const defaults = require('../util/defaults')

const EventSource = typeof window !== 'undefined' && window.EventSource
  ? window.EventSource // Native browser EventSource
  : require('@sanity/eventsource') // Node.js, IE etc

// Temporarily(?) needed because the node eventsource doesn't expose removeEventListener
const removeListener = (evtSrc, evt, handler) => {
  if (evtSrc.removeEventListener) {
    evtSrc.removeEventListener(evt, handler, false)
  } else {
    evtSrc.removeListener(evt, handler)
  }
}

const possibleOptions = ['includePreviousRevision', 'includeResult']
const defaultOptions = {
  includeResult: true
}

module.exports = function listen(query, params, opts = {}) {
  const options = defaults(opts, defaultOptions)
  const listenOpts = pick(options, possibleOptions)
  const qs = encodeQueryString({query, params, options: listenOpts})
  const dataset = validators.hasDataset(this.clientConfig)
  const uri = this.getUrl(`/data/listen/${dataset}${qs}`)
  const token = this.clientConfig.token
  const listenFor = options.events ? options.events : ['document']
  const shouldEmitReconnect = listenFor.indexOf('reconnect') !== -1

  const es = new EventSource(uri, assign(
    {withCredentials: true},
    token ? {headers: {'Sanity-Token': token}} : {}
  ))

  return new Observable(observer => {
    es.addEventListener('error', onError, false)
    es.addEventListener('disconnect', onDisconnect, false)
    listenFor.forEach(type => es.addEventListener(type, onMessage, false))

    function onError(err) {
      if (err.data) {
        observer.error(cooerceError(err))
      } else if (es.readyState === EventSource.CLOSED) {
        observer.complete()
      } else if (es.readyState === EventSource.CONNECTING) {
        emitReconnect()
      }
    }

    function onMessage(evt) {
      const event = parseEvent(evt)
      return event instanceof Error
        ? observer.error(event)
        : observer.next(event)
    }

    function onDisconnect(evt) {
      observer.complete()
      unsubscribe()
    }

    function unsubscribe() {
      listenFor.forEach(type => removeListener(es, type, onMessage))
      removeListener(es, 'error', onError)
      removeListener(es, 'disconnect', onDisconnect)
      es.close()
    }

    function emitReconnect() {
      if (shouldEmitReconnect) {
        observer.next({type: 'reconnect'})
      }
    }

    return unsubscribe
  })
}

function parseEvent(event) {
  try {
    const data = (event.data && JSON.parse(event.data)) || {}
    return assign({type: event.type}, data)
  } catch (err) {
    return err
  }
}

function cooerceError(err) {
  if (err instanceof Error) {
    return err
  }

  const evt = parseEvent(err)
  return evt instanceof Error
    ? evt
    : new Error(evt.error || 'Unknown listener error')
}

@@ -1295,16 +1295,16 @@
 : %5B'
-document
+mutation
 '%5D%0A 
@@ -1583,24 +1583,87 @@
 ror, false)%0A
+    es.addEventListener('channelError', onChannelError, false)%0A
     es.addEv
@@ -1802,27 +1802,24 @@
 ion onError(
-err
 ) %7B%0A      if
@@ -1823,79 +1823,8 @@
  if 
-(err.data) %7B%0A        observer.error(cooerceError(err))%0A      %7D else if 
 (es.
@@ -1980,32 +1980,114 @@
 %0A      %7D%0A    %7D%0A%0A
+    function onChannelError(err) %7B%0A      observer.error(cooerceError(err))%0A    %7D%0A%0A
     function onM
@@ -2469,16 +2469,73 @@
 nError)%0A
+      removeListener(es, 'channelError', onChannelError)%0A
       re
@@ -3088,16 +3088,16 @@
   ? evt%0A
-
     : ne
@@ -3116,16 +3116,31 @@
 error %7C%7C
+ evt.message %7C%7C
  'Unknow

Update listen to match new event names