Initial commit
This commit is contained in:
@@ -9,6 +9,11 @@ function createRealtimeServer(server, { clientOrigin, database, onError }) {
|
||||
},
|
||||
})
|
||||
|
||||
const typingByRoom = new Map()
|
||||
|
||||
// io.use: Middleware die bei jeder eingehenden WebSocket-Verbindung läuft.
|
||||
// Prüft ob room und userId im Handshake vorhanden sind und ob der User in der DB existiert.
|
||||
// Wer vorher nicht POST /join aufgerufen hat, hat keine gültige userId und wird abgelehnt.
|
||||
io.use(async (socket, next) => {
|
||||
try {
|
||||
const auth = socket.handshake.auth || {}
|
||||
@@ -25,6 +30,7 @@ function createRealtimeServer(server, { clientOrigin, database, onError }) {
|
||||
return next(new Error('User not authorized for this room'))
|
||||
}
|
||||
|
||||
// room und userId in socket.data speichern — so sind sie in registerConnection verfügbar.
|
||||
socket.data.room = room
|
||||
socket.data.userId = userId
|
||||
next()
|
||||
@@ -34,7 +40,7 @@ function createRealtimeServer(server, { clientOrigin, database, onError }) {
|
||||
})
|
||||
|
||||
io.on('connection', (socket) => {
|
||||
registerConnection(io, socket, database, onError).catch((error) => {
|
||||
registerConnection(io, socket, database, onError, typingByRoom).catch((error) => {
|
||||
onError('Socket connection setup failed', error)
|
||||
socket.disconnect(true)
|
||||
})
|
||||
@@ -43,8 +49,10 @@ function createRealtimeServer(server, { clientOrigin, database, onError }) {
|
||||
return io
|
||||
}
|
||||
|
||||
async function registerConnection(io, socket, database, onError) {
|
||||
async function registerConnection(io, socket, database, onError, typingByRoom) {
|
||||
const { room, userId } = socket.data
|
||||
|
||||
// socket.join: Fügt den Socket einem Room-Kanal hinzu — io.to(room).emit() erreicht nur diese Sockets.
|
||||
socket.join(room)
|
||||
|
||||
const user = await database.setUserOnline(userId, room)
|
||||
@@ -53,10 +61,25 @@ async function registerConnection(io, socket, database, onError) {
|
||||
return
|
||||
}
|
||||
|
||||
socket.data.username = user.username
|
||||
|
||||
socket.emit('connection:ready', { userId, room })
|
||||
socket.emit('history:init', await database.getMessages(room))
|
||||
|
||||
// Teilnehmerliste an alle im Room senden — User sieht sofort wer online ist.
|
||||
await broadcastParticipants(io, database, room)
|
||||
|
||||
// typingByRoom: Map pro Room — speichert wer gerade tippt { userId → { userId, username } }.
|
||||
socket.on('typing:start', () => {
|
||||
if (!typingByRoom.has(room)) typingByRoom.set(room, new Map())
|
||||
typingByRoom.get(room).set(userId, { userId, username: socket.data.username })
|
||||
io.to(room).emit('typing:update', [...typingByRoom.get(room).values()])
|
||||
})
|
||||
|
||||
socket.on('typing:stop', () => {
|
||||
typingByRoom.get(room)?.delete(userId)
|
||||
io.to(room).emit('typing:update', [...(typingByRoom.get(room)?.values() ?? [])])
|
||||
})
|
||||
|
||||
socket.on('message:send', async (payload) => {
|
||||
try {
|
||||
const text =
|
||||
@@ -65,11 +88,16 @@ async function registerConnection(io, socket, database, onError) {
|
||||
return
|
||||
}
|
||||
|
||||
// Nochmal prüfen ob der User noch im richtigen Room ist — könnte sich seit dem Verbinden geändert haben.
|
||||
const activeUser = await database.getUser(userId)
|
||||
if (!activeUser || activeUser.room !== room) {
|
||||
return
|
||||
}
|
||||
|
||||
// Tipp-Indikator sofort entfernen wenn eine Nachricht gesendet wird.
|
||||
typingByRoom.get(room)?.delete(userId)
|
||||
io.to(room).emit('typing:update', [...(typingByRoom.get(room)?.values() ?? [])])
|
||||
|
||||
const message = await database.addMessage({
|
||||
id: randomUUID(),
|
||||
room,
|
||||
@@ -77,6 +105,7 @@ async function registerConnection(io, socket, database, onError) {
|
||||
username: activeUser.username,
|
||||
text: text.trim(),
|
||||
})
|
||||
// Nachricht an alle im Room senden — inkl. Sender selbst.
|
||||
io.to(room).emit('message:new', message)
|
||||
} catch (error) {
|
||||
onError('Failed to handle socket message', error)
|
||||
@@ -84,6 +113,9 @@ async function registerConnection(io, socket, database, onError) {
|
||||
})
|
||||
|
||||
socket.on('disconnect', () => {
|
||||
typingByRoom.get(room)?.delete(userId)
|
||||
io.to(room).emit('typing:update', [...(typingByRoom.get(room)?.values() ?? [])])
|
||||
// User als offline markieren und Teilnehmerliste aktualisieren.
|
||||
database
|
||||
.setUserOffline(userId)
|
||||
.then(() => broadcastParticipants(io, database, room))
|
||||
@@ -100,6 +132,8 @@ async function broadcastParticipants(io, database, room) {
|
||||
io.to(room).emit('participants:update', participants)
|
||||
}
|
||||
|
||||
// Liest einen Parameter zuerst aus der primären Quelle (auth), dann aus dem Fallback (query).
|
||||
// Schützt gegen leere Strings — nur echte Werte werden akzeptiert.
|
||||
function readParam(primary, fallback) {
|
||||
if (typeof primary === 'string' && primary.trim()) return primary
|
||||
if (typeof fallback === 'string' && fallback.trim()) return fallback
|
||||
|
||||
@@ -1,33 +1,44 @@
|
||||
const fs = require('fs')
|
||||
const path = require('path')
|
||||
|
||||
// JsonStore: Dateibasierter Fallback-Store wenn kein MongoDB vorhanden ist.
|
||||
// Alle Daten liegen als JSON-Datei auf der Festplatte — kein externer Datenbankserver nötig.
|
||||
// Nachteil gegenüber MongoDB: Kein Concurrency-Schutz bei mehreren gleichzeitigen Schreiboperationen.
|
||||
class JsonStore {
|
||||
constructor(filePath) {
|
||||
this.filePath = filePath
|
||||
this.state = { users: [], messages: [] }
|
||||
this.state = { users: [], messages: [] } // In-Memory-State — alle Operationen laufen darauf.
|
||||
this._ensureStore()
|
||||
}
|
||||
|
||||
// Stellt sicher dass die JSON-Datei und das Verzeichnis existieren.
|
||||
// Wenn die Datei beschädigt ist (kein gültiges JSON), wird sie neu erstellt.
|
||||
_ensureStore() {
|
||||
const dir = path.dirname(this.filePath)
|
||||
fs.mkdirSync(dir, { recursive: true })
|
||||
fs.mkdirSync(dir, { recursive: true }) // Verzeichnis anlegen falls nicht vorhanden.
|
||||
if (!fs.existsSync(this.filePath)) {
|
||||
this._persist()
|
||||
return
|
||||
}
|
||||
try {
|
||||
const contents = fs.readFileSync(this.filePath, 'utf-8')
|
||||
this.state = JSON.parse(contents)
|
||||
this.state = JSON.parse(contents) // Datei in den In-Memory-State laden.
|
||||
} catch (error) {
|
||||
// Datei vorhanden aber nicht lesbar oder kein gültiges JSON — neu erstellen.
|
||||
console.warn('JSON store damaged, recreating', error)
|
||||
this._persist()
|
||||
}
|
||||
}
|
||||
|
||||
// Schreibt den aktuellen In-Memory-State in die JSON-Datei.
|
||||
// null, 2 formatiert das JSON mit Einrückung — besser lesbar beim Debugging.
|
||||
_persist() {
|
||||
fs.writeFileSync(this.filePath, JSON.stringify(this.state, null, 2), 'utf-8')
|
||||
}
|
||||
|
||||
// Upsert: User aktualisieren wenn vorhanden, sonst neu anlegen.
|
||||
// Spread-Operator {...this.state.users[index], ...} übernimmt alle bestehenden Felder
|
||||
// und überschreibt nur die geänderten — joinedAt bleibt beim Update erhalten.
|
||||
async upsertUser({ id, username, room }) {
|
||||
const trimmed = username?.trim()
|
||||
if (!trimmed) {
|
||||
@@ -60,6 +71,7 @@ class JsonStore {
|
||||
return user
|
||||
}
|
||||
|
||||
// Object.assign mutiert das gefundene Objekt direkt im Array — kein Index-Lookup nötig.
|
||||
async setUserOnline(userId, room) {
|
||||
const user = this.state.users.find((entry) => entry.id === userId)
|
||||
if (!user) {
|
||||
@@ -74,6 +86,7 @@ class JsonStore {
|
||||
return user
|
||||
}
|
||||
|
||||
// lastActiveAt wird auch beim Offline-Setzen aktualisiert — zeigt wann der User zuletzt aktiv war.
|
||||
async setUserOffline(userId) {
|
||||
const user = this.state.users.find((entry) => entry.id === userId)
|
||||
if (!user) {
|
||||
@@ -96,6 +109,8 @@ class JsonStore {
|
||||
return user
|
||||
}
|
||||
|
||||
// splice(index, 1) entfernt genau ein Element an der gefundenen Position.
|
||||
// Destructuring [removed] holt das gelöschte Element aus dem zurückgegebenen Array.
|
||||
async removeUser(userId) {
|
||||
const index = this.state.users.findIndex((entry) => entry.id === userId)
|
||||
if (index === -1) return null
|
||||
@@ -108,6 +123,8 @@ class JsonStore {
|
||||
return this.state.users.find((entry) => entry.id === userId) || null
|
||||
}
|
||||
|
||||
// .map() gibt nur die öffentlichen Felder zurück — interne Felder werden nicht nach außen gegeben.
|
||||
// Sortierung: Online-User zuerst, dann nach letzter Aktivität absteigend.
|
||||
async getParticipants(room) {
|
||||
return this.state.users
|
||||
.filter((entry) => entry.room === room)
|
||||
@@ -126,6 +143,7 @@ class JsonStore {
|
||||
})
|
||||
}
|
||||
|
||||
// createdAt wird serverseitig gesetzt — dem Client-Timestamp wird nicht vertraut.
|
||||
async addMessage({ id, room, userId, username, text }) {
|
||||
const message = {
|
||||
id,
|
||||
@@ -140,11 +158,25 @@ class JsonStore {
|
||||
return message
|
||||
}
|
||||
|
||||
// slice(-limit) gibt die letzten N Nachrichten zurück — neueste am Ende, älteste zuerst.
|
||||
async getMessages(room, limit = 100) {
|
||||
return this.state.messages
|
||||
.filter((message) => message.room === room)
|
||||
.slice(-limit)
|
||||
}
|
||||
|
||||
// roomMap: Objekt das pro Room die Teilnehmerzahl zählt — schneller als mehrere filter()-Aufrufe.
|
||||
// localeCompare: Alphabetische Sortierung mit Sprachunterstützung (Umlaute etc.).
|
||||
async getRooms() {
|
||||
const roomMap = {}
|
||||
for (const user of this.state.users) {
|
||||
if (!roomMap[user.room]) {
|
||||
roomMap[user.room] = { name: user.room, participantCount: 0 }
|
||||
}
|
||||
if (user.isOnline) roomMap[user.room].participantCount++
|
||||
}
|
||||
return Object.values(roomMap).sort((a, b) => a.name.localeCompare(b.name))
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = JsonStore
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
const config = require('../../config/database')
|
||||
|
||||
|
||||
// Sicheres Laden des MongoDB-Treibers — wenn das Paket nicht installiert ist,
|
||||
// wird der Fehler erst in createMongoStore() mit einer klaren Meldung geworfen.
|
||||
let MongoClient
|
||||
try {
|
||||
;({ MongoClient } = require('mongodb'))
|
||||
@@ -8,12 +9,14 @@ try {
|
||||
MongoClient = null
|
||||
}
|
||||
|
||||
|
||||
// Collection-Namen als Konstanten — wenn sich ein Name ändert, muss er nur hier angepasst werden.
|
||||
const collectionNames = {
|
||||
users: 'users',
|
||||
messages: 'messages',
|
||||
}
|
||||
|
||||
// MongoDB speichert den Primärschlüssel intern als "_id".
|
||||
// Diese Funktion übersetzt das in "id" damit der Rest der App nichts von MongoDB-Interna wissen muss.
|
||||
function normalizeUser(doc) {
|
||||
if (!doc) return null
|
||||
const {
|
||||
@@ -34,6 +37,8 @@ function normalizeUser(doc) {
|
||||
}
|
||||
}
|
||||
|
||||
// Gleiche Logik wie normalizeUser — username wird redundant in der Nachricht gespeichert
|
||||
// damit beim Laden keine extra DB-Abfrage für den Usernamen nötig ist.
|
||||
function normalizeMessage(doc) {
|
||||
if (!doc) return null
|
||||
const {
|
||||
@@ -54,6 +59,8 @@ function normalizeMessage(doc) {
|
||||
}
|
||||
}
|
||||
|
||||
// Kompatibilitätsfunktion für ältere MongoDB-Treiber (< 4.0).
|
||||
// findOneAndUpdate gibt dort das Ergebnis in einem "value"-Wrapper zurück — ab 4.0 direkt.
|
||||
function extractDocument(result) {
|
||||
if (!result) return null
|
||||
return Object.prototype.hasOwnProperty.call(result, 'value')
|
||||
@@ -61,8 +68,9 @@ function extractDocument(result) {
|
||||
: result
|
||||
}
|
||||
|
||||
// async damit der Start der App nicht blockiert während MongoDB-Verbindung und Indizes aufgebaut werden.
|
||||
async function createMongoStore() {
|
||||
// Guard: fail fast with an actionable message if the driver is missing.
|
||||
|
||||
if (!MongoClient) {
|
||||
const message =
|
||||
'MongoDB driver is not installed. Run `npm install mongodb` in Backend/.'
|
||||
@@ -73,18 +81,20 @@ async function createMongoStore() {
|
||||
|
||||
const client = new MongoClient(config.uri, {
|
||||
...config.options,
|
||||
appName: 'chat-app',
|
||||
appName: 'chat-app', // Wird in MongoDB Atlas angezeigt um zu sehen welche App verbunden ist.
|
||||
})
|
||||
|
||||
await client.connect()
|
||||
const db = client.db(config.dbName)
|
||||
const db = client.db(config.dbName) // Entspricht "USE chatapp" in SQL.
|
||||
|
||||
const users = db.collection(collectionNames.users)
|
||||
const messages = db.collection(collectionNames.messages)
|
||||
|
||||
// Indizes werden einmalig beim Start angelegt — ohne Index durchsucht MongoDB jede Abfrage
|
||||
// alle Dokumente (Full Collection Scan). Promise.all legt beide parallel an.
|
||||
await Promise.all([
|
||||
users.createIndex({ room: 1, isOnline: -1, lastActiveAt: -1 }),
|
||||
messages.createIndex({ room: 1, createdAt: 1 }),
|
||||
users.createIndex({ room: 1, isOnline: -1, lastActiveAt: -1 }), // Für getParticipants()
|
||||
messages.createIndex({ room: 1, createdAt: 1 }), // Für getMessages()
|
||||
])
|
||||
|
||||
console.log(
|
||||
@@ -93,7 +103,12 @@ async function createMongoStore() {
|
||||
}/${config.dbName} (collections: ${Object.values(collectionNames).join(', ')})`,
|
||||
)
|
||||
|
||||
// Öffentliches Interface des Stores — der Rest der App kennt keine MongoDB-Details (Repository-Pattern).
|
||||
return {
|
||||
|
||||
// Upsert = Update + Insert in einem Schritt (atomare Operation).
|
||||
// Wenn User existiert: Felder aktualisieren. Wenn nicht: neu anlegen.
|
||||
// $setOnInsert stellt sicher dass joinedAt nur beim ersten Anlegen gesetzt wird, nicht bei Updates.
|
||||
async upsertUser({ id, username, room }) {
|
||||
const now = new Date()
|
||||
const result = await users.findOneAndUpdate(
|
||||
@@ -112,7 +127,7 @@ async function createMongoStore() {
|
||||
},
|
||||
{
|
||||
upsert: true,
|
||||
returnDocument: 'after',
|
||||
returnDocument: 'after', // Aktualisiertes Dokument zurückgeben, nicht das alte.
|
||||
},
|
||||
)
|
||||
const document = extractDocument(result)
|
||||
@@ -120,11 +135,12 @@ async function createMongoStore() {
|
||||
return normalizeUser(document)
|
||||
}
|
||||
|
||||
|
||||
// Fallback: Manche MongoDB-Versionen geben bei einem Upsert-Insert kein Dokument zurück.
|
||||
const inserted = await users.findOne({ _id: id })
|
||||
return normalizeUser(inserted)
|
||||
},
|
||||
|
||||
// $set aktualisiert nur die angegebenen Felder — der Rest (z.B. username) bleibt unberührt.
|
||||
async setUserOnline(userId, room) {
|
||||
const result = await users.findOneAndUpdate(
|
||||
{ _id: userId },
|
||||
@@ -140,6 +156,7 @@ async function createMongoStore() {
|
||||
return normalizeUser(extractDocument(result))
|
||||
},
|
||||
|
||||
// lastActiveAt wird auch beim Offline-Setzen aktualisiert — zeigt wann der User zuletzt aktiv war.
|
||||
async setUserOffline(userId) {
|
||||
const result = await users.findOneAndUpdate(
|
||||
{ _id: userId },
|
||||
@@ -154,6 +171,7 @@ async function createMongoStore() {
|
||||
return normalizeUser(extractDocument(result))
|
||||
},
|
||||
|
||||
// trim() auf Datenbankebene — auch wenn die Frontend-Validierung fehlt, landen keine schmutzigen Daten in der DB.
|
||||
async updateUsername(userId, username) {
|
||||
const result = await users.findOneAndUpdate(
|
||||
{ _id: userId },
|
||||
@@ -168,6 +186,7 @@ async function createMongoStore() {
|
||||
return normalizeUser(extractDocument(result))
|
||||
},
|
||||
|
||||
// findOneAndDelete löscht und gibt das Dokument in einem Schritt zurück — kein extra findOne() nötig.
|
||||
async removeUser(userId) {
|
||||
const result = await users.findOneAndDelete({ _id: userId })
|
||||
return normalizeUser(extractDocument(result))
|
||||
@@ -178,14 +197,19 @@ async function createMongoStore() {
|
||||
return normalizeUser(doc)
|
||||
},
|
||||
|
||||
// Online-User zuerst, dann nach letzter Aktivität sortiert.
|
||||
// toArray() lädt alle Cursor-Ergebnisse in den Speicher — bei sehr großen Datenmengen wäre ein Cursor besser.
|
||||
async getParticipants(room) {
|
||||
const docs = await users
|
||||
.find({ room })
|
||||
.sort({ isOnline: -1, lastActiveAt: -1 })
|
||||
.sort({ isOnline: -1, lastActiveAt: -1 }) // -1 = absteigend, 1 = aufsteigend
|
||||
.toArray()
|
||||
return docs.map(normalizeUser)
|
||||
},
|
||||
|
||||
// _id wird von außen als UUID übergeben statt von MongoDB generiert —
|
||||
// so ist die ID bereits bekannt bevor die DB antwortet (z.B. für den WebSocket-Broadcast).
|
||||
// createdAt wird serverseitig gesetzt — dem Client-Timestamp wird nicht vertraut.
|
||||
async addMessage({ id, room, userId, username, text }) {
|
||||
const message = {
|
||||
_id: id,
|
||||
@@ -196,9 +220,11 @@ async function createMongoStore() {
|
||||
createdAt: new Date(),
|
||||
}
|
||||
await messages.insertOne(message)
|
||||
// Lokales Objekt zurückgeben statt nochmal aus der DB lesen — spart einen Query.
|
||||
return normalizeMessage(message)
|
||||
},
|
||||
|
||||
// limit = 100 verhindert dass bei einem langen Chatverlauf alles auf einmal geladen wird.
|
||||
async getMessages(room, limit = 100) {
|
||||
const docs = await messages
|
||||
.find({ room })
|
||||
@@ -208,6 +234,27 @@ async function createMongoStore() {
|
||||
return docs.map(normalizeMessage)
|
||||
},
|
||||
|
||||
// Aggregation Pipeline: MongoDB berechnet das Ergebnis serverseitig — effizienter als
|
||||
// alle Daten laden und in JavaScript auswerten.
|
||||
// $group: Gruppiert User nach Room (wie GROUP BY in SQL).
|
||||
// $cond: Zählt nur Online-User pro Room.
|
||||
// $project: Blendet _id aus und benennt es in "name" um.
|
||||
async getRooms() {
|
||||
const result = await users.aggregate([
|
||||
{
|
||||
$group: {
|
||||
_id: '$room',
|
||||
participantCount: { $sum: { $cond: ['$isOnline', 1, 0] } },
|
||||
},
|
||||
},
|
||||
{ $project: { _id: 0, name: '$_id', participantCount: 1 } },
|
||||
{ $sort: { name: 1 } },
|
||||
]).toArray()
|
||||
return result
|
||||
},
|
||||
|
||||
// Verbindung sauber trennen beim Herunterfahren der App — ohne close() bleiben
|
||||
// offene Verbindungen auf dem MongoDB-Server hängen.
|
||||
async disconnect() {
|
||||
await client.close()
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user