feat(zigbee): import to pending section with edge persistence

Coordinator auto-approves to a canvas Node; routers/end devices land in pending_devices keyed by IEEE. Discovered parent->child edges are stored in pending_device_links so that approving a pending device later auto-creates the Edge once both endpoints exist as canvas Nodes. - new POST /api/v1/zigbee/import-pending (default mode in modal) - new pending_device_links table; ieee_address on nodes + pending_devices - pending_devices.ip migrated to nullable (table rebuild on existing DBs) - approve / bulk-approve return auto-created edges; sidebar pushes them into the canvas store with bottom -> top-t handles - ZigbeeImportModal: radio toggle pending vs canvas; reset on close - PendingDeviceModal: zigbee badge, IEEE/LQI/vendor/model rows, services hidden for zigbee - Sidebar pending row: ZIG source badge, LQI badge, friendly_name fallback - SearchBar: null-safe IP, also searches friendly_name and ieee_address - Tooltip trigger uses asChild to avoid nested-button hydration error
2026-05-07 23:02:03 +02:00
parent 5e567d4628
commit 3ae159d8d6
15 changed files with 988 additions and 40 deletions
@@ -11,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.api.deps import get_current_user
 from app.core.config import settings
 from app.db.database import AsyncSessionLocal, get_db
-from app.db.models import Node, PendingDevice, ScanRun
+from app.db.models import Edge, Node, PendingDevice, PendingDeviceLink, ScanRun
 from app.schemas.nodes import NodeCreate
 from app.schemas.scan import PendingDeviceResponse, ScanRunResponse
 from app.services.scanner import request_cancel, run_scan
@@ -126,23 +126,31 @@ async def bulk_approve_devices(
    for device in devices:
        device.status = "approved"
        node = Node(
-            label=device.hostname or device.ip,
+            label=device.hostname or device.friendly_name or device.ip or "device",
            type=device.suggested_type or "generic",
            ip=device.ip,
            hostname=device.hostname,
            status="unknown",
            services=device.services or [],
+            ieee_address=device.ieee_address,
        )
        db.add(node)
        created_nodes.append(node)
    await db.flush()  # populates node.id from Python-side default before reading
    node_ids = [n.id for n in created_nodes]
    approved_device_ids = [d.id for d in devices]
+
+    all_edges: list[dict[str, str]] = []
+    for device in devices:
+        all_edges.extend(await _resolve_pending_links_for_ieee(db, device.ieee_address))
+
    await db.commit()
    return {
        "approved": len(node_ids),
        "node_ids": node_ids,
        "device_ids": approved_device_ids,
+        "edges_created": len(all_edges),
+        "edges": all_edges,
        "skipped": len(payload.device_ids) - len(node_ids),
    }

@@ -186,12 +194,102 @@ async def approve_device(
        hostname=node_data.hostname,
        status=node_data.status,
        services=node_data.services or [],
+        ieee_address=device.ieee_address,
    )
    db.add(node)
    await db.flush()
    node_id = node.id
+
+    edges = await _resolve_pending_links_for_ieee(db, device.ieee_address)
+
    await db.commit()
-    return {"approved": True, "node_id": node_id}
+    return {
+        "approved": True,
+        "node_id": node_id,
+        "edges_created": len(edges),
+        "edges": edges,
+    }
+
+
+async def _resolve_pending_links_for_ieee(
+    db: AsyncSession, ieee: str | None
+) -> list[dict[str, str]]:
+    """Materialize edges for any pending_device_links involving ``ieee``.
+
+    For each link where the other endpoint already exists as a canvas Node
+    (matched by ``Node.ieee_address``), create the Edge and drop the link
+    row. Links where the other endpoint is still pending are kept so they
+    can resolve when that endpoint is approved later.
+    """
+    if not ieee:
+        return []
+
+    links_q = await db.execute(
+        select(PendingDeviceLink).where(
+            (PendingDeviceLink.source_ieee == ieee)
+            | (PendingDeviceLink.target_ieee == ieee)
+        )
+    )
+    links = list(links_q.scalars().all())
+    if not links:
+        return []
+
+    # Map every relevant ieee → Node (single query).
+    other_ieees = {
+        link.target_ieee if link.source_ieee == ieee else link.source_ieee
+        for link in links
+    }
+    other_ieees.add(ieee)
+    nodes_q = await db.execute(
+        select(Node).where(Node.ieee_address.in_(other_ieees))
+    )
+    by_ieee = {n.ieee_address: n for n in nodes_q.scalars().all() if n.ieee_address}
+
+    self_node = by_ieee.get(ieee)
+    if self_node is None:
+        return []
+
+    # Pre-fetch existing edges between these node ids so we don't create dups
+    # if the user re-approves a device or had drawn the link manually.
+    candidate_node_ids = [n.id for n in by_ieee.values()]
+    existing_q = await db.execute(
+        select(Edge).where(
+            Edge.source.in_(candidate_node_ids),
+            Edge.target.in_(candidate_node_ids),
+        )
+    )
+    existing_pairs = {(e.source, e.target) for e in existing_q.scalars().all()}
+
+    created: list[dict[str, str]] = []
+    for link in links:
+        other_ieee = (
+            link.target_ieee if link.source_ieee == ieee else link.source_ieee
+        )
+        other_node = by_ieee.get(other_ieee)
+        if other_node is None:
+            continue
+        if link.source_ieee == ieee:
+            src_id, tgt_id = self_node.id, other_node.id
+        else:
+            src_id, tgt_id = other_node.id, self_node.id
+        # Skip if either direction already exists.
+        if (src_id, tgt_id) in existing_pairs or (tgt_id, src_id) in existing_pairs:
+            await db.delete(link)
+            continue
+        edge = Edge(
+            source=src_id,
+            target=tgt_id,
+            type="iot",
+            source_handle="bottom",
+            target_handle="top-t",
+        )
+        db.add(edge)
+        await db.flush()
+        existing_pairs.add((src_id, tgt_id))
+        created.append({"id": edge.id, "source": src_id, "target": tgt_id})
+        await db.delete(link)
+
+    return created


@router.post("/pending/{device_id}/hide")
@@ -1,12 +1,20 @@
 """FastAPI router for Zigbee2MQTT import."""

 import logging
+from typing import Any

 from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import delete as sa_delete
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession

 from app.api.deps import get_current_user
+from app.db.database import get_db
+from app.db.models import Node, PendingDevice, PendingDeviceLink
 from app.schemas.zigbee import (
+    ZigbeeCoordinatorOut,
    ZigbeeEdgeOut,
+    ZigbeeImportPendingResponse,
    ZigbeeImportRequest,
    ZigbeeImportResponse,
    ZigbeeNodeOut,
@@ -58,6 +66,162 @@ async def import_zigbee_network(
    return ZigbeeImportResponse(nodes=nodes, edges=edges, device_count=len(nodes))


+@router.post("/import-pending", response_model=ZigbeeImportPendingResponse)
+async def import_zigbee_to_pending(
+    payload: ZigbeeImportRequest,
+    db: AsyncSession = Depends(get_db),
+    _: str = Depends(get_current_user),
+) -> ZigbeeImportPendingResponse:
+    """Fetch the Z2M networkmap and store devices in the pending section.
+
+    Coordinator is auto-approved (creates a canvas Node directly with
+    ``ieee_address`` set). Routers and end devices are upserted into
+    ``pending_devices`` keyed by IEEE address. The discovered parent→child
+    edges are persisted as ``pending_device_links`` rows so that approving a
+    pending device later can auto-create the corresponding Edge when the
+    other endpoint already exists as a canvas Node.
+
+    Re-importing replaces all zigbee-discovered links and updates pending
+    rows in place; pending devices not present in the new map are kept
+    untouched (the user may be mid-approval).
+    """
+    try:
+        nodes_raw, edges_raw = await fetch_networkmap(
+            mqtt_host=payload.mqtt_host,
+            mqtt_port=payload.mqtt_port,
+            base_topic=payload.base_topic,
+            username=payload.mqtt_username,
+            password=payload.mqtt_password,
+            tls=payload.mqtt_tls,
+            tls_insecure=payload.mqtt_tls_insecure,
+        )
+    except ImportError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+    except ConnectionError as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    except TimeoutError as exc:
+        raise HTTPException(status_code=504, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    except Exception as exc:
+        logger.exception("Unexpected error during Zigbee pending import")
+        raise HTTPException(status_code=500, detail="Unexpected error during Zigbee import") from exc
+
+    return await _persist_pending_import(db, nodes_raw, edges_raw)
+
+
+async def _persist_pending_import(
+    db: AsyncSession,
+    nodes_raw: list[dict[str, Any]],
+    edges_raw: list[dict[str, Any]],
+) -> ZigbeeImportPendingResponse:
+    """Upsert nodes/edges into pending_devices + pending_device_links.
+
+    Coordinator auto-approves to a canvas Node. Other devices upsert by IEEE.
+    All zigbee-source links are wiped and re-inserted from the new map.
+    """
+    coordinator_out: ZigbeeCoordinatorOut | None = None
+    coordinator_existed = False
+    pending_created = 0
+    pending_updated = 0
+
+    for n in nodes_raw:
+        ieee = n.get("ieee_address")
+        if not ieee:
+            continue
+        if n.get("device_type") == "Coordinator":
+            existing = await db.execute(select(Node).where(Node.ieee_address == ieee))
+            existing_node = existing.scalar_one_or_none()
+            if existing_node:
+                coordinator_out = ZigbeeCoordinatorOut(
+                    id=existing_node.id,
+                    label=existing_node.label,
+                    ieee_address=ieee,
+                )
+                coordinator_existed = True
+                continue
+            label = n.get("friendly_name") or ieee
+            node = Node(
+                label=label,
+                type=n.get("type") or "zigbee_coordinator",
+                status="unknown",
+                ieee_address=ieee,
+                services=[],
+            )
+            db.add(node)
+            await db.flush()
+            coordinator_out = ZigbeeCoordinatorOut(
+                id=node.id, label=label, ieee_address=ieee
+            )
+            continue
+
+        result = await db.execute(
+            select(PendingDevice).where(PendingDevice.ieee_address == ieee)
+        )
+        pending = result.scalar_one_or_none()
+        if pending is None:
+            db.add(
+                PendingDevice(
+                    ieee_address=ieee,
+                    friendly_name=n.get("friendly_name"),
+                    hostname=n.get("friendly_name"),
+                    suggested_type=n.get("type"),
+                    device_subtype=n.get("device_type"),
+                    model=n.get("model"),
+                    vendor=n.get("vendor"),
+                    lqi=n.get("lqi"),
+                    status="pending",
+                    discovery_source="zigbee",
+                )
+            )
+            pending_created += 1
+        else:
+            pending.friendly_name = n.get("friendly_name") or pending.friendly_name
+            pending.suggested_type = n.get("type") or pending.suggested_type
+            pending.device_subtype = n.get("device_type") or pending.device_subtype
+            pending.model = n.get("model") or pending.model
+            pending.vendor = n.get("vendor") or pending.vendor
+            if n.get("lqi") is not None:
+                pending.lqi = n.get("lqi")
+            if pending.status == "hidden":
+                # Re-imported a hidden device → leave it hidden, just refresh fields.
+                pass
+            pending_updated += 1
+
+    # Replace all zigbee-source links with the freshly discovered set.
+    await db.execute(
+        sa_delete(PendingDeviceLink).where(PendingDeviceLink.discovery_source == "zigbee")
+    )
+
+    links_recorded = 0
+    seen: set[tuple[str, str]] = set()
+    for e in edges_raw:
+        src = e.get("source")
+        tgt = e.get("target")
+        if not src or not tgt or (src, tgt) in seen:
+            continue
+        seen.add((src, tgt))
+        db.add(
+            PendingDeviceLink(
+                source_ieee=src,
+                target_ieee=tgt,
+                discovery_source="zigbee",
+            )
+        )
+        links_recorded += 1
+
+    await db.commit()
+
+    return ZigbeeImportPendingResponse(
+        pending_created=pending_created,
+        pending_updated=pending_updated,
+        coordinator=coordinator_out,
+        coordinator_already_existed=coordinator_existed,
+        links_recorded=links_recorded,
+        device_count=len(nodes_raw),
+    )
+
+
@router.post("/test-connection", response_model=ZigbeeTestConnectionResponse)
 async def test_zigbee_connection(
    payload: ZigbeeTestConnectionRequest,
@@ -80,6 +80,72 @@ async def init_db() -> None:
            await conn.exec_driver_sql("ALTER TABLE nodes ADD COLUMN bottom_handles INTEGER NOT NULL DEFAULT 1")
        with suppress(OperationalError):
            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN discovery_source TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE nodes ADD COLUMN ieee_address TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("CREATE INDEX IF NOT EXISTS ix_nodes_ieee_address ON nodes(ieee_address)")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN ieee_address TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql(
+                "CREATE INDEX IF NOT EXISTS ix_pending_devices_ieee_address "
+                "ON pending_devices(ieee_address)"
+            )
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN friendly_name TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN device_subtype TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN model TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN vendor TEXT")
+        with suppress(OperationalError):
+            await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN lqi INTEGER")
+        # Drop NOT NULL on pending_devices.ip (Zigbee devices have no IP).
+        # SQLite can't ALTER column nullability — rebuild the table if needed.
+        with suppress(OperationalError):
+            info = await conn.exec_driver_sql("PRAGMA table_info(pending_devices)")
+            cols = info.fetchall()
+            ip_col = next((c for c in cols if c[1] == "ip"), None)
+            # PRAGMA table_info row layout: (cid, name, type, notnull, dflt, pk)
+            if ip_col and ip_col[3] == 1:
+                logger.info("Migrating pending_devices: dropping NOT NULL on ip column")
+                await conn.exec_driver_sql("PRAGMA foreign_keys = OFF")
+                await conn.exec_driver_sql(
+                    "CREATE TABLE pending_devices_new ("
+                    "id VARCHAR PRIMARY KEY,"
+                    "ip VARCHAR,"
+                    "mac VARCHAR, hostname VARCHAR, os VARCHAR, services JSON,"
+                    "suggested_type VARCHAR,"
+                    "status VARCHAR,"
+                    "discovery_source VARCHAR,"
+                    "ieee_address VARCHAR,"
+                    "friendly_name VARCHAR,"
+                    "device_subtype VARCHAR,"
+                    "model VARCHAR,"
+                    "vendor VARCHAR,"
+                    "lqi INTEGER,"
+                    "discovered_at DATETIME"
+                    ")"
+                )
+                await conn.exec_driver_sql(
+                    "INSERT INTO pending_devices_new "
+                    "(id, ip, mac, hostname, os, services, suggested_type, status, "
+                    "discovery_source, ieee_address, friendly_name, device_subtype, "
+                    "model, vendor, lqi, discovered_at) "
+                    "SELECT id, ip, mac, hostname, os, services, suggested_type, status, "
+                    "discovery_source, ieee_address, friendly_name, device_subtype, "
+                    "model, vendor, lqi, discovered_at FROM pending_devices"
+                )
+                await conn.exec_driver_sql("DROP TABLE pending_devices")
+                await conn.exec_driver_sql(
+                    "ALTER TABLE pending_devices_new RENAME TO pending_devices"
+                )
+                await conn.exec_driver_sql(
+                    "CREATE INDEX IF NOT EXISTS ix_pending_devices_ieee_address "
+                    "ON pending_devices(ieee_address)"
+                )
+                await conn.exec_driver_sql("PRAGMA foreign_keys = ON")
        with suppress(OperationalError):
            await conn.exec_driver_sql("ALTER TABLE edges ADD COLUMN waypoints JSON")
        with suppress(OperationalError):
@@ -46,6 +46,7 @@ class Node(Base):
    width: Mapped[float | None] = mapped_column(Float, nullable=True)
    height: Mapped[float | None] = mapped_column(Float, nullable=True)
    bottom_handles: Mapped[int] = mapped_column(Integer, default=1)
+    ieee_address: Mapped[str | None] = mapped_column(String, index=True, nullable=True)
    last_seen: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    response_time_ms: Mapped[int | None] = mapped_column(Integer)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
@@ -86,7 +87,7 @@ class PendingDevice(Base):
    __tablename__ = "pending_devices"

    id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid)
-    ip: Mapped[str] = mapped_column(String, nullable=False)
+    ip: Mapped[str | None] = mapped_column(String, nullable=True)
    mac: Mapped[str | None] = mapped_column(String)
    hostname: Mapped[str | None] = mapped_column(String)
    os: Mapped[str | None] = mapped_column(String)
@@ -94,6 +95,31 @@ class PendingDevice(Base):
    suggested_type: Mapped[str | None] = mapped_column(String)
    status: Mapped[str] = mapped_column(String, default="pending")
    discovery_source: Mapped[str | None] = mapped_column(String)
+    ieee_address: Mapped[str | None] = mapped_column(String, index=True, nullable=True, unique=True)
+    friendly_name: Mapped[str | None] = mapped_column(String, nullable=True)
+    device_subtype: Mapped[str | None] = mapped_column(String, nullable=True)
+    model: Mapped[str | None] = mapped_column(String, nullable=True)
+    vendor: Mapped[str | None] = mapped_column(String, nullable=True)
+    lqi: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+
+
+class PendingDeviceLink(Base):
+    """Link between two Zigbee endpoints discovered during import.
+
+    Endpoints are addressed by IEEE (stable across re-imports). Either side may
+    already exist as a canvas Node (resolved via Node.ieee_address) or still be
+    a PendingDevice. On approval, the matching Edge is auto-created when both
+    endpoints exist as canvas Nodes.
+    """
+
+    __tablename__ = "pending_device_links"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid)
+    source_ieee: Mapped[str] = mapped_column(String, nullable=False, index=True)
+    target_ieee: Mapped[str] = mapped_column(String, nullable=False, index=True)
+    lqi: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    discovery_source: Mapped[str] = mapped_column(String, nullable=False, default="zigbee")
    discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)


@@ -6,7 +6,7 @@ from pydantic import BaseModel

 class PendingDeviceResponse(BaseModel):
    id: str
-    ip: str
+    ip: str | None
    mac: str | None
    hostname: str | None
    os: str | None
@@ -14,6 +14,12 @@ class PendingDeviceResponse(BaseModel):
    suggested_type: str | None
    status: str
    discovery_source: str | None
+    ieee_address: str | None = None
+    friendly_name: str | None = None
+    device_subtype: str | None = None
+    model: str | None = None
+    vendor: str | None = None
+    lqi: int | None = None
    discovered_at: datetime

    model_config = {"from_attributes": True}
@@ -76,3 +76,20 @@ class ZigbeeImportResponse(BaseModel):
 class ZigbeeTestConnectionResponse(BaseModel):
    connected: bool
    message: str
+
+
+class ZigbeeCoordinatorOut(BaseModel):
+    id: str
+    label: str
+    ieee_address: str
+
+
+class ZigbeeImportPendingResponse(BaseModel):
+    """Result of importing a Z2M network into the pending section."""
+
+    pending_created: int
+    pending_updated: int
+    coordinator: ZigbeeCoordinatorOut | None = None
+    coordinator_already_existed: bool = False
+    links_recorded: int
+    device_count: int