Search - MatildaTv

const express = require('express'); const crypto = require('crypto'); const cookie = require('cookie'); const { v4: uuidv4 } = require('uuid'); const axios = require('axios'); const cheerio = require('cheerio'); const DEFAULT_PAGE_SIZE = 25; const MAX_PAGE_SIZE = 100; const MAX_ATTACHMENT_SIZE_BYTES = 2.5 * 1024 * 1024; // 2.5 MB per attachment const SUPPORTED_ATTACHMENT_TYPES = new Set(['image/png', 'image/jpeg', 'image/gif', 'image/webp']); const SYSTEM_USER_ID = 0; /** * Initialize messaging module * @param {Object} deps * @param {express.Application} deps.app * @param {import('socket.io').Server} deps.io * @param {Function} deps.query - promisified MySQL query (same shape as mysql2) * @param {Function} deps.log - logger from main server * @param {Object} deps.config - configuration object from app_start * @param {Function} deps.getUserData - helper to fetch user objects * @param {Map>} deps.onlineUsers - map maintained by core server * @returns {{ registerSocketHandlers: (socket: import('socket.io').Socket) => void }} */ function init({ app, io, query, log, config, getUserData, onlineUsers }) { log('INFO', 'Initializing messaging module...'); const masterSecret = config.messaging_master_secret || process.env.MESSAGING_MASTER_SECRET; const oneSignalAppId = config.onesignal_app_id || process.env.ONESIGNAL_APP_ID; const oneSignalRestKey = config.onesignal_rest_key || process.env.ONESIGNAL_REST_KEY; const pushSound = config.messaging_push_sound || 'stream_notification.mp3'; if (!masterSecret) { log('WARN', 'Messaging master secret is not configured. Messages will fail to encrypt.'); } /** * Helpers */ const deriveConversationKey = (userIdA, userIdB) => { if (!masterSecret) { throw new Error('Missing messaging master secret'); } const [minId, maxId] = [Number(userIdA), Number(userIdB)].sort((a, b) => a - b); return crypto .createHmac('sha256', masterSecret) .update(`${minId}:${maxId}`) .digest(); }; const encryptPayload = (payload, key, type = 'text', meta = {}) => { const iv = crypto.randomBytes(12); const cipher = crypto.createCipheriv('aes-256-gcm', key, iv); const serialized = Buffer.from(JSON.stringify(payload), 'utf8'); const encrypted = Buffer.concat([cipher.update(serialized), cipher.final()]); const authTag = cipher.getAuthTag(); return { v: 1, type, iv: iv.toString('base64'), cipher: encrypted.toString('base64'), tag: authTag.toString('base64'), meta }; }; const decryptPayload = (envelope, key) => { if (!envelope || typeof envelope !== 'object') { return null; } try { if (!envelope.iv || !envelope.cipher || !envelope.tag) { log('ERROR', 'Missing envelope fields', { has_iv: !!envelope.iv, has_cipher: !!envelope.cipher, has_tag: !!envelope.tag }); return null; } const iv = Buffer.from(envelope.iv, 'base64'); const cipherText = Buffer.from(envelope.cipher, 'base64'); const tag = Buffer.from(envelope.tag, 'base64'); const decipher = crypto.createDecipheriv('aes-256-gcm', key, iv); decipher.setAuthTag(tag); const decrypted = Buffer.concat([decipher.update(cipherText), decipher.final()]); const parsed = JSON.parse(decrypted.toString('utf8')); return parsed; } catch (err) { const errorType = err.code === 'ERR_CRYPTO_INVALID_TAG' ? 'AUTH_TAG_MISMATCH' : err.code === 'ERR_OSSL_BAD_DECRYPT' ? 'DECRYPT_FAILED' : err.name === 'SyntaxError' ? 'JSON_PARSE_ERROR' : 'UNKNOWN'; log('ERROR', 'Failed to decrypt message payload', { error: err.message, error_type: errorType, error_code: err.code, has_key: !!key && key.length === 32 }); return null; } }; const isEnvelope = (value) => { if (!value) return false; if (typeof value !== 'string') return false; if (!value.trim().startsWith('{')) return false; try { const parsed = JSON.parse(value); return parsed && parsed.v === 1 && parsed.cipher && parsed.iv && parsed.tag; } catch (err) { return false; } }; const safeJSONParse = (value, fallback = null) => { if (!value) return fallback; try { return JSON.parse(value); } catch (err) { return fallback; } }; const resolvePrivacy = (userRow) => { if (!userRow) return {}; if (userRow.privacy && typeof userRow.privacy === 'object') { return userRow.privacy; } return safeJSONParse(userRow.privacy, {}) || {}; }; const conversationRoom = (userIdA, userIdB) => { const [minId, maxId] = [Number(userIdA), Number(userIdB)].sort((a, b) => a - b); return `conversation:${minId}:${maxId}`; }; const sanitizeAttachments = (attachments = []) => { if (!Array.isArray(attachments)) return []; return attachments .slice(0, 5) .map((file) => { if (!file || typeof file !== 'object') return null; let { data, mime, name } = file; if (!data || !mime) return null; if (!SUPPORTED_ATTACHMENT_TYPES.has(mime)) return null; // Handle both base64 strings and data URLs let base64Data = data; if (typeof data === 'string' && data.startsWith('data:')) { // Extract base64 from data URL const commaIndex = data.indexOf(','); if (commaIndex > 0) { base64Data = data.substring(commaIndex + 1); } else { return null; // Invalid data URL } } const payloadSize = Buffer.byteLength(base64Data, 'base64'); if (payloadSize > MAX_ATTACHMENT_SIZE_BYTES) return null; return { id: uuidv4(), mime, name: (name && String(name).slice(0, 64)) || 'attachment', data: `data:${mime};base64,${base64Data}` }; }) .filter(Boolean); }; const normalizeMessageRow = async (row, viewerId) => { let envelope = null; let payload = { text: row.text || '' }; // First check if row.text looks like an envelope (even if isEnvelope fails) const textStr = String(row.text || '').trim(); const looksLikeEnvelope = textStr.startsWith('{') && (textStr.includes('"v":1') || textStr.includes('"v": 1')) && textStr.includes('"cipher"') && textStr.includes('"iv"'); if (isEnvelope(row.text) || looksLikeEnvelope) { try { // Check for potential truncation (JSON parse errors at position 65535) if (textStr.length >= 65530 && textStr.length <= 65540) { log('WARN', 'Message text is near TEXT column limit, may be truncated', { message_id: row.id, text_length: textStr.length, from_id: row.from_id, to_id: row.to_id }); } envelope = JSON.parse(row.text); // Ensure we never return the envelope as payload if (envelope && (envelope.v === 1 || envelope.cipher || envelope.iv)) { // This is definitely an envelope, decrypt it const decrypted = decryptPayload(envelope, deriveConversationKey(row.from_id, row.to_id)); if (decrypted && typeof decrypted === 'object' && !decrypted.v && !decrypted.cipher && !decrypted.iv) { // Valid decrypted payload (not an envelope) payload = decrypted; } else { log('WARN', 'Decryption returned invalid payload or envelope, using fallback', { message_id: row.id, has_v: !!decrypted?.v, has_cipher: !!decrypted?.cipher, from_id: row.from_id, to_id: row.to_id }); payload = { text: '[Message could not be decrypted]' }; } } else { log('WARN', 'Invalid envelope structure', { message_id: row.id }); payload = { text: '[Message could not be decrypted]' }; } } catch (err) { log('ERROR', 'Failed to parse or decrypt message envelope', { message_id: row.id, error: err.message, stack: err.stack, from_id: row.from_id, to_id: row.to_id }); payload = { text: '[Message could not be decrypted]' }; } } else { // Not an envelope, but check if it accidentally contains envelope JSON if (looksLikeEnvelope) { log('WARN', 'Text looks like envelope but isEnvelope returned false', { message_id: row.id, text_preview: textStr.substring(0, 100) }); payload = { text: '[Message could not be decrypted]' }; } } // Final safety check: ensure payload is never an envelope object if (payload && (payload.v === 1 || payload.cipher || payload.iv)) { log('ERROR', 'Payload contains envelope structure, replacing with fallback', { message_id: row.id }); payload = { text: '[Message could not be decrypted]' }; } // Ensure payload always has expected structure if (!payload || typeof payload !== 'object') { payload = { text: '[Invalid message format]' }; } if (typeof payload.text !== 'string') { payload.text = payload.text ? String(payload.text) : ''; } // Final check: ensure payload.text is never a stringified envelope if (typeof payload.text === 'string') { const textTrimmed = payload.text.trim(); const isEnvelopeString = textTrimmed.startsWith('{') && (textTrimmed.includes('"v":1') || textTrimmed.includes('"v": 1') || textTrimmed.includes('"v":1,')) && (textTrimmed.includes('"cipher"') || textTrimmed.includes('"cipher":')) && (textTrimmed.includes('"iv"') || textTrimmed.includes('"iv":')); if (isEnvelopeString) { log('ERROR', 'Payload text contains envelope JSON string, replacing with fallback', { message_id: row.id, text_length: textTrimmed.length, text_preview: textTrimmed.substring(0, 150) }); payload.text = '[Message could not be decrypted]'; } } const meta = envelope?.meta || {}; const likes = meta.likes || { count: 0, users: [] }; const thread = meta.thread || null; const deleted = meta.deleted || null; // Add image from separate column to payload (mirroring RTMP stream comments) if (row.image) { // Ensure image is in data URL format const imageData = row.image.includes('data:image') ? row.image : `data:image/jpeg;base64,${row.image}`; // Add image to payload as an attachment (for compatibility with existing frontend) if (!payload.attachments) { payload.attachments = []; } // Only add if not already present (to avoid duplicates) const hasImage = payload.attachments.some(att => att && att.mime && att.mime.startsWith('image/')); if (!hasImage) { payload.attachments.unshift({ id: uuidv4(), mime: 'image/jpeg', name: 'image.jpg', data: imageData }); } } // Fetch sender user data const sender = await getUserData(row.from_id, { withPrivacy: false, withAliases: true }); return { id: row.id, client_id: meta.client_id || null, sender_id: row.from_id, sender: sender ? { id: sender.id, username: sender.username, avatar: sender.avatar, verified: sender.verified, admin: sender.admin || 0 } : null, recipient_id: row.to_id, conversation_key: conversationRoom(row.from_id, row.to_id), time: row.time, seen_at: row.seen || 0, type: envelope?.type || 'text', payload, image: row.image ? (row.image.includes('data:image') ? row.image : `data:image/jpeg;base64,${row.image}`) : null, meta: { likes: { count: likes.count || likes.user_ids?.length || 0, user_ids: likes.user_ids || likes.users || [] }, thread, deleted, attachments_summary: payload?.attachments?.map((item) => ({ id: item.id, mime: item.mime, name: item.name })) || [], giphy: payload?.giphy || null, og: payload?.og || null }, is_legacy: !envelope, capabilities: { can_delete: row.from_id === viewerId, can_like: row.from_id !== viewerId, can_reply: true, is_owner: row.from_id === viewerId } }; }; const authenticateRequest = async (req, requireAdmin = false) => { try { const cookies = cookie.parse(req.headers.cookie || ''); const sessionId = cookies.user_id; if (!sessionId || !/^[a-zA-Z0-9-]+$/.test(sessionId)) { return null; } const [rows] = await query( 'SELECT user_id FROM sessions WHERE session_id = ? AND time > UNIX_TIMESTAMP() - 2592000', [sessionId] ); const numericUserId = rows?.[0]?.user_id; if (!numericUserId) return null; const user = await getUserData(numericUserId, { withPrivacy: true, withAliases: true }); if (!user) return null; if (requireAdmin && !(user.admin > 0)) { return null; } return user; } catch (err) { log('ERROR', 'Failed to authenticate messaging request', { error: err.message }); return null; } }; const ensureConversationRows = async (userId, partnerId) => { const combos = [ [userId, partnerId], [partnerId, userId] ]; for (const [one, two] of combos) { const [existing] = await query( 'SELECT id FROM conversations WHERE user_one = ? AND user_two = ? LIMIT 1', [one, two] ); if (!existing?.[0]) { await query('INSERT INTO conversations (user_one, user_two, time) VALUES (?, ?, UNIX_TIMESTAMP())', [ one, two ]); } } }; const updateConversationTimestamp = async (userId, partnerId) => { const updates = [ query('UPDATE conversations SET time = UNIX_TIMESTAMP() WHERE user_one = ? AND user_two = ?', [ userId, partnerId ]), query('UPDATE conversations SET time = UNIX_TIMESTAMP() WHERE user_one = ? AND user_two = ?', [ partnerId, userId ]) ]; await Promise.all(updates); }; const fetchConversationList = async (userId, { limit = DEFAULT_PAGE_SIZE, cursor = null } = {}) => { const size = Math.min(Math.max(Number(limit) || DEFAULT_PAGE_SIZE, 1), MAX_PAGE_SIZE); const cursorClause = cursor ? 'AND c.time < ?' : ''; const params = [userId]; if (cursor) { params.push(Number(cursor)); } params.push(size + 1); const [rows] = await query( ` SELECT c.user_two AS partner_id, c.time, u.username, u.avatar, u.verified, u.last_active, u.admin FROM conversations c INNER JOIN users u ON u.id = c.user_two WHERE c.user_one = ? ${cursorClause} ORDER BY c.time DESC LIMIT ? `, params ); const hasMore = rows.length > size; if (hasMore) { rows.pop(); } const partners = rows.map((row) => row.partner_id); const lastMessages = partners.length ? await Promise.all( partners.map((partnerId) => query( ` SELECT id, from_id, to_id, text, time, seen, from_deleted, to_deleted FROM messages WHERE ( (from_id = ? AND to_id = ? AND from_deleted = 0) OR (from_id = ? AND to_id = ? AND to_deleted = 0) ) ORDER BY id DESC LIMIT 1 `, [userId, partnerId, partnerId, userId] ).then(([messageRows]) => messageRows?.[0] || null) ) ) : []; const unreadCounts = partners.length ? await Promise.all( partners.map((partnerId) => query( ` SELECT COUNT(*) AS unread FROM messages WHERE from_id = ? AND to_id = ? AND to_deleted = 0 AND (seen IS NULL OR seen = 0) `, [partnerId, userId] ).then(([countRows]) => countRows?.[0]?.unread || 0) ) ) : []; const conversations = await Promise.all( rows.map(async (row, index) => { const partner = await getUserData(row.partner_id, { withPrivacy: false, withAliases: true }); const normalizedLastMessage = lastMessages[index] ? await normalizeMessageRow(lastMessages[index], userId) : null; return { partner_id: row.partner_id, partner: partner ? { id: partner.id, username: partner.username, avatar: partner.avatar, verified: partner.verified, admin: partner.admin || 0, last_active: partner.last_active || row.last_active || 0 } : { id: row.partner_id, username: row.username, avatar: row.avatar, verified: row.verified, admin: row.admin || 0, last_active: row.last_active || 0 }, updated_at: row.time, last_message: normalizedLastMessage, unread: unreadCounts[index] || 0 }; }) ); return { conversations, next_cursor: hasMore ? rows[rows.length - 1].time : null }; }; const fetchMessages = async ( userId, partnerId, { limit = DEFAULT_PAGE_SIZE, before = null, after = null, threadRoot = null } = {} ) => { const size = Math.min(Math.max(Number(limit) || DEFAULT_PAGE_SIZE, 1), MAX_PAGE_SIZE); const where = [ `(from_id = ? AND to_id = ? AND from_deleted = 0) OR (from_id = ? AND to_id = ? AND to_deleted = 0)` ]; const params = [userId, partnerId, partnerId, userId]; if (before) { where.push('id < ?'); params.push(Number(before)); } if (after) { where.push('id > ?'); params.push(Number(after)); } const order = after ? 'ASC' : 'DESC'; const [rows] = await query( ` SELECT id, from_id, to_id, text, image, time, seen, from_deleted, to_deleted FROM messages WHERE ${where.map((clause) => `(${clause})`).join(' AND ')} ORDER BY id ${order} LIMIT ? `, [...params, size] ); const normalized = await Promise.all(rows.map((row) => normalizeMessageRow(row, userId))); if (!after) { normalized.reverse(); } // Don't automatically mark as read when fetching messages // Messages should only be marked as read when user actually views them // (handled by messaging:mark-read socket event) return normalized; }; const validateMessagePermissions = async (sender, recipient) => { if (!recipient) { throw new Error('Recipient not found'); } if (recipient.id === sender.id) { throw new Error('Cannot message yourself'); } const privacy = resolvePrivacy(recipient); const messageRule = privacy.who_can_message_me || 'all'; if (messageRule === 'no_one') { throw new Error('Recipient is not accepting messages'); } if (messageRule === 'subscribers') { const [rows] = await query( 'SELECT COUNT(*) as count FROM subscriptions WHERE user_id = ? AND subscriber_id = ? AND active = 1', [recipient.id, sender.id] ); if (!rows?.[0] || rows[0].count === 0) { throw new Error('Recipient only accepts messages from subscribers'); } } }; const buildMessageEnvelope = (senderId, recipientId, payload, options = {}) => { const type = options.type || 'text'; const meta = { client_id: options.clientId || uuidv4(), likes: { count: 0, user_ids: [] }, thread: options.thread || null, attachments: payload.attachments?.length || 0, deleted: null }; return encryptPayload(payload, deriveConversationKey(senderId, recipientId), type, meta); }; const sendPushNotification = async (targetUserId, payload) => { if (!oneSignalAppId || !oneSignalRestKey) { log('DEBUG', 'Skipping OneSignal notification - missing credentials'); return; } try { const targetUser = await getUserData(targetUserId, { withPrivacy: true, withAliases: true }); if (!targetUser) return; const privacy = resolvePrivacy(targetUser); if (privacy && privacy.message_notifications === 'off') { log('DEBUG', 'User disabled messaging push notifications', { user_id: targetUserId }); return; } const alias = `oneSignal_${targetUserId}`; const body = { app_id: oneSignalAppId, include_aliases: { external_id: [alias] }, target_channel: 'push', headings: { en: payload.title || 'New message' }, contents: { en: payload.body || 'You have a new message' }, data: payload.data || {}, ios_sound: pushSound, android_sound: pushSound, web_push_topic: `message_${targetUserId}_${Date.now()}` }; if (payload.url) { body.url = payload.url; } await fetch('https://onesignal.com/api/v1/notifications', { method: 'POST', headers: { 'Content-Type': 'application/json; charset=utf-8', Authorization: `Basic ${oneSignalRestKey}` }, body: JSON.stringify(body) }); } catch (err) { log('WARN', 'Failed to dispatch OneSignal push', { user_id: targetUserId, error: err.message }); } }; const socketBroadcast = (partnerId, event, payload) => { const sockets = onlineUsers.get(partnerId); if (!sockets || sockets.size === 0) return; sockets.forEach((socketId) => { const targetSocket = io.sockets.sockets.get(socketId); if (targetSocket) { targetSocket.emit(event, payload); } }); }; /** * Socket Event Handlers (replacing HTTP routes) */ const handleBootstrap = async (socket, payload = {}) => { try { const conversations = await fetchConversationList(socket.userId, { limit: DEFAULT_PAGE_SIZE }); const user = await getUserData(socket.userId, { withPrivacy: true, withAliases: true }); socket.emit('messaging:bootstrap', { status: 200, data: { user: { id: user.id, username: user.username, avatar: user.avatar, verified: user.verified, admin: user.admin || 0 }, conversations } }); } catch (err) { log('ERROR', 'Failed to bootstrap messaging', { user_id: socket.userId, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Failed to bootstrap messaging' }); } }; const handleGetConversations = async (socket, payload = {}) => { try { const { cursor, limit } = payload; const conversations = await fetchConversationList(socket.userId, { cursor, limit }); socket.emit('messaging:conversations', { status: 200, data: conversations }); } catch (err) { log('ERROR', 'Failed to fetch conversations', { user_id: socket.userId, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Could not load conversations' }); } }; const handleGetMessages = async (socket, payload = {}) => { try { const partnerId = Number(payload.partner_id); if (!Number.isFinite(partnerId)) { socket.emit('messaging:error', { status: 400, error: 'Invalid partner id' }); return; } const options = { limit: payload.limit, before: payload.before, after: payload.after, threadRoot: payload.thread_root }; const messages = await fetchMessages(socket.userId, partnerId, options); socket.emit('messaging:messages', { status: 200, data: { messages } }); } catch (err) { log('ERROR', 'Failed to fetch messages', { user_id: socket.userId, partner_id: payload.partner_id, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Could not load messages' }); } }; const handleSendMessage = async (socket, payload = {}) => { try { const partnerId = Number(payload.partner_id); if (!Number.isFinite(partnerId)) { socket.emit('messaging:error', { status: 400, error: 'Invalid partner id' }); return; } const partner = await getUserData(partnerId, { withPrivacy: true, withAliases: true }); const user = await getUserData(socket.userId, { withPrivacy: true, withAliases: true }); await validateMessagePermissions(user, partner); const text = (payload.text || '').trim(); const sanitizedAttachments = sanitizeAttachments(payload.attachments); const giphy = payload.giphy && typeof payload.giphy === 'object' ? payload.giphy : null; const og = payload.og && typeof payload.og === 'object' ? payload.og : null; const thread = payload.thread || null; const clientId = payload.client_id || uuidv4(); // Extract first image from attachments (mirroring RTMP stream comments approach) // Store it separately in the image column instead of in the encrypted envelope let imageData = null; let remainingAttachments = sanitizedAttachments; if (sanitizedAttachments.length > 0) { // Find first image attachment const firstImage = sanitizedAttachments.find(att => att && att.mime && att.mime.startsWith('image/') ); if (firstImage && firstImage.data) { // Extract base64 data (remove data:image/...;base64, prefix if present) imageData = firstImage.data.includes(',') ? firstImage.data.split(',')[1] : firstImage.data; // Remove the image from attachments array (it will be stored separately) remainingAttachments = sanitizedAttachments.filter(att => att !== firstImage); } } const messageType = payload.type || (imageData || remainingAttachments.length > 0 ? 'media' : giphy ? 'giphy' : 'text'); if (!text && !imageData && remainingAttachments.length === 0 && !giphy && !og) { socket.emit('messaging:error', { status: 400, error: 'Message body is empty' }); return; } await ensureConversationRows(socket.userId, partnerId); // If replying, fetch the original message to get reply info let replyInfo = null; if (thread?.reply_to) { try { const [replyRows] = await query( `SELECT id, from_id, to_id, text, image FROM messages WHERE id = ? LIMIT 1`, [thread.reply_to] ); if (replyRows && replyRows.length > 0) { const replyMessage = await normalizeMessageRow(replyRows[0], socket.userId); if (replyMessage && replyMessage.sender) { // Get image from reply message if it exists let replyImage = null; if (replyRows[0].image) { replyImage = `data:image/jpeg;base64,${replyRows[0].image}`; } else if (replyMessage.payload?.attachments?.length > 0) { const firstImage = replyMessage.payload.attachments.find(att => att.mime?.startsWith('image/')); if (firstImage) { replyImage = firstImage.data; } } replyInfo = { message_id: thread.reply_to, username: replyMessage.sender.username || 'User', text: replyMessage.payload?.text || '[Message]', avatar: replyMessage.sender.avatar || '', image: replyImage }; } } } catch (err) { log('WARN', 'Failed to fetch reply info', { reply_to: thread.reply_to, error: err.message }); } } // Build payload without the image (it's stored separately) const messagePayload = { text, attachments: remainingAttachments, // Only non-image attachments or additional images giphy, og, reply_to: thread?.reply_to || null, thread_root: thread?.root || null }; // Enhanced thread meta with reply info const enhancedThread = thread ? { ...thread, reply_username: replyInfo?.username, reply_text: replyInfo?.text, reply_avatar: replyInfo?.avatar, reply_image: replyInfo?.image } : null; const envelope = buildMessageEnvelope(socket.userId, partnerId, messagePayload, { type: messageType, thread: enhancedThread, clientId }); const envelopeJson = JSON.stringify(envelope); // Insert message with image stored separately (like RTMP stream comments) const [result] = await query( ` INSERT INTO messages (from_id, to_id, text, image, time, seen, from_deleted, to_deleted) VALUES (?, ?, ?, ?, UNIX_TIMESTAMP(), 0, 0, 0) `, [socket.userId, partnerId, envelopeJson, imageData] ); const messageId = result.insertId; await updateConversationTimestamp(socket.userId, partnerId); const [insertedRows] = await query( ` SELECT id, from_id, to_id, text, image, time, seen, from_deleted, to_deleted FROM messages WHERE id = ? LIMIT 1 `, [messageId] ); const normalizedMessage = await normalizeMessageRow(insertedRows[0], socket.userId); const roomName = conversationRoom(socket.userId, partnerId); io.to(roomName).emit('messaging:new-message', normalizedMessage); socketBroadcast(partnerId, 'messaging:conversation-updated', { partner_id: socket.userId, last_message: normalizedMessage }); if (!onlineUsers.get(partnerId) || onlineUsers.get(partnerId).size === 0) { await sendPushNotification(partnerId, { title: `${user.username || 'Someone'} sent you a message`, body: text || 'New attachment', data: { type: 'messaging', partner_id: socket.userId } }); } socket.emit('messaging:message-sent', { status: 200, data: { message: normalizedMessage } }); } catch (err) { log('ERROR', 'Failed to send message', { user_id: socket.userId, partner_id: payload.partner_id, error: err.message }); socket.emit('messaging:error', { status: 400, error: err.message || 'Could not send message' }); } }; const handleLikeMessage = async (socket, payload = {}) => { try { const messageId = Number(payload.message_id); const enabled = payload.enabled !== false; const [rows] = await query( ` SELECT id, from_id, to_id, text, image, time, seen FROM messages WHERE id = ? LIMIT 1 `, [messageId] ); const messageRow = rows?.[0]; if (!messageRow) { socket.emit('messaging:error', { status: 404, error: 'Message not found' }); return; } const envelope = isEnvelope(messageRow.text) ? JSON.parse(messageRow.text) : null; if (!envelope) { socket.emit('messaging:error', { status: 400, error: 'Cannot like legacy message' }); return; } const likes = envelope.meta.likes || { count: 0, user_ids: [] }; const likedSet = new Set(likes.user_ids || []); if (enabled) { likedSet.add(socket.userId); } else { likedSet.delete(socket.userId); } envelope.meta.likes = { count: likedSet.size, user_ids: Array.from(likedSet) }; await query('UPDATE messages SET text = ? WHERE id = ?', [JSON.stringify(envelope), messageId]); const normalizedMessage = await normalizeMessageRow( { ...messageRow, text: JSON.stringify(envelope) }, socket.userId ); const partnerId = messageRow.from_id === socket.userId ? messageRow.to_id : messageRow.from_id; const roomName = conversationRoom(messageRow.from_id, messageRow.to_id); io.to(roomName).emit('messaging:message-updated', normalizedMessage); socket.emit('messaging:message-liked', { status: 200, data: { message: normalizedMessage } }); } catch (err) { log('ERROR', 'Failed to toggle like', { user_id: socket.userId, message_id: payload.message_id, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Could not apply like' }); } }; const handleDeleteMessage = async (socket, payload = {}) => { try { const messageId = Number(payload.message_id); const [rows] = await query( ` SELECT id, from_id, to_id, text, image, time, seen FROM messages WHERE id = ? LIMIT 1 `, [messageId] ); const messageRow = rows?.[0]; if (!messageRow) { socket.emit('messaging:error', { status: 404, error: 'Message not found' }); return; } if (messageRow.from_id !== socket.userId && !(socket.admin > 0)) { socket.emit('messaging:error', { status: 403, error: 'Not allowed to delete this message' }); return; } // Actually delete the message from database await query('DELETE FROM messages WHERE id = ?', [messageId]); const roomName = conversationRoom(messageRow.from_id, messageRow.to_id); // Emit to room that message was removed io.to(roomName).emit('messaging:message-removed', { message_id: messageId, partner_id: messageRow.from_id === socket.userId ? messageRow.to_id : messageRow.from_id }); socket.emit('messaging:message-deleted', { status: 200, data: { message_id: messageId } }); } catch (err) { log('ERROR', 'Failed to delete message', { user_id: socket.userId, message_id: payload.message_id, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Could not delete message' }); } }; const handleEditMessage = async (socket, payload = {}, callback) => { try { const messageId = Number(payload.message_id); const text = (payload.text || '').trim(); const sanitizedAttachments = sanitizeAttachments(payload.attachments || []); const giphy = payload.giphy && typeof payload.giphy === 'object' ? payload.giphy : null; const og = payload.og && typeof payload.og === 'object' ? payload.og : null; if (!text && sanitizedAttachments.length === 0 && !giphy && !og) { if (typeof callback === 'function') { callback({ status: 400, error: 'Message body is empty' }); } else { socket.emit('messaging:error', { status: 400, error: 'Message body is empty' }); } return; } const [rows] = await query( ` SELECT id, from_id, to_id, text, image, time, seen FROM messages WHERE id = ? LIMIT 1 `, [messageId] ); const messageRow = rows?.[0]; if (!messageRow) { socket.emit('messaging:error', { status: 404, error: 'Message not found' }); return; } if (messageRow.from_id !== socket.userId) { socket.emit('messaging:error', { status: 403, error: 'Not allowed to edit this message' }); return; } const envelope = isEnvelope(messageRow.text) ? JSON.parse(messageRow.text) : null; if (!envelope) { socket.emit('messaging:error', { status: 400, error: 'Cannot edit legacy message' }); return; } // Update payload const messagePayload = { text, attachments: sanitizedAttachments, giphy, og, reply_to: envelope.payload?.reply_to || null, thread_root: envelope.payload?.thread_root || null }; // Extract image if present let imageData = messageRow.image; if (sanitizedAttachments.length > 0) { const firstImage = sanitizedAttachments.find(att => att && att.mime && att.mime.startsWith('image/') ); if (firstImage && firstImage.data) { imageData = firstImage.data.includes(',') ? firstImage.data.split(',')[1] : firstImage.data; } } // Rebuild envelope with updated payload const updatedEnvelope = { ...envelope, payload: messagePayload }; await query('UPDATE messages SET text = ?, image = ? WHERE id = ?', [ JSON.stringify(updatedEnvelope), imageData, messageId ]); const normalizedMessage = await normalizeMessageRow( { ...messageRow, text: JSON.stringify(updatedEnvelope), image: imageData }, socket.userId ); const roomName = conversationRoom(messageRow.from_id, messageRow.to_id); io.to(roomName).emit('messaging:message-updated', normalizedMessage); // Call callback if provided if (typeof callback === 'function') { callback({ status: 200, data: { message: normalizedMessage } }); } else { socket.emit('messaging:message-edited', { status: 200, data: { message: normalizedMessage } }); } } catch (err) { log('ERROR', 'Failed to edit message', { user_id: socket.userId, message_id: payload.message_id, error: err.message }); if (typeof callback === 'function') { callback({ status: 500, error: 'Could not edit message' }); } else { socket.emit('messaging:error', { status: 500, error: 'Could not edit message' }); } } }; const handleSearchMessages = async (socket, payload = {}) => { try { const partnerId = Number(payload.partner_id); const term = (payload.q || '').trim().toLowerCase(); if (!term || term.length < 2) { socket.emit('messaging:error', { status: 400, error: 'Search term too short' }); return; } if (!Number.isFinite(partnerId)) { socket.emit('messaging:error', { status: 400, error: 'Invalid partner id' }); return; } const [rows] = await query( ` SELECT id, from_id, to_id, text, time, seen FROM messages WHERE ((from_id = ? AND to_id = ?) OR (from_id = ? AND to_id = ?)) ORDER BY id DESC LIMIT 500 `, [socket.userId, partnerId, partnerId, socket.userId] ); const matches = []; for (const row of rows) { const normalized = await normalizeMessageRow(row, socket.userId); const plaintext = (normalized.payload?.text || '').toLowerCase(); if (plaintext.includes(term)) { matches.push(normalized); } } socket.emit('messaging:search-results', { status: 200, data: { matches } }); } catch (err) { log('ERROR', 'Failed to search messages', { user_id: socket.userId, partner_id: payload.partner_id, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Could not search messages' }); } }; const handleUpdatePreferences = async (socket, payload = {}) => { try { const enablePush = payload.push_notifications !== false; const user = await getUserData(socket.userId, { withPrivacy: true, withAliases: true }); const privacy = resolvePrivacy(user); privacy.message_notifications = enablePush ? 'on' : 'off'; await query('UPDATE users SET privacy = ? WHERE id = ?', [JSON.stringify(privacy), socket.userId]); socket.emit('messaging:preferences-updated', { status: 200, data: { message_notifications: privacy.message_notifications } }); } catch (err) { log('ERROR', 'Failed to update messaging preferences', { user_id: socket.userId, error: err.message }); socket.emit('messaging:error', { status: 500, error: 'Could not update preferences' }); } }; /** * Socket handlers */ const registerSocketHandlers = (socket) => { if (!socket.userId) { return; } socket.messagingRooms = new Set(); // Bootstrap messaging socket.on('messaging:bootstrap', (payload) => handleBootstrap(socket, payload)); // Get conversations list socket.on('messaging:get-conversations', (payload) => handleGetConversations(socket, payload)); // Get messages for a conversation socket.on('messaging:get-messages', (payload) => handleGetMessages(socket, payload)); // Join a conversation (existing functionality) socket.on('messaging:join', async (payload = {}) => { const partnerId = Number(payload.partner_id); if (!Number.isFinite(partnerId)) return; const room = conversationRoom(socket.userId, partnerId); socket.join(room); socket.messagingRooms.add(room); try { await ensureConversationRows(socket.userId, partnerId); const messages = await fetchMessages(socket.userId, partnerId, { limit: DEFAULT_PAGE_SIZE }); socket.emit('messaging:history', { partner_id: partnerId, messages }); } catch (err) { log('ERROR', 'Socket history load failed', { user_id: socket.userId, partner_id: partnerId, error: err.message }); } }); // Leave a conversation (existing functionality) socket.on('messaging:leave', (payload = {}) => { const partnerId = Number(payload.partner_id); if (!Number.isFinite(partnerId)) return; const room = conversationRoom(socket.userId, partnerId); socket.leave(room); socket.messagingRooms.delete(room); }); // Send a message socket.on('messaging:send-message', (payload) => handleSendMessage(socket, payload)); // Like/unlike a message socket.on('messaging:like-message', (payload) => handleLikeMessage(socket, payload)); // Delete a message socket.on('messaging:delete-message', (payload) => handleDeleteMessage(socket, payload)); // Edit a message socket.on('messaging:edit-message', (payload, callback) => handleEditMessage(socket, payload, callback)); // Search messages socket.on('messaging:search', (payload) => handleSearchMessages(socket, payload)); // Update preferences socket.on('messaging:update-preferences', (payload) => handleUpdatePreferences(socket, payload)); // Typing indicator (existing functionality) socket.on('messaging:typing', (payload = {}) => { const partnerId = Number(payload.partner_id); if (!Number.isFinite(partnerId)) return; const room = conversationRoom(socket.userId, partnerId); // Ensure the sender is in the room (Socket.IO handles duplicate joins gracefully) socket.join(room); if (socket.messagingRooms) { socket.messagingRooms.add(room); } const typingPayload = { partner_id: socket.userId, active: !!payload.active, at: Date.now() }; socket.to(room).emit('messaging:typing', typingPayload); }); // Mark messages as read (existing functionality) socket.on('messaging:mark-read', async (payload = {}) => { const partnerId = Number(payload.partner_id); if (!Number.isFinite(partnerId)) return; try { const seenAt = Math.floor(Date.now() / 1000); await query( ` UPDATE messages SET seen = ? WHERE from_id = ? AND to_id = ? AND to_deleted = 0 AND (seen IS NULL OR seen = 0) `, [seenAt, partnerId, socket.userId] ); const room = conversationRoom(socket.userId, partnerId); // Ensure the sender is in the room (Socket.IO handles duplicate joins gracefully) socket.join(room); if (socket.messagingRooms) { socket.messagingRooms.add(room); } socket.to(room).emit('messaging:read-receipt', { partner_id: socket.userId, seen_at: seenAt }); } catch (err) { log('ERROR', 'Failed to mark messages read', { user_id: socket.userId, partner_id: partnerId, error: err.message }); } }); // Admin moderation (existing functionality) socket.on('messaging:moderate-delete', async (payload = {}) => { if (!(socket.isAdmin || socket.admin > 0)) { return; } const messageId = Number(payload.message_id); if (!Number.isFinite(messageId)) return; try { const [rows] = await query( ` SELECT id, from_id, to_id, text, image, time, seen FROM messages WHERE id = ? LIMIT 1 `, [messageId] ); const messageRow = rows?.[0]; if (!messageRow) return; const envelope = isEnvelope(messageRow.text) ? JSON.parse(messageRow.text) : null; if (envelope) { envelope.meta.deleted = { by: socket.userId || SYSTEM_USER_ID, at: Math.floor(Date.now() / 1000), reason: payload.reason || 'moderated' }; await query('UPDATE messages SET text = ? WHERE id = ?', [JSON.stringify(envelope), messageId]); const normalized = await normalizeMessageRow( { ...messageRow, text: JSON.stringify(envelope) }, socket.userId ); const room = conversationRoom(messageRow.from_id, messageRow.to_id); io.to(room).emit('messaging:message-updated', normalized); } else { await query('UPDATE messages SET from_deleted = 1, to_deleted = 1 WHERE id = ?', [messageId]); const room = conversationRoom(messageRow.from_id, messageRow.to_id); io.to(room).emit('messaging:message-removed', { message_id: messageId }); } socket.emit('messaging:moderated', { message_id: messageId }); } catch (err) { log('ERROR', 'Failed to moderate delete', { message_id: messageId, error: err.message }); } }); const handleFetchOg = async (socket, payload = {}, callback) => { try { const url = payload.url; if (!url || typeof url !== 'string' || !url.match(/^https?:\/\//)) { if (typeof callback === 'function') { callback({ status: 400, error: 'Invalid URL' }); } return; } try { const response = await axios.get(url, { timeout: 5000, headers: { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }, maxRedirects: 5 }); const $ = cheerio.load(response.data); const ogData = { title: $('meta[property="og:title"]').attr('content') || $('title').text() || '', description: $('meta[property="og:description"]').attr('content') || $('meta[name="description"]').attr('content') || '', image: $('meta[property="og:image"]').attr('content') || '', url: $('meta[property="og:url"]').attr('content') || url }; // Clean up image URL (make absolute if relative) if (ogData.image && !ogData.image.startsWith('http')) { try { const baseUrl = new URL(url); ogData.image = new URL(ogData.image, baseUrl.origin).href; } catch (e) { ogData.image = ''; } } if (typeof callback === 'function') { callback({ status: 200, data: ogData }); } else { socket.emit('messaging:og-fetched', { status: 200, data: ogData }); } } catch (err) { log('WARN', 'Failed to fetch OG metadata', { url, error: err.message }); if (typeof callback === 'function') { callback({ status: 500, error: 'Failed to fetch metadata' }); } } } catch (err) { log('ERROR', 'OG fetch handler error', { error: err.message }); if (typeof callback === 'function') { callback({ status: 500, error: 'Internal error' }); } } }; socket.on('messaging:fetch-og', (payload, callback) => { handleFetchOg(socket, payload, callback); }); socket.on('disconnect', () => { if (socket.messagingRooms) { socket.messagingRooms.forEach((room) => socket.leave(room)); socket.messagingRooms.clear(); } }); }; return { registerSocketHandlers }; } module.exports = { init };

Upgrade to a premium experience with MatildaTv PRO

Subscribe to enjoy ad-free viewing and much more!

Sign Up Maybe Later