Merge pull request #23 from ipnet-mesh/claude/plan-event-deduplication-01NWhrJaWzQiGi1xLzmk1udg

Deduplication for multiple receiver nodes

.env.example

@@ -5,29 +5,33 @@
 # Docker Image
 # ===================
 
-# Leave empty to build from local Dockerfile, or set to use a pre-built image:
-# MESHCORE_IMAGE=ghcr.io/ipnet-mesh/meshcore-hub:latest
-# MESHCORE_IMAGE=ghcr.io/ipnet-mesh/meshcore-hub:main
-# MESHCORE_IMAGE=ghcr.io/ipnet-mesh/meshcore-hub:v1.0.0
-MESHCORE_IMAGE=
+# Docker image version tag to use
+# Options: latest, main, v1.0.0, etc.
+IMAGE_VERSION=latest
 
 # ===================
-# Data Directory
+# Data & Seed Directories
 # ===================
 
-# Base directory for all service data (collector DB, tags, members, etc.)
+# Base directory for runtime data (database, etc.)
 # Default: ./data (relative to docker-compose.yml location)
 # Inside containers this is mapped to /data
 #
 # Structure:
 #   ${DATA_HOME}/
-#   ├── collector/
-#   │   ├── meshcore.db    # SQLite database
-#   │   └── tags.json      # Node tags for import
-#   └── web/
-#       └── members.json   # Network members list
+#   └── meshcore.db    # SQLite database
 DATA_HOME=./data
 
+# Directory containing seed data files for import
+# Default: ./seed (relative to docker-compose.yml location)
+# Inside containers this is mapped to /seed
+#
+# Structure:
+#   ${SEED_HOME}/
+#   ├── node_tags.yaml    # Node tags for import
+#   └── members.yaml      # Network members for import
+SEED_HOME=./seed
+
 # ===================
 # Common Settings
 # ===================
@@ -35,12 +39,20 @@ DATA_HOME=./data
 # Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
 LOG_LEVEL=INFO
 
-# MQTT Broker Settings (internal use)
+# ===================
+# MQTT Settings
+# ===================
+
+# MQTT Broker connection (for interface/collector/api services)
+# When using the local MQTT broker (--profile mqtt), use "mqtt" as host
+# When using an external broker, set the hostname/IP
+MQTT_HOST=mqtt
+MQTT_PORT=1883
 MQTT_USERNAME=
 MQTT_PASSWORD=
 MQTT_PREFIX=meshcore
 
-# External MQTT port mapping
+# External port mappings for local MQTT broker (--profile mqtt only)
 MQTT_EXTERNAL_PORT=1883
 MQTT_WS_PORT=9001
 
@@ -58,6 +70,7 @@ SERIAL_PORT_SENDER=/dev/ttyUSB1
 SERIAL_BAUD=115200
 
 # Optional node address override (64-char hex string)
+# Only set if you need to override the device's public key
 NODE_ADDRESS=
 NODE_ADDRESS_SENDER=
 
@@ -84,25 +97,21 @@ WEB_PORT=8080
 NETWORK_NAME=MeshCore Network
 NETWORK_CITY=
 NETWORK_COUNTRY=
-NETWORK_LOCATION=
 
 # Radio configuration (comma-delimited)
 # Format: <profile>,<frequency>,<bandwidth>,<spreading_factor>,<coding_rate>,<tx_power>
 # Example: EU/UK Narrow,869.618MHz,62.5kHz,8,8,22dBm
 NETWORK_RADIO_CONFIG=
 
+# Contact information
 NETWORK_CONTACT_EMAIL=
 NETWORK_CONTACT_DISCORD=
+NETWORK_CONTACT_GITHUB=
 
 # Welcome text displayed on the homepage (plain text, optional)
 # If not set, a default welcome message is shown
 NETWORK_WELCOME_TEXT=
 
-# Members file location (optional override)
-# Default: ${DATA_HOME}/web/members.json
-# Only set this if you want to use a different location
-# MEMBERS_FILE=/custom/path/to/members.json
-
 # ===================
 # Webhook Settings
 # ===================

README.md

@@ -82,8 +82,8 @@ cd meshcore-hub
 cp .env.example .env
 # Edit .env: set SERIAL_PORT to your device (e.g., /dev/ttyUSB0 or /dev/ttyACM0)
 
-# Start the entire stack including the interface receiver
-docker compose --profile interface-receiver up -d
+# Start the entire stack with local MQTT broker
+docker compose --profile mqtt --profile core --profile receiver up -d
 
 # View the web dashboard
 open http://localhost:8080
@@ -105,7 +105,7 @@ For larger deployments, you can separate receiver nodes from the central infrast
 │  │   Device     │   │   Device     │   │    Device    │             │
 │  └──────┬───────┘   └──────┬───────┘   └──────┬───────┘             │
 │         │                  │                  │                      │
-│         │ interface-receiver only             │                      │
+│         │ receiver profile only               │                      │
 │         └──────────────────┼──────────────────┘                      │
 │                            │                                         │
 │                     MQTT (port 1883)                                 │
@@ -126,18 +126,21 @@ For larger deployments, you can separate receiver nodes from the central infrast
 
 **On each receiver node (Raspberry Pi, etc.):**
 ```bash
-# Only run the interface-receiver component
+# Only run the receiver component
 # Configure .env with MQTT_HOST pointing to your central server
 MQTT_HOST=your-community-server.com
 SERIAL_PORT=/dev/ttyUSB0
 
-docker compose --profile interface-receiver up -d
+docker compose --profile receiver up -d
 ```
 
 **On the central server (VPS/cloud):**
 ```bash
-# Run the core infrastructure (no interface needed)
-docker compose up -d
+# Run the core infrastructure with local MQTT broker
+docker compose --profile mqtt --profile core up -d
+
+# Or connect to an existing MQTT broker (set MQTT_HOST in .env)
+docker compose --profile core up -d
 ```
 
 This architecture allows:
@@ -150,26 +153,19 @@ This architecture allows:
 
 ### Using Docker Compose (Recommended)
 
-Docker Compose runs core services by default and uses **profiles** for optional components:
+Docker Compose uses **profiles** to select which services to run:
 
-**Default Services (always run):**
+| Profile | Services | Use Case |
+|---------|----------|----------|
+| `core` | collector, api, web | Central server infrastructure |
+| `receiver` | interface-receiver | Receiver node (events to MQTT) |
+| `sender` | interface-sender | Sender node (MQTT to device) |
+| `mqtt` | mosquitto broker | Local MQTT broker (optional) |
+| `mock` | interface-mock-receiver | Testing without hardware |
+| `migrate` | db-migrate | One-time database migration |
+| `seed` | seed | One-time seed data import |
 
-| Service | Description |
-|---------|-------------|
-| `mqtt` | Eclipse Mosquitto MQTT broker |
-| `collector` | MQTT subscriber + database storage (auto-seeds on startup) |
-| `api` | REST API server |
-| `web` | Web dashboard |
-
-**Optional Profiles:**
-
-| Profile | Services |
-|---------|----------|
-| `interface-receiver` | MeshCore device receiver (events to MQTT) |
-| `interface-sender` | MeshCore device sender (MQTT to device) |
-| `mock` | Mock device receiver (for testing without hardware) |
-| `migrate` | One-time database migration runner |
-| `seed` | One-time seed data import (also runs automatically on collector startup) |
+**Note:** Most deployments connect to an external MQTT broker. Add `--profile mqtt` only if you need a local broker.
 
 ```bash
 # Clone the repository
@@ -180,24 +176,24 @@ cd meshcore-hub
 cp .env.example .env
 # Edit .env with your settings (API keys, serial port, network info)
 
-# Option 1: Start core services (mqtt, collector, api, web)
-docker compose up -d
+# Create database schema
+docker compose --profile migrate run --rm db-migrate
 
-# Option 2: Start with mock device for testing
-docker compose --profile mock up -d
+# Seed the database
+docker compose --profile seed run --rm seed
 
-# Option 3: Start with real MeshCore device
-docker compose --profile interface-receiver up -d
+# Start core services with local MQTT broker
+docker compose --profile mqtt --profile core up -d
+
+# Or connect to external MQTT (configure MQTT_HOST in .env)
+docker compose --profile core up -d
+
+# Start just the receiver (connects to MQTT_HOST from .env)
+docker compose --profile receiver up -d
 
 # View logs
 docker compose logs -f
 
-# Run database migrations (one-time)
-docker compose --profile migrate up
-
-# Import seed data manually (also runs on collector startup)
-docker compose --profile seed up
-
 # Stop services
 docker compose down
 ```

alembic/versions/20241206_0001_003_add_event_hash.py

@@ -0,0 +1,67 @@
+"""Add event_hash column to event tables for deduplication
+
+Revision ID: 003
+Revises: 002
+Create Date: 2024-12-06
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "003"
+down_revision: Union[str, None] = "002"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add event_hash column to messages table
+    op.add_column(
+        "messages",
+        sa.Column("event_hash", sa.String(32), nullable=True),
+    )
+    op.create_index("ix_messages_event_hash", "messages", ["event_hash"])
+
+    # Add event_hash column to advertisements table
+    op.add_column(
+        "advertisements",
+        sa.Column("event_hash", sa.String(32), nullable=True),
+    )
+    op.create_index("ix_advertisements_event_hash", "advertisements", ["event_hash"])
+
+    # Add event_hash column to trace_paths table
+    op.add_column(
+        "trace_paths",
+        sa.Column("event_hash", sa.String(32), nullable=True),
+    )
+    op.create_index("ix_trace_paths_event_hash", "trace_paths", ["event_hash"])
+
+    # Add event_hash column to telemetry table
+    op.add_column(
+        "telemetry",
+        sa.Column("event_hash", sa.String(32), nullable=True),
+    )
+    op.create_index("ix_telemetry_event_hash", "telemetry", ["event_hash"])
+
+
+def downgrade() -> None:
+    # Remove event_hash from telemetry
+    op.drop_index("ix_telemetry_event_hash", table_name="telemetry")
+    op.drop_column("telemetry", "event_hash")
+
+    # Remove event_hash from trace_paths
+    op.drop_index("ix_trace_paths_event_hash", table_name="trace_paths")
+    op.drop_column("trace_paths", "event_hash")
+
+    # Remove event_hash from advertisements
+    op.drop_index("ix_advertisements_event_hash", table_name="advertisements")
+    op.drop_column("advertisements", "event_hash")
+
+    # Remove event_hash from messages
+    op.drop_index("ix_messages_event_hash", table_name="messages")
+    op.drop_column("messages", "event_hash")

alembic/versions/20241206_0002_004_event_receivers.py

@@ -0,0 +1,63 @@
+"""Add event_receivers junction table for multi-receiver tracking
+
+Revision ID: 004
+Revises: 003
+Create Date: 2024-12-06
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "004"
+down_revision: Union[str, None] = "003"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "event_receivers",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("event_type", sa.String(20), nullable=False),
+        sa.Column("event_hash", sa.String(32), nullable=False),
+        sa.Column(
+            "receiver_node_id",
+            sa.String(36),
+            sa.ForeignKey("nodes.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("snr", sa.Float, nullable=True),
+        sa.Column("received_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.UniqueConstraint(
+            "event_hash", "receiver_node_id", name="uq_event_receivers_hash_node"
+        ),
+    )
+    op.create_index(
+        "ix_event_receivers_event_hash",
+        "event_receivers",
+        ["event_hash"],
+    )
+    op.create_index(
+        "ix_event_receivers_receiver_node_id",
+        "event_receivers",
+        ["receiver_node_id"],
+    )
+    op.create_index(
+        "ix_event_receivers_type_hash",
+        "event_receivers",
+        ["event_type", "event_hash"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_event_receivers_type_hash", table_name="event_receivers")
+    op.drop_index("ix_event_receivers_receiver_node_id", table_name="event_receivers")
+    op.drop_index("ix_event_receivers_event_hash", table_name="event_receivers")
+    op.drop_table("event_receivers")

alembic/versions/20241206_0003_005_unique_event_hash.py

@@ -0,0 +1,63 @@
+"""Make event_hash columns unique for race condition prevention
+
+Revision ID: 005
+Revises: 004
+Create Date: 2024-12-06
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "005"
+down_revision: Union[str, None] = "004"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Drop existing non-unique indexes and create unique constraints
+    # Note: SQLite handles NULL values as unique (each NULL is distinct)
+
+    # Messages: drop index, create unique constraint
+    op.drop_index("ix_messages_event_hash", table_name="messages")
+    op.create_unique_constraint("uq_messages_event_hash", "messages", ["event_hash"])
+
+    # Advertisements: drop index, create unique constraint
+    op.drop_index("ix_advertisements_event_hash", table_name="advertisements")
+    op.create_unique_constraint(
+        "uq_advertisements_event_hash", "advertisements", ["event_hash"]
+    )
+
+    # Trace paths: drop index, create unique constraint
+    op.drop_index("ix_trace_paths_event_hash", table_name="trace_paths")
+    op.create_unique_constraint(
+        "uq_trace_paths_event_hash", "trace_paths", ["event_hash"]
+    )
+
+    # Telemetry: drop index, create unique constraint
+    op.drop_index("ix_telemetry_event_hash", table_name="telemetry")
+    op.create_unique_constraint("uq_telemetry_event_hash", "telemetry", ["event_hash"])
+
+
+def downgrade() -> None:
+    # Restore non-unique indexes
+
+    # Telemetry
+    op.drop_constraint("uq_telemetry_event_hash", "telemetry", type_="unique")
+    op.create_index("ix_telemetry_event_hash", "telemetry", ["event_hash"])
+
+    # Trace paths
+    op.drop_constraint("uq_trace_paths_event_hash", "trace_paths", type_="unique")
+    op.create_index("ix_trace_paths_event_hash", "trace_paths", ["event_hash"])
+
+    # Advertisements
+    op.drop_constraint("uq_advertisements_event_hash", "advertisements", type_="unique")
+    op.create_index("ix_advertisements_event_hash", "advertisements", ["event_hash"])
+
+    # Messages
+    op.drop_constraint("uq_messages_event_hash", "messages", type_="unique")
+    op.create_index("ix_messages_event_hash", "messages", ["event_hash"])

docker-compose.yml

@@ -1,10 +1,14 @@
 services:
   # ==========================================================================
-  # MQTT Broker - Eclipse Mosquitto
+  # MQTT Broker - Eclipse Mosquitto (optional, use --profile mqtt)
+  # Most users will connect to an external MQTT broker instead
   # ==========================================================================
   mqtt:
     image: eclipse-mosquitto:2
     container_name: meshcore-mqtt
+    profiles:
+      - all
+      - mqtt
     restart: unless-stopped
     ports:
       - "${MQTT_EXTERNAL_PORT:-1883}:1883"
@@ -24,17 +28,15 @@ services:
   # Interface Receiver - MeshCore device to MQTT bridge (events)
   # ==========================================================================
   interface-receiver:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-interface-receiver
     profiles:
-      - interface-receiver
+      - all
+      - receiver
     restart: unless-stopped
-    depends_on:
-      mqtt:
-        condition: service_healthy
     devices:
       - "${SERIAL_PORT:-/dev/ttyUSB0}:${SERIAL_PORT:-/dev/ttyUSB0}"
     user: root  # Required for device access
@@ -60,17 +62,15 @@ services:
   # Interface Sender - MQTT to MeshCore device bridge (commands)
   # ==========================================================================
   interface-sender:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-interface-sender
     profiles:
-      - interface-sender
+      - all
+      - sender
     restart: unless-stopped
-    depends_on:
-      mqtt:
-        condition: service_healthy
     devices:
       - "${SERIAL_PORT_SENDER:-/dev/ttyUSB1}:${SERIAL_PORT_SENDER:-/dev/ttyUSB1}"
     user: root  # Required for device access
@@ -96,17 +96,15 @@ services:
   # Interface Mock Receiver - For testing without real devices
   # ==========================================================================
   interface-mock-receiver:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-interface-mock-receiver
     profiles:
+      - all
       - mock
     restart: unless-stopped
-    depends_on:
-      mqtt:
-        condition: service_healthy
     environment:
       - LOG_LEVEL=${LOG_LEVEL:-INFO}
       - MQTT_HOST=${MQTT_HOST:-mqtt}
@@ -128,15 +126,15 @@ services:
   # Collector - MQTT subscriber and database storage
   # ==========================================================================
   collector:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-collector
+    profiles:
+      - all
+      - core
     restart: unless-stopped
-    depends_on:
-      mqtt:
-        condition: service_healthy
     volumes:
       - ${DATA_HOME:-./data}:/data
       - ${SEED_HOME:-./seed}:/seed
@@ -175,15 +173,16 @@ services:
   # API Server - REST API for querying data and sending commands
   # ==========================================================================
   api:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-api
+    profiles:
+      - all
+      - core
     restart: unless-stopped
     depends_on:
-      mqtt:
-        condition: service_healthy
       collector:
         condition: service_started
     ports:
@@ -217,11 +216,14 @@ services:
   # Web Dashboard - Web interface for network visualization
   # ==========================================================================
   web:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-web
+    profiles:
+      - all
+      - core
     restart: unless-stopped
     depends_on:
       api:
@@ -254,12 +256,13 @@ services:
   # Database Migrations - Run Alembic migrations
   # ==========================================================================
   db-migrate:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-db-migrate
     profiles:
+      - all
       - migrate
     volumes:
       # Mount data directory (uses collector/meshcore.db)
@@ -274,12 +277,13 @@ services:
   # Seed Data - Import node_tags.json and members.json from SEED_HOME
   # ==========================================================================
   seed:
-    image: ghcr.io/ipnet-mesh/meshcore-hub:main
+    image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest}
     build:
       context: .
       dockerfile: Dockerfile
     container_name: meshcore-seed
     profiles:
+      - all
       - seed
     volumes:
       # Mount data directory for database (read-write)

src/meshcore_hub/__main__.py

@@ -174,6 +174,40 @@ def db_history() -> None:
     command.history(alembic_cfg)
 
 
+@db.command("stamp")
+@click.option(
+    "--revision",
+    type=str,
+    default="head",
+    help="Target revision to stamp (default: head)",
+)
+@click.option(
+    "--database-url",
+    type=str,
+    default=None,
+    envvar="DATABASE_URL",
+    help="Database connection URL",
+)
+def db_stamp(revision: str, database_url: str | None) -> None:
+    """Stamp database with revision without running migrations.
+
+    Use this to mark an existing database as up-to-date when the schema
+    was created before Alembic migrations were introduced.
+    """
+    import os
+    from alembic import command
+    from alembic.config import Config
+
+    click.echo(f"Stamping database with revision: {revision}")
+
+    alembic_cfg = Config("alembic.ini")
+    if database_url:
+        os.environ["DATABASE_URL"] = database_url
+
+    command.stamp(alembic_cfg, revision)
+    click.echo("Database stamped successfully.")
+
+
 # Health check commands for Docker HEALTHCHECK
 @cli.group()
 def health() -> None:

src/meshcore_hub/api/app.py

@@ -32,10 +32,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # Get database URL from app state
     database_url = getattr(app.state, "database_url", "sqlite:///./meshcore.db")
 
-    # Initialize database
+    # Initialize database (schema managed by Alembic migrations)
     logger.info(f"Initializing database: {database_url}")
     _db_manager = DatabaseManager(database_url)
-    _db_manager.create_tables()
 
     yield
 

src/meshcore_hub/api/routes/advertisements.py

@@ -9,8 +9,12 @@ from sqlalchemy.orm import aliased, selectinload
 
 from meshcore_hub.api.auth import RequireRead
 from meshcore_hub.api.dependencies import DbSession
-from meshcore_hub.common.models import Advertisement, Node
-from meshcore_hub.common.schemas.messages import AdvertisementList, AdvertisementRead
+from meshcore_hub.common.models import Advertisement, EventReceiver, Node, NodeTag
+from meshcore_hub.common.schemas.messages import (
+    AdvertisementList,
+    AdvertisementRead,
+    ReceiverInfo,
+)
 
 router = APIRouter()
 
@@ -25,6 +29,62 @@ def _get_friendly_name(node: Optional[Node]) -> Optional[str]:
     return None
 
 
+def _fetch_receivers_for_events(
+    session: DbSession,
+    event_type: str,
+    event_hashes: list[str],
+) -> dict[str, list[ReceiverInfo]]:
+    """Fetch receiver info for a list of events by their hashes."""
+    if not event_hashes:
+        return {}
+
+    query = (
+        select(
+            EventReceiver.event_hash,
+            EventReceiver.snr,
+            EventReceiver.received_at,
+            Node.id.label("node_id"),
+            Node.public_key,
+            Node.name,
+        )
+        .join(Node, EventReceiver.receiver_node_id == Node.id)
+        .where(EventReceiver.event_type == event_type)
+        .where(EventReceiver.event_hash.in_(event_hashes))
+        .order_by(EventReceiver.received_at)
+    )
+
+    results = session.execute(query).all()
+    receivers_by_hash: dict[str, list[ReceiverInfo]] = {}
+
+    node_ids = [r.node_id for r in results]
+    friendly_names: dict[str, str] = {}
+    if node_ids:
+        fn_query = (
+            select(NodeTag.node_id, NodeTag.value)
+            .where(NodeTag.node_id.in_(node_ids))
+            .where(NodeTag.key == "friendly_name")
+        )
+        for node_id, value in session.execute(fn_query).all():
+            friendly_names[node_id] = value
+
+    for row in results:
+        if row.event_hash not in receivers_by_hash:
+            receivers_by_hash[row.event_hash] = []
+
+        receivers_by_hash[row.event_hash].append(
+            ReceiverInfo(
+                node_id=row.node_id,
+                public_key=row.public_key,
+                name=row.name,
+                friendly_name=friendly_names.get(row.node_id),
+                snr=row.snr,
+                received_at=row.received_at,
+            )
+        )
+
+    return receivers_by_hash
+
+
 @router.get("", response_model=AdvertisementList)
 async def list_advertisements(
     _: RequireRead,
@@ -97,6 +157,12 @@ async def list_advertisements(
         nodes = session.execute(nodes_query).scalars().all()
         nodes_by_id = {n.id: n for n in nodes}
 
+    # Fetch all receivers for these advertisements
+    event_hashes = [r[0].event_hash for r in results if r[0].event_hash]
+    receivers_by_hash = _fetch_receivers_for_events(
+        session, "advertisement", event_hashes
+    )
+
     # Build response with node details
     items = []
     for row in results:
@@ -116,6 +182,9 @@ async def list_advertisements(
             "flags": adv.flags,
             "received_at": adv.received_at,
             "created_at": adv.created_at,
+            "receivers": (
+                receivers_by_hash.get(adv.event_hash, []) if adv.event_hash else []
+            ),
         }
         items.append(AdvertisementRead(**data))
 
@@ -175,6 +244,14 @@ async def get_advertisement(
     receiver_node = nodes_by_id.get(result.receiver_id) if result.receiver_id else None
     source_node = nodes_by_id.get(result.source_id) if result.source_id else None
 
+    # Fetch receivers for this advertisement
+    receivers = []
+    if adv.event_hash:
+        receivers_by_hash = _fetch_receivers_for_events(
+            session, "advertisement", [adv.event_hash]
+        )
+        receivers = receivers_by_hash.get(adv.event_hash, [])
+
     data = {
         "received_by": result.receiver_pk,
         "receiver_name": result.receiver_name,
@@ -187,5 +264,6 @@ async def get_advertisement(
         "flags": adv.flags,
         "received_at": adv.received_at,
         "created_at": adv.created_at,
+        "receivers": receivers,
     }
     return AdvertisementRead(**data)

src/meshcore_hub/api/routes/messages.py

@@ -9,8 +9,8 @@ from sqlalchemy.orm import aliased, selectinload
 
 from meshcore_hub.api.auth import RequireRead
 from meshcore_hub.api.dependencies import DbSession
-from meshcore_hub.common.models import Message, Node, NodeTag
-from meshcore_hub.common.schemas.messages import MessageList, MessageRead
+from meshcore_hub.common.models import EventReceiver, Message, Node, NodeTag
+from meshcore_hub.common.schemas.messages import MessageList, MessageRead, ReceiverInfo
 
 router = APIRouter()
 
@@ -25,6 +25,75 @@ def _get_friendly_name(node: Optional[Node]) -> Optional[str]:
     return None
 
 
+def _fetch_receivers_for_events(
+    session: DbSession,
+    event_type: str,
+    event_hashes: list[str],
+) -> dict[str, list[ReceiverInfo]]:
+    """Fetch receiver info for a list of events by their hashes.
+
+    Args:
+        session: Database session
+        event_type: Type of event ('message', 'advertisement', etc.)
+        event_hashes: List of event hashes to fetch receivers for
+
+    Returns:
+        Dict mapping event_hash to list of ReceiverInfo objects
+    """
+    if not event_hashes:
+        return {}
+
+    # Query event_receivers with receiver node info
+    query = (
+        select(
+            EventReceiver.event_hash,
+            EventReceiver.snr,
+            EventReceiver.received_at,
+            Node.id.label("node_id"),
+            Node.public_key,
+            Node.name,
+        )
+        .join(Node, EventReceiver.receiver_node_id == Node.id)
+        .where(EventReceiver.event_type == event_type)
+        .where(EventReceiver.event_hash.in_(event_hashes))
+        .order_by(EventReceiver.received_at)
+    )
+
+    results = session.execute(query).all()
+
+    # Group by event_hash
+    receivers_by_hash: dict[str, list[ReceiverInfo]] = {}
+
+    # Get friendly names for receiver nodes
+    node_ids = [r.node_id for r in results]
+    friendly_names: dict[str, str] = {}
+    if node_ids:
+        fn_query = (
+            select(NodeTag.node_id, NodeTag.value)
+            .where(NodeTag.node_id.in_(node_ids))
+            .where(NodeTag.key == "friendly_name")
+        )
+        for node_id, value in session.execute(fn_query).all():
+            friendly_names[node_id] = value
+
+    for row in results:
+        if row.event_hash not in receivers_by_hash:
+            receivers_by_hash[row.event_hash] = []
+
+        receivers_by_hash[row.event_hash].append(
+            ReceiverInfo(
+                node_id=row.node_id,
+                public_key=row.public_key,
+                name=row.name,
+                friendly_name=friendly_names.get(row.node_id),
+                snr=row.snr,
+                received_at=row.received_at,
+            )
+        )
+
+    return receivers_by_hash
+
+
 @router.get("", response_model=MessageList)
 async def list_messages(
     _: RequireRead,
@@ -126,6 +195,10 @@ async def list_messages(
         receivers = session.execute(receivers_query).scalars().all()
         receivers_by_id = {n.id: n for n in receivers}
 
+    # Fetch all receivers for these messages
+    event_hashes = [r[0].event_hash for r in results if r[0].event_hash]
+    receivers_by_hash = _fetch_receivers_for_events(session, "message", event_hashes)
+
     # Build response with sender info and received_by
     items = []
     for row in results:
@@ -159,6 +232,9 @@ async def list_messages(
             "sender_timestamp": m.sender_timestamp,
             "received_at": m.received_at,
             "created_at": m.created_at,
+            "receivers": (
+                receivers_by_hash.get(m.event_hash, []) if m.event_hash else []
+            ),
         }
         items.append(MessageRead(**msg_dict))
 
@@ -189,6 +265,15 @@ async def get_message(
         raise HTTPException(status_code=404, detail="Message not found")
 
     message, receiver_pk = result
+
+    # Fetch receivers for this message
+    receivers = []
+    if message.event_hash:
+        receivers_by_hash = _fetch_receivers_for_events(
+            session, "message", [message.event_hash]
+        )
+        receivers = receivers_by_hash.get(message.event_hash, [])
+
     data = {
         "id": message.id,
         "receiver_node_id": message.receiver_node_id,
@@ -204,5 +289,6 @@ async def get_message(
         "sender_timestamp": message.sender_timestamp,
         "received_at": message.received_at,
         "created_at": message.created_at,
+        "receivers": receivers,
     }
     return MessageRead(**data)

src/meshcore_hub/collector/cli.py

@@ -182,11 +182,10 @@ def _run_collector_service(
     click.echo(f"MQTT: {mqtt_host}:{mqtt_port} (prefix: {prefix})")
     click.echo(f"Database: {database_url}")
 
-    # Initialize database and run seed import on startup
+    # Initialize database (schema managed by Alembic migrations)
     from meshcore_hub.common.database import DatabaseManager
 
     db = DatabaseManager(database_url)
-    db.create_tables()
 
     # Auto-seed from seed files on startup
     click.echo("")
@@ -294,9 +293,8 @@ def seed_cmd(
 
     from meshcore_hub.common.database import DatabaseManager
 
-    # Initialize database
+    # Initialize database (schema managed by Alembic migrations)
     db = DatabaseManager(ctx.obj["database_url"])
-    db.create_tables()
 
     # Run seed import
     imported_any = _run_seed_import(
@@ -448,9 +446,8 @@ def import_tags_cmd(
     from meshcore_hub.common.database import DatabaseManager
     from meshcore_hub.collector.tag_import import import_tags
 
-    # Initialize database
+    # Initialize database (schema managed by Alembic migrations)
     db = DatabaseManager(ctx.obj["database_url"])
-    db.create_tables()
 
     # Import tags
     stats = import_tags(
@@ -529,9 +526,8 @@ def import_members_cmd(
     from meshcore_hub.common.database import DatabaseManager
     from meshcore_hub.collector.member_import import import_members
 
-    # Initialize database
+    # Initialize database (schema managed by Alembic migrations)
     db = DatabaseManager(ctx.obj["database_url"])
-    db.create_tables()
 
     # Import members
     stats = import_members(

src/meshcore_hub/collector/handlers/advertisement.py

@@ -5,9 +5,11 @@ from datetime import datetime, timezone
 from typing import Any
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 
 from meshcore_hub.common.database import DatabaseManager
-from meshcore_hub.common.models import Advertisement, Node
+from meshcore_hub.common.hash_utils import compute_advertisement_hash
+from meshcore_hub.common.models import Advertisement, Node, add_event_receiver
 
 logger = logging.getLogger(__name__)
 
@@ -40,8 +42,17 @@ def handle_advertisement(
     flags = payload.get("flags")
     now = datetime.now(timezone.utc)
 
+    # Compute event hash for deduplication (30-second time bucket)
+    event_hash = compute_advertisement_hash(
+        public_key=adv_public_key,
+        name=name,
+        adv_type=adv_type,
+        flags=flags,
+        received_at=now,
+    )
+
     with db.session_scope() as session:
-        # Find or create receiver node
+        # Find or create receiver node first (needed for both new and duplicate events)
         receiver_node = None
         if public_key:
             receiver_query = select(Node).where(Node.public_key == public_key)
@@ -55,6 +66,37 @@ def handle_advertisement(
                 )
                 session.add(receiver_node)
                 session.flush()
+            else:
+                receiver_node.last_seen = now
+
+        # Check if advertisement with same hash already exists
+        existing = session.execute(
+            select(Advertisement.id).where(Advertisement.event_hash == event_hash)
+        ).scalar_one_or_none()
+
+        if existing:
+            # Still update advertised node's last_seen even for duplicate advertisements
+            node_query = select(Node).where(Node.public_key == adv_public_key)
+            node = session.execute(node_query).scalar_one_or_none()
+            if node:
+                node.last_seen = now
+
+            # Add this receiver to the junction table
+            if receiver_node:
+                added = add_event_receiver(
+                    session=session,
+                    event_type="advertisement",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=None,  # Advertisements don't have SNR
+                    received_at=now,
+                )
+                if added:
+                    logger.debug(
+                        f"Added receiver {public_key[:12]}... to advertisement "
+                        f"(hash={event_hash[:8]}...)"
+                    )
+            return
 
         # Find or create advertised node
         node_query = select(Node).where(Node.public_key == adv_public_key)
@@ -91,9 +133,43 @@ def handle_advertisement(
             adv_type=adv_type,
             flags=flags,
             received_at=now,
+            event_hash=event_hash,
         )
         session.add(advertisement)
 
+        # Add first receiver to junction table
+        if receiver_node:
+            add_event_receiver(
+                session=session,
+                event_type="advertisement",
+                event_hash=event_hash,
+                receiver_node_id=receiver_node.id,
+                snr=None,
+                received_at=now,
+            )
+
+        # Flush to check for duplicate constraint violation (race condition)
+        try:
+            session.flush()
+        except IntegrityError:
+            # Race condition: another request inserted the same event_hash
+            session.rollback()
+            logger.debug(
+                f"Duplicate advertisement skipped (race condition, "
+                f"hash={event_hash[:8]}...)"
+            )
+            # Re-add receiver to existing event in a new transaction
+            if receiver_node:
+                add_event_receiver(
+                    session=session,
+                    event_type="advertisement",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=None,
+                    received_at=now,
+                )
+            return
+
     logger.info(
         f"Stored advertisement from {name or adv_public_key[:12]!r} "
         f"(type={adv_type})"

src/meshcore_hub/collector/handlers/message.py

@@ -5,9 +5,11 @@ from datetime import datetime, timezone
 from typing import Any
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 
 from meshcore_hub.common.database import DatabaseManager
-from meshcore_hub.common.models import Message, Node
+from meshcore_hub.common.hash_utils import compute_message_hash
+from meshcore_hub.common.models import Message, Node, add_event_receiver
 
 logger = logging.getLogger(__name__)
 
@@ -84,8 +86,17 @@ def _handle_message(
         except (ValueError, OSError):
             pass
 
+    # Compute event hash for deduplication
+    event_hash = compute_message_hash(
+        text=text,
+        pubkey_prefix=pubkey_prefix,
+        channel_idx=channel_idx,
+        sender_timestamp=sender_timestamp,
+        txt_type=txt_type,
+    )
+
     with db.session_scope() as session:
-        # Find receiver node
+        # Find or create receiver node first (needed for both new and duplicate events)
         receiver_node = None
         if public_key:
             receiver_query = select(Node).where(Node.public_key == public_key)
@@ -102,6 +113,29 @@ def _handle_message(
             else:
                 receiver_node.last_seen = now
 
+        # Check if message with same hash already exists
+        existing = session.execute(
+            select(Message.id).where(Message.event_hash == event_hash)
+        ).scalar_one_or_none()
+
+        if existing:
+            # Event already exists - just add this receiver to the junction table
+            if receiver_node:
+                added = add_event_receiver(
+                    session=session,
+                    event_type="message",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=snr,
+                    received_at=now,
+                )
+                if added:
+                    logger.debug(
+                        f"Added receiver {public_key[:12]}... to message "
+                        f"(hash={event_hash[:8]}...)"
+                    )
+            return
+
         # Create message record
         message = Message(
             receiver_node_id=receiver_node.id if receiver_node else None,
@@ -115,9 +149,42 @@ def _handle_message(
             snr=snr,
             sender_timestamp=sender_timestamp,
             received_at=now,
+            event_hash=event_hash,
         )
         session.add(message)
 
+        # Add first receiver to junction table
+        if receiver_node:
+            add_event_receiver(
+                session=session,
+                event_type="message",
+                event_hash=event_hash,
+                receiver_node_id=receiver_node.id,
+                snr=snr,
+                received_at=now,
+            )
+
+        # Flush to check for duplicate constraint violation (race condition)
+        try:
+            session.flush()
+        except IntegrityError:
+            # Race condition: another request inserted the same event_hash
+            session.rollback()
+            logger.debug(
+                f"Duplicate message skipped (race condition, hash={event_hash[:8]}...)"
+            )
+            # Re-add receiver to existing event in a new transaction
+            if receiver_node:
+                add_event_receiver(
+                    session=session,
+                    event_type="message",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=snr,
+                    received_at=now,
+                )
+            return
+
     if message_type == "contact":
         logger.info(
             f"Stored contact message from {pubkey_prefix!r}: "

src/meshcore_hub/collector/handlers/telemetry.py

@@ -5,9 +5,11 @@ from datetime import datetime, timezone
 from typing import Any
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 
 from meshcore_hub.common.database import DatabaseManager
-from meshcore_hub.common.models import Node, Telemetry
+from meshcore_hub.common.hash_utils import compute_telemetry_hash
+from meshcore_hub.common.models import Node, Telemetry, add_event_receiver
 
 logger = logging.getLogger(__name__)
 
@@ -49,8 +51,15 @@ def handle_telemetry(
             except ValueError:
                 lpp_bytes = lpp_data.encode()
 
+    # Compute event hash for deduplication (30-second time bucket)
+    event_hash = compute_telemetry_hash(
+        node_public_key=node_public_key,
+        parsed_data=parsed_data,
+        received_at=now,
+    )
+
     with db.session_scope() as session:
-        # Find receiver node
+        # Find or create receiver node first (needed for both new and duplicate events)
         receiver_node = None
         if public_key:
             receiver_query = select(Node).where(Node.public_key == public_key)
@@ -67,6 +76,29 @@ def handle_telemetry(
             else:
                 receiver_node.last_seen = now
 
+        # Check if telemetry with same hash already exists
+        existing = session.execute(
+            select(Telemetry.id).where(Telemetry.event_hash == event_hash)
+        ).scalar_one_or_none()
+
+        if existing:
+            # Event already exists - just add this receiver to the junction table
+            if receiver_node:
+                added = add_event_receiver(
+                    session=session,
+                    event_type="telemetry",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=None,
+                    received_at=now,
+                )
+                if added:
+                    logger.debug(
+                        f"Added receiver {public_key[:12]}... to telemetry "
+                        f"(node={node_public_key[:12]}...)"
+                    )
+            return
+
         # Find or create reporting node
         reporting_node = None
         if node_public_key:
@@ -92,9 +124,43 @@ def handle_telemetry(
             lpp_data=lpp_bytes,
             parsed_data=parsed_data,
             received_at=now,
+            event_hash=event_hash,
         )
         session.add(telemetry)
 
+        # Add first receiver to junction table
+        if receiver_node:
+            add_event_receiver(
+                session=session,
+                event_type="telemetry",
+                event_hash=event_hash,
+                receiver_node_id=receiver_node.id,
+                snr=None,
+                received_at=now,
+            )
+
+        # Flush to check for duplicate constraint violation (race condition)
+        try:
+            session.flush()
+        except IntegrityError:
+            # Race condition: another request inserted the same event_hash
+            session.rollback()
+            logger.debug(
+                f"Duplicate telemetry skipped (race condition, "
+                f"node={node_public_key[:12]}...)"
+            )
+            # Re-add receiver to existing event in a new transaction
+            if receiver_node:
+                add_event_receiver(
+                    session=session,
+                    event_type="telemetry",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=None,
+                    received_at=now,
+                )
+            return
+
     # Log telemetry values
     if parsed_data:
         values = ", ".join(f"{k}={v}" for k, v in parsed_data.items())

src/meshcore_hub/collector/handlers/trace.py

@@ -5,9 +5,11 @@ from datetime import datetime, timezone
 from typing import Any
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 
 from meshcore_hub.common.database import DatabaseManager
-from meshcore_hub.common.models import Node, TracePath
+from meshcore_hub.common.hash_utils import compute_trace_hash
+from meshcore_hub.common.models import Node, TracePath, add_event_receiver
 
 logger = logging.getLogger(__name__)
 
@@ -40,8 +42,11 @@ def handle_trace_data(
     snr_values = payload.get("snr_values")
     hop_count = payload.get("hop_count")
 
+    # Compute event hash for deduplication (initiator_tag is unique per trace)
+    event_hash = compute_trace_hash(initiator_tag=initiator_tag)
+
     with db.session_scope() as session:
-        # Find receiver node
+        # Find or create receiver node first (needed for both new and duplicate events)
         receiver_node = None
         if public_key:
             receiver_query = select(Node).where(Node.public_key == public_key)
@@ -58,6 +63,29 @@ def handle_trace_data(
             else:
                 receiver_node.last_seen = now
 
+        # Check if trace with same hash already exists
+        existing = session.execute(
+            select(TracePath.id).where(TracePath.event_hash == event_hash)
+        ).scalar_one_or_none()
+
+        if existing:
+            # Event already exists - just add this receiver to the junction table
+            if receiver_node:
+                added = add_event_receiver(
+                    session=session,
+                    event_type="trace",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=None,  # Trace events don't have a single SNR value
+                    received_at=now,
+                )
+                if added:
+                    logger.debug(
+                        f"Added receiver {public_key[:12]}... to trace "
+                        f"(tag={initiator_tag})"
+                    )
+            return
+
         # Create trace path record
         trace_path = TracePath(
             receiver_node_id=receiver_node.id if receiver_node else None,
@@ -69,7 +97,40 @@ def handle_trace_data(
             snr_values=snr_values,
             hop_count=hop_count,
             received_at=now,
+            event_hash=event_hash,
         )
         session.add(trace_path)
 
+        # Add first receiver to junction table
+        if receiver_node:
+            add_event_receiver(
+                session=session,
+                event_type="trace",
+                event_hash=event_hash,
+                receiver_node_id=receiver_node.id,
+                snr=None,
+                received_at=now,
+            )
+
+        # Flush to check for duplicate constraint violation (race condition)
+        try:
+            session.flush()
+        except IntegrityError:
+            # Race condition: another request inserted the same event_hash
+            session.rollback()
+            logger.debug(
+                f"Duplicate trace skipped (race condition, tag={initiator_tag})"
+            )
+            # Re-add receiver to existing event in a new transaction
+            if receiver_node:
+                add_event_receiver(
+                    session=session,
+                    event_type="trace",
+                    event_hash=event_hash,
+                    receiver_node_id=receiver_node.id,
+                    snr=None,
+                    received_at=now,
+                )
+            return
+
     logger.info(f"Stored trace data: tag={initiator_tag}, hops={hop_count}")

src/meshcore_hub/collector/subscriber.py

@@ -206,14 +206,16 @@ class Subscriber:
         """Start the subscriber."""
         logger.info("Starting collector subscriber")
 
-        # Create database tables if needed
+        # Verify database connection (schema managed by Alembic migrations)
         try:
-            self.db.create_tables()
+            # Test connection by getting a session
+            session = self.db.get_session()
+            session.close()
             self._db_connected = True
-            logger.info("Database initialized")
+            logger.info("Database connection verified")
         except Exception as e:
             self._db_connected = False
-            logger.error(f"Failed to initialize database: {e}")
+            logger.error(f"Failed to connect to database: {e}")
             raise
 
         # Connect to MQTT broker

src/meshcore_hub/common/hash_utils.py

@@ -0,0 +1,142 @@
+"""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_seconds: int = 30,
+) -> 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_seconds: Time bucket size in seconds (default 30)
+
+    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
+        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_seconds: int = 30,
+) -> 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_seconds: Time bucket size in seconds (default 30)
+
+    Returns:
+        32-character hex hash string
+    """
+    # Bucket the time
+    time_bucket = ""
+    if received_at:
+        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()

src/meshcore_hub/common/models/__init__.py

@@ -10,6 +10,7 @@ from meshcore_hub.common.models.telemetry import Telemetry
 from meshcore_hub.common.models.event_log import EventLog
 from meshcore_hub.common.models.member import Member
 from meshcore_hub.common.models.member_node import MemberNode
+from meshcore_hub.common.models.event_receiver import EventReceiver, add_event_receiver
 
 __all__ = [
     "Base",
@@ -23,4 +24,6 @@ __all__ = [
     "EventLog",
     "Member",
     "MemberNode",
+    "EventReceiver",
+    "add_event_receiver",
 ]

src/meshcore_hub/common/models/advertisement.py

@@ -58,6 +58,11 @@ class Advertisement(Base, UUIDMixin, TimestampMixin):
         default=utc_now,
         nullable=False,
     )
+    event_hash: Mapped[Optional[str]] = mapped_column(
+        String(32),
+        nullable=True,
+        unique=True,
+    )
 
     __table_args__ = (Index("ix_advertisements_received_at", "received_at"),)
 

src/meshcore_hub/common/models/event_receiver.py

@@ -0,0 +1,127 @@
+"""EventReceiver model for tracking which nodes received each event."""
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Optional
+from uuid import uuid4
+
+from sqlalchemy import DateTime, Float, ForeignKey, Index, String, UniqueConstraint
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+from sqlalchemy.orm import Mapped, Session, mapped_column, relationship
+
+from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
+
+if TYPE_CHECKING:
+    from meshcore_hub.common.models.node import Node
+
+
+class EventReceiver(Base, UUIDMixin, TimestampMixin):
+    """Junction model tracking which receivers observed each event.
+
+    This table enables multi-receiver tracking for deduplicated events.
+    When multiple receiver nodes observe the same mesh event, each receiver
+    gets an entry in this table linked by the event_hash.
+
+    Attributes:
+        id: UUID primary key
+        event_type: Type of event ('message', 'advertisement', 'trace', 'telemetry')
+        event_hash: Hash identifying the unique event (links to event tables)
+        receiver_node_id: FK to the node that received this event
+        snr: Signal-to-noise ratio at this receiver (if available)
+        received_at: When this specific receiver saw the event
+        created_at: Record creation timestamp
+        updated_at: Record update timestamp
+    """
+
+    __tablename__ = "event_receivers"
+
+    event_type: Mapped[str] = mapped_column(
+        String(20),
+        nullable=False,
+    )
+    event_hash: Mapped[str] = mapped_column(
+        String(32),
+        nullable=False,
+        index=True,
+    )
+    receiver_node_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("nodes.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    snr: Mapped[Optional[float]] = mapped_column(
+        Float,
+        nullable=True,
+    )
+    received_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=utc_now,
+        nullable=False,
+    )
+
+    # Relationship to receiver node
+    receiver_node: Mapped["Node"] = relationship(
+        "Node",
+        foreign_keys=[receiver_node_id],
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "event_hash", "receiver_node_id", name="uq_event_receivers_hash_node"
+        ),
+        Index("ix_event_receivers_type_hash", "event_type", "event_hash"),
+    )
+
+    def __repr__(self) -> str:
+        return (
+            f"<EventReceiver(type={self.event_type}, "
+            f"hash={self.event_hash[:8]}..., "
+            f"node={self.receiver_node_id[:8]}...)>"
+        )
+
+
+def add_event_receiver(
+    session: Session,
+    event_type: str,
+    event_hash: str,
+    receiver_node_id: str,
+    snr: Optional[float] = None,
+    received_at: Optional[datetime] = None,
+) -> bool:
+    """Add a receiver to an event, handling duplicates gracefully.
+
+    Uses INSERT OR IGNORE to handle the unique constraint on (event_hash, receiver_node_id).
+
+    Args:
+        session: SQLAlchemy session
+        event_type: Type of event ('message', 'advertisement', 'trace', 'telemetry')
+        event_hash: Hash identifying the unique event
+        receiver_node_id: UUID of the receiver node
+        snr: Signal-to-noise ratio at this receiver (optional)
+        received_at: When this receiver saw the event (defaults to now)
+
+    Returns:
+        True if a new receiver entry was added, False if it already existed.
+    """
+    from datetime import timezone
+
+    now = received_at or datetime.now(timezone.utc)
+
+    stmt = (
+        sqlite_insert(EventReceiver)
+        .values(
+            id=str(uuid4()),
+            event_type=event_type,
+            event_hash=event_hash,
+            receiver_node_id=receiver_node_id,
+            snr=snr,
+            received_at=now,
+            created_at=now,
+            updated_at=now,
+        )
+        .on_conflict_do_nothing(index_elements=["event_hash", "receiver_node_id"])
+    )
+    result = session.execute(stmt)
+    # CursorResult has rowcount attribute
+    rowcount = getattr(result, "rowcount", 0)
+    return bool(rowcount and rowcount > 0)

src/meshcore_hub/common/models/message.py

@@ -76,6 +76,11 @@ class Message(Base, UUIDMixin, TimestampMixin):
         default=utc_now,
         nullable=False,
     )
+    event_hash: Mapped[Optional[str]] = mapped_column(
+        String(32),
+        nullable=True,
+        unique=True,
+    )
 
     __table_args__ = (
         Index("ix_messages_message_type", "message_type"),

src/meshcore_hub/common/models/telemetry.py

@@ -54,6 +54,11 @@ class Telemetry(Base, UUIDMixin, TimestampMixin):
         default=utc_now,
         nullable=False,
     )
+    event_hash: Mapped[Optional[str]] = mapped_column(
+        String(32),
+        nullable=True,
+        unique=True,
+    )
 
     __table_args__ = (Index("ix_telemetry_received_at", "received_at"),)
 

src/meshcore_hub/common/models/trace_path.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from typing import Optional
 
-from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, String
 from sqlalchemy.dialects.sqlite import JSON
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -67,6 +67,11 @@ class TracePath(Base, UUIDMixin, TimestampMixin):
         default=utc_now,
         nullable=False,
     )
+    event_hash: Mapped[Optional[str]] = mapped_column(
+        String(32),
+        nullable=True,
+        unique=True,
+    )
 
     __table_args__ = (
         Index("ix_trace_paths_initiator_tag", "initiator_tag"),

src/meshcore_hub/common/schemas/__init__.py

@@ -20,6 +20,7 @@ from meshcore_hub.common.schemas.nodes import (
     NodeTagRead,
 )
 from meshcore_hub.common.schemas.messages import (
+    ReceiverInfo,
     MessageRead,
     MessageList,
     MessageFilters,
@@ -57,7 +58,8 @@ __all__ = [
     "NodeTagCreate",
     "NodeTagUpdate",
     "NodeTagRead",
-    # Messages
+    # Messages & Events
+    "ReceiverInfo",
     "MessageRead",
     "MessageList",
     "MessageFilters",

src/meshcore_hub/common/schemas/messages.py

@@ -6,6 +6,24 @@ from typing import Literal, Optional
 from pydantic import BaseModel, Field
 
 
+class ReceiverInfo(BaseModel):
+    """Information about a receiver that observed an event."""
+
+    node_id: str = Field(..., description="Receiver node UUID")
+    public_key: str = Field(..., description="Receiver node public key")
+    name: Optional[str] = Field(default=None, description="Receiver node name")
+    friendly_name: Optional[str] = Field(
+        default=None, description="Receiver friendly name from tags"
+    )
+    snr: Optional[float] = Field(
+        default=None, description="Signal-to-noise ratio at this receiver"
+    )
+    received_at: datetime = Field(..., description="When this receiver saw the event")
+
+    class Config:
+        from_attributes = True
+
+
 class MessageRead(BaseModel):
     """Schema for reading a message."""
 
@@ -37,6 +55,9 @@ class MessageRead(BaseModel):
     )
     received_at: datetime = Field(..., description="When received by interface")
     created_at: datetime = Field(..., description="Record creation timestamp")
+    receivers: list[ReceiverInfo] = Field(
+        default_factory=list, description="All receivers that observed this message"
+    )
 
     class Config:
         from_attributes = True
@@ -104,6 +125,10 @@ class AdvertisementRead(BaseModel):
     flags: Optional[int] = Field(default=None, description="Capability flags")
     received_at: datetime = Field(..., description="When received")
     created_at: datetime = Field(..., description="Record creation timestamp")
+    receivers: list[ReceiverInfo] = Field(
+        default_factory=list,
+        description="All receivers that observed this advertisement",
+    )
 
     class Config:
         from_attributes = True
@@ -137,6 +162,10 @@ class TracePathRead(BaseModel):
     hop_count: Optional[int] = Field(default=None, description="Total hops")
     received_at: datetime = Field(..., description="When received")
     created_at: datetime = Field(..., description="Record creation timestamp")
+    receivers: list[ReceiverInfo] = Field(
+        default_factory=list,
+        description="All receivers that observed this trace",
+    )
 
     class Config:
         from_attributes = True
@@ -163,6 +192,10 @@ class TelemetryRead(BaseModel):
     )
     received_at: datetime = Field(..., description="When received")
     created_at: datetime = Field(..., description="Record creation timestamp")
+    receivers: list[ReceiverInfo] = Field(
+        default_factory=list,
+        description="All receivers that observed this telemetry",
+    )
 
     class Config:
         from_attributes = True

src/meshcore_hub/interface/receiver.py

@@ -197,17 +197,19 @@ class Receiver:
             logger.error(f"Failed to connect to MQTT broker: {e}")
             raise
 
-        # Connect to device
-        if not self.device.connect():
-            self._device_connected = False
-            logger.error("Failed to connect to MeshCore device")
-            self.mqtt.stop()
-            self.mqtt.disconnect()
-            self._mqtt_connected = False
-            raise RuntimeError("Failed to connect to MeshCore device")
+        # Device should already be connected (from create_receiver)
+        # but handle case where start() is called directly
+        if not self.device.is_connected:
+            if not self.device.connect():
+                self._device_connected = False
+                logger.error("Failed to connect to MeshCore device")
+                self.mqtt.stop()
+                self.mqtt.disconnect()
+                self._mqtt_connected = False
+                raise RuntimeError("Failed to connect to MeshCore device")
+            logger.info(f"Connected to MeshCore device: {self.device.public_key}")
 
         self._device_connected = True
-        logger.info(f"Connected to MeshCore device: {self.device.public_key}")
 
         # Initialize device: set time and send local advertisement
         self._initialize_device()
@@ -291,17 +293,22 @@ def create_receiver(
     Returns:
         Configured Receiver instance
     """
-    # Create device
+    # Create and connect device first to get public key
     device = create_device(port=port, baud=baud, mock=mock, node_address=node_address)
 
-    # Create MQTT client
+    if not device.connect():
+        raise RuntimeError("Failed to connect to MeshCore device")
+
+    logger.info(f"Connected to MeshCore device: {device.public_key}")
+
+    # Create MQTT client with device's public key for unique client ID
     mqtt_config = MQTTConfig(
         host=mqtt_host,
         port=mqtt_port,
         username=mqtt_username,
         password=mqtt_password,
         prefix=mqtt_prefix,
-        client_id=f"meshcore-receiver-{device.public_key[:8] if device.public_key else 'unknown'}",
+        client_id=f"meshcore-receiver-{device.public_key[:12] if device.public_key else 'unknown'}",
     )
     mqtt_client = MQTTClient(mqtt_config)
 

src/meshcore_hub/interface/sender.py

@@ -200,14 +200,16 @@ class Sender:
         """Start the sender."""
         logger.info("Starting SENDER mode")
 
-        # Connect to device first
-        if not self.device.connect():
-            self._device_connected = False
-            logger.error("Failed to connect to MeshCore device")
-            raise RuntimeError("Failed to connect to MeshCore device")
+        # Device should already be connected (from create_sender)
+        # but handle case where start() is called directly
+        if not self.device.is_connected:
+            if not self.device.connect():
+                self._device_connected = False
+                logger.error("Failed to connect to MeshCore device")
+                raise RuntimeError("Failed to connect to MeshCore device")
+            logger.info(f"Connected to MeshCore device: {self.device.public_key}")
 
         self._device_connected = True
-        logger.info(f"Connected to MeshCore device: {self.device.public_key}")
 
         # Connect to MQTT broker
         try:
@@ -307,17 +309,22 @@ def create_sender(
     Returns:
         Configured Sender instance
     """
-    # Create device
+    # Create and connect device first to get public key
     device = create_device(port=port, baud=baud, mock=mock, node_address=node_address)
 
-    # Create MQTT client
+    if not device.connect():
+        raise RuntimeError("Failed to connect to MeshCore device")
+
+    logger.info(f"Connected to MeshCore device: {device.public_key}")
+
+    # Create MQTT client with device's public key for unique client ID
     mqtt_config = MQTTConfig(
         host=mqtt_host,
         port=mqtt_port,
         username=mqtt_username,
         password=mqtt_password,
         prefix=mqtt_prefix,
-        client_id=f"meshcore-sender-{device.public_key[:8] if device.public_key else 'unknown'}",
+        client_id=f"meshcore-sender-{device.public_key[:12] if device.public_key else 'unknown'}",
     )
     mqtt_client = MQTTClient(mqtt_config)
 

src/meshcore_hub/web/templates/advertisements.html

@@ -71,7 +71,31 @@
                     {% endif %}
                 </td>
                 <td>
-                    {% if ad.received_by %}
+                    {% if ad.receivers and ad.receivers|length > 1 %}
+                    <div class="dropdown dropdown-hover dropdown-end">
+                        <label tabindex="0" class="badge badge-outline badge-sm cursor-pointer">
+                            {{ ad.receivers|length }} receivers
+                        </label>
+                        <ul tabindex="0" class="dropdown-content z-[1] menu p-2 shadow bg-base-100 rounded-box w-56">
+                            {% for recv in ad.receivers %}
+                            <li>
+                                <a href="/nodes/{{ recv.public_key }}" class="text-sm">
+                                    {{ recv.friendly_name or recv.name or recv.public_key[:12] + '...' }}
+                                </a>
+                            </li>
+                            {% endfor %}
+                        </ul>
+                    </div>
+                    {% elif ad.receivers and ad.receivers|length == 1 %}
+                    <a href="/nodes/{{ ad.receivers[0].public_key }}" class="link link-hover">
+                        {% if ad.receivers[0].friendly_name or ad.receivers[0].name %}
+                        <div class="font-medium">{{ ad.receivers[0].friendly_name or ad.receivers[0].name }}</div>
+                        <div class="text-xs font-mono opacity-70">{{ ad.receivers[0].public_key[:16] }}...</div>
+                        {% else %}
+                        <span class="font-mono text-sm">{{ ad.receivers[0].public_key[:16] }}...</span>
+                        {% endif %}
+                    </a>
+                    {% elif ad.received_by %}
                     <a href="/nodes/{{ ad.received_by }}" class="link link-hover">
                         {% if ad.receiver_friendly_name or ad.receiver_name %}
                         <div class="font-medium">{{ ad.receiver_friendly_name or ad.receiver_name }}</div>

src/meshcore_hub/web/templates/messages.html

@@ -87,7 +87,34 @@
                 </td>
                 <td class="break-words max-w-md" style="white-space: pre-wrap;">{{ msg.text or '-' }}</td>
                 <td>
-                    {% if msg.received_by %}
+                    {% if msg.receivers and msg.receivers|length > 1 %}
+                    <div class="dropdown dropdown-hover dropdown-end">
+                        <label tabindex="0" class="badge badge-outline badge-sm cursor-pointer">
+                            {{ msg.receivers|length }} receivers
+                        </label>
+                        <ul tabindex="0" class="dropdown-content z-[1] menu p-2 shadow bg-base-100 rounded-box w-56">
+                            {% for recv in msg.receivers %}
+                            <li>
+                                <a href="/nodes/{{ recv.public_key }}" class="text-sm">
+                                    <span class="flex-1">{{ recv.friendly_name or recv.name or recv.public_key[:12] + '...' }}</span>
+                                    {% if recv.snr is not none %}
+                                    <span class="badge badge-ghost badge-xs">{{ "%.1f"|format(recv.snr) }}</span>
+                                    {% endif %}
+                                </a>
+                            </li>
+                            {% endfor %}
+                        </ul>
+                    </div>
+                    {% elif msg.receivers and msg.receivers|length == 1 %}
+                    <a href="/nodes/{{ msg.receivers[0].public_key }}" class="link link-hover">
+                        {% if msg.receivers[0].friendly_name or msg.receivers[0].name %}
+                        <div class="font-medium">{{ msg.receivers[0].friendly_name or msg.receivers[0].name }}</div>
+                        <div class="text-xs font-mono opacity-70">{{ msg.receivers[0].public_key[:16] }}...</div>
+                        {% else %}
+                        <span class="font-mono text-sm">{{ msg.receivers[0].public_key[:16] }}...</span>
+                        {% endif %}
+                    </a>
+                    {% elif msg.received_by %}
                     <a href="/nodes/{{ msg.received_by }}" class="link link-hover">
                         {% if msg.receiver_friendly_name or msg.receiver_name %}
                         <div class="font-medium">{{ msg.receiver_friendly_name or msg.receiver_name }}</div>

tests/test_common/test_hash_utils.py

@@ -0,0 +1,266 @@
+"""Tests for hash utilities for event deduplication."""
+
+from datetime import datetime, timezone
+
+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 (300 second) 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_seconds=300,  # 5 minutes
+        )
+        hash2 = compute_advertisement_hash(
+            public_key="a" * 64,
+            name="Node1",
+            adv_type="chat",
+            flags=128,
+            received_at=time2,
+            bucket_seconds=300,  # 5 minutes
+        )
+
+        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 (300 second) 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_seconds=300,  # 5 minutes
+        )
+        hash2 = compute_advertisement_hash(
+            public_key="a" * 64,
+            name="Node1",
+            received_at=time2,
+            bucket_seconds=300,  # 5 minutes
+        )
+
+        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 (300s) bucket, these should be in different buckets
+        hash1_5min = compute_advertisement_hash(
+            public_key="a" * 64,
+            received_at=time1,
+            bucket_seconds=300,  # 5 minutes
+        )
+        hash2_5min = compute_advertisement_hash(
+            public_key="a" * 64,
+            received_at=time2,
+            bucket_seconds=300,  # 5 minutes
+        )
+        assert hash1_5min != hash2_5min
+
+        # With 10-minute (600s) bucket, these should be in the same bucket
+        hash1_10min = compute_advertisement_hash(
+            public_key="a" * 64,
+            received_at=time1,
+            bucket_seconds=600,  # 10 minutes
+        )
+        hash2_10min = compute_advertisement_hash(
+            public_key="a" * 64,
+            received_at=time2,
+            bucket_seconds=600,  # 10 minutes
+        )
+        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_seconds=300,  # 5 minutes
+        )
+        hash2 = compute_telemetry_hash(
+            node_public_key="a" * 64,
+            parsed_data=data,
+            received_at=time2,
+            bucket_seconds=300,  # 5 minutes
+        )
+
+        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