esp32-cluster/coordinator/dashboard.py

#!/usr/bin/env python3
"""
WiFi Recon Dashboard
Reads from events.db (written by udp_ingest.py), serves a live web UI.
"""

import asyncio
import sqlite3
import datetime
import os
import json
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import HTMLResponse, StreamingResponse, FileResponse

DB_PATH      = os.path.join(os.path.dirname(__file__), "events.db")
HTML_PATH    = os.path.join(os.path.dirname(__file__), "templates", "dashboard.html")
CSS_PATH     = os.path.join(os.path.dirname(__file__), "static", "dashboard.css")
JS_PATH      = os.path.join(os.path.dirname(__file__), "static", "dashboard.js")
OUI_PATH     = os.path.join(os.path.dirname(__file__), "oui.txt")
ONLINE_SECS  = 30   # node considered online if heartbeat within this many seconds
EVENT_CAP    = 100  # max events returned / shown in dashboard
HOT_DAYS     = 7    # dashboard queries cover this many days; Search is always all-time

# ─── Deauth reason codes ─────────────────────────────────────────────────────

# classification: 'normal', 'suspicious', 'attack'
DEAUTH_REASONS: dict[int, tuple[str, str]] = {
    0:   ("Reserved",                                    "normal"),
    1:   ("Unspecified",                                 "normal"),
    2:   ("Previous auth no longer valid",               "attack"),
    3:   ("Station leaving BSS",                         "normal"),
    4:   ("Inactivity / idle timeout",                   "normal"),
    5:   ("Too many associated stations",                "normal"),
    6:   ("Class 2 frame from non-authenticated station","attack"),
    7:   ("Class 3 frame from non-associated station",   "attack"),
    8:   ("Station leaving BSS (disassoc)",              "normal"),
    9:   ("Station not authenticated",                   "suspicious"),
    10:  ("Power capability unacceptable",               "normal"),
    11:  ("Supported channels unacceptable",             "normal"),
    13:  ("Invalid information element",                 "suspicious"),
    14:  ("MIC failure",                                 "suspicious"),
    15:  ("4-way handshake timeout",                     "normal"),
    16:  ("Group key handshake timeout",                 "normal"),
    17:  ("IE in 4-way handshake differs",               "suspicious"),
    18:  ("Invalid group cipher",                        "suspicious"),
    19:  ("Invalid pairwise cipher",                     "suspicious"),
    20:  ("Invalid AKMP",                                "suspicious"),
    21:  ("Unsupported RSNE version",                    "suspicious"),
    22:  ("Invalid RSNE capabilities",                   "suspicious"),
    23:  ("802.1X authentication failed",                "suspicious"),
    24:  ("Cipher suite rejected by policy",             "suspicious"),
    34:  ("Low ACK",                                     "normal"),
    39:  ("BSS transition management request",           "normal"),
    45:  ("Peer stakey negotiation failed",              "suspicious"),
    47:  ("Authorized access limit reached",             "normal"),
    99:  ("Peer abandoned session",                      "normal"),
    103: ("Disassociated — excessive frames",            "suspicious"),
    147: ("Vendor-specific",                             "normal"),
}

def deauth_reason_info(code: int) -> tuple[str, str]:
    return DEAUTH_REASONS.get(code, ("Unknown / vendor-specific", "normal"))

# ─── OUI lookup ──────────────────────────────────────────────────────────────

def _load_oui(path: str) -> dict[str, str]:
    db: dict[str, str] = {}
    if not os.path.exists(path):
        return db
    with open(path, encoding="utf-8", errors="ignore") as f:
        for line in f:
            if "(hex)" not in line:
                continue
            parts = line.split("(hex)")
            prefix = parts[0].strip().replace("-", ":").upper()   # "AA-BB-CC" → "AA:BB:CC"
            vendor = parts[1].strip()
            if prefix and vendor:
                db[prefix] = vendor
    return db

_OUI_DB: dict[str, str] = _load_oui(OUI_PATH)


def oui_lookup(mac: str) -> tuple[str, bool]:
    """
    Returns (vendor_name, is_randomized).
    Randomized MACs have the locally-administered bit set (bit 1 of first byte).
    OUI lookup is meaningless for randomized MACs.
    """
    if not mac or len(mac) < 8:
        return ("Unknown", False)
    try:
        first_byte = int(mac.replace(":", "").replace("-", "")[:2], 16)
        randomized = bool(first_byte & 0x02)
    except ValueError:
        return ("Unknown", False)
    if randomized:
        return ("Randomized MAC", True)
    prefix = mac.upper()[:8]
    return (_OUI_DB.get(prefix, "Unknown"), False)

# ─── DB helpers ──────────────────────────────────────────────────────────────

def get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA wal_autocheckpoint=500")  # auto-checkpoint every 500 pages (~2MB)
    return conn


def query(sql: str, params: tuple = ()) -> list[dict]:
    with get_conn() as conn:
        rows = conn.execute(sql, params).fetchall()
    return [dict(r) for r in rows]


def _hot_cutoff() -> str:
    """ISO timestamp HOT_DAYS ago — used to scope dashboard queries."""
    return (datetime.datetime.utcnow() - datetime.timedelta(days=HOT_DAYS)).isoformat(timespec="seconds")

def _node_status(last_seen: str | None, threshold: int = ONLINE_SECS) -> str:
    if not last_seen:
        return "offline"
    try:
        dt = datetime.datetime.fromisoformat(last_seen)
        delta = (datetime.datetime.utcnow() - dt).total_seconds()
        return "online" if delta <= threshold else "offline"
    except ValueError:
        return "offline"

# ─── API data builders ────────────────────────────────────────────────────────

# ─── BLE classification ──────────────────────────────────────────────────────

BLE_COMPANIES: dict[int, str] = {
    0x004C: "Apple",
    0x0075: "Samsung",
    0x0006: "Microsoft",
    0x00E0: "Google",
    0x0059: "Nordic Semiconductor",
    0x0138: "Garmin",
    0x0499: "Ruuvi Innovations",
    0x0157: "Polar Electro",
    0x00BD: "Fitbit",
    0x0171: "Amazon",
    0x004F: "MediaTek",
    0x001D: "Qualcomm",
}

# Apple manufacturer data: after the 2-byte company ID, byte 0 of mfr_data is the type.
APPLE_TYPES: dict[str, str] = {
    "02": "iBeacon",
    "05": "AirDrop",
    "07": "HomeKit",
    "09": "AirPods",
    "0a": "AirPods Pro",
    "0b": "AirPods 2nd gen",
    "0c": "AirPods case / Watch",
    "0f": "AirPods 3rd gen",
    "12": "FindMy / AirTag",
    "13": "AirPods 4th gen",
    "15": "Proximity Pair",
    "1e": "AirPods Pro 2nd gen",
}


def ble_device_type(mfr_id: int | None, mfr_data: str | None) -> str:
    if mfr_id is None:
        return "Unknown"
    if mfr_id == 0x004C and mfr_data and len(mfr_data) >= 2:
        return APPLE_TYPES.get(mfr_data[:2].lower(), "Apple device")
    vendor = BLE_COMPANIES.get(mfr_id)
    return f"{vendor} device" if vendor else f"Unknown (0x{mfr_id:04X})"


_nodes_cache: tuple[float, list] | None = None
_networks_cache: tuple[float, list] | None = None
_clients_cache: tuple[float, list] | None = None
_cross_node_cache: tuple[float, dict] | None = None
_presence_cache: tuple[float, dict] | None = None
_alerts_cache: tuple[float, dict] | None = None
_sessions_cache: tuple[float, list] | None = None
_ble_cache: tuple[float, dict] | None = None
_NODES_CACHE_TTL      = 10   # seconds
_NETWORKS_CACHE_TTL   = 30   # seconds
_CLIENTS_CACHE_TTL    = 30   # seconds
_CROSS_NODE_CACHE_TTL = 60   # seconds
_PRESENCE_CACHE_TTL   = 60   # seconds
_ALERTS_CACHE_TTL     = 60   # seconds
_SESSIONS_CACHE_TTL   = 60   # seconds
_BLE_CACHE_TTL        = 30   # seconds

def build_nodes() -> list[dict]:
    global _nodes_cache
    now = datetime.datetime.utcnow().timestamp()
    if _nodes_cache and (now - _nodes_cache[0]) < _NODES_CACHE_TTL:
        return _nodes_cache[1]

    # WiFi nodes: beacon-derived stats
    beacon_rows = query("""
        SELECT
            node_id,
            MAX(received_at)  AS last_beacon,
            COUNT(*)          AS total_events,
            MIN(received_at)  AS first_seen
        FROM beacon_events
        GROUP BY node_id
    """)

    # Latest heartbeat per node — covers ALL node types (WiFi and BLE)
    hb_rows = query("""
        SELECT node_id, received_at AS last_heartbeat,
               uptime_ms, free_heap, wifi_rssi
        FROM heartbeat_events
        WHERE id IN (SELECT MAX(id) FROM heartbeat_events GROUP BY node_id)
    """)
    hb_map = {r["node_id"]: r for r in hb_rows}

    node_map: dict = {r["node_id"]: r for r in beacon_rows}

    # BLE-only nodes have no beacon events — add them from heartbeat data
    for node_id, hb in hb_map.items():
        if node_id not in node_map:
            node_map[node_id] = {
                "node_id":      node_id,
                "last_beacon":  None,
                "total_events": 0,
                "first_seen":   hb.get("last_heartbeat"),
            }

    rows = sorted(node_map.values(), key=lambda r: r["node_id"])

    for r in rows:
        hb = hb_map.get(r["node_id"], {})
        r["last_heartbeat"] = hb.get("last_heartbeat")
        r["uptime_ms"]      = hb.get("uptime_ms")
        r["free_heap"]      = hb.get("free_heap")
        r["wifi_rssi"]      = hb.get("wifi_rssi")
        r["last_seen"]      = r["last_heartbeat"] or r["last_beacon"]
        if r["last_heartbeat"]:
            r["status"] = _node_status(r["last_heartbeat"], threshold=30)
        else:
            r["status"] = _node_status(r["last_beacon"], threshold=60)

    _nodes_cache = (now, rows)
    return rows


def build_events(limit: int = EVENT_CAP, node_id: str | None = None) -> list[dict]:
    if node_id:
        return query("""
            SELECT * FROM beacon_events
            WHERE node_id = ?
            ORDER BY id DESC LIMIT ?
        """, (node_id, limit))
    return query("""
        SELECT * FROM beacon_events
        ORDER BY id DESC LIMIT ?
    """, (limit,))


def build_networks() -> list[dict]:
    global _networks_cache
    now = datetime.datetime.utcnow().timestamp()
    if _networks_cache and (now - _networks_cache[0]) < _NETWORKS_CACHE_TTL:
        return _networks_cache[1]
    rows = query("""
        SELECT
            ssid,
            bssid,
            MAX(rssi)                AS best_rssi,
            ROUND(AVG(rssi), 1)      AS avg_rssi,
            channel,
            encryption,
            COUNT(*)                 AS times_seen,
            COUNT(DISTINCT node_id)  AS node_count,
            MAX(received_at)         AS last_seen
        FROM beacon_events
        WHERE bssid IS NOT NULL AND received_at > ?
        GROUP BY bssid
        ORDER BY times_seen DESC
        LIMIT 300
    """, (_hot_cutoff(),))
    _networks_cache = (now, rows)
    return rows


def build_clients() -> list[dict]:
    global _clients_cache
    now = datetime.datetime.utcnow().timestamp()
    if _clients_cache and (now - _clients_cache[0]) < _CLIENTS_CACHE_TTL:
        return _clients_cache[1]
    """
    Per unique src_mac: aggregated stats and the SSIDs they were probing for.
    """
    cutoff = _hot_cutoff()
    rows = query("""
        SELECT
            src_mac,
            COUNT(*)                        AS times_seen,
            COUNT(DISTINCT node_id)         AS node_count,
            MAX(rssi)                        AS best_rssi,
            ROUND(AVG(rssi), 1)             AS avg_rssi,
            COUNT(DISTINCT NULLIF(ssid,'')) AS unique_ssids,
            MIN(received_at)                AS first_seen,
            MAX(received_at)                AS last_seen
        FROM probe_events
        WHERE src_mac IS NOT NULL AND received_at > ?
        GROUP BY src_mac
        ORDER BY times_seen DESC
        LIMIT 300
    """, (cutoff,))
    # Fetch SSIDs only for the MACs we actually returned
    if not rows:
        return []
    mac_list     = [r["src_mac"] for r in rows]
    placeholders = ",".join("?" * len(mac_list))
    ssid_rows = query(f"""
        SELECT DISTINCT src_mac, ssid FROM probe_events
        WHERE src_mac IN ({placeholders}) AND ssid IS NOT NULL AND ssid != ''
          AND received_at > ?
        ORDER BY src_mac, ssid
    """, tuple(mac_list) + (cutoff,))
    ssids_by_mac: dict = {}
    for s in ssid_rows:
        ssids_by_mac.setdefault(s["src_mac"], []).append(s["ssid"])

    for r in rows:
        r["probed_ssids"] = ssids_by_mac.get(r["src_mac"], [])
        vendor, randomized = oui_lookup(r["src_mac"])
        r["vendor"]     = vendor
        r["randomized"] = randomized
    _clients_cache = (now, rows)
    return rows


def build_cross_node() -> dict:
    """
    For every BSSID, return per-node RSSI stats side by side.
    Result: { node_ids: [...], networks: [{bssid, ssid, channel, encryption,
              node_count, nodes: {node_id: {best_rssi, avg_rssi, times_seen}}}] }
    """
    global _cross_node_cache
    now = datetime.datetime.utcnow().timestamp()
    if _cross_node_cache and (now - _cross_node_cache[0]) < _CROSS_NODE_CACHE_TTL:
        return _cross_node_cache[1]

    node_rows = query("SELECT DISTINCT node_id FROM beacon_events ORDER BY node_id")
    node_ids  = [r["node_id"] for r in node_rows]

    cutoff = _hot_cutoff()
    rows = query("""
        SELECT
            b.bssid,
            MAX(b.ssid)            AS ssid,
            MAX(b.channel)         AS channel,
            MAX(b.encryption)      AS encryption,
            b.node_id,
            MAX(b.rssi)            AS best_rssi,
            ROUND(AVG(b.rssi), 1)  AS avg_rssi,
            COUNT(*)               AS times_seen
        FROM beacon_events b
        WHERE b.received_at > ? AND b.bssid IN (
            SELECT bssid FROM beacon_events
            WHERE bssid IS NOT NULL AND received_at > ?
            GROUP BY bssid
            ORDER BY COUNT(DISTINCT node_id) DESC, COUNT(*) DESC
            LIMIT 300
        )
        GROUP BY b.bssid, b.node_id
        ORDER BY b.bssid, b.node_id
    """, (cutoff, cutoff))

    networks: dict = {}
    for r in rows:
        bssid = r["bssid"]
        if bssid not in networks:
            networks[bssid] = {
                "bssid":      bssid,
                "ssid":       r["ssid"],
                "channel":    r["channel"],
                "encryption": r["encryption"],
                "nodes":      {},
            }
        networks[bssid]["nodes"][r["node_id"]] = {
            "best_rssi":  r["best_rssi"],
            "avg_rssi":   r["avg_rssi"],
            "times_seen": r["times_seen"],
        }

    result = sorted(
        networks.values(),
        key=lambda x: len(x["nodes"]),
        reverse=True,
    )
    for net in result:
        net["node_count"] = len(net["nodes"])

    result_dict = {"node_ids": node_ids, "networks": result}
    _cross_node_cache = (now, result_dict)
    return result_dict


def build_presence() -> dict:
    global _presence_cache
    _now_ts = datetime.datetime.utcnow().timestamp()
    if _presence_cache and (_now_ts - _presence_cache[0]) < _PRESENCE_CACHE_TTL:
        return _presence_cache[1]

    now        = datetime.datetime.utcnow()
    cutoff_1h  = (now - datetime.timedelta(hours=1)).isoformat(timespec="seconds")
    cutoff_24h = (now - datetime.timedelta(hours=24)).isoformat(timespec="seconds")

    # ── Present now: real MACs seen 2+ times in the last hour ────────────────
    present_rows = query("""
        SELECT src_mac,
               COUNT(*)              AS times_seen,
               MAX(rssi)             AS best_rssi,
               ROUND(AVG(rssi), 1)   AS avg_rssi,
               MAX(received_at)      AS last_seen,
               COUNT(DISTINCT node_id) AS node_count
        FROM probe_events
        WHERE src_mac IS NOT NULL AND received_at > ?
        GROUP BY src_mac
        HAVING COUNT(*) > 1
        ORDER BY last_seen DESC
    """, (cutoff_1h,))
    present = []
    for r in present_rows:
        vendor, randomized = oui_lookup(r["src_mac"])
        if not randomized:
            r["vendor"] = vendor
            present.append(r)

    # ── New arrivals: first seen in last 24h ─────────────────────────────────
    new_mac_rows = query("""
        SELECT src_mac,
               MIN(received_at)      AS first_seen,
               MAX(rssi)             AS best_rssi,
               COUNT(*)              AS times_seen,
               COUNT(DISTINCT node_id) AS node_count
        FROM probe_events
        WHERE src_mac IS NOT NULL
        GROUP BY src_mac
        HAVING MIN(received_at) > ?
        ORDER BY first_seen DESC
        LIMIT 100
    """, (cutoff_24h,))
    new_macs = []
    for r in new_mac_rows:
        vendor, randomized = oui_lookup(r["src_mac"])
        r["vendor"]     = vendor
        r["randomized"] = randomized
        new_macs.append(r)

    cutoff_hot = _hot_cutoff()
    new_networks = query("""
        SELECT bssid,
               MAX(ssid)             AS ssid,
               MIN(received_at)      AS first_seen,
               MAX(rssi)             AS best_rssi,
               MAX(channel)          AS channel,
               MAX(encryption)       AS encryption,
               COUNT(DISTINCT node_id) AS node_count
        FROM beacon_events
        WHERE bssid IS NOT NULL AND received_at > ?
        GROUP BY bssid
        HAVING MIN(received_at) > ?
        ORDER BY first_seen DESC
        LIMIT 100
    """, (cutoff_hot, cutoff_24h,))

    # ── Regulars: seen on 2+ distinct calendar days ───────────────────────────
    reg_mac_rows = query("""
        SELECT src_mac,
               COUNT(DISTINCT date(received_at)) AS days_seen,
               MIN(received_at)                  AS first_seen,
               MAX(received_at)                  AS last_seen,
               MAX(rssi)                         AS best_rssi,
               COUNT(*)                          AS times_seen,
               COUNT(DISTINCT node_id)           AS node_count
        FROM probe_events
        WHERE src_mac IS NOT NULL
        GROUP BY src_mac
        HAVING COUNT(DISTINCT date(received_at)) >= 2
        ORDER BY days_seen DESC, times_seen DESC
        LIMIT 50
    """)
    reg_macs = []
    for r in reg_mac_rows:
        vendor, randomized = oui_lookup(r["src_mac"])
        r["vendor"]     = vendor
        r["randomized"] = randomized
        reg_macs.append(r)

    reg_networks = query("""
        SELECT bssid,
               MAX(ssid)                         AS ssid,
               COUNT(DISTINCT date(received_at)) AS days_seen,
               MIN(received_at)                  AS first_seen,
               MAX(received_at)                  AS last_seen,
               MAX(rssi)                         AS best_rssi,
               MAX(channel)                      AS channel,
               MAX(encryption)                   AS encryption,
               COUNT(DISTINCT node_id)           AS node_count
        FROM beacon_events
        WHERE bssid IS NOT NULL
        GROUP BY bssid
        HAVING COUNT(DISTINCT date(received_at)) >= 2
        ORDER BY days_seen DESC, best_rssi DESC
        LIMIT 50
    """)

    result = {
        "present":      present,
        "new_macs":     new_macs,
        "new_networks": new_networks,
        "reg_macs":     reg_macs,
        "reg_networks": reg_networks,
    }
    _presence_cache = (_now_ts, result)
    return result


SESSION_GAP_SECS = 120  # gap > 2 min between events from same src = new session


def _finalize_session(active: dict) -> dict:
    reasons   = active["reasons"]
    dominant  = max(set(reasons), key=reasons.count) if reasons else 0
    _, classification = deauth_reason_info(dominant)

    rssi_vals = [r for r in active["rssi_vals"] if r is not None]
    start_dt  = datetime.datetime.fromisoformat(active["start"])
    end_dt    = datetime.datetime.fromisoformat(active["end"])
    duration  = int((end_dt - start_dt).total_seconds())
    frames    = len(active["frames"])

    if classification == "attack" and frames >= 5:
        severity = "attack"
    elif frames >= 3 or classification in ("attack", "suspicious"):
        severity = "suspicious"
    else:
        severity = "normal"

    desc, _ = deauth_reason_info(dominant)
    src_vendor, src_rand = oui_lookup(active["src"])
    return {
        "src":                  active["src"],
        "src_vendor":           src_vendor,
        "src_randomized":       src_rand,
        "start":                active["start"],
        "end":                  active["end"],
        "duration_secs":        duration,
        "frame_count":          frames,
        "unique_bssids":        list(active["bssids"]),
        "unique_targets":       list(active["targets"]),
        "node_count":           len(active["nodes"]),
        "nodes":                list(active["nodes"]),
        "peak_rssi":            max(rssi_vals) if rssi_vals else None,
        "avg_rssi":             round(sum(rssi_vals) / len(rssi_vals), 1) if rssi_vals else None,
        "dominant_reason":      dominant,
        "dominant_reason_desc": desc,
        "dominant_reason_class": classification,
        "severity":             severity,
        "events":               active["frames"][:50],
    }


def build_sessions() -> list[dict]:
    global _sessions_cache
    now = datetime.datetime.utcnow().timestamp()
    if _sessions_cache and (now - _sessions_cache[0]) < _SESSIONS_CACHE_TTL:
        return _sessions_cache[1]

    cutoff = _hot_cutoff()

    rows = query("""
        SELECT src, dst, bssid, reason, rssi, received_at, node_id, subtype
        FROM deauth_events
        WHERE received_at > ?
        ORDER BY src, received_at
    """, (cutoff,))

    if not rows:
        return []

    sessions: list[dict] = []
    active: dict | None  = None

    for row in rows:
        src = row["src"]
        ts  = row["received_at"]

        new_session = False
        if active is None or active["src"] != src:
            new_session = True
        else:
            last_dt = datetime.datetime.fromisoformat(active["end"])
            curr_dt = datetime.datetime.fromisoformat(ts)
            if (curr_dt - last_dt).total_seconds() > SESSION_GAP_SECS:
                new_session = True

        if new_session:
            if active is not None:
                sessions.append(_finalize_session(active))
            active = {
                "src":      src,
                "start":    ts,
                "end":      ts,
                "bssids":   set(),
                "targets":  set(),
                "nodes":    set(),
                "reasons":  [],
                "rssi_vals": [],
                "frames":   [],
            }

        active["end"] = ts
        active["bssids"].add(row["bssid"])
        active["targets"].add(row["dst"])
        active["nodes"].add(row["node_id"])
        active["reasons"].append(row["reason"])
        active["rssi_vals"].append(row["rssi"])
        active["frames"].append(row)

    if active is not None:
        sessions.append(_finalize_session(active))

    sessions.sort(key=lambda s: s["start"], reverse=True)
    result = sessions[:100]
    _sessions_cache = (now, result)
    return result


def build_alerts() -> dict:
    global _alerts_cache
    _now_ts = datetime.datetime.utcnow().timestamp()
    if _alerts_cache and (_now_ts - _alerts_cache[0]) < _ALERTS_CACHE_TTL:
        return _alerts_cache[1]

    now         = datetime.datetime.utcnow()
    cutoff_5m   = (now - datetime.timedelta(minutes=5)).isoformat(timespec="seconds")
    cutoff_1h   = (now - datetime.timedelta(hours=1)).isoformat(timespec="seconds")
    cutoff_7d   = (now - datetime.timedelta(days=7)).isoformat(timespec="seconds")

    # Recent raw deauth/disassoc events
    recent = query("""
        SELECT * FROM deauth_events
        ORDER BY id DESC LIMIT 100
    """)

    # Burst detection: >= 10 frames for the same BSSID within the last 5 minutes.
    # node_count > 1 means multiple sensors confirmed the burst — much higher confidence.
    bursts = query("""
        SELECT
            bssid,
            subtype,
            COUNT(*)                AS count,
            COUNT(DISTINCT node_id) AS node_count,
            COUNT(DISTINCT src)     AS unique_srcs,
            MIN(received_at)        AS first_seen,
            MAX(received_at)        AS last_seen
        FROM deauth_events
        WHERE received_at > ?
        GROUP BY bssid, subtype
        HAVING COUNT(*) >= 10
        ORDER BY count DESC
    """, (cutoff_5m,))

    # Correlate burst BSSIDs with known network SSIDs
    if bursts:
        bssid_list = [b["bssid"] for b in bursts]
        placeholders = ",".join("?" * len(bssid_list))
        ssid_rows = query(f"""
            SELECT bssid, MAX(ssid) AS ssid, MAX(encryption) AS encryption
            FROM beacon_events WHERE bssid IN ({placeholders})
            GROUP BY bssid
        """, tuple(bssid_list))
        ssid_map = {r["bssid"]: r for r in ssid_rows}
        for b in bursts:
            net = ssid_map.get(b["bssid"], {})
            b["ssid"]       = net.get("ssid")
            b["encryption"] = net.get("encryption")

    # Summary counts for the last hour
    summary = query("""
        SELECT
            COUNT(*)                AS total,
            COUNT(DISTINCT bssid)   AS unique_bssids,
            COUNT(DISTINCT src)     AS unique_srcs,
            SUM(CASE WHEN subtype='deauth' THEN 1 ELSE 0 END)   AS deauth_count,
            SUM(CASE WHEN subtype='disassoc' THEN 1 ELSE 0 END) AS disassoc_count
        FROM deauth_events
        WHERE received_at > ?
    """, (cutoff_1h,))

    # Most targeted BSSIDs — last 7 days
    top_targets = query("""
        SELECT
            bssid,
            COUNT(*)                AS total_frames,
            COUNT(DISTINCT src)     AS unique_srcs,
            COUNT(DISTINCT node_id) AS node_count,
            MIN(received_at)        AS first_seen,
            MAX(received_at)        AS last_seen
        FROM deauth_events
        WHERE received_at > ?
        GROUP BY bssid
        ORDER BY total_frames DESC
        LIMIT 10
    """, (cutoff_7d,))
    if top_targets:
        bssid_list   = [t["bssid"] for t in top_targets]
        placeholders = ",".join("?" * len(bssid_list))
        ssid_rows    = query(f"""
            SELECT bssid, MAX(ssid) AS ssid
            FROM beacon_events WHERE bssid IN ({placeholders})
            GROUP BY bssid
        """, tuple(bssid_list))
        ssid_map = {r["bssid"]: r["ssid"] for r in ssid_rows}
        # Dominant reason per impersonated BSSID
        reason_rows_t = query(f"""
            SELECT bssid, reason, COUNT(*) AS cnt
            FROM deauth_events
            WHERE received_at > ? AND bssid IN ({placeholders})
            GROUP BY bssid, reason
            ORDER BY bssid, cnt DESC
        """, (cutoff_7d,) + tuple(bssid_list))
        dom_reason_by_bssid: dict = {}
        for r in reason_rows_t:
            if r["bssid"] not in dom_reason_by_bssid:
                dom_reason_by_bssid[r["bssid"]] = r["reason"]
        for t in top_targets:
            t["ssid"] = ssid_map.get(t["bssid"])
            code = dom_reason_by_bssid.get(t["bssid"])
            if code is not None:
                desc, cls = deauth_reason_info(code)
                t["dominant_reason"]      = code
                t["dominant_reason_desc"] = desc
                t["dominant_reason_class"] = cls
            else:
                t["dominant_reason"]      = None
                t["dominant_reason_desc"] = None
                t["dominant_reason_class"] = None

    # Most targeted devices — last 7 days (grouped by dst, the actual victim)
    top_targeted_devices = query("""
        SELECT
            dst,
            COUNT(*)                    AS total_frames,
            COUNT(DISTINCT bssid)       AS networks_used,
            COUNT(DISTINCT node_id)     AS node_count,
            MIN(received_at)            AS first_seen,
            MAX(received_at)            AS last_seen
        FROM deauth_events
        WHERE received_at > ?
        GROUP BY dst
        ORDER BY total_frames DESC
        LIMIT 10
    """, (cutoff_7d,))
    if top_targeted_devices:
        dst_list     = [d["dst"] for d in top_targeted_devices]
        placeholders = ",".join("?" * len(dst_list))
        # Dominant reason per targeted dst MAC
        reason_rows_d = query(f"""
            SELECT dst, reason, COUNT(*) AS cnt
            FROM deauth_events
            WHERE received_at > ? AND dst IN ({placeholders})
            GROUP BY dst, reason
            ORDER BY dst, cnt DESC
        """, (cutoff_7d,) + tuple(dst_list))
        dom_reason_by_dst: dict = {}
        for r in reason_rows_d:
            if r["dst"] not in dom_reason_by_dst:
                dom_reason_by_dst[r["dst"]] = r["reason"]
    else:
        dom_reason_by_dst = {}

    for d in top_targeted_devices:
        vendor, randomized = oui_lookup(d["dst"])
        d["vendor"]     = vendor
        d["randomized"] = randomized
        code = dom_reason_by_dst.get(d["dst"])
        if code is not None:
            desc, cls = deauth_reason_info(code)
            d["dominant_reason"]       = code
            d["dominant_reason_desc"]  = desc
            d["dominant_reason_class"] = cls
        else:
            d["dominant_reason"]       = None
            d["dominant_reason_desc"]  = None
            d["dominant_reason_class"] = None

    # Reason code breakdown — last 7 days
    reason_rows = query("""
        SELECT reason,
               COUNT(*)                AS total_frames,
               COUNT(DISTINCT bssid)   AS unique_bssids,
               COUNT(DISTINCT dst)     AS unique_targets,
               MIN(received_at)        AS first_seen,
               MAX(received_at)        AS last_seen
        FROM deauth_events
        WHERE received_at > ?
        GROUP BY reason
        ORDER BY total_frames DESC
    """, (cutoff_7d,))
    for r in reason_rows:
        desc, classification = deauth_reason_info(r["reason"])
        r["description"]    = desc
        r["classification"] = classification

    # Activity heatmap — frame count per day × hour, last 7 days
    heatmap_rows = query("""
        SELECT date(received_at)                            AS day,
               CAST(strftime('%H', received_at) AS INTEGER) AS hour,
               COUNT(*)                                     AS frames
        FROM deauth_events
        WHERE received_at > ?
        GROUP BY day, hour
        ORDER BY day, hour
    """, (cutoff_7d,))

    result = {
        "recent":               recent,
        "bursts":               bursts,
        "summary":              summary[0] if summary else {},
        "top_targets":          top_targets,
        "top_targeted_devices": top_targeted_devices,
        "reason_stats":         reason_rows,
        "heatmap":              heatmap_rows,
    }
    _alerts_cache = (_now_ts, result)
    return result


def build_search(q: str) -> dict:
    if not q or len(q) < 2:
        return {}
    like = f"%{q}%"
    results = {}

    # As a network — BSSID or SSID match in beacon_events
    as_network = query("""
        SELECT bssid, MAX(ssid) AS ssid,
               COUNT(*) AS times_seen,
               MAX(rssi) AS best_rssi, ROUND(AVG(rssi),1) AS avg_rssi,
               MAX(channel) AS channel, MAX(encryption) AS encryption,
               COUNT(DISTINCT node_id) AS node_count,
               MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
        FROM beacon_events
        WHERE bssid LIKE ? OR ssid LIKE ?
        GROUP BY bssid ORDER BY times_seen DESC LIMIT 20
    """, (like, like))
    if as_network:
        results["as_network"] = as_network

    # As a client device — src_mac match in probe_events
    as_client = query("""
        SELECT src_mac,
               COUNT(*) AS times_seen,
               MAX(rssi) AS best_rssi, ROUND(AVG(rssi),1) AS avg_rssi,
               COUNT(DISTINCT node_id) AS node_count,
               MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
        FROM probe_events
        WHERE src_mac LIKE ?
        GROUP BY src_mac LIMIT 10
    """, (like,))
    if as_client:
        ssid_rows = query("""
            SELECT DISTINCT src_mac, ssid FROM probe_events
            WHERE src_mac LIKE ? AND ssid IS NOT NULL AND ssid != ''
        """, (like,))
        ssids_by_mac: dict = {}
        for s in ssid_rows:
            ssids_by_mac.setdefault(s["src_mac"], []).append(s["ssid"])
        for r in as_client:
            vendor, randomized = oui_lookup(r["src_mac"])
            r["vendor"]       = vendor
            r["randomized"]   = randomized
            r["probed_ssids"] = ssids_by_mac.get(r["src_mac"], [])
        results["as_client"] = as_client

    # Devices probing for this SSID
    probing_for = query("""
        SELECT src_mac,
               COUNT(*) AS times_seen,
               MAX(rssi) AS best_rssi,
               MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
        FROM probe_events
        WHERE ssid LIKE ?
        GROUP BY src_mac ORDER BY times_seen DESC LIMIT 20
    """, (like,))
    if probing_for:
        for r in probing_for:
            vendor, randomized = oui_lookup(r["src_mac"])
            r["vendor"]     = vendor
            r["randomized"] = randomized
        results["probing_for"] = probing_for

    # As a deauth victim — dst match
    as_victim = query("""
        SELECT dst,
               COUNT(*) AS total_frames,
               COUNT(DISTINCT bssid) AS networks_used,
               COUNT(DISTINCT node_id) AS node_count,
               MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
        FROM deauth_events
        WHERE dst LIKE ?
        GROUP BY dst LIMIT 10
    """, (like,))
    if as_victim:
        for r in as_victim:
            vendor, randomized = oui_lookup(r["dst"])
            r["vendor"]     = vendor
            r["randomized"] = randomized
        results["as_victim"] = as_victim

    # As an impersonated AP — bssid match in deauth_events
    as_impersonated = query("""
        SELECT bssid,
               COUNT(*) AS total_frames,
               COUNT(DISTINCT dst) AS unique_targets,
               COUNT(DISTINCT node_id) AS node_count,
               MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
        FROM deauth_events
        WHERE bssid LIKE ?
        GROUP BY bssid LIMIT 10
    """, (like,))
    if as_impersonated:
        results["as_impersonated"] = as_impersonated

    # Vendor name search — match query against OUI database, then find those MACs
    q_lower = q.lower()
    matching_prefixes = [
        prefix for prefix, vendor in _OUI_DB.items()
        if q_lower in vendor.lower()
    ][:60]  # cap at 60 prefixes to keep query manageable

    if matching_prefixes:
        ph = ",".join("?" * len(matching_prefixes))

        vendor_clients = query(f"""
            SELECT src_mac,
                   COUNT(*) AS times_seen,
                   MAX(rssi) AS best_rssi, ROUND(AVG(rssi),1) AS avg_rssi,
                   COUNT(DISTINCT node_id) AS node_count,
                   MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
            FROM probe_events
            WHERE substr(src_mac,1,8) IN ({ph})
            GROUP BY src_mac ORDER BY times_seen DESC LIMIT 30
        """, tuple(matching_prefixes))
        if vendor_clients:
            ssid_rows = query(f"""
                SELECT DISTINCT src_mac, ssid FROM probe_events
                WHERE substr(src_mac,1,8) IN ({ph})
                  AND ssid IS NOT NULL AND ssid != ''
            """, tuple(matching_prefixes))
            ssids_by_mac: dict = {}
            for s in ssid_rows:
                ssids_by_mac.setdefault(s["src_mac"], []).append(s["ssid"])
            for r in vendor_clients:
                vendor, randomized = oui_lookup(r["src_mac"])
                r["vendor"]       = vendor
                r["randomized"]   = randomized
                r["probed_ssids"] = ssids_by_mac.get(r["src_mac"], [])
            results["vendor_clients"] = vendor_clients

        vendor_networks = query(f"""
            SELECT bssid, MAX(ssid) AS ssid,
                   COUNT(*) AS times_seen,
                   MAX(rssi) AS best_rssi, ROUND(AVG(rssi),1) AS avg_rssi,
                   MAX(channel) AS channel, MAX(encryption) AS encryption,
                   COUNT(DISTINCT node_id) AS node_count,
                   MIN(received_at) AS first_seen, MAX(received_at) AS last_seen
            FROM beacon_events
            WHERE substr(bssid,1,8) IN ({ph})
            GROUP BY bssid ORDER BY times_seen DESC LIMIT 30
        """, tuple(matching_prefixes))
        if vendor_networks:
            results["vendor_networks"] = vendor_networks

        if "vendor_clients" in results or "vendor_networks" in results:
            results["vendor_name"] = next(
                (v for k, v in _OUI_DB.items() if q_lower in v.lower()), q
            )

    return results


def build_node_detail(node_id: str) -> dict | None:
    # WiFi node path — beacon-derived stats
    rows = query("""
        SELECT
            node_id,
            MIN(received_at) AS first_seen,
            MAX(received_at) AS last_seen,
            COUNT(*)         AS total_events,
            COUNT(DISTINCT ssid)  AS unique_ssids,
            COUNT(DISTINCT bssid) AS unique_bssids,
            AVG(rssi)        AS avg_rssi,
            MIN(rssi)        AS min_rssi,
            MAX(rssi)        AS max_rssi
        FROM beacon_events
        WHERE node_id = ?
        GROUP BY node_id
    """, (node_id,))

    if rows:
        detail = rows[0]
        detail["node_type"] = "wifi"
    else:
        # BLE node path — stats from ble_events
        ble_rows = query("""
            SELECT
                node_id,
                MIN(received_at)    AS first_seen,
                MAX(received_at)    AS last_seen,
                COUNT(*)            AS total_events,
                COUNT(DISTINCT mac) AS unique_macs,
                AVG(rssi)           AS avg_rssi,
                MIN(rssi)           AS min_rssi,
                MAX(rssi)           AS max_rssi
            FROM ble_events
            WHERE node_id = ?
            GROUP BY node_id
        """, (node_id,))

        if ble_rows:
            detail = ble_rows[0]
        else:
            # Node exists (heartbeats only) but no scan data yet
            hb_check = query(
                "SELECT node_id FROM heartbeat_events WHERE node_id = ? LIMIT 1", (node_id,)
            )
            if not hb_check:
                return None
            detail = {
                "node_id":      node_id,
                "first_seen":   None,
                "last_seen":    None,
                "total_events": 0,
                "unique_macs":  0,
                "avg_rssi":     None,
                "min_rssi":     None,
                "max_rssi":     None,
            }

        detail["node_type"]    = "ble"
        detail["unique_ssids"] = 0
        detail["unique_bssids"] = 0

    hb_rows = query("""
        SELECT received_at AS last_heartbeat, uptime_ms, free_heap, wifi_rssi
        FROM heartbeat_events
        WHERE node_id = ?
        ORDER BY id DESC LIMIT 1
    """, (node_id,))
    hb = hb_rows[0] if hb_rows else {}
    detail["last_heartbeat"] = hb.get("last_heartbeat")
    detail["uptime_ms"]      = hb.get("uptime_ms")
    detail["free_heap"]      = hb.get("free_heap")
    detail["wifi_rssi"]      = hb.get("wifi_rssi")
    if detail["last_heartbeat"]:
        detail["status"] = _node_status(detail["last_heartbeat"], threshold=30)
    else:
        detail["status"] = _node_status(detail.get("last_seen"), threshold=60)
    detail["avg_rssi"] = round(detail["avg_rssi"], 1) if detail.get("avg_rssi") else None
    detail["events"]   = build_events(limit=EVENT_CAP, node_id=node_id)
    return detail

# ─── Routes ──────────────────────────────────────────────────────────────────

def ensure_schema():
    """Create tables if they don't exist. Runs once at startup."""
    with get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS beacon_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT, received_at TEXT NOT NULL,
                node_id TEXT, node_ts INTEGER, ssid TEXT, bssid TEXT,
                rssi INTEGER, channel INTEGER, encryption TEXT, importance TEXT
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS probe_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT, received_at TEXT NOT NULL,
                node_id TEXT, node_ts INTEGER, src_mac TEXT, ssid TEXT,
                rssi INTEGER, importance TEXT
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS heartbeat_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT, received_at TEXT NOT NULL,
                node_id TEXT, node_ts INTEGER,
                uptime_ms INTEGER, free_heap INTEGER, wifi_rssi INTEGER
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS deauth_events (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                received_at TEXT NOT NULL,
                node_id     TEXT,
                node_ts     INTEGER,
                subtype     TEXT,
                src         TEXT,
                dst         TEXT,
                bssid       TEXT,
                reason      INTEGER,
                rssi        INTEGER
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS assoc_events (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                received_at TEXT NOT NULL,
                node_id     TEXT,
                node_ts     INTEGER,
                subtype     TEXT,
                src         TEXT,
                bssid       TEXT,
                ssid        TEXT,
                rssi        INTEGER
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS ble_events (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                received_at TEXT NOT NULL,
                node_id     TEXT,
                node_ts     INTEGER,
                mac         TEXT,
                addr_type   TEXT,
                name        TEXT,
                rssi        INTEGER,
                mfr_id      INTEGER,
                mfr_data    TEXT
            )
        """)
        conn.executescript("""
            CREATE INDEX IF NOT EXISTS idx_beacon_received_at      ON beacon_events  (received_at);
            CREATE INDEX IF NOT EXISTS idx_beacon_node_id          ON beacon_events  (node_id);
            CREATE INDEX IF NOT EXISTS idx_beacon_bssid            ON beacon_events  (bssid);
            CREATE INDEX IF NOT EXISTS idx_beacon_bssid_time       ON beacon_events  (bssid, received_at);
            CREATE INDEX IF NOT EXISTS idx_probe_received_at       ON probe_events   (received_at);
            CREATE INDEX IF NOT EXISTS idx_probe_node_id           ON probe_events   (node_id);
            CREATE INDEX IF NOT EXISTS idx_probe_src_mac           ON probe_events   (src_mac);
            CREATE INDEX IF NOT EXISTS idx_probe_src_time          ON probe_events   (src_mac, received_at);
            CREATE INDEX IF NOT EXISTS idx_heartbeat_node_id       ON heartbeat_events (node_id);
            CREATE INDEX IF NOT EXISTS idx_deauth_received_at      ON deauth_events  (received_at);
            CREATE INDEX IF NOT EXISTS idx_deauth_node_id          ON deauth_events  (node_id);
            CREATE INDEX IF NOT EXISTS idx_deauth_bssid            ON deauth_events  (bssid);
            CREATE INDEX IF NOT EXISTS idx_deauth_bssid_time       ON deauth_events  (bssid, received_at);
            CREATE INDEX IF NOT EXISTS idx_deauth_src              ON deauth_events  (src);
            CREATE INDEX IF NOT EXISTS idx_deauth_dst              ON deauth_events  (dst);
            CREATE INDEX IF NOT EXISTS idx_assoc_received_at       ON assoc_events   (received_at);
            CREATE INDEX IF NOT EXISTS idx_assoc_node_id           ON assoc_events   (node_id);
            CREATE INDEX IF NOT EXISTS idx_assoc_src               ON assoc_events   (src);
            CREATE INDEX IF NOT EXISTS idx_assoc_bssid             ON assoc_events   (bssid);
            CREATE INDEX IF NOT EXISTS idx_ble_received_at         ON ble_events     (received_at);
            CREATE INDEX IF NOT EXISTS idx_ble_node_id             ON ble_events     (node_id);
            CREATE INDEX IF NOT EXISTS idx_ble_mac                 ON ble_events     (mac);
            CREATE INDEX IF NOT EXISTS idx_ble_mfr_id              ON ble_events     (mfr_id);
        """)
        conn.commit()


PRUNE_INTERVAL_SECS      = 6 * 3600  # run pruning every 6 hours
CHECKPOINT_INTERVAL_SECS = 30 * 60   # force WAL checkpoint every 30 minutes


def prune_old_events() -> None:
    """Delete events older than HOT_DAYS from all event tables, then checkpoint."""
    cutoff = _hot_cutoff()
    tables = ["beacon_events", "probe_events", "deauth_events", "assoc_events", "ble_events", "heartbeat_events"]
    with get_conn() as conn:
        for table in tables:
            conn.execute(f"DELETE FROM {table} WHERE received_at < ?", (cutoff,))
        conn.commit()
    checkpoint_wal()


def checkpoint_wal() -> None:
    """
    Force a RESTART checkpoint — writes all WAL frames to the main DB file
    and resets the WAL back to the beginning. Prevents the WAL from growing
    unbounded under sustained write load, which causes DB corruption over time.
    RESTART blocks until all current readers finish, then checkpoints fully.
    """
    with get_conn() as conn:
        conn.execute("PRAGMA wal_checkpoint(RESTART)")


app = FastAPI(title="ESP32 Recon Dashboard")
ensure_schema()


@app.on_event("startup")
async def start_background_tasks():
    async def _pruner():
        while True:
            await asyncio.sleep(PRUNE_INTERVAL_SECS)
            await asyncio.to_thread(prune_old_events)

    async def _checkpointer():
        while True:
            await asyncio.sleep(CHECKPOINT_INTERVAL_SECS)
            await asyncio.to_thread(checkpoint_wal)

    asyncio.create_task(_pruner())
    asyncio.create_task(_checkpointer())


@app.get("/", response_class=HTMLResponse)
async def index():
    return FileResponse(HTML_PATH)


@app.get("/static/dashboard.css")
async def serve_css():
    return FileResponse(CSS_PATH, media_type="text/css")


@app.get("/static/dashboard.js")
async def serve_js():
    return FileResponse(JS_PATH, media_type="application/javascript")


@app.get("/api/nodes")
async def api_nodes():
    return await asyncio.to_thread(build_nodes)


@app.get("/api/events")
async def api_events(limit: int = EVENT_CAP):
    limit = min(limit, EVENT_CAP)
    return await asyncio.to_thread(build_events, limit)


@app.get("/api/networks")
async def api_networks():
    return await asyncio.to_thread(build_networks)


@app.get("/api/clients")
async def api_clients():
    return await asyncio.to_thread(build_clients)


@app.get("/api/cross-node")
async def api_cross_node():
    return await asyncio.to_thread(build_cross_node)


@app.get("/api/presence")
async def api_presence():
    return await asyncio.to_thread(build_presence)


@app.get("/api/rssi-history")
async def api_rssi_history(bssid: str, hours: int = 2):
    hours = min(max(hours, 1), 48)
    cutoff = (datetime.datetime.utcnow() - datetime.timedelta(hours=hours)).isoformat(timespec="seconds")

    def _fetch():
        rows = query("""
            SELECT node_id, received_at, rssi
            FROM beacon_events
            WHERE bssid = ? AND received_at >= ?
            ORDER BY received_at ASC
        """, (bssid, cutoff))
        series: dict = {}
        for r in rows:
            nid = r["node_id"]
            if nid not in series:
                series[nid] = []
            series[nid].append({"t": r["received_at"], "rssi": r["rssi"]})
        return series

    return await asyncio.to_thread(_fetch)


@app.get("/api/sessions")
async def api_sessions():
    return await asyncio.to_thread(build_sessions)


@app.get("/api/search")
async def api_search(q: str = ""):
    return await asyncio.to_thread(build_search, q)


@app.get("/api/alerts")
async def api_alerts():
    return await asyncio.to_thread(build_alerts)


@app.get("/api/heartbeats")
async def api_heartbeats():
    """Latest heartbeat per node."""
    return await asyncio.to_thread(query, """
        SELECT node_id, received_at, uptime_ms, free_heap, wifi_rssi
        FROM heartbeat_events
        WHERE id IN (SELECT MAX(id) FROM heartbeat_events GROUP BY node_id)
    """)


@app.get("/api/nodes/{node_id}")
async def api_node_detail(node_id: str):
    detail = await asyncio.to_thread(build_node_detail, node_id)
    if detail is None:
        raise HTTPException(status_code=404, detail="Node not found")
    return detail


@app.get("/api/ble")
async def api_ble():
    data = await asyncio.to_thread(build_ble)
    return data


@app.get("/api/stream")
async def api_stream():
    """SSE stream — polls DB every 2 s, pushes new events to the browser."""
    async def event_generator():
        try:
            last_id = 0
            # Start from the current latest event so we don't replay history on connect
            rows = await asyncio.to_thread(query, "SELECT MAX(id) AS max_id FROM beacon_events")
            if rows and rows[0]["max_id"]:
                last_id = rows[0]["max_id"]

            while True:
                await asyncio.sleep(2)
                new_rows = await asyncio.to_thread(query, """
                    SELECT * FROM beacon_events
                    WHERE id > ?
                    ORDER BY id ASC
                """, (last_id,))
                for row in new_rows:
                    last_id = row["id"]
                    payload = json.dumps(row)
                    yield f"data: {payload}\n\n"
        except asyncio.CancelledError:
            pass  # client disconnected or server shutting down — expected, not an error

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )


def build_ble() -> dict:
    global _ble_cache
    now = datetime.datetime.utcnow().timestamp()
    if _ble_cache and (now - _ble_cache[0]) < _BLE_CACHE_TTL:
        return _ble_cache[1]

    cutoff = _hot_cutoff()

    devices = query("""
        SELECT
            mac, addr_type,
            MAX(name)        AS name,
            mfr_id,
            MAX(mfr_data)    AS mfr_data,
            MAX(rssi)        AS best_rssi,
            COUNT(*)         AS times_seen,
            MIN(received_at) AS first_seen,
            MAX(received_at) AS last_seen,
            COUNT(DISTINCT node_id) AS node_count
        FROM ble_events
        WHERE received_at >= ?
        GROUP BY mac
        ORDER BY times_seen DESC
        LIMIT 300
    """, (cutoff,))

    for d in devices:
        d["type"]   = ble_device_type(d["mfr_id"], d["mfr_data"])
        d["vendor"] = BLE_COMPANIES.get(d["mfr_id"], "Unknown") if d["mfr_id"] is not None else "Unknown"

    feed = query("""
        SELECT received_at, node_id, mac, addr_type, name, rssi, mfr_id, mfr_data
        FROM ble_events
        WHERE received_at >= ?
        ORDER BY id DESC
        LIMIT 100
    """, (cutoff,))

    for f in feed:
        f["type"] = ble_device_type(f["mfr_id"], f["mfr_data"])

    # Manufacturer breakdown
    breakdown = query("""
        SELECT mfr_id, COUNT(DISTINCT mac) AS unique_devices
        FROM ble_events
        WHERE received_at >= ?
        GROUP BY mfr_id
        ORDER BY unique_devices DESC
        LIMIT 20
    """, (cutoff,))

    for b in breakdown:
        b["vendor"] = BLE_COMPANIES.get(b["mfr_id"], "Unknown") if b["mfr_id"] is not None else "No mfr data"

    result = {"devices": devices, "feed": feed, "breakdown": breakdown}
    _ble_cache = (now, result)
    return result


# ─── Entry point ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("dashboard:app", host="0.0.0.0", port=8080, reload=False)