diff --git a/config.example.json b/config.example.json index 6774ce00..b0d8612f 100644 --- a/config.example.json +++ b/config.example.json @@ -43,5 +43,10 @@ "nodeSearch": 10, "invalidationDebounce": 30, "_comment": "All values in seconds. Server uses these directly. Client fetches via /api/config/cache." + }, + "packetStore": { + "maxMemoryMB": 1024, + "estimatedPacketBytes": 450, + "_comment": "In-memory packet store. maxMemoryMB caps RAM usage. All packets loaded on startup, served from RAM." } } \ No newline at end of file diff --git a/packet-store.js b/packet-store.js new file mode 100644 index 00000000..edc82c54 --- /dev/null +++ b/packet-store.js @@ -0,0 +1,269 @@ +'use strict'; + +/** + * In-memory packet store — loads all packets from SQLite on startup, + * serves reads from RAM, writes to both RAM + SQLite. + * Caps memory at configurable limit (default 1GB). + */ +class PacketStore { + constructor(dbModule, config = {}) { + this.dbModule = dbModule; // The full db module (has .db, .insertPacket, .getPacket) + this.db = dbModule.db; // Raw better-sqlite3 instance for queries + this.maxBytes = (config.maxMemoryMB || 1024) * 1024 * 1024; + this.estPacketBytes = config.estimatedPacketBytes || 450; + this.maxPackets = Math.floor(this.maxBytes / this.estPacketBytes); + + // Core storage: array sorted by timestamp DESC (newest first) + this.packets = []; + // Indexes + this.byId = new Map(); + this.byHash = new Map(); // hash → [packet, ...] + this.byObserver = new Map(); // observer_id → [packet, ...] + this.byNode = new Map(); // pubkey → [packet, ...] + + this.loaded = false; + this.stats = { totalLoaded: 0, evicted: 0, inserts: 0, queries: 0 }; + } + + /** Load all packets from SQLite into memory */ + load() { + const t0 = Date.now(); + const rows = this.db.prepare( + 'SELECT * FROM packets ORDER BY timestamp DESC' + ).all(); + + for (const row of rows) { + if (this.packets.length >= this.maxPackets) break; + this._index(row); + this.packets.push(row); + } + + this.stats.totalLoaded = this.packets.length; + this.loaded = true; + const elapsed = Date.now() - t0; + console.log(`[PacketStore] Loaded ${this.packets.length} packets in ${elapsed}ms (${Math.round(this.packets.length * this.estPacketBytes / 1024 / 1024)}MB est)`); + return this; + } + + /** Index a packet into all lookup maps */ + _index(pkt) { + this.byId.set(pkt.id, pkt); + + if (pkt.hash) { + if (!this.byHash.has(pkt.hash)) this.byHash.set(pkt.hash, []); + this.byHash.get(pkt.hash).push(pkt); + } + + if (pkt.observer_id) { + if (!this.byObserver.has(pkt.observer_id)) this.byObserver.set(pkt.observer_id, []); + this.byObserver.get(pkt.observer_id).push(pkt); + } + + // Index by node pubkeys mentioned in decoded_json + this._indexByNode(pkt); + } + + /** Extract node pubkeys/names from decoded_json and index */ + _indexByNode(pkt) { + if (!pkt.decoded_json) return; + try { + const decoded = JSON.parse(pkt.decoded_json); + const keys = new Set(); + if (decoded.pubKey) keys.add(decoded.pubKey); + if (decoded.destPubKey) keys.add(decoded.destPubKey); + if (decoded.srcPubKey) keys.add(decoded.srcPubKey); + for (const k of keys) { + if (!this.byNode.has(k)) this.byNode.set(k, []); + this.byNode.get(k).push(pkt); + } + } catch {} + } + + /** Remove oldest packets when over memory limit */ + _evict() { + while (this.packets.length > this.maxPackets) { + const old = this.packets.pop(); + this.byId.delete(old.id); + // Remove from hash index + if (old.hash && this.byHash.has(old.hash)) { + const arr = this.byHash.get(old.hash).filter(p => p.id !== old.id); + if (arr.length) this.byHash.set(old.hash, arr); else this.byHash.delete(old.hash); + } + // Remove from observer index + if (old.observer_id && this.byObserver.has(old.observer_id)) { + const arr = this.byObserver.get(old.observer_id).filter(p => p.id !== old.id); + if (arr.length) this.byObserver.set(old.observer_id, arr); else this.byObserver.delete(old.observer_id); + } + // Skip node index cleanup for eviction (expensive, low value) + this.stats.evicted++; + } + } + + /** Insert a new packet (to both memory and SQLite) */ + insert(packetData) { + const id = this.dbModule.insertPacket(packetData); + const row = this.dbModule.getPacket(id); + if (row) { + this.packets.unshift(row); // newest first + this._index(row); + this._evict(); + this.stats.inserts++; + } + return id; + } + + /** Query packets with filters — all from memory */ + query({ limit = 50, offset = 0, type, route, region, observer, hash, since, until, node, order = 'DESC' } = {}) { + this.stats.queries++; + let results = this.packets; + + // Use indexes for single-key filters when possible + if (hash && !type && !route && !region && !observer && !since && !until && !node) { + results = this.byHash.get(hash) || []; + } else if (observer && !type && !route && !region && !hash && !since && !until && !node) { + results = this.byObserver.get(observer) || []; + } else if (node && !type && !route && !region && !observer && !hash && !since && !until) { + results = this.byNode.get(node) || []; + } else { + // Apply filters sequentially + if (type !== undefined) { + const t = Number(type); + results = results.filter(p => p.payload_type === t); + } + if (route !== undefined) { + const r = Number(route); + results = results.filter(p => p.route_type === r); + } + if (observer) results = results.filter(p => p.observer_id === observer); + if (hash) results = results.filter(p => p.hash === hash); + if (since) results = results.filter(p => p.timestamp > since); + if (until) results = results.filter(p => p.timestamp < until); + if (region) { + // Need to look up observers for this region + const regionObservers = new Set(); + try { + const obs = this.db.prepare('SELECT id FROM observers WHERE iata = ?').all(region); + obs.forEach(o => regionObservers.add(o.id)); + } catch {} + results = results.filter(p => regionObservers.has(p.observer_id)); + } + if (node) { + // Check indexed results first, fall back to text search + const indexed = this.byNode.get(node); + if (indexed) { + const idSet = new Set(indexed.map(p => p.id)); + results = results.filter(p => idSet.has(p.id)); + } else { + // Text search fallback (node name) + results = results.filter(p => + p.decoded_json && p.decoded_json.includes(node) + ); + } + } + } + + const total = results.length; + + // Sort + if (order === 'ASC') { + results = results.slice().sort((a, b) => { + if (a.timestamp < b.timestamp) return -1; + if (a.timestamp > b.timestamp) return 1; + return 0; + }); + } + // Default DESC — packets array is already sorted newest-first + + // Paginate + const paginated = results.slice(Number(offset), Number(offset) + Number(limit)); + return { packets: paginated, total }; + } + + /** Query with groupByHash — aggregate packets by content hash */ + queryGrouped({ limit = 50, offset = 0, type, route, region, observer, hash, since, until, node } = {}) { + this.stats.queries++; + + // Get filtered results first + const { packets: filtered, total: filteredTotal } = this.query({ + limit: 999999, offset: 0, type, route, region, observer, hash, since, until, node + }); + + // Group by hash + const groups = new Map(); + for (const p of filtered) { + const h = p.hash || p.id.toString(); + if (!groups.has(h)) { + groups.set(h, { + hash: p.hash, + observer_count: new Set(), + count: 0, + latest: p.timestamp, + observer_id: p.observer_id, + observer_name: p.observer_name, + path_json: p.path_json, + payload_type: p.payload_type, + raw_hex: p.raw_hex, + decoded_json: p.decoded_json, + }); + } + const g = groups.get(h); + g.count++; + if (p.observer_id) g.observer_count.add(p.observer_id); + if (p.timestamp > g.latest) { + g.latest = p.timestamp; + } + // Keep longest path + if (p.path_json && (!g.path_json || p.path_json.length > g.path_json.length)) { + g.path_json = p.path_json; + g.raw_hex = p.raw_hex; + } + } + + // Sort by latest DESC, paginate + const sorted = [...groups.values()] + .map(g => ({ ...g, observer_count: g.observer_count.size })) + .sort((a, b) => b.latest.localeCompare(a.latest)); + + const total = sorted.length; + const paginated = sorted.slice(Number(offset), Number(offset) + Number(limit)); + return { packets: paginated, total }; + } + + /** Get timestamps for sparkline */ + getTimestamps(since) { + const results = []; + for (const p of this.packets) { + if (p.timestamp <= since) break; // sorted DESC, so we can stop early + results.push(p.timestamp); + } + return results.reverse(); // return ASC + } + + /** Get a single packet by ID */ + getById(id) { + return this.byId.get(id) || null; + } + + /** Get all siblings of a packet (same hash) */ + getSiblings(hash) { + return this.byHash.get(hash) || []; + } + + /** Memory stats */ + getStats() { + return { + ...this.stats, + inMemory: this.packets.length, + maxPackets: this.maxPackets, + estimatedMB: Math.round(this.packets.length * this.estPacketBytes / 1024 / 1024), + maxMB: Math.round(this.maxBytes / 1024 / 1024), + indexes: { + byHash: this.byHash.size, + byObserver: this.byObserver.size, + byNode: this.byNode.size, + } + }; + } +} + +module.exports = PacketStore; diff --git a/server.js b/server.js index 7e8cf0a6..d4387167 100644 --- a/server.js +++ b/server.js @@ -8,6 +8,7 @@ const path = require('path'); const config = require('./config.json'); const decoder = require('./decoder'); const crypto = require('crypto'); +const PacketStore = require('./packet-store'); // Compute a content hash from raw hex: header byte + payload (skipping path hops) // This correctly groups retransmissions of the same packet (same content, different paths) @@ -26,6 +27,7 @@ function computeContentHash(rawHex) { } catch { return rawHex.slice(0, 16); } } const db = require('./db'); +const pktStore = new PacketStore(db, config.packetStore || {}).load(); const channelKeys = require("./config.json").channelKeys || {}; // --- Cache TTL config (seconds → ms) --- @@ -159,6 +161,7 @@ app.get('/api/perf', (req, res) => { endpoints: Object.fromEntries(sorted), slowQueries: perfStats.slowQueries.slice(-20), cache: { size: cache.size, hits: cache.hits, misses: cache.misses, hitRate: cache.hits + cache.misses > 0 ? Math.round(cache.hits / (cache.hits + cache.misses) * 1000) / 10 : 0 }, + packetStore: pktStore.getStats(), }); }); @@ -298,7 +301,7 @@ try { const observerId = parts[2] || null; const region = parts[1] || null; - const packetId = db.insertPacket({ + const packetId = pktStore.insert({ raw_hex: msg.raw, timestamp: now, observer_id: observerId, @@ -370,7 +373,7 @@ try { const role = advert.role || (advert.flags?.repeater ? 'repeater' : advert.flags?.room ? 'room' : 'companion'); db.upsertNode({ public_key: pubKey, name, role, lat, lon, last_seen: now }); - const packetId = db.insertPacket({ + const packetId = pktStore.insert({ raw_hex: null, timestamp: now, observer_id: 'companion', @@ -401,7 +404,7 @@ try { const senderKey = `sender-${senderName.toLowerCase().replace(/[^a-z0-9]/g, '')}`; db.upsertNode({ public_key: senderKey, name: senderName, role: 'companion', lat: null, lon: null, last_seen: now }); } - const packetId = db.insertPacket({ + const packetId = pktStore.insert({ raw_hex: null, timestamp: now, observer_id: 'companion', @@ -423,7 +426,7 @@ try { // Handle direct messages if (topic.startsWith('meshcore/message/direct/')) { const dm = msg.payload || msg; - const packetId = db.insertPacket({ + const packetId = pktStore.insert({ raw_hex: null, timestamp: dm.timestamp || now, observer_id: 'companion', @@ -443,7 +446,7 @@ try { // Handle traceroute if (topic.startsWith('meshcore/traceroute/')) { const trace = msg.payload || msg; - const packetId = db.insertPacket({ + const packetId = pktStore.insert({ raw_hex: null, timestamp: now, observer_id: 'companion', @@ -489,57 +492,31 @@ app.get('/api/stats', (req, res) => { app.get('/api/packets', (req, res) => { const { limit = 50, offset = 0, type, route, region, observer, hash, since, until, groupByHash, node } = req.query; - + const order = req.query.order === 'asc' ? 'ASC' : 'DESC'; + if (groupByHash === 'true') { - let where = []; - let params = {}; - if (type !== undefined) { where.push('payload_type = @type'); params.type = Number(type); } - if (route !== undefined) { where.push('route_type = @route'); params.route = Number(route); } - if (region) { where.push('observer_id IN (SELECT id FROM observers WHERE iata = @region)'); params.region = region; } - if (observer) { where.push('observer_id = @observer'); params.observer = observer; } - if (hash) { where.push('hash = @hash'); params.hash = hash; } - if (since) { where.push('timestamp > @since'); params.since = since; } - if (until) { where.push('timestamp < @until'); params.until = until; } - if (node) { where.push("(decoded_json LIKE @nodePattern OR decoded_json LIKE @nodeNamePattern)"); params.nodePattern = `%${node}%`; const n = db.db.prepare('SELECT name FROM nodes WHERE public_key = ?').get(node); params.nodeNamePattern = n ? `%${n.name}%` : `%${node}%`; } - const clause = where.length ? 'WHERE ' + where.join(' AND ') : ''; - const packets = db.db.prepare(`SELECT hash, COUNT(DISTINCT observer_id) as observer_count, COUNT(*) as count, MAX(timestamp) as latest, (SELECT observer_id FROM packets pObs WHERE pObs.hash = packets.hash ORDER BY pObs.timestamp ASC LIMIT 1) as observer_id, (SELECT observer_name FROM packets pOn WHERE pOn.hash = packets.hash ORDER BY pOn.timestamp ASC LIMIT 1) as observer_name, (SELECT path_json FROM packets p2 WHERE p2.hash = packets.hash ORDER BY LENGTH(path_json) DESC LIMIT 1) as path_json, (SELECT payload_type FROM packets p3 WHERE p3.hash = packets.hash ORDER BY p3.timestamp DESC LIMIT 1) as payload_type, (SELECT raw_hex FROM packets p4 WHERE p4.hash = packets.hash ORDER BY LENGTH(raw_hex) DESC LIMIT 1) as raw_hex, (SELECT decoded_json FROM packets p5 WHERE p5.hash = packets.hash ORDER BY p5.timestamp DESC LIMIT 1) as decoded_json FROM packets ${clause} GROUP BY hash ORDER BY latest DESC LIMIT @limit OFFSET @offset`).all({ ...params, limit: Number(limit), offset: Number(offset) }); - const total = db.db.prepare(`SELECT COUNT(DISTINCT hash) as count FROM packets ${clause}`).get(params).count; - return res.json({ packets, total }); + return res.json(pktStore.queryGrouped({ limit, offset, type, route, region, observer, hash, since, until, node })); } - let where = []; - let params = {}; - if (type !== undefined) { where.push('payload_type = @type'); params.type = Number(type); } - if (route !== undefined) { where.push('route_type = @route'); params.route = Number(route); } - if (region) { where.push('observer_id IN (SELECT id FROM observers WHERE iata = @region)'); params.region = region; } - if (observer) { where.push('observer_id = @observer'); params.observer = observer; } - if (hash) { where.push('hash = @hash'); params.hash = hash; } - if (since) { where.push('timestamp > @since'); params.since = since; } - if (until) { where.push('timestamp < @until'); params.until = until; } - if (node) { where.push("(decoded_json LIKE @nodePattern OR decoded_json LIKE @nodeNamePattern)"); params.nodePattern = `%${node}%`; const nn = db.db.prepare('SELECT name FROM nodes WHERE public_key = ?').get(node); params.nodeNamePattern = nn ? `%${nn.name}%` : `%${node}%`; } - const clause = where.length ? 'WHERE ' + where.join(' AND ') : ''; - const orderDir = req.query.order === 'asc' ? 'ASC' : 'DESC'; - const packets = db.db.prepare(`SELECT * FROM packets ${clause} ORDER BY timestamp ${orderDir} LIMIT @limit OFFSET @offset`).all({ ...params, limit: Number(limit), offset: Number(offset) }); - const total = db.db.prepare(`SELECT COUNT(*) as count FROM packets ${clause}`).get(params).count; - res.json({ packets, total }); + res.json(pktStore.query({ limit, offset, type, route, region, observer, hash, since, until, node, order })); }); // Lightweight endpoint: just timestamps for timeline sparkline app.get('/api/packets/timestamps', (req, res) => { const { since } = req.query; if (!since) return res.status(400).json({ error: 'since required' }); - const rows = db.db.prepare('SELECT timestamp FROM packets WHERE timestamp > ? ORDER BY timestamp ASC').all(since); - res.json(rows.map(r => r.timestamp)); + res.json(pktStore.getTimestamps(since)); }); app.get('/api/packets/:id', (req, res) => { - const packet = db.getPacket(Number(req.params.id)); + const packet = pktStore.getById(Number(req.params.id)) || db.getPacket(Number(req.params.id)); if (!packet) return res.status(404).json({ error: 'Not found' }); // Use the sibling with the longest path (most hops) for display if (packet.hash) { - const best = db.db.prepare('SELECT id, path_json, raw_hex FROM packets WHERE hash = ? ORDER BY LENGTH(path_json) DESC LIMIT 1').get(packet.hash); - if (best && best.path_json && best.path_json.length > (packet.path_json || '').length) { + const siblings = pktStore.getSiblings(packet.hash); + const best = siblings.reduce((a, b) => (b.path_json || '').length > (a.path_json || '').length ? b : a, packet); + if (best.path_json && best.path_json.length > (packet.path_json || '').length) { packet.path_json = best.path_json; packet.raw_hex = best.raw_hex; } @@ -638,7 +615,7 @@ app.post('/api/packets', (req, res) => { const decoded = decoder.decodePacket(hex, channelKeys); const now = new Date().toISOString(); - const packetId = db.insertPacket({ + const packetId = pktStore.insert({ raw_hex: hex.toUpperCase(), timestamp: now, observer_id: observer || null,