feature: handle streamable doc update policy in student groups

prisma/migrations/20260722070708_add_presentation_flag_to_student_group/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "student_groups" ADD COLUMN     "can_present" BOOLEAN NOT NULL DEFAULT false,
+ADD COLUMN     "presented_document" JSONB;

prisma/schema.prisma

@@ -139,6 +139,8 @@ model StudentGroup {
   parentId             String?               @map("parent_id") @db.Uuid
   createdAt            DateTime              @default(now()) @map("created_at")
   updatedAt            DateTime              @default(now()) @updatedAt @map("updated_at")
+  canPresent           Boolean               @default(false) @map("can_present")
+  presentedDocument    Json?                 @map("presented_document")
   rootGroupPermissions RootGroupPermission[] @relation("root_group_to_student_group_permission")
   parent               StudentGroup?         @relation("parent_student_group", fields: [parentId], references: [id], onDelete: Cascade)
   children             StudentGroup[]        @relation("parent_student_group")

prisma/view-migrations/tsconfig.json

@@ -4,12 +4,10 @@
         "target": "esnext",
         "module": "nodenext",
         "outDir": "../../dist/view-migrations",
-        "rootDir": "./",
         "esModuleInterop": true,
         "skipLibCheck": true,
-        "baseUrl": "../../",
-        "paths": {
-            "*": ["node_modules/*"]
-        }
+        "types": [
+            "node"
+        ],
     }
 }

src/controllers/studentGroups.ts

@@ -2,6 +2,7 @@ import { StudentGroup as DbStudentGroup } from '../../prisma/generated/client.js
 import { RequestHandler } from 'express';
 import StudentGroup from '../models/StudentGroup.js';
 import { IoEvent, RecordType } from '../routes/socketEventTypes.js';
+import { JsonObject } from '@prisma/client/runtime/client';
 
 export const find: RequestHandler<{ id: string }> = async (req, res, next) => {
     const group = await StudentGroup.findModel((req as any).user!, req.params.id);
@@ -14,11 +15,11 @@ export const create: RequestHandler<any, any, DbStudentGroup> = async (req, res,
     res.status(200).json(model);
 };
 
-export const update: RequestHandler<{ id: string }, any, { data: DbStudentGroup }> = async (
-    req,
-    res,
-    next
-) => {
+export const update: RequestHandler<
+    { id: string },
+    any,
+    { data: Partial<Omit<DbStudentGroup, 'presentedDocument'> & { presentedDocument: JsonObject }> }
+> = async (req, res, next) => {
     const model = await StudentGroup.updateModel((req as any).user!, req.params.id, req.body.data);
     if (!model) {
         return res.status(404).json({ message: 'Student group not found' });

src/models/StudentGroup.ts

@@ -3,14 +3,45 @@ import prisma from '../prisma.js';
 import { HTTP403Error, HTTP404Error } from '../utils/errors/Errors.js';
 import { createDataExtractor } from '../helpers/dataExtractor.js';
 import { hasElevatedAccess, Role } from './User.js';
+import { JsonObject } from '@prisma/client/runtime/client';
+import Logger from '../utils/logger.js';
 
 const getData = createDataExtractor<Prisma.StudentGroupUncheckedUpdateInput>(
     ['description', 'name'],
-    ['parentId']
+    ['parentId', 'canPresent', 'presentedDocument']
 );
 
 export type ApiStudentGroup = DbStudentGroup & { userIds: string[]; adminIds: string[] };
 
+export const StreamableGroupUserCacheStore = new Map<string, Set<string>>();
+
+const setStreamableGroupUsers = (group: ApiStudentGroup) => {
+    if (!group.canPresent) {
+        StreamableGroupUserCacheStore.delete(group.id);
+        return;
+    }
+    StreamableGroupUserCacheStore.set(group.id, new Set(group.userIds.concat(group.adminIds)));
+};
+
+let lastCacheRecreation: number | null = null;
+const MS_IN_45_MINUTES = 1000 * 60 * 45;
+export const recreateStreamableGroupUserCache = async () => {
+    if (lastCacheRecreation && Date.now() - lastCacheRecreation < MS_IN_45_MINUTES) {
+        return;
+    }
+    const all = await prisma.studentGroup.findMany({
+        include: { users: true }
+    });
+    StreamableGroupUserCacheStore.clear();
+    for (const group of all.map((record) => asApiRecord(record)!)) {
+        setStreamableGroupUsers(group);
+    }
+    Logger.info(
+        `☄️  Initialized StreamableGroupUserCacheStore with ${StreamableGroupUserCacheStore.size} groups`
+    );
+    lastCacheRecreation = Date.now();
+};
+
 function asApiRecord(
     record: DbStudentGroup & { users: { userId: string; isAdmin: boolean }[] }
 ): ApiStudentGroup;
@@ -65,13 +96,18 @@ function StudentGroup(db: PrismaClient['studentGroup']) {
             return asApiRecord(model);
         },
 
-        async updateModel(actor: User, id: string, data: Partial<DbStudentGroup>): Promise<ApiStudentGroup> {
+        async updateModel(
+            actor: User,
+            id: string,
+            data: Partial<Omit<DbStudentGroup, 'presentedDocument'> & { presentedDocument: JsonObject }>
+        ): Promise<ApiStudentGroup> {
             const record = await this.findModel(actor, id);
             if (!hasAdminAccess(actor, record)) {
                 throw new HTTP403Error('Not authorized');
             }
             /** remove fields not updatable*/
             const sanitized = getData(data, false, hasElevatedAccess(actor.role));
+
             const parentId =
                 typeof sanitized.parentId === 'string' ? sanitized.parentId : sanitized.parentId?.set;
             if (parentId && actor.role !== Role.ADMIN) {
@@ -87,7 +123,11 @@ function StudentGroup(db: PrismaClient['studentGroup']) {
                 data: sanitized,
                 include: { users: { select: { userId: true, isAdmin: true } } }
             });
-            return asApiRecord(result);
+            const resultApi = asApiRecord(result);
+            if (resultApi.canPresent !== record.canPresent) {
+                setStreamableGroupUsers(resultApi);
+            }
+            return resultApi;
         },
 
         async setAdminRole(
@@ -143,6 +183,7 @@ function StudentGroup(db: PrismaClient['studentGroup']) {
                 },
                 include: { users: { select: { userId: true, isAdmin: true } } }
             });
+            StreamableGroupUserCacheStore.get(id)?.add(userId);
             return asApiRecord(result)!;
         },
 
@@ -160,6 +201,7 @@ function StudentGroup(db: PrismaClient['studentGroup']) {
                 data: { users: { delete: { id: { userId: userId, studentGroupId: record.id } } } },
                 include: { users: { select: { userId: true, isAdmin: true } } }
             });
+            StreamableGroupUserCacheStore.get(id)?.delete(userId);
             return asApiRecord(result)!;
         },
 
@@ -204,14 +246,17 @@ function StudentGroup(db: PrismaClient['studentGroup']) {
                 },
                 include: { users: { select: { userId: true, isAdmin: true } } }
             });
-            return asApiRecord(model)!;
+            const result = asApiRecord(model)!;
+            setStreamableGroupUsers(result);
+            return result;
         },
 
         async deleteModel(actor: User, id: string): Promise<DbStudentGroup> {
             const record = await this.findModel(actor, id);
             if (!hasAdminAccess(actor, record)) {
                 throw new HTTP403Error('Not authorized');
             }
+            StreamableGroupUserCacheStore.delete(id);
             return db.delete({ where: { id: id } });
         }
     });

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

@@ -3,10 +3,11 @@ import { ClientToServerEvents, IoClientEvent, ServerToClientEvents } from '../so
 import type { DefaultEventsMap, Socket } from 'socket.io';
 import prisma from '../../prisma.js';
 import StudentGroup from '../../models/StudentGroup.js';
-import onStreamUpdate from './streamUpdate.handler.js';
+import onStreamUpdate, { onStreamDynamicRoomUpdate } from './streamUpdate.handler.js';
 import DocumentRoot from '../../models/DocumentRoot.js';
 import { highestAccess, RWAccess } from '../../helpers/accessPolicy.js';
 import { Role } from '../../models/User.js';
+import Logger from '../../utils/logger.js';
 type SocketType = Socket<ClientToServerEvents, ServerToClientEvents, DefaultEventsMap, any>;
 
 const isDocumentRoot = (roomId: string) => {
@@ -37,16 +38,16 @@ const findStudentGroup = (userId: string, roomId: string) => {
 const joinRoom = (socket: SocketType, roomId: string, joinStreamGroup: boolean) => {
     socket.join(roomId);
     if (joinStreamGroup) {
-        socket.on(IoClientEvent.STREAM_UPDATE, onStreamUpdate(roomId, socket));
+        socket.on(IoClientEvent.STREAM_UPDATE, onStreamDynamicRoomUpdate(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) {
-            return isDocumentRoot(roomId)
-                .then((docRoot) => {
-                    joinRoom(socket, roomId, !!docRoot);
+            return Promise.all([isDocumentRoot(roomId), StudentGroup.findModel(user, roomId)])
+                .then(([docRoot, group]) => {
+                    joinRoom(socket, roomId, !!docRoot || !!(group && group.canPresent));
                     callback(true);
                 })
                 .catch(() => {
@@ -55,7 +56,7 @@ const onJoinRoom: (user: User, socket: SocketType) => ClientToServerEvents[IoCli
         }
         StudentGroup.findModel(user, roomId).then((group) => {
             if (group) {
-                socket.join(roomId);
+                joinRoom(socket, roomId, group.canPresent);
                 callback(true);
             } else {
                 if (user.role === Role.TEACHER) {

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

@@ -1,18 +1,50 @@
-import { ClientToServerEvents, IoClientEvent, IoEvent, ServerToClientEvents } from '../socketEventTypes.js';
+import {
+    ChangedDocument,
+    ClientToServerEvents,
+    IoClientEvent,
+    IoEvent,
+    ServerToClientEvents
+} from '../socketEventTypes.js';
+import { User } from '../../../prisma/generated/client.js';
 import type { DefaultEventsMap, Socket } from 'socket.io';
+import { StreamableGroupUserCacheStore } from '../../models/StudentGroup.js';
 
 const onStreamUpdate: (
+    user: User,
+    socket: Socket<ClientToServerEvents, ServerToClientEvents, DefaultEventsMap, any>
+) => ClientToServerEvents[IoClientEvent.STREAM_UPDATE] = (user, socket) => (payload) => {
+    if (!StreamableGroupUserCacheStore.get(payload.roomId)?.has(user.id)) {
+        return;
+    }
+    const pkg: ChangedDocument = {
+        data: payload.data,
+        id: payload.id,
+        updatedAt: payload.updatedAt
+    };
+    if (payload.meta) {
+        pkg.meta = payload.meta;
+    }
+
+    socket.to(payload.roomId).emit(IoEvent.CHANGED_DOCUMENT, pkg);
+};
+
+export const onStreamDynamicRoomUpdate: (
     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, {
+    const pkg: ChangedDocument = {
         data: payload.data,
         id: payload.id,
         updatedAt: payload.updatedAt
-    });
+    };
+    if (payload.meta) {
+        pkg.meta = payload.meta;
+    }
+
+    socket.to(payload.roomId).emit(IoEvent.CHANGED_DOCUMENT, pkg);
 };
 
 export default onStreamUpdate;

src/routes/socketEventTypes.ts

@@ -47,13 +47,14 @@ export interface ChangedRecord<T extends RecordType> {
     record: TypeRecordMap[T];
 }
 
-export interface ChangedDocument {
+export interface ChangedDocument<T = any> {
     id: string;
     data: Prisma.JsonValue;
     updatedAt: Date;
+    meta?: T;
 }
 
-export interface StreamedDynamicDocument extends ChangedDocument {
+export interface StreamedDynamicDocument<T = any> extends ChangedDocument<T> {
     roomId: string;
 }
 

src/routes/socketEvents.ts

@@ -10,6 +10,7 @@ import onAction from './event-handlers/action.handler.js';
 import onJoinRoom from './event-handlers/joinRoom.handler.js';
 import onLeaveRoom from './event-handlers/leaveRoom.handler.js';
 import { auth } from '../auth.js';
+import onStreamUpdate from './event-handlers/streamUpdate.handler.js';
 
 export enum IoRoom {
     ADMIN = 'admin',
@@ -44,6 +45,7 @@ const EventRouter = (io: Server<ClientToServerEvents, ServerToClientEvents>) =>
         socket.join(IoRoom.ALL);
         socket.on(IoClientEvent.JOIN_ROOM, onJoinRoom(user, socket));
         socket.on(IoClientEvent.LEAVE_ROOM, onLeaveRoom(user, socket));
+        socket.on(IoClientEvent.STREAM_UPDATE, onStreamUpdate(user, socket));
         const groups = await StudentGroup.all(user);
         const groupIds = groups.map((group) => group.id);
         if (groupIds.length > 0) {

src/server.ts

@@ -4,20 +4,29 @@ import http from 'http';
 import * as Sentry from '@sentry/node';
 import Logger from './utils/logger.js';
 import dotenv from 'dotenv';
+import { recreateStreamableGroupUserCache } from './models/StudentGroup.js';
 dotenv.config();
 
 const PORT = process.env.PORT || 3002;
 
-const server = http.createServer(app);
-initializeSocketIo(server);
+const start = async () => {
+    const server = http.createServer(app);
+    initializeSocketIo(server);
 
-configure(app);
+    configure(app);
+    await recreateStreamableGroupUserCache();
 
-if (process.env.NODE_ENV === 'production' && process.env.SENTRY_DSN) {
-    Sentry.setupExpressErrorHandler(app);
-}
+    if (process.env.NODE_ENV === 'production' && process.env.SENTRY_DSN) {
+        Sentry.setupExpressErrorHandler(app);
+    }
 
-server.listen(PORT || 3002, () => {
-    Logger.info(`application is running at: http://localhost:${PORT}`);
-    Logger.info('Press Ctrl+C to quit.');
+    server.listen(PORT || 3002, () => {
+        Logger.info(`application is running at: http://localhost:${PORT}`);
+        Logger.info('Press Ctrl+C to quit.');
+    });
+};
+
+start().catch((error) => {
+    Logger.error('Failed to initialize server', error);
+    process.exit(1);
 });