Something went wrong. Try again.
Monorepo for Aesthetic.Computer aesthetic.computer
Something went wrong. Try again.
45 kB · 1288 lines
JavaScript
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289// 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 { createHmac, timingSafeEqual } from "crypto";
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).import { MAX_CHARS, chatCapabilities, profanityFiltered,} from "../shared/chat-capabilities.mjs";
const MAX_MESSAGES = 500;
// A dead or unreachable Mongo should cost a chat connection seconds, not the// driver's default 30s of server selection — several awaits sit on the path// between a socket opening and its `connected` handshake.const mongoClientOptions = { serverSelectionTimeoutMS: 8000 };
// 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", // Dedicated topic — sotce chat pings must never fan out to the shared // "mood" topic that aesthetic.computer devices subscribe to. topic: "chat-sotce", // And its own face in the notification, not the aesthetic.computer mark. icon: "https://assets.aesthetic.computer/sotce-net/cookie-192.png", }, "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, mongoClientOptions); 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"); }
// Chat must outlive its database: messages that fail to persist stay in // memory (`sub`, no `id`) and this flush retries them until Mongo returns. // Without it, recovery only happened at shutdown — a whole outage's worth // of messages rode on the process exiting cleanly. if (!this.dev && this.mongoConnectionString) { this.persistTimer = setInterval(async () => { const stranded = Object.values(this.instances).reduce( (n, instance) => n + instance.messages.filter((m) => m.sub && !m.id).length, 0, ); if (stranded === 0) return; console.log(`💬 Retrying persistence for ${stranded} stranded message(s)...`); try { await this.persistAllMessages(); } catch (err) { console.error("💬 Persistence retry failed:", err.message); } }, 60000); } }
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.via) msg.via = message.via; // 📺 e.g. "youtube" — service relays if (message.link) msg.link = message.link; // 📺 walk-back to the live chat 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. This await sits between // the socket opening and the `connected` packet — if it throws, the client // never finishes its handshake and chat reads as offline (the 2026-08-12 // credential-rotation outage). Hearts are decoration; degrade to none. let heartCounts = {}; if (this.db) { try { const messageIds = instance.messages.filter((m) => m.id).map((m) => m.id); heartCounts = await countHearts(this.db, instance.config.name, messageIds); } catch (err) { console.error(`💬 [${instance.config.name}] Heart counts unavailable, connecting without:`, err.message); } }
// 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, // The handshake: what this channel accepts. Clients read their limit // and syntax from here instead of hardcoding a copy of it. capabilities: chatCapabilities(instance.config.name), 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:service-message") { await this.handleServiceMessage(instance, ws, id, msg); } else if (msg.type === "chat:delete") { await this.handleDeleteMessage(instance, ws, id, msg); } else if (msg.type === "chat:edit") { await this.handleEditMessage(instance, ws, id, msg); } else if (msg.type === "chat:heart") { await this.handleChatHeart(instance, ws, id, msg); } }
// 📺 Service relays (the TV chat bridge) post on behalf of off-AC visitors — // e.g. YouTube live-chat viewers of the always-on broadcasts. They authorize // with a shared secret (CHAT_SERVICE_SECRET env), not a user token, and their // messages carry `via` (e.g. "youtube") + the visitor's display name as // `from`, so clients can render them as televised guests rather than handles. // No activity feed, no push notifications — guests are heard, not amplified. async handleServiceMessage(instance, ws, id, msg) { const secret = process.env.CHAT_SERVICE_SECRET; if (!secret || msg.content?.secret !== secret) { console.error(`💬 [${instance.config.name}] Service message rejected (bad secret)`); ws.send(this.pack("unauthorized", { message: "Bad service secret." }, id)); return; }
const via = String(msg.content.via || "youtube").slice(0, 16); // Visitor names are foreign input: strip control chars and our color-code // delimiter, clamp, and fall back rather than ever posting an empty name. const name = String(msg.content.name || "") .replace(/[\u0000-\u001f\\@]/g, "") .trim() .slice(0, 24) || "viewer"; let text = String(msg.content.text || "").trim(); if (!text) return; if (text.length > MAX_CHARS) text = text.slice(0, MAX_CHARS); if (profanityFiltered(instance.config.name)) text = filter(text, this.filterDebug);
// Optional walk-back link (the broadcast's popout chat) — YouTube only. let link; const rawLink = String(msg.content.link || "").slice(0, 160); if (/^https:\/\/(www\.youtube\.com|youtu\.be)\//.test(rawLink)) link = rawLink;
let when = new Date(); if (!this.dev) { try { const clockResponse = await fetch("https://aesthetic.computer/api/clock"); if (clockResponse.ok) when = new Date(await clockResponse.text()); } catch (err) { console.log("💬 Clock fetch failed, using local time"); } }
let insertedId; if (!this.dev && this.db) { try { const collection = this.db.collection(instance.config.name); const result = await collection.insertOne({ from: name, via, text, when, font: "font_1", ...(link ? { link } : {}) }); insertedId = result.insertedId?.toString(); } catch (err) { console.error(`💬 [${instance.config.name}] Service message store failed, broadcasting anyway:`, err.message); } }
const out = { from: name, via, text, when, font: "font_1" }; if (link) out.link = link; if (insertedId) out.id = insertedId; instance.messages.push(out); if (instance.messages.length > MAX_MESSAGES) instance.messages.shift(); this.broadcast(instance, this.pack("message", out)); console.log(`💬 [${instance.config.name}] 📺 via ${via}: ${name} (${text.length} chars)`); }
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. Counted in UTF-16 code units — see shared/chat-capabilities. if (msg.content.text.length > MAX_CHARS) { ws.send( this.pack("too-long", { message: `Please limit to ${MAX_CHARS} characters.`, maxChars: MAX_CHARS, countedAs: "utf16-code-units", was: msg.content.text.length, }), ); 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); } // An account waiting to be deleted is locked out of chat too // (system/backend/account-deletion.mjs). if (authorized?.sub && (await this.accountLocked(authorized.sub))) { console.log("💬 Refused locked account"); authorized = undefined; } 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 = profanityFiltered(instance.config.name) ? filter(message.text, this.filterDebug) : message.text; }
// 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). Persistence is // best-effort: a failure here must not stop the broadcast below — the // message stays in memory with `sub` and no `id`, which is exactly what // the periodic persistAllMessages flush looks for once Mongo recovers. let insertedId; if (!this.dev && !userIsMuted && this.db) { try { 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"); } catch (err) { console.error(`💬 [${instance.config.name}] Message store failed, broadcasting anyway:`, err.message); } }
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-sotce rides its // own "chat-sotce" topic, which only sotce.net readers opt into. if (!this.dev && !userIsMuted) { 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?.sub && (await this.accountLocked(authorized.sub))) authorized = undefined; 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 })); }
// ✏️ Re-edit your own message in place. Mirrors delete's auth + ownership // checks and message-send's length/profanity rules; DB keeps the raw text // (like handleChatMessage) while the broadcast carries the filtered copy. async handleEditMessage(instance, ws, id, msg) { const { token, sub, id: messageId } = msg.content || {}; let text = msg.content?.text; console.log( `💬 [${instance.config.name}] Edit request from ${sub} for message ${messageId}` );
if (!messageId || typeof text !== "string") { ws.send(this.pack("error", { message: "Missing message id or text." })); return; }
text = text.replace(/\s+$/, ""); if (text.length === 0) { ws.send(this.pack("error", { message: "Nothing to say." })); return; } if (text.length > MAX_CHARS) { ws.send( this.pack("too-long", { message: `Please limit to ${MAX_CHARS} characters.`, maxChars: MAX_CHARS, countedAs: "utf16-code-units", was: text.length, }), ); return; }
if (await this.isMuted(instance, sub)) { ws.send(this.pack("muted", { message: "Your user has been muted." })); 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?.sub && (await this.accountLocked(authorized.sub))) authorized = undefined; 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 edit your own messages." })); return; } if (message.deleted) { ws.send(this.pack("error", { message: "Deleted messages can't be edited." })); return; }
const filteredText = profanityFiltered(instance.config.name) ? filter(text, this.filterDebug) : text; const editedWhen = new Date();
// Update MongoDB (raw text, matching how sends are stored) if (this.db) { try { const collection = this.db.collection(instance.config.name); await collection.updateOne( { _id: new ObjectId(messageId) }, { $set: { text, edited: true, editedWhen } } ); console.log("💬 Message edited in DB"); } catch (err) { console.error("💬 Failed to edit message in DB:", err); } }
// Update in-memory message.text = filteredText; message.edited = true; message.editedWhen = editedWhen; delete message.redactedText;
// Broadcast the edit to all clients this.broadcast( instance, this.pack("message:edit", { id: messageId, text: filteredText, edited: true, editedWhen, }), ); }
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?.sub && (await this.accountLocked(authorized.sub))) authorized = undefined; 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", signed by // system/netlify/functions/device-token.mjs with JWT_SECRET over // "sub:timestamp" (JWT_SECRET, else AC_DEVICE_SECRET — lith's key). // Without the secret nothing can be verified, so device tokens are // refused rather than trusted. const secret = process.env.JWT_SECRET || process.env.AC_DEVICE_SECRET; const parts = /^([0-9a-f]{32})\.([0-9a-z]+)$/.exec(token || ""); if (!secret || !parts || !handle) return undefined; try { const cleanHandle = handle.replace("@", ""); const doc = await this.db.collection("@handles").findOne({ handle: cleanHandle }); const expected = doc?._id ? createHmac("sha256", secret).update(`${doc._id}:${parts[2]}`).digest("hex").slice(0, 32) : ""; if (expected && timingSafeEqual(Buffer.from(expected), Buffer.from(parts[1]))) { 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 accountLocked(sub) { try { const pending = await this.db?.collection("account-deletions").findOne( { _id: sub, state: { $in: ["scheduled", "running", "failed"] } }, { projection: { _id: 1 } }, ); return !!pending; } catch { return false; } }
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, icon: instance.config.icon, // undefined keeps the default mark 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'}`);
// A deleted account (system/backend/account-deletion.mjs) is not a log // line: drop its messages from memory instead of posting anything. if (parsed.action === "account:erase") { this.eraseAccount(instance, parsed.users?.[0]); return { status: 200, body: { status: "success", message: "Account erased" } }; }
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" } }; } }
eraseAccount(instance, sub) { if (!sub) return; instance.messages = instance.messages.filter( (msg) => msg.sub !== sub && !(msg.users || []).includes(sub), ); delete instance.subsToHandles[sub]; // Connected clients redact a muted user's messages; reuse that so the // erased text leaves open screens too. this.broadcast(instance, this.pack("chat-system:mute", { user: sub })); }
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, })); }
// The same handshake the `connected` packet carries, for callers with no // socket. `host` may be a chat host or a bare channel name; without one you // get every channel, since the profanity policy differs between them. getCapabilities(host = null) { if (host) { const instance = this.getInstance(host) || Object.values(this.instances).find((i) => i.config.name === host); if (!instance) return null; return { host: instance.config.allowedHost, ...chatCapabilities(instance.config.name), }; }
return Object.values(this.instances).map((instance) => ({ host: instance.config.allowedHost, ...chatCapabilities(instance.config.name), })); }
// 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, mongoClientOptions); 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; // Stamp each message with its new id so it stops matching the // unpersisted filter — this runs on a timer now, and without the // stamp every pass would insert the same messages again. for (const [i, insertedId] of Object.entries(result.insertedIds)) { if (unpersisted[i]) unpersisted[i].id = insertedId.toString(); } console.log(`💬 [${collectionName}] Persisted ${result.insertedCount} messages`); } catch (err) { // insertMany with ordered:false continues on duplicate errors if (err.insertedCount) totalPersisted += err.insertedCount; if (err.result?.insertedIds) { for (const [i, insertedId] of Object.entries(err.result.insertedIds)) { if (unpersisted[i]) unpersisted[i].id = insertedId.toString(); } } console.error(`💬 [${collectionName}] Persistence error:`, err.message); } }
return totalPersisted; }
async shutdown() { console.log("💬 ChatManager shutting down..."); if (this.persistTimer) clearInterval(this.persistTimer); const count = await this.persistAllMessages(); console.log(`💬 Persisted ${count} total messages`);
if (this.mongoClient) { await this.mongoClient.close(); console.log("💬 MongoDB connection closed"); } }}