feat: Document presence indicator (#1114)

* Update websockets to allow joining document-based rooms

* dynamic websocket joining

* emit user.join/leave events when entering and exiting document rooms

* presence storage

* feat: frontend presence store

* lint

* UI updates

* First pass editing state

* refactoring

* Timeout per user/doc
lint

* Document data loading refactor to keep Socket mounted

* restore: Mark as viewed functionality
Add display of 'you' to collaborators

* fix: Socket/document remount when document slug changes due to title change

* Revert unneccessary package update

* Move editing ping interval to a shared constant

* fix: Flash of sidebar when loading page directly on editing mode

* separate document and revision loading

* add comments for socket events

* fix: Socket events getting bound multiple times on reconnect

* fix: Clear client side presence state on disconnect

* fix: Don't ignore server side error
Improved documentation

* More comments / why comments

* rename Socket -> SocketPresence

* fix: Handle redis is down
remove unneccessary join

* fix: PR feedback

server/index.js

@@ -1,13 +1,17 @@
 // @flow
+import { promisify } from 'util';
 import http from 'http';
 import IO from 'socket.io';
 import SocketAuth from 'socketio-auth';
 import socketRedisAdapter from 'socket.io-redis';
 import { getUserForJWT } from './utils/jwt';
-import { Collection } from './models';
+import { Document, Collection, View } from './models';
+import { client } from './redis';
 import app from './app';
 import policy from './policies';
 
+const redisHget = promisify(client.hget).bind(client);
+const redisHset = promisify(client.hset).bind(client);
 const server = http.createServer(app.callback());
 let io;
 
@@ -30,6 +34,10 @@ if (process.env.WEBSOCKETS_ENABLED === 'true') {
         const user = await getUserForJWT(token);
         socket.client.user = user;
 
+        // store the mapping between socket id and user id in redis
+        // so that it is accessible across multiple server nodes
+        await redisHset(socket.id, 'userId', user.id);
+
         return callback(null, true);
       } catch (err) {
         return callback(err);
@@ -37,33 +45,129 @@ if (process.env.WEBSOCKETS_ENABLED === 'true') {
     },
     postAuthenticate: async (socket, data) => {
       const { user } = socket.client;
-      // join the rooms associated with the current team
-      // and user so we can send authenticated events
-      socket.join(`team-${user.teamId}`);
-      socket.join(`user-${user.id}`);
 
-      // join rooms associated with collections this user
+      // the rooms associated with the current team
+      // and user so we can send authenticated events
+      let rooms = [`team-${user.teamId}`, `user-${user.id}`];
+
+      // the rooms associated with collections this user
       // has access to on connection. New collection subscriptions
-      // are managed from the client as needed
+      // are managed from the client as needed through the 'join' event
       const collectionIds = await user.collectionIds();
       collectionIds.forEach(collectionId =>
-        socket.join(`collection-${collectionId}`)
+        rooms.push(`collection-${collectionId}`)
       );
 
-      // allow the client to request to join rooms based on
-      // new collections being created.
-      socket.on('join', async event => {
-        const collection = await Collection.scope({
-          method: ['withMembership', user.id],
-        }).findByPk(event.roomId);
+      // join all of the rooms at once
+      socket.join(rooms);
 
-        if (can(user, 'read', collection)) {
-          socket.join(`collection-${event.roomId}`);
+      // allow the client to request to join rooms
+      socket.on('join', async event => {
+        // user is joining a collection channel, because their permissions have
+        // changed, granting them access.
+        if (event.collectionId) {
+          const collection = await Collection.scope({
+            method: ['withMembership', user.id],
+          }).findByPk(event.collectionId);
+
+          if (can(user, 'read', collection)) {
+            socket.join(`collection-${event.collectionId}`);
+          }
+        }
+
+        // user is joining a document channel, because they have navigated to
+        // view a document.
+        if (event.documentId) {
+          const document = await Document.findByPk(event.documentId, {
+            userId: user.id,
+          });
+
+          if (can(user, 'read', document)) {
+            const room = `document-${event.documentId}`;
+
+            await View.touch(event.documentId, user.id, event.isEditing);
+            const editing = await View.findRecentlyEditingByDocument(
+              event.documentId
+            );
+
+            socket.join(room, () => {
+              // let everyone else in the room know that a new user joined
+              io.to(room).emit('user.join', {
+                userId: user.id,
+                documentId: event.documentId,
+                isEditing: event.isEditing,
+              });
+
+              // let this user know who else is already present in the room
+              io.in(room).clients(async (err, sockets) => {
+                if (err) throw err;
+
+                // because a single user can have multiple socket connections we
+                // need to make sure that only unique userIds are returned. A Map
+                // makes this easy.
+                let userIds = new Map();
+                for (const socketId of sockets) {
+                  const userId = await redisHget(socketId, 'userId');
+                  userIds.set(userId, userId);
+                }
+                socket.emit('document.presence', {
+                  documentId: event.documentId,
+                  userIds: Array.from(userIds.keys()),
+                  editingIds: editing.map(view => view.userId),
+                });
+              });
+            });
+          }
         }
       });
 
+      // allow the client to request to leave rooms
       socket.on('leave', event => {
-        socket.leave(`collection-${event.roomId}`);
+        if (event.collectionId) {
+          socket.leave(`collection-${event.collectionId}`);
+        }
+        if (event.documentId) {
+          const room = `document-${event.documentId}`;
+          socket.leave(room, () => {
+            io.to(room).emit('user.leave', {
+              userId: user.id,
+              documentId: event.documentId,
+            });
+          });
+        }
+      });
+
+      socket.on('disconnecting', () => {
+        const rooms = Object.keys(socket.rooms);
+
+        rooms.forEach(room => {
+          if (room.startsWith('document-')) {
+            const documentId = room.replace('document-', '');
+            io.to(room).emit('user.leave', {
+              userId: user.id,
+              documentId,
+            });
+          }
+        });
+      });
+
+      socket.on('presence', async event => {
+        const room = `document-${event.documentId}`;
+
+        if (event.documentId && socket.rooms[room]) {
+          const view = await View.touch(
+            event.documentId,
+            user.id,
+            event.isEditing
+          );
+          view.user = user;
+
+          io.to(room).emit('user.presence', {
+            userId: user.id,
+            documentId: event.documentId,
+            isEditing: event.isEditing,
+          });
+        }
       });
     },
   });