First commit

.gitignore

@@ -0,0 +1,3 @@
+node_modules
+*.swp
+chats.db

main.js

@@ -0,0 +1,112 @@
+var html = require('yo-yo')
+var through = require('through2')
+var ws = require('websocket-stream')('ws://localhost:5000' )
+
+var root = document.body.appendChild(document.createElement('div'))
+var state = {
+  username: 'hey_joe',
+  rooms: []
+}
+
+update(state)
+
+ws.pipe(through(function (buff, enc, next) {
+  var msg = buff.toString()
+  if (/create_chat/.test(msg)) {
+    var room = JSON.parse(msg.split('!')[1])
+    state.rooms.push(room)
+  } else {
+    var data = msg.split('!')
+    var chat_id = data[1]
+    var message = data[2]
+
+    var room = state.rooms.find(function (room) {
+      return room.id === chat_id
+    })
+
+    if (room) room.messages.push(message)
+  }
+
+  update(state)
+  next()
+}))
+
+ws.write('connection!' + state.username)
+
+fetch('http://localhost:5000/chats?user=' + state.username)
+  .then(function (res) {
+    return res.json()
+  })
+  .then(function updateRooms (rooms) {
+    state.rooms = rooms || []
+
+    update(state)
+  })
+
+function update(state) {
+  html.update(root, html`<div>
+    <h1>Chat rooms</h1>
+    <h2>Current user ${state.username}</h2>
+
+    <form onsubmit=${handleChangeUser}>
+      <h3>Change user</h3>
+      <input name="username" >
+    </form>
+
+    <button onclick=${handleCreateChat}>Create chat</button>
+
+    ${state.rooms.map((room) => html`
+      <div>
+        <h3>Chat - ${room.id}</h3>
+
+        <form onsubmit=${handleAddUser} data-chat=${room.id}>
+          <h4>Add user to the chat</h4>
+          <input name="username" />
+        </form>
+
+        <pre>${room.messages.join('\n')}</pre>
+
+        <form onsubmit=${handleSendMessage} data-chat=${room.id}>
+          <input name="message">
+        </form>
+      </div>
+    `)}
+
+   </div>`)
+
+  function handleChangeUser (evt) {
+    evt.preventDefault()
+    var username = this.elements.username.value
+
+    state.username = username
+    update(state)
+    this.reset()
+  }
+
+  function handleAddUser (evt) {
+    evt.preventDefault()
+    var username = this.elements.username.value
+    var chat_id = this.getAttribute('data-chat')
+
+    if (!username) return
+
+    ws.write('add_user!' + username + '!' + chat_id)
+    this.reset()
+  }
+
+  function handleCreateChat (evt) {
+    fetch('http://localhost:5000/chats?user=' + state.username, { method: 'POST' })
+  }
+
+  function handleSendMessage (evt) {
+    evt.preventDefault()
+    var username = state.username
+    var chat_id = this.getAttribute('data-chat')
+    var message = this.elements.message.value
+
+    if (!message) return
+
+    ws.write('add_message!' + username + '!' + chat_id + '!' + message)
+    this.reset()
+  }
+}

models/chats.js

@@ -0,0 +1,117 @@
+const level = require('level')
+const onend = require('end-of-stream')
+const through = require('through2')
+const to = require('to2')
+const crypto = require('crypto')
+
+const db = level('chats.db')
+
+module.exports = {
+  getAllChatsByUsername,
+  createChat,
+  addUserIntoChat,
+  getAllUsersIntoChat,
+  getUser,
+  addUser
+}
+
+function getAllChatsByUsername (username) {
+  return new Promise(function (resolve, reject) {
+    const chats = []
+    const stream = db.createReadStream({
+      gt: 'user!' + username + '!chat!',
+      lt: 'user!' + username + '!chat!~'
+    }).pipe(to.obj(function (row, enc, next) {
+      const chat_id = row.key.split('!')[3]
+      const chat = {
+        id: chat_id,
+        messages: []
+      }
+
+      const messagesStream = db.createReadStream({
+        gt: 'user!' + username + '!chat!' + chat_id + '!messages!',
+        lt: 'user!' + username + '!chat!' + chat_id + '!messages!~'
+      }).pipe(to.obj(function (row, enc, next) {
+        chat.messages.push(row.value)
+        next()
+      }))
+
+      onend(messagesStream, function () {
+        chats.push(chat)
+        next()
+      })
+    }))
+
+    stream.on('error', reject)
+
+    onend(stream, function () {
+      resolve(chats)
+    })
+  })
+}
+
+function getAllUsersIntoChat (chat_id) {
+  return new Promise(function (resolve) {
+    const users = []
+
+    onend(db.createReadStream({
+      gt: 'chat!' + chat_id + '!user!',
+      lt: 'chat!' + chat_id + '!user!~'
+    }).pipe(to.obj(function (row, enc, next) {
+      var username = row.key.split('!')[3]
+      users.push(username)
+
+      next()
+    })), function (error) {
+      if (error) return reject(error)
+      resolve(users)
+    })
+  })
+}
+
+function createChat(username) {
+  const chat_id = crypto.randomBytes(16).toString('hex')
+  return addUserIntoChat(chat_id, username)
+}
+
+function addUserIntoChat(chat_id, username) {
+  return new Promise(function (resolve, reject) {
+    const batch = [
+      { key: 'user!' + username + '!chat!' + chat_id, value: chat_id },
+      { key: 'chat!' + chat_id + '!user!' + username, value: chat_id }
+    ]
+
+    db.batch(batch, function (error) {
+      if (error) return reject(error)
+
+      const chat = {
+        id: chat_id,
+        messages: ['Welcome to the new chat ' + username]
+      }
+
+      resolve(chat)
+    })
+  })
+}
+
+function getUser (username) {
+  return new Promise(function (resolve, reject) {
+    db.get('user!' + username, function (error, value) {
+      if (error || !value) {
+        reject(error)
+      } else {
+        resolve(value)
+      }
+    })
+  })
+}
+
+function addUser (username) {
+  return new Promise(function (resolve, reject) {
+    db.put('user!' + username, username, function (error) {
+      if (error) return reject(error)
+
+      resolve(username)
+    })
+  })
+}

package.json

@@ -0,0 +1,6 @@
+{
+  "dependencies": {
+    "to2": "^1.0.0",
+    "websocket-stream": "^4.0.0"
+  }
+}

server.js

@@ -0,0 +1,86 @@
+var http = require('http')
+var websock = require('websocket-stream')
+var onend = require('end-of-stream')
+var through = require('through2')
+
+var chatModel = require('./models/chats.js')
+
+var server = http.createServer(function (req, res) {
+  res.setHeader('Access-Control-Allow-Origin', 'http://localhost:9966')
+
+  if (/GET/i.test(req.method) && /chats/.test(req.url)) {
+    var username = require('url').parse(req.url).query.split('=')[1]
+
+    chatModel.getAllChatsByUsername(username)
+      .then(function (chats) {
+        res.setHeader('Content-Type', 'application/json')
+        res.end(JSON.stringify(chats))
+      })
+  } else if (/POST/i.test(req.method) && /chats/.test(req.url)) {
+    var username = require('url').parse(req.url).query.split('=')[1]
+
+    chatModel.createChat(username)
+      .then(function (chat) {
+        res.setHeader('Content-Type', 'application/json')
+        res.end(JSON.stringify(chat))
+      })
+  }
+})
+
+var count = 0
+var streamByUser = {}
+
+var ws = websock.createServer({ server: server }, function (stream) {
+  stream.pipe(handleChat(stream))
+
+  onend(stream, function () {
+    console.log('DISCONNECTED')
+  })
+})
+
+function handleChat (stream) {
+  return through(function (buff, enc, next) {
+    // type!username!chat_id!message
+    var data = buff.toString().split('!')
+    var typeOfMessage = data[0]
+    var username = data[1]
+    var chat_id = data[2]
+    var message = data[3]
+
+    var runMessage = {
+      connection: function connection(username) {
+        chatModel
+          .getUser(username)
+          .catch(function () {
+            return chatModel.addUser(username)
+          })
+          .then(function () {
+            streamByUser[username] = stream
+            stream.write('connected!' + username + '\n')
+          })
+      },
+      add_user: function add_user(username, chat_id) {
+        chatModel.addUserIntoChat(chat_id, username)
+      },
+      add_message: function add_message(username, chat_id, message) {
+        chatModel.getAllUsersIntoChat(chat_id)
+          .then(function (users) {
+            users
+              .map(user => ({
+                name: user,
+                stream: streamByUser[user]
+              }))
+              .filter(user => user.stream)
+              .forEach(user => {
+                user.stream.write('chat_id!' + chat_id + '!<' + username + '> ' + message + '\n')
+              })
+          })
+      }
+    }
+
+    runMessage[typeOfMessage](username, chat_id, message)
+    next()
+  })
+}
+
+server.listen(5000)