From c80986fe67d0b531070d690201429398d6f7d1e5 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 6 Dec 2025 12:08:07 +0000 Subject: [PATCH] Add event deduplication at presentation layer When multiple receiver nodes are running, the same mesh events (messages, advertisements) are reported multiple times. This causes duplicate entries in the Web UI. Changes: - Add hash_utils.py with deterministic hash functions for each event type - Add `dedupe` parameter to messages and advertisements API endpoints (default: True) - Update dashboard stats to use distinct counts for messages/advertisements - Deduplicate recent advertisements and channel messages in dashboard - Add comprehensive tests for hash utilities Hash strategy: - Messages: hash of text + pubkey_prefix + channel_idx + sender_timestamp + txt_type - Advertisements: hash of public_key + name + adv_type + flags + 5-minute time bucket --- src/meshcore_hub/api/routes/advertisements.py | 64 ++++- src/meshcore_hub/api/routes/dashboard.py | 141 +++++++-- src/meshcore_hub/api/routes/messages.py | 59 +++- src/meshcore_hub/common/hash_utils.py | 144 ++++++++++ tests/test_common/test_hash_utils.py | 268 ++++++++++++++++++ 5 files changed, 644 insertions(+), 32 deletions(-) create mode 100644 src/meshcore_hub/common/hash_utils.py create mode 100644 tests/test_common/test_hash_utils.py diff --git a/src/meshcore_hub/api/routes/advertisements.py b/src/meshcore_hub/api/routes/advertisements.py index 55a57ea..94e74ff 100644 --- a/src/meshcore_hub/api/routes/advertisements.py +++ b/src/meshcore_hub/api/routes/advertisements.py @@ -9,6 +9,7 @@ from sqlalchemy.orm import aliased, selectinload from meshcore_hub.api.auth import RequireRead from meshcore_hub.api.dependencies import DbSession +from meshcore_hub.common.hash_utils import compute_advertisement_hash from meshcore_hub.common.models import Advertisement, Node from meshcore_hub.common.schemas.messages import AdvertisementList, AdvertisementRead @@ -35,6 +36,9 @@ async def list_advertisements( ), since: Optional[datetime] = Query(None, description="Start timestamp"), until: Optional[datetime] = Query(None, description="End timestamp"), + dedupe: bool = Query( + True, description="Deduplicate advertisements from multiple receivers" + ), limit: int = Query(50, ge=1, le=100, description="Page size"), offset: int = Query(0, ge=0, description="Page offset"), ) -> AdvertisementList: @@ -70,12 +74,37 @@ async def list_advertisements( if until: query = query.where(Advertisement.received_at <= until) - # Get total count - count_query = select(func.count()).select_from(query.subquery()) - total = session.execute(count_query).scalar() or 0 + # When deduplicating, we need to fetch more results and compute distinct count + if dedupe: + # For deduplicated count, count distinct by public_key within time buckets + # We use a 5-minute time bucket for advertisements + distinct_subquery = ( + select( + Advertisement.public_key, + Advertisement.name, + Advertisement.adv_type, + Advertisement.flags, + # Use date truncation for time bucketing (5 min = 300 seconds) + (func.strftime("%s", Advertisement.received_at) / 300).label( + "time_bucket" + ), + ) + .distinct() + .select_from(query.subquery()) + ) + count_query = select(func.count()).select_from(distinct_subquery.subquery()) + total = session.execute(count_query).scalar() or 0 - # Apply pagination - query = query.order_by(Advertisement.received_at.desc()).offset(offset).limit(limit) + # Fetch extra results to account for duplicates + fetch_limit = (limit + offset) * 3 + query = query.order_by(Advertisement.received_at.desc()).limit(fetch_limit) + else: + # Standard count and pagination + count_query = select(func.count()).select_from(query.subquery()) + total = session.execute(count_query).scalar() or 0 + query = ( + query.order_by(Advertisement.received_at.desc()).offset(offset).limit(limit) + ) # Execute results = session.execute(query).all() @@ -99,8 +128,25 @@ async def list_advertisements( # Build response with node details items = [] + seen_hashes: set[str] = set() + for row in results: adv = row[0] + + # Compute hash for deduplication + if dedupe: + adv_hash = compute_advertisement_hash( + public_key=adv.public_key, + name=adv.name, + adv_type=adv.adv_type, + flags=adv.flags, + received_at=adv.received_at, + bucket_minutes=5, + ) + if adv_hash in seen_hashes: + continue + seen_hashes.add(adv_hash) + receiver_node = nodes_by_id.get(row.receiver_id) if row.receiver_id else None source_node = nodes_by_id.get(row.source_id) if row.source_id else None @@ -119,6 +165,14 @@ async def list_advertisements( } items.append(AdvertisementRead(**data)) + # Stop once we have enough items (for dedupe mode with pagination) + if dedupe and len(items) >= offset + limit: + break + + # Apply offset for dedupe mode (we fetched from beginning) + if dedupe: + items = items[offset : offset + limit] + return AdvertisementList( items=items, total=total, diff --git a/src/meshcore_hub/api/routes/dashboard.py b/src/meshcore_hub/api/routes/dashboard.py index 9afecdb..d2bebf0 100644 --- a/src/meshcore_hub/api/routes/dashboard.py +++ b/src/meshcore_hub/api/routes/dashboard.py @@ -8,6 +8,10 @@ from sqlalchemy import func, select from meshcore_hub.api.auth import RequireRead from meshcore_hub.api.dependencies import DbSession +from meshcore_hub.common.hash_utils import ( + compute_advertisement_hash, + compute_message_hash, +) from meshcore_hub.common.models import Advertisement, Message, Node, NodeTag from meshcore_hub.common.schemas.messages import ( ChannelMessage, @@ -43,45 +47,110 @@ async def get_stats( or 0 ) - # Total messages + # Total messages (deduplicated by content hash) + distinct_messages = ( + select( + Message.text, + Message.pubkey_prefix, + Message.channel_idx, + Message.sender_timestamp, + Message.txt_type, + ) + .distinct() + .subquery() + ) total_messages = ( - session.execute(select(func.count()).select_from(Message)).scalar() or 0 + session.execute(select(func.count()).select_from(distinct_messages)).scalar() + or 0 ) - # Messages today + # Messages today (deduplicated) + distinct_messages_today = ( + select( + Message.text, + Message.pubkey_prefix, + Message.channel_idx, + Message.sender_timestamp, + Message.txt_type, + ) + .where(Message.received_at >= today_start) + .distinct() + .subquery() + ) messages_today = ( session.execute( - select(func.count()) - .select_from(Message) - .where(Message.received_at >= today_start) + select(func.count()).select_from(distinct_messages_today) ).scalar() or 0 ) - # Total advertisements + # Total advertisements (deduplicated by public_key + 5min time bucket) + distinct_advertisements = ( + select( + Advertisement.public_key, + Advertisement.name, + Advertisement.adv_type, + Advertisement.flags, + (func.strftime("%s", Advertisement.received_at) / 300).label("time_bucket"), + ) + .distinct() + .subquery() + ) total_advertisements = ( - session.execute(select(func.count()).select_from(Advertisement)).scalar() or 0 + session.execute( + select(func.count()).select_from(distinct_advertisements) + ).scalar() + or 0 ) - # Advertisements in last 24h + # Advertisements in last 24h (deduplicated) + distinct_advertisements_24h = ( + select( + Advertisement.public_key, + Advertisement.name, + Advertisement.adv_type, + Advertisement.flags, + (func.strftime("%s", Advertisement.received_at) / 300).label("time_bucket"), + ) + .where(Advertisement.received_at >= yesterday) + .distinct() + .subquery() + ) advertisements_24h = ( session.execute( - select(func.count()) - .select_from(Advertisement) - .where(Advertisement.received_at >= yesterday) + select(func.count()).select_from(distinct_advertisements_24h) ).scalar() or 0 ) - # Recent advertisements (last 10) - recent_ads = ( + # Recent advertisements (last 10, deduplicated) + # Fetch more to ensure we have 10 unique after deduplication + recent_ads_raw = ( session.execute( - select(Advertisement).order_by(Advertisement.received_at.desc()).limit(10) + select(Advertisement).order_by(Advertisement.received_at.desc()).limit(30) ) .scalars() .all() ) + # Deduplicate by hash + seen_ad_hashes: set[str] = set() + recent_ads = [] + for ad in recent_ads_raw: + ad_hash = compute_advertisement_hash( + public_key=ad.public_key, + name=ad.name, + adv_type=ad.adv_type, + flags=ad.flags, + received_at=ad.received_at, + bucket_minutes=5, + ) + if ad_hash not in seen_ad_hashes: + seen_ad_hashes.add(ad_hash) + recent_ads.append(ad) + if len(recent_ads) >= 10: + break + # Get node names, adv_types, and friendly_name tags for the advertised nodes ad_public_keys = [ad.public_key for ad in recent_ads] node_names: dict[str, str] = {} @@ -119,29 +188,57 @@ async def get_stats( for ad in recent_ads ] - # Channel message counts - channel_counts_query = ( - select(Message.channel_idx, func.count()) + # Channel message counts (deduplicated) + distinct_channel_messages = ( + select( + Message.channel_idx, + Message.text, + Message.pubkey_prefix, + Message.sender_timestamp, + Message.txt_type, + ) .where(Message.message_type == "channel") .where(Message.channel_idx.isnot(None)) - .group_by(Message.channel_idx) + .distinct() + .subquery() ) + channel_counts_query = select( + distinct_channel_messages.c.channel_idx, func.count() + ).group_by(distinct_channel_messages.c.channel_idx) channel_results = session.execute(channel_counts_query).all() channel_message_counts = { int(channel): int(count) for channel, count in channel_results } - # Get latest 5 messages for each channel that has messages + # Get latest 5 messages for each channel that has messages (deduplicated) channel_messages: dict[int, list[ChannelMessage]] = {} for channel_idx, _ in channel_results: + # Fetch more messages to deduplicate messages_query = ( select(Message) .where(Message.message_type == "channel") .where(Message.channel_idx == channel_idx) .order_by(Message.received_at.desc()) - .limit(5) + .limit(15) ) - channel_msgs = session.execute(messages_query).scalars().all() + channel_msgs_raw = session.execute(messages_query).scalars().all() + + # Deduplicate + seen_msg_hashes: set[str] = set() + channel_msgs = [] + for m in channel_msgs_raw: + msg_hash = compute_message_hash( + text=m.text, + pubkey_prefix=m.pubkey_prefix, + channel_idx=m.channel_idx, + sender_timestamp=m.sender_timestamp, + txt_type=m.txt_type, + ) + if msg_hash not in seen_msg_hashes: + seen_msg_hashes.add(msg_hash) + channel_msgs.append(m) + if len(channel_msgs) >= 5: + break # Look up sender names for these messages msg_prefixes = [m.pubkey_prefix for m in channel_msgs if m.pubkey_prefix] diff --git a/src/meshcore_hub/api/routes/messages.py b/src/meshcore_hub/api/routes/messages.py index 79a461a..f9ee382 100644 --- a/src/meshcore_hub/api/routes/messages.py +++ b/src/meshcore_hub/api/routes/messages.py @@ -9,6 +9,7 @@ from sqlalchemy.orm import aliased, selectinload from meshcore_hub.api.auth import RequireRead from meshcore_hub.api.dependencies import DbSession +from meshcore_hub.common.hash_utils import compute_message_hash from meshcore_hub.common.models import Message, Node, NodeTag from meshcore_hub.common.schemas.messages import MessageList, MessageRead @@ -38,6 +39,9 @@ async def list_messages( since: Optional[datetime] = Query(None, description="Start timestamp"), until: Optional[datetime] = Query(None, description="End timestamp"), search: Optional[str] = Query(None, description="Search in message text"), + dedupe: bool = Query( + True, description="Deduplicate messages from multiple receivers" + ), limit: int = Query(50, ge=1, le=100, description="Page size"), offset: int = Query(0, ge=0, description="Page offset"), ) -> MessageList: @@ -74,12 +78,33 @@ async def list_messages( if search: query = query.where(Message.text.ilike(f"%{search}%")) - # Get total count - count_query = select(func.count()).select_from(query.subquery()) - total = session.execute(count_query).scalar() or 0 + # When deduplicating, we need to fetch more results to ensure we have enough + # after removing duplicates, and compute distinct count differently + if dedupe: + # For deduplicated count, we need to count distinct content + # Use a subquery that groups by content-identifying fields + distinct_subquery = ( + select( + Message.text, + Message.pubkey_prefix, + Message.channel_idx, + Message.sender_timestamp, + Message.txt_type, + ) + .distinct() + .select_from(query.subquery()) + ) + count_query = select(func.count()).select_from(distinct_subquery.subquery()) + total = session.execute(count_query).scalar() or 0 - # Apply pagination - query = query.order_by(Message.received_at.desc()).offset(offset).limit(limit) + # Fetch extra results to account for duplicates (3x limit + offset) + fetch_limit = (limit + offset) * 3 + query = query.order_by(Message.received_at.desc()).limit(fetch_limit) + else: + # Standard count and pagination + count_query = select(func.count()).select_from(query.subquery()) + total = session.execute(count_query).scalar() or 0 + query = query.order_by(Message.received_at.desc()).offset(offset).limit(limit) # Execute results = session.execute(query).all() @@ -128,8 +153,24 @@ async def list_messages( # Build response with sender info and received_by items = [] + seen_hashes: set[str] = set() + for row in results: m = row[0] + + # Compute hash for deduplication + if dedupe: + msg_hash = compute_message_hash( + text=m.text, + pubkey_prefix=m.pubkey_prefix, + channel_idx=m.channel_idx, + sender_timestamp=m.sender_timestamp, + txt_type=m.txt_type, + ) + if msg_hash in seen_hashes: + continue + seen_hashes.add(msg_hash) + receiver_pk = row.receiver_pk receiver_name = row.receiver_name receiver_node = ( @@ -162,6 +203,14 @@ async def list_messages( } items.append(MessageRead(**msg_dict)) + # Stop once we have enough items (for dedupe mode with pagination) + if dedupe and len(items) >= offset + limit: + break + + # Apply offset for dedupe mode (we fetched from beginning) + if dedupe: + items = items[offset : offset + limit] + return MessageList( items=items, total=total, diff --git a/src/meshcore_hub/common/hash_utils.py b/src/meshcore_hub/common/hash_utils.py new file mode 100644 index 0000000..5be1825 --- /dev/null +++ b/src/meshcore_hub/common/hash_utils.py @@ -0,0 +1,144 @@ +"""Event hash utilities for deduplication. + +This module provides functions to compute deterministic hashes for events, +allowing deduplication when multiple receiver nodes report the same event. +""" + +import hashlib +from datetime import datetime +from typing import Optional + + +def compute_message_hash( + text: str, + pubkey_prefix: Optional[str] = None, + channel_idx: Optional[int] = None, + sender_timestamp: Optional[datetime] = None, + txt_type: Optional[int] = None, +) -> str: + """Compute a deterministic hash for a message. + + The hash is computed from fields that uniquely identify a message's content + and sender, excluding receiver-specific data. + + Args: + text: Message content + pubkey_prefix: Sender's public key prefix (12 chars) + channel_idx: Channel index for channel messages + sender_timestamp: Sender's timestamp + txt_type: Message type indicator + + Returns: + 32-character hex hash string + """ + # Build a canonical string from the relevant fields + parts = [ + text or "", + pubkey_prefix or "", + str(channel_idx) if channel_idx is not None else "", + sender_timestamp.isoformat() if sender_timestamp else "", + str(txt_type) if txt_type is not None else "", + ] + canonical = "|".join(parts) + return hashlib.md5(canonical.encode("utf-8")).hexdigest() + + +def compute_advertisement_hash( + public_key: str, + name: Optional[str] = None, + adv_type: Optional[str] = None, + flags: Optional[int] = None, + received_at: Optional[datetime] = None, + bucket_minutes: int = 5, +) -> str: + """Compute a deterministic hash for an advertisement. + + Advertisements are bucketed by time since the same node may advertise + periodically and we want to deduplicate within a time window. + + Args: + public_key: Advertised node's public key + name: Advertised name + adv_type: Node type + flags: Capability flags + received_at: When received (used for time bucketing) + bucket_minutes: Time bucket size in minutes (default 5) + + Returns: + 32-character hex hash string + """ + # Bucket the time to allow deduplication within a window + time_bucket = "" + if received_at: + # Round down to nearest bucket + bucket_seconds = bucket_minutes * 60 + epoch = int(received_at.timestamp()) + bucket_epoch = (epoch // bucket_seconds) * bucket_seconds + time_bucket = str(bucket_epoch) + + parts = [ + public_key, + name or "", + adv_type or "", + str(flags) if flags is not None else "", + time_bucket, + ] + canonical = "|".join(parts) + return hashlib.md5(canonical.encode("utf-8")).hexdigest() + + +def compute_trace_hash(initiator_tag: int) -> str: + """Compute a deterministic hash for a trace path. + + Trace paths have a unique initiator_tag that serves as the identifier. + + Args: + initiator_tag: Unique trace identifier + + Returns: + 32-character hex hash string + """ + return hashlib.md5(str(initiator_tag).encode("utf-8")).hexdigest() + + +def compute_telemetry_hash( + node_public_key: str, + parsed_data: Optional[dict] = None, + received_at: Optional[datetime] = None, + bucket_minutes: int = 5, +) -> str: + """Compute a deterministic hash for a telemetry record. + + Telemetry is bucketed by time since nodes report periodically. + + Args: + node_public_key: Reporting node's public key + parsed_data: Decoded sensor readings + received_at: When received (used for time bucketing) + bucket_minutes: Time bucket size in minutes (default 5) + + Returns: + 32-character hex hash string + """ + # Bucket the time + time_bucket = "" + if received_at: + bucket_seconds = bucket_minutes * 60 + epoch = int(received_at.timestamp()) + bucket_epoch = (epoch // bucket_seconds) * bucket_seconds + time_bucket = str(bucket_epoch) + + # Serialize parsed_data deterministically + data_str = "" + if parsed_data: + # Sort keys for deterministic serialization + sorted_items = sorted(parsed_data.items()) + data_str = str(sorted_items) + + parts = [ + node_public_key, + data_str, + time_bucket, + ] + canonical = "|".join(parts) + return hashlib.md5(canonical.encode("utf-8")).hexdigest() diff --git a/tests/test_common/test_hash_utils.py b/tests/test_common/test_hash_utils.py new file mode 100644 index 0000000..6ba9e48 --- /dev/null +++ b/tests/test_common/test_hash_utils.py @@ -0,0 +1,268 @@ +"""Tests for hash utilities for event deduplication.""" + +from datetime import datetime, timezone + +import pytest + +from meshcore_hub.common.hash_utils import ( + compute_advertisement_hash, + compute_message_hash, + compute_telemetry_hash, + compute_trace_hash, +) + + +class TestComputeMessageHash: + """Tests for compute_message_hash function.""" + + def test_same_content_produces_same_hash(self) -> None: + """Identical messages should produce the same hash.""" + timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + hash1 = compute_message_hash( + text="Hello World", + pubkey_prefix="01ab2186c4d5", + channel_idx=4, + sender_timestamp=timestamp, + txt_type=1, + ) + hash2 = compute_message_hash( + text="Hello World", + pubkey_prefix="01ab2186c4d5", + channel_idx=4, + sender_timestamp=timestamp, + txt_type=1, + ) + + assert hash1 == hash2 + + def test_different_text_produces_different_hash(self) -> None: + """Messages with different text should have different hashes.""" + timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + hash1 = compute_message_hash( + text="Hello World", + pubkey_prefix="01ab2186c4d5", + sender_timestamp=timestamp, + ) + hash2 = compute_message_hash( + text="Goodbye World", + pubkey_prefix="01ab2186c4d5", + sender_timestamp=timestamp, + ) + + assert hash1 != hash2 + + def test_different_sender_produces_different_hash(self) -> None: + """Messages from different senders should have different hashes.""" + timestamp = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + hash1 = compute_message_hash( + text="Hello", + pubkey_prefix="01ab2186c4d5", + sender_timestamp=timestamp, + ) + hash2 = compute_message_hash( + text="Hello", + pubkey_prefix="99ff8877aabb", + sender_timestamp=timestamp, + ) + + assert hash1 != hash2 + + def test_different_channel_produces_different_hash(self) -> None: + """Messages on different channels should have different hashes.""" + hash1 = compute_message_hash(text="Hello", channel_idx=1) + hash2 = compute_message_hash(text="Hello", channel_idx=2) + + assert hash1 != hash2 + + def test_handles_none_values(self) -> None: + """Hash function should handle None values gracefully.""" + hash1 = compute_message_hash( + text="Test", + pubkey_prefix=None, + channel_idx=None, + sender_timestamp=None, + txt_type=None, + ) + + assert hash1 is not None + assert len(hash1) == 32 # MD5 hex digest length + + +class TestComputeAdvertisementHash: + """Tests for compute_advertisement_hash function.""" + + def test_same_content_same_bucket_produces_same_hash(self) -> None: + """Advertisements within the same time bucket should match.""" + # Two times within the same 5-minute bucket + time1 = datetime(2024, 1, 15, 10, 31, 0, tzinfo=timezone.utc) + time2 = datetime(2024, 1, 15, 10, 33, 0, tzinfo=timezone.utc) + + hash1 = compute_advertisement_hash( + public_key="a" * 64, + name="Node1", + adv_type="chat", + flags=128, + received_at=time1, + bucket_minutes=5, + ) + hash2 = compute_advertisement_hash( + public_key="a" * 64, + name="Node1", + adv_type="chat", + flags=128, + received_at=time2, + bucket_minutes=5, + ) + + assert hash1 == hash2 + + def test_different_bucket_produces_different_hash(self) -> None: + """Advertisements in different time buckets should not match.""" + # Two times in different 5-minute buckets + time1 = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + time2 = datetime(2024, 1, 15, 10, 36, 0, tzinfo=timezone.utc) + + hash1 = compute_advertisement_hash( + public_key="a" * 64, + name="Node1", + received_at=time1, + bucket_minutes=5, + ) + hash2 = compute_advertisement_hash( + public_key="a" * 64, + name="Node1", + received_at=time2, + bucket_minutes=5, + ) + + assert hash1 != hash2 + + def test_different_public_key_produces_different_hash(self) -> None: + """Advertisements from different nodes should have different hashes.""" + time = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + hash1 = compute_advertisement_hash( + public_key="a" * 64, + received_at=time, + ) + hash2 = compute_advertisement_hash( + public_key="b" * 64, + received_at=time, + ) + + assert hash1 != hash2 + + def test_configurable_bucket_size(self) -> None: + """Bucket size should be configurable.""" + time1 = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + time2 = datetime(2024, 1, 15, 10, 35, 0, tzinfo=timezone.utc) + + # With 5-minute bucket, these should be in different buckets + hash1_5min = compute_advertisement_hash( + public_key="a" * 64, + received_at=time1, + bucket_minutes=5, + ) + hash2_5min = compute_advertisement_hash( + public_key="a" * 64, + received_at=time2, + bucket_minutes=5, + ) + assert hash1_5min != hash2_5min + + # With 10-minute bucket, these should be in the same bucket + hash1_10min = compute_advertisement_hash( + public_key="a" * 64, + received_at=time1, + bucket_minutes=10, + ) + hash2_10min = compute_advertisement_hash( + public_key="a" * 64, + received_at=time2, + bucket_minutes=10, + ) + assert hash1_10min == hash2_10min + + +class TestComputeTraceHash: + """Tests for compute_trace_hash function.""" + + def test_same_tag_produces_same_hash(self) -> None: + """Same initiator_tag should produce same hash.""" + hash1 = compute_trace_hash(initiator_tag=123456789) + hash2 = compute_trace_hash(initiator_tag=123456789) + + assert hash1 == hash2 + + def test_different_tag_produces_different_hash(self) -> None: + """Different initiator_tag should produce different hash.""" + hash1 = compute_trace_hash(initiator_tag=123456789) + hash2 = compute_trace_hash(initiator_tag=987654321) + + assert hash1 != hash2 + + +class TestComputeTelemetryHash: + """Tests for compute_telemetry_hash function.""" + + def test_same_content_same_bucket_produces_same_hash(self) -> None: + """Telemetry within the same time bucket should match.""" + time1 = datetime(2024, 1, 15, 10, 31, 0, tzinfo=timezone.utc) + time2 = datetime(2024, 1, 15, 10, 33, 0, tzinfo=timezone.utc) + data = {"temperature": 22.5, "humidity": 65} + + hash1 = compute_telemetry_hash( + node_public_key="a" * 64, + parsed_data=data, + received_at=time1, + bucket_minutes=5, + ) + hash2 = compute_telemetry_hash( + node_public_key="a" * 64, + parsed_data=data, + received_at=time2, + bucket_minutes=5, + ) + + assert hash1 == hash2 + + def test_different_data_produces_different_hash(self) -> None: + """Different sensor readings should produce different hashes.""" + time = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + hash1 = compute_telemetry_hash( + node_public_key="a" * 64, + parsed_data={"temperature": 22.5}, + received_at=time, + ) + hash2 = compute_telemetry_hash( + node_public_key="a" * 64, + parsed_data={"temperature": 25.0}, + received_at=time, + ) + + assert hash1 != hash2 + + def test_deterministic_dict_serialization(self) -> None: + """Dict serialization should be deterministic regardless of key order.""" + time = datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + # Same data, different key order in source dicts + data1 = {"a": 1, "b": 2, "c": 3} + data2 = {"c": 3, "a": 1, "b": 2} + + hash1 = compute_telemetry_hash( + node_public_key="a" * 64, + parsed_data=data1, + received_at=time, + ) + hash2 = compute_telemetry_hash( + node_public_key="a" * 64, + parsed_data=data2, + received_at=time, + ) + + assert hash1 == hash2