Implement Phase 3: Collector Component

This commit adds the complete Collector component for storing MeshCore events:

MQTT Subscriber (collector/subscriber.py):
- Subscribes to all event topics from MQTT broker
- Routes events to appropriate handlers
- Manages database persistence

Event Handlers:
- advertisement.py: Handles node advertisements, upserts nodes
- message.py: Handles contact and channel messages
- trace.py: Handles network trace path data
- telemetry.py: Handles sensor telemetry responses
- contacts.py: Handles contacts sync events
- event_log.py: Generic handler for informational events

CLI (collector/cli.py):
- Click command for running the collector
- Environment variable support for all options
- Integrated with main CLI

Tests:
- Subscriber tests
- Handler tests for advertisement, message, telemetry

src/meshcore_hub/__main__.py

@@ -29,55 +29,12 @@ def cli(ctx: click.Context, log_level: str) -> None:
     configure_logging(level=ctx.obj["log_level"])
 
 
-# Import and register interface CLI
+# Import and register component CLIs
 from meshcore_hub.interface.cli import interface
+from meshcore_hub.collector.cli import collector
 
 cli.add_command(interface)
-
-
-@cli.command()
-@click.option(
-    "--mqtt-host",
-    type=str,
-    default="localhost",
-    envvar="MQTT_HOST",
-    help="MQTT broker host",
-)
-@click.option(
-    "--mqtt-port",
-    type=int,
-    default=1883,
-    envvar="MQTT_PORT",
-    help="MQTT broker port",
-)
-@click.option(
-    "--prefix",
-    type=str,
-    default="meshcore",
-    envvar="MQTT_PREFIX",
-    help="MQTT topic prefix",
-)
-@click.option(
-    "--database-url",
-    type=str,
-    default="sqlite:///./meshcore.db",
-    envvar="DATABASE_URL",
-    help="Database connection URL",
-)
-def collector(
-    mqtt_host: str,
-    mqtt_port: int,
-    prefix: str,
-    database_url: str,
-) -> None:
-    """Run the collector component.
-
-    Subscribes to MQTT broker and stores events in database.
-    """
-    click.echo("Starting collector...")
-    click.echo(f"MQTT: {mqtt_host}:{mqtt_port} (prefix: {prefix})")
-    click.echo(f"Database: {database_url}")
-    click.echo("Collector component not yet implemented.")
+cli.add_command(collector)
 
 
 @cli.command()

src/meshcore_hub/collector/cli.py

@@ -0,0 +1,94 @@
+"""CLI for the Collector component."""
+
+import click
+
+from meshcore_hub.common.logging import configure_logging
+
+
+@click.command("collector")
+@click.option(
+    "--mqtt-host",
+    type=str,
+    default="localhost",
+    envvar="MQTT_HOST",
+    help="MQTT broker host",
+)
+@click.option(
+    "--mqtt-port",
+    type=int,
+    default=1883,
+    envvar="MQTT_PORT",
+    help="MQTT broker port",
+)
+@click.option(
+    "--mqtt-username",
+    type=str,
+    default=None,
+    envvar="MQTT_USERNAME",
+    help="MQTT username",
+)
+@click.option(
+    "--mqtt-password",
+    type=str,
+    default=None,
+    envvar="MQTT_PASSWORD",
+    help="MQTT password",
+)
+@click.option(
+    "--prefix",
+    type=str,
+    default="meshcore",
+    envvar="MQTT_PREFIX",
+    help="MQTT topic prefix",
+)
+@click.option(
+    "--database-url",
+    type=str,
+    default="sqlite:///./meshcore.db",
+    envvar="DATABASE_URL",
+    help="Database connection URL",
+)
+@click.option(
+    "--log-level",
+    type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]),
+    default="INFO",
+    envvar="LOG_LEVEL",
+    help="Log level",
+)
+def collector(
+    mqtt_host: str,
+    mqtt_port: int,
+    mqtt_username: str | None,
+    mqtt_password: str | None,
+    prefix: str,
+    database_url: str,
+    log_level: str,
+) -> None:
+    """Run the collector component.
+
+    The collector subscribes to MQTT broker and stores
+    MeshCore events in the database for later retrieval.
+
+    Events stored include:
+    - Node advertisements
+    - Contact and channel messages
+    - Trace path data
+    - Telemetry responses
+    - Informational events (battery, status, etc.)
+    """
+    configure_logging(level=log_level)
+
+    click.echo("Starting MeshCore Collector")
+    click.echo(f"MQTT: {mqtt_host}:{mqtt_port} (prefix: {prefix})")
+    click.echo(f"Database: {database_url}")
+
+    from meshcore_hub.collector.subscriber import run_collector
+
+    run_collector(
+        mqtt_host=mqtt_host,
+        mqtt_port=mqtt_port,
+        mqtt_username=mqtt_username,
+        mqtt_password=mqtt_password,
+        mqtt_prefix=prefix,
+        database_url=database_url,
+    )

src/meshcore_hub/collector/handlers/__init__.py

@@ -1 +1,37 @@
 """Event handlers for processing MQTT messages."""
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from meshcore_hub.collector.subscriber import Subscriber
+
+
+def register_all_handlers(subscriber: "Subscriber") -> None:
+    """Register all event handlers with the subscriber.
+
+    Args:
+        subscriber: Subscriber instance
+    """
+    from meshcore_hub.collector.handlers.advertisement import handle_advertisement
+    from meshcore_hub.collector.handlers.message import (
+        handle_contact_message,
+        handle_channel_message,
+    )
+    from meshcore_hub.collector.handlers.trace import handle_trace_data
+    from meshcore_hub.collector.handlers.telemetry import handle_telemetry
+    from meshcore_hub.collector.handlers.contacts import handle_contacts
+    from meshcore_hub.collector.handlers.event_log import handle_event_log
+
+    # Persisted events with specific handlers
+    subscriber.register_handler("advertisement", handle_advertisement)
+    subscriber.register_handler("contact_msg_recv", handle_contact_message)
+    subscriber.register_handler("channel_msg_recv", handle_channel_message)
+    subscriber.register_handler("trace_data", handle_trace_data)
+    subscriber.register_handler("telemetry_response", handle_telemetry)
+    subscriber.register_handler("contacts", handle_contacts)
+
+    # Informational events (logged only)
+    subscriber.register_handler("send_confirmed", handle_event_log)
+    subscriber.register_handler("status_response", handle_event_log)
+    subscriber.register_handler("battery", handle_event_log)
+    subscriber.register_handler("path_updated", handle_event_log)

src/meshcore_hub/collector/handlers/advertisement.py

@@ -0,0 +1,100 @@
+"""Handler for advertisement events."""
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import Advertisement, Node
+
+logger = logging.getLogger(__name__)
+
+
+def handle_advertisement(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle an advertisement event.
+
+    1. Upserts the node in the nodes table
+    2. Creates an advertisement record
+    3. Updates node last_seen timestamp
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Advertisement payload
+        db: Database manager
+    """
+    adv_public_key = payload.get("public_key")
+    if not adv_public_key:
+        logger.warning("Advertisement missing public_key")
+        return
+
+    name = payload.get("name")
+    adv_type = payload.get("adv_type")
+    flags = payload.get("flags")
+    now = datetime.now(timezone.utc)
+
+    with db.session_scope() as session:
+        # Find or create receiver node
+        receiver_node = None
+        if public_key:
+            receiver_query = select(Node).where(Node.public_key == public_key)
+            receiver_node = session.execute(receiver_query).scalar_one_or_none()
+
+            if not receiver_node:
+                receiver_node = Node(
+                    public_key=public_key,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(receiver_node)
+                session.flush()
+
+        # Find or create advertised node
+        node_query = select(Node).where(Node.public_key == adv_public_key)
+        node = session.execute(node_query).scalar_one_or_none()
+
+        if node:
+            # Update existing node
+            if name:
+                node.name = name
+            if adv_type:
+                node.adv_type = adv_type
+            if flags is not None:
+                node.flags = flags
+            node.last_seen = now
+        else:
+            # Create new node
+            node = Node(
+                public_key=adv_public_key,
+                name=name,
+                adv_type=adv_type,
+                flags=flags,
+                first_seen=now,
+                last_seen=now,
+            )
+            session.add(node)
+            session.flush()
+
+        # Create advertisement record
+        advertisement = Advertisement(
+            receiver_node_id=receiver_node.id if receiver_node else None,
+            node_id=node.id,
+            public_key=adv_public_key,
+            name=name,
+            adv_type=adv_type,
+            flags=flags,
+            received_at=now,
+        )
+        session.add(advertisement)
+
+    logger.info(
+        f"Stored advertisement from {name or adv_public_key[:12]!r} "
+        f"(type={adv_type})"
+    )

src/meshcore_hub/collector/handlers/contacts.py

@@ -0,0 +1,75 @@
+"""Handler for contacts sync events."""
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import Node
+
+logger = logging.getLogger(__name__)
+
+
+def handle_contacts(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle a contacts sync event.
+
+    Upserts all contacts in the contacts list.
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Contacts payload
+        db: Database manager
+    """
+    contacts = payload.get("contacts", [])
+    if not contacts:
+        logger.debug("Empty contacts list received")
+        return
+
+    now = datetime.now(timezone.utc)
+    created_count = 0
+    updated_count = 0
+
+    with db.session_scope() as session:
+        for contact in contacts:
+            contact_key = contact.get("public_key")
+            if not contact_key:
+                continue
+
+            name = contact.get("name")
+            node_type = contact.get("node_type")
+
+            # Find or create node
+            node_query = select(Node).where(Node.public_key == contact_key)
+            node = session.execute(node_query).scalar_one_or_none()
+
+            if node:
+                # Update existing node
+                if name and not node.name:
+                    node.name = name
+                if node_type and not node.adv_type:
+                    node.adv_type = node_type
+                node.last_seen = now
+                updated_count += 1
+            else:
+                # Create new node
+                node = Node(
+                    public_key=contact_key,
+                    name=name,
+                    adv_type=node_type,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(node)
+                created_count += 1
+
+    logger.info(
+        f"Processed contacts sync: {created_count} new, {updated_count} updated"
+    )

src/meshcore_hub/collector/handlers/event_log.py

@@ -0,0 +1,61 @@
+"""Generic event log handler for informational events."""
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import EventLog, Node
+
+logger = logging.getLogger(__name__)
+
+
+def handle_event_log(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle an event by logging it to the events_log table.
+
+    This is used for informational events that don't need
+    specific processing but should be recorded.
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Event payload
+        db: Database manager
+    """
+    now = datetime.now(timezone.utc)
+
+    with db.session_scope() as session:
+        # Find receiver node
+        receiver_node = None
+        if public_key:
+            receiver_query = select(Node).where(Node.public_key == public_key)
+            receiver_node = session.execute(receiver_query).scalar_one_or_none()
+
+            if not receiver_node:
+                receiver_node = Node(
+                    public_key=public_key,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(receiver_node)
+                session.flush()
+            else:
+                receiver_node.last_seen = now
+
+        # Create event log record
+        event_log = EventLog(
+            receiver_node_id=receiver_node.id if receiver_node else None,
+            event_type=event_type,
+            payload=payload,
+            received_at=now,
+        )
+        session.add(event_log)
+
+    logger.debug(f"Logged event: {event_type}")

src/meshcore_hub/collector/handlers/message.py

@@ -0,0 +1,130 @@
+"""Handler for message events."""
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import Message, Node
+
+logger = logging.getLogger(__name__)
+
+
+def handle_contact_message(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle a contact message event.
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Message payload
+        db: Database manager
+    """
+    _handle_message(public_key, "contact", payload, db)
+
+
+def handle_channel_message(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle a channel message event.
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Message payload
+        db: Database manager
+    """
+    _handle_message(public_key, "channel", payload, db)
+
+
+def _handle_message(
+    public_key: str,
+    message_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle a message event (contact or channel).
+
+    Args:
+        public_key: Receiver node's public key
+        message_type: Message type ('contact' or 'channel')
+        payload: Message payload
+        db: Database manager
+    """
+    text = payload.get("text")
+    if not text:
+        logger.warning(f"Message missing text content")
+        return
+
+    now = datetime.now(timezone.utc)
+
+    # Extract fields based on message type
+    pubkey_prefix = payload.get("pubkey_prefix") if message_type == "contact" else None
+    channel_idx = payload.get("channel_idx") if message_type == "channel" else None
+    path_len = payload.get("path_len")
+    txt_type = payload.get("txt_type")
+    signature = payload.get("signature")
+    snr = payload.get("SNR") or payload.get("snr")
+
+    # Parse sender timestamp
+    sender_ts = payload.get("sender_timestamp")
+    sender_timestamp = None
+    if sender_ts:
+        try:
+            sender_timestamp = datetime.fromtimestamp(sender_ts, tz=timezone.utc)
+        except (ValueError, OSError):
+            pass
+
+    with db.session_scope() as session:
+        # Find receiver node
+        receiver_node = None
+        if public_key:
+            receiver_query = select(Node).where(Node.public_key == public_key)
+            receiver_node = session.execute(receiver_query).scalar_one_or_none()
+
+            if not receiver_node:
+                receiver_node = Node(
+                    public_key=public_key,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(receiver_node)
+                session.flush()
+            else:
+                receiver_node.last_seen = now
+
+        # Create message record
+        message = Message(
+            receiver_node_id=receiver_node.id if receiver_node else None,
+            message_type=message_type,
+            pubkey_prefix=pubkey_prefix,
+            channel_idx=channel_idx,
+            text=text,
+            path_len=path_len,
+            txt_type=txt_type,
+            signature=signature,
+            snr=snr,
+            sender_timestamp=sender_timestamp,
+            received_at=now,
+        )
+        session.add(message)
+
+    if message_type == "contact":
+        logger.info(
+            f"Stored contact message from {pubkey_prefix!r}: "
+            f"{text[:30]}{'...' if len(text) > 30 else ''}"
+        )
+    else:
+        logger.info(
+            f"Stored channel {channel_idx} message: "
+            f"{text[:30]}{'...' if len(text) > 30 else ''}"
+        )

src/meshcore_hub/collector/handlers/telemetry.py

@@ -0,0 +1,103 @@
+"""Handler for telemetry events."""
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import Node, Telemetry
+
+logger = logging.getLogger(__name__)
+
+
+def handle_telemetry(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle a telemetry response event.
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Telemetry payload
+        db: Database manager
+    """
+    node_public_key = payload.get("node_public_key")
+    if not node_public_key:
+        logger.warning("Telemetry missing node_public_key")
+        return
+
+    now = datetime.now(timezone.utc)
+
+    lpp_data = payload.get("lpp_data")
+    parsed_data = payload.get("parsed_data")
+
+    # Convert lpp_data to bytes if it's a string or list
+    lpp_bytes = None
+    if lpp_data:
+        if isinstance(lpp_data, bytes):
+            lpp_bytes = lpp_data
+        elif isinstance(lpp_data, list):
+            lpp_bytes = bytes(lpp_data)
+        elif isinstance(lpp_data, str):
+            try:
+                lpp_bytes = bytes.fromhex(lpp_data)
+            except ValueError:
+                lpp_bytes = lpp_data.encode()
+
+    with db.session_scope() as session:
+        # Find receiver node
+        receiver_node = None
+        if public_key:
+            receiver_query = select(Node).where(Node.public_key == public_key)
+            receiver_node = session.execute(receiver_query).scalar_one_or_none()
+
+            if not receiver_node:
+                receiver_node = Node(
+                    public_key=public_key,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(receiver_node)
+                session.flush()
+            else:
+                receiver_node.last_seen = now
+
+        # Find or create reporting node
+        reporting_node = None
+        if node_public_key:
+            node_query = select(Node).where(Node.public_key == node_public_key)
+            reporting_node = session.execute(node_query).scalar_one_or_none()
+
+            if not reporting_node:
+                reporting_node = Node(
+                    public_key=node_public_key,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(reporting_node)
+                session.flush()
+            else:
+                reporting_node.last_seen = now
+
+        # Create telemetry record
+        telemetry = Telemetry(
+            receiver_node_id=receiver_node.id if receiver_node else None,
+            node_id=reporting_node.id if reporting_node else None,
+            node_public_key=node_public_key,
+            lpp_data=lpp_bytes,
+            parsed_data=parsed_data,
+            received_at=now,
+        )
+        session.add(telemetry)
+
+    # Log telemetry values
+    if parsed_data:
+        values = ", ".join(f"{k}={v}" for k, v in parsed_data.items())
+        logger.info(f"Stored telemetry from {node_public_key[:12]!r}: {values}")
+    else:
+        logger.info(f"Stored telemetry from {node_public_key[:12]!r}")

src/meshcore_hub/collector/handlers/trace.py

@@ -0,0 +1,77 @@
+"""Handler for trace data events."""
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import Node, TracePath
+
+logger = logging.getLogger(__name__)
+
+
+def handle_trace_data(
+    public_key: str,
+    event_type: str,
+    payload: dict[str, Any],
+    db: DatabaseManager,
+) -> None:
+    """Handle a trace data event.
+
+    Args:
+        public_key: Receiver node's public key (from MQTT topic)
+        event_type: Event type name
+        payload: Trace data payload
+        db: Database manager
+    """
+    initiator_tag = payload.get("initiator_tag")
+    if initiator_tag is None:
+        logger.warning("Trace data missing initiator_tag")
+        return
+
+    now = datetime.now(timezone.utc)
+
+    path_len = payload.get("path_len")
+    flags = payload.get("flags")
+    auth = payload.get("auth")
+    path_hashes = payload.get("path_hashes")
+    snr_values = payload.get("snr_values")
+    hop_count = payload.get("hop_count")
+
+    with db.session_scope() as session:
+        # Find receiver node
+        receiver_node = None
+        if public_key:
+            receiver_query = select(Node).where(Node.public_key == public_key)
+            receiver_node = session.execute(receiver_query).scalar_one_or_none()
+
+            if not receiver_node:
+                receiver_node = Node(
+                    public_key=public_key,
+                    first_seen=now,
+                    last_seen=now,
+                )
+                session.add(receiver_node)
+                session.flush()
+            else:
+                receiver_node.last_seen = now
+
+        # Create trace path record
+        trace_path = TracePath(
+            receiver_node_id=receiver_node.id if receiver_node else None,
+            initiator_tag=initiator_tag,
+            path_len=path_len,
+            flags=flags,
+            auth=auth,
+            path_hashes=path_hashes,
+            snr_values=snr_values,
+            hop_count=hop_count,
+            received_at=now,
+        )
+        session.add(trace_path)
+
+    logger.info(
+        f"Stored trace data: tag={initiator_tag}, hops={hop_count}"
+    )

src/meshcore_hub/collector/subscriber.py

@@ -0,0 +1,230 @@
+"""MQTT Subscriber for collecting MeshCore events.
+
+The subscriber:
+1. Connects to MQTT broker
+2. Subscribes to all event topics
+3. Routes events to appropriate handlers
+4. Persists data to database
+"""
+
+import logging
+import signal
+import threading
+import time
+from typing import Any, Callable, Optional
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.mqtt import MQTTClient, MQTTConfig
+
+logger = logging.getLogger(__name__)
+
+
+# Handler type: receives (public_key, event_type, payload, db_manager)
+EventHandler = Callable[[str, str, dict[str, Any], DatabaseManager], None]
+
+
+class Subscriber:
+    """MQTT Subscriber for collecting and storing MeshCore events."""
+
+    def __init__(
+        self,
+        mqtt_client: MQTTClient,
+        db_manager: DatabaseManager,
+    ):
+        """Initialize subscriber.
+
+        Args:
+            mqtt_client: MQTT client instance
+            db_manager: Database manager instance
+        """
+        self.mqtt = mqtt_client
+        self.db = db_manager
+        self._running = False
+        self._shutdown_event = threading.Event()
+        self._handlers: dict[str, EventHandler] = {}
+
+    def register_handler(self, event_type: str, handler: EventHandler) -> None:
+        """Register a handler for an event type.
+
+        Args:
+            event_type: Event type name (e.g., 'advertisement')
+            handler: Handler function
+        """
+        self._handlers[event_type] = handler
+        logger.debug(f"Registered handler for {event_type}")
+
+    def _handle_mqtt_message(
+        self,
+        topic: str,
+        pattern: str,
+        payload: dict[str, Any],
+    ) -> None:
+        """Handle incoming MQTT event message.
+
+        Args:
+            topic: MQTT topic
+            pattern: Subscription pattern
+            payload: Message payload
+        """
+        # Parse event from topic
+        parsed = self.mqtt.topic_builder.parse_event_topic(topic)
+        if not parsed:
+            logger.warning(f"Could not parse event topic: {topic}")
+            return
+
+        public_key, event_type = parsed
+        logger.debug(f"Received event: {event_type} from {public_key[:12]}...")
+
+        # Find and call handler
+        handler = self._handlers.get(event_type)
+        if handler:
+            try:
+                handler(public_key, event_type, payload, self.db)
+            except Exception as e:
+                logger.error(f"Error handling {event_type}: {e}")
+        else:
+            # Use generic event log handler if no specific handler
+            from meshcore_hub.collector.handlers.event_log import handle_event_log
+
+            try:
+                handle_event_log(public_key, event_type, payload, self.db)
+            except Exception as e:
+                logger.error(f"Error logging event {event_type}: {e}")
+
+    def start(self) -> None:
+        """Start the subscriber."""
+        logger.info("Starting collector subscriber")
+
+        # Create database tables if needed
+        self.db.create_tables()
+
+        # Connect to MQTT broker
+        try:
+            self.mqtt.connect()
+            self.mqtt.start_background()
+            logger.info("Connected to MQTT broker")
+        except Exception as e:
+            logger.error(f"Failed to connect to MQTT broker: {e}")
+            raise
+
+        # Subscribe to all event topics
+        event_topic = self.mqtt.topic_builder.all_events_topic()
+        self.mqtt.subscribe(event_topic, self._handle_mqtt_message)
+        logger.info(f"Subscribed to event topic: {event_topic}")
+
+        self._running = True
+
+    def run(self) -> None:
+        """Run the subscriber event loop (blocking)."""
+        if not self._running:
+            self.start()
+
+        logger.info("Collector running. Press Ctrl+C to stop.")
+
+        try:
+            while self._running and not self._shutdown_event.is_set():
+                time.sleep(0.1)
+        except KeyboardInterrupt:
+            logger.info("Keyboard interrupt received")
+        finally:
+            self.stop()
+
+    def stop(self) -> None:
+        """Stop the subscriber."""
+        if not self._running:
+            return
+
+        logger.info("Stopping collector subscriber")
+        self._running = False
+        self._shutdown_event.set()
+
+        # Stop MQTT
+        self.mqtt.stop()
+        self.mqtt.disconnect()
+
+        logger.info("Collector subscriber stopped")
+
+
+def create_subscriber(
+    mqtt_host: str = "localhost",
+    mqtt_port: int = 1883,
+    mqtt_username: Optional[str] = None,
+    mqtt_password: Optional[str] = None,
+    mqtt_prefix: str = "meshcore",
+    database_url: str = "sqlite:///./meshcore.db",
+) -> Subscriber:
+    """Create a configured subscriber instance.
+
+    Args:
+        mqtt_host: MQTT broker host
+        mqtt_port: MQTT broker port
+        mqtt_username: MQTT username
+        mqtt_password: MQTT password
+        mqtt_prefix: MQTT topic prefix
+        database_url: Database connection URL
+
+    Returns:
+        Configured Subscriber instance
+    """
+    # Create MQTT client
+    mqtt_config = MQTTConfig(
+        host=mqtt_host,
+        port=mqtt_port,
+        username=mqtt_username,
+        password=mqtt_password,
+        prefix=mqtt_prefix,
+        client_id="meshcore-collector",
+    )
+    mqtt_client = MQTTClient(mqtt_config)
+
+    # Create database manager
+    db_manager = DatabaseManager(database_url)
+
+    # Create subscriber
+    subscriber = Subscriber(mqtt_client, db_manager)
+
+    # Register handlers
+    from meshcore_hub.collector.handlers import register_all_handlers
+
+    register_all_handlers(subscriber)
+
+    return subscriber
+
+
+def run_collector(
+    mqtt_host: str = "localhost",
+    mqtt_port: int = 1883,
+    mqtt_username: Optional[str] = None,
+    mqtt_password: Optional[str] = None,
+    mqtt_prefix: str = "meshcore",
+    database_url: str = "sqlite:///./meshcore.db",
+) -> None:
+    """Run the collector (blocking).
+
+    Args:
+        mqtt_host: MQTT broker host
+        mqtt_port: MQTT broker port
+        mqtt_username: MQTT username
+        mqtt_password: MQTT password
+        mqtt_prefix: MQTT topic prefix
+        database_url: Database connection URL
+    """
+    subscriber = create_subscriber(
+        mqtt_host=mqtt_host,
+        mqtt_port=mqtt_port,
+        mqtt_username=mqtt_username,
+        mqtt_password=mqtt_password,
+        mqtt_prefix=mqtt_prefix,
+        database_url=database_url,
+    )
+
+    # Set up signal handlers
+    def signal_handler(signum: int, frame: Any) -> None:
+        logger.info(f"Received signal {signum}")
+        subscriber.stop()
+
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    # Run
+    subscriber.run()

tests/test_collector/conftest.py

@@ -0,0 +1,25 @@
+"""Fixtures for collector component tests."""
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from meshcore_hub.common.database import DatabaseManager
+from meshcore_hub.common.models import Base
+
+
+@pytest.fixture
+def db_manager():
+    """Create an in-memory database manager for testing."""
+    manager = DatabaseManager("sqlite:///:memory:")
+    manager.create_tables()
+    yield manager
+    manager.dispose()
+
+
+@pytest.fixture
+def db_session(db_manager):
+    """Create a database session for testing."""
+    session = db_manager.get_session()
+    yield session
+    session.close()

tests/test_collector/test_handlers/test_advertisement.py

@@ -0,0 +1,87 @@
+"""Tests for advertisement handler."""
+
+import pytest
+from sqlalchemy import select
+
+from meshcore_hub.common.models import Advertisement, Node
+from meshcore_hub.collector.handlers.advertisement import handle_advertisement
+
+
+class TestHandleAdvertisement:
+    """Tests for handle_advertisement."""
+
+    def test_creates_new_node(self, db_manager, db_session):
+        """Test that new nodes are created."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+            "flags": 218,
+        }
+
+        handle_advertisement("b" * 64, "advertisement", payload, db_manager)
+
+        # Check node was created
+        node = db_session.execute(
+            select(Node).where(Node.public_key == "a" * 64)
+        ).scalar_one_or_none()
+
+        assert node is not None
+        assert node.name == "TestNode"
+        assert node.adv_type == "chat"
+        assert node.flags == 218
+
+    def test_updates_existing_node(self, db_manager, db_session):
+        """Test that existing nodes are updated."""
+        # Create initial node
+        node = Node(public_key="a" * 64, name="OldName", adv_type="repeater")
+        db_session.add(node)
+        db_session.commit()
+
+        # Handle advertisement with new data
+        payload = {
+            "public_key": "a" * 64,
+            "name": "NewName",
+            "adv_type": "chat",
+            "flags": 100,
+        }
+
+        handle_advertisement("b" * 64, "advertisement", payload, db_manager)
+
+        # Refresh node
+        db_session.refresh(node)
+
+        assert node.name == "NewName"
+        assert node.adv_type == "chat"
+        assert node.flags == 100
+
+    def test_creates_advertisement_record(self, db_manager, db_session):
+        """Test that advertisement records are created."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+        }
+
+        handle_advertisement("b" * 64, "advertisement", payload, db_manager)
+
+        # Check advertisement was created
+        ad = db_session.execute(select(Advertisement)).scalar_one_or_none()
+
+        assert ad is not None
+        assert ad.public_key == "a" * 64
+        assert ad.name == "TestNode"
+
+    def test_handles_missing_public_key(self, db_manager, db_session):
+        """Test that missing public_key is handled gracefully."""
+        payload = {
+            "name": "TestNode",
+            "adv_type": "chat",
+        }
+
+        # Should not raise
+        handle_advertisement("b" * 64, "advertisement", payload, db_manager)
+
+        # No advertisement should be created
+        ads = db_session.execute(select(Advertisement)).scalars().all()
+        assert len(ads) == 0

tests/test_collector/test_handlers/test_message.py

@@ -0,0 +1,89 @@
+"""Tests for message handlers."""
+
+import pytest
+from sqlalchemy import select
+
+from meshcore_hub.common.models import Message, Node
+from meshcore_hub.collector.handlers.message import (
+    handle_contact_message,
+    handle_channel_message,
+)
+
+
+class TestHandleContactMessage:
+    """Tests for handle_contact_message."""
+
+    def test_creates_contact_message(self, db_manager, db_session):
+        """Test that contact messages are stored."""
+        payload = {
+            "pubkey_prefix": "01ab2186c4d5",
+            "text": "Hello World!",
+            "path_len": 3,
+            "SNR": 15.5,
+        }
+
+        handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager)
+
+        # Check message was created
+        msg = db_session.execute(select(Message)).scalar_one_or_none()
+
+        assert msg is not None
+        assert msg.message_type == "contact"
+        assert msg.pubkey_prefix == "01ab2186c4d5"
+        assert msg.text == "Hello World!"
+        assert msg.path_len == 3
+        assert msg.snr == 15.5
+
+    def test_handles_missing_text(self, db_manager, db_session):
+        """Test that missing text is handled gracefully."""
+        payload = {
+            "pubkey_prefix": "01ab2186c4d5",
+            "path_len": 3,
+        }
+
+        handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager)
+
+        # No message should be created
+        msgs = db_session.execute(select(Message)).scalars().all()
+        assert len(msgs) == 0
+
+
+class TestHandleChannelMessage:
+    """Tests for handle_channel_message."""
+
+    def test_creates_channel_message(self, db_manager, db_session):
+        """Test that channel messages are stored."""
+        payload = {
+            "channel_idx": 4,
+            "text": "Channel broadcast",
+            "path_len": 10,
+            "SNR": 8.5,
+        }
+
+        handle_channel_message("a" * 64, "channel_msg_recv", payload, db_manager)
+
+        # Check message was created
+        msg = db_session.execute(select(Message)).scalar_one_or_none()
+
+        assert msg is not None
+        assert msg.message_type == "channel"
+        assert msg.channel_idx == 4
+        assert msg.text == "Channel broadcast"
+        assert msg.path_len == 10
+        assert msg.snr == 8.5
+
+    def test_creates_receiver_node_if_needed(self, db_manager, db_session):
+        """Test that receiver node is created if it doesn't exist."""
+        payload = {
+            "channel_idx": 4,
+            "text": "Test message",
+        }
+
+        handle_channel_message("a" * 64, "channel_msg_recv", payload, db_manager)
+
+        # Check receiver node was created
+        node = db_session.execute(
+            select(Node).where(Node.public_key == "a" * 64)
+        ).scalar_one_or_none()
+
+        assert node is not None

tests/test_collector/test_handlers/test_telemetry.py

@@ -0,0 +1,61 @@
+"""Tests for telemetry handler."""
+
+import pytest
+from sqlalchemy import select
+
+from meshcore_hub.common.models import Node, Telemetry
+from meshcore_hub.collector.handlers.telemetry import handle_telemetry
+
+
+class TestHandleTelemetry:
+    """Tests for handle_telemetry."""
+
+    def test_creates_telemetry_record(self, db_manager, db_session):
+        """Test that telemetry records are stored."""
+        payload = {
+            "node_public_key": "b" * 64,
+            "parsed_data": {
+                "temperature": 22.5,
+                "humidity": 65,
+                "battery": 3.8,
+            },
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+
+        # Check telemetry was created
+        telemetry = db_session.execute(select(Telemetry)).scalar_one_or_none()
+
+        assert telemetry is not None
+        assert telemetry.node_public_key == "b" * 64
+        assert telemetry.parsed_data["temperature"] == 22.5
+        assert telemetry.parsed_data["humidity"] == 65
+        assert telemetry.parsed_data["battery"] == 3.8
+
+    def test_creates_reporting_node(self, db_manager, db_session):
+        """Test that reporting node is created if needed."""
+        payload = {
+            "node_public_key": "b" * 64,
+            "parsed_data": {"temperature": 20.0},
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+
+        # Check node was created
+        node = db_session.execute(
+            select(Node).where(Node.public_key == "b" * 64)
+        ).scalar_one_or_none()
+
+        assert node is not None
+
+    def test_handles_missing_node_public_key(self, db_manager, db_session):
+        """Test that missing node_public_key is handled gracefully."""
+        payload = {
+            "parsed_data": {"temperature": 20.0},
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+
+        # No telemetry should be created
+        records = db_session.execute(select(Telemetry)).scalars().all()
+        assert len(records) == 0

tests/test_collector/test_subscriber.py

@@ -0,0 +1,78 @@
+"""Tests for the collector subscriber."""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+from meshcore_hub.collector.subscriber import Subscriber, create_subscriber
+
+
+class TestSubscriber:
+    """Tests for Subscriber class."""
+
+    @pytest.fixture
+    def mock_mqtt_client(self):
+        """Create a mock MQTT client."""
+        client = MagicMock()
+        client.topic_builder = MagicMock()
+        client.topic_builder.all_events_topic.return_value = "meshcore/+/event/#"
+        client.topic_builder.parse_event_topic.return_value = ("a" * 64, "advertisement")
+        return client
+
+    @pytest.fixture
+    def subscriber(self, mock_mqtt_client, db_manager):
+        """Create a subscriber instance."""
+        return Subscriber(mock_mqtt_client, db_manager)
+
+    def test_register_handler(self, subscriber):
+        """Test handler registration."""
+        handler = MagicMock()
+
+        subscriber.register_handler("advertisement", handler)
+
+        assert "advertisement" in subscriber._handlers
+
+    def test_start_connects_mqtt(self, subscriber, mock_mqtt_client):
+        """Test that start connects to MQTT."""
+        subscriber.start()
+
+        mock_mqtt_client.connect.assert_called_once()
+        mock_mqtt_client.start_background.assert_called_once()
+        mock_mqtt_client.subscribe.assert_called_once()
+
+    def test_stop_disconnects_mqtt(self, subscriber, mock_mqtt_client):
+        """Test that stop disconnects MQTT."""
+        subscriber.start()
+        subscriber.stop()
+
+        mock_mqtt_client.stop.assert_called_once()
+        mock_mqtt_client.disconnect.assert_called_once()
+
+    def test_handle_mqtt_message_calls_handler(self, subscriber, mock_mqtt_client, db_manager):
+        """Test that MQTT messages are routed to handlers."""
+        handler = MagicMock()
+        subscriber.register_handler("advertisement", handler)
+        subscriber.start()
+
+        subscriber._handle_mqtt_message(
+            topic="meshcore/abc/event/advertisement",
+            pattern="meshcore/+/event/#",
+            payload={"public_key": "b" * 64, "name": "Test"},
+        )
+
+        handler.assert_called_once()
+
+
+class TestCreateSubscriber:
+    """Tests for create_subscriber factory function."""
+
+    def test_creates_subscriber(self):
+        """Test creating a subscriber."""
+        with patch("meshcore_hub.collector.subscriber.MQTTClient") as MockMQTT:
+            subscriber = create_subscriber(
+                mqtt_host="localhost",
+                mqtt_port=1883,
+                database_url="sqlite:///:memory:",
+            )
+
+            assert subscriber is not None
+            MockMQTT.assert_called_once()