diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 4895676a..850014bd 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -32,6 +32,20 @@ type Config struct { LogLevel string `json:"logLevel,omitempty"` ChannelKeysPath string `json:"channelKeysPath,omitempty"` ChannelKeys map[string]string `json:"channelKeys,omitempty"` + Retention *RetentionConfig `json:"retention,omitempty"` +} + +// RetentionConfig controls how long stale nodes are kept before being moved to inactive_nodes. +type RetentionConfig struct { + NodeDays int `json:"nodeDays"` +} + +// NodeDaysOrDefault returns the configured retention.nodeDays or 7 if not set. +func (c *Config) NodeDaysOrDefault() int { + if c.Retention != nil && c.Retention.NodeDays > 0 { + return c.Retention.NodeDays + } + return 7 } // LoadConfig reads configuration from a JSON file, with env var overrides. diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index 5860b4c0..45fff738 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -87,6 +87,19 @@ func applySchema(db *sql.DB) error { CREATE INDEX IF NOT EXISTS idx_nodes_last_seen ON nodes(last_seen); CREATE INDEX IF NOT EXISTS idx_observers_last_seen ON observers(last_seen); + CREATE TABLE IF NOT EXISTS inactive_nodes ( + public_key TEXT PRIMARY KEY, + name TEXT, + role TEXT, + lat REAL, + lon REAL, + last_seen TEXT, + first_seen TEXT, + advert_count INTEGER DEFAULT 0 + ); + + CREATE INDEX IF NOT EXISTS idx_inactive_nodes_last_seen ON inactive_nodes(last_seen); + CREATE TABLE IF NOT EXISTS transmissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, raw_hex TEXT NOT NULL, @@ -331,6 +344,34 @@ func (s *Store) Close() error { return s.db.Close() } +// MoveStaleNodes moves nodes not seen in nodeDays to the inactive_nodes table. +// Returns the number of nodes moved. +func (s *Store) MoveStaleNodes(nodeDays int) (int64, error) { + cutoff := time.Now().UTC().AddDate(0, 0, -nodeDays).Format(time.RFC3339) + tx, err := s.db.Begin() + if err != nil { + return 0, fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback() + + _, err = tx.Exec(`INSERT OR REPLACE INTO inactive_nodes SELECT * FROM nodes WHERE last_seen < ?`, cutoff) + if err != nil { + return 0, fmt.Errorf("insert inactive: %w", err) + } + result, err := tx.Exec(`DELETE FROM nodes WHERE last_seen < ?`, cutoff) + if err != nil { + return 0, fmt.Errorf("delete stale: %w", err) + } + moved, _ := result.RowsAffected() + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("commit: %w", err) + } + if moved > 0 { + log.Printf("Moved %d node(s) to inactive_nodes (not seen in %d days)", moved, nodeDays) + } + return moved, nil +} + // PacketData holds the data needed to insert a packet into the DB. type PacketData struct { RawHex string diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 59632e20..95ef01a1 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -42,6 +42,18 @@ func main() { defer store.Close() log.Printf("SQLite opened: %s", cfg.DBPath) + // Node retention: move stale nodes to inactive_nodes on startup + nodeDays := cfg.NodeDaysOrDefault() + store.MoveStaleNodes(nodeDays) + + // Daily ticker for node retention + retentionTicker := time.NewTicker(24 * time.Hour) + go func() { + for range retentionTicker.C { + store.MoveStaleNodes(nodeDays) + } + }() + channelKeys := loadChannelKeys(cfg, *configPath) if len(channelKeys) > 0 { log.Printf("Loaded %d channel keys for GRP_TXT decryption", len(channelKeys)) @@ -124,6 +136,7 @@ func main() { <-sig log.Println("Shutting down...") + retentionTicker.Stop() for _, c := range clients { c.Disconnect(1000) } diff --git a/cmd/server/config.go b/cmd/server/config.go index 3085e116..d8afe52a 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -43,6 +43,20 @@ type Config struct { } `json:"liveMap"` CacheTTL map[string]interface{} `json:"cacheTTL"` + + Retention *RetentionConfig `json:"retention,omitempty"` +} + +type RetentionConfig struct { + NodeDays int `json:"nodeDays"` +} + +// NodeDaysOrDefault returns the configured retention.nodeDays or 7 if not set. +func (c *Config) NodeDaysOrDefault() int { + if c.Retention != nil && c.Retention.NodeDays > 0 { + return c.Retention.NodeDays + } + return 7 } type HealthThresholds struct { diff --git a/config.example.json b/config.example.json index 1a6b7df5..6991efe8 100644 --- a/config.example.json +++ b/config.example.json @@ -1,6 +1,10 @@ { "port": 3000, "apiKey": "your-secret-api-key-here", + "retention": { + "nodeDays": 7, + "_comment": "Nodes not seen in this many days are moved to inactive_nodes table. Default 7." + }, "https": { "cert": "/path/to/cert.pem", "key": "/path/to/key.pem" diff --git a/db.js b/db.js index 504c8a16..f96c728c 100644 --- a/db.js +++ b/db.js @@ -52,8 +52,20 @@ db.exec(` noise_floor INTEGER ); + CREATE TABLE IF NOT EXISTS inactive_nodes ( + public_key TEXT PRIMARY KEY, + name TEXT, + role TEXT, + lat REAL, + lon REAL, + last_seen TEXT, + first_seen TEXT, + advert_count INTEGER DEFAULT 0 + ); + CREATE INDEX IF NOT EXISTS idx_nodes_last_seen ON nodes(last_seen); CREATE INDEX IF NOT EXISTS idx_observers_last_seen ON observers(last_seen); + CREATE INDEX IF NOT EXISTS idx_inactive_nodes_last_seen ON inactive_nodes(last_seen); CREATE TABLE IF NOT EXISTS transmissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -870,4 +882,20 @@ function getNodeAnalytics(pubkey, days) { }; } -module.exports = { db, schemaVersion, observerIdToRowid, resolveObserverIdx, insertTransmission, upsertNode, incrementAdvertCount, upsertObserver, updateObserverStatus, getPackets, getPacket, getTransmission, getNodes, getNode, getObservers, getStats, searchNodes, getNodeHealth, getNodeAnalytics, removePhantomNodes }; +// Move stale nodes to inactive_nodes table based on retention.nodeDays config. +function moveStaleNodes(nodeDays) { + if (!nodeDays || nodeDays <= 0) return 0; + const cutoff = new Date(Date.now() - nodeDays * 24 * 3600000).toISOString(); + const move = db.transaction(() => { + db.prepare(`INSERT OR REPLACE INTO inactive_nodes SELECT * FROM nodes WHERE last_seen < ?`).run(cutoff); + const result = db.prepare(`DELETE FROM nodes WHERE last_seen < ?`).run(cutoff); + return result.changes; + }); + const moved = move(); + if (moved > 0) { + console.log(`[retention] Moved ${moved} node(s) to inactive_nodes (not seen in ${nodeDays} days)`); + } + return moved; +} + +module.exports = { db, schemaVersion, observerIdToRowid, resolveObserverIdx, insertTransmission, upsertNode, incrementAdvertCount, upsertObserver, updateObserverStatus, getPackets, getPacket, getTransmission, getNodes, getNode, getObservers, getStats, searchNodes, getNodeHealth, getNodeAnalytics, removePhantomNodes, moveStaleNodes }; diff --git a/server.js b/server.js index 8f8ec0dd..422d9c43 100644 --- a/server.js +++ b/server.js @@ -476,6 +476,13 @@ setInterval(() => { } }, 60000).unref(); +// --- Node Retention: move stale nodes to inactive_nodes --- +const RETENTION_NODE_DAYS = (config.retention && config.retention.nodeDays) || 7; +db.moveStaleNodes(RETENTION_NODE_DAYS); +setInterval(() => { + db.moveStaleNodes(RETENTION_NODE_DAYS); +}, 24 * 3600000).unref(); + // --- Health / Telemetry Endpoint --- app.get('/api/health', (req, res) => { const mem = process.memoryUsage(); diff --git a/test-db.js b/test-db.js index f53b4f20..4b69e002 100644 --- a/test-db.js +++ b/test-db.js @@ -447,6 +447,61 @@ console.log('\nstats exclude phantom nodes:'); assert(statsAfter.totalNodesAllTime === countBefore, 'phantom removed from totalNodesAllTime'); } +// --- moveStaleNodes --- +console.log('\nmoveStaleNodes:'); +{ + // Verify inactive_nodes table exists + const tables = db.db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all().map(r => r.name); + assert(tables.includes('inactive_nodes'), 'inactive_nodes table exists'); + + // Verify inactive_nodes has same columns as nodes + const nodesCols = db.db.pragma('table_info(nodes)').map(c => c.name).sort(); + const inactiveCols = db.db.pragma('table_info(inactive_nodes)').map(c => c.name).sort(); + assert(JSON.stringify(nodesCols) === JSON.stringify(inactiveCols), 'inactive_nodes has same columns as nodes'); + + // Insert a stale node (last_seen 30 days ago) and a fresh node + const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 3600000).toISOString(); + const now = new Date().toISOString(); + db.upsertNode({ public_key: 'stale00000000000000000000stale000', name: 'StaleNode', role: 'repeater', last_seen: thirtyDaysAgo, first_seen: thirtyDaysAgo }); + db.upsertNode({ public_key: 'fresh00000000000000000000fresh000', name: 'FreshNode', role: 'companion', last_seen: now, first_seen: now }); + + // Verify both exist in nodes + assert(db.getNode('stale00000000000000000000stale000') !== null, 'stale node exists before move'); + assert(db.getNode('fresh00000000000000000000fresh000') !== null, 'fresh node exists before move'); + + // Move stale nodes (7 day threshold) + const moved = db.moveStaleNodes(7); + assert(moved >= 1, `moveStaleNodes moved at least 1 node (got ${moved})`); + + // Stale node should be gone from nodes + assert(db.getNode('stale00000000000000000000stale000') === null, 'stale node removed from nodes'); + + // Fresh node should still be in nodes + assert(db.getNode('fresh00000000000000000000fresh000') !== null, 'fresh node still in nodes'); + + // Stale node should be in inactive_nodes + const inactive = db.db.prepare('SELECT * FROM inactive_nodes WHERE public_key = ?').get('stale00000000000000000000stale000'); + assert(inactive !== null, 'stale node exists in inactive_nodes'); + assert(inactive.name === 'StaleNode', 'stale node name preserved in inactive_nodes'); + assert(inactive.role === 'repeater', 'stale node role preserved in inactive_nodes'); + + // Fresh node should NOT be in inactive_nodes + const freshInactive = db.db.prepare('SELECT * FROM inactive_nodes WHERE public_key = ?').get('fresh00000000000000000000fresh000'); + assert(!freshInactive, 'fresh node not in inactive_nodes'); + + // Running again should move 0 (already moved) + const moved2 = db.moveStaleNodes(7); + assert(moved2 === 0, 'second moveStaleNodes moves nothing'); + + // With nodeDays=0 should be a no-op + const moved3 = db.moveStaleNodes(0); + assert(moved3 === 0, 'moveStaleNodes(0) is a no-op'); + + // With null should be a no-op + const moved4 = db.moveStaleNodes(null); + assert(moved4 === 0, 'moveStaleNodes(null) is a no-op'); +} + cleanup(); delete process.env.DB_PATH;