Something went wrong. Try again.
Monorepo for Aesthetic.Computer aesthetic.computer
Something went wrong. Try again.
32 kB · 973 lines
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974// Chat Manager, 25.11.28// Multi-instance chat support for session-server// Adapted from nanos/chat.mjs to run multiple chat instances in one process
import { WebSocket } from "ws";import { promises as fs } from "fs";import fetch from "node-fetch";import https from "https";
import { filter } from "./filter.mjs";import { redact, unredact } from "./redact.mjs";import { ensureIndexes as ensureHeartsIndexes, toggleHeart, countHearts } from "./hearts.mjs";
import { MongoClient, ObjectId } from "mongodb";import { broadcastToTopic } from "../shared/push.mjs"; // Standard push (no Firebase).
const MAX_MESSAGES = 500;
// Chat instance configurationsexport const chatInstances = { "chat-system.aesthetic.computer": { name: "chat-system", allowedHost: "chat-system.aesthetic.computer", userInfoEndpoint: "https://aesthetic.us.auth0.com/userinfo", topic: "mood", }, "chat.sotce.net": { name: "chat-sotce", allowedHost: "chat.sotce.net", userInfoEndpoint: "https://sotce.us.auth0.com/userinfo", topic: "mood", }, "chat-clock.aesthetic.computer": { name: "chat-clock", allowedHost: "chat-clock.aesthetic.computer", userInfoEndpoint: "https://aesthetic.us.auth0.com/userinfo", topic: "mood", },};
// Development host mappings (localhost ports)const devHostMappings = { "localhost:8083": "chat-system.aesthetic.computer", "localhost:8084": "chat.sotce.net", "localhost:8085": "chat-clock.aesthetic.computer",};
export class ChatManager { constructor(options = {}) { this.dev = options.dev || false; this.filterDebug = options.filterDebug || false; this.loggerKey = options.loggerKey || process.env.LOGGER_KEY; this.activityEmitter = options.activityEmitter || null; // MongoDB this.mongoClient = null; this.db = null; this.mongoConnectionString = options.mongoConnectionString || process.env.MONGODB_CONNECTION_STRING; this.mongoDbName = options.mongoDbName || process.env.MONGODB_NAME || "aesthetic"; // HTTPS agent for dev mode this.agent = this.dev ? new https.Agent({ rejectUnauthorized: false }) : null; // Per-instance state this.instances = {}; for (const [host, config] of Object.entries(chatInstances)) { this.instances[host] = { config, messages: [], connections: {}, connectionId: 0, authorizedConnections: {}, subsToHandles: {}, subsToSubscribers: {}, }; } console.log("💬 ChatManager initialized with instances:", Object.keys(chatInstances).join(", ")); }
setActivityEmitter(emitter) { this.activityEmitter = typeof emitter === "function" ? emitter : null; }
emitActivity(payload) { if (!this.activityEmitter) return; try { this.activityEmitter(payload); } catch (err) { console.error("💬 Activity emitter failure:", err); } }
async init() { // Connect to MongoDB if (this.mongoConnectionString) { try { console.log("💬 Connecting to MongoDB..."); this.mongoClient = new MongoClient(this.mongoConnectionString); await this.mongoClient.connect(); this.db = this.mongoClient.db(this.mongoDbName); console.log("💬 MongoDB connected!"); // Ensure hearts indexes exist await ensureHeartsIndexes(this.db);
// Load messages for each instance for (const [host, instance] of Object.entries(this.instances)) { await this.loadMessages(instance); } } catch (err) { console.error("💬 MongoDB connection failed:", err); } } else { console.log("💬 No MongoDB connection string, running without persistence"); } }
async loadMessages(instance) { if (!this.db) return; const collectionName = instance.config.name; console.log(`💬 Loading messages for ${collectionName}...`); try { const chatCollection = this.db.collection(collectionName); let combinedMessages;
if (collectionName === "chat-sotce") { combinedMessages = (await chatCollection .find({}) .sort({ when: -1 }) .limit(MAX_MESSAGES) .toArray()).reverse(); } else if (collectionName !== "chat-system") { combinedMessages = (await chatCollection .find({}) .sort({ when: -1 }) .limit(MAX_MESSAGES) .toArray()).reverse(); } else { // chat-system includes logs combinedMessages = ( await chatCollection .aggregate([ { $unionWith: { coll: "logs", pipeline: [{ $match: {} }], }, }, { $sort: { when: -1 } }, { $limit: MAX_MESSAGES }, ]) .toArray() ).reverse(); }
// Deduplicate messages (removes duplicates caused by persist-on-shutdown bug) { const seen = new Set(); combinedMessages = combinedMessages.filter((msg) => { const key = `${msg.user || msg.from}:${msg.text}:${msg.when?.getTime?.() ?? msg.when}`; if (seen.has(key)) return false; seen.add(key); return true; }); }
// Check mutes for chat-system if (collectionName === "chat-system") { for (const msg of combinedMessages) { if (await this.isMuted(instance, msg.user)) { redact(msg); } } }
instance.messages = []; for (const message of combinedMessages) { let from; if (message.user) { from = await this.getHandleFromSub(instance, message.user); } else { from = message.from || "deleted"; }
const msg = { from, text: message.deleted ? "[deleted]" : (instance.config.name === "chat-clock" ? message.text : filter(message.text, this.filterDebug)) || "message forgotten", redactedText: message.redactedText, when: message.when, sub: message.user || undefined, font: message.font || "font_1", // 🔤 Include font from DB (default for old messages) }; if (message._id) msg.id = message._id.toString(); if (message.deleted) msg.deleted = true; instance.messages.push(msg); } console.log(`💬 Loaded ${instance.messages.length} messages for ${collectionName}`); } catch (err) { console.error(`💬 Failed to load messages for ${collectionName}:`, err); } }
// Check if a host should be handled by chat manager isChatHost(host) { // Direct match if (chatInstances[host]) return true; // Dev mapping if (this.dev && devHostMappings[host]) return true; return false; }
// Get the instance for a host getInstance(host) { // Direct match if (this.instances[host]) return this.instances[host]; // Dev mapping if (this.dev && devHostMappings[host]) { return this.instances[devHostMappings[host]]; } return null; }
// Handle a new WebSocket connection async handleConnection(ws, req) { const host = req.headers.host; const instance = this.getInstance(host); if (!instance) { console.log("💬 Unknown chat host:", host); ws.close(1008, "Unknown host"); return; }
// Validate host in production if (!this.dev && host !== instance.config.allowedHost) { ws.close(1008, "Policy violation"); return; }
const id = instance.connectionId++; instance.connections[id] = ws; const ip = req.socket.remoteAddress || "localhost"; ws.isAlive = true;
ws.on("pong", () => { ws.isAlive = true; });
console.log( `💬 [${instance.config.name}] Connection ${id} from ${ip}, online: ${Object.keys(instance.connections).length}` );
// Set up ping interval for this connection const pingInterval = setInterval(() => { if (ws.isAlive === false) { clearInterval(pingInterval); return ws.terminate(); } ws.isAlive = false; ws.ping(); }, 15000);
ws.on("message", async (data) => { await this.handleMessage(instance, ws, id, data); });
ws.on("close", () => { clearInterval(pingInterval); delete instance.connections[id]; delete instance.authorizedConnections[id];
console.log( `💬 [${instance.config.name}] Connection ${id} closed, online: ${Object.keys(instance.connections).length}` );
// Broadcast updated online handles after removing this connection this.broadcastOnlineHandles(instance); this.broadcast(instance, this.pack("left", { chatters: Object.keys(instance.connections).length, handles: this.getOnlineHandles(instance) }, id)); });
// Load heart counts for current message window let heartCounts = {}; if (this.db) { const messageIds = instance.messages.filter((m) => m.id).map((m) => m.id); heartCounts = await countHearts(this.db, instance.config.name, messageIds); }
// Send welcome message ws.send( this.pack( "connected", { message: `Joined \`${instance.config.name}\` • 🧑🤝🧑 ${Object.keys(instance.connections).length}`, chatters: Object.keys(instance.connections).length, handles: this.getOnlineHandles(instance), messages: instance.messages, heartCounts, id, }, id, ), );
// Notify others this.broadcastOthers(instance, ws, this.pack( "joined", { text: `${id} has joined. Connections open: ${Object.keys(instance.connections).length}`, chatters: Object.keys(instance.connections).length, handles: this.getOnlineHandles(instance), }, id, )); }
async handleMessage(instance, ws, id, data) { let msg; try { msg = JSON.parse(data.toString()); } catch (err) { console.log("💬 Failed to parse message:", err); return; }
msg.id = id;
if (msg.type === "logout") { console.log(`💬 [${instance.config.name}] User logged out`); delete instance.authorizedConnections[id]; } else if (msg.type === "chat:message") { await this.handleChatMessage(instance, ws, id, msg); } else if (msg.type === "chat:delete") { await this.handleDeleteMessage(instance, ws, id, msg); } else if (msg.type === "chat:heart") { await this.handleChatHeart(instance, ws, id, msg); } }
async handleChatMessage(instance, ws, id, msg) { console.log( `💬 [${instance.config.name}] Message from ${msg.content.sub} (${msg.content.text.length} chars)` );
// Mute check if (await this.isMuted(instance, msg.content.sub)) { ws.send(this.pack("muted", { message: "Your user has been muted." })); return; }
// Length limit const len = 128; if (msg.content.text.length > len) { ws.send(this.pack("too-long", { message: `Please limit to ${len} characters.` })); return; }
// Authorization (Auth0 token or AC device token) let authorized; if (instance.authorizedConnections[id]?.token === msg.content.token) { authorized = instance.authorizedConnections[id].user; console.log("💬 Pre-authorized"); } else { console.log("💬 Authorizing...", "handle:", msg.content.handle, "token:", msg.content.token?.slice(0, 10) + "...", "content-type:", typeof msg.content); authorized = await this.authorize(instance, msg.content.token); // Fallback: AC device token (from ac-native device-token API) if (!authorized && msg.content.handle && msg.content.token) { console.log("💬 Trying device token auth for @" + msg.content.handle); authorized = await this.authorizeDeviceToken(instance, msg.content.handle, msg.content.token); } if (authorized) { instance.authorizedConnections[id] = { token: msg.content.token, user: authorized, }; } }
// Get handle let handle, subscribed; if (authorized) { handle = await this.getHandleFromSub(instance, authorized.sub); // Store handle in authorizedConnections for online list instance.authorizedConnections[id].handle = handle; // Broadcast updated online handles to all clients this.broadcastOnlineHandles(instance); // Subscription check for sotce if (instance.config.name === "chat-sotce") { subscribed = await this.checkSubscription(instance, authorized.sub, msg.content.token); } else { subscribed = true; } }
if (!authorized || !handle || !subscribed) { console.error("💬 Unauthorized:", { authorized: !!authorized, handle, subscribed }); ws.send(this.pack("unauthorized", { message: "Please login and/or subscribe." }, id)); return; }
// Process and store message try { const message = msg.content; const fromSub = message.sub; let filteredText; const userIsMuted = await this.isMuted(instance, fromSub);
if (userIsMuted) { redact(message); filteredText = message.text; } else { filteredText = instance.config.name === "chat-clock" ? message.text : filter(message.text, this.filterDebug); }
// Get server time let when = new Date(); if (!this.dev) { try { const clockResponse = await fetch("https://aesthetic.computer/api/clock"); if (clockResponse.ok) { const serverTimeISO = await clockResponse.text(); when = new Date(serverTimeISO); } } catch (err) { console.log("💬 Clock fetch failed, using local time"); } }
// Store in MongoDB (production only, non-muted users) let insertedId; if (!this.dev && !userIsMuted && this.db) { const dbmsg = { user: message.sub, text: message.text, when, font: message.font || "font_1", // 🔤 Store user's font preference }; const collection = this.db.collection(instance.config.name); await collection.createIndex({ when: 1 }); const result = await collection.insertOne(dbmsg); insertedId = result.insertedId?.toString(); console.log("💬 Message stored"); }
const out = { from: handle, text: filteredText, redactedText: message.redactedText, when, sub: fromSub, font: message.font || "font_1", // 🔤 Include font in broadcast }; if (insertedId) out.id = insertedId;
// Duplicate detection const lastMsg = instance.messages[instance.messages.length - 1]; if (lastMsg && lastMsg.sub === out.sub && lastMsg.text === out.text && !lastMsg.count) { lastMsg.count = (lastMsg.count || 1) + 1; this.broadcast(instance, this.pack("message:update", { index: instance.messages.length - 1, count: lastMsg.count })); } else { instance.messages.push(out); if (instance.messages.length > MAX_MESSAGES) instance.messages.shift(); this.broadcast(instance, this.pack("message", out)); }
if (handle && handle.startsWith("@")) { const compactText = `${filteredText || ""}`.replace(/\s+/g, " ").trim(); const preview = compactText.length > 80 ? `${compactText.slice(0, 77)}...` : compactText;
this.emitActivity({ handle, event: { type: "chat", when: when?.getTime?.() || Date.now(), label: `Chat ${instance.config.name.replace("chat-", "")}: ${preview}`, piece: instance.config.name, ref: insertedId || null, text: compactText, }, countsDelta: { chats: 1 }, }); }
// Push notification (production only, non-muted, chat-system and chat-clock only) // Note: chat-sotce is intentionally excluded from push notifications if (!this.dev && !userIsMuted && instance.config.name !== "chat-sotce") { this.notify(instance, handle, filteredText, when); } } catch (err) { console.error("💬 Message handling error:", err); } }
async handleDeleteMessage(instance, ws, id, msg) { const { token, sub, id: messageId } = msg.content; console.log( `💬 [${instance.config.name}] Delete request from ${sub} for message ${messageId}` );
if (!messageId) { ws.send(this.pack("error", { message: "Missing message id." })); return; }
// Authorize the user let authorized; if (instance.authorizedConnections[id]?.token === token) { authorized = instance.authorizedConnections[id].user; } else { authorized = await this.authorize(instance, token); if (authorized) { instance.authorizedConnections[id] = { token, user: authorized }; } }
if (!authorized || authorized.sub !== sub) { ws.send(this.pack("unauthorized", { message: "Please login." }, id)); return; }
// Find the message in memory and verify ownership const message = instance.messages.find((m) => m.id === messageId); if (!message) { ws.send(this.pack("error", { message: "Message not found." })); return; } if (message.sub !== sub) { ws.send(this.pack("error", { message: "You can only delete your own messages." })); return; }
// Already deleted if (message.deleted) return;
// Soft-delete in MongoDB if (this.db) { try { const collection = this.db.collection(instance.config.name); await collection.updateOne( { _id: new ObjectId(messageId) }, { $set: { deleted: true } } ); console.log("💬 Message soft-deleted in DB"); } catch (err) { console.error("💬 Failed to delete message in DB:", err); } }
// Update in-memory message.text = "[deleted]"; message.deleted = true; delete message.redactedText;
// Broadcast deletion to all clients this.broadcast(instance, this.pack("message:delete", { id: messageId })); }
async handleChatHeart(instance, ws, id, msg) { const { for: forId, token } = msg.content || {}; if (!forId || !token) return;
// Reuse cached auth or re-authorize let authorized; if (instance.authorizedConnections[id]?.token === token) { authorized = instance.authorizedConnections[id].user; } else { authorized = await this.authorize(instance, token); if (authorized) { instance.authorizedConnections[id] = { token, user: authorized }; } }
if (!authorized) { ws.send(this.pack("unauthorized", { message: "Please login." }, id)); return; }
if (!this.db) return;
try { const { hearted, count } = await toggleHeart(this.db, { user: authorized.sub, type: instance.config.name, for: forId, }); console.log(`💬 [${instance.config.name}] Heart ${hearted ? "+" : "-"} on ${forId} → ${count}`); this.broadcast(instance, this.pack("message:hearts", { for: forId, count })); } catch (err) { console.error("💬 Heart error:", err); } }
async authorizeDeviceToken(instance, handle, token) { // AC device tokens are "hmac.timestamp" — validate by looking up handle in MongoDB if (!token || !token.includes(".") || !handle) return undefined; try { const cleanHandle = handle.replace("@", ""); const doc = await this.db.collection("@handles").findOne({ handle: cleanHandle }); if (doc && doc._id) { console.log("💬 Device token authorized for @" + cleanHandle + " (sub: " + doc._id + ")"); return { sub: doc._id }; } console.log("💬 Device token: handle @" + cleanHandle + " not found in DB"); return undefined; } catch (err) { console.error("💬 Device token auth error:", err); return undefined; } }
async authorize(instance, token) { try { const response = await fetch(instance.config.userInfoEndpoint, { headers: { Authorization: "Bearer " + token, "Content-Type": "application/json", }, });
if (response.status === 200) { return response.json(); } return undefined; } catch (err) { console.error("💬 Authorization error:", err); return undefined; } }
async getHandleFromSub(instance, fromSub) { if (instance.subsToHandles[fromSub]) { return "@" + instance.subsToHandles[fromSub]; }
try { let prefix = instance.config.name === "chat-sotce" ? "sotce-" : ""; let host = this.dev ? "https://localhost:8888" : (instance.config.name === "chat-sotce" ? "https://sotce.net" : "https://aesthetic.computer");
const options = {}; if (this.dev) options.agent = this.agent;
const response = await fetch(`${host}/handle?for=${prefix}${fromSub}`, options); if (response.status === 200) { const data = await response.json(); if (data.handle) { instance.subsToHandles[fromSub] = data.handle; return "@" + data.handle; } } } catch (err) { console.error("💬 Handle lookup error:", err); }
return "@unknown"; }
async checkSubscription(instance, sub, token) { if (instance.subsToSubscribers[sub] !== undefined) { return instance.subsToSubscribers[sub]; }
const host = this.dev ? "https://localhost:8888" : "https://sotce.net"; const options = { method: "POST", body: JSON.stringify({ retrieve: "subscription" }), headers: { Authorization: "Bearer " + token, "Content-Type": "application/json", }, };
if (this.dev) options.agent = this.agent;
try { const response = await fetch(`${host}/sotce-net/subscribed`, options); if (response.status === 200) { const data = await response.json(); instance.subsToSubscribers[sub] = data.subscribed; return data.subscribed; } } catch (err) { console.error("💬 Subscription check error:", err); }
instance.subsToSubscribers[sub] = false; return false; }
async isMuted(instance, sub) { if (!sub || !this.db) return false; try { const mutesCollection = this.db.collection(instance.config.name + "-mutes"); const mute = await mutesCollection.findOne({ user: sub }); return !!mute; } catch (err) { return false; } }
notify(instance, handle, text, when) { let title = handle + " 💬"; if (instance.config.name === "chat-clock") { const getClockEmoji = (date) => { let hour = date.getHours() % 12 || 12; const minutes = date.getMinutes(); const emojiCode = minutes < 30 ? (0x1f550 + hour - 1) : (0x1f55c + hour - 1); return String.fromCodePoint(emojiCode); }; title = handle + " " + getClockEmoji(when); }
if (!this.db) return; broadcastToTopic(this.db, instance.config.topic, { title, body: text, urgent: true, // time-sensitive on iOS, Urgency: high on web ttl: 0, // don't store undelivered chat pings data: { piece: "chat" }, }).then((summary) => { console.log("💬 Notification sent:", summary); }).catch((err) => { console.log("💬 Notification error:", err); }); }
// Handle HTTP log endpoint async handleLog(instance, body, authHeader) { const token = authHeader?.split(" ")[1]; if (token !== this.loggerKey) { return { status: 403, body: { status: "error", message: "Forbidden" } }; }
try { const parsed = typeof body === "string" ? JSON.parse(body) : body; console.log(`💬 [${instance.config.name}] Log received from: ${parsed.from || 'unknown'}`);
instance.messages.push(parsed); if (instance.messages.length > MAX_MESSAGES) instance.messages.shift();
// Handle actions (mute/unmute, handle updates) if (parsed.action) { await this.handleLogAction(instance, parsed); }
this.broadcast(instance, this.pack("message", parsed));
if (instance.config.name === "chat-system") { this.notify(instance, "log 🪵", parsed.text, new Date()); }
return { status: 200, body: { status: "success", message: "Log received" } }; } catch (err) { return { status: 400, body: { status: "error", message: "Malformed log JSON" } }; } }
async handleLogAction(instance, parsed) { let [object, behavior] = (parsed.action || "").split(":"); if (!behavior) { behavior = object; object = null; }
if (object === "chat-system" && (behavior === "mute" || behavior === "unmute")) { const user = parsed.users[0]; if (behavior === "mute") { instance.messages.forEach((msg) => { if (msg.sub === user) redact(msg); }); } else { instance.messages.forEach((msg) => { if (msg.sub === user) unredact(msg); }); } this.broadcast(instance, this.pack(parsed.action, { user })); }
if (object === "handle") { if (behavior === "colors") { // Broadcast handle color changes to all connected clients. const data = JSON.parse(parsed.value); this.broadcast(instance, this.pack("handle:colors", { user: parsed.users[0], handle: data.handle, colors: data.colors })); } else { instance.subsToHandles[parsed.users[0]] = parsed.value;
if (behavior === "update" || behavior === "strip") { const from = behavior === "update" ? "@" + parsed.value : "nohandle"; instance.messages.forEach((msg) => { if (msg.sub === parsed.users[0]) { msg.from = from; } }); this.broadcast(instance, this.pack(parsed.action, { user: parsed.users[0], handle: from })); } } } }
pack(type, content, id) { if (typeof content === "object") content = JSON.stringify(content); return JSON.stringify({ type, content, id }); }
broadcast(instance, message) { Object.values(instance.connections).forEach((c) => { if (c?.readyState === WebSocket.OPEN) c.send(message); }); }
broadcastOthers(instance, exclude, message) { Object.values(instance.connections).forEach((c) => { if (c !== exclude && c?.readyState === WebSocket.OPEN) c.send(message); }); }
// Set the presence resolver function (called from session.mjs) // This allows us to query which users are on a specific piece setPresenceResolver(resolver) { this.presenceResolver = resolver; }
// Get list of online handles for an instance (all connected & authorized) getOnlineHandles(instance) { const handles = []; for (const [id, auth] of Object.entries(instance.authorizedConnections)) { if (auth.handle && instance.connections[id]) { handles.push(auth.handle); } } return [...new Set(handles)]; // Remove duplicates }
// Get handles of users actually viewing the chat piece right now getHereHandles(instance) { if (!this.presenceResolver) return []; // Get all users on "chat" piece from session server const onChatPiece = this.presenceResolver("chat"); // Intersect with authorized handles for this chat instance const onlineHandles = this.getOnlineHandles(instance); return onChatPiece.filter(h => onlineHandles.includes(h)); }
// Broadcast presence data (online + here) to all clients broadcastOnlineHandles(instance) { const online = this.getOnlineHandles(instance); const here = this.getHereHandles(instance); // Send both for backwards compatibility and new "here" feature this.broadcast(instance, this.pack("presence", { handles: online, // backwards compat with old "online-handles" online, // all authorized chat connections here // only those actually on chat piece })); }
// Get status for a specific instance or all getStatus(host = null) { if (host) { const instance = this.getInstance(host); if (!instance) return null; return { name: instance.config.name, connections: Object.keys(instance.connections).length, messages: instance.messages.length, }; }
return Object.entries(this.instances).map(([host, instance]) => ({ host, name: instance.config.name, connections: Object.keys(instance.connections).length, messages: instance.messages.length, })); }
// Get recent messages for a specific instance (for dashboard) getRecentMessages(host, count = 10) { const instance = this.getInstance(host); if (!instance) return []; // Return the most recent messages, but don't expose sensitive data return instance.messages.slice(-count).map(msg => ({ from: msg.from || 'unknown', text: msg.text || '', when: msg.when })).reverse(); // Most recent first }
// Persist any in-memory messages that weren't saved to MongoDB. // Messages received while this.db was null (broken env) have a `sub` field // but were never inserted. Messages loaded from DB on startup do not have `sub`. async persistAllMessages() { // Establish a connection if we don't have one if (!this.db && this.mongoConnectionString) { try { console.log("💬 Connecting to MongoDB for message persistence..."); this.mongoClient = new MongoClient(this.mongoConnectionString); await this.mongoClient.connect(); this.db = this.mongoClient.db(this.mongoDbName); console.log("💬 MongoDB connected for persistence!"); } catch (err) { console.error("💬 Failed to connect to MongoDB for persistence:", err); return 0; } }
if (!this.db) { console.error("💬 No MongoDB connection available, cannot persist messages"); return 0; }
let totalPersisted = 0;
for (const [, instance] of Object.entries(this.instances)) { const collectionName = instance.config.name; // Only persist messages that were never saved to DB. // DB-loaded and successfully-stored messages have an `id` (from MongoDB _id). // Messages received while DB was unavailable have `sub` but no `id`. const unpersisted = instance.messages.filter(msg => msg.sub && !msg.id);
if (unpersisted.length === 0) { console.log(`💬 [${collectionName}] No unpersisted messages`); continue; }
try { const collection = this.db.collection(collectionName); const docs = unpersisted.map(msg => ({ user: msg.sub, text: msg.text, when: msg.when, font: msg.font || "font_1", }));
const result = await collection.insertMany(docs, { ordered: false }); totalPersisted += result.insertedCount; console.log(`💬 [${collectionName}] Persisted ${result.insertedCount} messages`); } catch (err) { // insertMany with ordered:false continues on duplicate errors if (err.insertedCount) totalPersisted += err.insertedCount; console.error(`💬 [${collectionName}] Persistence error:`, err.message); } }
return totalPersisted; }
async shutdown() { console.log("💬 ChatManager shutting down..."); const count = await this.persistAllMessages(); console.log(`💬 Persisted ${count} total messages`);
if (this.mongoClient) { await this.mongoClient.close(); console.log("💬 MongoDB connection closed"); } }}