diff --git a/.env.example b/.env.example index fc3310d..4972949 100644 --- a/.env.example +++ b/.env.example @@ -411,8 +411,15 @@ PROMETHEUS_PORT=9090 # Default cache TTL in seconds (matches web auto-refresh interval) # REDIS_CACHE_TTL=30 -# Cache TTL for dashboard endpoints (seconds) -# REDIS_CACHE_TTL_DASHBOARD=30 +# Cache TTL for dashboard endpoints (seconds; covers /dashboard/* and /routes/{id}/history) +# REDIS_CACHE_TTL_DASHBOARD=300 + +# Cache TTL for /routes/{id} detail endpoint (seconds) +# REDIS_CACHE_TTL_ROUTE_DETAIL=300 + +# Emit HTTP Cache-Control on /api/v1/* responses + ETag/If-None-Match on +# cached endpoints. Disable to suppress all client-side caching directives. +# API_CACHE_CONTROL_ENABLED=true # External Alertmanager port (when using --profile metrics) ALERTMANAGER_PORT=9093 diff --git a/alembic/versions/20260718_1200_6b3430fd84f4_add_event_hash_to_packets.py b/alembic/versions/20260718_1200_6b3430fd84f4_add_event_hash_to_packets.py new file mode 100644 index 0000000..70bdf4e --- /dev/null +++ b/alembic/versions/20260718_1200_6b3430fd84f4_add_event_hash_to_packets.py @@ -0,0 +1,63 @@ +"""add event_hash to raw packets and path hops + +Revision ID: 6b3430fd84f4 +Revises: ec40c67c8c83 +Create Date: 2026-07-18 12:00:00.000000+00:00 + +Adds a nullable ``event_hash`` column to ``raw_packets`` and +``packet_path_hops``. The column denormalizes the underlying structured +event's identity (advertisement / message / telemetry / trace +``event_hash``) so the route evaluator can deduplicate matches by their +underlying event rather than by per-transmission wire hash. + +Background: a single advert or message retransmitted (or flooded) through +the mesh produces one ``RawPacket`` row per on-air copy, each with a +fresh wire ``packet_hash``. Counting those as distinct matches let a +single underlying event satisfy ``packet_count_threshold`` within seconds +and bias the route's health. Joining through ``event_hash`` collapses +all receptions of the same underlying event into one match. + +Rows captured before this migration (and any unclassified wire packets) +keep ``event_hash IS NULL``; the evaluator falls back to the wire +``packet_hash`` for those, preserving today's behaviour until they age +out of the configured ``window_hours``. No data backfill is performed. + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "6b3430fd84f4" +down_revision: Union[str, None] = "ec40c67c8c83" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("raw_packets", schema=None) as batch_op: + batch_op.add_column( + sa.Column("event_hash", sa.String(length=32), nullable=True) + ) + batch_op.create_index("ix_raw_packets_event_hash", ["event_hash"], unique=False) + + with op.batch_alter_table("packet_path_hops", schema=None) as batch_op: + batch_op.add_column( + sa.Column("event_hash", sa.String(length=32), nullable=True) + ) + batch_op.create_index( + "ix_packet_path_hops_event_hash_received_at", + ["event_hash", "received_at"], + unique=False, + ) + + +def downgrade() -> None: + with op.batch_alter_table("packet_path_hops", schema=None) as batch_op: + batch_op.drop_index("ix_packet_path_hops_event_hash_received_at") + batch_op.drop_column("event_hash") + + with op.batch_alter_table("raw_packets", schema=None) as batch_op: + batch_op.drop_index("ix_raw_packets_event_hash") + batch_op.drop_column("event_hash") diff --git a/docker-compose.yml b/docker-compose.yml index 4b785c4..531fb57 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -320,7 +320,9 @@ services: - REDIS_PASSWORD=${REDIS_PASSWORD:-} - REDIS_KEY_PREFIX=${REDIS_KEY_PREFIX:-hub} - REDIS_CACHE_TTL=${REDIS_CACHE_TTL:-30} - - REDIS_CACHE_TTL_DASHBOARD=${REDIS_CACHE_TTL_DASHBOARD:-30} + - REDIS_CACHE_TTL_DASHBOARD=${REDIS_CACHE_TTL_DASHBOARD:-300} + - REDIS_CACHE_TTL_ROUTE_DETAIL=${REDIS_CACHE_TTL_ROUTE_DETAIL:-300} + - API_CACHE_CONTROL_ENABLED=${API_CACHE_CONTROL_ENABLED:-true} # Spam detection: hide-filter switch + threshold (bridged from # FEATURE_SPAM_DETECTION so it tracks the collector and web toggle). On by # default; opt out with FEATURE_SPAM_DETECTION=false. diff --git a/docs/configuration.md b/docs/configuration.md index 74a10ce..57fb494 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -62,7 +62,28 @@ Optional Redis-backed caching for API responses. When disabled or unavailable, t | `REDIS_PASSWORD` | _(none)_ | Redis password (optional) | | `REDIS_KEY_PREFIX` | `hub` | Cache key prefix for multi-instance isolation | | `REDIS_CACHE_TTL` | `30` | Default cache TTL in seconds | -| `REDIS_CACHE_TTL_DASHBOARD` | `30` | Cache TTL for dashboard endpoints in seconds | +| `REDIS_CACHE_TTL_DASHBOARD` | `300` | Cache TTL for `/dashboard/*` endpoints and `/routes/{id}/history` (seconds). Trend/aggregation data tolerates longer staleness than the default TTL. | +| `REDIS_CACHE_TTL_ROUTE_DETAIL` | `300` | Cache TTL for `/routes/{id}` detail endpoint in seconds | + +### HTTP Cache-Control + +On top of the Redis server cache, the API emits HTTP-layer caching directives so browsers and HTTP clients can reuse recent responses without re-fetching. Disabled with `API_CACHE_CONTROL_ENABLED=false` (default `true`). + +| Variable | Default | Description | +| --- | --- | --- | +| `API_CACHE_CONTROL_ENABLED` | `true` | Emit HTTP `Cache-Control` on `/api/v1/*` responses and `ETag` / `If-None-Match` handling on `@cached` endpoints | + +Responses fall into three buckets: + +| Bucket | Endpoints | `Cache-Control` | `ETag` | +| --- | --- | --- | --- | +| `@cached` GETs | All Redis-cached endpoints (`/nodes`, `/routes`, `/routes/{id}`, `/dashboard/*`, `/packets`, `/packet-groups`, `/messages`, `/advertisements`, `/channels`, `/user/profiles`) | `private, max-age=` | Strong SHA-256 hash of the body; `If-None-Match` returns `304 Not Modified` | +| Other GETs | Per-id detail endpoints (`/nodes/{key}`, `/packets/{id}`, `/messages/{id}`, `/user/profile/{id}`, `/trace-paths`, `/telemetry`, etc.) | `private, max-age=0, must-revalidate` | _(none)_ | +| Mutating + health | `POST`/`PUT`/`DELETE` + `/health*` | `no-store` | _(none)_ | + +All API responses use `private` because several `@cached` endpoints are role-aware — their response shape/redaction varies by trusted-proxy `X-User-Id` / `X-User-Roles` headers, so shared/CDN caches must never store them. Browser caches key by URL + request headers and so remain correct. + +Client `max-age` matches the configured Redis TTL for the endpoint (e.g. 300 s on `/routes/{id}`, 30 s on `/dashboard/*`). The `X-Cache: HIT|MISS` observability header continues to be emitted regardless of this setting. ## Collector diff --git a/docs/routes.md b/docs/routes.md index 98d95fc..6c794ef 100644 --- a/docs/routes.md +++ b/docs/routes.md @@ -10,13 +10,15 @@ For the environment-variable reference, see [configuration.md → Feature Flags] A route is an ordered list of two or more nodes (the configured path). For each captured packet reception, the evaluator walks the reception's path-hash sequence and checks whether the route's nodes appear **in order, as a subsequence** (intermediate hops are allowed). When `reversible` is set (the default), the reverse-ordered path is also accepted, so a packet travelling `B → ... → A` counts toward an `A → B` route. +Matches are deduplicated by their **underlying event identity**, not per on-air transmission. The collector denormalizes the structured event's `event_hash` (the same key used to dedup advertisements, messages, telemetry, and traces at the structured-event layer) onto each captured raw packet at ingest time. The evaluator prefers `event_hash` when set, so retransmissions or floods of the same underlying advert/message count once toward `packet_count_threshold` instead of once per on-air copy. Packets captured before this column existed (and any unclassified wire packets) have `event_hash IS NULL` and fall back to the wire `packet_hash`, preserving the previous behaviour until they age out of the configured `window_hours`. + Each route carries these knobs: | Field | Default | Description | | --- | --- | --- | | `match_width` | `1` | Path-hash prefix width in bytes (1/2/3). Higher widths disambiguate nodes that share a short public-key prefix. | | `window_hours` | `24` | Rolling lookback window for the live status card. | -| `packet_count_threshold` | `3` | Distinct matching packets at/above which the route is `healthy`. | +| `packet_count_threshold` | `3` | Distinct matching packets at/above which the route is `healthy`. "Distinct" is per underlying event, not per transmission — see [How health is evaluated](#how-health-is-evaluated) above. | | `clear_threshold` | _(2× threshold)_ | Comfort bar for the `clear`/`marginal` split. Omit/null to use twice the threshold. | | `max_hop_span` | _(unlimited)_ | Caps the position gap between the first and last matched node, to reject matches that wander too far. | | `reversible` | `true` | Also match the path in reverse direction. | diff --git a/docs/upgrading.md b/docs/upgrading.md index b6e4877..0355f06 100644 --- a/docs/upgrading.md +++ b/docs/upgrading.md @@ -40,6 +40,39 @@ Remote observers contribute to the Hub by publishing decoded packets to your MQT In Docker Compose, set `OBSERVER_ALLOWLIST` / `OBSERVER_DENYLIST` in your `.env`; they are wired into the collector service. See [configuration.md → Observer Ingestion Filters](configuration.md#observer-ingestion-filters) and [observer.md](observer.md). +### HTTP Cache-Control on API responses + +The API now emits HTTP `Cache-Control` headers on every `/api/v1/*` response and supports `ETag` / `If-None-Match` → `304 Not Modified` on Redis-cached endpoints. **On by default**, non-breaking, no migration required. + +- `@cached` GET endpoints get `Cache-Control: private, max-age=` (matching their existing Redis TTL) plus a strong `ETag`. On a matching `If-None-Match`, the server returns `304 Not Modified` with an empty body — saving bandwidth on cache hits. +- Uncached GETs under `/api/v1/*` get `Cache-Control: private, max-age=0, must-revalidate` so browsers may store but always revalidate. +- `POST`/`PUT`/`DELETE`/`PATCH` and `/health*` get `Cache-Control: no-store`. +- All API responses are `private` (browser cache only) because several `@cached` endpoints are role-aware — their response shape varies by `X-User-Id`/`X-User-Roles`. Shared/CDN caches must not store them. + +Existing Redis entries written in the legacy bare-JSON format are still read; they're transparently upgraded to the new `{"body": ..., "etag": ...}` envelope on the next cache miss. No manual Redis flush is required; old entries expire within one TTL period (≤ 300 s by default). + +To suppress all client-side caching directives (e.g. for debugging): + +| Variable | Default | Description | +| --- | --- | --- | +| `API_CACHE_CONTROL_ENABLED` | `true` | Emit `Cache-Control` on `/api/v1/*` + `ETag`/`If-None-Match` on `@cached` endpoints. Set `false` to disable. | + +The `X-Cache: HIT|MISS` observability header is unaffected — it is still emitted whenever a request flows through the `@cached` decorator, regardless of this setting. + +### Dashboard Cache TTL default raised (30s → 300s) + +The default for `REDIS_CACHE_TTL_DASHBOARD` has been raised from **30 s** to **300 s** (5 minutes). This setting covers all `/dashboard/*` endpoints (`stats`, `activity`, `packet-activity`, `packet-breakdown`, `message-activity`, `node-count`) **and** the per-route health strip `GET /api/v1/routes/{id}/history`. These all return trend/aggregation data where minute-level staleness is invisible to operators but every cache miss costs seconds of SQL/aggregation work — the old 30 s default was leaving most of that work unpaid on typical dashboards. + +| Variable | Old default | New default | +| --- | --- | --- | +| `REDIS_CACHE_TTL_DASHBOARD` | `30` | `300` | + +- **Operators who already set the env var explicitly:** no change — your value still wins. +- **Operators on the default:** silently bumped to 5 min staleness on the dashboard and route health strips. If you relied on the old 30 s behaviour (e.g. for development or tight freshness SLOs), set `REDIS_CACHE_TTL_DASHBOARD=30` to restore it. +- **Existing Redis entries at deploy time:** keep their original TTL and expire naturally within ≤ 30 s. No flush is required. +- **Browser `max-age`:** the HTTP `Cache-Control: max-age=` emitted by the middleware follows the Redis TTL, so browser-side caching on these endpoints also extends to 5 min. Clicking around the dashboard will be instant for that window. A hard refresh (`Ctrl+Shift+R`) bypasses this if you need the latest data. +- **Side-effect fix:** the reported short-TTL behaviour on the per-route healthchart (which shares this setting) is resolved. + ## v0.15.0 ### Spam Detection (score, hide, and toggle likely-spam messages) @@ -260,6 +293,7 @@ A new optional Redis-backed caching layer reduces database load for read-heavy A | `REDIS_KEY_PREFIX` | `hub` | Cache key prefix (change per instance for multi-instance setups) | | `REDIS_CACHE_TTL` | `30` | Default cache TTL in seconds | | `REDIS_CACHE_TTL_DASHBOARD` | `30` | Cache TTL for dashboard endpoints | +| `REDIS_CACHE_TTL_ROUTE_DETAIL` | `300` | Cache TTL for `/routes/{id}` detail endpoint | **Docker Compose:** Redis is available via the `cache` profile: diff --git a/src/meshcore_hub/api/app.py b/src/meshcore_hub/api/app.py index 7da423b..c4c2fba 100644 --- a/src/meshcore_hub/api/app.py +++ b/src/meshcore_hub/api/app.py @@ -90,7 +90,9 @@ def create_app( redis_password: str | None = None, redis_key_prefix: str = "hub", redis_cache_ttl: int = 30, - redis_cache_ttl_dashboard: int = 30, + redis_cache_ttl_dashboard: int = 300, + redis_cache_ttl_route_detail: int = 300, + api_cache_control_enabled: bool = True, spam_detection_enabled: bool = False, spam_score_threshold: float = 0.65, ) -> FastAPI: @@ -119,6 +121,9 @@ def create_app( redis_key_prefix: Prefix for all cache keys redis_cache_ttl: Default cache TTL in seconds redis_cache_ttl_dashboard: Cache TTL for dashboard endpoints + redis_cache_ttl_route_detail: Cache TTL for /routes/{id} detail endpoint + api_cache_control_enabled: Emit HTTP Cache-Control on /api/v1/* and + ETag/If-None-Match handling on @cached endpoints. Returns: Configured FastAPI application @@ -154,6 +159,8 @@ def create_app( app.state.redis_key_prefix = redis_key_prefix app.state.redis_cache_ttl = redis_cache_ttl app.state.redis_cache_ttl_dashboard = redis_cache_ttl_dashboard + app.state.redis_cache_ttl_route_detail = redis_cache_ttl_route_detail + app.state.api_cache_control_enabled = api_cache_control_enabled app.state.spam_detection_enabled = spam_detection_enabled app.state.spam_score_threshold = spam_score_threshold @@ -170,11 +177,61 @@ def create_app( ) @app.middleware("http") - async def cache_header_middleware(request: Request, call_next: Any) -> Any: + async def api_cache_middleware(request: Request, call_next: Any) -> Any: + """Apply X-Cache observability + HTTP Cache-Control / ETag headers. + + Buckets (only when ``app.state.api_cache_control_enabled`` is True): + * ``@cached`` endpoints (``request.state.cache_control_ttl`` set by + the decorator): ``private, max-age=`` + ``ETag`` echoed back. + * Uncached GETs under ``/api/v1``: ``private, max-age=0, + must-revalidate`` so browsers revalidate but may store. + * Mutating methods (POST/PUT/DELETE/PATCH): ``no-store``. + * ``/health*`` endpoints: ``no-store``. + + ``private`` is used everywhere because several cached endpoints are + role-aware — their response shape/redaction varies by trusted-proxy + ``X-User-Id`` / ``X-User-Roles`` headers, so shared/CDN caches must + never store them. + """ response = await call_next(request) + + # X-Cache observability header (always emitted, even when the kill + # switch is on, so monitoring can still see hit/miss ratios). cache_status = getattr(request.state, "cache_status", None) if cache_status is not None: response.headers["X-Cache"] = cache_status + + # ETag from the @cached decorator (304 responses already carry it; + # this branch covers the 200 path where the decorator returned a + # plain model and FastAPI serialized it). + etag = getattr(request.state, "api_etag", None) + if etag is not None and "etag" not in response.headers: + response.headers["ETag"] = etag + + if not getattr(app.state, "api_cache_control_enabled", True): + return response + + # Don't overwrite a Cache-Control header set explicitly by a handler + # or by the @cached decorator's 304 Response. + if "cache-control" in response.headers: + return response + + method = request.method.upper() + path = request.url.path + + if method in ("POST", "PUT", "DELETE", "PATCH"): + response.headers["Cache-Control"] = "no-store" + elif path.startswith("/health"): + response.headers["Cache-Control"] = "no-store" + elif path.startswith("/api/"): + ttl = getattr(request.state, "cache_control_ttl", 0) + if isinstance(ttl, int) and ttl > 0: + response.headers["Cache-Control"] = f"private, max-age={ttl}" + else: + response.headers["Cache-Control"] = ( + "private, max-age=0, must-revalidate" + ) + return response # Include routers @@ -272,6 +329,8 @@ def create_app_from_env() -> FastAPI: redis_key_prefix=settings.redis_key_prefix, redis_cache_ttl=settings.redis_cache_ttl, redis_cache_ttl_dashboard=settings.redis_cache_ttl_dashboard, + redis_cache_ttl_route_detail=settings.redis_cache_ttl_route_detail, + api_cache_control_enabled=settings.api_cache_control_enabled, spam_detection_enabled=settings.spam_detection_enabled, spam_score_threshold=settings.spam_score_threshold, ) diff --git a/src/meshcore_hub/api/cache.py b/src/meshcore_hub/api/cache.py index 18a3eff..b3f783b 100644 --- a/src/meshcore_hub/api/cache.py +++ b/src/meshcore_hub/api/cache.py @@ -1,6 +1,7 @@ """Cache decorator for API endpoints.""" import functools +import hashlib import inspect import json import logging @@ -8,6 +9,7 @@ from typing import Any, Callable, Optional from urllib.parse import urlencode from fastapi import Request +from fastapi.responses import Response logger = logging.getLogger(__name__) @@ -50,36 +52,105 @@ def _build_cache_key( return f"{endpoint_name}:{sorted_query_string(request)}" -def _lookup(cache: Any, cache_key: str, request: Request) -> Any: - """Return the cached value, or _MISS, and record the cache status.""" +def _compute_etag(serialized_body: str) -> str: + """Compute a strong ETag header value (quoted hex hash) for a body. + + Uses the first 32 hex chars of SHA-256 — collision-safe for cache keys + and short enough to fit comfortably in a request header. + """ + digest = hashlib.sha256(serialized_body.encode("utf-8")).hexdigest()[:32] + return f'"{digest}"' + + +def _etag_matches(if_none_match: str, etag: str) -> bool: + """Return True if the client's If-None-Match header matches the ETag. + + Per RFC 7232 the header may carry a comma-separated list or the wildcard + ``*``. Weak indicators (``W/``) are accepted on the client side. We do not + emit weak ETags, but clients are allowed to send them. + """ + if if_none_match.strip() == "*": + return True + for token in if_none_match.split(","): + candidate = token.strip() + if candidate.startswith("W/"): + candidate = candidate[2:] + if candidate == etag: + return True + return False + + +def _serialize_for_cache(result: Any) -> tuple[Any, str]: + """Convert a handler result into (body_json, serialized_json) for storage. + + Returns the JSON-compatible body (suitable for JSONResponse) and the + canonical serialized string used to compute the ETag. Pydantic models are + dumped via ``model_dump(mode="json")`` so datetimes become ISO strings + deterministically. + """ + if hasattr(result, "model_dump"): + body = result.model_dump(mode="json") + elif isinstance(result, dict): + body = result + else: + body = result + serialized = json.dumps(body, default=str) + return body, serialized + + +def _store(cache: Any, cache_key: str, result: Any, ttl: int) -> tuple[Any, str]: + """Serialize, ETag, and store a handler result in the cache. + + Returns ``(body, etag)`` so the caller can return the body to FastAPI + without re-serializing. Stores the new envelope format + ``{"body": ..., "etag": "..."}`` so future hits can return the ETag + without re-hashing. + """ + body, serialized = _serialize_for_cache(result) + etag = _compute_etag(serialized) + envelope = json.dumps({"body": body, "etag": etag}) try: - cached_value = cache.get(cache_key) - except Exception as e: - logger.warning("Redis GET error for %s: %s", cache_key, e) - cached_value = None - - if cached_value is not None: - logger.debug("Cache HIT: %s", cache_key) - request.state.cache_status = "HIT" - return json.loads(cached_value) - - logger.debug("Cache MISS: %s", cache_key) - request.state.cache_status = "MISS" - return _MISS - - -def _store(cache: Any, cache_key: str, result: Any, ttl: int) -> None: - """Serialize and store a handler result in the cache.""" - try: - if hasattr(result, "model_dump"): - serialized = json.dumps(result.model_dump(mode="json")) - elif isinstance(result, dict): - serialized = json.dumps(result) - else: - serialized = json.dumps(result, default=str) - cache.set(cache_key, serialized, ttl) + cache.set(cache_key, envelope, ttl) except Exception as e: logger.warning("Cache store error for %s: %s", cache_key, e) + return body, etag + + +def _lookup(cache: Any, cache_key: str, request: Request) -> tuple[Any, str]: + """Return ``(body, etag)`` on hit, or ``(_MISS, "")`` on miss. + + Reads both the new envelope format ``{"body": ..., "etag": "..."}`` and + the legacy bare-body format (which carried no ETag). Legacy entries are + hashed on read so they still serve correctly; natural expiry migrates + them to the envelope format on the next write. + """ + try: + raw = cache.get(cache_key) + except Exception as e: + logger.warning("Redis GET error for %s: %s", cache_key, e) + return _MISS, "" + + if raw is None: + logger.debug("Cache MISS: %s", cache_key) + request.state.cache_status = "MISS" + return _MISS, "" + + logger.debug("Cache HIT: %s", cache_key) + request.state.cache_status = "HIT" + try: + parsed = json.loads(raw) + except (json.JSONDecodeError, TypeError): + # Garbage in cache — treat as miss and let the next write repair it. + request.state.cache_status = "MISS" + return _MISS, "" + + if isinstance(parsed, dict) and "body" in parsed and "etag" in parsed: + return parsed["body"], parsed["etag"] + # Legacy entry: body stored bare, no ETag envelope. Hash on the fly so + # the client still gets a usable ETag; the next MISS overwrites it with + # the envelope format. + serialized = json.dumps(parsed, default=str) + return parsed, _compute_etag(serialized) def cached( @@ -106,19 +177,37 @@ def cached( async def async_wrapper(*args: Any, **kwargs: Any) -> Any: request = _find_request(kwargs) cache = getattr(request.app.state, "redis_cache", None) + ttl = getattr(request.app.state, ttl_setting, 30) + # Always advertise the client-cache TTL so the middleware can + # emit Cache-Control regardless of whether Redis is configured. + request.state.cache_control_ttl = ttl + if cache is None: return await func(*args, **kwargs) - ttl = getattr(request.app.state, ttl_setting, 30) cache_key = _build_cache_key(request, endpoint_name, key_builder) + body, etag = _lookup(cache, cache_key, request) - cached = _lookup(cache, cache_key, request) - if cached is not _MISS: - return cached + if body is _MISS: + # MISS: call the handler, store, and return the original + # result so FastAPI's response_model handling still sees + # the Pydantic model (not the JSON-dumped dict). + result = await func(*args, **kwargs) + _, etag = _store(cache, cache_key, result, ttl) + return_value = result + else: + # HIT: only have the JSON-deserialized body. + return_value = body - result = await func(*args, **kwargs) - _store(cache, cache_key, result, ttl) - return result + # ETag + If-None-Match handling. + request.state.api_etag = etag + if_none_match = request.headers.get("if-none-match") + if if_none_match and _etag_matches(if_none_match, etag): + return Response( + status_code=304, + headers={"ETag": etag}, + ) + return return_value return async_wrapper @@ -126,19 +215,30 @@ def cached( def sync_wrapper(*args: Any, **kwargs: Any) -> Any: request = _find_request(kwargs) cache = getattr(request.app.state, "redis_cache", None) + ttl = getattr(request.app.state, ttl_setting, 30) + request.state.cache_control_ttl = ttl + if cache is None: return func(*args, **kwargs) - ttl = getattr(request.app.state, ttl_setting, 30) cache_key = _build_cache_key(request, endpoint_name, key_builder) + body, etag = _lookup(cache, cache_key, request) - cached = _lookup(cache, cache_key, request) - if cached is not _MISS: - return cached + if body is _MISS: + result = func(*args, **kwargs) + _, etag = _store(cache, cache_key, result, ttl) + return_value = result + else: + return_value = body - result = func(*args, **kwargs) - _store(cache, cache_key, result, ttl) - return result + request.state.api_etag = etag + if_none_match = request.headers.get("if-none-match") + if if_none_match and _etag_matches(if_none_match, etag): + return Response( + status_code=304, + headers={"ETag": etag}, + ) + return return_value return sync_wrapper diff --git a/src/meshcore_hub/api/cli.py b/src/meshcore_hub/api/cli.py index b17f415..256ea36 100644 --- a/src/meshcore_hub/api/cli.py +++ b/src/meshcore_hub/api/cli.py @@ -173,10 +173,26 @@ import click @click.option( "--redis-cache-ttl-dashboard", type=int, - default=30, + default=300, envvar="REDIS_CACHE_TTL_DASHBOARD", help="Cache TTL for dashboard endpoints (seconds)", ) +@click.option( + "--redis-cache-ttl-route-detail", + type=int, + default=300, + envvar="REDIS_CACHE_TTL_ROUTE_DETAIL", + help="Cache TTL for /routes/{id} detail endpoint (seconds)", +) +@click.option( + "--api-cache-control-enabled/--no-api-cache-control", + default=True, + envvar="API_CACHE_CONTROL_ENABLED", + help=( + "Emit HTTP Cache-Control headers on /api/v1/* responses and ETag / " + "If-None-Match handling on @cached endpoints" + ), +) @click.option( "--reload", is_flag=True, @@ -223,6 +239,8 @@ def api( redis_key_prefix: str, redis_cache_ttl: int, redis_cache_ttl_dashboard: int, + redis_cache_ttl_route_detail: int, + api_cache_control_enabled: bool, reload: bool, workers: int, ) -> None: @@ -280,8 +298,11 @@ def api( click.echo(f"Redis: {redis_host}:{redis_port}/{redis_db}") click.echo(f"Redis key prefix: {redis_key_prefix}") click.echo( - f"Redis cache TTL: {redis_cache_ttl}s (dashboard: {redis_cache_ttl_dashboard}s)" + f"Redis cache TTL: {redis_cache_ttl}s " + f"(dashboard: {redis_cache_ttl_dashboard}s, " + f"route detail: {redis_cache_ttl_route_detail}s)" ) + click.echo(f"API Cache-Control enabled: {api_cache_control_enabled}") click.echo(f"Reload mode: {reload}") click.echo(f"Workers: {workers}") click.echo("=" * 50) @@ -343,6 +364,8 @@ def api( redis_key_prefix=redis_key_prefix, redis_cache_ttl=redis_cache_ttl, redis_cache_ttl_dashboard=redis_cache_ttl_dashboard, + redis_cache_ttl_route_detail=redis_cache_ttl_route_detail, + api_cache_control_enabled=api_cache_control_enabled, ) click.echo("\nStarting API server...") diff --git a/src/meshcore_hub/api/routes/routes.py b/src/meshcore_hub/api/routes/routes.py index 8bccde0..923a4c4 100644 --- a/src/meshcore_hub/api/routes/routes.py +++ b/src/meshcore_hub/api/routes/routes.py @@ -214,6 +214,11 @@ def create_route( @router.get("/{route_id}", response_model=RouteDetail) +@cached( + "routes/{id}", + ttl_setting="redis_cache_ttl_route_detail", + key_builder=_routes_key_builder, +) def get_route( _: RequireRead, session: DbSession, diff --git a/src/meshcore_hub/collector/handlers/advertisement.py b/src/meshcore_hub/collector/handlers/advertisement.py index d198732..0a8392a 100644 --- a/src/meshcore_hub/collector/handlers/advertisement.py +++ b/src/meshcore_hub/collector/handlers/advertisement.py @@ -2,7 +2,7 @@ import logging from datetime import datetime, timedelta, timezone -from typing import Any +from typing import Any, Optional from sqlalchemy import select from sqlalchemy.exc import IntegrityError @@ -33,7 +33,7 @@ def handle_advertisement( event_type: str, payload: dict[str, Any], db: DatabaseManager, -) -> None: +) -> Optional[str]: """Handle an advertisement event. 1. Upserts the node in the nodes table @@ -45,11 +45,16 @@ def handle_advertisement( event_type: Event type name payload: Advertisement payload db: Database manager + + Returns: + The ``event_hash`` of the underlying advertisement (used by the + subscriber to denormalize the identity onto the captured raw + packet), or ``None`` on early exit. """ adv_public_key = payload.get("public_key") if not adv_public_key: logger.warning("Advertisement missing public_key") - return + return None name = payload.get("name") adv_type = payload.get("adv_type") @@ -154,7 +159,7 @@ def handle_advertisement( f"Added receiver {public_key[:12]}... to advertisement " f"(hash={event_hash[:8]}...)" ) - return + return event_hash # Find or create advertised node node_query = select(Node).where(Node.public_key == adv_public_key) @@ -237,8 +242,9 @@ def handle_advertisement( path_len=path_len, observed_at=now, ) - return + return event_hash logger.info( f"Stored advertisement from {name or adv_public_key[:12]!r} (type={adv_type})" ) + return event_hash diff --git a/src/meshcore_hub/collector/handlers/message.py b/src/meshcore_hub/collector/handlers/message.py index 9c027b4..f5444f0 100644 --- a/src/meshcore_hub/collector/handlers/message.py +++ b/src/meshcore_hub/collector/handlers/message.py @@ -2,7 +2,7 @@ import logging from datetime import datetime, timezone -from typing import Any +from typing import Any, Optional from sqlalchemy import select from sqlalchemy.exc import IntegrityError @@ -25,7 +25,7 @@ def handle_contact_message( event_type: str, payload: dict[str, Any], db: DatabaseManager, -) -> None: +) -> Optional[str]: """Handle a contact message event. Args: @@ -33,8 +33,12 @@ def handle_contact_message( event_type: Event type name payload: Message payload db: Database manager + + Returns: + The ``event_hash`` of the underlying message, or ``None`` on early + exit. """ - _handle_message(public_key, "contact", payload, db) + return _handle_message(public_key, "contact", payload, db) def handle_channel_message( @@ -42,7 +46,7 @@ def handle_channel_message( event_type: str, payload: dict[str, Any], db: DatabaseManager, -) -> None: +) -> Optional[str]: """Handle a channel message event. Args: @@ -50,8 +54,12 @@ def handle_channel_message( event_type: Event type name payload: Message payload db: Database manager + + Returns: + The ``event_hash`` of the underlying message, or ``None`` on early + exit. """ - _handle_message(public_key, "channel", payload, db) + return _handle_message(public_key, "channel", payload, db) def _handle_message( @@ -59,7 +67,7 @@ def _handle_message( message_type: str, payload: dict[str, Any], db: DatabaseManager, -) -> None: +) -> Optional[str]: """Handle a message event (contact or channel). Args: @@ -67,11 +75,15 @@ def _handle_message( message_type: Message type ('contact' or 'channel') payload: Message payload db: Database manager + + Returns: + The ``event_hash`` of the underlying message, or ``None`` on early + exit. """ text = payload.get("text") if not text: logger.warning("Message missing text content") - return + return None now = datetime.now(timezone.utc) @@ -144,7 +156,7 @@ def _handle_message( f"Added receiver {public_key[:12]}... to message " f"(hash={event_hash[:8]}...)" ) - return + return event_hash # Spam scoring (online): only on the first insert of an event_hash, and # only when the feature is switched on. When off, the columns stay null @@ -227,7 +239,7 @@ def _handle_message( path_len=path_len, observed_at=now, ) - return + return event_hash # Surface the spam score (and the signals that drove it) in the log so it is # observable without querying the DB. Likely-spam (>= threshold) is logged at @@ -252,3 +264,5 @@ def _handle_message( logger.warning(line) else: logger.info(line) + + return event_hash diff --git a/src/meshcore_hub/collector/handlers/raw_packet.py b/src/meshcore_hub/collector/handlers/raw_packet.py index e8586ab..bbd603e 100644 --- a/src/meshcore_hub/collector/handlers/raw_packet.py +++ b/src/meshcore_hub/collector/handlers/raw_packet.py @@ -11,7 +11,7 @@ import logging from datetime import datetime, timezone from typing import Any, Optional -from sqlalchemy import select +from sqlalchemy import select, update from meshcore_hub.collector.letsmesh_normalizer import LetsMeshNormalizer from meshcore_hub.common.database import DatabaseManager @@ -56,7 +56,7 @@ def store_raw_packet( decoded_packet: Optional[dict[str, Any]], event_type: str, db: DatabaseManager, -) -> None: +) -> Optional[str]: """Capture a single raw packet from the LetsMesh ``packets`` feed. Args: @@ -65,6 +65,11 @@ def store_raw_packet( decoded_packet: Decoder output already produced during normalization event_type: How the collector classified the packet db: Database manager + + Returns: + The ``raw_packet.id`` of the inserted row, or ``None`` on failure. + The caller uses the id to backfill ``event_hash`` after the + structured handler resolves the underlying event identity. """ now = datetime.now(timezone.utc) @@ -168,3 +173,34 @@ def store_raw_packet( ) logger.debug("Captured raw packet: %s (%s)", packet_hash or "unknown", event_type) + return raw_packet.id + + +def update_raw_packet_event_hash( + raw_packet_id: str, + event_hash: str, + db: DatabaseManager, +) -> None: + """Backfill ``event_hash`` onto a captured raw packet and its hops. + + Called by the subscriber after the structured handler resolves the + underlying event identity. Two single-row UPDATEs scoped by + ``raw_packet_id``; no-op if the row no longer exists (e.g. retention + cleanup ran between capture and dispatch). + + Args: + raw_packet_id: The id returned by :func:`store_raw_packet`. + event_hash: The ``event_hash`` returned by the structured handler. + db: Database manager. + """ + with db.session_scope() as session: + session.execute( + update(RawPacket) + .where(RawPacket.id == raw_packet_id) + .values(event_hash=event_hash) + ) + session.execute( + update(PacketPathHop) + .where(PacketPathHop.raw_packet_id == raw_packet_id) + .values(event_hash=event_hash) + ) diff --git a/src/meshcore_hub/collector/handlers/telemetry.py b/src/meshcore_hub/collector/handlers/telemetry.py index 0883d27..9aec534 100644 --- a/src/meshcore_hub/collector/handlers/telemetry.py +++ b/src/meshcore_hub/collector/handlers/telemetry.py @@ -2,7 +2,7 @@ import logging from datetime import datetime, timezone -from typing import Any +from typing import Any, Optional from sqlalchemy import select from sqlalchemy.exc import IntegrityError @@ -19,7 +19,7 @@ def handle_telemetry( event_type: str, payload: dict[str, Any], db: DatabaseManager, -) -> None: +) -> Optional[str]: """Handle a telemetry response event. Args: @@ -27,11 +27,15 @@ def handle_telemetry( event_type: Event type name payload: Telemetry payload db: Database manager + + Returns: + The ``event_hash`` of the underlying telemetry record, or ``None`` + on early exit. """ node_public_key = payload.get("node_public_key") if not node_public_key: logger.warning("Telemetry missing node_public_key") - return + return None now = datetime.now(timezone.utc) @@ -100,7 +104,7 @@ def handle_telemetry( f"Added receiver {public_key[:12]}... to telemetry " f"(node={node_public_key[:12]}...)" ) - return + return event_hash # Find or create reporting node reporting_node = None @@ -164,7 +168,7 @@ def handle_telemetry( path_len=path_len, observed_at=now, ) - return + return event_hash # Log telemetry values if parsed_data: @@ -172,3 +176,5 @@ def handle_telemetry( logger.info(f"Stored telemetry from {node_public_key[:12]!r}: {values}") else: logger.info(f"Stored telemetry from {node_public_key[:12]!r}") + + return event_hash diff --git a/src/meshcore_hub/collector/handlers/trace.py b/src/meshcore_hub/collector/handlers/trace.py index 2b3c7e4..3ba7210 100644 --- a/src/meshcore_hub/collector/handlers/trace.py +++ b/src/meshcore_hub/collector/handlers/trace.py @@ -2,7 +2,7 @@ import logging from datetime import datetime, timezone -from typing import Any +from typing import Any, Optional from sqlalchemy import select from sqlalchemy.exc import IntegrityError @@ -19,7 +19,7 @@ def handle_trace_data( event_type: str, payload: dict[str, Any], db: DatabaseManager, -) -> None: +) -> Optional[str]: """Handle a trace data event. Args: @@ -27,11 +27,15 @@ def handle_trace_data( event_type: Event type name payload: Trace data payload db: Database manager + + Returns: + The ``event_hash`` of the underlying trace record, or ``None`` on + early exit. """ initiator_tag = payload.get("initiator_tag") if initiator_tag is None: logger.warning("Trace data missing initiator_tag") - return + return None now = datetime.now(timezone.utc) @@ -86,7 +90,7 @@ def handle_trace_data( f"Added receiver {public_key[:12]}... to trace " f"(tag={initiator_tag})" ) - return + return event_hash # Create trace path record trace_path = TracePath( @@ -135,6 +139,7 @@ def handle_trace_data( path_len=path_len, observed_at=now, ) - return + return event_hash logger.info(f"Stored trace data: tag={initiator_tag}, hops={hop_count}") + return event_hash diff --git a/src/meshcore_hub/collector/routes.py b/src/meshcore_hub/collector/routes.py index 9cdb52d..6761c8c 100644 --- a/src/meshcore_hub/collector/routes.py +++ b/src/meshcore_hub/collector/routes.py @@ -74,6 +74,23 @@ def derive_quality( return RouteQuality.UNKNOWN.value +def _match_identity(hops: list[dict[str, Any]]) -> Optional[str]: + """Identity used to dedup matched receptions into one match per event. + + Prefers the denormalized ``event_hash`` (the underlying structured + event's identity, populated at ingest) so that retransmissions of the + same advert/message/telemetry/trace count once instead of once per + on-air copy. Falls back to the per-transmission wire ``packet_hash`` + when ``event_hash`` is NULL (legacy rows captured before the column + existed, or unclassified packets that didn't trigger a structured + handler). + """ + if not hops: + return None + first = hops[0] + return first.get("event_hash") or first.get("packet_hash") + + def _subsequence_indices( path: list[dict[str, Any]], expected: list[str], @@ -281,6 +298,7 @@ def fetch_candidate_paths( PacketPathHop.position, PacketPathHop.node_hash, PacketPathHop.packet_hash, + PacketPathHop.event_hash, PacketPathHop.received_at, PacketPathHop.observer_node_id, ) @@ -298,6 +316,7 @@ def fetch_candidate_paths( "position": row.position, "node_hash": row.node_hash, "packet_hash": row.packet_hash, + "event_hash": row.event_hash, "received_at": row.received_at, "observer_node_id": row.observer_node_id, } @@ -337,6 +356,7 @@ def _fetch_matching_hops( PacketPathHop.position, PacketPathHop.node_hash, PacketPathHop.packet_hash, + PacketPathHop.event_hash, PacketPathHop.received_at, ) .where( @@ -359,6 +379,7 @@ def _fetch_matching_hops( "position": row.position, "node_hash": row.node_hash, "packet_hash": row.packet_hash, + "event_hash": row.event_hash, "received_at": row.received_at, } ) @@ -475,9 +496,9 @@ def evaluate_route( matched_packets: set[str] = set() for hops in paths.values(): if _match_hops(hops, expected, route.max_hop_span, reversible): - ph = hops[0]["packet_hash"] - if ph: - matched_packets.add(ph) + identity = _match_identity(hops) + if identity: + matched_packets.add(identity) if len(matched_packets) >= eff_clear: return ( RouteState.HEALTHY.value, @@ -528,9 +549,9 @@ def evaluate_route_day( matched_packets: set[str] = set() for hops in paths.values(): if _match_hops(hops, expected, route.max_hop_span, reversible): - ph = hops[0]["packet_hash"] - if ph: - matched_packets.add(ph) + identity = _match_identity(hops) + if identity: + matched_packets.add(identity) if len(matched_packets) >= eff_clear: return ( RouteQuality.CLEAR.value, @@ -673,9 +694,9 @@ def evaluate_route_history( matched_packets: set[str] = set() for hops in day_paths[i].values(): if _match_hops(hops, expected, route.max_hop_span, reversible): - ph = hops[0]["packet_hash"] - if ph: - matched_packets.add(ph) + identity = _match_identity(hops) + if identity: + matched_packets.add(identity) if len(matched_packets) >= eff_clear: break @@ -765,7 +786,13 @@ def recent_matches( limit: int = 3, now: Optional[datetime] = None, ) -> list[dict[str, Any]]: - """Return the latest *limit* matching paths for a route.""" + """Return the latest *limit* matching paths for a route. + + Deduplicates by event identity (preferring ``event_hash``, falling back + to wire ``packet_hash``) so the UI shows one row per underlying event + rather than one row per retransmission. When multiple receptions share + an identity, the newest is returned. + """ expected = _route_expected_hashes(route) if len(expected) < 2: return [] @@ -781,22 +808,37 @@ def recent_matches( session, expected, since, observer_ids, reversible ) - matches: list[dict[str, Any]] = [] + # Keep the newest match per identity so the UI lists distinct underlying + # events rather than every retransmission of the same event. + matches_by_identity: dict[str, dict[str, Any]] = {} for hops in paths.values(): subpath = _matched_subpath(hops, expected, route.max_hop_span, reversible) if not subpath: continue + identity = _match_identity(subpath) + if identity is None: + # Fall back to a synthetic unique key so unmatched-identity + # receptions still surface (one row each). + identity = f"__rawid_{id(subpath)}" first = subpath[0] if subpath else {} - matches.append( - { - "packet_hash": first.get("packet_hash"), - "hops": subpath, - "received_at": first.get("received_at"), - "observer_node_id": first.get("observer_node_id"), - } + received_at = first.get("received_at") or datetime.min.replace( + tzinfo=timezone.utc ) + candidate = { + "packet_hash": first.get("packet_hash"), + "event_hash": first.get("event_hash"), + "hops": subpath, + "received_at": first.get("received_at"), + "observer_node_id": first.get("observer_node_id"), + } + existing = matches_by_identity.get(identity) + if existing is None or received_at > ( + existing.get("received_at") or datetime.min.replace(tzinfo=timezone.utc) + ): + matches_by_identity[identity] = candidate - matches.sort( + matches = sorted( + matches_by_identity.values(), key=lambda m: m["received_at"] or datetime.min.replace(tzinfo=timezone.utc), reverse=True, ) @@ -878,9 +920,9 @@ def preview_route( for hops in paths.values(): if _match_hops(hops, expected, max_hop_span, reversible): - ph = hops[0]["packet_hash"] - if ph: - matched_packets.add(ph) + identity = _match_identity(hops) + if identity: + matched_packets.add(identity) obs = hops[0]["observer_node_id"] if obs: contributing[obs] = contributing.get(obs, 0) + 1 diff --git a/src/meshcore_hub/collector/subscriber.py b/src/meshcore_hub/collector/subscriber.py index fefd6af..651e2dc 100644 --- a/src/meshcore_hub/collector/subscriber.py +++ b/src/meshcore_hub/collector/subscriber.py @@ -31,8 +31,10 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Handler type: receives (public_key, event_type, payload, db_manager) -EventHandler = Callable[[str, str, dict[str, Any], DatabaseManager], None] +# Handler type: receives (public_key, event_type, payload, db_manager), +# returns the underlying event's event_hash (or None) so the subscriber can +# denormalize the identity onto the captured raw packet. +EventHandler = Callable[[str, str, dict[str, Any], DatabaseManager], Optional[str]] class Subscriber(LetsMeshNormalizer): @@ -253,13 +255,22 @@ class Subscriber(LetsMeshNormalizer): public_key, event_type, normalized_payload = parsed logger.debug("Received event: %s from %s...", event_type, public_key[:12]) - # Capture the raw packet (packets feed only) independent of, and before, - # structured dispatch so the raw_packets table is complete. The boolean - # short-circuit avoids the insert entirely when capture is disabled. + # Capture the raw packet (packets feed only) independent of, and + # before, structured dispatch so the raw_packets table is complete. + # The boolean short-circuit avoids the insert entirely when capture + # is disabled. The returned id lets us backfill ``event_hash`` onto + # the row (and its hops) once the structured handler resolves the + # underlying event identity. + raw_packet_id: str | None = None if self._raw_packet_capture_enabled: - self._maybe_capture_raw_packet(topic, public_key, event_type, payload) + raw_packet_id = self._maybe_capture_raw_packet( + topic, public_key, event_type, payload + ) - self._dispatch_event(public_key, event_type, normalized_payload) + event_hash = self._dispatch_event(public_key, event_type, normalized_payload) + + if raw_packet_id and event_hash: + self._backfill_raw_packet_event_hash(raw_packet_id, event_hash) def _maybe_capture_raw_packet( self, @@ -267,41 +278,76 @@ class Subscriber(LetsMeshNormalizer): public_key: str, event_type: str, payload: dict[str, Any], - ) -> None: + ) -> str | None: """Persist a raw_packets row for a packets-feed reception. Reuses the decode the normalizer already performed (the decoder caches per raw hex, so this ``decode_payload`` call is a cache hit). Capture failures are logged and never block event dispatch. + + Returns: + The inserted ``raw_packet.id``, or ``None`` when the topic is + not a packets-feed message or capture fails. The caller uses the + id to backfill ``event_hash`` after dispatch. """ try: parsed_topic = self.mqtt.topic_builder.parse_letsmesh_upload_topic(topic) if not parsed_topic: - return + return None _, feed_type = parsed_topic if feed_type != "packets": - return + return None from meshcore_hub.collector.handlers.raw_packet import store_raw_packet decoded_packet = self._letsmesh_decoder.decode_payload(payload) - store_raw_packet(public_key, payload, decoded_packet, event_type, self.db) + return store_raw_packet( + public_key, payload, decoded_packet, event_type, self.db + ) except Exception as e: logger.error("Error capturing raw packet: %s", e) + return None + + def _backfill_raw_packet_event_hash( + self, + raw_packet_id: str, + event_hash: str, + ) -> None: + """Propagate ``event_hash`` onto a captured raw packet and its hops. + + Runs after structured dispatch resolves the underlying event. Two + single-row UPDATEs scoped by ``raw_packet_id``; failures are logged + and swallowed so they never affect subsequent ingest. + """ + try: + from meshcore_hub.collector.handlers.raw_packet import ( + update_raw_packet_event_hash, + ) + + update_raw_packet_event_hash(raw_packet_id, event_hash, self.db) + except Exception as e: + logger.error("Error backfilling raw packet event_hash: %s", e) def _dispatch_event( self, public_key: str, event_type: str, payload: dict[str, Any], - ) -> None: - """Route a normalized event to the appropriate handler.""" + ) -> str | None: + """Route a normalized event to the appropriate handler. + + Returns: + The ``event_hash`` returned by the structured handler (used by + the caller to denormalize the identity onto the captured raw + packet), or ``None`` when no handler ran or it returned None. + """ + event_hash: str | None = None # Find and call handler handler = self._handlers.get(event_type) if handler: try: - handler(public_key, event_type, payload, self.db) + event_hash = handler(public_key, event_type, payload, self.db) except Exception as e: logger.error(f"Error handling {event_type}: {e}") else: @@ -317,6 +363,8 @@ class Subscriber(LetsMeshNormalizer): if self._webhook_dispatcher and self._webhook_dispatcher.webhooks: self._queue_webhook_event(event_type, payload, public_key) + return event_hash + def _queue_webhook_event( self, event_type: str, payload: dict[str, Any], public_key: str ) -> None: diff --git a/src/meshcore_hub/common/config.py b/src/meshcore_hub/common/config.py index b086ecb..0f2e93b 100644 --- a/src/meshcore_hub/common/config.py +++ b/src/meshcore_hub/common/config.py @@ -418,8 +418,25 @@ class APISettings(CommonSettings): description="Default cache TTL in seconds", ) redis_cache_ttl_dashboard: int = Field( - default=30, - description="Cache TTL for dashboard endpoints (seconds)", + default=300, + description=( + "Cache TTL in seconds for dashboard endpoints and per-route " + "health history (trend/aggregation data tolerates longer staleness)" + ), + ) + redis_cache_ttl_route_detail: int = Field( + default=300, + description="Cache TTL for /routes/{id} detail endpoint (seconds)", + ) + + # HTTP Cache-Control headers + api_cache_control_enabled: bool = Field( + default=True, + description=( + "Emit HTTP Cache-Control headers on /api/v1/* responses and ETag / " + "If-None-Match handling on @cached endpoints. Disable to suppress " + "all client-side caching directives." + ), ) diff --git a/src/meshcore_hub/common/models/packet_path_hop.py b/src/meshcore_hub/common/models/packet_path_hop.py index 52bcd74..4143bac 100644 --- a/src/meshcore_hub/common/models/packet_path_hop.py +++ b/src/meshcore_hub/common/models/packet_path_hop.py @@ -27,6 +27,12 @@ class PacketPathHop(Base, UUIDMixin, TimestampMixin): position: Zero-based hop position in the ordered path node_hash: Normalized (uppercase) hex prefix for this hop packet_hash: Denormalized packet hash from raw_packets + event_hash: Denormalized identity of the underlying structured event + from raw_packets. The route evaluator prefers ``event_hash`` over + ``packet_hash`` when deduplicating matches, so retransmissions + of the same underlying advert/message/telemetry/trace count once + instead of once per on-air copy. ``NULL`` for legacy rows and + unclassified packets; the evaluator falls back to ``packet_hash``. received_at: Denormalized reception timestamp from raw_packets observer_node_id: Denormalized observer node FK """ @@ -49,6 +55,10 @@ class PacketPathHop(Base, UUIDMixin, TimestampMixin): String(32), nullable=True, ) + event_hash: Mapped[Optional[str]] = mapped_column( + String(32), + nullable=True, + ) received_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, @@ -73,6 +83,11 @@ class PacketPathHop(Base, UUIDMixin, TimestampMixin): "ix_packet_path_hops_received_at", "received_at", ), + Index( + "ix_packet_path_hops_event_hash_received_at", + "event_hash", + "received_at", + ), ) def __repr__(self) -> str: diff --git a/src/meshcore_hub/common/models/raw_packet.py b/src/meshcore_hub/common/models/raw_packet.py index 7247077..8dc0abe 100644 --- a/src/meshcore_hub/common/models/raw_packet.py +++ b/src/meshcore_hub/common/models/raw_packet.py @@ -22,6 +22,14 @@ class RawPacket(Base, UUIDMixin, TimestampMixin): observer_node_id: FK to nodes (the receiving interface) packet_hash: LetsMesh packet hash, links rows to structured records and groups multi-observer receptions + event_hash: Denormalized identity of the underlying structured event + (advertisement / message / telemetry / trace) computed by the + structured handler. Populated after dispatch by the subscriber. + ``NULL`` for packets that did not trigger a structured handler and + for rows captured before this column was added. The route + evaluator prefers ``event_hash`` over ``packet_hash`` when + deduplicating matches, so retransmissions of the same underlying + event count once instead of once per on-air copy. raw_hex: The on-air bytes from ``payload["raw"]`` packet_type: Wire packet type payload_type: Decoder payload type @@ -51,6 +59,11 @@ class RawPacket(Base, UUIDMixin, TimestampMixin): nullable=True, index=True, ) + event_hash: Mapped[Optional[str]] = mapped_column( + String(32), + nullable=True, + index=True, + ) raw_hex: Mapped[Optional[str]] = mapped_column( Text, nullable=True, diff --git a/src/meshcore_hub/web/static/js/spa/pages/routes.js b/src/meshcore_hub/web/static/js/spa/pages/routes.js index 3028db4..3f2084e 100644 --- a/src/meshcore_hub/web/static/js/spa/pages/routes.js +++ b/src/meshcore_hub/web/static/js/spa/pages/routes.js @@ -200,15 +200,41 @@ function renderDetailContent(route, detail, { navigate, packetsEnabled, history const detailUrl = packetDetailUrl(m.packet_hash); return html`
{ e.stopPropagation(); navigate(detailUrl); } : undefined}> - ${(m.hops || []).map((h, i) => { - const rn = pathLookup.get((h.node_hash || '').toLowerCase().slice(0, prefixLen)); - return html` - ${i > 0 ? html`\u2192` : nothing} - ${rn - ? html`${(h.node_hash || '').toLowerCase()}` - : html`${(h.node_hash || '').toLowerCase()}`} - `; - })} + ${(() => { + const hops = m.hops || []; + const PATH_MAX = 5; + const PATH_HEAD = 2; + const PATH_TAIL = 2; + let entries; + if (hops.length > PATH_MAX) { + const hidden = hops.length - PATH_HEAD - PATH_TAIL; + const head = hops.slice(0, PATH_HEAD); + const tail = hops.slice(-PATH_TAIL); + const ellipsis = html``; + entries = [ + ...head.map(h => [h, true]), + [ellipsis, false], + ...tail.map(h => [h, true]), + ]; + } else { + entries = hops.map(h => [h, true]); + } + return entries.map(([h, isHop], i) => { + if (!isHop) { + return html` + ${i > 0 ? html`\u2192` : nothing} + ${h} + `; + } + const rn = pathLookup.get((h.node_hash || '').toLowerCase().slice(0, prefixLen)); + return html` + ${i > 0 ? html`\u2192` : nothing} + ${rn + ? html`${(h.node_hash || '').toLowerCase()}` + : html`${(h.node_hash || '').toLowerCase()}`} + `; + }); + })()} ${m.received_at ? html`${new Date(m.received_at).toLocaleString()}` : nothing}
`; })} diff --git a/tests/test_api/test_cache.py b/tests/test_api/test_cache.py index 6bc9d10..80baaf1 100644 --- a/tests/test_api/test_cache.py +++ b/tests/test_api/test_cache.py @@ -430,6 +430,33 @@ class TestCachedDecorator: call_args = mock_cache.set.call_args assert call_args[0][2] == 60 # TTL should be 60, not 30 + async def test_route_detail_ttl_override(self): + app = FastAPI() + mock_cache = MagicMock() + mock_cache.get.return_value = None + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + app.state.redis_cache_ttl_route_detail = 90 + + @cached("routes/{id}", ttl_setting="redis_cache_ttl_route_detail") + async def handler(request: Request): + return {"id": "abc", "matches": []} + + scope = { + "type": "http", + "query_string": b"", + "headers": [], + "app": app, + } + from starlette.datastructures import State + + request = Request(scope) + request._state = State() + + await handler(request=request) + call_args = mock_cache.set.call_args + assert call_args[0][2] == 90 # TTL should be 90, not 30 + async def test_serializes_pydantic_model_result(self): app = FastAPI() mock_cache = MagicMock() @@ -455,8 +482,10 @@ class TestCachedDecorator: result = await handler(request=request) assert result.items == [1, 2] set_call = mock_cache.set.call_args - stored = json.loads(set_call[0][1]) - assert stored == {"items": [1, 2], "total": 2} + envelope = json.loads(set_call[0][1]) + assert envelope["body"] == {"items": [1, 2], "total": 2} + assert isinstance(envelope["etag"], str) + assert envelope["etag"].startswith('"') async def test_serializes_dict_result(self): app = FastAPI() @@ -483,7 +512,9 @@ class TestCachedDecorator: result = await handler(request=request) assert result == {"key": "value"} set_call = mock_cache.set.call_args - assert set_call[0][1] == '{"key": "value"}' + envelope = json.loads(set_call[0][1]) + assert envelope["body"] == {"key": "value"} + assert isinstance(envelope["etag"], str) async def test_serializes_other_result_with_default_str(self): app = FastAPI() @@ -510,7 +541,8 @@ class TestCachedDecorator: result = await handler(request=request) assert result == ["a", "b"] set_call = mock_cache.set.call_args - assert json.loads(set_call[0][1]) == ["a", "b"] + envelope = json.loads(set_call[0][1]) + assert envelope["body"] == ["a", "b"] async def test_cache_set_error_falls_through(self): app = FastAPI() @@ -560,6 +592,394 @@ class TestCachedDecorator: assert result == {"items": ["direct"]} +class TestCachedEtag: + """ETag / If-None-Match behavior of @cached.""" + + @staticmethod + def _make_request(app, headers=None): + scope = { + "type": "http", + "query_string": b"", + "headers": [(k.lower().encode(), v.encode()) for k, v in (headers or [])], + "app": app, + } + from starlette.datastructures import State + + request = Request(scope) + request._state = State() + return request + + async def test_miss_sets_request_state_etag(self): + app = FastAPI() + mock_cache = MagicMock() + mock_cache.get.return_value = None + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["a"], "total": 1} + + request = self._make_request(app) + result = await handler(request=request) + assert result == {"items": ["a"], "total": 1} + etag = request.state.api_etag + assert isinstance(etag, str) + assert etag.startswith('"') + + async def test_hit_reads_etag_from_envelope(self): + app = FastAPI() + mock_cache = MagicMock() + body = {"items": [], "total": 0} + etag = '"deadbeefdeadbeefdeadbeefdeadbeef"' + mock_cache.get.return_value = json.dumps({"body": body, "etag": etag}) + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["should not appear"], "total": 1} + + request = self._make_request(app) + result = await handler(request=request) + assert result == body + assert request.state.api_etag == etag + assert request.state.cache_status == "HIT" + + async def test_hit_computes_etag_for_legacy_entry(self): + """Legacy bare-JSON cache entries (no etag envelope) still serve and + get an ETag computed on the fly.""" + app = FastAPI() + mock_cache = MagicMock() + body = {"items": [], "total": 0} + mock_cache.get.return_value = json.dumps(body) + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["should not appear"], "total": 1} + + request = self._make_request(app) + result = await handler(request=request) + assert result == body + assert request.state.cache_status == "HIT" + # ETag is computed deterministically from the legacy body bytes. + assert request.state.api_etag.startswith('"') + + async def test_garbage_cache_entry_treated_as_miss(self): + app = FastAPI() + mock_cache = MagicMock() + mock_cache.get.return_value = "not valid json {" + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["real"], "total": 1} + + request = self._make_request(app) + result = await handler(request=request) + assert result == {"items": ["real"], "total": 1} + assert request.state.cache_status == "MISS" + + async def test_if_none_match_returns_304_on_miss(self): + app = FastAPI() + mock_cache = MagicMock() + mock_cache.get.return_value = None + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["a"], "total": 1} + + # First call to learn the ETag. + request = self._make_request(app) + await handler(request=request) + etag = request.state.api_etag + + # Second call with matching If-None-Match. + request2 = self._make_request(app, headers=[("if-none-match", etag)]) + response = await handler(request=request2) + from fastapi.responses import Response + + assert isinstance(response, Response) + assert response.status_code == 304 + assert response.headers["ETag"] == etag + + async def test_if_none_match_returns_304_on_hit(self): + app = FastAPI() + mock_cache = MagicMock() + body = {"items": [], "total": 0} + etag = '"abc123abc123abc123abc123abc123ab"' + mock_cache.get.return_value = json.dumps({"body": body, "etag": etag}) + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["should not appear"], "total": 1} + + request = self._make_request(app, headers=[("if-none-match", etag)]) + response = await handler(request=request) + from fastapi.responses import Response + + assert isinstance(response, Response) + assert response.status_code == 304 + assert response.headers["ETag"] == etag + + async def test_if_none_match_non_matching_returns_body(self): + app = FastAPI() + mock_cache = MagicMock() + body = {"items": [], "total": 0} + etag = '"abc123abc123abc123abc123abc123ab"' + mock_cache.get.return_value = json.dumps({"body": body, "etag": etag}) + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["should not appear"], "total": 1} + + request = self._make_request( + app, headers=[("if-none-match", '"different-etag-value"')] + ) + result = await handler(request=request) + assert result == body + + async def test_if_none_match_wildcard_matches_any_etag(self): + app = FastAPI() + mock_cache = MagicMock() + body = {"items": [], "total": 0} + etag = '"abc123abc123abc123abc123abc123ab"' + mock_cache.get.return_value = json.dumps({"body": body, "etag": etag}) + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["nope"], "total": 1} + + request = self._make_request(app, headers=[("if-none-match", "*")]) + from fastapi.responses import Response + + response = await handler(request=request) + assert isinstance(response, Response) + assert response.status_code == 304 + + async def test_if_none_match_accepts_weak_indicator(self): + app = FastAPI() + mock_cache = MagicMock() + body = {"items": [], "total": 0} + etag = '"abc123abc123abc123abc123abc123ab"' + mock_cache.get.return_value = json.dumps({"body": body, "etag": etag}) + app.state.redis_cache = mock_cache + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": ["nope"], "total": 1} + + request = self._make_request(app, headers=[("if-none-match", f"W/{etag}")]) + from fastapi.responses import Response + + response = await handler(request=request) + assert isinstance(response, Response) + assert response.status_code == 304 + + async def test_cache_control_ttl_always_set(self): + """Even when Redis is NullCache, request.state.cache_control_ttl + is set so the middleware can emit Cache-Control.""" + app = FastAPI() + app.state.redis_cache = NullCache() + app.state.redis_cache_ttl = 30 + + @cached("nodes") + async def handler(request: Request): + return {"items": [], "total": 0} + + request = self._make_request(app) + await handler(request=request) + assert request.state.cache_control_ttl == 30 + + async def test_cache_control_ttl_uses_overridden_setting(self): + app = FastAPI() + app.state.redis_cache = NullCache() + app.state.redis_cache_ttl = 30 + app.state.redis_cache_ttl_route_detail = 90 + + @cached("routes/{id}", ttl_setting="redis_cache_ttl_route_detail") + async def handler(request: Request): + return {"id": "abc"} + + request = self._make_request(app) + await handler(request=request) + assert request.state.cache_control_ttl == 90 + + +class TestCachedEtagHelpers: + """Direct unit tests for the etag helpers.""" + + def test_compute_etag_is_quoted_hex(self): + from meshcore_hub.api.cache import _compute_etag + + etag = _compute_etag('{"a": 1}') + assert etag.startswith('"') + assert etag.endswith('"') + assert len(etag) == 34 # 32 hex chars + 2 quotes + + def test_compute_etag_is_deterministic(self): + from meshcore_hub.api.cache import _compute_etag + + assert _compute_etag("body") == _compute_etag("body") + + def test_compute_etag_changes_with_input(self): + from meshcore_hub.api.cache import _compute_etag + + assert _compute_etag("body1") != _compute_etag("body2") + + def test_etag_matches_strong_equality(self): + from meshcore_hub.api.cache import _etag_matches + + etag = '"abc123"' + assert _etag_matches('"abc123"', etag) + + def test_etag_matches_wildcard(self): + from meshcore_hub.api.cache import _etag_matches + + assert _etag_matches("*", '"any"') + + def test_etag_matches_weak_indicator(self): + from meshcore_hub.api.cache import _etag_matches + + etag = '"abc123"' + assert _etag_matches('W/"abc123"', etag) + + def test_etag_matches_one_of_list(self): + from meshcore_hub.api.cache import _etag_matches + + etag = '"abc123"' + assert _etag_matches('"xyz", "abc123", "def"', etag) + + def test_etag_no_match(self): + from meshcore_hub.api.cache import _etag_matches + + assert not _etag_matches('"other"', '"abc123"') + + +class TestCacheControlMiddleware: + """End-to-end Cache-Control / ETag tests via the FastAPI test client.""" + + def test_cached_get_emits_cache_control_and_etag(self, client_no_auth): + mock_cache = MagicMock() + mock_cache.get.return_value = None + client_no_auth.app.state.redis_cache = mock_cache + client_no_auth.app.state.redis_cache_ttl = 30 + response = client_no_auth.get("/api/v1/nodes") + assert response.status_code == 200 + assert response.headers["cache-control"] == "private, max-age=30" + assert "etag" in response.headers + + def test_cached_get_cache_control_uses_overridden_ttl(self, client_no_auth): + """Route detail endpoint should use the route-detail TTL (300s).""" + mock_cache = MagicMock() + mock_cache.get.return_value = None + client_no_auth.app.state.redis_cache = mock_cache + client_no_auth.app.state.redis_cache_ttl = 30 + client_no_auth.app.state.redis_cache_ttl_route_detail = 300 + # Need a route to exist; this is a happy-path check of the header value + # so we stub the cache and just hit the endpoint with a fake id. The + # endpoint will return 404 but still flow through the @cached decorator + # and the middleware. + response = client_no_auth.get("/api/v1/routes/nonexistent-id") + # 404 is fine; we only care about the header applied by the decorator + # via request.state.cache_control_ttl. + assert response.headers.get("cache-control") == "private, max-age=300" + + def test_cached_get_304_on_matching_if_none_match(self, client_no_auth): + mock_cache = MagicMock() + body = { + "items": [], + "total": 0, + "limit": 50, + "offset": 0, + } + etag = '"abc123abc123abc123abc123abc123ab"' + mock_cache.get.return_value = json.dumps({"body": body, "etag": etag}) + client_no_auth.app.state.redis_cache = mock_cache + client_no_auth.app.state.redis_cache_ttl = 30 + + response = client_no_auth.get("/api/v1/nodes", headers={"If-None-Match": etag}) + assert response.status_code == 304 + assert response.headers["ETag"] == etag + assert response.headers["cache-control"] == "private, max-age=30" + assert response.headers["x-cache"] == "HIT" + # 304 must not carry a body. + assert response.content in (b"", b"null") + + def test_uncached_get_emits_must_revalidate(self, client_no_auth, sample_node): + """Uncached GET detail endpoints get max-age=0, must-revalidate.""" + # Force the @cached list endpoint to NOT be the target by hitting the + # per-id endpoint, which is not cached. + if hasattr(client_no_auth.app.state, "redis_cache"): + del client_no_auth.app.state.redis_cache + response = client_no_auth.get(f"/api/v1/nodes/{sample_node.public_key}") + assert response.status_code == 200 + assert ( + response.headers["cache-control"] == "private, max-age=0, must-revalidate" + ) + + def test_post_emits_no_store(self, client_no_auth, api_db_session): + """POST endpoints always get Cache-Control: no-store. + + Depends on ``api_db_session`` so its teardown truncates the channel + row even if the create succeeds (channel creates mutate the DB and + would otherwise leak into later tests in the same module). + """ + response = client_no_auth.post( + "/api/v1/channels", + json={ + "name": "CacheControlTestChan", + "key_hex": "AABBCCDDEEFF00112233445566778899", + "visibility": "community", + }, + ) + # May succeed (201) or fail (400/500) depending on validation; we + # only care that the middleware set no-store on the response. + assert response.headers["cache-control"] == "no-store" + + def test_health_emits_no_store(self, client_no_auth): + response = client_no_auth.get("/health") + assert response.status_code == 200 + assert response.headers["cache-control"] == "no-store" + + def test_health_ready_emits_no_store(self, client_no_auth): + response = client_no_auth.get("/health/ready") + assert response.status_code == 200 + assert response.headers["cache-control"] == "no-store" + + def test_kill_switch_suppresses_cache_control(self, client_no_auth): + """When api_cache_control_enabled is False, no Cache-Control is added.""" + client_no_auth.app.state.api_cache_control_enabled = False + if hasattr(client_no_auth.app.state, "redis_cache"): + del client_no_auth.app.state.redis_cache + response = client_no_auth.get("/api/v1/nodes") + assert "cache-control" not in response.headers + + def test_kill_switch_preserves_x_cache_header(self, client_no_auth): + """X-Cache is observability, not a client-caching directive, so the + kill switch should not suppress it.""" + mock_cache = MagicMock() + mock_cache.get.return_value = None + client_no_auth.app.state.redis_cache = mock_cache + client_no_auth.app.state.redis_cache_ttl = 30 + client_no_auth.app.state.api_cache_control_enabled = False + response = client_no_auth.get("/api/v1/nodes") + assert response.headers.get("x-cache") == "MISS" + assert "cache-control" not in response.headers + + class TestLifespanRedis: async def test_lifespan_creates_null_cache_when_disabled(self): import meshcore_hub.api.app as app_module @@ -750,6 +1170,9 @@ class TestCliRedis: "60", "--redis-cache-ttl-dashboard", "120", + "--redis-cache-ttl-route-detail", + "90", + "--no-api-cache-control", ], catch_exceptions=False, ) @@ -762,6 +1185,50 @@ class TestCliRedis: assert call_kwargs["redis_key_prefix"] == "pre" assert call_kwargs["redis_cache_ttl"] == 60 assert call_kwargs["redis_cache_ttl_dashboard"] == 120 + assert call_kwargs["redis_cache_ttl_route_detail"] == 90 + assert call_kwargs["api_cache_control_enabled"] is False + + def test_api_cache_control_enabled_default(self): + """Without --no-api-cache-control, the flag defaults to True.""" + from click.testing import CliRunner + + from meshcore_hub.api.cli import api + + runner = CliRunner() + with patch("uvicorn.run"): + with patch("meshcore_hub.common.config.get_api_settings") as mock_settings: + mock_settings.return_value = MagicMock( + data_home="/tmp/test", + effective_database_url="sqlite:///test.db", + ) + with patch("meshcore_hub.api.app.create_app") as mock_create_app: + mock_create_app.return_value = MagicMock() + runner.invoke(api, catch_exceptions=False) + call_kwargs = mock_create_app.call_args[1] + assert call_kwargs["api_cache_control_enabled"] is True + + def test_redis_cache_ttl_dashboard_default(self): + """--redis-cache-ttl-dashboard defaults to 300 (raised from 30). + + Covers /dashboard/* endpoints and /routes/{id}/history. Trend data + tolerates longer staleness than the original 30s default. + """ + from click.testing import CliRunner + + from meshcore_hub.api.cli import api + + runner = CliRunner() + with patch("uvicorn.run"): + with patch("meshcore_hub.common.config.get_api_settings") as mock_settings: + mock_settings.return_value = MagicMock( + data_home="/tmp/test", + effective_database_url="sqlite:///test.db", + ) + with patch("meshcore_hub.api.app.create_app") as mock_create_app: + mock_create_app.return_value = MagicMock() + runner.invoke(api, catch_exceptions=False) + call_kwargs = mock_create_app.call_args[1] + assert call_kwargs["redis_cache_ttl_dashboard"] == 300 class TestKeyBuilders: diff --git a/tests/test_api/test_routes.py b/tests/test_api/test_routes.py index e4e5d07..9d191b4 100644 --- a/tests/test_api/test_routes.py +++ b/tests/test_api/test_routes.py @@ -247,6 +247,86 @@ class TestGetRouteDetail: resp = client_no_auth.get("/api/v1/routes/nonexistent") assert resp.status_code == 404 + def test_detail_response_is_cached(self, client_no_auth, api_db_session): + """Detail endpoint writes its response to the cache after a miss.""" + import json + from unittest.mock import MagicMock + + nodes = _sample_nodes(api_db_session, 2) + route = Route(from_label="Alpha", to_label="Beta") + api_db_session.add(route) + api_db_session.flush() + for pos, n in enumerate(nodes): + api_db_session.add( + RouteNode( + route_id=route.id, + node_id=n.id, + position=pos, + expected_hash=n.public_key[:2].upper(), + ) + ) + api_db_session.commit() + + mock_cache = MagicMock() + mock_cache.get.return_value = None + client_no_auth.app.state.redis_cache = mock_cache + client_no_auth.app.state.redis_cache_ttl_route_detail = 90 + + resp = client_no_auth.get(f"/api/v1/routes/{route.id}") + assert resp.status_code == 200 + assert resp.headers.get("x-cache") == "MISS" + + mock_cache.set.assert_called_once() + cache_key, serialized, ttl = mock_cache.set.call_args[0] + assert cache_key.startswith(f"/api/v1/routes/{route.id}:") + assert "role=anonymous" in cache_key + assert ttl == 90 + envelope = json.loads(serialized) + assert envelope["body"]["from_label"] == "Alpha" + assert isinstance(envelope["etag"], str) + + def test_detail_serves_from_cache_on_hit(self, client_no_auth, api_db_session): + """A second call within the TTL window is served from the cache.""" + nodes = _sample_nodes(api_db_session, 2) + route = Route(from_label="Alpha", to_label="Beta") + api_db_session.add(route) + api_db_session.flush() + for pos, n in enumerate(nodes): + api_db_session.add( + RouteNode( + route_id=route.id, + node_id=n.id, + position=pos, + expected_hash=n.public_key[:2].upper(), + ) + ) + api_db_session.commit() + + store: dict[str, str] = {} + + class _FakeCache: + def get(self, key): + return store.get(key) + + def set(self, key, value, ttl): + store[key] = value + + def ping(self): + return True + + client_no_auth.app.state.redis_cache = _FakeCache() + client_no_auth.app.state.redis_cache_ttl_route_detail = 60 + + first = client_no_auth.get(f"/api/v1/routes/{route.id}") + assert first.status_code == 200 + assert first.headers.get("x-cache") == "MISS" + first_body = first.json() + + second = client_no_auth.get(f"/api/v1/routes/{route.id}") + assert second.status_code == 200 + assert second.headers.get("x-cache") == "HIT" + assert second.json() == first_body + class TestUpdateRoute: def test_update_from_to(self, client_no_auth, api_db_session): diff --git a/tests/test_collector/test_handlers/test_raw_packet.py b/tests/test_collector/test_handlers/test_raw_packet.py index 16a6dce..8dff309 100644 --- a/tests/test_collector/test_handlers/test_raw_packet.py +++ b/tests/test_collector/test_handlers/test_raw_packet.py @@ -2,7 +2,10 @@ from sqlalchemy import select -from meshcore_hub.collector.handlers.raw_packet import store_raw_packet +from meshcore_hub.collector.handlers.raw_packet import ( + store_raw_packet, + update_raw_packet_event_hash, +) from meshcore_hub.common.models import Node, PacketPathHop, RawPacket @@ -28,7 +31,7 @@ class TestStoreRawPacket: """A single packet writes exactly one raw_packets row.""" payload = {"raw": "0011223344", "hash": "deadbeef", "SNR": 9.5, "path": "0102"} - store_raw_packet( + rp_id = store_raw_packet( "a" * 64, payload, _channel_decoded(), "channel_msg_recv", db_manager ) @@ -39,6 +42,7 @@ class TestStoreRawPacket: assert rp.packet_hash == "deadbeef" assert rp.event_type == "channel_msg_recv" assert rp.snr == 9.5 + assert rp_id == rp.id def test_derives_channel_idx_and_source_prefix(self, db_manager, db_session): """channel_idx from channelHash; source prefix from sourceHash.""" @@ -309,3 +313,69 @@ class TestStoreRawPacketPathHops: ) assert len(hops) == 2 assert [h.node_hash for h in hops] == ["AABB", "CCDD"] + + +class TestUpdateRawPacketEventHash: + """Tests for the post-dispatch event_hash backfill.""" + + def test_backfills_raw_packet_and_hops(self, db_manager, db_session): + """update_raw_packet_event_hash writes event_hash onto both rows.""" + decoded = { + "payloadType": 1, + "path": ["aa", "bb"], + "payload": {"decoded": {}}, + } + rp_id = store_raw_packet( + "a" * 64, {"raw": "00", "hash": "wire1"}, decoded, "flood", db_manager + ) + assert rp_id is not None + + update_raw_packet_event_hash(rp_id, "evt-aaa", db_manager) + + rp = db_session.execute(select(RawPacket)).scalar_one() + db_session.refresh(rp) + assert rp.event_hash == "evt-aaa" + + hops = db_session.execute(select(PacketPathHop)).scalars().all() + assert len(hops) == 2 + for hop in hops: + db_session.refresh(hop) + assert hop.event_hash == "evt-aaa" + + def test_backfill_idempotent_on_replay(self, db_manager, db_session): + """Calling backfill twice with the same hash is a no-op.""" + decoded = {"payloadType": 1, "path": ["aa"], "payload": {"decoded": {}}} + rp_id = store_raw_packet( + "a" * 64, {"raw": "00", "hash": "wire1"}, decoded, "flood", db_manager + ) + assert rp_id is not None + + update_raw_packet_event_hash(rp_id, "evt-aaa", db_manager) + update_raw_packet_event_hash(rp_id, "evt-aaa", db_manager) + + rp = db_session.execute(select(RawPacket)).scalar_one() + db_session.refresh(rp) + assert rp.event_hash == "evt-aaa" + + def test_backfill_overwrites_prior_value(self, db_manager, db_session): + """A second backfill with a different hash replaces the prior one.""" + decoded = {"payloadType": 1, "path": ["aa"], "payload": {"decoded": {}}} + rp_id = store_raw_packet( + "a" * 64, {"raw": "00", "hash": "wire1"}, decoded, "flood", db_manager + ) + assert rp_id is not None + + update_raw_packet_event_hash(rp_id, "evt-first", db_manager) + update_raw_packet_event_hash(rp_id, "evt-second", db_manager) + + rp = db_session.execute(select(RawPacket)).scalar_one() + db_session.refresh(rp) + assert rp.event_hash == "evt-second" + + def test_backfill_no_op_when_row_missing(self, db_manager, db_session): + """Backfilling a non-existent id silently does nothing (retention + cleanup may have removed the row between capture and dispatch).""" + # No row created; just verify it doesn't raise. + update_raw_packet_event_hash("nonexistent-id", "evt-aaa", db_manager) + + assert db_session.execute(select(RawPacket)).scalars().all() == [] diff --git a/tests/test_collector/test_routes.py b/tests/test_collector/test_routes.py index 57bd4a5..3bb2e9c 100644 --- a/tests/test_collector/test_routes.py +++ b/tests/test_collector/test_routes.py @@ -51,6 +51,7 @@ def _make_reception( packet_hash: str, path_hashes: list[str], received_at: datetime | None = None, + event_hash: str | None = None, ) -> str: """Insert a RawPacket + PacketPathHop rows for a test reception.""" ts = received_at or _NOW @@ -59,6 +60,7 @@ def _make_reception( id=rp_id, observer_node_id=observer_node_id, packet_hash=packet_hash, + event_hash=event_hash, received_at=ts, ) db_session.add(rp) @@ -70,6 +72,7 @@ def _make_reception( position=pos, node_hash=nh, packet_hash=packet_hash, + event_hash=event_hash, received_at=ts, observer_node_id=observer_node_id, ) @@ -326,6 +329,92 @@ class TestEvaluateRoute: assert count == 1 assert state == RouteState.HEALTHY.value + def test_retransmissions_dedup_by_event_hash(self, db_session): + """Three retransmissions of the same underlying event count once. + + Each reception has a distinct wire ``packet_hash`` (the LetsMesh + ``payload["hash"]`` is unique per transmission) but shares the same + underlying ``event_hash``. Without the event_hash dedup a single + advert broadcast three times would satisfy ``threshold=3`` within + seconds and bias the route's health. + """ + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b], threshold=3) + + _make_reception( + db_session, None, "wire1", ["AA", "BB"], event_hash="evt-shared" + ) + _make_reception( + db_session, None, "wire2", ["AA", "BB"], event_hash="evt-shared" + ) + _make_reception( + db_session, None, "wire3", ["AA", "BB"], event_hash="evt-shared" + ) + db_session.commit() + + since = _NOW - timedelta(hours=24) + state, quality, count = evaluate_route(db_session, route, since) + assert count == 1 + assert state == RouteState.UNHEALTHY.value + + def test_distinct_event_hashes_count_separately(self, db_session): + """Two underlying events (distinct event_hash) count as two matches.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b], threshold=2) + + _make_reception(db_session, None, "wire1", ["AA", "BB"], event_hash="evt-aaa") + _make_reception(db_session, None, "wire2", ["AA", "BB"], event_hash="evt-bbb") + db_session.commit() + + since = _NOW - timedelta(hours=24) + state, _, count = evaluate_route(db_session, route, since) + assert count == 2 + assert state == RouteState.HEALTHY.value + + def test_null_event_hash_falls_back_to_packet_hash(self, db_session): + """Legacy rows with NULL event_hash dedup by wire packet_hash only.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b], threshold=3) + + # Three distinct wire hashes, no event_hash (legacy/NULL). + _make_reception(db_session, None, "wire1", ["AA", "BB"]) + _make_reception(db_session, None, "wire2", ["AA", "BB"]) + _make_reception(db_session, None, "wire3", ["AA", "BB"]) + db_session.commit() + + since = _NOW - timedelta(hours=24) + state, _, count = evaluate_route(db_session, route, since) + assert count == 3 + assert state == RouteState.HEALTHY.value + + def test_mixed_event_and_wire_identities(self, db_session): + """Mix of NULL and shared event_hash: each NULL is unique, shared + event_hash collapses to one.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b], threshold=3) + + # Two legacy rows (distinct wire hashes, NULL event_hash) → 2 matches. + _make_reception(db_session, None, "wire1", ["AA", "BB"]) + _make_reception(db_session, None, "wire2", ["AA", "BB"]) + # Three retransmissions of one event → 1 match. + for i in range(3): + _make_reception( + db_session, + None, + f"wire-shared-{i}", + ["AA", "BB"], + event_hash="evt-shared", + ) + db_session.commit() + + since = _NOW - timedelta(hours=24) + _, _, count = evaluate_route(db_session, route, since) + assert count == 3 # 2 legacy + 1 collapsed event + def test_observer_scope_filter(self, db_session): """Only in-scope observers are considered.""" node_a = _make_node(db_session, "aa" + "0" * 62) @@ -497,6 +586,84 @@ class TestRecentMatches: hops = matches[0]["hops"] assert [h["node_hash"] for h in hops] == ["BB", "YY", "AA"] + def test_dedup_by_event_hash_keeps_newest(self, db_session): + """Multiple retransmissions of one event return one row, newest first. + + Without this dedup, the recent-matches list showed three rows with + near-identical timestamps whenever a single advert traversed the + route multiple times — the symptom that surfaced this bug. + """ + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b]) + + # Three retransmissions of one event with monotonic timestamps. + base = _NOW - timedelta(hours=5) + for i in range(3): + _make_reception( + db_session, + None, + f"wire{i}", + ["AA", "BB"], + received_at=base + timedelta(minutes=i), + event_hash="evt-shared", + ) + db_session.commit() + + matches = recent_matches(db_session, route, limit=3, now=_NOW) + assert len(matches) == 1 + # The newest reception of the shared event is the one returned. + assert matches[0]["packet_hash"] == "wire2" + assert matches[0]["event_hash"] == "evt-shared" + + def test_distinct_events_returned_in_newest_first_order(self, db_session): + """Two distinct events surface as two rows, newest first.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b]) + + _make_reception( + db_session, + None, + "wire-older", + ["AA", "BB"], + received_at=_NOW - timedelta(hours=2), + event_hash="evt-older", + ) + _make_reception( + db_session, + None, + "wire-newer", + ["AA", "BB"], + received_at=_NOW - timedelta(hours=1), + event_hash="evt-newer", + ) + db_session.commit() + + matches = recent_matches(db_session, route, limit=3, now=_NOW) + assert len(matches) == 2 + assert matches[0]["event_hash"] == "evt-newer" + assert matches[1]["event_hash"] == "evt-older" + + def test_legacy_null_event_hash_one_row_per_wire_hash(self, db_session): + """NULL event_hash rows preserve today's per-wire-hash behaviour.""" + node_a = _make_node(db_session, "aa" + "0" * 62) + node_b = _make_node(db_session, "bb" + "0" * 62) + route = _make_route(db_session, "R1", [node_a, node_b]) + + for i in range(3): + _make_reception( + db_session, + None, + f"wire{i}", + ["AA", "BB"], + received_at=_NOW - timedelta(hours=i), + ) + db_session.commit() + + matches = recent_matches(db_session, route, limit=3, now=_NOW) + assert len(matches) == 3 + class TestPreviewRoute: def test_normal_preview(self, db_session): diff --git a/tests/test_collector/test_subscriber.py b/tests/test_collector/test_subscriber.py index 4deda9b..1882651 100644 --- a/tests/test_collector/test_subscriber.py +++ b/tests/test_collector/test_subscriber.py @@ -154,6 +154,136 @@ class TestSubscriber: finally: session.close() + def test_event_hash_backfilled_after_dispatch(self, mock_mqtt_client, db_manager): + """When the structured handler returns an event_hash, the subscriber + propagates it onto the captured raw_packet (and its hops).""" + from sqlalchemy import select + + from meshcore_hub.common.models import PacketPathHop, RawPacket + + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "a" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, db_manager, raw_packet_capture_enabled=True + ) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("a" * 64, "channel_msg_recv", {"text": "hi"}) + ) + subscriber._letsmesh_decoder.decode_payload = MagicMock( # type: ignore[method-assign] + return_value={ + "payloadType": 4, + "path": ["aa", "bb"], + "payload": {"decoded": {}}, + } + ) + # Structured handler resolves the underlying event and returns its hash. + subscriber.register_handler( + "channel_msg_recv", lambda pk, et, payload, db: "evt-resolved" + ) + + subscriber._handle_mqtt_message( + topic="meshcore/STN/abc/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "0011", "hash": "wire1"}, + ) + + session = db_manager.get_session() + try: + rp = session.execute(select(RawPacket)).scalar_one() + session.refresh(rp) + assert rp.event_hash == "evt-resolved" + + hops = session.execute(select(PacketPathHop)).scalars().all() + assert len(hops) == 2 + for hop in hops: + session.refresh(hop) + assert hop.event_hash == "evt-resolved" + finally: + session.close() + + def test_event_hash_skipped_when_handler_returns_none( + self, mock_mqtt_client, db_manager + ): + """A handler that returns None (or no handler) leaves event_hash NULL.""" + from sqlalchemy import select + + from meshcore_hub.common.models import RawPacket + + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "a" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, db_manager, raw_packet_capture_enabled=True + ) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("a" * 64, "channel_msg_recv", {"text": "hi"}) + ) + subscriber._letsmesh_decoder.decode_payload = MagicMock( # type: ignore[method-assign] + return_value={"payloadType": 4, "payload": {"decoded": {}}} + ) + # Handler returns None (event unresolvable / unclassified). + subscriber.register_handler( + "channel_msg_recv", lambda pk, et, payload, db: None + ) + + subscriber._handle_mqtt_message( + topic="meshcore/STN/abc/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "0011", "hash": "wire1"}, + ) + + session = db_manager.get_session() + try: + rp = session.execute(select(RawPacket)).scalar_one() + session.refresh(rp) + assert rp.event_hash is None + finally: + session.close() + + def test_event_hash_skipped_when_handler_raises(self, mock_mqtt_client, db_manager): + """A throwing handler leaves event_hash NULL but the raw_packet is + still captured (capture happens before dispatch).""" + from sqlalchemy import select + + from meshcore_hub.common.models import RawPacket + + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "a" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, db_manager, raw_packet_capture_enabled=True + ) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("a" * 64, "channel_msg_recv", {"text": "hi"}) + ) + subscriber._letsmesh_decoder.decode_payload = MagicMock( # type: ignore[method-assign] + return_value={"payloadType": 4, "payload": {"decoded": {}}} + ) + + def _boom(pk, et, payload, db): + raise RuntimeError("handler failed") + + subscriber.register_handler("channel_msg_recv", _boom) + + subscriber._handle_mqtt_message( + topic="meshcore/STN/abc/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "0011", "hash": "wire1"}, + ) + + session = db_manager.get_session() + try: + rp = session.execute(select(RawPacket)).scalar_one() + session.refresh(rp) + assert rp.event_hash is None + assert rp.raw_hex == "0011" # capture survived the handler failure + finally: + session.close() + def test_raw_capture_skips_status_feed(self, mock_mqtt_client, db_manager): """Capture is packets-feed only; status feed writes no raw rows.""" from sqlalchemy import select