From 7e3ab04b538e69b98a505d482a165eb53aeb2cf5 Mon Sep 17 00:00:00 2001 From: agessaman Date: Wed, 29 Jul 2026 21:56:09 -0700 Subject: [PATCH] feat(web-viewer): rebuild dashboard on a background snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The landing page re-ran ~50 aggregate queries five times per load, then repeated the whole sequence every 30 seconds forever — including in backgrounded tabs. Against the live 1.44 GB database that was roughly 20 seconds of SQLite work per page load. Move the work off the request path. A refresher thread in the viewer process (which already runs migrations, so it works for a split-DB install) writes two tables: daily_rollup, one row per local date, and dashboard_snapshot, a single JSON row. A page load now reads one row. Measured on the live database: first paint 6 requests -> 2, /api/dashboard/summary p50 1.1 ms (304 in 0.8 ms), /api/stats 130 ms, and a 0.32 s refresh once a minute in the background. Make the numbers mean what they say: - Window selectors are built from each source's retention. The page offered "30d" and "All" against tables pruned at 7 days, so three of four choices returned the same figure under a label that denied it. - The incoming-packet chart reports its measured window instead of claiming 7 days for a table pruned at 3 — it sat beside a genuine 7-day contacts chart inviting an invalid comparison. - Days with no source data store NULL and render as gaps. Writing 0 would put a fake cliff at every retention boundary. - Signal metrics are stored as sums and counts, never means, so any window re-aggregates correctly. - Delta chips compare the last two complete calendar days and say so; the headline above them is a rolling 24 hours. - Unmapped role ordinals (type0..type15) bucket into "Unknown". - SNR comes from message_stats, where it is populated on every row, not from complete_contact_tracking, where it is populated on 7%. Kill the json_extract scans: packet_stream gains denormalized route_type_name, payload_type_name, path_len and bytes_per_hop, written at capture time. Aggregating those from JSON cost 3-6 s per query. Existing rows convert a bounded batch per tick rather than in one migration that would rewrite ~180 MB into the WAL and stall bot startup. A partial index serves as the backfill worklist — without it the "any rows left?" probe is a full scan costing 4.6 s per tick, and it costs that after the backfill finishes, because finding nothing still means reading everything. Also: replace the per-contact hop-prefix scan with the existing bucketed matcher and memoize the 7-day chunk set (264 ms -> 35 ms on a synthetic 100k-row database, regression-locked by a test); move the dashboard's JS and CSS to static files, which removes the CSP nonce requirement for the bulk of the page; and give cleanup_old_stats a future-timestamp guard, without which rows dated 2103 are never older than the cutoff and so live forever. Deletes the orphaned /stats page, unreachable from the nav and rendering stub charts that never populated. /api/stats stays as a shim with every key name intact plus Deprecation and Sunset headers. All schema changes are additive, so a downgraded codebase can read the data; it would however need the new schema_version rows removed, since MigrationRunner rejects versions it does not know. --- CHANGELOG.md | 67 + TODO.md | 2 + config.ini.example | 25 + docs/data-retention.md | 14 + docs/web-viewer.md | 63 +- modules/commands/stats_command.py | 40 +- modules/config_schema.py | 7 + modules/db_migrations.py | 94 + modules/web_viewer/app.py | 445 +++- modules/web_viewer/dashboard_stats.py | 1296 ++++++++++++ modules/web_viewer/integration.py | 148 +- modules/web_viewer/static/css/dashboard.css | 394 ++++ modules/web_viewer/static/js/dashboard.js | 1048 ++++++++++ .../web_viewer/templates/api_explorer.html | 9 +- modules/web_viewer/templates/index.html | 1853 ++++------------- modules/web_viewer/templates/stats.html | 355 ---- package.json | 2 +- tests/test_dashboard_stats.py | 717 +++++++ tests/test_settings_ui.py | 1 + tests/test_web_viewer.py | 28 +- tests/test_web_viewer_app.py | 3 + tests/test_web_viewer_integration.py | 20 +- tests/test_web_viewer_xss_hardening.py | 43 +- 23 files changed, 4722 insertions(+), 1952 deletions(-) create mode 100644 modules/web_viewer/dashboard_stats.py create mode 100644 modules/web_viewer/static/css/dashboard.css create mode 100644 modules/web_viewer/static/js/dashboard.js delete mode 100644 modules/web_viewer/templates/stats.html create mode 100644 tests/test_dashboard_stats.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8aa67fd..11d2258 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,73 @@ All notable changes to this project are documented here. The format loosely foll [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and the project adheres to semantic versioning. +## [Unreleased] + +### Added + +- Rebuilt web-viewer dashboard, served from a background snapshot instead of + recomputing statistics on every request. A refresher thread in the viewer + process writes `daily_rollup` (one row per local date) and + `dashboard_snapshot` (a single JSON row), so a page load reads one row rather + than running ~50 aggregate queries five times over. +- Daily trends that outlive retention: `message_stats` and friends are pruned at + 7 days and `packet_stream` at 3, so a 30-day chart had nothing to draw from + until now. Signal metrics are stored as sums and counts, never means, so any + window re-aggregates correctly. +- New `/api/dashboard/{summary,series,top,windows,refresh}` endpoints. `summary` + sends a strong `ETag`, so the page's 30-second poll is normally a bodyless + `304`, and polling stops entirely while the tab is hidden. +- New dashboard tiles and charts: routing mix (flood vs direct), hop-count and + path-length histograms, an SNR/RSSI distribution, a 30-day multibyte adoption + trend, busiest repeaters, and a role/device mix. +- New `[Web_Viewer]` settings: `dashboard_snapshot_enabled`, + `dashboard_snapshot_interval_seconds`, `dashboard_snapshot_history_days`, and + `dashboard_packet_backfill_rows`. +- Partial index `idx_packet_stream_undimensioned` as the backfill worklist. + Probing for remaining un-dimensioned rows was otherwise a full table scan, + and it cost the same 4.6 seconds *after* the backfill finished as during it, + because finding nothing still meant reading everything. + +### Changed + +- Time-window selectors are now built from each source's configured retention. + The dashboard previously offered "30d" and "All" against tables pruned at 7 + days, so three of the four choices returned the same number under a label + that claimed otherwise. +- The incoming-packet chart no longer claims to show 7 days. `packet_stream` is + pruned at 3 days, so the covered window is now measured and labelled from the + data — and shown beside a genuinely 7-day contacts chart it can be compared + to. +- `packet_stream` gained denormalized `route_type_name`, `payload_type_name`, + `path_len`, and `bytes_per_hop` columns, written at capture time. Aggregating + these via `json_extract` cost 3–6 seconds per query on a large database. + Existing rows are converted a bounded batch per refresh rather than in one + table-rewriting migration. +- Dashboard JavaScript and CSS moved to static files, removing the CSP nonce + requirement for the bulk of the page's code. +- `cleanup_old_stats` now also deletes rows dated implausibly far in the future. + Such rows are never older than the retention cutoff, so they were immortal — + observed in the wild dated 2103. + +### Removed + +- The orphaned `/stats` page, which was unreachable from the navigation and + rendered stub charts that never populated. + +### Deprecated + +- `GET /api/stats`. Every key name is preserved and the response now carries + `Deprecation` and `Sunset` headers; use `/api/dashboard/*` instead. It will be + removed at the next major version. + +### Notes for downgrades + +All schema changes are additive — two new tables, four new nullable columns, and +new indexes — so the data itself is safe to read with older code. However, +`MigrationRunner` fails startup on encountering an applied migration version it +does not know about, so downgrading below this release requires deleting the +corresponding `schema_version` rows. + ## [1.0.0] — 2026-07-28 v1.0.0 marks the first stable release. It adds transport recovery, location and rain diff --git a/TODO.md b/TODO.md index 6b4f28a..4e9c651 100644 --- a/TODO.md +++ b/TODO.md @@ -104,6 +104,8 @@ by running `python scripts/update_todos.py` (see [Auto-Update](#auto-update)). - [x] (2026-03-15) **Configuration tab** — `/config` page; SMTP + nightly email toggle; log rotation; DB backup; stored in `bot_metadata` - [x] (2026-03-15) **Real-time log viewer** — `/logs` page; SocketIO `subscribe_logs`/`log_line`; level-based coloring; pause/clear/filter; log tail thread; "Logs" nav link - [ ] **Mobile-responsive improvements** — optimize layout for small screens +- [x] (2026-07-29) **Dashboard modernization** — landing page rebuilt on a background snapshot instead of live aggregation. Migrations 0018 (`daily_rollup`, `dashboard_snapshot`) and 0019 (`packet_stream` denormalized `route_type_name`/`payload_type_name`/`path_len`/`bytes_per_hop`, backfilled a bounded batch per tick rather than in one table-rewriting UPDATE); new `modules/web_viewer/dashboard_stats.py`; refresher thread in the viewer process with a `bot_metadata` lease; new `/api/dashboard/{summary,series,top,windows,refresh}`; JS/CSS extracted to `static/` (no CSP nonce needed); `/stats` page and `stats.html` deleted; `/api/stats` kept as a deprecated shim. Measured on the live 1.44 GB DB: first paint 6 requests → 2, `/api/dashboard/summary` p50 **1.1 ms** (304s in 0.8 ms), `/api/stats` 130 ms, and a 0.32 s refresh once a minute off the request path. Window labels and the packet chart's window are now derived from retention instead of asserted. 37 tests added +- [ ] **Dashboard follow-ups** — expand-chart interaction wired to `/api/dashboard/series` (endpoint exists, no UI yet); consider the deferred `observed_paths(last_seen, path_length, bytes_per_hop, public_key)` covering index if refresher timing shows the path-length histogram dominating - [x] (2026-07-11) **Radio page: Node Settings card** — path hash size (response path hash bytes, mode 0–2), name/location/advert-policy, multi-ACKs/telemetry permissions, RX-delay/airtime tuning, zero-hop/flood advert buttons; extended `/api/radio/params`, new `/api/radio/advert`, scheduler write ops; config-managed settings are not writable from the panel (new-contact mode owned by `auto_manage_contacts`; name locked when `bot_name` + `auto_update_device_name`); loop.detect dropped from firmware endpoints (repeater CLI config, not a companion setting); 23 tests added - [x] (TASK-01 2026-03-15) **Remove firmware config + reboot UI** — radio.html: Firmware Configuration card and Reboot Radio button removed; JS handlers removed; 4 tests added - [x] (TASK-02 2026-03-15) **Fix realtime stream blank on load** — added 50-row history replay to `subscribe_commands`; fixed `last_timestamp = 0` → `time.time() - 300` in polling thread; 5 tests added (BUG-023 fixed) diff --git a/config.ini.example b/config.ini.example index 684f613..ae237c8 100644 --- a/config.ini.example +++ b/config.ini.example @@ -1605,6 +1605,31 @@ auto_start = false # true: expose page/nav/API endpoint multibyte_monitor_enabled = false +# Dashboard snapshot refresher +# A background thread in the viewer process recomputes the landing page's +# statistics on an interval and stores them in two tables (daily_rollup and +# dashboard_snapshot), so page loads read one row instead of running ~50 +# aggregate queries. It also accumulates daily trends that outlive the raw +# tables' retention — message/command/path stats are pruned at 7 days and +# packet_stream at 3, so without this there is no history to chart. +# Disabling it leaves /api/dashboard/summary returning 503 and the dashboard +# unable to render. +# dashboard_snapshot_enabled = true +# +# How often to recompute. Each refresh recomputes today plus a 3-day trailing +# window; older days are frozen. +# dashboard_snapshot_interval_seconds = 60 +# +# How many days of daily_rollup history to retain (~1 row/day, so 400 days is +# well under a megabyte). +# dashboard_snapshot_history_days = 400 +# +# packet_stream rows written before this feature existed keep their route type +# and hop encoding only inside a JSON blob. Each refresh converts this many of +# them to real columns rather than rewriting the whole table at once, which on +# an SD-card Pi would stall the bot for minutes. Set to 0 to disable. +# dashboard_packet_backfill_rows = 2000 + # Optional timeout overrides for web viewer integration (seconds) # edge_post_timeout_sec = 1.0 # node_post_timeout_sec = 0.5 diff --git a/docs/data-retention.md b/docs/data-retention.md index e1aad68..fb7a622 100644 --- a/docs/data-retention.md +++ b/docs/data-retention.md @@ -28,8 +28,22 @@ Stats tables (message_stats, command_stats, path_stats) use **`[Stats_Command]`* | **purging_log** | Audit trail for repeater purges | 90 days | | **mesh_connections** | Path graph edges (in-memory + DB); should be ≥ Path_Command `graph_edge_expiration_days` | 7 days | | **message_stats, command_stats, path_stats** | Stats command data | 7 days (`[Stats_Command]` `data_retention_days`) | +| **daily_rollup** | Per-day dashboard rollups, so trends outlive the tables above | 400 days (`[Web_Viewer]` `dashboard_snapshot_history_days`) | +| **dashboard_snapshot** | Single-row current-state payload for the dashboard | Overwritten in place | | **geocoding_cache, generic_cache** | Expired entries removed by scheduler | By expiry time | +`daily_rollup` exists precisely because the retention above is short: the +dashboard charts 30 days of message and packet activity from tables pruned at 7 +and 3 days respectively. One row per day is roughly 200 bytes, so a year of +history costs well under a megabyte. Pruning it is handled by the web viewer's +snapshot refresher rather than the bot scheduler, because the viewer may point +at a different database file. + +Note that cleanup also removes rows dated implausibly far in the **future**. +A node with a bad clock can write a timestamp years ahead; because such a row is +never older than the cutoff, a lower-bound-only delete would keep it forever and +stretch every chart axis to match. + Shorter retention (e.g. 2–3 days for `packet_stream`) is enough for the web viewer and transmission_tracker; longer retention is only needed if you want more history. ## How cleanup runs diff --git a/docs/web-viewer.md b/docs/web-viewer.md index 17078b5..2ff2cec 100644 --- a/docs/web-viewer.md +++ b/docs/web-viewer.md @@ -137,9 +137,45 @@ proxy_set_header X-Forwarded-Proto $scheme; ## Pages Overview ### Dashboard -- Database status and statistics -- Contact counts and cache information -- Quick navigation to other sections +- Health strip: bot status, database size, uptime, connected clients, radio state, + and snapshot age with a manual refresh control +- **Mesh**: nodes heard, adverts, new nodes, nodes gone quiet, tracked/known + contacts, and geographic coverage — each with a 30-day sparkline and a + change chip +- Routing mix (flood vs direct), hop-count and path-length histograms, and an + SNR/RSSI distribution +- Path encoding: multibyte share among contacts and among incoming packets, + plus a 30-day multibyte adoption trend +- Busiest repeaters and the role/device mix +- **Bot**: messages, commands, reply rate, and unique users, plus the top + commands/users/channels and longest paths +- Live activity feed + +The dashboard is served from a **snapshot**, not from live aggregation. A +background thread in the viewer process recomputes it every +`dashboard_snapshot_interval_seconds` (default 60) and writes two tables: + +| Table | Contents | +|-------|----------| +| `daily_rollup` | One row per local date. Retains counts whose raw sources are pruned long before the dashboard's 30-day window. Signal metrics are stored as sums and counts, never means, so any window re-aggregates correctly. | +| `dashboard_snapshot` | A single JSON row describing current state. | + +Two consequences worth knowing: + +- **Gaps are not zeros.** A day with no source data stores NULL and renders as a + break in the line. Days seeded when the feature was first enabled have real + advert counts but no message, command, path, or packet figures — those raw + rows were already pruned and cannot be recovered. +- **Window labels come from retention.** The time-window selectors are built + from each source's configured retention, so the list cannot offer "30 days" + against a table pruned at 7. + +The **multibyte share** trend is accumulated forward, not derived. `observed_paths` +is deduplicated with a lifetime `observation_count` and a `last_seen` that is +bumped on every re-observation, so historical per-day shares cannot be +reconstructed from it — nearly half the observation volume would be attributed +to the wrong day. Each refresh recomputes today plus a three-day trailing +window; older days stay frozen at the value recorded then. ### Repeater Contacts - Active repeater contacts @@ -192,7 +228,24 @@ proxy_set_header X-Forwarded-Proto $scheme; The viewer also provides JSON API endpoints: -- `GET /api/stats` - Database statistics +- `GET /api/dashboard/summary` - Snapshot-backed dashboard payload, including + 30-day sparkline series and change figures. Sends a strong `ETag`; poll with + `If-None-Match` to get a bodyless `304` while the snapshot is unchanged. +- `GET /api/dashboard/series?metric=&days=` - Full-history points for one + metric. `metric` is one of `messages`, `commands`, `adverts`, `nodes`, + `new_nodes`, `packets`, `multibyte_share`. +- `GET /api/dashboard/top?kind=&window=&limit=` - One leaderboard. + `kind` is one of `users`, `commands`, `channels`, `paths`, `repeaters`. The + response carries `window_label`, `retention_days`, and + `truncated_by_retention`. +- `GET /api/dashboard/windows` - Selector options derived from each source's + retention. +- `POST /api/dashboard/refresh` - Force a snapshot recomputation. +- `GET /api/stats` - **Deprecated.** The whole-database statistics payload the + dashboard used to call five times per page load. Every key name is preserved + for external consumers, and the response carries `Deprecation` and `Sunset` + headers. Use the `/api/dashboard/*` endpoints instead; this one is removed at + the next major version. - `GET /api/contacts` - Repeater contacts data. The contacts page uses optional `page`, `page_size` (maximum 200), `search`, `sort`, and `direction` parameters; callers that omit pagination retain the legacy full-list response. @@ -200,7 +253,7 @@ The viewer also provides JSON API endpoints: Example usage: ```bash -curl http://localhost:5000/api/stats +curl http://localhost:5000/api/dashboard/summary ``` ## Database Requirements diff --git a/modules/commands/stats_command.py b/modules/commands/stats_command.py index b07bcab..9a1ea9e 100644 --- a/modules/commands/stats_command.py +++ b/modules/commands/stats_command.py @@ -789,35 +789,47 @@ class StatsCommand(BaseCommand): self.logger.error(f"Error getting adverts leaderboard: {e}") return self.translate('commands.stats.error_adverts', error=str(e)) + # Rows timestamped beyond this margin ahead of now came from a bad clock on + # the sending node, not from the future. Without an upper bound they are + # never older than the cutoff, so retention never removes them: observed in + # the wild dated 2103-08-15, which also stretches any chart's axis to match. + FUTURE_TIMESTAMP_GRACE_SECONDS = 24 * 60 * 60 + def cleanup_old_stats(self, days_to_keep: int = 7) -> None: """Clean up old stats data to prevent database bloat. + Deletes rows older than the retention cutoff *and* rows dated + implausibly far in the future. + Args: days_to_keep: Number of days of data to retain. """ try: - cutoff_time = int(time.time()) - (days_to_keep * 24 * 60 * 60) + now = int(time.time()) + cutoff_time = now - (days_to_keep * 24 * 60 * 60) + future_cutoff = now + self.FUTURE_TIMESTAMP_GRACE_SECONDS with self.bot.db_manager.connection() as conn: cursor = conn.cursor() - # Clean up old message stats - cursor.execute('DELETE FROM message_stats WHERE timestamp < ?', (cutoff_time,)) - messages_deleted = cursor.rowcount - - # Clean up old command stats - cursor.execute('DELETE FROM command_stats WHERE timestamp < ?', (cutoff_time,)) - commands_deleted = cursor.rowcount - - # Clean up old path stats - cursor.execute('DELETE FROM path_stats WHERE timestamp < ?', (cutoff_time,)) - paths_deleted = cursor.rowcount + deleted = {} + for table in ('message_stats', 'command_stats', 'path_stats'): + cursor.execute( + f'DELETE FROM {table} WHERE timestamp < ? OR timestamp > ?', # noqa: S608 - fixed table names + (cutoff_time, future_cutoff), + ) + deleted[table] = cursor.rowcount conn.commit() - total_deleted = messages_deleted + commands_deleted + paths_deleted + total_deleted = sum(deleted.values()) if total_deleted > 0: - self.logger.info(f"Cleaned up {total_deleted} old stats entries ({messages_deleted} messages, {commands_deleted} commands, {paths_deleted} paths)") + self.logger.info( + f"Cleaned up {total_deleted} old stats entries " + f"({deleted['message_stats']} messages, " + f"{deleted['command_stats']} commands, " + f"{deleted['path_stats']} paths)" + ) except Exception as e: self.logger.error(f"Error cleaning up old stats: {e}") diff --git a/modules/config_schema.py b/modules/config_schema.py index f54b23e..6684d11 100644 --- a/modules/config_schema.py +++ b/modules/config_schema.py @@ -201,6 +201,13 @@ SECTIONS: dict[str, SectionMeta] = { "sqlite_busy_timeout_ms": KeyMeta(type="int", default="60000"), "sqlite_foreign_keys": KeyMeta(type="bool", default="true"), "sqlite_journal_mode": KeyMeta(default="WAL"), + # Dashboard snapshot refresher — moves the landing page's aggregate + # queries off the request path and accumulates trends that outlive the + # raw tables' retention. + "dashboard_snapshot_enabled": KeyMeta(type="bool", default="true"), + "dashboard_snapshot_interval_seconds": KeyMeta(type="int", default="60"), + "dashboard_snapshot_history_days": KeyMeta(type="int", default="400"), + "dashboard_packet_backfill_rows": KeyMeta(type="int", default="2000"), }), "Localization": SectionMeta(keys={ "language": KeyMeta(default="en"), diff --git a/modules/db_migrations.py b/modules/db_migrations.py index 5ce2279..a66f494 100644 --- a/modules/db_migrations.py +++ b/modules/db_migrations.py @@ -586,6 +586,98 @@ def _m0017_feed_queue_item_uniqueness(cursor: sqlite3.Cursor) -> None: ) +def _m0018_dashboard_rollup_tables(cursor: sqlite3.Cursor) -> None: + """Durable dashboard storage: per-day rollups plus a current-state snapshot. + + ``daily_rollup`` holds one row per local date for the metrics whose raw + sources are pruned long before the dashboard's 30-day window (message_stats + and friends at 7 days, packet_stream at 3). Signal metrics are stored as + sums and counts rather than means so an arbitrary window re-aggregates + correctly — averaging daily averages weights a quiet day the same as a busy + one. Every numeric column is nullable on purpose: NULL means "no source + data for this day", which the UI must render as a gap, while 0 means "the + source was present and the count really was zero". + + ``dashboard_snapshot`` is a single row of JSON. The current-state half of + the dashboard is ~40 heterogeneous scalars and small arrays that are read + together and never queried by field, so columnizing it would cost a + migration per new tile for no query benefit. + """ + cursor.executescript( + """ + CREATE TABLE IF NOT EXISTS daily_rollup ( + date TEXT PRIMARY KEY, + messages_total INTEGER, messages_dm INTEGER, messages_channel INTEGER, + unique_senders INTEGER, unique_channels INTEGER, + commands_total INTEGER, commands_replied INTEGER, unique_command_users INTEGER, + path_obs_total INTEGER, path_len_max INTEGER, path_len_sum INTEGER, + snr_sum REAL, snr_count INTEGER, rssi_sum REAL, rssi_count INTEGER, + hops_sum INTEGER, hops_count INTEGER, + packets_total INTEGER, packets_flood INTEGER, packets_direct INTEGER, + packets_multibyte INTEGER, + adverts_total INTEGER, nodes_active INTEGER, nodes_new INTEGER, + adverts_from_multibyte INTEGER, adverts_from_singlebyte INTEGER, + contacts_known INTEGER, contacts_tracked INTEGER, + sources_present INTEGER NOT NULL DEFAULT 0, + is_backfilled INTEGER NOT NULL DEFAULT 0, + is_final INTEGER NOT NULL DEFAULT 0, + computed_at TEXT + ); + CREATE INDEX IF NOT EXISTS idx_daily_rollup_computed ON daily_rollup(computed_at); + + CREATE TABLE IF NOT EXISTS dashboard_snapshot ( + id INTEGER PRIMARY KEY CHECK (id = 1), + generated_at REAL NOT NULL, + duration_ms INTEGER, + schema_rev INTEGER NOT NULL DEFAULT 1, + payload TEXT NOT NULL + ); + """ + ) + + +def _m0019_packet_stream_denorm_dims(cursor: sqlite3.Cursor) -> None: + """Denormalize the four packet dimensions the dashboard aggregates on. + + Counting packets by bytes_per_hop or route type via + ``json_extract(data, '$.…')`` is a full scan that parses every row's JSON: + ~3s for bytes_per_hop and ~6s for route_type_name on a 178k-row table. + The writer already holds these values as parsed fields, so storing them as + columns removes the parse entirely. + + Deliberately no bulk UPDATE here. packet_stream rows average ~1KB, so + widening 178k of them rewrites ~180MB into the WAL inside MigrationRunner's + transaction — a multi-minute stall blocking bot startup on an SD-card Pi. + The dashboard refresher backfills a bounded batch per tick instead, and the + table's 3-day retention makes the gap self-healing regardless. + """ + if not _table_exists(cursor, "packet_stream"): + return + _add_column(cursor, "packet_stream", "route_type_name", "TEXT") + _add_column(cursor, "packet_stream", "payload_type_name", "TEXT") + _add_column(cursor, "packet_stream", "path_len", "INTEGER") + _add_column(cursor, "packet_stream", "bytes_per_hop", "INTEGER") + cursor.executescript( + """ + CREATE INDEX IF NOT EXISTS idx_packet_stream_dims + ON packet_stream(timestamp, bytes_per_hop, route_type_name, payload_type_name) + WHERE type = 'packet'; + + -- Worklist for the refresher's incremental backfill. Without it, the + -- "any rows left to dimension?" probe is a full table scan that costs + -- ~4.6s on a 180MB packet_stream — and it costs that on every tick + -- *after* the backlog is empty, because finding nothing still means + -- reading everything. The index shrinks toward empty as rows are + -- dimensioned, so the probe converges to a no-op. It must carry + -- ``type`` as well: non-packet rows keep a NULL route type forever and + -- would otherwise leave the index permanently non-empty. + CREATE INDEX IF NOT EXISTS idx_packet_stream_undimensioned + ON packet_stream(type, id) + WHERE route_type_name IS NULL; + """ + ) + + # --------------------------------------------------------------------------- # Migration registry — append new entries here, never remove or reorder. # --------------------------------------------------------------------------- @@ -610,6 +702,8 @@ MIGRATIONS: list[MigrationEntry] = [ (15, "channel_operations: claimed_at", _m0015_channel_operations_claimed_at), (16, "channel_operations: claim owner identity", _m0016_channel_operations_claim_owner), (17, "feed_message_queue: unique identifiable items", _m0017_feed_queue_item_uniqueness), + (18, "dashboard rollup and snapshot tables", _m0018_dashboard_rollup_tables), + (19, "packet_stream: denormalized packet dimensions", _m0019_packet_stream_denorm_dims), ] diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 1bae2d9..f1235da 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -65,6 +65,15 @@ from modules.settings_schema import ( ) from modules.settings_store import get_settings_store from modules.version_info import resolve_runtime_version +from modules.web_viewer.dashboard_stats import ( + SERIES_METRICS, + TOP_KINDS, + DashboardStatsService, + humanize_span, +) + +# RFC 8594 Sunset date advertised on the deprecated /api/stats endpoint. +STATS_ENDPOINT_SUNSET = "Fri, 01 Jan 2027 00:00:00 GMT" def _validate_dynamic_key(key: str) -> "str | None": @@ -213,6 +222,8 @@ class BotDataViewer: 'path_stats', 'schema_version', 'greeter_rollout', + 'daily_rollup', + 'dashboard_snapshot', } def __init__(self, db_path="meshcore_bot.db", repeater_db_path=None, config_path="config.ini"): @@ -287,9 +298,10 @@ class BotDataViewer: # The contacts list needs all-time multibyte hop-prefix evidence for its # capability badge. Cache that derived set between requests and invalidate # it when the underlying multibyte-path population changes. + # Keyed by the recent_days window (None = all time) so the contacts list + # and the dashboard's 7-day view do not evict each other. self._contacts_badge_cache_lock = threading.Lock() - self._contacts_badge_cache_signature = None - self._contacts_badge_cache_chunks: set[str] = set() + self._contacts_badge_cache: dict[int | None, tuple[tuple, set[str]]] = {} # Use [Bot] db_path when [Web_Viewer] db_path is unset bot_db = self.config.get('Bot', 'db_path', fallback='meshcore_bot.db') @@ -319,6 +331,8 @@ class BotDataViewer: except (configparser.NoSectionError, configparser.NoOptionError, ValueError, TypeError): self.multibyte_monitor_enabled = False + self._init_dashboard_service() + # Configure CORS for SocketIO — default to same-origin (no cross-origin) cors_raw = self.config.get('Web_Viewer', 'cors_allowed_origins', fallback='').strip() if cors_raw: @@ -349,6 +363,10 @@ class BotDataViewer: # Start periodic cleanup self._start_cleanup_scheduler() + # Start the dashboard snapshot refresher (moves the landing page's + # aggregate queries off the request path) + self._start_dashboard_refresher() + self.logger.info("BotDataViewer initialized with Flask-SocketIO 5.x best practices") def _setup_logging(self): @@ -792,7 +810,6 @@ class BotDataViewer: 'radio', 'realtime', 'contacts', - 'stats', 'plugins_page', 'greeter', 'logs', @@ -859,8 +876,23 @@ class BotDataViewer: @self.app.route('/') def index(): - """Main dashboard""" - return render_template('index.html') + """Main dashboard. + + Server-side config reaches the page as a JSON data block rather than + an inline script, so the dashboard's JavaScript can live in a static + file that ``script-src 'self'`` already covers. + """ + database_size = None + with suppress(OSError): + database_size = os.path.getsize(self.db_path) + return render_template( + 'index.html', + dashboard_boot={ + 'database_size': database_size, + 'snapshot_interval_seconds': self.dashboard_snapshot_interval, + 'snapshot_enabled': self.dashboard_snapshot_enabled, + }, + ) @self.app.route('/realtime') def realtime(): @@ -883,11 +915,6 @@ class BotDataViewer: return redirect('/config#database') - @self.app.route('/stats') - def stats(): - """Statistics page""" - return render_template('stats.html') - @self.app.route('/multibyte-rollout') def multibyte_rollout(): """Multibyte hash rollout analytics page""" @@ -2094,7 +2121,13 @@ class BotDataViewer: @self.app.route('/api/stats') def api_stats(): - """Get comprehensive database statistics for dashboard""" + """Deprecated: whole-database statistics in one payload. + + Superseded by /api/dashboard/summary (snapshot-backed) and + /api/dashboard/top (one narrow query per leaderboard). Retained + with every key name intact for external consumers, and scheduled for + removal at the next major version. + """ try: # Get optional time window parameters for analytics top_users_window = request.args.get('top_users_window', 'all') @@ -2107,13 +2140,124 @@ class BotDataViewer: top_paths_window=top_paths_window, top_channels_window=top_channels_window ) - return jsonify(stats) + stats['deprecated'] = True + response = jsonify(stats) + response.headers['Deprecation'] = 'true' + response.headers['Sunset'] = STATS_ENDPOINT_SUNSET + response.headers['Link'] = '; rel="successor-version"' + return response except Exception as e: self.logger.error(f"Error getting stats: {e}") return jsonify({'error': str(e)}), 500 + @self.app.route('/api/dashboard/summary') + def api_dashboard_summary(): + """Snapshot-backed dashboard payload — one row read, no aggregation. + + Sparkline series are folded in so first paint costs two requests + (/api/health plus this) instead of the six the old page made. + """ + try: + with self._with_db_connection() as conn: + payload = self.dashboard_stats.read_summary(conn) + if payload is None: + # No snapshot yet: the refresher runs a couple of seconds + # after startup, so tell the client to retry rather than + # recomputing everything on the request path. + response = jsonify({ + 'error': 'Dashboard snapshot not generated yet', + 'pending': True, + }) + response.status_code = 503 + response.headers['Retry-After'] = '5' + return response + + # ETags.contains() wants the bare tag; the quotes belong only in + # the header itself. + tag = str(payload['generated_at']) + if request.if_none_match.contains(tag): + response = self.app.response_class(status=304) + else: + response = jsonify(payload) + response.headers['ETag'] = f'"{tag}"' + response.headers['Cache-Control'] = 'private, max-age=15' + return response + except Exception as e: + self.logger.error(f"Error reading dashboard summary: {e}") + return jsonify({'error': str(e)}), 500 + + @self.app.route('/api/dashboard/series') + def api_dashboard_series(): + """Full-history points for one metric, for the expand-chart interaction.""" + metric = request.args.get('metric', 'messages') + if metric not in SERIES_METRICS: + return jsonify({'error': f'Unknown metric: {metric}'}), 400 + try: + days = int(request.args.get('days', 30)) + except (TypeError, ValueError): + days = 30 + try: + with self._with_db_connection() as conn: + payload = self.dashboard_stats.read_series(conn, metric, days) + tag = f'{metric}:{days}:{payload.pop("etag_source", None)}' + if request.if_none_match.contains(tag): + response = self.app.response_class(status=304) + else: + response = jsonify(payload) + response.headers['ETag'] = f'"{tag}"' + response.headers['Cache-Control'] = 'private, max-age=300' + return response + except Exception as e: + self.logger.error(f"Error reading dashboard series: {e}") + return jsonify({'error': str(e)}), 500 + + @self.app.route('/api/dashboard/top') + def api_dashboard_top(): + """One leaderboard, one narrow query — replaces four whole-payload reads.""" + kind = request.args.get('kind', 'users') + if kind not in TOP_KINDS: + return jsonify({'error': f'Unknown kind: {kind}'}), 400 + window = request.args.get('window', 'all') + if window not in ('24h', '7d', '30d', '90d', 'all'): + window = 'all' + try: + limit = int(request.args.get('limit', 15)) + except (TypeError, ValueError): + limit = 15 + try: + with self._with_db_connection() as conn: + payload = self.dashboard_stats.read_top(conn, kind, window, limit) + response = jsonify(payload) + response.headers['Cache-Control'] = 'private, max-age=30' + return response + except Exception as e: + self.logger.error(f"Error reading dashboard top {kind}: {e}") + return jsonify({'error': str(e)}), 500 + + @self.app.route('/api/dashboard/windows') + def api_dashboard_windows(): + """Selector options derived from retention, so no label overclaims.""" + try: + response = jsonify(self.dashboard_stats.derive_windows()) + response.headers['Cache-Control'] = 'private, max-age=300' + return response + except Exception as e: + self.logger.error(f"Error deriving dashboard windows: {e}") + return jsonify({'error': str(e)}), 500 + + @self.app.route('/api/dashboard/refresh', methods=['POST']) + def api_dashboard_refresh(): + """Force a snapshot refresh. Used by the health strip's manual control.""" + try: + with closing(self._dashboard_connection()) as conn: + result = self.dashboard_stats.refresh(conn) + return jsonify({'success': True, **result}) + except Exception as e: + self.logger.error(f"Error refreshing dashboard snapshot: {e}") + return jsonify({'success': False, 'error': str(e)}), 500 + @self.app.route('/api/stats/rate_limiters') def api_rate_limiter_stats(): """Return current rate limiter statistics from the running bot.""" @@ -4457,6 +4601,159 @@ class BotDataViewer: polling_thread.start() self.logger.info("Database polling started") + def _config_int(self, section: str, option: str, fallback: int) -> int: + """Read an int config value, falling back on a missing or malformed entry.""" + try: + return self.config.getint(section, option, fallback=fallback) + except (configparser.Error, ValueError, TypeError): + return fallback + + def _init_dashboard_service(self): + """Build the dashboard rollup/snapshot service from config.""" + try: + self.dashboard_snapshot_enabled = self.config.getboolean( + 'Web_Viewer', 'dashboard_snapshot_enabled', fallback=True + ) + except (configparser.Error, ValueError, TypeError): + self.dashboard_snapshot_enabled = True + + self.dashboard_snapshot_interval = max( + 15, self._config_int('Web_Viewer', 'dashboard_snapshot_interval_seconds', 60) + ) + self.dashboard_stats = DashboardStatsService( + self.logger, + history_days=self._config_int('Web_Viewer', 'dashboard_snapshot_history_days', 400), + packet_backfill_rows=self._config_int('Web_Viewer', 'dashboard_packet_backfill_rows', 2000), + interval_seconds=self.dashboard_snapshot_interval, + # Retention drives which window labels the UI is allowed to offer, + # so read the same keys the cleanup jobs enforce. + stats_retention_days=self._config_int('Stats_Command', 'data_retention_days', 7), + packet_retention_days=self._config_int('Data_Retention', 'packet_stream_retention_days', 3), + adverts_retention_days=self._config_int('Data_Retention', 'daily_stats_retention_days', 90), + multibyte_contacts_fn=self._count_contacts_7d_multibyte, + ) + + def _count_contacts_7d_multibyte(self, cursor) -> tuple[int, int] | None: + """(multibyte, total) contacts heard in the last 7 days, or None if unavailable. + + Injected into DashboardStatsService so the snapshot reuses the viewer's + memoized hop-prefix evidence instead of rebuilding it. + """ + try: + cursor.execute( + """ + SELECT COUNT(*) FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-7 days') + """ + ) + total = cursor.fetchone()[0] or 0 + except sqlite3.Error as e: + self.logger.debug(f"Could not count 7d contacts: {e}") + return None + + chunk_buckets = self._bucket_hop_chunks( + self._get_cached_contact_multibyte_hop_chunks(cursor, recent_days=7) + ) + mb_advert_pks: set[str] = set() + try: + cursor.execute( + """ + SELECT DISTINCT public_key FROM observed_paths + WHERE packet_type = 'advert' AND public_key IS NOT NULL + AND bytes_per_hop IN (2, 3) + AND date(last_seen) >= date('now', 'localtime', '-7 days') + """ + ) + mb_advert_pks = {row[0] for row in cursor.fetchall() if row[0]} + except sqlite3.Error as e: + self.logger.debug(f"Could not load 7d multibyte advert keys: {e}") + + try: + cursor.execute( + """ + SELECT public_key, role, out_bytes_per_hop + FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-7 days') + """ + ) + multibyte = sum( + 1 + for row in cursor.fetchall() + if self._contact_has_multibyte_path_evidence( + row[0], row[1], row[2], mb_advert_pks, chunk_buckets + ) + ) + except sqlite3.Error as e: + self.logger.debug(f"Could not compute contacts_7d_multibyte_path: {e}") + return None + return multibyte, total + + def _dashboard_connection(self): + """Connection for the refresher: autocommit, so BEGIN IMMEDIATE is ours. + + The reader connections leave transaction control to pysqlite, but the + refresher deliberately brackets its own write phase and must not have an + implicit transaction opened underneath it. + """ + conn = sqlite3.connect(self.db_path, timeout=60, isolation_level=None) + conn.row_factory = sqlite3.Row + with suppress(sqlite3.OperationalError): + conn.execute(f"PRAGMA busy_timeout={self._config_int('Web_Viewer', 'sqlite_busy_timeout_ms', 60000)}") + conn.execute("PRAGMA journal_mode=WAL") + return conn + + def _start_dashboard_refresher(self): + """Recompute the dashboard snapshot on an interval, in this process. + + The refresher lives in the viewer rather than the bot for two reasons: + the viewer may point at a different database ([Web_Viewer] db_path), and + it already runs migrations itself, so the rollup tables exist in + whichever file this process opens. A bot-side scheduler job would write + to the wrong file in a split-DB install and would not run at all for a + standalone viewer. + """ + if not self.dashboard_snapshot_enabled: + self.logger.info("Dashboard snapshot refresher disabled by config") + return + + def refresher(): + consecutive_errors = 0 + # Refresh once at startup so the first page load has a snapshot. + delay = 2.0 + while True: + time.sleep(delay) + delay = self.dashboard_snapshot_interval + try: + with closing(self._dashboard_connection()) as conn: + if not self.dashboard_stats.try_claim_lease(conn): + self.logger.debug( + "Another viewer holds the dashboard snapshot lease; skipping tick" + ) + continue + result = self.dashboard_stats.refresh(conn) + consecutive_errors = 0 + self.logger.debug( + "Dashboard snapshot refreshed in %sms (%s days, %s packet rows backfilled)", + result['duration_ms'], + result['days'], + result['backfilled_packet_rows'], + ) + except Exception as e: + consecutive_errors += 1 + if consecutive_errors == 1: + self.logger.error(f"Dashboard snapshot refresh failed: {e}", exc_info=True) + else: + self.logger.warning( + f"Dashboard snapshot refresh failed ({consecutive_errors}): {e}" + ) + delay = min(600, self.dashboard_snapshot_interval * (2 ** min(consecutive_errors, 5))) + + thread = threading.Thread(target=refresher, name="dashboard-snapshot", daemon=True) + thread.start() + self.logger.info( + f"Dashboard snapshot refresher started (every {self.dashboard_snapshot_interval}s)" + ) + def _start_cleanup_scheduler(self): """Start background thread for periodic database cleanup""" import threading @@ -4600,12 +4897,14 @@ class BotDataViewer: # the pie chart matches "last 7 days" (lifetime paths + stale out_bytes_per_hop # otherwise inflated the percentage). stats['contacts_7d_multibyte_path'] = 0 - multibyte_chunks: set[str] = set() + chunk_buckets = self._bucket_hop_chunks(set()) mb_advert_pks: set[str] = set() if 'observed_paths' in tables: try: - multibyte_chunks = self._collect_multibyte_hop_chunks( - cursor, recent_days=7 + chunk_buckets = self._bucket_hop_chunks( + self._get_cached_contact_multibyte_hop_chunks( + cursor, recent_days=7 + ) ) # Use date() — julianday(iso8601) often returns NULL for Python isoformat() strings cursor.execute( @@ -4636,7 +4935,7 @@ class BotDataViewer: row["role"], row["out_bytes_per_hop"], mb_advert_pks, - multibyte_chunks, + chunk_buckets, ): mb_7d += 1 stats['contacts_7d_multibyte_path'] = mb_7d @@ -4674,7 +4973,10 @@ class BotDataViewer: """) stats['unique_device_types'] = cursor.fetchone()[0] - # Incoming packets (packet_stream): multibyte path share, last 7 days (decoded bytes_per_hop) + # Incoming packets: multibyte path share over whatever packet_stream + # actually retains. The key names still say 7d for compatibility, + # but the window is reported honestly alongside them — + # packet_stream is pruned at 3 days, so the old label was never true. stats['incoming_packets_7d'] = 0 stats['incoming_packets_7d_multibyte_path'] = 0 if 'packet_stream' in tables: @@ -4682,28 +4984,23 @@ class BotDataViewer: cutoff_ts = time.time() - 7 * 86400 cursor.execute( """ - SELECT COUNT(*) FROM packet_stream - WHERE type = ? AND timestamp > ? + SELECT COUNT(*), + SUM(CASE WHEN bytes_per_hop IN (2, 3) THEN 1 ELSE 0 END), + MIN(timestamp) + FROM packet_stream + WHERE type = ? AND timestamp > ? AND route_type_name IS NOT NULL """, ("packet", cutoff_ts), ) - stats['incoming_packets_7d'] = cursor.fetchone()[0] or 0 - mb_pk = 0 - try: - cursor.execute( - """ - SELECT COUNT(*) FROM packet_stream - WHERE type = ? AND timestamp > ? - AND CAST(json_extract(data, '$.bytes_per_hop') AS INTEGER) IN (2, 3) - """, - ("packet", cutoff_ts), - ) - mb_pk = cursor.fetchone()[0] or 0 - except sqlite3.OperationalError: - mb_pk = self._count_multibyte_packets_from_stream_json(cursor, cutoff_ts) - stats['incoming_packets_7d_multibyte_path'] = mb_pk + row = cursor.fetchone() + stats['incoming_packets_7d'] = row[0] or 0 + stats['incoming_packets_7d_multibyte_path'] = row[1] or 0 + stats['incoming_packets_from'] = row[2] + stats['incoming_packets_window_label'] = humanize_span( + time.time() - row[2] if row[2] else None + ) except Exception as e: - self.logger.debug(f"Could not compute incoming_packets_7d multibyte stats: {e}") + self.logger.debug(f"Could not compute incoming packet multibyte stats: {e}") # Advertisement statistics using daily tracking table if 'daily_stats' in tables: @@ -5251,38 +5548,6 @@ class BotDataViewer: out.append(seg.lower()) return out - def _count_multibyte_packets_from_stream_json(self, cursor, cutoff_ts: float) -> int: - """Count packet_stream rows (type=packet) since cutoff with bytes_per_hop in (2, 3). JSON parse fallback.""" - import json - - n = 0 - try: - cursor.execute( - """ - SELECT data FROM packet_stream - WHERE type = ? AND timestamp > ? - """, - ("packet", cutoff_ts), - ) - for row in cursor.fetchall(): - raw = row["data"] - if not raw: - continue - try: - d = json.loads(raw) - except (json.JSONDecodeError, TypeError): - continue - bph = d.get("bytes_per_hop") - try: - bph_i = int(bph) if bph is not None else None - except (TypeError, ValueError): - bph_i = None - if bph_i in (2, 3): - n += 1 - except Exception as e: - self.logger.debug(f"packet_stream JSON scan for multibyte: {e}") - return n - def _collect_multibyte_hop_chunks( self, cursor, recent_days: int | None = None ) -> set[str]: @@ -5322,8 +5587,15 @@ class BotDataViewer: self.logger.debug(f"Could not load multibyte hop chunks: {e}") return chunks - def _get_cached_contact_multibyte_hop_chunks(self, cursor) -> set[str]: - """Return all-time contact badge evidence without rebuilding it per page request.""" + def _get_cached_contact_multibyte_hop_chunks( + self, cursor, recent_days: int | None = None + ) -> set[str]: + """Return contact badge evidence without rebuilding it per page request. + + Memoized per ``recent_days`` window: the contacts list wants all-time + evidence (None) while the dashboard wants the 7-day set, and both are + rebuilt often enough that sharing one slot would thrash the cache. + """ try: # last_seen changes when an existing path is observed again; count changes on # inserts and retention deletes. Together they cheaply invalidate the cache while @@ -5336,19 +5608,27 @@ class BotDataViewer: ) signature_row = cursor.fetchone() signature = tuple(signature_row) if signature_row else (None, 0) + if recent_days is not None: + # A windowed set also changes when the window itself slides, which + # leaves the row fingerprint untouched — pin it to the local date. + cursor.execute("SELECT date('now', 'localtime')") + signature = (*signature, cursor.fetchone()[0]) except Exception as e: self.logger.debug(f"Could not fingerprint multibyte hop chunks: {e}") - return self._collect_multibyte_hop_chunks(cursor) + return self._collect_multibyte_hop_chunks(cursor, recent_days=recent_days) lock = getattr(self, '_contacts_badge_cache_lock', None) if lock is None: lock = self._contacts_badge_cache_lock = threading.Lock() with lock: - if getattr(self, '_contacts_badge_cache_signature', None) == signature: - return self._contacts_badge_cache_chunks - chunks = self._collect_multibyte_hop_chunks(cursor) - self._contacts_badge_cache_signature = signature - self._contacts_badge_cache_chunks = chunks + cache = getattr(self, '_contacts_badge_cache', None) + if cache is None: + cache = self._contacts_badge_cache = {} + cached = cache.get(recent_days) + if cached is not None and cached[0] == signature: + return cached[1] + chunks = self._collect_multibyte_hop_chunks(cursor, recent_days=recent_days) + cache[recent_days] = (signature, chunks) return chunks @staticmethod @@ -5437,9 +5717,13 @@ class BotDataViewer: role: str | None, out_bytes_per_hop: Any, multibyte_advert_public_keys: set[str], - multibyte_hop_chunks: set[str], + chunk_buckets: dict[int, set[str]], ) -> bool: - """Multibyte detection for dashboard 7d stats (observed_paths scoped by date in SQL).""" + """Multibyte detection for dashboard 7d stats (observed_paths scoped by date in SQL). + + ``chunk_buckets`` is the length-bucketed form of the multibyte hop chunks + (see _bucket_hop_chunks) — build it once per request, not per contact. + """ pk = public_key or "" role_l = (role or "").lower() obph: int | None @@ -5455,10 +5739,11 @@ class BotDataViewer: if pk and pk in multibyte_advert_public_keys: return True if role_l in ("repeater", "roomserver") and pk: + # Chunks are fixed-length (4 or 6), so startswith reduces to prefix-equality + # lookups — O(1) per contact instead of a scan of the whole chunk set. pk_low = pk.lower() - for chunk in multibyte_hop_chunks: - if pk_low.startswith(chunk): - return True + if pk_low[:4] in chunk_buckets[4] or pk_low[:6] in chunk_buckets[6]: + return True return False def _compute_single_byte_relay_metrics( diff --git a/modules/web_viewer/dashboard_stats.py b/modules/web_viewer/dashboard_stats.py new file mode 100644 index 0000000..4cd9d57 --- /dev/null +++ b/modules/web_viewer/dashboard_stats.py @@ -0,0 +1,1296 @@ +#!/usr/bin/env python3 +"""Dashboard statistics: daily rollups, the current-state snapshot, and readers. + +The web viewer's landing page used to recompute ~50 aggregate queries on every +request, five times per page load. This module moves that work off the request +path into a background refresher that writes two tables: + +``daily_rollup`` + One row per local date. Holds the metrics whose raw sources are pruned + long before the dashboard's 30-day window — message/command/path stats at 7 + days, packet_stream at 3 — so trends outlive retention. Signal metrics are + stored as sums and counts, never means, so any window re-aggregates + correctly. + +``dashboard_snapshot`` + A single row of JSON describing current state, regenerated whole each tick. + +Every method takes a ``sqlite3.Connection`` rather than owning one, so the +service unit-tests without Flask and the caller controls transaction scope. + +Two conventions matter throughout: + +* **NULL is not zero.** A NULL column means "no source data for this day" and + must render as a gap. Writing 0 instead produces a fake cliff at the + retention boundary that reads as an outage. +* **Local dates, clamped timestamps.** Rollup dates are local to match + ``daily_stats`` (the bot writes ``date('now','localtime')``), while the stats + tables store UTC epochs. Those tables also contain future-dated rows + (observed up to year 2103), so every query clamps to now — otherwise the + rollup grows a row for 2103 and the x-axis spans 77 years. +""" + +from __future__ import annotations + +import json +import math +import os +import socket +import sqlite3 +import time +from contextlib import suppress +from datetime import datetime, timedelta +from typing import Any, Callable + +SCHEMA_REV = 1 + +LEASE_KEY = "dashboard.snapshot_lease" + +# Source-presence bitmask stored on each daily_rollup row. A missing table is +# genuinely possible: message_stats/command_stats/path_stats are created by +# stats_command, not by migrations, so `collect_stats = false` installs lack them. +SOURCE_MESSAGE_STATS = 1 << 0 +SOURCE_COMMAND_STATS = 1 << 1 +SOURCE_PATH_STATS = 1 << 2 +SOURCE_PACKET_STREAM = 1 << 3 +SOURCE_DAILY_STATS = 1 << 4 +SOURCE_CONTACT_TRACKING = 1 << 5 +SOURCE_OBSERVED_PATHS = 1 << 6 + +SOURCE_NAMES = { + SOURCE_MESSAGE_STATS: "message_stats", + SOURCE_COMMAND_STATS: "command_stats", + SOURCE_PATH_STATS: "path_stats", + SOURCE_PACKET_STREAM: "packet_stream", + SOURCE_DAILY_STATS: "daily_stats", + SOURCE_CONTACT_TRACKING: "complete_contact_tracking", + SOURCE_OBSERVED_PATHS: "observed_paths", +} + +# Series exposed by /api/dashboard/series and folded into the summary payload. +# multibyte_share is a ratio the UI plots on a 0-100 axis; the rest are counts. +SERIES_METRICS: dict[str, str] = { + "messages": "messages_total", + "commands": "commands_total", + "adverts": "adverts_total", + "nodes": "nodes_active", + "new_nodes": "nodes_new", + "packets": "packets_total", + "multibyte_share": ( + "CASE WHEN COALESCE(adverts_from_multibyte, 0) + COALESCE(adverts_from_singlebyte, 0) > 0 " + "THEN ROUND(adverts_from_multibyte * 100.0 / " + "(adverts_from_multibyte + adverts_from_singlebyte), 1) END" + ), +} + +SUMMARY_SERIES_POINTS = 30 + +# Metrics that are already a ratio: a period "total" has to be the mean of the +# daily values, not their sum — adding percentages together means nothing. +RATIO_METRICS = frozenset({"multibyte_share"}) + +TOP_KINDS = ("users", "commands", "channels", "paths", "repeaters") + +# Roles the firmware reports as an unmapped enum ordinal. They are real +# contacts, so they belong in the mix — just not as sixteen singleton slices. +_UNKNOWN_ROLE_PREFIX = "type" + + +def local_date_str(when: float | None = None) -> str: + """Local YYYY-MM-DD for *when* (default now), matching date('now','localtime').""" + return datetime.fromtimestamp(time.time() if when is None else when).strftime("%Y-%m-%d") + + +def day_bounds(date_str: str) -> tuple[float, float]: + """Local-midnight epoch bounds [start, end) for *date_str*. + + Computed in Python rather than with date(timestamp,'unixepoch','localtime') + so the comparison stays sargable against the timestamp indexes, and so DST + transitions produce a genuine 23- or 25-hour day. + """ + start = datetime.strptime(date_str, "%Y-%m-%d") + return start.timestamp(), (start + timedelta(days=1)).timestamp() + + +def _table_exists(conn: sqlite3.Connection, table: str) -> bool: + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + return row is not None + + +def _percentile(sorted_values: list[float], fraction: float) -> float | None: + """Linear-interpolated percentile over a pre-sorted list.""" + if not sorted_values: + return None + if len(sorted_values) == 1: + return round(float(sorted_values[0]), 1) + position = fraction * (len(sorted_values) - 1) + low = math.floor(position) + high = math.ceil(position) + if low == high: + return round(float(sorted_values[low]), 1) + weight = position - low + value = sorted_values[low] * (1 - weight) + sorted_values[high] * weight + return round(float(value), 1) + + +def _change_pct(current: float | None, previous: float | None) -> float | None: + """Percent change, or None when the baseline cannot support one. + + Computed server-side precisely so the client never has to decide what + dividing by zero means. + """ + if current is None or previous is None: + return None + if previous == 0: + return None + return round(((current - previous) / previous) * 100, 1) + + +def normalize_role(role: str | None) -> str: + """Map firmware role values to display buckets, folding unmapped ordinals. + + A handful of contacts report role as the raw enum ordinal ('type0'…'type15') + rather than a name. Charting each as its own slice buries the real roles. + """ + text = (role or "").strip() + if not text: + return "Unknown" + lowered = text.lower() + if lowered.startswith(_UNKNOWN_ROLE_PREFIX) and lowered[len(_UNKNOWN_ROLE_PREFIX):].isdigit(): + return "Unknown" + return lowered + + +def humanize_span(seconds: float | None) -> str: + """Render a covered-window length as 'last 2d 6h' / 'last 45m'.""" + if seconds is None or seconds <= 0: + return "no data" + days, remainder = divmod(int(seconds), 86400) + hours, remainder = divmod(remainder, 3600) + minutes = remainder // 60 + if days: + return f"last {days}d {hours}h" if hours else f"last {days}d" + if hours: + return f"last {hours}h {minutes}m" if minutes else f"last {hours}h" + return f"last {max(1, minutes)}m" + + +class DashboardStatsService: + """Computes and reads the dashboard's rollup and snapshot tables. + + ``multibyte_contacts_fn`` is injected by the app so the service can reuse + the viewer's cached hop-prefix evidence without importing Flask. It takes a + cursor and returns ``(multibyte_count, total_count)`` for contacts heard in + the last 7 days, or None if it cannot be computed. + """ + + def __init__( + self, + logger, + *, + history_days: int = 400, + packet_backfill_rows: int = 2000, + interval_seconds: int = 60, + stats_retention_days: int = 7, + packet_retention_days: int = 3, + adverts_retention_days: int = 90, + multibyte_contacts_fn: Callable[[sqlite3.Cursor], tuple[int, int] | None] | None = None, + ) -> None: + self.logger = logger + self.history_days = max(7, int(history_days)) + self.packet_backfill_rows = max(0, int(packet_backfill_rows)) + self.interval_seconds = max(5, int(interval_seconds)) + self.stats_retention_days = max(1, int(stats_retention_days)) + self.packet_retention_days = max(1, int(packet_retention_days)) + self.adverts_retention_days = max(1, int(adverts_retention_days)) + self.multibyte_contacts_fn = multibyte_contacts_fn + # Recompute today plus a trailing window, bounded by each source's own + # retention: recomputing a day whose raw rows were just pruned would + # overwrite a real value with a zero. + self.trailing_days = 3 + + # -- source availability ------------------------------------------------- + + def detect_sources(self, conn: sqlite3.Connection) -> int: + """Bitmask of the source tables actually present in this database.""" + present = 0 + for bit, name in SOURCE_NAMES.items(): + if _table_exists(conn, name): + present |= bit + return present + + def _retention_for(self, bit: int) -> int: + if bit == SOURCE_PACKET_STREAM: + return self.packet_retention_days + if bit in (SOURCE_MESSAGE_STATS, SOURCE_COMMAND_STATS, SOURCE_PATH_STATS): + return self.stats_retention_days + return self.adverts_retention_days + + def _recompute_dates(self, today: str) -> list[str]: + """Dates to recompute this tick: today plus the trailing window.""" + base = datetime.strptime(today, "%Y-%m-%d") + return [ + (base - timedelta(days=offset)).strftime("%Y-%m-%d") + for offset in range(self.trailing_days + 1) + ] + + # -- packet dimension backfill ------------------------------------------ + + def backfill_packet_dims(self, conn: sqlite3.Connection) -> int: + """Populate denormalized packet columns for a bounded batch of old rows. + + Rows written before migration 0019 carry their dimensions only inside + the JSON blob. Widening all of them at once rewrites the whole table + into the WAL, so this walks backwards a batch per tick instead; with + 3-day retention the backlog clears long before it matters. + + route_type_name doubles as the "already dimensioned" marker, so a row + whose JSON has no route type is stamped UNKNOWN_ROUTE rather than left + NULL — otherwise the same batch would be reselected on every tick and + the backlog would never advance. + + Returns the number of rows updated. + """ + if not self.packet_backfill_rows or not _table_exists(conn, "packet_stream"): + return 0 + try: + cursor = conn.execute( + """ + UPDATE packet_stream SET + route_type_name = COALESCE(json_extract(data, '$.route_type_name'), + 'UNKNOWN_ROUTE'), + payload_type_name = json_extract(data, '$.payload_type_name'), + path_len = CAST(json_extract(data, '$.path_len') AS INTEGER), + bytes_per_hop = CAST(json_extract(data, '$.bytes_per_hop') AS INTEGER) + WHERE id IN ( + SELECT id FROM packet_stream + WHERE type = 'packet' AND route_type_name IS NULL + ORDER BY id DESC LIMIT ? + ) + """, + (self.packet_backfill_rows,), + ) + return cursor.rowcount or 0 + except sqlite3.OperationalError as exc: + # Older SQLite builds without the JSON1 extension, or a viewer DB + # that predates the migration. Packet tiles degrade to "unknown". + self.logger.debug(f"packet_stream dimension backfill unavailable: {exc}") + return 0 + + def packet_coverage(self, conn: sqlite3.Connection) -> dict[str, Any]: + """Actual time span covered by packet rows carrying denormalized dims. + + The old dashboard labelled this window "last 7 days" while packet_stream + retention is 3 — reporting the measured span instead of a config value + is the whole point. + """ + coverage: dict[str, Any] = { + "packets_from": None, + "packets_to": None, + "packets_window_label": "no data", + "packets_with_dims": 0, + } + if not _table_exists(conn, "packet_stream"): + return coverage + row = conn.execute( + """ + SELECT MIN(timestamp), MAX(timestamp), COUNT(*) + FROM packet_stream + WHERE type = 'packet' AND route_type_name IS NOT NULL + """ + ).fetchone() + if not row or not row[2]: + return coverage + coverage["packets_from"] = row[0] + coverage["packets_to"] = row[1] + coverage["packets_with_dims"] = row[2] + coverage["packets_window_label"] = humanize_span(time.time() - row[0]) + return coverage + + # -- daily rollup -------------------------------------------------------- + + def _first_advert_dates(self, conn: sqlite3.Connection) -> dict[str, str]: + """public_key -> earliest daily_stats date, for new-node attribution. + + Computed once per refresh: the alternative is a HAVING MIN(date) scan + per day, which the backfill would run ninety times. + """ + if not _table_exists(conn, "daily_stats"): + return {} + return { + row[0]: row[1] + for row in conn.execute( + "SELECT public_key, MIN(date) FROM daily_stats GROUP BY public_key" + ) + } + + def _message_metrics( + self, conn: sqlite3.Connection, start: float, end: float + ) -> dict[str, Any]: + row = conn.execute( + """ + SELECT COUNT(*), + SUM(CASE WHEN is_dm THEN 1 ELSE 0 END), + SUM(CASE WHEN channel IS NOT NULL AND channel != '' THEN 1 ELSE 0 END), + COUNT(DISTINCT sender_id), + COUNT(DISTINCT channel), + SUM(snr), SUM(CASE WHEN snr IS NOT NULL THEN 1 ELSE 0 END), + SUM(rssi), SUM(CASE WHEN rssi IS NOT NULL THEN 1 ELSE 0 END), + SUM(hops), SUM(CASE WHEN hops IS NOT NULL THEN 1 ELSE 0 END) + FROM message_stats + WHERE timestamp >= ? AND timestamp < ? + """, + (start, end), + ).fetchone() + return { + "messages_total": row[0] or 0, + "messages_dm": row[1] or 0, + "messages_channel": row[2] or 0, + "unique_senders": row[3] or 0, + "unique_channels": row[4] or 0, + "snr_sum": row[5], + "snr_count": row[6] or 0, + "rssi_sum": row[7], + "rssi_count": row[8] or 0, + "hops_sum": row[9], + "hops_count": row[10] or 0, + } + + def _command_metrics( + self, conn: sqlite3.Connection, start: float, end: float + ) -> dict[str, Any]: + row = conn.execute( + """ + SELECT COUNT(*), + SUM(CASE WHEN response_sent THEN 1 ELSE 0 END), + COUNT(DISTINCT sender_id) + FROM command_stats + WHERE timestamp >= ? AND timestamp < ? + """, + (start, end), + ).fetchone() + return { + "commands_total": row[0] or 0, + "commands_replied": row[1] or 0, + "unique_command_users": row[2] or 0, + } + + def _path_metrics( + self, conn: sqlite3.Connection, start: float, end: float + ) -> dict[str, Any]: + row = conn.execute( + """ + SELECT COUNT(*), MAX(path_length), SUM(path_length) + FROM path_stats + WHERE timestamp >= ? AND timestamp < ? + """, + (start, end), + ).fetchone() + return { + "path_obs_total": row[0] or 0, + "path_len_max": row[1], + "path_len_sum": row[2], + } + + def _packet_metrics( + self, conn: sqlite3.Connection, start: float, end: float + ) -> dict[str, Any]: + row = conn.execute( + """ + SELECT COUNT(*), + SUM(CASE WHEN route_type_name LIKE '%FLOOD' THEN 1 ELSE 0 END), + SUM(CASE WHEN route_type_name LIKE '%DIRECT' THEN 1 ELSE 0 END), + SUM(CASE WHEN bytes_per_hop IN (2, 3) THEN 1 ELSE 0 END) + FROM packet_stream + WHERE type = 'packet' AND timestamp >= ? AND timestamp < ? + """, + (start, end), + ).fetchone() + return { + "packets_total": row[0] or 0, + "packets_flood": row[1] or 0, + "packets_direct": row[2] or 0, + "packets_multibyte": row[3] or 0, + } + + def _advert_metrics( + self, + conn: sqlite3.Connection, + date_str: str, + first_seen: dict[str, str], + multibyte_keys: set[str] | None, + ) -> dict[str, Any]: + row = conn.execute( + "SELECT SUM(advert_count), COUNT(DISTINCT public_key) FROM daily_stats WHERE date = ?", + (date_str,), + ).fetchone() + metrics: dict[str, Any] = { + "adverts_total": row[0] or 0, + "nodes_active": row[1] or 0, + "nodes_new": sum(1 for first in first_seen.values() if first == date_str), + "adverts_from_multibyte": None, + "adverts_from_singlebyte": None, + } + if multibyte_keys is None: + return metrics + # Classification is only knowable as of now, so this split is written + # for the recompute window and then frozen — see the tooltip copy. + multibyte = single = 0 + for key, count in conn.execute( + "SELECT public_key, advert_count FROM daily_stats WHERE date = ?", (date_str,) + ): + if key in multibyte_keys: + multibyte += count or 0 + else: + single += count or 0 + metrics["adverts_from_multibyte"] = multibyte + metrics["adverts_from_singlebyte"] = single + return metrics + + def _multibyte_advert_keys(self, conn: sqlite3.Connection) -> set[str] | None: + """Public keys currently classified as advertising over multibyte paths.""" + if not _table_exists(conn, "observed_paths"): + return None + try: + return { + row[0] + for row in conn.execute( + """ + SELECT DISTINCT public_key FROM observed_paths + WHERE packet_type = 'advert' AND public_key IS NOT NULL + AND bytes_per_hop IN (2, 3) + """ + ) + if row[0] + } + except sqlite3.Error as exc: + self.logger.debug(f"Could not load multibyte advert keys: {exc}") + return None + + def compute_day( + self, + conn: sqlite3.Connection, + date_str: str, + *, + sources: int, + today: str, + first_seen: dict[str, str], + multibyte_keys: set[str] | None, + backfill: bool = False, + ) -> dict[str, Any]: + """Recompute every rollup column for one local date. + + Families whose raw rows have already aged out contribute NULL rather + than 0, so an upsert preserves whatever history is already stored. + """ + start, end = day_bounds(date_str) + now = time.time() + # Future-dated rows exist in the wild (max observed: 2103-08-15); an + # unclamped upper bound would mint a rollup row for that year. + end = min(end, now) + values: dict[str, Any] = {"date": date_str} + present = 0 + + if end <= start: + # Entirely in the future — nothing observable belongs to this date. + return {"values": values, "sources": 0} + + age_days = (datetime.strptime(today, "%Y-%m-%d") - datetime.strptime(date_str, "%Y-%m-%d")).days + + def _in_retention(bit: int) -> bool: + return age_days < self._retention_for(bit) + + for bit, compute in ( + (SOURCE_MESSAGE_STATS, self._message_metrics), + (SOURCE_COMMAND_STATS, self._command_metrics), + (SOURCE_PATH_STATS, self._path_metrics), + (SOURCE_PACKET_STREAM, self._packet_metrics), + ): + if not (sources & bit) or not _in_retention(bit): + continue + try: + values.update(compute(conn, start, end)) + present |= bit + except sqlite3.Error as exc: + self.logger.debug(f"Rollup {SOURCE_NAMES[bit]} for {date_str}: {exc}") + + if sources & SOURCE_DAILY_STATS: + try: + values.update( + self._advert_metrics(conn, date_str, first_seen, multibyte_keys) + ) + present |= SOURCE_DAILY_STATS + except sqlite3.Error as exc: + self.logger.debug(f"Rollup daily_stats for {date_str}: {exc}") + + # Contact totals are a point-in-time reading with no historical form, so + # only today's row gets them; older rows keep whatever was stored then. + if (sources & SOURCE_CONTACT_TRACKING) and date_str == today and not backfill: + try: + row = conn.execute( + """ + SELECT COUNT(*), SUM(CASE WHEN is_currently_tracked = 1 THEN 1 ELSE 0 END) + FROM complete_contact_tracking + """ + ).fetchone() + values["contacts_known"] = row[0] or 0 + values["contacts_tracked"] = row[1] or 0 + present |= SOURCE_CONTACT_TRACKING + except sqlite3.Error as exc: + self.logger.debug(f"Rollup contact totals for {date_str}: {exc}") + + return {"values": values, "sources": present} + + ROLLUP_COLUMNS = ( + "messages_total", "messages_dm", "messages_channel", + "unique_senders", "unique_channels", + "commands_total", "commands_replied", "unique_command_users", + "path_obs_total", "path_len_max", "path_len_sum", + "snr_sum", "snr_count", "rssi_sum", "rssi_count", + "hops_sum", "hops_count", + "packets_total", "packets_flood", "packets_direct", "packets_multibyte", + "adverts_total", "nodes_active", "nodes_new", + "adverts_from_multibyte", "adverts_from_singlebyte", + "contacts_known", "contacts_tracked", + ) + + def upsert_day( + self, + conn: sqlite3.Connection, + computed: dict[str, Any], + *, + is_backfilled: bool = False, + is_final: bool = False, + ) -> None: + """Write one recomputed date. + + Idempotent: every value is a full recomputation from raw rows, not an + increment. Columns arrive as NULL when their source is absent or aged + out, and COALESCE keeps the stored value rather than erasing history. + """ + values = computed["values"] + columns = ("date", *self.ROLLUP_COLUMNS, "sources_present", "is_backfilled", "is_final", "computed_at") + params = [values["date"]] + params.extend(values.get(column) for column in self.ROLLUP_COLUMNS) + params.extend([ + computed["sources"], + 1 if is_backfilled else 0, + 1 if is_final else 0, + datetime.now().isoformat(timespec="seconds"), + ]) + assignments = ", ".join( + f"{column} = COALESCE(excluded.{column}, daily_rollup.{column})" + for column in self.ROLLUP_COLUMNS + ) + conn.execute( + f""" + INSERT INTO daily_rollup ({', '.join(columns)}) + VALUES ({', '.join('?' * len(columns))}) + ON CONFLICT(date) DO UPDATE SET + {assignments}, + sources_present = daily_rollup.sources_present | excluded.sources_present, + is_backfilled = CASE WHEN excluded.is_backfilled = 0 THEN 0 + ELSE daily_rollup.is_backfilled END, + is_final = MAX(daily_rollup.is_final, excluded.is_final), + computed_at = excluded.computed_at + """, + params, + ) + + def plan_backfill(self, conn: sqlite3.Connection) -> list[dict[str, Any]]: + """Compute (read-only) one rollup row per date present in daily_stats. + + Message, command, path and packet columns stay NULL: those raw rows were + pruned days ago and cannot be recovered. The UI renders them as gaps. + Separated from the write so the ~90-day pass never holds a write lock + while it reads. + """ + if not _table_exists(conn, "daily_stats"): + return [] + today = local_date_str() + horizon = ( + datetime.strptime(today, "%Y-%m-%d") - timedelta(days=self.history_days) + ).strftime("%Y-%m-%d") + dates = [ + row[0] + for row in conn.execute( + "SELECT DISTINCT date FROM daily_stats WHERE date >= ? AND date <= ? ORDER BY date", + (horizon, today), + ) + if row[0] + ] + if not dates: + return [] + first_seen = self._first_advert_dates(conn) + sources = self.detect_sources(conn) & SOURCE_DAILY_STATS + return [ + self.compute_day( + conn, + date_str, + sources=sources, + today=today, + first_seen=first_seen, + multibyte_keys=None, + backfill=True, + ) + for date_str in dates + ] + + def write_backfill(self, conn: sqlite3.Connection, planned: list[dict[str, Any]]) -> int: + """Persist a planned backfill. Caller owns the transaction.""" + today = local_date_str() + for computed in planned: + self.upsert_day( + conn, + computed, + is_backfilled=True, + is_final=computed["values"]["date"] < today, + ) + return len(planned) + + def has_history(self, conn: sqlite3.Connection) -> bool: + """True once daily_rollup holds anything — backfill runs only when empty.""" + row = conn.execute("SELECT 1 FROM daily_rollup LIMIT 1").fetchone() + return row is not None + + def prune_history(self, conn: sqlite3.Connection) -> int: + cutoff = (datetime.now() - timedelta(days=self.history_days)).strftime("%Y-%m-%d") + cursor = conn.execute("DELETE FROM daily_rollup WHERE date < ?", (cutoff,)) + return cursor.rowcount or 0 + + # -- snapshot ------------------------------------------------------------ + + def _mesh_snapshot(self, conn: sqlite3.Connection, sources: int) -> dict[str, Any]: + mesh: dict[str, Any] = {} + today = local_date_str() + + if sources & SOURCE_DAILY_STATS: + row = conn.execute( + "SELECT SUM(advert_count), COUNT(DISTINCT public_key) FROM daily_stats WHERE date = ?", + (today,), + ).fetchone() + mesh["adverts_24h"] = row[0] or 0 + mesh["nodes_24h"] = row[1] or 0 + week_ago = (datetime.now() - timedelta(days=6)).strftime("%Y-%m-%d") + row = conn.execute( + """ + SELECT SUM(CASE WHEN first_date = ? THEN 1 ELSE 0 END), + SUM(CASE WHEN first_date >= ? THEN 1 ELSE 0 END) + FROM (SELECT MIN(date) AS first_date FROM daily_stats GROUP BY public_key) + """, + (today, week_ago), + ).fetchone() + mesh["new_nodes_24h"] = row[0] or 0 + mesh["new_nodes_7d"] = row[1] or 0 + + if sources & SOURCE_CONTACT_TRACKING: + row = conn.execute( + """ + SELECT COUNT(*), + SUM(CASE WHEN is_currently_tracked = 1 THEN 1 ELSE 0 END), + SUM(CASE WHEN last_heard < datetime('now','localtime','-7 days') + AND last_heard >= datetime('now','localtime','-14 days') + THEN 1 ELSE 0 END) + FROM complete_contact_tracking + """ + ).fetchone() + mesh["contacts_known"] = row[0] or 0 + mesh["contacts_tracked"] = row[1] or 0 + # "Heard last week, silent this week" — a churn signal, not a total. + mesh["gone_quiet_7d"] = row[2] or 0 + + # Each column carries its own emptiness test rather than sharing one + # in the WHERE clause: filtering the row set on country would drop + # contacts that have a state or city but no country, undercounting + # both. COUNT(DISTINCT ...) already ignores the NULLs. + row = conn.execute( + """ + SELECT COUNT(DISTINCT CASE + WHEN country IS NULL OR country = '' THEN NULL + WHEN country IN ('United States','United States of America','US','USA') + THEN 'United States' ELSE country END), + COUNT(DISTINCT CASE WHEN state IS NOT NULL AND state != '' THEN state END), + COUNT(DISTINCT CASE WHEN city IS NOT NULL AND city != '' THEN city END) + FROM complete_contact_tracking + WHERE last_heard > datetime('now','localtime','-30 days') + AND is_currently_tracked = 1 + """ + ).fetchone() + mesh["countries"] = row[0] or 0 + mesh["states"] = row[1] or 0 + mesh["cities"] = row[2] or 0 + + mesh["hop_histogram"] = [ + [int(hop), int(count)] + for hop, count in conn.execute( + """ + SELECT hop_count, COUNT(*) FROM complete_contact_tracking + WHERE hop_count IS NOT NULL AND hop_count BETWEEN 0 AND 32 + GROUP BY hop_count ORDER BY hop_count + """ + ) + ] + mesh["role_mix"] = self._mix(conn, "role") + mesh["device_mix"] = self._mix(conn, "device_type") + + if sources & SOURCE_OBSERVED_PATHS: + mesh["path_len_histogram"] = [ + [int(length), int(count)] + for length, count in conn.execute( + """ + SELECT path_length, COUNT(*) FROM observed_paths + WHERE packet_type = 'advert' AND path_length IS NOT NULL + AND path_length BETWEEN 0 AND 32 + AND last_seen >= datetime('now','localtime','-7 days') + GROUP BY path_length ORDER BY path_length + """ + ) + ] + + if sources & SOURCE_PACKET_STREAM: + row = conn.execute( + """ + SELECT SUM(CASE WHEN route_type_name LIKE '%FLOOD' THEN 1 ELSE 0 END), + SUM(CASE WHEN route_type_name LIKE '%DIRECT' THEN 1 ELSE 0 END), + SUM(CASE WHEN bytes_per_hop IN (2, 3) THEN 1 ELSE 0 END), + COUNT(*) + FROM packet_stream + WHERE type = 'packet' AND route_type_name IS NOT NULL + """ + ).fetchone() + mesh["route_mix"] = {"flood": row[0] or 0, "direct": row[1] or 0} + mesh.setdefault("encoding", {})["packets"] = { + "multibyte": row[2] or 0, + "total": row[3] or 0, + } + + if sources & SOURCE_MESSAGE_STATS: + mesh["snr"], mesh["rssi"] = self._signal_distribution(conn) + + if self.multibyte_contacts_fn is not None: + try: + result = self.multibyte_contacts_fn(conn.cursor()) + except Exception as exc: # noqa: BLE001 - injected callback + self.logger.debug(f"Multibyte contact evidence unavailable: {exc}") + result = None + if result is not None: + multibyte, total = result + mesh.setdefault("encoding", {})["contacts_7d"] = { + "multibyte": multibyte, + "total": total, + } + return mesh + + def _mix(self, conn: sqlite3.Connection, column: str) -> list[list[Any]]: + """Contact counts per role/device_type, unmapped ordinals folded into Unknown.""" + buckets: dict[str, int] = {} + for value, count in conn.execute( + f"SELECT {column}, COUNT(*) FROM complete_contact_tracking GROUP BY {column}" # noqa: S608 - fixed identifiers + ): + key = normalize_role(value) if column == "role" else (value or "Unknown") + buckets[key] = buckets.get(key, 0) + (count or 0) + return [[name, count] for name, count in sorted(buckets.items(), key=lambda kv: -kv[1])] + + def _signal_distribution(self, conn: sqlite3.Connection) -> tuple[dict, dict]: + """SNR/RSSI distribution over the last 7 days of received messages. + + Deliberately sourced from message_stats, where snr is populated on every + row, rather than complete_contact_tracking, where it is populated on ~7% + — presenting that biased sample as network-wide would be a lie. + """ + cutoff = time.time() - 7 * 86400 + snr_values = sorted( + float(row[0]) + for row in conn.execute( + "SELECT snr FROM message_stats WHERE snr IS NOT NULL AND timestamp >= ? AND timestamp <= ?", + (cutoff, time.time()), + ) + ) + rssi_values = sorted( + float(row[0]) + for row in conn.execute( + "SELECT rssi FROM message_stats WHERE rssi IS NOT NULL AND timestamp >= ? AND timestamp <= ?", + (cutoff, time.time()), + ) + ) + snr = { + "p10": _percentile(snr_values, 0.10), + "p50": _percentile(snr_values, 0.50), + "p90": _percentile(snr_values, 0.90), + "n": len(snr_values), + } + rssi = {"p50": _percentile(rssi_values, 0.50), "n": len(rssi_values)} + return snr, rssi + + def _bot_snapshot(self, conn: sqlite3.Connection, sources: int) -> dict[str, Any]: + """Rolling 24-hour bot activity (distinct from the calendar-day rollup).""" + bot: dict[str, Any] = {} + now = time.time() + cutoff = now - 86400 + if sources & SOURCE_MESSAGE_STATS: + row = conn.execute( + """ + SELECT COUNT(*), COUNT(DISTINCT sender_id), COUNT(DISTINCT channel) + FROM message_stats WHERE timestamp >= ? AND timestamp <= ? + """, + (cutoff, now), + ).fetchone() + bot["messages_24h"] = row[0] or 0 + bot["unique_users_24h"] = row[1] or 0 + bot["unique_channels_24h"] = row[2] or 0 + if sources & SOURCE_COMMAND_STATS: + row = conn.execute( + """ + SELECT COUNT(*), SUM(CASE WHEN response_sent THEN 1 ELSE 0 END) + FROM command_stats WHERE timestamp >= ? AND timestamp <= ? + """, + (cutoff, now), + ).fetchone() + total = row[0] or 0 + bot["commands_24h"] = total + bot["reply_rate_24h"] = round(((row[1] or 0) / total) * 100, 1) if total else None + return bot + + def _series_from_rollup( + self, conn: sqlite3.Connection, points: int = SUMMARY_SERIES_POINTS + ) -> dict[str, list[dict[str, Any]]]: + expressions = ", ".join(f"{sql} AS {name}" for name, sql in SERIES_METRICS.items()) + rows = conn.execute( + f""" + SELECT date, is_final, {expressions} + FROM daily_rollup ORDER BY date DESC LIMIT ? + """, + (points,), + ).fetchall() + series: dict[str, list[dict[str, Any]]] = {name: [] for name in SERIES_METRICS} + for row in reversed(rows): + for name in SERIES_METRICS: + series[name].append( + { + "date": row["date"], + "value": row[name], + "complete": bool(row["is_final"]), + } + ) + return series + + def _deltas_from_rollup(self, conn: sqlite3.Connection) -> dict[str, float | None]: + """Percent change of the last complete day against the day before it. + + Explicitly calendar days, not a rolling 24 hours: quoting one against + the other is the classic dashboard lie, so the UI says which it is. + """ + expressions = ", ".join(f"{sql} AS {name}" for name, sql in SERIES_METRICS.items()) + rows = conn.execute( + f""" + SELECT date, {expressions} FROM daily_rollup + WHERE is_final = 1 ORDER BY date DESC LIMIT 2 + """ + ).fetchall() + if len(rows) < 2: + return {} + current, previous = rows[0], rows[1] + return {name: _change_pct(current[name], previous[name]) for name in SERIES_METRICS} + + def build_snapshot(self, conn: sqlite3.Connection) -> dict[str, Any]: + """Assemble the whole current-state payload. Read-only.""" + sources = self.detect_sources(conn) + payload: dict[str, Any] = { + "coverage": { + **self.packet_coverage(conn), + "stats_retention_days": self.stats_retention_days, + "packet_retention_days": self.packet_retention_days, + "adverts_retention_days": self.adverts_retention_days, + "sources_present": [ + name for bit, name in SOURCE_NAMES.items() if sources & bit + ], + }, + "mesh": self._mesh_snapshot(conn, sources), + "bot": self._bot_snapshot(conn, sources), + } + payload["series"] = self._series_from_rollup(conn) + payload["deltas"] = self._deltas_from_rollup(conn) + return payload + + # -- refresh orchestration ---------------------------------------------- + + def refresh(self, conn: sqlite3.Connection, *, backfill: bool | None = None) -> dict[str, Any]: + """Recompute rollups and the snapshot. + + The whole read phase runs outside any explicit transaction and every + write goes in one short BEGIN IMMEDIATE at the end, so a refresh never + holds a write lock while it reads and cannot stall the bot's inserts. + + ``backfill`` defaults to "only if daily_rollup is empty". + """ + started = time.perf_counter() + today = local_date_str() + + # --- write phase 0: dimension a batch of older packet rows ----------- + # First, not last: every packet figure below reads the denormalized + # columns, so backfilling afterwards would make each tick describe the + # previous tick's coverage and leave the counts drifting upward until + # the backlog cleared. + conn.execute("BEGIN IMMEDIATE") + try: + backfilled_packets = self.backfill_packet_dims(conn) + conn.commit() + except Exception: + conn.rollback() + raise + + # --- read phase 1: everything the rollup rows need ------------------- + sources = self.detect_sources(conn) + first_seen = self._first_advert_dates(conn) + multibyte_keys = self._multibyte_advert_keys(conn) + + if backfill is None: + backfill = not self.has_history(conn) + planned_backfill = self.plan_backfill(conn) if backfill else [] + + computed_days = [ + self.compute_day( + conn, + date_str, + sources=sources, + today=today, + first_seen=first_seen, + multibyte_keys=multibyte_keys, + ) + for date_str in self._recompute_dates(today) + ] + oldest_recomputed = min(day["values"]["date"] for day in computed_days) + + # --- write phase 1: rollups and the packet dimension backfill -------- + conn.execute("BEGIN IMMEDIATE") + try: + if planned_backfill: + self.write_backfill(conn, planned_backfill) + for day in computed_days: + self.upsert_day(conn, day, is_final=day["values"]["date"] < today) + # A day that has left the recompute window will never change again. + conn.execute( + "UPDATE daily_rollup SET is_final = 1 WHERE date < ? AND is_final = 0", + (oldest_recomputed,), + ) + self.prune_history(conn) + conn.commit() + except Exception: + conn.rollback() + raise + + # --- read phase 2: the snapshot, which reads what we just wrote ------ + # Series and deltas come from daily_rollup and the packet tiles from the + # freshly dimensioned rows, so this has to follow the write, not precede + # it — otherwise every snapshot describes the previous tick. + payload = self.build_snapshot(conn) + generated_at = time.time() + duration_ms = int((time.perf_counter() - started) * 1000) + + # --- write phase 2: the snapshot row --------------------------------- + conn.execute("BEGIN IMMEDIATE") + try: + conn.execute( + """ + INSERT INTO dashboard_snapshot (id, generated_at, duration_ms, schema_rev, payload) + VALUES (1, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + generated_at = excluded.generated_at, + duration_ms = excluded.duration_ms, + schema_rev = excluded.schema_rev, + payload = excluded.payload + """, + (generated_at, duration_ms, SCHEMA_REV, json.dumps(payload)), + ) + conn.commit() + except Exception: + conn.rollback() + raise + + return { + "generated_at": generated_at, + "duration_ms": duration_ms, + "days": len(computed_days), + "backfilled_days": len(planned_backfill), + "backfilled_packet_rows": backfilled_packets, + } + + # -- lease --------------------------------------------------------------- + + def try_claim_lease(self, conn: sqlite3.Connection) -> bool: + """Best-effort exclusion between viewers sharing one database. + + Refreshes are idempotent full recomputations, so a double refresh wastes + CPU and corrupts nothing. This exists to skip that waste, not to + guarantee correctness — treat a lost race as "someone else has it". + """ + if not _table_exists(conn, "bot_metadata"): + return True + now = time.time() + owner = f"{os.getpid()}:{socket.gethostname()}:{now + self.interval_seconds * 2:.0f}" + try: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT value FROM bot_metadata WHERE key = ?", (LEASE_KEY,) + ).fetchone() + if row and row[0]: + parts = str(row[0]).rsplit(":", 1) + try: + expires = float(parts[-1]) + except (TypeError, ValueError): + expires = 0.0 + held_by_other = parts[0] != f"{os.getpid()}:{socket.gethostname()}" + if expires > now and held_by_other: + conn.rollback() + return False + conn.execute( + """ + INSERT INTO bot_metadata (key, value, updated_at) + VALUES (?, ?, CURRENT_TIMESTAMP) + ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP + """, + (LEASE_KEY, owner), + ) + conn.commit() + return True + except sqlite3.Error as exc: + # Losing the race to another viewer surfaces here as a lock error; + # treat any failure as "someone else has it" and skip the tick. + with suppress(sqlite3.Error): + conn.rollback() + self.logger.debug(f"Snapshot lease not claimed: {exc}") + return False + + # -- readers ------------------------------------------------------------- + + def read_summary(self, conn: sqlite3.Connection) -> dict[str, Any] | None: + """Return the stored snapshot with freshness metadata, or None if unset.""" + if not _table_exists(conn, "dashboard_snapshot"): + return None + row = conn.execute( + "SELECT generated_at, duration_ms, schema_rev, payload FROM dashboard_snapshot WHERE id = 1" + ).fetchone() + if not row: + return None + try: + payload = json.loads(row["payload"]) + except (TypeError, ValueError) as exc: + self.logger.warning(f"Discarding unreadable dashboard snapshot: {exc}") + return None + age = max(0.0, time.time() - row["generated_at"]) + payload.update( + { + "generated_at": row["generated_at"], + "duration_ms": row["duration_ms"], + "schema_rev": row["schema_rev"], + "snapshot_age_seconds": round(age, 1), + "stale": age > self.interval_seconds * 3, + } + ) + return payload + + def read_series(self, conn: sqlite3.Connection, metric: str, days: int) -> dict[str, Any]: + """Points for one metric plus period-over-period totals.""" + if metric not in SERIES_METRICS: + raise ValueError(f"Unknown metric: {metric}") + days = max(1, min(int(days), self.history_days)) + expression = SERIES_METRICS[metric] + start = (datetime.now() - timedelta(days=days - 1)).strftime("%Y-%m-%d") + previous_start = (datetime.now() - timedelta(days=days * 2 - 1)).strftime("%Y-%m-%d") + + rows = conn.execute( + f"SELECT date, is_final, {expression} AS value, computed_at " + "FROM daily_rollup WHERE date >= ? ORDER BY date", + (start,), + ).fetchall() + points = [ + {"date": row["date"], "value": row["value"], "complete": bool(row["is_final"])} + for row in rows + ] + previous_rows = conn.execute( + f"SELECT {expression} AS value FROM daily_rollup WHERE date >= ? AND date < ?", + (previous_start, start), + ).fetchall() + + def aggregate(source) -> float | None: + values = [row["value"] for row in source if row["value"] is not None] + if not values: + return None + if metric in RATIO_METRICS: + return round(sum(values) / len(values), 1) + return sum(values) + + current_total = aggregate(rows) + previous_total = aggregate(previous_rows) + etag_row = conn.execute( + "SELECT MAX(computed_at) FROM daily_rollup WHERE date >= ?", (previous_start,) + ).fetchone() + + return { + "metric": metric, + "days": days, + "is_ratio": metric in RATIO_METRICS, + "points": points, + "current_period_total": current_total, + "previous_period_total": previous_total, + "change_pct": _change_pct(current_total, previous_total), + "etag_source": etag_row[0] if etag_row else None, + } + + def _window_cutoff(self, window: str) -> float | None: + """Epoch lower bound for a window token, or None for 'all'.""" + return { + "24h": time.time() - 86400, + "7d": time.time() - 7 * 86400, + "30d": time.time() - 30 * 86400, + "90d": time.time() - 90 * 86400, + }.get(window) + + def read_top( + self, conn: sqlite3.Connection, kind: str, window: str, limit: int + ) -> dict[str, Any]: + """One narrow leaderboard query, instead of the whole stats payload.""" + if kind not in TOP_KINDS: + raise ValueError(f"Unknown top kind: {kind}") + limit = max(1, min(int(limit), 50)) + cutoff = self._window_cutoff(window) + now = time.time() + retention = self.stats_retention_days + items: list[dict[str, Any]] = [] + + if kind == "repeaters": + retention = self.adverts_retention_days + days = {"24h": 1, "7d": 7, "30d": 30, "90d": 90}.get(window, retention) + since = (datetime.now() - timedelta(days=days - 1)).strftime("%Y-%m-%d") + if _table_exists(conn, "daily_stats") and _table_exists(conn, "complete_contact_tracking"): + items = [ + { + "name": row["name"] or row["public_key"][:12], + "public_key": row["public_key"], + "role": normalize_role(row["role"]), + "count": row["adverts"] or 0, + } + for row in conn.execute( + """ + SELECT ds.public_key, c.name, c.role, SUM(ds.advert_count) AS adverts + FROM daily_stats ds + JOIN complete_contact_tracking c ON c.public_key = ds.public_key + WHERE ds.date >= ? AND c.role IN ('repeater', 'roomserver') + GROUP BY ds.public_key ORDER BY adverts DESC LIMIT ? + """, + (since, limit), + ) + ] + elif kind == "users" and _table_exists(conn, "message_stats"): + items = [ + {"name": row[0], "count": row[1]} + for row in conn.execute( + """ + SELECT sender_id, COUNT(*) AS n FROM message_stats + WHERE timestamp <= ? AND timestamp >= ? + GROUP BY sender_id ORDER BY n DESC LIMIT ? + """, + (now, cutoff if cutoff is not None else 0, limit), + ) + ] + elif kind == "commands" and _table_exists(conn, "command_stats"): + items = [ + {"name": row[0], "count": row[1]} + for row in conn.execute( + """ + SELECT command_name, COUNT(*) AS n FROM command_stats + WHERE timestamp <= ? AND timestamp >= ? + GROUP BY command_name ORDER BY n DESC LIMIT ? + """, + (now, cutoff if cutoff is not None else 0, limit), + ) + ] + elif kind == "channels" and _table_exists(conn, "message_stats"): + items = [ + {"name": row[0], "count": row[1], "users": row[2]} + for row in conn.execute( + """ + SELECT channel, COUNT(*) AS n, COUNT(DISTINCT sender_id) AS users + FROM message_stats + WHERE channel IS NOT NULL AND channel != '' + AND timestamp <= ? AND timestamp >= ? + GROUP BY channel ORDER BY n DESC LIMIT ? + """, + (now, cutoff if cutoff is not None else 0, limit), + ) + ] + elif kind == "paths" and _table_exists(conn, "path_stats"): + items = [ + { + "name": row[0], + "path_length": row[1], + "path_string": row[2], + "timestamp": row[3], + } + for row in conn.execute( + """ + SELECT sender_id, path_length, path_string, timestamp FROM path_stats + WHERE timestamp <= ? AND timestamp >= ? + ORDER BY path_length DESC, timestamp DESC LIMIT ? + """, + (now, cutoff if cutoff is not None else 0, limit), + ) + ] + + requested_days = {"24h": 1, "7d": 7, "30d": 30, "90d": 90}.get(window) + return { + "kind": kind, + "window": window, + "window_label": self._window_label(window, retention), + "retention_days": retention, + "truncated_by_retention": bool( + requested_days is None or requested_days > retention + ), + "items": items, + } + + @staticmethod + def _window_label(window: str, retention_days: int) -> str: + return { + "24h": "Last 24 hours", + "7d": "Last 7 days", + "30d": "Last 30 days", + "90d": "Last 90 days", + }.get(window, f"All retained ({retention_days}d)") + + def derive_windows(self, conn: sqlite3.Connection | None = None) -> dict[str, Any]: + """Build the selector options from retention, so labels cannot overclaim. + + The old UI offered "30d" and "All" against tables pruned at 7 days: every + one of those options returned the same 7-day number under a wrong label. + """ + retention = { + "stats_days": self.stats_retention_days, + "packet_stream_days": self.packet_retention_days, + "daily_stats_days": self.adverts_retention_days, + "observed_paths_days": self.adverts_retention_days, + } + + def options(retention_days: int) -> list[dict[str, str]]: + candidates = [("24h", 1), ("7d", 7), ("30d", 30), ("90d", 90)] + out = [ + {"value": value, "label": self._window_label(value, retention_days)} + for value, days in candidates + if days <= retention_days or value == "24h" + ] + out.append({"value": "all", "label": f"All retained ({retention_days}d)"}) + return out + + stats_options = options(self.stats_retention_days) + return { + "retention": retention, + "windows": { + "users": stats_options, + "commands": stats_options, + "channels": stats_options, + "paths": stats_options, + "repeaters": options(self.adverts_retention_days), + }, + } diff --git a/modules/web_viewer/integration.py b/modules/web_viewer/integration.py index 8c13f40..98fcfb3 100644 --- a/modules/web_viewer/integration.py +++ b/modules/web_viewer/integration.py @@ -211,11 +211,21 @@ class BotIntegration: return resolve_path(raw, base_dir) return str(Path(self.bot.db_manager.db_path).resolve()) + # Denormalized packet dimensions written alongside the JSON blob. Order + # matters: it is the INSERT column order and the queue tuple order. + PACKET_DIM_COLUMNS = ("route_type_name", "payload_type_name", "path_len", "bytes_per_hop") + def _init_packet_stream_table(self): """Backward-compatible initializer (now handled by migrations). Kept for older call sites and tests that patch this method. Safe to call multiple times and safe to ignore failures. + + The table-exists path still runs the ALTERs: this process writes to + ``[Web_Viewer] db_path``, which in a split-DB install is a file the bot's + own MigrationRunner never touches. If the viewer has not started yet + that DB has an old three-column packet_stream, and skipping the ALTERs + would leave every insert failing on unknown columns. """ try: import sqlite3 @@ -226,6 +236,8 @@ class BotIntegration: "SELECT name FROM sqlite_master WHERE type='table' AND name='packet_stream'" ) if cur.fetchone() is not None: + self._add_missing_packet_dim_columns(conn) + conn.commit() return try: foreign_keys = self.bot.config.getboolean("Web_Viewer", "sqlite_foreign_keys", fallback=True) @@ -247,7 +259,11 @@ class BotIntegration: id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, data TEXT NOT NULL, - type TEXT NOT NULL + type TEXT NOT NULL, + route_type_name TEXT, + payload_type_name TEXT, + path_len INTEGER, + bytes_per_hop INTEGER ) """ ) @@ -257,10 +273,62 @@ class BotIntegration: conn.execute( "CREATE INDEX IF NOT EXISTS idx_packet_stream_type ON packet_stream(type)" ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_dims " + "ON packet_stream(timestamp, bytes_per_hop, route_type_name, payload_type_name) " + "WHERE type = 'packet'" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_undimensioned " + "ON packet_stream(type, id) WHERE route_type_name IS NULL" + ) conn.commit() except Exception: pass + def _add_missing_packet_dim_columns(self, conn) -> None: + """ALTER in any denormalized dimension column packet_stream is missing.""" + existing = {row[1] for row in conn.execute("PRAGMA table_info(packet_stream)")} + types = {"path_len": "INTEGER", "bytes_per_hop": "INTEGER"} + for column in self.PACKET_DIM_COLUMNS: + if column not in existing: + conn.execute( + f"ALTER TABLE packet_stream ADD COLUMN {column} {types.get(column, 'TEXT')}" + ) + if self.PACKET_DIM_COLUMNS[0] not in existing: + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_dims " + "ON packet_stream(timestamp, bytes_per_hop, route_type_name, payload_type_name) " + "WHERE type = 'packet'" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_undimensioned " + "ON packet_stream(type, id) WHERE route_type_name IS NULL" + ) + + def _packet_stream_has_dim_columns(self, conn) -> bool: + """Probe once per process whether the target DB carries the dimension columns. + + The viewer DB may be a separate file this process never migrates, so the + writer cannot assume the wide schema — it falls back to the three-column + INSERT rather than failing every flush. + """ + cached = getattr(self, '_packet_dims_supported', None) + if cached is not None: + return cached + try: + existing = {row[1] for row in conn.execute("PRAGMA table_info(packet_stream)")} + supported = all(c in existing for c in self.PACKET_DIM_COLUMNS) + except Exception: + supported = False + self._packet_dims_supported = supported + if not supported: + self.bot.logger.info( + "packet_stream lacks denormalized dimension columns; " + "writing legacy rows (dashboard packet tiles will backfill from JSON)" + ) + return supported + def _start_drain_thread(self) -> None: """Start the background thread that flushes the write queue every DRAIN_INTERVAL seconds.""" self._drain_thread = threading.Thread( @@ -276,7 +344,7 @@ class BotIntegration: self._drain_stop.wait(timeout=self.DRAIN_INTERVAL) self._flush_write_queue() - def _requeue_rows(self, rows: list[tuple[float, str, str]]) -> None: + def _requeue_rows(self, rows: list[tuple]) -> None: """Restore rows to the queue after a failed flush (FIFO). Logs if the queue stays full.""" for i, row in enumerate(rows): try: @@ -296,7 +364,7 @@ class BotIntegration: with self._flush_lock: if self._write_queue.empty(): return - rows: list[tuple[float, str, str]] = [] + rows: list[tuple] = [] while not self._write_queue.empty(): try: rows.append(self._write_queue.get_nowait()) @@ -309,10 +377,18 @@ class BotIntegration: for attempt in range(max_retries): try: with closing(sqlite3.connect(str(db_path), timeout=self.sqlite_connect_timeout_sec)) as conn: - conn.executemany( - 'INSERT INTO packet_stream (timestamp, data, type) VALUES (?, ?, ?)', - rows, - ) + if self._packet_stream_has_dim_columns(conn): + conn.executemany( + 'INSERT INTO packet_stream ' + '(timestamp, data, type, route_type_name, payload_type_name, ' + 'path_len, bytes_per_hop) VALUES (?, ?, ?, ?, ?, ?, ?)', + rows, + ) + else: + conn.executemany( + 'INSERT INTO packet_stream (timestamp, data, type) VALUES (?, ?, ?)', + [row[:3] for row in rows], + ) conn.commit() return except sqlite3.OperationalError as e: @@ -331,9 +407,20 @@ class BotIntegration: self._requeue_rows(rows) return - def _insert_packet_stream_row(self, data_json: str, row_type: str, log_prefix: str = "packet data"): - """Queue one row for batched insertion into packet_stream by the drain thread.""" - item = (time.time(), data_json, row_type) + def _insert_packet_stream_row( + self, + data_json: str, + row_type: str, + log_prefix: str = "packet data", + dims: tuple | None = None, + ): + """Queue one row for batched insertion into packet_stream by the drain thread. + + ``dims`` carries the denormalized (route_type_name, payload_type_name, + path_len, bytes_per_hop) values for packet rows; command/message/other + row types pass None and store NULLs. + """ + item = (time.time(), data_json, row_type, *(dims or (None, None, None, None))) try: self._write_queue.put(item, timeout=self._WRITE_QUEUE_PUT_TIMEOUT_SEC) except queue.Full: @@ -382,11 +469,50 @@ class BotIntegration: serializable_data = self._make_json_serializable(packet_data) # Store in database for web viewer to read (retries on database is locked) - self._insert_packet_stream_row(json.dumps(serializable_data), 'packet', "packet data") + self._insert_packet_stream_row( + json.dumps(serializable_data), + 'packet', + "packet data", + dims=self._packet_dims_from_data(serializable_data), + ) except Exception as e: self.bot.logger.warning(f"Error storing packet data for web viewer: {e}") + @staticmethod + def _packet_dims_from_data(packet_data: dict) -> tuple: + """Pull the denormalized dimension values out of an already-parsed packet. + + Returns (route_type_name, payload_type_name, path_len, bytes_per_hop), + with None for anything absent or non-coercible — the dashboard reads + these as "unknown" rather than mis-bucketing the packet. + """ + def _text(key: str) -> str | None: + value = packet_data.get(key) + if value is None or isinstance(value, (dict, list)): + return None + text = str(value).strip() + return text[:32] or None + + def _number(key: str) -> int | None: + value = packet_data.get(key) + if value is None or isinstance(value, bool): + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + return ( + # Never NULL: route_type_name is the dashboard's "this row has been + # dimensioned" marker, and a NULL would put the row back in the + # refresher's backfill queue on every tick. + _text('route_type_name') or 'UNKNOWN_ROUTE', + _text('payload_type_name'), + _number('path_len'), + _number('bytes_per_hop'), + ) + def capture_command(self, message, command_name, response, success, command_id=None): """Capture command data and store in database for web viewer""" try: diff --git a/modules/web_viewer/static/css/dashboard.css b/modules/web_viewer/static/css/dashboard.css new file mode 100644 index 0000000..594fcaa --- /dev/null +++ b/modules/web_viewer/static/css/dashboard.css @@ -0,0 +1,394 @@ +/* Dashboard (landing page) styles. + * + * Only the theme tokens declared in base.html are used (--bg-*, --text-*, + * --border-color, --card-*, --shadow-color), and every rule that hard-codes a + * colour has a matching [data-theme="dark"] override — Bootstrap 5.1.3 has no + * native dark theme here, so nothing inherits one for free. + */ + +/* ── Health strip ──────────────────────────────────────────────────────── */ + +.health-strip { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 0.25rem 1.5rem; + min-height: 56px; + padding: 0.5rem 1rem; + margin-bottom: 1.25rem; + background-color: var(--card-bg); + border: 1px solid var(--border-color); + border-radius: 10px; + box-shadow: 0 1px 3px var(--shadow-color); +} + +.health-strip__item { + display: flex; + align-items: center; + gap: 0.45rem; + font-size: 0.875rem; + color: var(--text-color); + white-space: nowrap; +} + +.health-strip__label { + color: var(--text-muted); + text-transform: uppercase; + font-size: 0.7rem; + letter-spacing: 0.04em; +} + +.health-strip__spacer { + flex: 1 1 auto; +} + +.health-strip .btn-refresh { + padding: 0.1rem 0.4rem; + line-height: 1; + border: none; + background: transparent; + color: var(--text-muted); + cursor: pointer; +} + +.health-strip .btn-refresh:hover, +.health-strip .btn-refresh:focus { + color: #0d6efd; +} + +.health-strip .btn-refresh[disabled] { + opacity: 0.5; + cursor: progress; +} + +/* ── Section headings ──────────────────────────────────────────────────── */ + +.dashboard-section-title { + display: flex; + align-items: baseline; + gap: 0.6rem; + margin: 1.75rem 0 0.75rem; + font-size: 0.8rem; + font-weight: 700; + text-transform: uppercase; + letter-spacing: 0.08em; + color: var(--text-muted); +} + +.dashboard-section-title::after { + content: ""; + flex: 1 1 auto; + height: 1px; + background-color: var(--border-color); +} + +/* ── Stat tiles ────────────────────────────────────────────────────────── */ + +.stat-tile { + display: flex; + flex-direction: column; + height: 100%; + padding: 0.75rem 0.9rem 0.5rem; + background-color: var(--card-bg); + border: 1px solid var(--border-color); + border-radius: 10px; + box-shadow: 0 1px 3px var(--shadow-color); +} + +.stat-tile__label { + font-size: 0.7rem; + font-weight: 600; + text-transform: uppercase; + letter-spacing: 0.04em; + color: var(--text-muted); + margin-bottom: 0.2rem; +} + +.stat-tile__row { + display: flex; + align-items: baseline; + gap: 0.5rem; + flex-wrap: wrap; +} + +.stat-tile__value { + font-size: 1.6rem; + font-weight: 700; + line-height: 1.1; + color: var(--text-color); + font-variant-numeric: tabular-nums; +} + +.stat-tile__value--muted { + color: var(--text-muted); +} + +.stat-tile__sub { + font-size: 0.75rem; + color: var(--text-muted); +} + +.stat-tile__spark { + position: relative; + height: 48px; + margin-top: auto; +} + +/* Sparkline canvases are sized by their container, not by their attributes. */ +.stat-tile__spark canvas { + width: 100% !important; + height: 48px !important; +} + +/* ── Delta chips ───────────────────────────────────────────────────────── */ + +.delta-chip { + display: inline-flex; + align-items: center; + gap: 0.15rem; + padding: 0.05rem 0.35rem; + border-radius: 999px; + font-size: 0.72rem; + font-weight: 600; + font-variant-numeric: tabular-nums; + background-color: var(--bg-tertiary); + color: var(--text-muted); +} + +.delta-chip--up { + background-color: rgba(25, 135, 84, 0.12); + color: #146c43; +} + +.delta-chip--down { + background-color: rgba(220, 53, 69, 0.12); + color: #b02a37; +} + +[data-theme="dark"] .delta-chip--up { + background-color: rgba(25, 135, 84, 0.22); + color: #75b798; +} + +[data-theme="dark"] .delta-chip--down { + background-color: rgba(220, 53, 69, 0.22); + color: #ea868f; +} + +/* ── Mini bar (tracked vs known) ───────────────────────────────────────── */ + +.mini-bar { + height: 6px; + width: 100%; + margin-top: 0.4rem; + border-radius: 3px; + background-color: var(--bg-tertiary); + overflow: hidden; +} + +.mini-bar__fill { + height: 100%; + background-color: #0d6efd; + border-radius: 3px; +} + +/* ── Stacked route-mix bar ─────────────────────────────────────────────── */ + +.stacked-bar { + display: flex; + height: 26px; + width: 100%; + border-radius: 6px; + overflow: hidden; + background-color: var(--bg-tertiary); +} + +.stacked-bar__segment { + display: flex; + align-items: center; + justify-content: center; + font-size: 0.72rem; + font-weight: 600; + color: #fff; + white-space: nowrap; + overflow: hidden; + min-width: 0; +} + +.stacked-bar__legend { + display: flex; + flex-wrap: wrap; + gap: 0.25rem 1rem; + margin-top: 0.5rem; + font-size: 0.78rem; + color: var(--text-muted); +} + +.stacked-bar__swatch { + display: inline-block; + width: 10px; + height: 10px; + border-radius: 2px; + margin-right: 0.35rem; + vertical-align: baseline; +} + +/* ── Horizontal mix rows (role / device) ───────────────────────────────── */ + +.mix-row { + display: grid; + grid-template-columns: minmax(5.5rem, 8rem) 1fr auto; + align-items: center; + gap: 0.6rem; + margin-bottom: 0.35rem; + font-size: 0.82rem; +} + +.mix-row__name { + color: var(--text-color); + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.mix-row__track { + height: 10px; + border-radius: 5px; + background-color: var(--bg-tertiary); + overflow: hidden; +} + +.mix-row__fill { + height: 100%; + border-radius: 5px; +} + +.mix-row__count { + color: var(--text-muted); + font-variant-numeric: tabular-nums; +} + +/* ── Signal percentile readout ─────────────────────────────────────────── */ + +.signal-grid { + display: grid; + grid-template-columns: repeat(3, 1fr); + gap: 0.5rem; + text-align: center; +} + +.signal-grid__value { + font-size: 1.1rem; + font-weight: 700; + color: var(--text-color); + font-variant-numeric: tabular-nums; +} + +.signal-grid__label { + font-size: 0.68rem; + text-transform: uppercase; + letter-spacing: 0.04em; + color: var(--text-muted); +} + +/* ── Leaderboard rows ──────────────────────────────────────────────────── */ + +.top-list-row { + display: flex; + justify-content: space-between; + align-items: center; + gap: 0.5rem; + padding: 0.25rem 0; +} + +.top-list-row + .top-list-row { + border-top: 1px solid var(--border-color); +} + +.top-list-row__name { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +/* ── Chart wrappers ────────────────────────────────────────────────────── */ + +.chart-box { + position: relative; + width: 100%; + height: 200px; +} + +.chart-box--tall { + height: 240px; +} + +.doughnut-box { + position: relative; + max-width: 190px; + width: 100%; + margin: 0 auto; +} + +.dashboard-note { + font-size: 0.75rem; + color: var(--text-muted); +} + +.dashboard-empty { + display: flex; + align-items: center; + justify-content: center; + height: 100%; + min-height: 100px; + color: var(--text-muted); + font-size: 0.85rem; + font-style: italic; +} + +/* Info affordance shared by tooltip triggers in card headers. */ +.dashboard-info-btn { + padding: 0; + border: 0; + background: none; + color: var(--text-muted); + line-height: 1; +} + +.dashboard-info-btn:hover, +.dashboard-info-btn:focus { + color: #0dcaf0; + text-decoration: none; +} + +.tooltip.dashboard-hint .tooltip-inner { + max-width: min(22rem, 92vw); + text-align: left; +} + +/* ── Live activity feed ────────────────────────────────────────────────── */ + +.live-feed { + height: 260px; + overflow-y: auto; + background-color: var(--bg-secondary); + padding: 0.5rem; +} + +.live-feed__entry { + padding: 3px 8px; + margin-bottom: 3px; + font-size: 0.82rem; + border-radius: 2px; + border-left: 3px solid var(--border-color); +} + +@media (max-width: 575.98px) { + .stat-tile__value { + font-size: 1.35rem; + } + + .health-strip { + gap: 0.25rem 1rem; + } +} diff --git a/modules/web_viewer/static/js/dashboard.js b/modules/web_viewer/static/js/dashboard.js new file mode 100644 index 0000000..71f6243 --- /dev/null +++ b/modules/web_viewer/static/js/dashboard.js @@ -0,0 +1,1048 @@ +/* MeshCore Bot — landing page dashboard. + * + * Served as a static file so `script-src 'self'` covers it and it needs no CSP + * nonce. Server-provided configuration arrives in a + * - - - - +{# Data, not code: a JSON block needs no CSP nonce and cannot execute. #} + + {% endblock %} diff --git a/modules/web_viewer/templates/stats.html b/modules/web_viewer/templates/stats.html deleted file mode 100644 index 6523892..0000000 --- a/modules/web_viewer/templates/stats.html +++ /dev/null @@ -1,355 +0,0 @@ -{% extends "base.html" %} - -{% block title %}Statistics - MeshCore Bot Data Viewer{% endblock %} - -{% block content %} -
-
-

- Statistics -

-
-
- - -
-
-
-
- Total Contacts -
-
-

0

- All time -
-
-
- -
-
-
- Total Adverts -
-
-

0

- All time -
-
-
- -
-
-
- Recent Activity -
-
-

0

- Last 24 hours -
-
-
- -
-
-
- Growth Rate -
-
-

0%

- Daily growth -
-
-
-
- - -
-
-
-
- Activity Distribution -
-
-
-
Loading activity chart...
-
-
-
-
- -
-
-
- Usage Trends -
-
- -
-
-
-
- - -
-
-
-
- Detailed Statistics -
-
-
-
-
Contact Statistics
-
    -
  • Loading contact statistics...
    • -
-
-
-
Advert Statistics
- -
-
-
-
-
-
- - -
-
-
-
- System Performance -
-
-
-
Loading system performance...
-
-
-
-
- -
-
-
- Configuration -
-
-
-
Loading system configuration...
-
-
-
-
-
-{% endblock %} - -{% block extra_js %} - -{% endblock %} diff --git a/package.json b/package.json index bda27ba..51771bf 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "description": "Frontend linting for meshcore-bot web viewer templates", "scripts": { "lint:html": "htmlhint \"modules/web_viewer/templates/**/*.html\"", - "lint:js": "eslint \"modules/web_viewer/templates/**/*.html\"", + "lint:js": "eslint \"modules/web_viewer/templates/**/*.html\" modules/web_viewer/static/js/dashboard.js modules/web_viewer/static/js/channel_operations.js", "lint:frontend": "npm run lint:html && npm run lint:js", "test:contacts": "node --test tests/js/contacts_manager.test.mjs" }, diff --git a/tests/test_dashboard_stats.py b/tests/test_dashboard_stats.py new file mode 100644 index 0000000..edf018f --- /dev/null +++ b/tests/test_dashboard_stats.py @@ -0,0 +1,717 @@ +#!/usr/bin/env python3 +"""Tests for the web viewer dashboard: rollup service, snapshot API, and the +performance floor of the legacy /api/stats payload. + +The perf test seeds a synthetic database roughly the shape of a live install +(~100k rows across the tables the dashboard reads) so the request-path cost is +regression-locked rather than eyeballed. +""" + +import json +import logging +import sqlite3 +import time +from configparser import ConfigParser +from contextlib import closing +from datetime import UTC, datetime, timedelta +from unittest.mock import patch + +import pytest + +from modules.web_viewer.app import BotDataViewer +from modules.web_viewer.dashboard_stats import ( + LEASE_KEY, + TOP_KINDS, + day_bounds, + local_date_str, + normalize_role, +) + +# --------------------------------------------------------------------------- +# Synthetic database +# --------------------------------------------------------------------------- + +CONTACTS = 2000 +OBSERVED_PATHS = 20000 +PACKETS = 40000 +MESSAGES = 20000 +COMMANDS = 5000 +PATHS = 5000 +DAILY_STATS_DAYS = 90 +DAILY_STATS_NODES = 200 + +# Only the first ADVERTISING_CONTACTS contacts appear in observed_paths, so the +# rest fall through the cheap public-key check into the hop-prefix matcher — +# the quadratic step this suite exists to keep out. +ADVERTISING_CONTACTS = 500 + + +def _pk(i: int) -> str: + """Deterministic 64-hex-char public key for row *i*.""" + return f"{i:064x}" + + +def _create_stats_tables(conn: sqlite3.Connection) -> None: + """Create the three tables stats_command owns (migrations do not).""" + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS message_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL, + sender_id TEXT NOT NULL, + channel TEXT, + content TEXT NOT NULL, + is_dm BOOLEAN NOT NULL, + hops INTEGER, + snr REAL, + rssi INTEGER, + path TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE IF NOT EXISTS command_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL, + sender_id TEXT NOT NULL, + command_name TEXT NOT NULL, + channel TEXT, + is_dm BOOLEAN NOT NULL, + response_sent BOOLEAN NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE IF NOT EXISTS path_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL, + sender_id TEXT NOT NULL, + channel TEXT, + path_length INTEGER NOT NULL, + path_string TEXT NOT NULL, + hops INTEGER, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE INDEX IF NOT EXISTS idx_message_timestamp ON message_stats(timestamp); + CREATE INDEX IF NOT EXISTS idx_command_timestamp ON command_stats(timestamp); + """ + ) + + +def seed_database(db_path: str, now: float | None = None) -> None: + """Populate *db_path* with a synthetic mesh roughly the shape of a live install.""" + now = time.time() if now is None else now + roles = ["repeater", "roomserver", "companion", "sensor", "type3"] + devices = ["Repeater", "Companion", "RoomServer", "Sensor"] + + with sqlite3.connect(db_path, timeout=60) as conn: + _create_stats_tables(conn) + + conn.executemany( + """ + INSERT INTO complete_contact_tracking + (public_key, name, role, device_type, first_heard, last_heard, + advert_count, snr, signal_strength, hop_count, is_currently_tracked, + out_bytes_per_hop, city, state, country) + VALUES (?, ?, ?, ?, datetime('now','localtime','-40 days'), + datetime('now','localtime', ?), ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + _pk(i), + f"node-{i}", + roles[i % len(roles)], + devices[i % len(devices)], + f"-{i % 10} days", + 1 + (i % 50), + -10.0 + (i % 30), + -120.0 + (i % 60), + i % 6, + 1 if i % 3 == 0 else 0, + 3 if i % 4 == 0 else 1, + f"city-{i % 90}", + f"state-{i % 14}", + "United States" if i % 4 else "Canada", + ) + for i in range(CONTACTS) + ], + ) + + conn.executemany( + """ + INSERT INTO observed_paths + (public_key, from_prefix, to_prefix, path_hex, path_length, + bytes_per_hop, packet_type, first_seen, last_seen, observation_count) + VALUES (?, ?, ?, ?, ?, ?, 'advert', + datetime('now','localtime','-20 days'), + datetime('now','localtime', ?), ?) + """, + [ + ( + _pk(i % ADVERTISING_CONTACTS), + f"{i % 65536:04x}", + f"{(i * 7) % 65536:04x}", + f"{i % 65536:04x}{(i * 7) % 65536:04x}", + 2, + 2 if i % 5 else 1, + f"-{i % 12} days", + 1 + (i % 9), + ) + for i in range(OBSERVED_PATHS) + ], + ) + + conn.executemany( + "INSERT INTO packet_stream (timestamp, data, type) VALUES (?, ?, 'packet')", + [ + ( + now - (i % (3 * 86400)), + json.dumps( + { + "bytes_per_hop": 2 if i % 4 else 1, + "route_type_name": "FLOOD" if i % 3 else "DIRECT", + "payload_type_name": "ADVERT" if i % 2 else "TXT_MSG", + "path_len": i % 5, + "payload_hex": "ab" * 40, + } + ), + ) + for i in range(PACKETS) + ], + ) + + conn.executemany( + """ + INSERT INTO message_stats + (timestamp, sender_id, channel, content, is_dm, hops, snr, rssi) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + int(now - (i % (7 * 86400))), + f"user-{i % 120}", + None if i % 5 == 0 else f"channel-{i % 4}", + f"message {i}", + 1 if i % 5 == 0 else 0, + i % 5, + -5.0 + (i % 20), + -110 + (i % 50), + ) + for i in range(MESSAGES) + ], + ) + + conn.executemany( + """ + INSERT INTO command_stats + (timestamp, sender_id, command_name, channel, is_dm, response_sent) + VALUES (?, ?, ?, ?, ?, ?) + """, + [ + ( + int(now - (i % (7 * 86400))), + f"user-{i % 60}", + f"cmd{i % 12}", + f"channel-{i % 4}", + i % 2, + 0 if i % 20 == 0 else 1, + ) + for i in range(COMMANDS) + ], + ) + + conn.executemany( + """ + INSERT INTO path_stats + (timestamp, sender_id, channel, path_length, path_string, hops) + VALUES (?, ?, ?, ?, ?, ?) + """, + [ + ( + int(now - (i % (7 * 86400))), + f"user-{i % 60}", + f"channel-{i % 4}", + i % 8, + ",".join(f"{(i + h) % 256:02x}" for h in range(i % 8)), + i % 8, + ) + for i in range(PATHS) + ], + ) + + conn.executemany( + """ + INSERT INTO daily_stats (date, public_key, advert_count) + VALUES (date('now','localtime', ?), ?, ?) + """, + [ + (f"-{day} days", _pk(node), 1 + ((day * node) % 20)) + for day in range(DAILY_STATS_DAYS) + for node in range(DAILY_STATS_NODES) + ], + ) + conn.commit() + + +def _fake_setup_logging(self: BotDataViewer) -> None: + self.logger = logging.getLogger("test_dashboard_stats") + self.logger.setLevel(logging.DEBUG) + if not self.logger.handlers: + self.logger.addHandler(logging.NullHandler()) + self.logger.propagate = False + + +def build_viewer(tmp_path, extra_web_viewer: dict[str, str] | None = None) -> BotDataViewer: + """Create a BotDataViewer over a fresh temp DB with all background threads off.""" + db_path = str(tmp_path / "meshcore_bot.db") + config_path = str(tmp_path / "config.ini") + + config = ConfigParser() + config.add_section("Bot") + config.set("Bot", "db_path", db_path) + config.add_section("Web_Viewer") + config.set("Web_Viewer", "enabled", "false") + config.set("Web_Viewer", "web_viewer_password", "") + for key, value in (extra_web_viewer or {}).items(): + config.set("Web_Viewer", key, value) + with open(config_path, "w") as handle: + config.write(handle) + + with ( + patch.object(BotDataViewer, "_setup_logging", _fake_setup_logging), + patch.object(BotDataViewer, "_start_database_polling", lambda self: None), + patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), + patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), + patch.object(BotDataViewer, "_setup_socketio_handlers", lambda self: None), + patch("modules.web_viewer.app.RepeaterManager"), + ): + viewer = BotDataViewer(db_path=db_path, config_path=config_path) + + viewer.app.config["TESTING"] = True + return viewer + + +@pytest.fixture(scope="module") +def seeded_viewer(tmp_path_factory): + """Module-scoped viewer over a ~100k-row synthetic database.""" + tmp_path = tmp_path_factory.mktemp("dashboard_perf") + viewer = build_viewer(tmp_path) + seed_database(viewer.db_path) + with sqlite3.connect(viewer.db_path, timeout=60) as conn: + conn.execute("ANALYZE") + yield viewer + + +# --------------------------------------------------------------------------- +# Performance floor +# --------------------------------------------------------------------------- + + +class TestStatsPerformance: + """Regression lock on the request-path cost of the dashboard payload. + + Thresholds are deliberately loose relative to measured times (~10x headroom) + so they fail on an algorithmic regression, not on a slow CI runner. + """ + + def test_api_stats_under_threshold(self, seeded_viewer): + """~35ms after the rework, against ~264ms for the linear hop-prefix scan.""" + with seeded_viewer.app.test_client() as client: + client.get("/api/stats") # warm the page cache + started = time.perf_counter() + response = client.get("/api/stats") + elapsed = time.perf_counter() - started + + assert response.status_code == 200 + payload = response.get_json() + assert "error" not in payload + assert payload["total_contacts"] == CONTACTS + assert elapsed < 0.2, f"/api/stats took {elapsed:.2f}s on a synthetic 100k-row DB" + + def test_dashboard_summary_is_a_single_row_read(self, seeded_viewer): + with closing(seeded_viewer._dashboard_connection()) as conn: + seeded_viewer.dashboard_stats.refresh(conn) + with seeded_viewer.app.test_client() as client: + client.get("/api/dashboard/summary") + started = time.perf_counter() + response = client.get("/api/dashboard/summary") + elapsed = time.perf_counter() - started + + assert response.status_code == 200 + assert elapsed < 0.05, f"/api/dashboard/summary took {elapsed * 1000:.0f}ms" + + +# --------------------------------------------------------------------------- +# Rollup correctness +# --------------------------------------------------------------------------- + + +@pytest.fixture +def viewer(tmp_path): + """Function-scoped viewer over an empty migrated database.""" + return build_viewer(tmp_path) + + +def _refresh(viewer, **kwargs) -> dict: + with closing(viewer._dashboard_connection()) as conn: + return viewer.dashboard_stats.refresh(conn, **kwargs) + + +def _rollup(viewer, date_str: str) -> dict | None: + with closing(viewer._dashboard_connection()) as conn: + row = conn.execute("SELECT * FROM daily_rollup WHERE date = ?", (date_str,)).fetchone() + return dict(row) if row else None + + +class TestRollupCorrectness: + + def test_backfilled_days_are_null_not_zero(self, viewer): + """A pruned source must read as a gap; zero would look like an outage.""" + seed_database(viewer.db_path) + _refresh(viewer) + + old_date = (datetime.now() - timedelta(days=60)).strftime("%Y-%m-%d") + row = _rollup(viewer, old_date) + assert row is not None + assert row["is_backfilled"] == 1 + # daily_stats survives 90 days, so adverts are real... + assert row["adverts_total"] > 0 + # ...while the 7-day and 3-day sources are unrecoverable, not empty. + for column in ("messages_total", "commands_total", "path_obs_total", "packets_total"): + assert row[column] is None, f"{column} should be NULL for a backfilled day" + + def test_future_timestamps_do_not_create_rollup_rows(self, viewer): + """Rows dated 2103 exist in the wild; an unclamped bound charts them.""" + seed_database(viewer.db_path) + with sqlite3.connect(viewer.db_path) as conn: + conn.execute( + """ + INSERT INTO message_stats (timestamp, sender_id, channel, content, is_dm) + VALUES (?, 'ghost', 'general', 'from the future', 0) + """, + (int(datetime(2103, 8, 15).timestamp()),), + ) + _refresh(viewer) + + with closing(viewer._dashboard_connection()) as conn: + latest = conn.execute("SELECT MAX(date) FROM daily_rollup").fetchone()[0] + assert latest <= local_date_str() + + def test_today_excludes_future_rows_from_its_count(self, viewer): + seed_database(viewer.db_path) + today = local_date_str() + before = _refresh(viewer) and _rollup(viewer, today)["messages_total"] + + with sqlite3.connect(viewer.db_path) as conn: + conn.execute( + """ + INSERT INTO message_stats (timestamp, sender_id, channel, content, is_dm) + VALUES (?, 'ghost', 'general', 'later today but not yet', 0) + """, + (int(time.time()) + 3600,), + ) + _refresh(viewer) + assert _rollup(viewer, today)["messages_total"] == before + + def test_refresh_is_idempotent(self, viewer): + seed_database(viewer.db_path) + # Clear the packet backfill backlog first: while it drains, successive + # refreshes legitimately see more dimensioned rows than the last. + viewer.dashboard_stats.packet_backfill_rows = PACKETS + _refresh(viewer) + first = _rollup(viewer, local_date_str()) + _refresh(viewer) + second = _rollup(viewer, local_date_str()) + for column, value in first.items(): + if column == "computed_at": + continue + assert second[column] == value, column + + def test_signal_stored_as_sums_and_counts(self, viewer): + """Means cannot be re-aggregated across a window; sums and counts can.""" + seed_database(viewer.db_path) + _refresh(viewer) + row = _rollup(viewer, local_date_str()) + assert row["snr_count"] == MESSAGES + assert row["rssi_count"] == MESSAGES + with sqlite3.connect(viewer.db_path) as conn: + expected = conn.execute("SELECT SUM(snr) FROM message_stats").fetchone()[0] + assert row["snr_sum"] == pytest.approx(expected) + + def test_missing_stats_tables_degrade_to_null(self, viewer): + """`collect_stats = false` leaves the tables absent entirely.""" + seed_database(viewer.db_path) + with sqlite3.connect(viewer.db_path) as conn: + for table in ("message_stats", "command_stats", "path_stats"): + conn.execute(f"DROP TABLE {table}") + _refresh(viewer) + + row = _rollup(viewer, local_date_str()) + assert row["messages_total"] is None + assert row["commands_total"] is None + assert row["adverts_total"] > 0 # unaffected sources still populate + + with viewer.app.test_client() as client: + payload = client.get("/api/dashboard/summary").get_json() + assert "message_stats" not in payload["coverage"]["sources_present"] + assert payload["bot"] == {} + + def test_local_date_attribution_under_non_utc_timezone(self, viewer, monkeypatch): + """The rollup keys on local dates while message_stats stores UTC epochs. + + Get this wrong and the message series is offset from the advert series + by several hours of traffic — silently, and for months. + """ + monkeypatch.setenv("TZ", "America/Los_Angeles") + time.tzset() + try: + # 03:00 UTC is 20:00 the previous day in Los Angeles. + utc_moment = datetime(2026, 7, 29, 3, 0, tzinfo=UTC) + local_date = utc_moment.astimezone().strftime("%Y-%m-%d") + assert local_date == "2026-07-28", "fixture assumes a UTC-7/8 offset" + + start, end = day_bounds(local_date) + assert start <= utc_moment.timestamp() < end + + with sqlite3.connect(viewer.db_path) as conn: + _create_stats_tables(conn) + conn.execute( + """ + INSERT INTO message_stats (timestamp, sender_id, channel, content, is_dm) + VALUES (?, 'tz-probe', 'general', 'evening message', 0) + """, + (int(utc_moment.timestamp()),), + ) + + with closing(viewer._dashboard_connection()) as conn: + computed = viewer.dashboard_stats.compute_day( + conn, + local_date, + sources=viewer.dashboard_stats.detect_sources(conn), + today=local_date, + first_seen={}, + multibyte_keys=None, + ) + assert computed["values"]["messages_total"] == 1 + finally: + monkeypatch.undo() + time.tzset() + + +class TestPacketDimensions: + + def test_column_counts_match_json_extract(self, viewer): + """The denormalized columns must agree with the JSON they replace.""" + seed_database(viewer.db_path) + viewer.dashboard_stats.packet_backfill_rows = PACKETS + _refresh(viewer) + + with sqlite3.connect(viewer.db_path) as conn: + from_column = conn.execute( + "SELECT COUNT(*) FROM packet_stream " + "WHERE type='packet' AND bytes_per_hop IN (2, 3)" + ).fetchone()[0] + from_json = conn.execute( + "SELECT COUNT(*) FROM packet_stream WHERE type='packet' " + "AND CAST(json_extract(data, '$.bytes_per_hop') AS INTEGER) IN (2, 3)" + ).fetchone()[0] + flood_column = conn.execute( + "SELECT COUNT(*) FROM packet_stream " + "WHERE type='packet' AND route_type_name LIKE '%FLOOD'" + ).fetchone()[0] + flood_json = conn.execute( + "SELECT COUNT(*) FROM packet_stream WHERE type='packet' " + "AND json_extract(data, '$.route_type_name') LIKE '%FLOOD'" + ).fetchone()[0] + + assert from_column == from_json + assert flood_column == flood_json + + def test_backfill_advances_and_terminates(self, viewer): + """Rows whose JSON has no route type must still be marked processed.""" + with sqlite3.connect(viewer.db_path) as conn: + conn.executemany( + "INSERT INTO packet_stream (timestamp, data, type) VALUES (?, ?, 'packet')", + [(time.time(), json.dumps({"hops": 1}), ) for _ in range(10)], + ) + viewer.dashboard_stats.packet_backfill_rows = 100 + with closing(viewer._dashboard_connection()) as conn: + first = viewer.dashboard_stats.backfill_packet_dims(conn) + second = viewer.dashboard_stats.backfill_packet_dims(conn) + assert first == 10 + assert second == 0, "rows without a route type were reselected forever" + + def test_backlog_probe_uses_the_worklist_index(self, viewer): + """The 'anything left to dimension?' probe must not scan the table. + + Finding nothing still means reading everything without this index, and + that cost is paid on every tick *after* the backfill completes — 4.6s + per minute forever on a 180MB packet_stream. + """ + with sqlite3.connect(viewer.db_path) as conn: + plan = conn.execute( + "EXPLAIN QUERY PLAN " + "SELECT id FROM packet_stream " + "WHERE type = 'packet' AND route_type_name IS NULL " + "ORDER BY id DESC LIMIT 2000" + ).fetchall() + assert any("idx_packet_stream_undimensioned" in str(step) for step in plan), plan + + def test_coverage_reports_the_measured_window(self, viewer): + """The old page labelled a 3-day table 'last 7 days'.""" + now = time.time() + with sqlite3.connect(viewer.db_path) as conn: + conn.executemany( + "INSERT INTO packet_stream (timestamp, data, type, route_type_name) " + "VALUES (?, '{}', 'packet', 'FLOOD')", + [(now - 2.25 * 86400,), (now - 60,)], + ) + with closing(viewer._dashboard_connection()) as conn: + coverage = viewer.dashboard_stats.packet_coverage(conn) + assert coverage["packets_window_label"] == "last 2d 6h" + assert coverage["packets_with_dims"] == 2 + + +class TestDerivedWindows: + + @pytest.mark.parametrize( + ("retention", "expected"), + [ + (7, ["24h", "7d", "all"]), + (30, ["24h", "7d", "30d", "all"]), + (1, ["24h", "all"]), + ], + ) + def test_options_never_outrun_retention(self, tmp_path, retention, expected): + viewer = build_viewer(tmp_path) + viewer.dashboard_stats.stats_retention_days = retention + windows = viewer.dashboard_stats.derive_windows() + assert [o["value"] for o in windows["windows"]["users"]] == expected + assert windows["windows"]["users"][-1]["label"] == f"All retained ({retention}d)" + + +class TestRoleBucketing: + + @pytest.mark.parametrize( + ("raw", "expected"), + [ + ("repeater", "repeater"), + ("RoomServer", "roomserver"), + ("type0", "Unknown"), + ("type15", "Unknown"), + ("", "Unknown"), + (None, "Unknown"), + ("typewriter", "typewriter"), + ], + ) + def test_unmapped_ordinals_fold_into_unknown(self, raw, expected): + assert normalize_role(raw) == expected + + +class TestSnapshotLease: + + def test_second_holder_is_refused_until_expiry(self, viewer): + with closing(viewer._dashboard_connection()) as conn: + assert viewer.dashboard_stats.try_claim_lease(conn) is True + # Same process reclaims its own lease; a different owner does not. + assert viewer.dashboard_stats.try_claim_lease(conn) is True + conn.execute( + "UPDATE bot_metadata SET value = ? WHERE key = ?", + (f"999999:other-host:{time.time() + 600:.0f}", LEASE_KEY), + ) + assert viewer.dashboard_stats.try_claim_lease(conn) is False + + def test_expired_lease_is_reclaimed(self, viewer): + with closing(viewer._dashboard_connection()) as conn: + viewer.dashboard_stats.try_claim_lease(conn) + conn.execute( + "UPDATE bot_metadata SET value = ? WHERE key = ?", + (f"999999:other-host:{time.time() - 10:.0f}", LEASE_KEY), + ) + assert viewer.dashboard_stats.try_claim_lease(conn) is True + + def test_concurrent_refreshes_agree(self, viewer): + """A double refresh wastes CPU but must not corrupt anything.""" + seed_database(viewer.db_path) + viewer.dashboard_stats.packet_backfill_rows = PACKETS + _refresh(viewer) + first = _rollup(viewer, local_date_str()) + _refresh(viewer) # simulates a second viewer that won the race + second = _rollup(viewer, local_date_str()) + assert {k: v for k, v in first.items() if k != "computed_at"} == { + k: v for k, v in second.items() if k != "computed_at" + } + + +class TestDashboardApi: + + def test_summary_returns_503_before_first_refresh(self, viewer): + with viewer.app.test_client() as client: + response = client.get("/api/dashboard/summary") + assert response.status_code == 503 + assert response.get_json()["pending"] is True + + def test_summary_etag_yields_304(self, viewer): + seed_database(viewer.db_path) + _refresh(viewer) + with viewer.app.test_client() as client: + first = client.get("/api/dashboard/summary") + second = client.get( + "/api/dashboard/summary", + headers={"If-None-Match": first.headers["ETag"]}, + ) + assert first.status_code == 200 + assert second.status_code == 304 + assert second.data == b"" + + def test_series_gaps_survive_as_null(self, viewer): + seed_database(viewer.db_path) + _refresh(viewer) + with viewer.app.test_client() as client: + payload = client.get("/api/dashboard/series?metric=messages&days=60").get_json() + values = [point["value"] for point in payload["points"]] + assert None in values, "backfilled days must stay null, not become zero" + assert any(v is not None for v in values) + + def test_ratio_series_averages_rather_than_sums(self, viewer): + seed_database(viewer.db_path) + _refresh(viewer) + with viewer.app.test_client() as client: + payload = client.get( + "/api/dashboard/series?metric=multibyte_share&days=30" + ).get_json() + assert payload["is_ratio"] is True + if payload["current_period_total"] is not None: + assert 0 <= payload["current_period_total"] <= 100 + + def test_unknown_metric_and_kind_rejected(self, viewer): + with viewer.app.test_client() as client: + assert client.get("/api/dashboard/series?metric=drop%20table").status_code == 400 + assert client.get("/api/dashboard/top?kind=drop%20table").status_code == 400 + + @pytest.mark.parametrize("kind", list(TOP_KINDS)) + def test_top_endpoints_return_items(self, viewer, kind): + seed_database(viewer.db_path) + with viewer.app.test_client() as client: + payload = client.get(f"/api/dashboard/top?kind={kind}&window=7d&limit=5").get_json() + assert payload["kind"] == kind + assert len(payload["items"]) <= 5 + assert payload["window_label"] == "Last 7 days" + + def test_manual_refresh_advances_generated_at(self, viewer): + seed_database(viewer.db_path) + _refresh(viewer) + with viewer.app.test_client() as client: + before = client.get("/api/dashboard/summary").get_json()["generated_at"] + posted = client.post( + "/api/dashboard/refresh", + headers={"X-Requested-With": "XMLHttpRequest"}, + ) + after = client.get("/api/dashboard/summary").get_json()["generated_at"] + assert posted.status_code == 200 + assert after > before + + def test_stats_shim_is_marked_deprecated(self, viewer): + with viewer.app.test_client() as client: + response = client.get("/api/stats") + assert response.status_code == 200 + assert response.headers["Deprecation"] == "true" + assert "Sunset" in response.headers + assert response.get_json()["deprecated"] is True diff --git a/tests/test_settings_ui.py b/tests/test_settings_ui.py index 0ea17a0..1e0dbd7 100644 --- a/tests/test_settings_ui.py +++ b/tests/test_settings_ui.py @@ -305,6 +305,7 @@ def viewer(tmp_path): with patch.object(BotDataViewer, "_start_database_polling"), \ patch.object(BotDataViewer, "_start_log_tailing"), \ patch.object(BotDataViewer, "_start_cleanup_scheduler"), \ + patch.object(BotDataViewer, "_start_dashboard_refresher"), \ patch.object(BotDataViewer, "_setup_socketio_handlers"), \ patch("modules.web_viewer.app.RepeaterManager"): v = BotDataViewer(db_path=str(tmp_path / "meshcore_bot.db"), config_path=config_path) diff --git a/tests/test_web_viewer.py b/tests/test_web_viewer.py index 0187838..5738a60 100644 --- a/tests/test_web_viewer.py +++ b/tests/test_web_viewer.py @@ -65,6 +65,7 @@ def viewer(tmp_path_factory): patch.object(BotDataViewer, "_start_database_polling", lambda self: None), patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), ): v = BotDataViewer(db_path=db_path, config_path=config_path) @@ -103,6 +104,7 @@ def auth_viewer(tmp_path_factory): patch.object(BotDataViewer, "_start_database_polling", lambda self: None), patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), ): v = BotDataViewer(db_path=db_path, config_path=config_path) @@ -171,14 +173,24 @@ class TestPageRoutes: # Scroll buttons assert 'id="live-scroll-top"' in html assert 'id="live-scroll-bottom"' in html - assert 'scrollLiveFeed' in html # Filter checkboxes with data-type attributes assert 'data-type="packet"' in html assert 'data-type="command"' in html assert 'data-type="message"' in html assert 'live-filter-cb' in html - # [#channel] prefix logic present in JS - assert 'applyFilters' in html + # Behaviour lives in the external, nonce-free dashboard script + assert 'js/dashboard.js' in html + + def test_dashboard_script_wires_live_feed_controls(self): + """The extracted dashboard script still implements the feed behaviour.""" + script = ( + Path(__file__).resolve().parents[1] + / "modules" / "web_viewer" / "static" / "js" / "dashboard.js" + ).read_text(encoding="utf-8") + assert "applyFilters" in script + assert "live-scroll-top" in script + assert "live-scroll-bottom" in script + assert "initLiveFeed" in script def test_realtime(self, client): resp = client.get("/realtime") @@ -223,9 +235,10 @@ class TestPageRoutes: assert resp.status_code == 302 assert resp.headers["Location"].endswith("/config#database") - def test_stats(self, client): + def test_stats_page_removed(self, client): + """The orphaned /stats stub page was folded into the dashboard.""" resp = client.get("/stats") - assert resp.status_code == 200 + assert resp.status_code == 404 def test_greeter(self, client): resp = client.get("/greeter") @@ -2111,6 +2124,7 @@ def socketio_viewer(tmp_path_factory): _patch.object(BotDataViewer, "_start_database_polling", lambda self: None), _patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), _patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + _patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), ): v = BotDataViewer(db_path=db_path, config_path=config_path) @@ -2325,7 +2339,7 @@ class TestConnectedClientsApi: html = resp.data.decode() assert "connectedClientsModal" in html assert "connected-clients-table" in html - assert "loadConnectedClients" in html + assert 'id="health-clients-link"' in html # --------------------------------------------------------------------------- @@ -3491,6 +3505,7 @@ class TestDbPathResolutionFromConfigDir: patch.object(BotDataViewer, "_start_database_polling", lambda self: None), patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), ): return BotDataViewer(config_path=config_path) @@ -3573,6 +3588,7 @@ class TestWebViewerLoggingRespectsLogFile: patch.object(BotDataViewer, "_start_database_polling", lambda self: None), patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), ): return BotDataViewer(config_path=config_path) diff --git a/tests/test_web_viewer_app.py b/tests/test_web_viewer_app.py index 24ec51c..eac58b0 100644 --- a/tests/test_web_viewer_app.py +++ b/tests/test_web_viewer_app.py @@ -69,6 +69,7 @@ def viewer_with_db(tmp_path): with patch.object(BotDataViewer, "_start_database_polling"), \ patch.object(BotDataViewer, "_start_log_tailing"), \ patch.object(BotDataViewer, "_start_cleanup_scheduler"), \ + patch.object(BotDataViewer, "_start_dashboard_refresher"), \ patch.object(BotDataViewer, "_setup_socketio_handlers"), \ patch("modules.web_viewer.app.RepeaterManager"): viewer = BotDataViewer(db_path=db_path, config_path=config_path) @@ -123,6 +124,7 @@ def mock_viewer(tmp_path): with patch.object(BotDataViewer, "_start_database_polling"), \ patch.object(BotDataViewer, "_start_log_tailing"), \ patch.object(BotDataViewer, "_start_cleanup_scheduler"), \ + patch.object(BotDataViewer, "_start_dashboard_refresher"), \ patch.object(BotDataViewer, "_setup_socketio_handlers"), \ patch("modules.web_viewer.app.RepeaterManager"): viewer = BotDataViewer(db_path=db_path, config_path=config_path) @@ -150,6 +152,7 @@ class TestAllowedTables: 'greeted_users', 'feed_subscriptions', 'feed_activity', 'feed_errors', 'path_stats', 'unique_advert_packets', 'schema_version', 'channel_operations', 'channels', 'feed_message_queue', + 'daily_rollup', 'dashboard_snapshot', } assert expected_tables == BotDataViewer.ALLOWED_TABLES diff --git a/tests/test_web_viewer_integration.py b/tests/test_web_viewer_integration.py index b234944..e267ceb 100644 --- a/tests/test_web_viewer_integration.py +++ b/tests/test_web_viewer_integration.py @@ -205,7 +205,7 @@ class TestInsertPacketStreamRow: bi = _make_bot_integration() bi._insert_packet_stream_row('{"x": 1}', "packet") assert not bi._write_queue.empty() - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() assert data == '{"x": 1}' assert row_type == "packet" @@ -228,7 +228,7 @@ class TestCaptureFullPacketData: bi = _make_bot_integration() bi.capture_full_packet_data({"snr": -5.0, "path_len": 2}) assert not bi._write_queue.empty() - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() parsed = json.loads(data) assert row_type == "packet" assert parsed["hops"] == 2 @@ -236,21 +236,21 @@ class TestCaptureFullPacketData: def test_no_path_len_defaults_hops_to_0(self): bi = _make_bot_integration() bi.capture_full_packet_data({"snr": -5.0}) - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() parsed = json.loads(data) assert parsed["hops"] == 0 def test_existing_hops_not_overwritten(self): bi = _make_bot_integration() bi.capture_full_packet_data({"hops": 3, "path_len": 2}) - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() parsed = json.loads(data) assert parsed["hops"] == 3 def test_datetime_added(self): bi = _make_bot_integration() bi.capture_full_packet_data({"snr": 0}) - ts, data, _ = bi._write_queue.get_nowait() + ts, data, _, *_dims = bi._write_queue.get_nowait() parsed = json.loads(data) assert "datetime" in parsed @@ -276,7 +276,7 @@ class TestCaptureCommand: msg.content = "ping" bi.capture_command(msg, "ping", "Pong!", True) assert not bi._write_queue.empty() - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() assert row_type == "command" parsed = json.loads(data) assert parsed["command"] == "ping" @@ -290,7 +290,7 @@ class TestCaptureCommand: msg.channel = "ch" msg.content = "cmd" bi.capture_command(msg, "cmd", "resp", True) - ts, data, _ = bi._write_queue.get_nowait() + ts, data, _, *_dims = bi._write_queue.get_nowait() parsed = json.loads(data) assert parsed["repeat_count"] == 0 @@ -313,7 +313,7 @@ class TestCaptureChannelMessage: msg.is_dm = False bi.capture_channel_message(msg) assert not bi._write_queue.empty() - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() assert row_type == "message" parsed = json.loads(data) assert parsed["type"] == "message" @@ -330,7 +330,7 @@ class TestCaptureChannelMessage: msg.path = "" msg.is_dm = True bi.capture_channel_message(msg) - ts, data, _ = bi._write_queue.get_nowait() + ts, data, _, *_dims = bi._write_queue.get_nowait() parsed = json.loads(data) assert parsed["is_dm"] is True @@ -345,7 +345,7 @@ class TestCapturePacketRouting: bi = _make_bot_integration() bi.capture_packet_routing({"path_nodes": ["aa", "bb"]}) assert not bi._write_queue.empty() - ts, data, row_type = bi._write_queue.get_nowait() + ts, data, row_type, *_dims = bi._write_queue.get_nowait() assert row_type == "routing" diff --git a/tests/test_web_viewer_xss_hardening.py b/tests/test_web_viewer_xss_hardening.py index 4174216..93afc71 100644 --- a/tests/test_web_viewer_xss_hardening.py +++ b/tests/test_web_viewer_xss_hardening.py @@ -10,13 +10,19 @@ import pytest from modules.web_viewer.app import BotDataViewer -TEMPLATES = Path(__file__).resolve().parents[1] / "modules" / "web_viewer" / "templates" +_WEB_VIEWER = Path(__file__).resolve().parents[1] / "modules" / "web_viewer" +TEMPLATES = _WEB_VIEWER / "templates" +STATIC = _WEB_VIEWER / "static" def _source(name: str) -> str: return (TEMPLATES / name).read_text(encoding="utf-8") +def _static_source(name: str) -> str: + return (STATIC / name).read_text(encoding="utf-8") + + def _fake_setup_logging(viewer: BotDataViewer) -> None: viewer.logger = logging.getLogger("test_web_viewer_xss_hardening") viewer.logger.addHandler(logging.NullHandler()) @@ -47,6 +53,7 @@ def client(tmp_path_factory): patch.object(BotDataViewer, "_start_database_polling", lambda self: None), patch.object(BotDataViewer, "_start_log_tailing", lambda self: None), patch.object(BotDataViewer, "_start_cleanup_scheduler", lambda self: None), + patch.object(BotDataViewer, "_start_dashboard_refresher", lambda self: None), ): viewer = BotDataViewer(db_path=str(db_path), config_path=str(config_path)) viewer.app.config["TESTING"] = True @@ -55,19 +62,24 @@ def client(tmp_path_factory): def test_dashboard_has_no_html_parser_sink_for_api_or_mesh_values() -> None: - source = _source("index.html") - assert ".innerHTML" not in source + # The dashboard's script now lives in a static file (no CSP nonce needed), + # so the trust boundary to police moved with it. + template = _source("index.html") + script = _static_source("js/dashboard.js") + assert ".innerHTML" not in template + assert ".innerHTML" not in script # Representative attacker-controlled sender, channel/title, path, client, - # and error values all terminate in textContent-backed DOM construction. + # and role values all terminate in textContent-backed DOM construction. for safe_sink in ( - "entryLabel.textContent = String(label ?? '')", - "entryText.textContent = ` ${String(text ?? '')}`", - "channelName.title = String(channel.channel ?? '')", - "pathCode = this.makeTextElement('code', path.path_string", - "clientCell.appendChild(this.makeTextElement('code', c.client_id))", - "errorText.textContent = String(message ?? '')", + "node.textContent = String(value ?? '');", + "makeTextElement('strong', label)", + "makeTextElement('span', ' ' + String(text ?? ''), 'text-muted')", + "label.title = String(name ?? '');", + "makeTextElement('code', item.path_string, 'small text-break')", + "idCell.appendChild(makeTextElement('code', client.client_id))", + "makeTextElement('div', name, 'mix-row__name')", ): - assert safe_sink in source + assert safe_sink in script, safe_sink def test_feed_metadata_preview_details_and_alerts_use_text_nodes() -> None: @@ -162,9 +174,6 @@ def test_legacy_admin_pages_encode_or_text_render_untrusted_values() -> None: assert "header.innerHTML = `" not in plugins assert "description.textContent = String(plugin.description ?? '')" in plugins - stats = _source("stats.html") - assert ".innerHTML" not in stats - def test_nonce_hardened_templates_have_no_inline_event_attributes() -> None: event_attribute = re.compile(r"<[^>]*\son[a-z]+\s*=", re.IGNORECASE) @@ -176,7 +185,6 @@ def test_nonce_hardened_templates_have_no_inline_event_attributes() -> None: "radio.html", "realtime.html", "contacts.html", - "stats.html", "plugins.html", "greeter.html", "logs.html", @@ -196,7 +204,6 @@ def test_nonce_hardened_templates_have_no_inline_event_attributes() -> None: "/radio", "/realtime", "/contacts", - "/stats", "/plugins", "/greeter", "/logs", @@ -220,10 +227,14 @@ def test_csp_nonce_matches_every_rendered_inline_script(client, path: str) -> No expected_nonce = nonce_match.group(1) html = response.get_data(as_text=True) + # A block with a non-JavaScript type is a data island, not code: the browser + # never executes it, so CSP does not gate it and it needs no nonce. + data_block = re.compile(r'\btype\s*=\s*"(?!(?:text/javascript|module)\b)[^"]+"', re.IGNORECASE) inline_scripts = [ attributes for attributes in re.findall(r"]*)>", html, re.IGNORECASE) if not re.search(r"\bsrc\s*=", attributes, re.IGNORECASE) + and not data_block.search(attributes) ] assert inline_scripts for attributes in inline_scripts: