Hot Damn… 🔥🔥 🚨 RFK Jr. Just Blasted Congressman Pallone into the Stratosphere for being owned by Big Pharma

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

Enjoyed the video?

Help us improve! Leave feedback on your experience.

0 watching

Hot Damn… 🔥🔥 🚨 RFK Jr. Just Blasted Congressman Pallone into the Stratosphere for being owned by Big Pharma

16 Subscribers

1

0

Quick Live Chat

206.1K Views· 25/06/25· News & Politics

#Harsh #words #but #true

🚨 RFK Jr. Just Blasted Congressman Pallone into the Stratosphere for being owned by Big Pharma

It was so harsh, they had to cut him off and refrain from speaking.

“15 years ago you and I met, and you were at that time a champion for people who had suffered injuries from vaccines…>

Since then, you've accepted $2M from pharmaceutical companies, more than any other member of this committee.”

Show more

0 Comments

Film & Animation Music Crypto Sports Gaming People & Blogs Comedy Entertainment News & Politics How-to & Style Travel & Events Covid-19 Other

Up next

Autoplay

Film & Animation Music Crypto Sports Gaming People & Blogs Comedy Entertainment News & Politics How-to & Style Travel & Events Covid-19 Other