hyper-social / bridge-email / lib / keet-client.js

import http from 'http'
import net from 'net'

class KeetClient {
  constructor (opts = {}) {
    this.host = opts.host || 'localhost'
    this.port = opts.port || 3117
    this._watching = false
    this._watchAbort = null
  }

  _rawRequest (method, path, body) {
    return new Promise((resolve, reject) => {
      const socket = net.connect(this.port, this.host)
      let payload = ''

      if (body) {
        payload = JSON.stringify(body)
      }

      socket.on('connect', () => {
        let req = `${method} ${path} HTTP/1.1\r\n`
        req += `Host: ${this.host}:${this.port}\r\n`
        if (payload) {
          req += `Content-Type: application/json\r\n`
          req += `Content-Length: ${Buffer.byteLength(payload)}\r\n`
        }
        req += `Connection: close\r\n`
        req += `\r\n`
        if (payload) req += payload
        socket.write(req)
      })

      let data = ''
      socket.on('data', (chunk) => { data += chunk.toString() })

      socket.on('end', () => {
        const idx = data.indexOf('\r\n\r\n')
        if (idx === -1) return reject(new Error('No HTTP body'))
        const jsonBody = data.slice(idx + 4)

        try {
          const parsed = JSON.parse(jsonBody)
          if (!parsed.ok) reject(new Error(parsed.error || 'RPC error'))
          else resolve(parsed.data)
        } catch {
          reject(new Error('Invalid JSON: ' + jsonBody.slice(0, 200)))
        }
      })

      socket.on('error', reject)
      socket.setTimeout(10000, () => {
        socket.destroy(new Error('Request timeout'))
      })
    })
  }

  rpc (method, params = {}) {
    return this._rawRequest('POST', '/rpc', { method, params })
  }

  async send (roomId, text) {
    return this.rpc('send', { roomId, text })
  }

  async messages (roomId, limit = 20) {
    return this._rawRequest('GET', `/rpc/messages/${roomId}?limit=${limit}`)
  }

  async rooms () {
    return this._rawRequest('GET', '/rpc/rooms')
  }

  async identity () {
    return this._rawRequest('GET', '/rpc/identity')
  }

  async status () {
    return this._rawRequest('GET', '/rpc/status')
  }

  watch (roomId, onMessage) {
    if (this._watching) return
    this._watching = true
    this._doWatch(roomId, onMessage)
  }

  _doWatch (roomId, onMessage) {
    if (!this._watching) return

    const req = http.request({
      hostname: this.host,
      port: this.port,
      path: `/rpc/watch/${roomId}`,
      method: 'GET',
      headers: { Accept: 'text/event-stream' }
    }, (res) => {
      let buffer = ''

      res.on('data', (chunk) => {
        buffer += chunk.toString()
        const events = buffer.split('\n\n')
        buffer = events.pop()

        for (const event of events) {
          if (!event.trim()) continue
          const lines = event.split('\n')
          let type = 'message'
          let data = null

          for (const line of lines) {
            if (line.startsWith('event: ')) type = line.slice(7)
            if (line.startsWith('data: ')) {
              try { data = JSON.parse(line.slice(6)) } catch {}
            }
          }

          if (type === 'message' && data) {
            onMessage(data)
          }
        }
      })

      res.on('end', () => {
        if (this._watching) {
          setTimeout(() => this._doWatch(roomId, onMessage), 2000)
        }
      })

      res.on('error', () => {
        if (this._watching) {
          setTimeout(() => this._doWatch(roomId, onMessage), 5000)
        }
      })
    })

    req.on('error', () => {
      if (this._watching) {
        setTimeout(() => this._doWatch(roomId, onMessage), 5000)
      }
    })

    this._watchAbort = req
    req.end()
  }

  stopWatch () {
    this._watching = false
    if (this._watchAbort) {
      this._watchAbort.destroy()
      this._watchAbort = null
    }
  }
}

export { KeetClient }