From c361f593c35f835551ec17569011be579f0d40db Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rodrigo=20P=C3=A9rez?= <ce5rpy@qmd.cl>
Date: Mon, 8 Jun 2026 17:36:32 -0400
Subject: [PATCH] feat(proxy): ProxyService domain and use cases (V2-P3-001)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add ClientSlot/UpstreamPortRange, ProxyUseCases with in-memory store and
RPTO queue, plus peer_id packet helper — library only, not wired to runtime.
---
 src/adn_server/application/ports.py           |  45 ++++++++
 src/adn_server/application/proxy/__init__.py  |  10 ++
 .../application/proxy/packet_helpers.py       |  47 ++++++++
 src/adn_server/application/proxy/use_cases.py | 104 ++++++++++++++++++
 src/adn_server/domain/proxy.py                |  68 ++++++++++++
 .../infrastructure/proxy/__init__.py          |   9 ++
 .../infrastructure/proxy/rpto_queue.py        |  22 ++++
 .../infrastructure/proxy/slot_store.py        |  51 +++++++++
 tests/application/test_proxy_use_cases.py     | 100 +++++++++++++++++
 9 files changed, 456 insertions(+)
 create mode 100644 src/adn_server/application/proxy/__init__.py
 create mode 100644 src/adn_server/application/proxy/packet_helpers.py
 create mode 100644 src/adn_server/application/proxy/use_cases.py
 create mode 100644 src/adn_server/domain/proxy.py
 create mode 100644 src/adn_server/infrastructure/proxy/__init__.py
 create mode 100644 src/adn_server/infrastructure/proxy/rpto_queue.py
 create mode 100644 src/adn_server/infrastructure/proxy/slot_store.py
 create mode 100644 tests/application/test_proxy_use_cases.py

diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py
index 9f18c96..1390149 100644
--- a/src/adn_server/application/ports.py
+++ b/src/adn_server/application/ports.py
@@ -305,6 +305,50 @@ class SubscriptionStore(ABC):
         ...
 
 
+class ProxySlotStore(ABC):
+    """Hotspot session registry: peer_id ↔ upstream port (Phase 3)."""
+
+    @abstractmethod
+    def bind(self, slot: "ClientSlot") -> None:
+        ...
+
+    @abstractmethod
+    def update_client(self, peer_id: bytes, host: str, port: int) -> None:
+        ...
+
+    @abstractmethod
+    def unbind(self, peer_id: bytes) -> "ClientSlot | None":
+        ...
+
+    @abstractmethod
+    def get_by_peer(self, peer_id: bytes) -> "ClientSlot | None":
+        ...
+
+    @abstractmethod
+    def get_by_upstream(self, upstream_port: int) -> "ClientSlot | None":
+        ...
+
+    @abstractmethod
+    def free_upstream_ports(self) -> tuple[int, ...]:
+        ...
+
+    @abstractmethod
+    def list_slots(self) -> tuple["ClientSlot", ...]:
+        ...
+
+
+class PendingRptoQueue(ABC):
+    """Pending RPTO payloads for self-service / login options push."""
+
+    @abstractmethod
+    def enqueue(self, peer_id: bytes, payload: bytes) -> None:
+        ...
+
+    @abstractmethod
+    def dequeue(self) -> tuple[bytes, bytes] | None:
+        ...
+
+
 class PeerTransport(Protocol):
     """Built-in mesh codec (dmre_v5, obp_v1) — decode datagrams to ``MeshIngress``, encode ``MeshEgress``."""
 
@@ -321,3 +365,4 @@ class PeerTransport(Protocol):
 
 if TYPE_CHECKING:
     from adn_server.domain.mesh_routing import MeshEgress, MeshIngress, PeerMeshConfig
+    from adn_server.domain.proxy import ClientSlot
diff --git a/src/adn_server/application/proxy/__init__.py b/src/adn_server/application/proxy/__init__.py
new file mode 100644
index 0000000..41d6fff
--- /dev/null
+++ b/src/adn_server/application/proxy/__init__.py
@@ -0,0 +1,10 @@
+"""Proxy application layer (Phase 3)."""
+
+from .packet_helpers import peer_id_from_packet
+from .use_cases import ProxySlotError, ProxyUseCases
+
+__all__ = [
+    "ProxySlotError",
+    "ProxyUseCases",
+    "peer_id_from_packet",
+]
diff --git a/src/adn_server/application/proxy/packet_helpers.py b/src/adn_server/application/proxy/packet_helpers.py
new file mode 100644
index 0000000..413c9fb
--- /dev/null
+++ b/src/adn_server/application/proxy/packet_helpers.py
@@ -0,0 +1,47 @@
+"""Extract peer_id from Homebrew packets (legacy ``adn_proxy.application.packet_commands``)."""
+
+from __future__ import annotations
+
+# Homebrew command prefixes (wire vocabulary; no I/O)
+_DMRD = b"DMRD"
+_DMRA = b"DMRA"
+_MSTC = b"MSTC"
+_MSTN = b"MSTN"
+_MSTP = b"MSTP"
+_RPTA = b"RPTA"
+_RPTCL = b"RPTCL"
+_RPTK = b"RPTK"
+_RPTL = b"RPTL"
+_RPTC = b"RPTC"
+_RPTO = b"RPTO"
+_RPTP = b"RPTP"
+
+
+def peer_id_from_packet(data: bytes, *, from_master: bool) -> bytes | None:
+    """Return 4-byte peer_id from packet payload, or None if not applicable."""
+    if len(data) < 8:
+        return None
+    command = data[:4]
+    if from_master:
+        if command == _DMRD and len(data) >= 15:
+            return data[11:15]
+        if command == _RPTA and len(data) >= 10:
+            return data[6:10]
+        if command == _MSTN and len(data) >= 10:
+            return data[6:10]
+        if command == _MSTP and len(data) >= 11:
+            return data[7:11]
+        if command == _MSTC and len(data) >= 9:
+            return data[5:9]
+        return None
+    if command == _DMRD and len(data) >= 15:
+        return data[11:15]
+    if command in (_DMRA, _RPTL, _RPTK, _RPTO) and len(data) >= 8:
+        return data[4:8]
+    if command == _RPTC:
+        if len(data) >= 5 and data[:5] == _RPTCL:
+            return data[5:9] if len(data) >= 9 else None
+        return data[4:8] if len(data) >= 8 else None
+    if command == _RPTP and len(data) >= 11:
+        return data[7:11]
+    return None
diff --git a/src/adn_server/application/proxy/use_cases.py b/src/adn_server/application/proxy/use_cases.py
new file mode 100644
index 0000000..185b8f7
--- /dev/null
+++ b/src/adn_server/application/proxy/use_cases.py
@@ -0,0 +1,104 @@
+"""ProxyService use cases: client registration, upstream routing, pending RPTO (Phase 3)."""
+
+from __future__ import annotations
+
+import random
+from collections.abc import Sequence
+
+from adn_server.application.ports import PendingRptoQueue, ProxySlotStore
+from adn_server.domain.errors import DomainError
+from adn_server.domain.proxy import ClientEndpoint, ClientSlot, PendingRpto, UpstreamPortRange
+from adn_server.domain.result import Fail, Result, Success
+from adn_server.domain.value_objects import int_id
+
+
+class ProxySlotError(DomainError):
+    """Proxy session allocation or lookup failure."""
+
+
+class ProxyUseCases:
+    """Register hotspot clients, map them to upstream ports, queue RPTO for the master."""
+
+    def __init__(
+        self,
+        slot_store: ProxySlotStore,
+        port_range: UpstreamPortRange,
+        rpto_queue: PendingRptoQueue,
+        *,
+        black_list: Sequence[int] = (),
+        rng: random.Random | None = None,
+    ) -> None:
+        self._slots = slot_store
+        self._port_range = port_range
+        self._rpto_queue = rpto_queue
+        self._black_list = frozenset(black_list)
+        self._rng = rng or random.Random()
+
+    def attach_client(
+        self,
+        peer_id: bytes,
+        host: str,
+        port: int,
+    ) -> Result[ClientSlot, ProxySlotError]:
+        """Bind or refresh a hotspot session (legacy ``peer_track`` on client packet)."""
+        if len(peer_id) != 4:
+            return Fail(ProxySlotError("peer_id must be 4 bytes"))
+        existing = self._slots.get_by_peer(peer_id)
+        if existing is not None:
+            updated = existing.with_client(host, port)
+            self._slots.update_client(peer_id, host, port)
+            return Success(updated)
+        if int_id(peer_id) in self._black_list:
+            return Fail(ProxySlotError("peer is blacklisted"))
+        upstream_port = self._pick_upstream_port()
+        if upstream_port is None:
+            return Fail(ProxySlotError("no upstream ports available"))
+        slot = ClientSlot(
+            peer_id=peer_id,
+            client=ClientEndpoint(host=host, port=port),
+            upstream_port=upstream_port,
+        )
+        self._slots.bind(slot)
+        return Success(slot)
+
+    def detach_client(self, peer_id: bytes) -> ClientSlot | None:
+        """Release session and upstream port (legacy ``reaper`` without I/O side effects)."""
+        return self._slots.unbind(peer_id)
+
+    def resolve_upstream(self, peer_id: bytes) -> int | None:
+        """Upstream port for forwarding client → master."""
+        slot = self._slots.get_by_peer(peer_id)
+        return slot.upstream_port if slot else None
+
+    def resolve_client(self, upstream_port: int) -> ClientEndpoint | None:
+        """Client endpoint for forwarding master → hotspot."""
+        slot = self._slots.get_by_upstream(upstream_port)
+        return slot.client if slot else None
+
+    def schedule_rpto(self, peer_id: bytes, payload: bytes) -> bool:
+        """Queue RPTO body for a connected peer (self-service / login options)."""
+        slot = self._slots.get_by_peer(peer_id)
+        if slot is None:
+            return False
+        self._rpto_queue.enqueue(peer_id, payload)
+        return True
+
+    def next_pending_rpto(self) -> PendingRpto | None:
+        """Dequeue one pending RPTO with its upstream port (for master send loop)."""
+        item = self._rpto_queue.dequeue()
+        if item is None:
+            return None
+        peer_id, payload = item
+        slot = self._slots.get_by_peer(peer_id)
+        if slot is None:
+            return None
+        return PendingRpto(peer_id=peer_id, payload=payload, upstream_port=slot.upstream_port)
+
+    def list_slots(self) -> tuple[ClientSlot, ...]:
+        return self._slots.list_slots()
+
+    def _pick_upstream_port(self) -> int | None:
+        free = self._slots.free_upstream_ports()
+        if not free:
+            return None
+        return self._rng.choice(free)
diff --git a/src/adn_server/domain/proxy.py b/src/adn_server/domain/proxy.py
new file mode 100644
index 0000000..49c9458
--- /dev/null
+++ b/src/adn_server/domain/proxy.py
@@ -0,0 +1,68 @@
+"""Hotspot proxy domain: client sessions and upstream port bindings (Phase 3)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class ClientEndpoint:
+    """Repeater UDP endpoint (legacy ``shost`` / ``sport``)."""
+
+    host: str
+    port: int
+
+
+@dataclass(frozen=True, slots=True)
+class UpstreamPortRange:
+    """Fan-in UDP ports on the peer server (``PORT`` .. ``PORT+GENERATOR-1``)."""
+
+    port_start: int
+    port_count: int
+
+    def __post_init__(self) -> None:
+        if self.port_count < 1:
+            raise ValueError("port_count must be >= 1")
+        if self.port_start < 1:
+            raise ValueError("port_start must be >= 1")
+
+    @property
+    def port_end(self) -> int:
+        return self.port_start + self.port_count - 1
+
+    def ports(self) -> tuple[int, ...]:
+        return tuple(range(self.port_start, self.port_end + 1))
+
+
+@dataclass(frozen=True, slots=True)
+class UpstreamBinding:
+    """Maps one fan-in listen port to the internal master HBP endpoint."""
+
+    listen_port: int
+    master_host: str
+    system_name: str | None = None
+
+
+@dataclass(slots=True)
+class ClientSlot:
+    """Active hotspot session (legacy ``peer_track`` entry)."""
+
+    peer_id: bytes
+    client: ClientEndpoint
+    upstream_port: int
+
+    def with_client(self, host: str, port: int) -> ClientSlot:
+        return ClientSlot(
+            peer_id=self.peer_id,
+            client=ClientEndpoint(host=host, port=port),
+            upstream_port=self.upstream_port,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class PendingRpto:
+    """Options payload queued for delivery to the master on a peer upstream port."""
+
+    peer_id: bytes
+    payload: bytes
+    upstream_port: int
diff --git a/src/adn_server/infrastructure/proxy/__init__.py b/src/adn_server/infrastructure/proxy/__init__.py
new file mode 100644
index 0000000..8ee21ed
--- /dev/null
+++ b/src/adn_server/infrastructure/proxy/__init__.py
@@ -0,0 +1,9 @@
+"""Proxy infrastructure adapters (Phase 3)."""
+
+from .rpto_queue import InMemoryPendingRptoQueue
+from .slot_store import InMemoryProxySlotStore
+
+__all__ = [
+    "InMemoryPendingRptoQueue",
+    "InMemoryProxySlotStore",
+]
diff --git a/src/adn_server/infrastructure/proxy/rpto_queue.py b/src/adn_server/infrastructure/proxy/rpto_queue.py
new file mode 100644
index 0000000..476b5f3
--- /dev/null
+++ b/src/adn_server/infrastructure/proxy/rpto_queue.py
@@ -0,0 +1,22 @@
+"""In-memory pending RPTO queue (Phase 3)."""
+
+from __future__ import annotations
+
+from collections import deque
+
+from adn_server.application.ports import PendingRptoQueue
+
+
+class InMemoryPendingRptoQueue(PendingRptoQueue):
+    """FIFO queue of RPTO payloads keyed by peer_id."""
+
+    def __init__(self) -> None:
+        self._items: deque[tuple[bytes, bytes]] = deque()
+
+    def enqueue(self, peer_id: bytes, payload: bytes) -> None:
+        self._items.append((peer_id, payload))
+
+    def dequeue(self) -> tuple[bytes, bytes] | None:
+        if not self._items:
+            return None
+        return self._items.popleft()
diff --git a/src/adn_server/infrastructure/proxy/slot_store.py b/src/adn_server/infrastructure/proxy/slot_store.py
new file mode 100644
index 0000000..e33ff61
--- /dev/null
+++ b/src/adn_server/infrastructure/proxy/slot_store.py
@@ -0,0 +1,51 @@
+"""In-memory proxy slot store (Phase 3; not wired to UDP yet)."""
+
+from __future__ import annotations
+
+from adn_server.application.ports import ProxySlotStore
+from adn_server.domain.proxy import ClientSlot, UpstreamPortRange
+
+
+class InMemoryProxySlotStore(ProxySlotStore):
+    """Track ``ClientSlot`` by peer_id and upstream port occupancy."""
+
+    def __init__(self, port_range: UpstreamPortRange) -> None:
+        self._port_range = port_range
+        self._by_peer: dict[bytes, ClientSlot] = {}
+        self._by_upstream: dict[int, bytes] = {}
+
+    def bind(self, slot: ClientSlot) -> None:
+        if slot.upstream_port in self._by_upstream:
+            raise ValueError(f"upstream port {slot.upstream_port} already bound")
+        if slot.peer_id in self._by_peer:
+            raise ValueError("peer_id already bound")
+        self._by_peer[slot.peer_id] = slot
+        self._by_upstream[slot.upstream_port] = slot.peer_id
+
+    def update_client(self, peer_id: bytes, host: str, port: int) -> None:
+        slot = self._by_peer.get(peer_id)
+        if slot is None:
+            raise KeyError(peer_id)
+        self._by_peer[peer_id] = slot.with_client(host, port)
+
+    def unbind(self, peer_id: bytes) -> ClientSlot | None:
+        slot = self._by_peer.pop(peer_id, None)
+        if slot is None:
+            return None
+        self._by_upstream.pop(slot.upstream_port, None)
+        return slot
+
+    def get_by_peer(self, peer_id: bytes) -> ClientSlot | None:
+        return self._by_peer.get(peer_id)
+
+    def get_by_upstream(self, upstream_port: int) -> ClientSlot | None:
+        peer_id = self._by_upstream.get(upstream_port)
+        if peer_id is None:
+            return None
+        return self._by_peer.get(peer_id)
+
+    def free_upstream_ports(self) -> tuple[int, ...]:
+        return tuple(p for p in self._port_range.ports() if p not in self._by_upstream)
+
+    def list_slots(self) -> tuple[ClientSlot, ...]:
+        return tuple(self._by_peer.values())
diff --git a/tests/application/test_proxy_use_cases.py b/tests/application/test_proxy_use_cases.py
new file mode 100644
index 0000000..8f75a0f
--- /dev/null
+++ b/tests/application/test_proxy_use_cases.py
@@ -0,0 +1,100 @@
+"""Proxy use cases and packet helpers (Phase 3)."""
+
+from __future__ import annotations
+
+import random
+
+import pytest
+
+from adn_server.application.proxy import ProxyUseCases, peer_id_from_packet
+from adn_server.domain.proxy import ClientEndpoint, UpstreamPortRange
+from adn_server.domain.result import is_fail, is_ok
+from adn_server.domain.value_objects import bytes_4
+from adn_server.infrastructure.proxy import InMemoryPendingRptoQueue, InMemoryProxySlotStore
+
+_PORT_RANGE = UpstreamPortRange(port_start=56400, port_count=4)
+_PEER_A = bytes_4(1234567)
+_PEER_B = bytes_4(7654321)
+
+
+def _service(*, black_list: tuple[int, ...] = (), seed: int = 1) -> ProxyUseCases:
+    return ProxyUseCases(
+        InMemoryProxySlotStore(_PORT_RANGE),
+        _PORT_RANGE,
+        InMemoryPendingRptoQueue(),
+        black_list=black_list,
+        rng=random.Random(seed),
+    )
+
+
+def test_attach_allocates_upstream_port_and_refreshes_client() -> None:
+    svc = _service()
+    first = svc.attach_client(_PEER_A, "10.0.0.1", 62031)
+    assert is_ok(first)
+    slot = first.value
+    assert slot.upstream_port in _PORT_RANGE.ports()
+    assert slot.client.host == "10.0.0.1"
+
+    again = svc.attach_client(_PEER_A, "10.0.0.2", 62032)
+    assert is_ok(again)
+    assert again.value.upstream_port == slot.upstream_port
+    assert again.value.client == ClientEndpoint(host="10.0.0.2", port=62032)
+
+
+def test_attach_rejects_blacklisted_peer() -> None:
+    svc = _service(black_list=(1234567,))
+    result = svc.attach_client(_PEER_A, "10.0.0.1", 62031)
+    assert is_fail(result)
+
+
+def test_attach_fails_when_ports_exhausted() -> None:
+    svc = _service()
+    for n in range(4):
+        peer = bytes_4(1000 + n)
+        assert is_ok(svc.attach_client(peer, "10.0.0.1", 62031 + n))
+    assert is_fail(svc.attach_client(bytes_4(9999), "10.0.0.9", 62099))
+
+
+def test_detach_frees_upstream_port() -> None:
+    svc = _service()
+    slot = svc.attach_client(_PEER_A, "10.0.0.1", 62031).value
+    removed = svc.detach_client(_PEER_A)
+    assert removed == slot
+    assert svc.resolve_upstream(_PEER_A) is None
+    assert is_ok(svc.attach_client(_PEER_B, "10.0.0.5", 62035))
+
+
+def test_resolve_client_by_upstream_port() -> None:
+    svc = _service()
+    slot = svc.attach_client(_PEER_A, "192.168.1.10", 12345).value
+    client = svc.resolve_client(slot.upstream_port)
+    assert client is not None
+    assert client.host == "192.168.1.10"
+    assert client.port == 12345
+
+
+def test_schedule_and_dequeue_rpto() -> None:
+    svc = _service()
+    svc.attach_client(_PEER_A, "10.0.0.1", 62031)
+    payload = b"TS1=123;TS2=456;"
+    assert svc.schedule_rpto(_PEER_A, payload) is True
+    assert svc.schedule_rpto(bytes_4(999), payload) is False
+    pending = svc.next_pending_rpto()
+    assert pending is not None
+    assert pending.peer_id == _PEER_A
+    assert pending.payload == payload
+    assert pending.upstream_port == svc.resolve_upstream(_PEER_A)
+
+
+@pytest.mark.parametrize(
+    ("data", "from_master", "expected"),
+    [
+        (b"RPTL" + _PEER_A + b"\x00", False, _PEER_A),
+        (b"DMRD" + b"\x00" * 7 + _PEER_A + b"\x00", False, _PEER_A),
+        (b"DMRD" + b"\x00" * 7 + _PEER_B, True, _PEER_B),
+        (b"RPTA" + b"\x00" * 2 + _PEER_A, True, _PEER_A),
+        (b"RPTL\x00", False, None),
+    ],
+)
+def test_peer_id_from_packet(data: bytes, from_master: bool, expected: bytes | None) -> None:
+    assert peer_id_from_packet(data, from_master=from_master) == expected