diff --git a/backend/app/api/routes/scan.py b/backend/app/api/routes/scan.py index 1b3fd06..2b778d2 100644 --- a/backend/app/api/routes/scan.py +++ b/backend/app/api/routes/scan.py @@ -15,8 +15,12 @@ from app.db.database import AsyncSessionLocal, get_db from app.db.models import Design, Edge, Node, PendingDevice, PendingDeviceLink, ScanRun from app.schemas.nodes import NodeCreate from app.schemas.scan import PendingDeviceResponse, ScanRunResponse +from app.services.node_dedupe import dedupe_nodes_by_ieee from app.services.scanner import DeepScanOptions, _valid_port_range, request_cancel, run_scan -from app.services.zigbee_service import build_zigbee_properties +from app.services.zigbee_service import ( + build_zigbee_properties, + merge_zigbee_properties, +) from app.services.zwave_service import build_zwave_properties _ZIGBEE_TYPES = {"zigbee_coordinator", "zigbee_router", "zigbee_enddevice"} @@ -308,6 +312,9 @@ async def bulk_approve_devices( db: AsyncSession = Depends(get_db), _: str = Depends(get_current_user), ) -> dict[str, Any]: + # Repair any legacy same-canvas duplicate nodes before placing more. + await dedupe_nodes_by_ieee(db) + # Target the design the user is on; fall back to the first design. default_design_id = payload.design_id if default_design_id is None: @@ -469,6 +476,35 @@ async def approve_device( raise HTTPException(status_code=409, detail="Device already processed") device.status = "approved" wireless = _is_wireless(node_data.type) + + # Guard against a true duplicate: same IEEE already placed on THIS design. + # (The same device on a *different* design is valid — one Node per canvas.) + if device.ieee_address is not None: + dup = ( + await db.execute( + select(Node).where( + Node.ieee_address == device.ieee_address, + Node.design_id == node_design_id, + ) + ) + ).scalars().first() + if dup is not None: + dup.properties = merge_zigbee_properties( + dup.properties, + _wireless_properties( + node_data.type, device.ieee_address, + device.vendor, device.model, device.lqi, + ) if wireless else [], + ) + edges = await _resolve_pending_links_for_ieee(db, device.ieee_address) + await db.commit() + return { + "approved": True, + "node_id": dup.id, + "edges_created": len(edges), + "edges": edges, + } + # Prefer the MAC discovered during the scan (stored on the pending device); # fall back to whatever the approve payload carried. _mac = device.mac or node_data.mac diff --git a/backend/app/api/routes/zigbee.py b/backend/app/api/routes/zigbee.py index 9e69af1..dd62039 100644 --- a/backend/app/api/routes/zigbee.py +++ b/backend/app/api/routes/zigbee.py @@ -23,6 +23,7 @@ from app.schemas.zigbee import ( ZigbeeTestConnectionRequest, ZigbeeTestConnectionResponse, ) +from app.services.node_dedupe import dedupe_nodes_by_ieee from app.services.zigbee_service import ( build_zigbee_properties, fetch_networkmap, @@ -138,6 +139,10 @@ async def _persist_pending_import( 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. """ + # Repair any pre-existing duplicate nodes (same IEEE) before upserting, so + # the by-IEEE lookups below resolve to a single row. + await dedupe_nodes_by_ieee(db) + # Determine target design (use first design as fallback) first_design = (await db.execute(select(Design).order_by(Design.created_at).limit(1))).scalar() default_design_id = first_design.id if first_design else None @@ -156,15 +161,22 @@ async def _persist_pending_import( ) 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: - existing_node.properties = merge_zigbee_properties( - existing_node.properties, props + # The coordinator may sit on several canvases (one Node per design). + # Refresh props on every matching node, not just one. + existing_nodes = ( + await db.execute( + select(Node).where(Node.ieee_address == ieee).order_by(Node.id) ) + ).scalars().all() + if existing_nodes: + for existing_node in existing_nodes: + existing_node.properties = merge_zigbee_properties( + existing_node.properties, props + ) + first = existing_nodes[0] coordinator_out = ZigbeeCoordinatorOut( - id=existing_node.id, - label=existing_node.label, + id=first.id, + label=first.label, ieee_address=ieee, ) coordinator_existed = True @@ -188,16 +200,18 @@ async def _persist_pending_import( continue # If the device has already been approved as a canvas Node, refresh - # its properties and skip creating a pending row (keeps approved - # devices out of pending/hidden modals on re-import). - existing_node_q = await db.execute( - select(Node).where(Node.ieee_address == ieee) - ) - existing_node = existing_node_q.scalar_one_or_none() - if existing_node: - existing_node.properties = merge_zigbee_properties( - existing_node.properties, props + # its properties on every canvas it sits on and skip creating a pending + # row (keeps approved devices out of pending/hidden modals on re-import). + existing_nodes = ( + await db.execute( + select(Node).where(Node.ieee_address == ieee).order_by(Node.id) ) + ).scalars().all() + if existing_nodes: + for existing_node in existing_nodes: + existing_node.properties = merge_zigbee_properties( + existing_node.properties, props + ) continue result = await db.execute( diff --git a/backend/app/api/routes/zwave.py b/backend/app/api/routes/zwave.py index 241dc38..56fc355 100644 --- a/backend/app/api/routes/zwave.py +++ b/backend/app/api/routes/zwave.py @@ -23,6 +23,7 @@ from app.schemas.zwave import ( ZwaveTestConnectionRequest, ZwaveTestConnectionResponse, ) +from app.services.node_dedupe import dedupe_nodes_by_ieee from app.services.zwave_service import ( build_zwave_properties, fetch_zwave_network, @@ -134,6 +135,10 @@ async def _persist_pending_import( Coordinator auto-approves to a canvas Node. Other devices upsert by Z-Wave identity. All zwave-source links are wiped and re-inserted from the new map. """ + # Repair any pre-existing same-canvas duplicate nodes before upserting, so + # the by-IEEE lookups below resolve cleanly. + await dedupe_nodes_by_ieee(db) + first_design = (await db.execute(select(Design).order_by(Design.created_at).limit(1))).scalar() default_design_id = first_design.id if first_design else None @@ -149,15 +154,22 @@ async def _persist_pending_import( props = build_zwave_properties(ieee, n.get("vendor"), n.get("model")) if n.get("type") == "zwave_coordinator": - existing = await db.execute(select(Node).where(Node.ieee_address == ieee)) - existing_node = existing.scalar_one_or_none() - if existing_node: - existing_node.properties = merge_zwave_properties( - existing_node.properties, props + # The coordinator may sit on several canvases (one Node per design). + # Refresh props on every matching node, not just one. + existing_nodes = ( + await db.execute( + select(Node).where(Node.ieee_address == ieee).order_by(Node.id) ) + ).scalars().all() + if existing_nodes: + for existing_node in existing_nodes: + existing_node.properties = merge_zwave_properties( + existing_node.properties, props + ) + first = existing_nodes[0] coordinator_out = ZwaveCoordinatorOut( - id=existing_node.id, - label=existing_node.label, + id=first.id, + label=first.label, ieee_address=ieee, ) coordinator_existed = True @@ -180,15 +192,18 @@ async def _persist_pending_import( ) continue - # Already approved as a canvas Node → refresh props, skip pending row. - existing_node_q = await db.execute( - select(Node).where(Node.ieee_address == ieee) - ) - existing_node = existing_node_q.scalar_one_or_none() - if existing_node: - existing_node.properties = merge_zwave_properties( - existing_node.properties, props + # Already approved as a canvas Node → refresh props on every canvas it + # sits on, skip pending row. + existing_nodes = ( + await db.execute( + select(Node).where(Node.ieee_address == ieee).order_by(Node.id) ) + ).scalars().all() + if existing_nodes: + for existing_node in existing_nodes: + existing_node.properties = merge_zwave_properties( + existing_node.properties, props + ) continue result = await db.execute( diff --git a/backend/app/services/node_dedupe.py b/backend/app/services/node_dedupe.py new file mode 100644 index 0000000..76fa1cd --- /dev/null +++ b/backend/app/services/node_dedupe.py @@ -0,0 +1,161 @@ +"""Collapse *true* duplicate canvas nodes: same ``ieee_address`` **and** same +``design_id`` (i.e. the same device placed twice on the *same* canvas). + +The same device legitimately appears on multiple canvases — placing a device on +another design creates a second :class:`Node` for that IEEE by design (see the +per-design guard in ``bulk_approve_devices``). Those cross-design rows are NOT +duplicates and must be preserved. Only two rows sharing both the IEEE *and* the +design are corrupt, and only those are collapsed here. + +This module provides an idempotent, **loss-free** repair: per ``(ieee, +design_id)`` group with >1 node it keeps the oldest as canonical, merges the +extras' data into it (properties, missing scalar fields, services), re-points +every edge and ``parent_id`` reference onto the canonical node, then deletes the +extras. Edges are de-duplicated and self-loops dropped after re-pointing so no +dangling or redundant links remain. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import Edge, Node +from app.services.zigbee_service import merge_zigbee_properties + +logger = logging.getLogger(__name__) + +# Scalar Node fields worth carrying over from a duplicate when the canonical +# node has no value. Positions, design, parent and identity fields are left on +# the canonical node untouched (that's the row the user actually placed). +_FILLABLE_FIELDS = ( + "hostname", + "ip", + "mac", + "os", + "check_method", + "check_target", + "notes", + "cpu_count", + "cpu_model", + "ram_gb", + "disk_gb", + "custom_icon", + "last_seen", + "last_scan", +) + + +def _merge_services( + a: list[Any] | None, b: list[Any] | None +) -> list[Any]: + """Union two service lists, de-duplicated, order-stable.""" + out: list[Any] = list(a or []) + seen = {repr(s) for s in out} + for s in b or []: + if repr(s) not in seen: + out.append(s) + seen.add(repr(s)) + return out + + +def _merge_into_canonical(canonical: Node, dup: Node) -> None: + """Fold ``dup``'s data into ``canonical`` in place (no field lost).""" + canonical.properties = merge_zigbee_properties( + canonical.properties, dup.properties or [] + ) + canonical.services = _merge_services(canonical.services, dup.services) + for field in _FILLABLE_FIELDS: + if getattr(canonical, field, None) in (None, "") and getattr(dup, field, None) not in (None, ""): + setattr(canonical, field, getattr(dup, field)) + # Prefer a human label over a bare IEEE/hex fallback. + if (not canonical.label or canonical.label == canonical.ieee_address) and dup.label: + canonical.label = dup.label + + +async def dedupe_nodes_by_ieee(db: AsyncSession) -> int: + """Merge duplicate nodes sharing an ``ieee_address`` AND ``design_id``. + + Returns the number of nodes removed. Idempotent: a no-op when every + ``(ieee, design)`` pair maps to at most one node. Nodes with the same IEEE + on *different* designs are left untouched (valid cross-canvas placement). + Does not commit — the caller owns the transaction. + """ + rows = ( + await db.execute( + select(Node) + .where(Node.ieee_address.is_not(None)) + .order_by(Node.ieee_address, Node.created_at, Node.id) + ) + ).scalars().all() + + groups: dict[tuple[str, str | None], list[Node]] = {} + for node in rows: + groups.setdefault((node.ieee_address, node.design_id), []).append(node) # type: ignore[arg-type] + + removed = 0 + for (ieee, _design), nodes in groups.items(): + if len(nodes) < 2: + continue + canonical, *dups = nodes # oldest first (ordered above) + dup_ids = {d.id for d in dups} + + for dup in dups: + _merge_into_canonical(canonical, dup) + + # Re-point edges + parents, then drop self-loops / duplicates. + edges = ( + await db.execute( + select(Edge).where( + Edge.source.in_(dup_ids) | Edge.target.in_(dup_ids) + ) + ) + ).scalars().all() + for edge in edges: + if edge.source in dup_ids: + edge.source = canonical.id + if edge.target in dup_ids: + edge.target = canonical.id + + # Re-point children whose parent was a duplicate. + children = ( + await db.execute(select(Node).where(Node.parent_id.in_(dup_ids))) + ).scalars().all() + for child in children: + child.parent_id = canonical.id + + # Collapse self-loops and now-redundant parallel edges. + all_edges = ( + await db.execute( + select(Edge).where( + (Edge.source == canonical.id) | (Edge.target == canonical.id) + ) + ) + ).scalars().all() + seen_pairs: set[tuple[str, str, str]] = set() + for edge in all_edges: + if edge.source == edge.target: + await db.delete(edge) + continue + key = (edge.source, edge.target, edge.type) + if key in seen_pairs: + await db.delete(edge) + continue + seen_pairs.add(key) + + await db.flush() + for dup in dups: + await db.delete(dup) + removed += 1 + + logger.info( + "Deduped IEEE %s: merged %d duplicate node(s) into %s", + ieee, len(dups), canonical.id, + ) + + if removed: + await db.flush() + return removed diff --git a/backend/tests/test_node_dedupe.py b/backend/tests/test_node_dedupe.py new file mode 100644 index 0000000..be702f1 --- /dev/null +++ b/backend/tests/test_node_dedupe.py @@ -0,0 +1,117 @@ +"""Tests for the same-canvas node dedupe repair (app.services.node_dedupe).""" + +import pytest +from sqlalchemy import select + +from app.db.models import Design, Edge, Node +from app.services.node_dedupe import dedupe_nodes_by_ieee + + +async def _design(db, name="d1"): + d = Design(name=name) + db.add(d) + await db.flush() + return d + + +@pytest.mark.asyncio +async def test_collapses_same_ieee_same_design(db_session): + d = await _design(db_session) + keep = Node( + label="Sensor", type="zigbee_enddevice", design_id=d.id, + ieee_address="0xAAA", properties=[{"key": "IEEE", "value": "0xAAA", "visible": True}], + pos_x=100, pos_y=200, + ) + db_session.add(keep) + await db_session.flush() + dup = Node( + label="Sensor", type="zigbee_enddevice", design_id=d.id, + ieee_address="0xAAA", ip="10.0.0.5", + properties=[{"key": "LQI", "value": "88", "visible": False}], + ) + db_session.add(dup) + await db_session.flush() + + removed = await dedupe_nodes_by_ieee(db_session) + assert removed == 1 + + nodes = (await db_session.execute(select(Node).where(Node.ieee_address == "0xAAA"))).scalars().all() + assert len(nodes) == 1 + survivor = nodes[0] + assert survivor.id == keep.id # oldest kept + assert survivor.pos_x == 100 # canvas position preserved + assert survivor.ip == "10.0.0.5" # missing field filled from dup + keys = {p["key"] for p in survivor.properties} + assert keys == {"IEEE", "LQI"} # properties merged + + +@pytest.mark.asyncio +async def test_preserves_same_ieee_across_designs(db_session): + """Same device on two canvases is valid — must NOT be merged.""" + d1 = await _design(db_session, "d1") + d2 = await _design(db_session, "d2") + for d in (d1, d2): + db_session.add(Node(label="S", type="zigbee_enddevice", design_id=d.id, ieee_address="0xBBB")) + await db_session.flush() + + removed = await dedupe_nodes_by_ieee(db_session) + assert removed == 0 + + nodes = (await db_session.execute(select(Node).where(Node.ieee_address == "0xBBB"))).scalars().all() + assert len(nodes) == 2 + + +@pytest.mark.asyncio +async def test_repoints_edges_and_drops_dupes(db_session): + d = await _design(db_session) + keep = Node(label="A", type="server", design_id=d.id, ieee_address="0xCCC") + dup = Node(label="A", type="server", design_id=d.id, ieee_address="0xCCC") + other = Node(label="B", type="server", design_id=d.id) + db_session.add_all([keep, other]) + await db_session.flush() + db_session.add(dup) + await db_session.flush() + + # keep<->other and dup<->other (parallel after repoint), plus dup<->keep (self-loop). + db_session.add_all([ + Edge(source=keep.id, target=other.id, type="ethernet", design_id=d.id), + Edge(source=dup.id, target=other.id, type="ethernet", design_id=d.id), + Edge(source=dup.id, target=keep.id, type="ethernet", design_id=d.id), + ]) + await db_session.flush() + + removed = await dedupe_nodes_by_ieee(db_session) + assert removed == 1 + + edges = (await db_session.execute(select(Edge))).scalars().all() + # self-loop dropped, parallel edge collapsed -> a single keep<->other edge + assert len(edges) == 1 + e = edges[0] + assert {e.source, e.target} == {keep.id, other.id} + + +@pytest.mark.asyncio +async def test_repoints_child_parent(db_session): + d = await _design(db_session) + keep = Node(label="Host", type="proxmox", design_id=d.id, ieee_address="0xDDD") + dup = Node(label="Host", type="proxmox", design_id=d.id, ieee_address="0xDDD") + db_session.add(keep) + await db_session.flush() + db_session.add(dup) + await db_session.flush() + child = Node(label="VM", type="vm", design_id=d.id, parent_id=dup.id) + db_session.add(child) + await db_session.flush() + + await dedupe_nodes_by_ieee(db_session) + await db_session.refresh(child) + assert child.parent_id == keep.id + + +@pytest.mark.asyncio +async def test_idempotent_noop_when_unique(db_session): + d = await _design(db_session) + db_session.add(Node(label="X", type="server", design_id=d.id, ieee_address="0xEEE")) + await db_session.flush() + assert await dedupe_nodes_by_ieee(db_session) == 0 + assert await dedupe_nodes_by_ieee(db_session) == 0 diff --git a/backend/tests/test_scan.py b/backend/tests/test_scan.py index 2114bb1..8087e20 100644 --- a/backend/tests/test_scan.py +++ b/backend/tests/test_scan.py @@ -311,6 +311,44 @@ async def test_approve_device(client: AsyncClient, headers, pending_device): assert inventory[0]["status"] == "approved" +@pytest.mark.asyncio +async def test_approve_device_no_dupe_same_ieee_same_design( + client: AsyncClient, headers, db_session +): + """Approving a device whose IEEE is already on the target design must reuse + the existing node, not create a duplicate (source of the crash bug).""" + design = Design(name="d1") + db_session.add(design) + await db_session.flush() + existing = Node( + label="sensor", type="zigbee_enddevice", ieee_address="0xZZZ", + services=[], design_id=design.id, + ) + db_session.add(existing) + device = PendingDevice( + id=str(uuid.uuid4()), ieee_address="0xZZZ", suggested_type="zigbee_enddevice", + status="pending", discovery_source="zigbee", + ) + db_session.add(device) + await db_session.commit() + + res = await client.post( + f"/api/v1/scan/pending/{device.id}/approve", + json={ + "label": "sensor", "type": "zigbee_enddevice", + "status": "online", "services": [], "design_id": design.id, + }, + headers=headers, + ) + assert res.status_code == 200 + assert res.json()["node_id"] == existing.id # reused, not new + + nodes = ( + await db_session.execute(select(Node).where(Node.ieee_address == "0xZZZ")) + ).scalars().all() + assert len(nodes) == 1 # no duplicate created + + @pytest.mark.asyncio async def test_approve_nonexistent_device(client: AsyncClient, headers): node_payload = { diff --git a/backend/tests/test_zigbee_router.py b/backend/tests/test_zigbee_router.py index 5096531..7467f15 100644 --- a/backend/tests/test_zigbee_router.py +++ b/backend/tests/test_zigbee_router.py @@ -618,6 +618,43 @@ async def test_persist_pending_import_refreshes_existing_coordinator_properties( assert by_key["Model"]["visible"] is False +@pytest.mark.asyncio +async def test_persist_pending_import_device_on_multiple_canvases( + db_session, +) -> None: + """Regression: a device approved onto TWO designs (one Node each) must not + crash re-import with MultipleResultsFound — props refresh on both nodes.""" + from sqlalchemy import select + + from app.api.routes.zigbee import _persist_pending_import + from app.db.models import Design, Node + + d1 = Design(name="d1") + d2 = Design(name="d2") + db_session.add_all([d1, d2]) + await db_session.flush() + for d in (d1, d2): + db_session.add(Node( + label="router_1", type="zigbee_router", status="online", + check_method="none", ieee_address="0xR1", services=[], + properties=[], design_id=d.id, + )) + await db_session.commit() + + bumped = [dict(n) for n in _PENDING_NODES] + bumped[1]["lqi"] = 240 + # Must not raise. + await _persist_pending_import(db_session, bumped, _PENDING_EDGES) + + nodes = ( + await db_session.execute(select(Node).where(Node.ieee_address == "0xR1")) + ).scalars().all() + assert len(nodes) == 2 # both canvas placements preserved + for n in nodes: + lqi = {p["key"]: p["value"] for p in n.properties}.get("LQI") + assert lqi == "240" # refreshed on every canvas + + @pytest.mark.asyncio async def test_import_pending_requires_auth(client: AsyncClient) -> None: res = await client.post( diff --git a/backend/tests/test_zwave_router.py b/backend/tests/test_zwave_router.py index 6974738..3aa7130 100644 --- a/backend/tests/test_zwave_router.py +++ b/backend/tests/test_zwave_router.py @@ -445,3 +445,38 @@ async def test_persist_keeps_hidden_hidden(db_session) -> None: ) ).scalar_one() assert still_hidden.status == "hidden" + + +@pytest.mark.asyncio +async def test_persist_device_on_multiple_canvases(db_session) -> None: + """Regression: a device on TWO designs (one Node each) must not crash + re-import with MultipleResultsFound — props refresh on both nodes.""" + from sqlalchemy import select + + from app.api.routes.zwave import _persist_pending_import + from app.db.models import Design, Node + + d1 = Design(name="d1") + d2 = Design(name="d2") + db_session.add_all([d1, d2]) + await db_session.flush() + for d in (d1, d2): + db_session.add(Node( + label="Wall Plug", type="zwave_router", status="online", + check_method="none", ieee_address="zwave-0xh-2", services=[], + properties=[], design_id=d.id, + )) + await db_session.commit() + + bumped = [dict(n) for n in _PENDING_NODES] + bumped[1]["model"] = "ZW200" + # Must not raise. + await _persist_pending_import(db_session, bumped, _PENDING_EDGES) + + nodes = ( + await db_session.execute(select(Node).where(Node.ieee_address == "zwave-0xh-2")) + ).scalars().all() + assert len(nodes) == 2 # both canvas placements preserved + for n in nodes: + model = {p["key"]: p["value"] for p in n.properties}.get("Model") + assert model == "ZW200" # refreshed on every canvas