diff --git a/cmd/server/coverage_test.go b/cmd/server/coverage_test.go index eb2edafa..238ec918 100644 --- a/cmd/server/coverage_test.go +++ b/cmd/server/coverage_test.go @@ -3356,6 +3356,128 @@ func TestIngestNewFromDBDuplicateObs(t *testing.T) { } } +// --- IngestNewObservations (fixes #174) --- + +func TestIngestNewObservations(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + store := NewPacketStore(db) + store.Load() + + // Get initial observation count for transmission 1 (hash abc123def4567890) + initialTx := store.byHash["abc123def4567890"] + if initialTx == nil { + t.Fatal("expected to find transmission abc123def4567890 in store") + } + initialObsCount := initialTx.ObservationCount + if initialObsCount != 2 { + t.Fatalf("expected 2 initial observations, got %d", initialObsCount) + } + + // Record the max obs ID after initial load + maxObsID := db.GetMaxObservationID() + + // Simulate a new observation arriving for the existing transmission AFTER + // the poller has already advanced past its transmission ID + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (1, 2, 5.0, -100, '["aa","bb","cc"]', ?)`, time.Now().Unix()) + + // Verify IngestNewFromDB does NOT pick up the new observation (tx id hasn't changed) + txMax := store.MaxTransmissionID() + _, newTxMax := store.IngestNewFromDB(txMax, 100) + if initialTx.ObservationCount != initialObsCount { + t.Errorf("IngestNewFromDB should not have changed obs count, was %d now %d", + initialObsCount, initialTx.ObservationCount) + } + _ = newTxMax + + // IngestNewObservations should pick it up + newObsMax := store.IngestNewObservations(maxObsID, 500) + if newObsMax <= maxObsID { + t.Errorf("expected newObsMax > %d, got %d", maxObsID, newObsMax) + } + if initialTx.ObservationCount != initialObsCount+1 { + t.Errorf("expected obs count %d, got %d", initialObsCount+1, initialTx.ObservationCount) + } + if len(initialTx.Observations) != initialObsCount+1 { + t.Errorf("expected %d observations slice len, got %d", initialObsCount+1, len(initialTx.Observations)) + } + + // Best observation should have been re-picked (new obs has longer path) + if initialTx.PathJSON != `["aa","bb","cc"]` { + t.Errorf("expected best path to be updated to longer path, got %s", initialTx.PathJSON) + } + + t.Run("no new observations", func(t *testing.T) { + max := store.IngestNewObservations(newObsMax, 500) + if max != newObsMax { + t.Errorf("expected same max %d, got %d", newObsMax, max) + } + }) + + t.Run("dedup by observer+path", func(t *testing.T) { + // Insert duplicate observation (same observer + path as existing) + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (1, 1, 12.5, -90, '["aa","bb"]', ?)`, time.Now().Unix()) + prevCount := initialTx.ObservationCount + newMax2 := store.IngestNewObservations(newObsMax, 500) + if initialTx.ObservationCount != prevCount { + t.Errorf("duplicate obs should not increase count, was %d now %d", + prevCount, initialTx.ObservationCount) + } + _ = newMax2 + }) + + t.Run("default limit", func(t *testing.T) { + _ = store.IngestNewObservations(newObsMax, 0) + }) +} + +func TestIngestNewObservationsV2(t *testing.T) { + db := setupTestDBv2(t) + defer db.Close() + seedV2Data(t, db) + store := NewPacketStore(db) + store.Load() + + tx := store.byHash["abc123def4567890"] + if tx == nil { + t.Fatal("expected to find transmission in store") + } + initialCount := tx.ObservationCount + + maxObsID := db.GetMaxObservationID() + + // Add new observation for existing transmission + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_id, observer_name, snr, rssi, path_json, timestamp) + VALUES (1, 'obs2', 'Obs Two', 6.0, -98, '["dd","ee"]', ?)`, time.Now().Unix()) + + newMax := store.IngestNewObservations(maxObsID, 500) + if newMax <= maxObsID { + t.Errorf("expected newMax > %d, got %d", maxObsID, newMax) + } + if tx.ObservationCount != initialCount+1 { + t.Errorf("expected obs count %d, got %d", initialCount+1, tx.ObservationCount) + } +} + +func TestGetMaxObservationID(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + + maxID := db.GetMaxObservationID() + if maxID != 0 { + t.Errorf("expected 0 for empty table, got %d", maxID) + } + + seedTestData(t, db) + maxID = db.GetMaxObservationID() + if maxID <= 0 { + t.Errorf("expected positive max obs ID, got %d", maxID) + } +} + // --- perfMiddleware with endpoint normalization --- func TestPerfMiddlewareEndpointNormalization(t *testing.T) { diff --git a/cmd/server/db.go b/cmd/server/db.go index 9379851f..3f38b21f 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -1488,6 +1488,13 @@ func (db *DB) GetMaxTransmissionID() int { return maxID } +// GetMaxObservationID returns the current max observation ID for polling. +func (db *DB) GetMaxObservationID() int { + var maxID int + db.conn.QueryRow("SELECT COALESCE(MAX(id), 0) FROM observations").Scan(&maxID) + return maxID +} + // GetObserverPacketCounts returns packetsLastHour for all observers (batch query). func (db *DB) GetObserverPacketCounts(sinceEpoch int64) map[string]int { counts := make(map[string]int) diff --git a/cmd/server/store.go b/cmd/server/store.go index f213e5b6..2a108aaa 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -991,6 +991,159 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac return result, newMaxID } +// IngestNewObservations loads new observations for transmissions already in the +// store. This catches observations that arrive after IngestNewFromDB has already +// advanced past the transmission's ID (fixes #174). +func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) int { + if limit <= 0 { + limit = 500 + } + + var querySQL string + if s.db.isV3 { + querySQL = `SELECT o.id, o.transmission_id, obs.id, obs.name, o.direction, + o.snr, o.rssi, o.score, o.path_json, datetime(o.timestamp, 'unixepoch') + FROM observations o + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.id > ? + ORDER BY o.id ASC + LIMIT ?` + } else { + querySQL = `SELECT o.id, o.transmission_id, o.observer_id, o.observer_name, o.direction, + o.snr, o.rssi, o.score, o.path_json, o.timestamp + FROM observations o + WHERE o.id > ? + ORDER BY o.id ASC + LIMIT ?` + } + + rows, err := s.db.conn.Query(querySQL, sinceObsID, limit) + if err != nil { + log.Printf("[store] ingest observations query error: %v", err) + return sinceObsID + } + defer rows.Close() + + type obsRow struct { + obsID int + txID int + observerID string + observerName string + direction string + snr, rssi *float64 + score *int + pathJSON string + timestamp string + } + + var obsRows []obsRow + for rows.Next() { + var oid, txID int + var observerID, observerName, direction, pathJSON, ts sql.NullString + var snr, rssi sql.NullFloat64 + var score sql.NullInt64 + + if err := rows.Scan(&oid, &txID, &observerID, &observerName, &direction, + &snr, &rssi, &score, &pathJSON, &ts); err != nil { + continue + } + + obsRows = append(obsRows, obsRow{ + obsID: oid, + txID: txID, + observerID: nullStrVal(observerID), + observerName: nullStrVal(observerName), + direction: nullStrVal(direction), + snr: nullFloatPtr(snr), + rssi: nullFloatPtr(rssi), + score: nullIntPtr(score), + pathJSON: nullStrVal(pathJSON), + timestamp: nullStrVal(ts), + }) + } + + if len(obsRows) == 0 { + return sinceObsID + } + + s.mu.Lock() + defer s.mu.Unlock() + + newMaxObsID := sinceObsID + updatedTxs := make(map[int]*StoreTx) + + for _, r := range obsRows { + if r.obsID > newMaxObsID { + newMaxObsID = r.obsID + } + + // Already ingested (e.g. by IngestNewFromDB in same cycle) + if _, exists := s.byObsID[r.obsID]; exists { + continue + } + + tx := s.byTxID[r.txID] + if tx == nil { + continue // transmission not yet in store + } + + // Dedup by observer + path + isDupe := false + for _, existing := range tx.Observations { + if existing.ObserverID == r.observerID && existing.PathJSON == r.pathJSON { + isDupe = true + break + } + } + if isDupe { + continue + } + + obs := &StoreObs{ + ID: r.obsID, + TransmissionID: r.txID, + ObserverID: r.observerID, + ObserverName: r.observerName, + Direction: r.direction, + SNR: r.snr, + RSSI: r.rssi, + Score: r.score, + PathJSON: r.pathJSON, + Timestamp: r.timestamp, + } + tx.Observations = append(tx.Observations, obs) + tx.ObservationCount++ + s.byObsID[r.obsID] = obs + if r.observerID != "" { + s.byObserver[r.observerID] = append(s.byObserver[r.observerID], obs) + } + s.totalObs++ + updatedTxs[r.txID] = tx + } + + // Re-pick best observation for updated transmissions + for _, tx := range updatedTxs { + pickBestObservation(tx) + } + + if len(updatedTxs) > 0 { + // Invalidate analytics caches + s.cacheMu.Lock() + s.rfCache = make(map[string]*cachedResult) + s.topoCache = make(map[string]*cachedResult) + s.hashCache = make(map[string]*cachedResult) + s.chanCache = make(map[string]*cachedResult) + s.distCache = make(map[string]*cachedResult) + s.subpathCache = make(map[string]*cachedResult) + s.cacheMu.Unlock() + + log.Printf("[poller] IngestNewObservations: updated %d existing txs, maxObsID %d->%d", + len(updatedTxs), sinceObsID, newMaxObsID) + } + + return newMaxObsID +} + // MaxTransmissionID returns the highest transmission ID in the store. func (s *PacketStore) MaxTransmissionID() int { s.mu.RLock() diff --git a/cmd/server/websocket.go b/cmd/server/websocket.go index 108551b0..d8cb19da 100644 --- a/cmd/server/websocket.go +++ b/cmd/server/websocket.go @@ -165,7 +165,8 @@ func NewPoller(db *DB, hub *Hub, interval time.Duration) *Poller { func (p *Poller) Start() { lastID := p.db.GetMaxTransmissionID() - log.Printf("[poller] starting from transmission ID %d, interval %v", lastID, p.interval) + lastObsID := p.db.GetMaxObservationID() + log.Printf("[poller] starting from transmission ID %d, obs ID %d, interval %v", lastID, lastObsID, p.interval) ticker := time.NewTicker(p.interval) defer ticker.Stop() @@ -174,11 +175,16 @@ func (p *Poller) Start() { select { case <-ticker.C: if p.store != nil { - // Ingest into in-memory store and broadcast + // Ingest new transmissions into in-memory store and broadcast newTxs, newMax := p.store.IngestNewFromDB(lastID, 100) if newMax > lastID { lastID = newMax } + // Ingest new observations for existing transmissions (fixes #174) + newObsMax := p.store.IngestNewObservations(lastObsID, 500) + if newObsMax > lastObsID { + lastObsID = newObsMax + } if len(newTxs) > 0 { log.Printf("[broadcast] sending %d packets to %d clients (lastID now %d)", len(newTxs), p.hub.ClientCount(), lastID) }