api.js (7858B)
1 const express = require('express'); 2 const { db } = require('../firebase'); 3 const { chatsCache, messagesCache, isCacheReady } = require('../cache'); 4 const { clients, enforceConnectionCeiling } = require('../sseManager'); 5 const { verifyApiToken } = require('../auth'); 6 7 const router = express.Router(); 8 9 router.get('/ping', (req, res) => { 10 res.status(200).send('Pong'); 11 }); 12 13 // --- API: Server Sent Events --- 14 router.get('/api/chats/stream', verifyApiToken, async (req, res) => { 15 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 16 17 res.writeHead(200, { 18 'Content-Type': 'text/event-stream', 19 'Cache-Control': 'no-cache', 20 'Connection': 'keep-alive', 21 'X-Accel-Buffering': 'no' 22 }); 23 if (res.flushHeaders) res.flushHeaders(); 24 if (res.socket) res.socket.setNoDelay(true); 25 res.write('\n'); 26 27 const cleanup = () => { 28 clients.chats.delete(res); 29 }; 30 31 enforceConnectionCeiling(req, res, cleanup); 32 clients.chats.add(res); 33 34 try { 35 const grouped = {}; 36 37 chatsCache.forEach((data, docId) => { 38 const phone = data.phoneNumber || data.id.split('@')[0]; 39 40 if (!grouped[phone]) { 41 grouped[phone] = { ...data, ids: [data.id] }; 42 } else { 43 grouped[phone].ids.push(data.id); 44 if ((data.lastActive || 0) > (grouped[phone].lastActive || 0)) { 45 grouped[phone].lastActive = data.lastActive; 46 grouped[phone].preview = data.preview || grouped[phone].preview; 47 } 48 if (data.customName) grouped[phone].customName = data.customName; 49 } 50 }); 51 52 res.write(`event: initial\ndata: ${JSON.stringify(Object.values(grouped))}\n\n`); 53 } catch (e) { 54 console.error("Error sending initial chats:", e); 55 } 56 }); 57 58 router.get('/api/messages/stream', verifyApiToken, async (req, res) => { 59 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 60 61 const { chatId, since } = req.query; 62 if (!chatId) return res.status(400).send('Missing chatId'); 63 64 res.writeHead(200, { 65 'Content-Type': 'text/event-stream', 66 'Cache-Control': 'no-cache', 67 'Connection': 'keep-alive', 68 'X-Accel-Buffering': 'no' 69 }); 70 if (res.flushHeaders) res.flushHeaders(); 71 if (res.socket) res.socket.setNoDelay(true); 72 res.write('\n'); 73 74 const cleanup = () => { 75 const chatClients = clients.messages.get(chatId); 76 if (chatClients) { 77 chatClients.delete(res); 78 if (chatClients.size === 0) clients.messages.delete(chatId); 79 } 80 }; 81 82 enforceConnectionCeiling(req, res, cleanup); 83 84 if (!clients.messages.has(chatId)) { 85 clients.messages.set(chatId, new Set()); 86 } 87 clients.messages.get(chatId).add(res); 88 89 try { 90 let initialMessages = []; 91 const chatMsgs = messagesCache.get(chatId); 92 93 if (chatMsgs) { 94 const sinceTs = since ? parseInt(since, 10) : 0; 95 for (const msg of chatMsgs.values()) { 96 if (msg.timestamp > sinceTs) { 97 initialMessages.push(msg); 98 } 99 } 100 initialMessages.sort((a, b) => a.timestamp - b.timestamp); 101 } 102 103 res.write(`event: initial\ndata: ${JSON.stringify(initialMessages)}\n\n`); 104 } catch (e) { 105 console.error("Error sending initial messages:", e); 106 } 107 }); 108 109 // Global sync stream: one connection covers messages for ALL chats, so the 110 // client doesn't need to open a new SSE connection per chat it opens. 111 router.get('/api/sync/stream', verifyApiToken, async (req, res) => { 112 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 113 114 const { since } = req.query; 115 116 res.writeHead(200, { 117 'Content-Type': 'text/event-stream', 118 'Cache-Control': 'no-cache', 119 'Connection': 'keep-alive', 120 'X-Accel-Buffering': 'no' 121 }); 122 if (res.flushHeaders) res.flushHeaders(); 123 if (res.socket) res.socket.setNoDelay(true); 124 res.write('\n'); 125 126 const cleanup = () => { 127 clients.sync.delete(res); 128 }; 129 130 enforceConnectionCeiling(req, res, cleanup); 131 clients.sync.add(res); 132 133 try { 134 const sinceTs = since ? parseInt(since, 10) : 0; 135 const initialMessages = []; 136 137 messagesCache.forEach((msgMap, chatId) => { 138 for (const msg of msgMap.values()) { 139 if (msg.timestamp > sinceTs) { 140 initialMessages.push({ ...msg, chatId }); 141 } 142 } 143 }); 144 145 initialMessages.sort((a, b) => a.timestamp - b.timestamp); 146 147 res.write(`event: initial\ndata: ${JSON.stringify(initialMessages)}\n\n`); 148 } catch (e) { 149 console.error("Error sending initial sync payload:", e); 150 } 151 }); 152 153 // --- API: Full Resync (used by the client's "Reset" flow) --- 154 // Lightweight manifest: chat metadata + per-chat message counts, no message 155 // bodies. Lets the client know the true total up front so it can show real 156 // download progress instead of a fake/local-only percentage. 157 router.get('/api/export/chats', verifyApiToken, async (req, res) => { 158 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 159 160 try { 161 const chats = []; 162 let totalMessages = 0; 163 164 chatsCache.forEach((data, id) => { 165 const count = messagesCache.has(id) ? messagesCache.get(id).size : 0; 166 totalMessages += count; 167 chats.push({ ...data, messageCount: count }); 168 }); 169 170 res.json({ chats, totalMessages }); 171 } catch (err) { 172 res.status(500).json({ error: err.message }); 173 } 174 }); 175 176 // Plain JSON (not SSE) — all messages for a single chat. Called once per 177 // chat by the client's resync loop so progress can be updated after each 178 // chat finishes, instead of waiting on one all-chats blob. 179 router.get('/api/export/messages', verifyApiToken, async (req, res) => { 180 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 181 182 const { chatId } = req.query; 183 if (!chatId) return res.status(400).json({ error: 'Missing chatId' }); 184 185 try { 186 const msgMap = messagesCache.get(chatId); 187 const messages = msgMap ? Array.from(msgMap.values()) : []; 188 res.json({ chatId, messages }); 189 } catch (err) { 190 res.status(500).json({ error: err.message }); 191 } 192 }); 193 194 // --- API: Standard Actions --- 195 router.post('/api/rename', verifyApiToken, async (req, res) => { 196 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 197 198 const { id, customName } = req.body; 199 if (!id || !customName) return res.status(400).json({ error: 'Missing parameters' }); 200 201 try { 202 // Fire & Forget to DB 203 await db.collection('Chats').doc(id).set({ customName }, { merge: true }); 204 205 // Instant RAM update 206 if (chatsCache.has(id)) { 207 chatsCache.get(id).customName = customName; 208 } 209 210 res.json({ success: true }); 211 } catch (err) { 212 res.status(500).json({ error: err.message }); 213 } 214 }); 215 216 router.get('/api/export', verifyApiToken, async (req, res) => { 217 if (!isCacheReady()) return res.status(503).json({ error: 'Cache still warming, retry shortly' }); 218 219 try { 220 const exportData = { chats: {}, messages: {} }; 221 222 chatsCache.forEach((data, id) => exportData.chats[id] = data); 223 messagesCache.forEach((msgMap, chatId) => { 224 exportData.messages[chatId] = Array.from(msgMap.values()); 225 }); 226 227 res.json(exportData); 228 } catch (err) { 229 res.status(500).json({ error: err.message }); 230 } 231 }); 232 233 module.exports = router;