From e3e7cb26daf6cb8732934f178a924ead013498a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 13 Jun 2026 07:48:40 +0000 Subject: [PATCH] Add packet-groups API endpoint for deduplicated packet list - New GroupedPacketRead/PacketReceptionInfo/GroupedPacketList schemas - GET /api/v1/packet-groups: two-phase GROUP BY query, 7-day default window, lightweight Phase 2 (no raw_hex/decoded), role-aware cache - GET /api/v1/packet-groups/{hash}: full reception list with path_hashes extracted from decoded.payload.decoded.pathHashes - New (packet_hash, received_at) composite index via Alembic migration - i18n keys for receptions, path, observer counts https://claude.ai/code/session_01NH2rZzuHzasJj12SZeRhbJ --- ...4e5f6_add_packet_hash_received_at_index.py | 30 ++ src/meshcore_hub/api/routes/__init__.py | 2 + src/meshcore_hub/api/routes/packet_groups.py | 292 ++++++++++++++++++ src/meshcore_hub/common/models/raw_packet.py | 5 + .../common/schemas/raw_packets.py | 47 +++ src/meshcore_hub/web/static/locales/en.json | 9 +- 6 files changed, 384 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/20260613_0730_a1b2c3d4e5f6_add_packet_hash_received_at_index.py create mode 100644 src/meshcore_hub/api/routes/packet_groups.py diff --git a/alembic/versions/20260613_0730_a1b2c3d4e5f6_add_packet_hash_received_at_index.py b/alembic/versions/20260613_0730_a1b2c3d4e5f6_add_packet_hash_received_at_index.py new file mode 100644 index 0000000..6794f94 --- /dev/null +++ b/alembic/versions/20260613_0730_a1b2c3d4e5f6_add_packet_hash_received_at_index.py @@ -0,0 +1,30 @@ +"""add packet_hash_received_at composite index + +Revision ID: a1b2c3d4e5f6 +Revises: e9f0c4079540 +Create Date: 2026-06-13 07:30:00.000000+00:00 + +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "a1b2c3d4e5f6" +down_revision: Union[str, None] = "e9f0c4079540" +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.create_index( + "ix_raw_packets_packet_hash_received_at", + ["packet_hash", "received_at"], + unique=False, + ) + + +def downgrade() -> None: + with op.batch_alter_table("raw_packets", schema=None) as batch_op: + batch_op.drop_index("ix_raw_packets_packet_hash_received_at") diff --git a/src/meshcore_hub/api/routes/__init__.py b/src/meshcore_hub/api/routes/__init__.py index 301ac10..5e2fa43 100644 --- a/src/meshcore_hub/api/routes/__init__.py +++ b/src/meshcore_hub/api/routes/__init__.py @@ -13,6 +13,7 @@ from meshcore_hub.api.routes.user_profiles import router as user_profiles_router from meshcore_hub.api.routes.adoptions import router as adoptions_router from meshcore_hub.api.routes.channels import router as channels_router from meshcore_hub.api.routes.raw_packets import router as raw_packets_router +from meshcore_hub.api.routes.packet_groups import router as packet_groups_router api_router = APIRouter() @@ -32,3 +33,4 @@ api_router.include_router(user_profiles_router, prefix="/user", tags=["User"]) api_router.include_router(adoptions_router, prefix="/adoptions", tags=["Adoptions"]) api_router.include_router(channels_router, prefix="/channels", tags=["Channels"]) api_router.include_router(raw_packets_router, prefix="/packets", tags=["Packets"]) +api_router.include_router(packet_groups_router, prefix="/packet-groups", tags=["Packet Groups"]) diff --git a/src/meshcore_hub/api/routes/packet_groups.py b/src/meshcore_hub/api/routes/packet_groups.py new file mode 100644 index 0000000..086327a --- /dev/null +++ b/src/meshcore_hub/api/routes/packet_groups.py @@ -0,0 +1,292 @@ +"""Grouped raw packet routes — one entry per unique packet_hash.""" + +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +from fastapi import APIRouter, HTTPException, Query, Request +from sqlalchemy import func, or_, select +from sqlalchemy.orm import aliased, selectinload + +from meshcore_hub.api.auth import RequireRead +from meshcore_hub.api.cache import cached, sorted_query_string +from meshcore_hub.api.channel_visibility import ( + get_max_visibility_level, + get_visible_channel_indices, + resolve_user_role, +) +from meshcore_hub.api.dependencies import DbSession +from meshcore_hub.common.models import Node, RawPacket +from meshcore_hub.common.schemas.raw_packets import ( + GroupedPacketList, + GroupedPacketRead, + PacketReceptionInfo, +) + +router = APIRouter() + +SEARCH_DEFAULT_WINDOW_DAYS = 7 +VALID_SORT_COLUMNS = {"time", "event_type", "reception_count"} + + +def _group_key_builder(request: Request) -> str: + role = resolve_user_role(request) or "anonymous" + return f"packet_groups:role={role}:{sorted_query_string(request)}" + + +def _extract_path_hashes(decoded: dict[str, Any] | None) -> list[str] | None: + """Extract pathHashes from decoded.payload.decoded.pathHashes.""" + if not decoded: + return None + payload = decoded.get("payload") or {} + inner = payload.get("decoded") or {} + hashes = inner.get("pathHashes") + return hashes if isinstance(hashes, list) else None + + +def _get_tag_name(node: Optional[Node]) -> Optional[str]: + if not node or not node.tags: + return None + for tag in node.tags: + if tag.key == "name": + return tag.value + return None + + +@router.get("", response_model=GroupedPacketList) +@cached("packet_groups", key_builder=_group_key_builder) +def list_packet_groups( + _: RequireRead, + session: DbSession, + request: Request, + search: Optional[str] = Query(None, description="Search in packet hash or observer name"), + event_type: Optional[str] = Query(None, description="Filter by event type"), + channel_idx: Optional[int] = Query(None, description="Filter by channel index"), + since: Optional[datetime] = Query(None, description="Start timestamp"), + until: Optional[datetime] = Query(None, description="End timestamp"), + sort: Optional[str] = Query(None, description="Sort column: time, event_type, reception_count"), + order: Optional[str] = Query(None, description="asc or desc"), + limit: int = Query(50, ge=1, le=100), + offset: int = Query(0, ge=0), +) -> GroupedPacketList: + """List deduplicated packet groups — one row per unique packet_hash. + + Rows with no packet_hash are excluded (they cannot be meaningfully grouped). + The 7-day default window applies unless ``since`` is provided. + """ + sort = sort if sort in VALID_SORT_COLUMNS else "time" + order = order if order in ("asc", "desc") else "desc" + + role = resolve_user_role(request) + max_level = get_max_visibility_level(role) + visible_indices = get_visible_channel_indices(session, max_level) + + # Default time window — keeps GROUP BY bounded when no explicit since/until. + if since is None: + if search: + since = datetime.now(timezone.utc) - timedelta(days=SEARCH_DEFAULT_WINDOW_DAYS) + else: + since = datetime.now(timezone.utc) - timedelta(days=SEARCH_DEFAULT_WINDOW_DAYS) + + ObserverNode = aliased(Node) + + # ── Phase 1: GROUP BY packet_hash to paginate over distinct groups ──────── + group_query = ( + select( + RawPacket.packet_hash, + func.count(RawPacket.id).label("reception_count"), + func.count(RawPacket.observer_node_id.distinct()).label("observer_count"), + func.min(RawPacket.received_at).label("first_seen"), + ) + .outerjoin(ObserverNode, RawPacket.observer_node_id == ObserverNode.id) + .where(RawPacket.packet_hash.is_not(None)) + .where(RawPacket.received_at >= since) + ) + + if search: + pattern = f"%{search}%" + group_query = group_query.where( + or_( + RawPacket.packet_hash.ilike(pattern), + ObserverNode.name.ilike(pattern), + ) + ) + + if event_type: + group_query = group_query.where(RawPacket.event_type == event_type) + + if channel_idx is not None: + group_query = group_query.where(RawPacket.channel_idx == channel_idx) + + if until: + group_query = group_query.where(RawPacket.received_at <= until) + + group_query = group_query.group_by(RawPacket.packet_hash) + + count_query = select(func.count()).select_from(group_query.subquery()) + total = session.execute(count_query).scalar() or 0 + + sort_col = { + "time": func.min(RawPacket.received_at), + "event_type": func.min(RawPacket.event_type), + "reception_count": func.count(RawPacket.id), + }[sort] + group_query = group_query.order_by( + sort_col.desc() if order == "desc" else sort_col.asc() + ) + group_query = group_query.offset(offset).limit(limit) + + group_rows = session.execute(group_query).all() + hashes = [r.packet_hash for r in group_rows] + + if not hashes: + return GroupedPacketList(items=[], total=total, limit=limit, offset=offset) + + # ── Phase 2: Lightweight metadata fetch — no raw_hex, no decoded ────────── + meta_query = select( + RawPacket.id, + RawPacket.packet_hash, + RawPacket.event_type, + RawPacket.channel_idx, + RawPacket.packet_type, + RawPacket.payload_type, + RawPacket.route_type, + RawPacket.source_pubkey_prefix, + RawPacket.received_at, + ).where(RawPacket.packet_hash.in_(hashes)).order_by(RawPacket.received_at.asc()) + + meta_rows = session.execute(meta_query).all() + + # Pick first (oldest) row per hash as the representative display row + representative: dict[str, Any] = {} + for row in meta_rows: + if row.packet_hash not in representative: + representative[row.packet_hash] = row + + group_counts = {r.packet_hash: r for r in group_rows} + + items = [] + for h in hashes: + rep = representative.get(h) + grp = group_counts[h] + is_redacted = bool( + rep is not None + and rep.channel_idx is not None + and rep.channel_idx not in visible_indices + ) + items.append( + GroupedPacketRead( + packet_hash=h, + event_type=rep.event_type if rep else None, + channel_idx=rep.channel_idx if rep else None, + packet_type=rep.packet_type if rep else None, + payload_type=rep.payload_type if rep else None, + route_type=rep.route_type if rep else None, + source_pubkey_prefix=( + None if is_redacted else (rep.source_pubkey_prefix if rep else None) + ), + reception_count=grp.reception_count, + observer_count=grp.observer_count, + receptions=[], + first_seen=grp.first_seen, + redacted=is_redacted, + raw_hex=None, + decoded=None, + ) + ) + + return GroupedPacketList(items=items, total=total, limit=limit, offset=offset) + + +@router.get("/{packet_hash}", response_model=GroupedPacketRead) +def get_packet_group( + _: RequireRead, + session: DbSession, + request: Request, + packet_hash: str, +) -> GroupedPacketRead: + """Full detail for a packet group, including all (observer, path) receptions.""" + ObserverNode = aliased(Node) + + rows = session.execute( + select( + RawPacket, + ObserverNode.public_key.label("observer_pk"), + ObserverNode.name.label("observer_name"), + ObserverNode.id.label("observer_id"), + ) + .outerjoin(ObserverNode, RawPacket.observer_node_id == ObserverNode.id) + .where(RawPacket.packet_hash == packet_hash) + .order_by(RawPacket.received_at.asc()) + ).all() + + if not rows: + raise HTTPException(status_code=404, detail="Packet group not found") + + role = resolve_user_role(request) + max_level = get_max_visibility_level(role) + visible_indices = get_visible_channel_indices(session, max_level) + + observer_ids = {row.observer_id for row in rows if row.observer_id} + nodes_by_id: dict[str, Node] = {} + if observer_ids: + nodes = ( + session.execute( + select(Node) + .where(Node.id.in_(observer_ids)) + .options(selectinload(Node.tags)) + ) + .scalars() + .all() + ) + nodes_by_id = {n.id: n for n in nodes} + + receptions: list[PacketReceptionInfo] = [] + for row in rows: + packet = row[0] + is_redacted = ( + packet.channel_idx is not None + and packet.channel_idx not in visible_indices + ) + observer_node = nodes_by_id.get(row.observer_id) if row.observer_id else None + receptions.append( + PacketReceptionInfo( + packet_id=packet.id, + observed_by=row.observer_pk, + observer_name=row.observer_name, + observer_tag_name=_get_tag_name(observer_node), + snr=packet.snr, + path_len=packet.path_len, + path_hashes=None if is_redacted else _extract_path_hashes(packet.decoded), + received_at=packet.received_at, + redacted=is_redacted, + ) + ) + + first_packet = rows[0][0] + all_redacted = all(r.redacted for r in receptions) + unique_observers = len({r.observed_by for r in receptions if r.observed_by}) + + # Use first non-redacted row for the shared raw_hex/decoded panel + rep_packet = None + for row in rows: + p = row[0] + if p.channel_idx is None or p.channel_idx in visible_indices: + rep_packet = p + break + + return GroupedPacketRead( + packet_hash=packet_hash, + event_type=first_packet.event_type, + channel_idx=first_packet.channel_idx, + packet_type=first_packet.packet_type, + payload_type=first_packet.payload_type, + route_type=first_packet.route_type, + source_pubkey_prefix=None if all_redacted else first_packet.source_pubkey_prefix, + reception_count=len(receptions), + observer_count=unique_observers, + receptions=receptions, + first_seen=first_packet.received_at, + redacted=all_redacted, + raw_hex=rep_packet.raw_hex if rep_packet else None, + decoded=rep_packet.decoded if rep_packet else None, + ) diff --git a/src/meshcore_hub/common/models/raw_packet.py b/src/meshcore_hub/common/models/raw_packet.py index 9abc813..66576ab 100644 --- a/src/meshcore_hub/common/models/raw_packet.py +++ b/src/meshcore_hub/common/models/raw_packet.py @@ -113,6 +113,11 @@ class RawPacket(Base, UUIDMixin, TimestampMixin): "source_pubkey_prefix", "received_at", ), + Index( + "ix_raw_packets_packet_hash_received_at", + "packet_hash", + "received_at", + ), ) def __repr__(self) -> str: diff --git a/src/meshcore_hub/common/schemas/raw_packets.py b/src/meshcore_hub/common/schemas/raw_packets.py index 46d9291..e682fd1 100644 --- a/src/meshcore_hub/common/schemas/raw_packets.py +++ b/src/meshcore_hub/common/schemas/raw_packets.py @@ -65,3 +65,50 @@ class RawPacketList(BaseModel): total: int = Field(..., description="Total number of raw packets") limit: int = Field(..., description="Page size limit") offset: int = Field(..., description="Page offset") + + +class PacketReceptionInfo(BaseModel): + """One raw_packet row — a single (observer, path) reception.""" + + packet_id: str = Field(..., description="Raw packet UUID") + observed_by: Optional[str] = Field(default=None, description="Observer public key") + observer_name: Optional[str] = Field(default=None, description="Observer node name") + observer_tag_name: Optional[str] = Field(default=None, description="Observer name from tags") + snr: Optional[float] = Field(default=None, description="SNR at this observer") + path_len: Optional[int] = Field(default=None, description="Hop count") + path_hashes: Optional[list[str]] = Field( + default=None, description="Hop node hash sequence from decoded.payload.decoded.pathHashes" + ) + received_at: datetime = Field(..., description="When received") + redacted: bool = Field(default=False) + + +class GroupedPacketRead(BaseModel): + """One packet hash with aggregated reception info.""" + + packet_hash: Optional[str] = Field(default=None) + event_type: Optional[str] = Field(default=None) + channel_idx: Optional[int] = Field(default=None) + packet_type: Optional[int] = Field(default=None) + payload_type: Optional[int] = Field(default=None) + route_type: Optional[str] = Field(default=None) + source_pubkey_prefix: Optional[str] = Field(default=None) + reception_count: int = Field(..., description="Total rows (paths × observers)") + observer_count: int = Field(..., description="Distinct observer nodes") + receptions: list[PacketReceptionInfo] = Field( + default_factory=list, + description="Individual receptions (populated for detail, empty for list)", + ) + first_seen: datetime = Field(..., description="Earliest received_at across all receptions") + redacted: bool = Field(default=False, description="True when all receptions are redacted") + raw_hex: Optional[str] = Field(default=None, description="Representative raw hex (null for list view)") + decoded: Optional[dict[str, Any]] = Field(default=None, description="Representative decoded JSON (null for list view)") + + +class GroupedPacketList(BaseModel): + """Paginated list of grouped packets.""" + + items: list[GroupedPacketRead] = Field(..., description="List of packet groups") + total: int = Field(..., description="Total distinct packet hashes matching filters") + limit: int + offset: int diff --git a/src/meshcore_hub/web/static/locales/en.json b/src/meshcore_hub/web/static/locales/en.json index 95c771d..3b68532 100644 --- a/src/meshcore_hub/web/static/locales/en.json +++ b/src/meshcore_hub/web/static/locales/en.json @@ -227,12 +227,19 @@ "col_source": "Source", "col_route_type": "Route Type", "col_raw": "Raw", + "receptions_title": "Receptions", + "receptions_count": "Receptions", + "reception_singular": "path", + "reception_plural": "paths", + "col_path": "Path", + "col_receptions": "Receptions", "sort": { "newest": "Time (newest)", "oldest": "Time (oldest)", "event_az": "Event (A–Z)", "snr_high": "SNR (highest)", - "path_high": "Hops (most)" + "path_high": "Hops (most)", + "receptions_high": "Receptions (most)" } }, "map": {