// Naive — সহজ কিন্তু ভুল socket.on('message', async (data) => { // প্রতি message = একটা DB query await prisma.message.create({ data: { content: data.content, ... } }); // প্রতি message = সব client-কে blast io.emit('new-message', data); });

async function validateMessage(socket, data) { // Gate 1: auth (middleware-এ done) // Gate 2: enrollment const enrolled = await redis.get( `enroll:${socket.user.id}:${data.classId}` ); if (!enrolled) throw new Error('Not enrolled'); // Gate 3: ban const banned = await redis.get( `banned:${socket.user.id}` ); if (banned) throw new Error('Banned'); // Gate 4: token bucket await checkTokenBucket(socket.user.id); // Gate 5: content if (!data.content?.trim()) throw new Error('Empty message'); if (data.content.length > 1000) throw new Error('Too long'); return true; }

async function checkTokenBucket(userId) { const key = `bucket:chat:${userId}`; // Redis-এ atomic operation const [tokens] = await redis .multi() .get(key) .exec(); const current = tokens ? parseInt(tokens[1]) : 10; if (current <= 0) { throw new Error( 'বেশি দ্রুত message করছ, একটু থামো' ); } // Token কমাও, ৬০ সেকেন্ড TTL await redis.set(key, current - 1, 'EX' , 60); } // Refill — separate cron job // প্রতি সেকেন্ডে ২ token add করো // কিন্তু max 10-এর বেশি নয়

// Connection-এ load (Layer 1) io.use(async (socket, next) => { const banned = await redis.get( `banned:${socket.user.id}` ); socket.isBanned = !!banned; next(); }); // Message-এ check async function checkBan(socket) { // Layer 1: memory if (socket.isBanned) throw err; // Layer 2: Redis const redisBan = await redis.get( `banned:${socket.user.id}` ); if (redisBan) { socket.isBanned = true; throw err; } // Layer 3: DB (rare case) const dbBan = await prisma.ban.findFirst({ where: { userId: socket.user.id, isActive: true } }); if (dbBan) { await redis.setex( `banned:${socket.user.id}`, 300, '1' ); socket.isBanned = true; throw err; } }

// Producer — Socket server socket.on('send-message', async (data) => { await validateMessage(socket, data); // Optimistic broadcast (instant) io.to(`class:${data.classId}`) .emit('new-message', { ...data, senderId: socket.user.id, pending: true }); // Redis Stream-এ push await redis.xadd('chat:stream', '*', { content: data.content, senderId: socket.user.id, classId: data.classId, socketId: socket.id }); });

// Consumer — Dedicated Worker Process const buffer = []; async function worker() { while (true) { // Redis Stream থেকে read const msgs = await redis.xread( 'COUNT', 100, 'BLOCK', 120, // 120ms wait 'STREAMS', 'chat:stream', lastId ); if (msgs) buffer.push(...msgs); // 120ms পর bulk insert if (buffer.length > 0) { await prisma.message.createMany({ data: buffer.map(m => ({ content: m.content, senderId: m.senderId, classId: m.classId, })) }); // Acknowledge — blue tick buffer.forEach(m => { io.to(m.socketId).emit('msg-saved'); }); buffer.length = 0; } } } worker();

-- activeChats table CREATE TABLE activeChats ( id UUID PRIMARY KEY, userId UUID REFERENCES User(id), classId UUID REFERENCES Class(id), socketId VARCHAR(100), joinedAt TIMESTAMP DEFAULT NOW(), lastActive TIMESTAMP DEFAULT NOW(), UNIQUE(userId, classId) -- duplicate নেই ); -- Join করলে upsert INSERT INTO activeChats (userId, classId, socketId, lastActive) VALUES (...) ON CONFLICT (userId, classId) DO UPDATE SET socketId = EXCLUDED.socketId, lastActive = NOW(); -- Disconnect হলে delete DELETE FROM activeChats WHERE socketId = $1;

// SuperAdmin: সব class-এর activity const activeSummary = await prisma .activeChats.groupBy({ by: ['classId'], _count: { userId: true }, orderBy: { _count: { userId: 'desc' } } }); // [{ classId: 'c1', _count: { userId: 450 } }, // { classId: 'c2', _count: { userId: 230 } }]