fix(subscriptions): collapse fingerprint duplicates in upsert

Panels (notably Happ JSON bundles) return the same logical server
multiple times — one outbound per SNI/fingerprint variant for
domain-fronting resilience. Observed on s.stun.su returning 1256
outbounds for happ-macos UA across 303 unique (addr,port,uuid)
servers. The pre-1.3.6 delete-and-insert flow created 1256 Node rows;
the 1.3.6 upsert kept them around forever as orphans (their fp is in
seen_fps so not deleted, but only one row per fp is the matched
'existing' so others are never updated either).

Two fixes:

1. Dedup parsed list by fingerprint before the upsert loop — multiple
   SNI variants of the same (protocol, addr, port, uuid, password)
   collapse to one Node row, last-wins on mutable fields.

2. Collapse legacy duplicates from the DB on every refresh. Pick the
   smallest-id row as the survivor (stable choice — minimizes external
   reference breakage) and transparently remap any active_node_id /
   NodeCircle.node_ids reference from a dup to the survivor before
   deleting the dup. User never notices the cleanup.

Tests: 3 new in TestSubscriptionRefreshDedupsParsed covering parsed
dedup, legacy DB dedup, and cross-reference remap (active + circle
both pointing at dups that get collapsed).

Author: DaveBugg

backend/app/api/subscriptions.py

@@ -426,14 +426,98 @@ async def _fetch_subscription_unlocked(sub_id: int) -> None:
         #      remap is possible, pick the first remaining enabled +
         #      online node from this subscription as a fallback so the
         #      user doesn't lose proxy after a refresh.
+
+        # First: dedup `parsed` by fingerprint. Panels (especially Happ
+        # JSON bundles) often expose the SAME (addr, port, uuid) server
+        # under multiple SNI / fingerprint / sid combos for domain-
+        # fronting resilience — each is one outbound entry. Our Node
+        # model treats (protocol, addr, port, uuid, password) as the
+        # unit (see `_node_fingerprint`), so collapse variants to one
+        # row, last-wins. Without this, the upsert touches only one
+        # row per fingerprint and the other duplicates from the OLD
+        # delete-and-insert era stay forever as orphans (their fp is
+        # in `seen_fps` → not removed; never the matched `existing` →
+        # not updated). Seen in the wild on a Happ-macos subscription
+        # that returned 1256 outbounds for 303 unique servers; without
+        # this dedup the row count never collapsed back to 303.
+        deduped: dict[str, dict] = {}
+        for n in parsed:
+            fp = _node_fingerprint(n)
+            deduped[fp] = n  # last-wins
+        parsed_dedup_skipped = len(parsed) - len(deduped)
+        parsed = list(deduped.values())
+        if parsed_dedup_skipped > 0:
+            logger.info(
+                "Subscription %d: collapsed %d duplicate parsed entries "
+                "(same fingerprint, different SNI/fp variants)",
+                sub_id, parsed_dedup_skipped,
+            )
+
         old_nodes = (await session.exec(
             select(Node).where(Node.subscription_id == sub_id)
         )).all()
+        # `old_by_fp` keys by fingerprint; multiple old rows with the
+        # same fingerprint (legacy duplicates from pre-1.3.6 inserts)
+        # collapse here — we keep the survivor with the smallest id
+        # (so external references — active_node_id, NodeCircle,
+        # RoutingRule — that point at the lowest id of a fingerprint
+        # group keep working) and remap all references on the other
+        # rows to the survivor before deleting them.
         old_by_fp: dict = {}
         old_by_id: dict = {}
-        for n in old_nodes:
-            old_by_fp[_node_row_fingerprint(n)] = n
+        # Process in id-ascending order so the FIRST seen for each fp
+        # is the smallest id → "survivor" is stable.
+        for n in sorted(old_nodes, key=lambda r: r.id):
+            fp = _node_row_fingerprint(n)
+            if fp not in old_by_fp:
+                old_by_fp[fp] = n
             old_by_id[n.id] = n
+        # Build {legacy_dup_id → survivor_id} map for transparent remap.
+        legacy_dup_remap: dict[int, int] = {}
+        for n in old_nodes:
+            fp = _node_row_fingerprint(n)
+            survivor = old_by_fp[fp]
+            if survivor.id != n.id:
+                legacy_dup_remap[n.id] = survivor.id
+
+        if legacy_dup_remap:
+            # Rewrite NodeCircle.node_ids so legacy-dup ids are
+            # transparently swapped for their fingerprint survivor.
+            # Also dedup within the list (a circle that referenced
+            # both halves of a dup pair shouldn't end up with the
+            # same survivor id twice).
+            import json as _json
+            all_circles_pre = (await session.exec(select(NodeCircle))).all()
+            for circle in all_circles_pre:
+                try:
+                    ids = (
+                        _json.loads(circle.node_ids)
+                        if isinstance(circle.node_ids, str)
+                        else (circle.node_ids or [])
+                    )
+                except Exception:
+                    continue
+                remapped: list[int] = []
+                seen: set[int] = set()
+                for i in ids:
+                    new_i = legacy_dup_remap.get(i, i)
+                    if new_i not in seen:
+                        remapped.append(new_i)
+                        seen.add(new_i)
+                if remapped != ids:
+                    if circle.current_index >= len(remapped):
+                        circle.current_index = 0
+                    circle.node_ids = _json.dumps(remapped)
+                    session.add(circle)
+
+            # Delete the legacy dup rows now.
+            for dup_id in legacy_dup_remap:
+                await session.delete(old_by_id[dup_id])
+            logger.info(
+                "Subscription %d: removed %d legacy duplicate Node rows "
+                "(same fingerprint as another row, refs remapped to survivor)",
+                sub_id, len(legacy_dup_remap),
+            )
 
         # Snapshot active node id (may live in this subscription or in
         # another one — we only care if it's in THIS subscription's
@@ -450,6 +534,25 @@ async def _fetch_subscription_unlocked(sub_id: int) -> None:
         active_was_in_sub = (
             active_id_before is not None and active_id_before in old_by_id
         )
+        # If the active node was one of the legacy duplicates we just
+        # deleted, transparently remap to the surviving sibling with
+        # the same fingerprint and persist immediately. This keeps the
+        # user's "active" pin on the same logical server through the
+        # dedup pass, with no UI gap.
+        if (
+            active_id_before is not None
+            and active_id_before in legacy_dup_remap
+        ):
+            survivor_id = legacy_dup_remap[active_id_before]
+            logger.warning(
+                "Subscription %d: active_node_id %d was a legacy duplicate "
+                "of %d — remapping to survivor",
+                sub_id, active_id_before, survivor_id,
+            )
+            if active_row is not None:
+                active_row.value = str(survivor_id)
+                session.add(active_row)
+            active_id_before = survivor_id  # downstream heal sees survivor
 
         # Field copy list — keep in sync with Node ORM. We deliberately
         # don't blow away `order` / `last_check` / `latency_ms` /

backend/tests/test_subscriptions.py

@@ -422,6 +422,190 @@ def test_empty_subscription_response_does_not_touch_circle(
         assert circle_after.enabled is True
 
 
+class TestSubscriptionRefreshDedupsParsed:
+    """Panels (especially Happ JSON bundles) often return the SAME
+    (addr,port,uuid) server under multiple SNI/fingerprint variants.
+    Our Node model treats those as a single row (see
+    `_node_fingerprint`). The upsert must collapse parsed entries to
+    one-per-fingerprint, otherwise legacy duplicate rows accumulate
+    and never get cleaned up. Also: refs from active_node + circles
+    must remap from any deleted dup to the surviving sibling."""
+
+    def test_parsed_duplicates_collapse_to_unique_count(
+        self, client, admin_user, auth_headers, session,
+    ):
+        import asyncio, json
+        from app.api.subscriptions import _fetch_subscription_unlocked
+        from app.models import Subscription, Node
+
+        sub = Subscription(name="Test", url="http://example/sub", enabled=True)
+        session.add(sub)
+        session.commit()
+        session.refresh(sub)
+
+        # Panel returns 5 lines for the SAME (addr,port,uuid), differing
+        # only in SNI — these should collapse to 1 Node row.
+        body = "\n".join(
+            f"vless://uuid-A@server.example:443?type=tcp&sni=sni{i}.example#name{i}"
+            for i in range(5)
+        )
+
+        with mock.patch(
+            "app.api.subscriptions.httpx.AsyncClient"
+        ) as mock_client:
+            instance = mock_client.return_value.__aenter__.return_value
+            instance.get = AsyncMock(return_value=mock.Mock(
+                status_code=200, text=body,
+                raise_for_status=lambda: None,
+            ))
+            asyncio.run(_fetch_subscription_unlocked(sub.id))
+
+        session.expire_all()
+        nodes = session.query(Node).filter(
+            Node.subscription_id == sub.id
+        ).all()
+        assert len(nodes) == 1, (
+            f"5 SNI variants of one server should collapse to 1 Node, "
+            f"got {len(nodes)}"
+        )
+
+    def test_legacy_duplicates_in_db_collapse_on_refresh(
+        self, client, admin_user, auth_headers, session,
+    ):
+        """Inverse case: DB already carries 3 legacy duplicate rows
+        (same fingerprint) from pre-1.3.6 inserts; refresh must keep
+        the smallest-id row and delete the other two."""
+        import asyncio, json
+        from app.api.subscriptions import _fetch_subscription_unlocked
+        from app.models import Subscription, Node
+
+        sub = Subscription(name="Test", url="http://example/sub", enabled=True)
+        session.add(sub)
+        session.commit()
+        session.refresh(sub)
+
+        # 3 rows, all with same (protocol, addr, port, uuid) — only
+        # SNI varies. They're legacy dups that need collapsing.
+        rows = [
+            Node(name=f"dup{i}", protocol="vless", address="server.example",
+                 port=443, uuid="uuid-A", transport="tcp", sni=f"sni{i}",
+                 subscription_id=sub.id, enabled=True)
+            for i in range(3)
+        ]
+        for r in rows:
+            session.add(r)
+        session.commit()
+        for r in rows:
+            session.refresh(r)
+        ids_before = sorted(r.id for r in rows)
+        survivor_id = ids_before[0]
+
+        with mock.patch(
+            "app.api.subscriptions.httpx.AsyncClient"
+        ) as mock_client:
+            instance = mock_client.return_value.__aenter__.return_value
+            instance.get = AsyncMock(return_value=mock.Mock(
+                status_code=200,
+                text="vless://uuid-A@server.example:443?type=tcp&sni=fresh#name",
+                raise_for_status=lambda: None,
+            ))
+            asyncio.run(_fetch_subscription_unlocked(sub.id))
+
+        session.expire_all()
+        nodes = session.query(Node).filter(
+            Node.subscription_id == sub.id
+        ).all()
+        assert len(nodes) == 1, (
+            f"3 legacy dups should collapse to 1, got {len(nodes)}"
+        )
+        assert nodes[0].id == survivor_id, (
+            f"survivor should be smallest-id row {survivor_id}, "
+            f"got {nodes[0].id}"
+        )
+
+    def test_active_node_and_circle_remap_through_legacy_dup_collapse(
+        self, client, admin_user, auth_headers, session,
+    ):
+        """When the active node OR a circle member is one of the
+        deleted legacy dups, both must transparently remap to the
+        surviving sibling — user never notices."""
+        import asyncio, json
+        from app.api.subscriptions import _fetch_subscription_unlocked
+        from app.models import Subscription, Node, NodeCircle
+        from app.models import Settings as DBSettings
+
+        sub = Subscription(name="Test", url="http://example/sub", enabled=True)
+        session.add(sub)
+        session.commit()
+        session.refresh(sub)
+
+        # Two dup groups: group A (3 rows, same fingerprint) and
+        # group B (2 rows, same fingerprint).
+        a_rows = [
+            Node(name=f"A{i}", protocol="vless", address="srv-a",
+                 port=443, uuid="uuid-A", transport="tcp", sni=f"sni-a{i}",
+                 subscription_id=sub.id, enabled=True)
+            for i in range(3)
+        ]
+        b_rows = [
+            Node(name=f"B{i}", protocol="vless", address="srv-b",
+                 port=443, uuid="uuid-B", transport="tcp", sni=f"sni-b{i}",
+                 subscription_id=sub.id, enabled=True)
+            for i in range(2)
+        ]
+        for r in a_rows + b_rows:
+            session.add(r)
+        session.commit()
+        for r in a_rows + b_rows:
+            session.refresh(r)
+
+        a_survivor = sorted(r.id for r in a_rows)[0]
+        a_dup = sorted(r.id for r in a_rows)[2]  # one of the dups to die
+        b_survivor = sorted(r.id for r in b_rows)[0]
+        b_dup = sorted(r.id for r in b_rows)[1]
+
+        # Active node pinned at a dup that's about to die
+        session.add(DBSettings(key="active_node_id", value=str(a_dup)))
+        # Circle uses one dup of A and one dup of B (both must remap)
+        circle = NodeCircle(
+            name="cross-dup", node_ids=json.dumps([a_dup, b_dup]),
+            mode="sequential", interval_min=5, interval_max=10,
+            current_index=0, enabled=True,
+        )
+        session.add(circle)
+        session.commit()
+        session.refresh(circle)
+        circle_id = circle.id
+
+        body = (
+            "vless://uuid-A@srv-a:443?type=tcp&sni=fresh-a#A\n"
+            "vless://uuid-B@srv-b:443?type=tcp&sni=fresh-b#B\n"
+        )
+        with mock.patch(
+            "app.api.subscriptions.httpx.AsyncClient"
+        ) as mock_client:
+            instance = mock_client.return_value.__aenter__.return_value
+            instance.get = AsyncMock(return_value=mock.Mock(
+                status_code=200, text=body,
+                raise_for_status=lambda: None,
+            ))
+            asyncio.run(_fetch_subscription_unlocked(sub.id))
+
+        session.expire_all()
+        # Active remapped to A's survivor
+        active = session.query(DBSettings).filter(
+            DBSettings.key == "active_node_id"
+        ).first()
+        assert int(active.value) == a_survivor, (
+            f"active_node_id should remap from dup {a_dup} to "
+            f"survivor {a_survivor}, got {active.value}"
+        )
+        # Circle remapped both members
+        circle_after = session.get(NodeCircle, circle_id)
+        assert json.loads(circle_after.node_ids) == [a_survivor, b_survivor]
+        assert circle_after.enabled is True
+
+
 class TestSubscriptionRefreshMutex:
     """The endpoint must refuse a second `/refresh` while a previous
     one is still in flight. Without this, two clicks within ~100ms