meshcore-bot/modules/web_viewer/mesh_evidence.py

"""Mesh graph evidence for the web viewer: multi-byte path edges and neighbor-confirmed edges."""

from __future__ import annotations

import time
from datetime import datetime, timedelta, timezone
from typing import Any, NamedTuple


class NeighborEvidenceKeys(NamedTuple):
    """Directed edge identities that zero-hop neighbor discovery has confirmed.

    A ``mesh_connections`` edge counts as neighbor-confirmed if it matches on
    either key space; see BotDataViewer._neighbor_evidence_edge_keys.
    """

    prefixes: set[tuple[str, str]]
    public_keys: set[tuple[str, str]]


class MeshEvidenceMixin:
    """Mixed into BotDataViewer."""

    NEIGHBOR_PREFIX_HEX_CHARS: Any
    _mesh_graph_cache_seconds: Any
    _multibyte_graph_cache_computing: Any
    _multibyte_graph_cache_condition: Any
    _multibyte_graph_cache_created_at: Any
    _multibyte_graph_cache_edges: Any
    _multibyte_graph_cache_failure: Any
    _multibyte_graph_cache_failure_at: Any
    _multibyte_graph_cache_retry_seconds: Any
    _with_db_connection: Any
    logger: Any

    def _derive_multibyte_evidence_edges(
        self,
        days: int | None = None,
        min_observations: int | None = None,
        *,
        force_refresh: bool = False,
    ) -> list[dict[str, Any]]:
        """Return lifetime-derived multi-byte edges filtered for the API view."""
        all_edges = self._aggregate_multibyte_evidence_edges(
            force_refresh=force_refresh
        )
        return self._filter_multibyte_evidence_edges(
            all_edges,
            days=days,
            min_observations=min_observations,
        )

    def _derive_multibyte_evidence_graph(
        self,
        days: int | None = None,
        min_observations: int | None = None,
        *,
        force_refresh: bool = False,
    ) -> tuple[list[dict[str, Any]], int]:
        """Return filtered edges plus the lifetime graph's prefix resolution."""
        all_edges = self._aggregate_multibyte_evidence_edges(
            force_refresh=force_refresh
        )
        prefix_hex_chars = max(
            (len(edge['from_prefix']) for edge in all_edges),
            default=2,
        )
        return (
            self._filter_multibyte_evidence_edges(
                all_edges,
                days=days,
                min_observations=min_observations,
            ),
            max(2, prefix_hex_chars),
        )

    @staticmethod
    def _filter_multibyte_evidence_edges(
        edges: list[dict[str, Any]],
        days: int | None,
        min_observations: int | None,
    ) -> list[dict[str, Any]]:
        """Apply view filters without changing lifetime edge identity or counts."""
        cutoff_naive = datetime.now() - timedelta(days=days) if days is not None else None
        cutoff_utc = (
            datetime.now(timezone.utc) - timedelta(days=days)
            if days is not None
            else None
        )
        result = []
        for edge in edges:
            if cutoff_naive is not None and edge['last_seen']:
                try:
                    last_seen = datetime.fromisoformat(
                        str(edge['last_seen']).replace('Z', '+00:00')
                    )
                except (TypeError, ValueError):
                    # Preserve the historical behavior for malformed timestamps:
                    # they remain visible rather than silently losing graph data.
                    last_seen = None
                if last_seen is not None:
                    if last_seen.tzinfo is None:
                        if last_seen < cutoff_naive:
                            continue
                    elif cutoff_utc is not None and last_seen.astimezone(timezone.utc) < cutoff_utc:
                        continue
            if (
                min_observations is not None
                and edge['observation_count'] < min_observations
            ):
                continue
            result.append(edge)
        return result

    def _aggregate_multibyte_evidence_edges(
        self, *, force_refresh: bool = False
    ) -> list[dict[str, Any]]:
        """Return a bounded-age, single-flight lifetime multi-byte aggregate."""
        def previous_failure(message: str) -> RuntimeError:
            failure = self._multibyte_graph_cache_failure
            if failure is None:
                return RuntimeError(message)
            failure_type, failure_message = failure
            return RuntimeError(
                f"{message}: {failure_type}: {failure_message}"
            )

        now = time.monotonic()
        with self._multibyte_graph_cache_condition:
            cached = self._multibyte_graph_cache_edges
            cache_age = now - self._multibyte_graph_cache_created_at
            failure_age = now - self._multibyte_graph_cache_failure_at
            retry_suppressed = (
                self._multibyte_graph_cache_failure is not None
                and failure_age < self._multibyte_graph_cache_retry_seconds
            )
            if retry_suppressed:
                if cached is not None and not force_refresh:
                    return cached
                raise previous_failure(
                    "Multi-byte mesh aggregation retry suppressed after failure"
                )
            if (
                not force_refresh
                and cached is not None
                and cache_age < self._mesh_graph_cache_seconds
            ):
                return cached

            if self._multibyte_graph_cache_computing:
                # Prefer a slightly stale result to making concurrent clients
                # duplicate the same expensive SQLite aggregation.
                if cached is not None and not force_refresh:
                    return cached
                while self._multibyte_graph_cache_computing:
                    self._multibyte_graph_cache_condition.wait()
                cached = self._multibyte_graph_cache_edges
                if self._multibyte_graph_cache_failure is not None:
                    if cached is not None and not force_refresh:
                        return cached
                    raise previous_failure(
                        "Concurrent multi-byte mesh aggregation failed"
                    )
                if cached is not None:
                    return cached

            self._multibyte_graph_cache_computing = True
            stale = cached

        started_at = time.monotonic()
        try:
            computed = self._compute_multibyte_evidence_edges()
        except Exception as exc:
            self.logger.warning(
                "Multi-byte mesh aggregation failed%s",
                "; serving cached data"
                if stale is not None and not force_refresh
                else "",
                exc_info=True,
            )
            with self._multibyte_graph_cache_condition:
                self._multibyte_graph_cache_failure = (
                    type(exc).__name__,
                    str(exc),
                )
                self._multibyte_graph_cache_failure_at = time.monotonic()
                self._multibyte_graph_cache_computing = False
                self._multibyte_graph_cache_condition.notify_all()
            if stale is not None and not force_refresh:
                return stale
            raise

        elapsed = time.monotonic() - started_at
        with self._multibyte_graph_cache_condition:
            self._multibyte_graph_cache_edges = computed
            self._multibyte_graph_cache_created_at = time.monotonic()
            self._multibyte_graph_cache_failure = None
            self._multibyte_graph_cache_failure_at = 0.0
            self._multibyte_graph_cache_computing = False
            self._multibyte_graph_cache_condition.notify_all()

        self.logger.debug(
            "Computed %d multi-byte mesh edges in %.3fs",
            len(computed),
            elapsed,
        )
        return computed

    def _compute_multibyte_evidence_edges(self) -> list[dict[str, Any]]:
        """Derive mesh edges purely from multi-byte path evidence.

        Splits each observed_paths row with bytes_per_hop >= 2 into consecutive
        hop pairs and aggregates per directed pair. Unlike mesh_connections, this
        never mixes in single-byte observations, so edge identity is unambiguous
        (up to 2/3-byte prefix collisions, which are rare).

        Edges observed at 2-byte resolution are coalesced into a 3-byte edge when
        exactly one 3-byte edge prefix-matches both endpoints — the same
        unique-match rule MeshGraph.add_edge applies at write time.

        Returns edge dicts matching the /api/mesh/edges schema, plus:
          path_count — number of distinct observed paths crossing the edge
          evidence   — always 'multibyte'
        """
        # Split paths and aggregate directed hop pairs in SQLite. This preserves
        # lifetime counts and cross-resolution coalescing while avoiding one
        # Python row/dict/list per observed path (hundreds of thousands on busy
        # meshes). The selected timeframe is applied only after coalescing,
        # matching the historical client-side filter semantics.
        query = '''
            WITH RECURSIVE edge_parts(
                path_hex, step, observation_count, first_seen, last_seen,
                hop_position, from_prefix, to_prefix, next_offset
            ) AS (
                SELECT
                    LOWER(path_hex),
                    bytes_per_hop * 2,
                    CASE
                        WHEN observation_count IS NULL OR observation_count = 0 THEN 1
                        ELSE observation_count
                    END,
                    first_seen,
                    last_seen,
                    1,
                    SUBSTR(LOWER(path_hex), 1, bytes_per_hop * 2),
                    SUBSTR(LOWER(path_hex), bytes_per_hop * 2 + 1, bytes_per_hop * 2),
                    bytes_per_hop * 4 + 1
                FROM observed_paths
                WHERE bytes_per_hop >= 2
                  AND path_hex IS NOT NULL
                  AND LENGTH(path_hex) > 0
                  AND LENGTH(path_hex) % (bytes_per_hop * 2) = 0
                  AND LENGTH(path_hex) >= bytes_per_hop * 4

                UNION ALL

                SELECT
                    path_hex,
                    step,
                    observation_count,
                    first_seen,
                    last_seen,
                    hop_position + 1,
                    to_prefix,
                    SUBSTR(path_hex, next_offset, step),
                    next_offset + step
                FROM edge_parts
                WHERE LENGTH(path_hex) >= next_offset + step - 1
            )
            SELECT
                from_prefix,
                to_prefix,
                SUM(observation_count) AS observation_count,
                COUNT(*) AS path_count,
                MIN(first_seen) AS first_seen,
                MAX(last_seen) AS last_seen,
                SUM(hop_position * observation_count) AS hop_position_sum
            FROM edge_parts
            GROUP BY from_prefix, to_prefix
        '''

        with self._with_db_connection() as conn:
            rows = conn.execute(query).fetchall()

        edges: dict[tuple[str, str], dict[str, Any]] = {
            (row['from_prefix'], row['to_prefix']): {
                'observation_count': row['observation_count'],
                'path_count': row['path_count'],
                'first_seen': row['first_seen'],
                'last_seen': row['last_seen'],
                'hop_position_sum': row['hop_position_sum'],
            }
            for row in rows
        }

        # Coalesce 2-byte edges into a 3-byte edge when exactly one matches.
        # (Hops within a path share one resolution, so keys are homogeneous.)
        by_truncated_key: dict[tuple[str, str], list[tuple[str, str]]] = {}
        for key in edges:
            if len(key[0]) == 6:
                by_truncated_key.setdefault((key[0][:4], key[1][:4]), []).append(key)
        for key in [k for k in edges if len(k[0]) == 4]:
            candidates = by_truncated_key.get(key, [])
            if len(candidates) == 1:
                target = edges[candidates[0]]
                source = edges.pop(key)
                target['observation_count'] += source['observation_count']
                target['path_count'] += source['path_count']
                target['hop_position_sum'] += source['hop_position_sum']
                if source['first_seen'] and (target['first_seen'] is None or source['first_seen'] < target['first_seen']):
                    target['first_seen'] = source['first_seen']
                if source['last_seen'] and (target['last_seen'] is None or source['last_seen'] > target['last_seen']):
                    target['last_seen'] = source['last_seen']

        result = []
        for (from_prefix, to_prefix), agg in edges.items():
            result.append({
                'from_prefix': from_prefix,
                'to_prefix': to_prefix,
                'from_public_key': None,
                'to_public_key': None,
                'observation_count': agg['observation_count'],
                'path_count': agg['path_count'],
                'first_seen': agg['first_seen'],
                'last_seen': agg['last_seen'],
                'avg_hop_position': agg['hop_position_sum'] / agg['observation_count'],
                'geographic_distance': None,
                'evidence': 'multibyte',
            })
        result.sort(key=lambda e: e['last_seen'] or '', reverse=True)
        return result

    def _neighbor_evidence_edge_keys(
        self,
        days: int | None = None,
    ) -> NeighborEvidenceKeys:
        """Directed pairs that confirmed zero-hop discovery has proven.

        Used to upgrade the evidence label in the combined view, where the edge
        itself comes from ``mesh_connections`` and so has lost its provenance.
        Two key spaces are returned because a ``mesh_connections`` edge can be
        matched by either:

        * ``prefixes`` — 3-byte prefix pairs, matching edges the graph stores at
          the same resolution neighbor discovery feeds it.
        * ``public_keys`` — full-key pairs, for edges the graph deliberately keeps
          at a *shorter* prefix (see ``MeshGraph.add_edge``: a 1-byte edge with no
          public key is not promoted, so several nodes keep sharing it) while
          still filling in the public keys discovery supplied. Truncating our
          keys down to 2 chars instead would be wrong — it would relabel every
          other node sharing that byte.

        ``days`` windows the evidence the same way the caller windows its edges.
        ``neighbor_links`` is never pruned, so without it a link last seen years
        ago would keep labelling a recent path-derived edge a current neighbor.
        """
        try:
            edges = self._derive_neighbor_evidence_graph(days=days)[0]
        except Exception as exc:
            # A pre-migration-22 database simply has no neighbor evidence.
            self.logger.debug(f"Neighbor evidence keys unavailable: {exc}")
            return NeighborEvidenceKeys(set(), set())

        # Both directions are already emitted per link, so no reversing here.
        prefixes = {
            (edge['from_prefix'], edge['to_prefix'])
            for edge in edges
            if edge['from_prefix'] and edge['to_prefix']
        }
        public_keys = {
            (edge['from_public_key'], edge['to_public_key'])
            for edge in edges
            if edge['from_public_key'] and edge['to_public_key']
        }
        return NeighborEvidenceKeys(prefixes, public_keys)

    def _compute_neighbor_evidence_edges(self) -> list[dict[str, Any]]:
        """Derive mesh edges from confirmed zero-hop neighbor discovery.

        This is the strongest evidence class in the database: each row is a
        direct RF reception between two *full* public keys with a measured SNR,
        recorded by modules/neighbors_discovery.py. Two differences from the
        multi-byte path derivation are worth noting:

        * ``from_public_key``/``to_public_key`` are populated. Path-derived edges
          cannot fill these in, because a path carries prefixes only.
        * ``snr``/``best_snr`` are real measurements. Unlike the dashboard's
          one-hop panel, which withholds SNR unless two sources agree because
          ``complete_contact_tracking.hop_count`` over-claims zero-hop, a
          discover response *is* the authoritative first-party measurement.

        Both directions are emitted per link: a discover response proves we
        transmitted, they received, they transmitted, and we received.
        """
        chars = self.NEIGHBOR_PREFIX_HEX_CHARS
        query = '''
            SELECT
                self_public_key,
                neighbor_public_key,
                observation_count,
                snr_sum,
                snr_count,
                best_snr,
                last_snr,
                first_seen,
                last_seen
            FROM neighbor_links
        '''
        try:
            with self._with_db_connection() as conn:
                rows = conn.execute(query).fetchall()
        except Exception as exc:
            self.logger.debug(f"Neighbor evidence edges unavailable: {exc}")
            return []

        edges: list[dict[str, Any]] = []
        for row in rows:
            self_key = (row['self_public_key'] or '').lower()
            neighbor_key = (row['neighbor_public_key'] or '').lower()
            if not self_key or not neighbor_key:
                continue
            snr_count = row['snr_count'] or 0
            mean_snr = (row['snr_sum'] / snr_count) if snr_count else None
            for from_key, to_key in ((self_key, neighbor_key), (neighbor_key, self_key)):
                edges.append({
                    'from_prefix': from_key[:chars],
                    'to_prefix': to_key[:chars],
                    'from_public_key': from_key,
                    'to_public_key': to_key,
                    'observation_count': row['observation_count'] or 1,
                    'first_seen': row['first_seen'],
                    'last_seen': row['last_seen'],
                    # A direct link is by definition the first hop of any path
                    # that crosses it.
                    'avg_hop_position': 1.0,
                    'geographic_distance': None,
                    'snr': mean_snr,
                    'best_snr': row['best_snr'],
                    'last_snr': row['last_snr'],
                    'evidence': 'neighbors',
                })

        edges.sort(key=lambda e: e['last_seen'] or '', reverse=True)
        return edges

    def _derive_neighbor_evidence_graph(
        self,
        days: int | None = None,
        min_observations: int | None = None,
    ) -> tuple[list[dict[str, Any]], int]:
        """Filtered neighbor-evidence edges plus their prefix resolution.

        Reuses the multi-byte view filter: it only touches ``last_seen`` and
        ``observation_count`` (handling both naive and aware timestamps), which
        is exactly the filtering these edges need.
        """
        all_edges = self._compute_neighbor_evidence_edges()
        filtered = self._filter_multibyte_evidence_edges(
            all_edges, days=days, min_observations=min_observations
        )
        return filtered, self.NEIGHBOR_PREFIX_HEX_CHARS