feature: support streaming document updates

Author: lebalz

src/routes/event-handlers/joinRoom.handler.ts

@@ -3,48 +3,90 @@ import { ClientToServerEvents, IoClientEvent, ServerToClientEvents } from '../so
 import type { DefaultEventsMap, Socket } from 'socket.io';
 import prisma from '../../prisma';
 import StudentGroup from '../../models/StudentGroup';
+import onStreamUpdate from './streamUpdate.handler';
+import DocumentRoot from '../../models/DocumentRoot';
+import { highestAccess, RWAccess } from '../../helpers/accessPolicy';
+type SocketType = Socket<ClientToServerEvents, ServerToClientEvents, DefaultEventsMap, any>;
 
-const onJoinRoom: (
-    user: User,
-    socket: Socket<ClientToServerEvents, ServerToClientEvents, DefaultEventsMap, any>
-) => ClientToServerEvents[IoClientEvent.JOIN_ROOM] =
+const isDocumentRoot = (roomId: string) => {
+    return prisma.documentRoot.findFirst({ where: { id: roomId } });
+};
+
+const findDocumentRoot = (user: User, roomId: string) => {
+    return DocumentRoot.getPermissions(user, roomId).then((res) => {
+        if (!res) {
+            return false;
+        } else {
+            const access = new Set([
+                ...res.groupPermissions.map((p) => p.access),
+                ...res.userPermissions.map((p) => p.access)
+            ]);
+            const current = highestAccess(access);
+            return RWAccess.has(current);
+        }
+    });
+};
+
+const findStudentGroup = (userId: string, roomId: string) => {
+    return prisma.studentGroup.findFirst({
+        where: {
+            users: {
+                some: {
+                    AND: [
+                        {
+                            userId: userId,
+                            isAdmin: true
+                        },
+                        {
+                            userId: roomId
+                        }
+                    ]
+                }
+            }
+        }
+    });
+};
+
+const joinRoom = (socket: SocketType, roomId: string, joinStreamGroup: boolean) => {
+    socket.join(roomId);
+    if (joinStreamGroup) {
+        socket.on(IoClientEvent.STREAM_UPDATE, onStreamUpdate(roomId, socket));
+    }
+};
+
+const onJoinRoom: (user: User, socket: SocketType) => ClientToServerEvents[IoClientEvent.JOIN_ROOM] =
     (user, socket) => (roomId: string, callback: (joined: boolean) => void) => {
         if (user.role === Role.ADMIN) {
-            socket.join(roomId);
-            return callback(true);
+            return isDocumentRoot(roomId)
+                .then((docRoot) => {
+                    joinRoom(socket, roomId, !!docRoot);
+                    callback(true);
+                })
+                .catch(() => {
+                    callback(false);
+                });
         }
         StudentGroup.findModel(user, roomId).then((group) => {
             if (group) {
                 socket.join(roomId);
                 callback(true);
             } else {
                 if (user.role === Role.TEACHER) {
-                    prisma.studentGroup
-                        .findFirst({
-                            where: {
-                                users: {
-                                    some: {
-                                        AND: [
-                                            {
-                                                userId: user.id,
-                                                isAdmin: true
-                                            },
-                                            {
-                                                userId: roomId
-                                            }
-                                        ]
-                                    }
-                                }
-                            }
-                        })
-                        .then((userRoom) => {
-                            if (userRoom) {
-                                socket.join(roomId);
-                                callback(true);
-                            } else {
-                                callback(false);
-                            }
-                        });
+                    findStudentGroup(user.id, roomId).then((userRoom) => {
+                        if (userRoom) {
+                            joinRoom(socket, roomId, false);
+                            callback(true);
+                        } else {
+                            findDocumentRoot(user, roomId)
+                                .then((canJoin) => {
+                                    joinRoom(socket, roomId, canJoin);
+                                    callback(true);
+                                })
+                                .catch(() => {
+                                    callback(false);
+                                });
+                        }
+                    });
                 }
             }
         });

src/routes/event-handlers/streamUpdate.handler.ts

@@ -0,0 +1,18 @@
+import { ClientToServerEvents, IoClientEvent, IoEvent, ServerToClientEvents } from '../socketEventTypes';
+import type { DefaultEventsMap, Socket } from 'socket.io';
+
+const onStreamUpdate: (
+    roomId: string,
+    socket: Socket<ClientToServerEvents, ServerToClientEvents, DefaultEventsMap, any>
+) => ClientToServerEvents[IoClientEvent.STREAM_UPDATE] = (roomId, socket) => (payload) => {
+    if (roomId !== payload.roomId) {
+        return;
+    }
+    socket.to(payload.roomId).emit(IoEvent.CHANGED_DOCUMENT, {
+        data: payload.data,
+        id: payload.id,
+        updatedAt: payload.updatedAt
+    });
+};
+
+export default onStreamUpdate;

src/routes/socketEventTypes.ts

@@ -52,6 +52,10 @@ export interface ChangedDocument {
     updatedAt: Date;
 }
 
+export interface StreamedDynamicDocument extends ChangedDocument {
+    roomId: string;
+}
+
 export interface ConnectedClients {
     rooms: [string, number][];
     type: 'full' | 'update';
@@ -98,6 +102,7 @@ export type Notification =
 export enum IoClientEvent {
     JOIN_ROOM = 'JOIN_ROOM',
     LEAVE_ROOM = 'LEAVE_ROOM',
+    STREAM_UPDATE = 'STREAM_UPDATE',
     ACTION = 'ACTION'
 }
 
@@ -120,4 +125,5 @@ export interface ClientToServerEvents {
     [IoClientEvent.JOIN_ROOM]: (roomId: string, callback: (joined: boolean) => void) => void;
     [IoClientEvent.LEAVE_ROOM]: (roomId: string, callback: (left: boolean) => void) => void;
     [IoClientEvent.ACTION]: (action: Action, callback: (ok: boolean) => void) => void;
+    [IoClientEvent.STREAM_UPDATE]: (payload: StreamedDynamicDocument) => void;
 }