feat: persist peer dynamic TGs in MariaDB across reconnects

Add DATABASE config, async DynamicTgStore, and restore on RPTC so
SINGLE=0/1 dynamics survive hotspot disconnects and server restarts
without blocking the DMRD voice path.

adn-server.example.yaml

@@ -78,14 +78,17 @@ PROXY:
   BLACK_LIST: []
   IP_BLACK_LIST: {}
 
-# Self-service: same MySQL + PBKDF2 as monitor/adn-monitor.yaml (Clients table).
-SELF_SERVICE:
-  USE_SELFSERVICE: true
+# MariaDB (required): dynamic TG persistence + optional self-service (Clients table).
+DATABASE:
   DB_SERVER: localhost
   DB_USERNAME: hbmon
   DB_PASSWORD: "<set-in-adn-server.yaml>"
   DB_NAME: hbmon
   DB_PORT: 3306
+
+# Self-service: uses DATABASE for MariaDB; PBKDF2 must match adn-monitor.yaml.
+SELF_SERVICE:
+  USE_SELFSERVICE: true
   PBKDF2_SALT: "ADN"
   PBKDF2_ITERATIONS: 2000
 

src/adn_server/application/dynamic_tg_use_cases.py

@@ -0,0 +1,127 @@
+# ADN DMR Peer Server - application dynamic tg use cases
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""Persist/restore per-peer dynamic TGs via ``DynamicTgStore`` port (voice path non-blocking)."""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Callable
+from typing import Any
+
+from adn_server.application.ports import DynamicTgStore
+from adn_server.application.routing.helpers import (
+    _peer_ua_session_entry,
+    peer_receives_group_tgid,
+    peer_single_mode,
+    purge_expired_peer_ua_sessions,
+    restore_peer_ua_entries_to_memory,
+)
+from adn_server.domain import int_id
+from adn_server.domain.dynamic_tg import DynamicTgEntry
+
+logger = logging.getLogger(__name__)
+
+
+class DynamicTgUseCases:
+    def __init__(
+        self,
+        store: DynamicTgStore,
+        *,
+        on_restored: Callable[[bytes, str, dict[str, Any], list[DynamicTgEntry], float], None]
+        | None = None,
+    ) -> None:
+        self._store = store
+        self._on_restored = on_restored
+
+    def persist_after_register(
+        self,
+        peer: dict[str, Any],
+        peer_id: bytes,
+        slot: int,
+        tgid: int,
+        sys_cfg: dict[str, Any],
+        *,
+        system_name: str,
+    ) -> None:
+        """Enqueue DB write (async); memory already updated by register_peer_ua_session."""
+        if int(tgid) <= 0 or int(tgid) == 4000 or peer_receives_group_tgid(peer, slot, int(tgid)):
+            return
+        now = time.time()
+        peer_int = int_id(peer_id)
+        if peer_single_mode(peer, sys_cfg):
+            session = _peer_ua_session_entry(sys_cfg, peer_id, int(slot))
+            if not session:
+                return
+            self._store.replace_single_slot(
+                DynamicTgEntry(
+                    int_id=peer_int,
+                    system_name=system_name,
+                    slot=int(slot),
+                    tgid=int(tgid),
+                    single_mode=True,
+                    expires_at=float(session["expires"]),
+                    updated_at=now,
+                )
+            )
+        else:
+            self._store.upsert(
+                DynamicTgEntry(
+                    int_id=peer_int,
+                    system_name=system_name,
+                    slot=int(slot),
+                    tgid=int(tgid),
+                    single_mode=False,
+                    expires_at=None,
+                    updated_at=now,
+                )
+            )
+
+    def delete_peer_slot(self, peer_id: bytes, system_name: str, slot: int) -> None:
+        self._store.delete_peer_slot(int_id(peer_id), system_name, int(slot))
+
+    def restore_peer(self, peer_id: bytes, system_name: str, sys_cfg: dict[str, Any]) -> Any:
+        """Load from persistence on reconnect (RPTC); returns async handle from port."""
+        now = time.time()
+
+        def _apply(entries: list[DynamicTgEntry]) -> list[int]:
+            active = [
+                e for e in entries
+                if not e.single_mode or e.expires_at is None or float(e.expires_at) > now
+            ]
+            tgids = restore_peer_ua_entries_to_memory(sys_cfg, peer_id, active, now=now)
+            if self._on_restored is not None:
+                self._on_restored(peer_id, system_name, sys_cfg, active, now)
+            if tgids:
+                logger.info(
+                    "(DYNAMIC_TG) Restored %s TG(s) for peer %s on %s: %s",
+                    len(tgids), int_id(peer_id), system_name, sorted(set(tgids)),
+                )
+            return tgids
+
+        return self._store.load_peer(int_id(peer_id), system_name).addCallback(_apply)
+
+    def purge_expired(self, config: dict[str, Any]) -> None:
+        now = time.time()
+        self._store.purge_expired(now)
+        for sys_cfg in config.get("SYSTEMS", {}).values():
+            if isinstance(sys_cfg, dict):
+                purge_expired_peer_ua_sessions(sys_cfg, now=now)

src/adn_server/application/ports.py

@@ -30,6 +30,7 @@ from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any, Protocol
 
 if TYPE_CHECKING:
+    from adn_server.domain.dynamic_tg import DynamicTgEntry
     from adn_server.domain.subscription import AudioChannel, Subscription, SubscriptionId, SubscriptionPhase, SystemId
 
 
@@ -366,6 +367,30 @@ class ProxySelfServiceStore(ABC):
         """Returns Deferred."""
 
 
+class DynamicTgStore(ABC):
+    """Persist per-peer user-activated dynamic TGs (``peer_dynamic_tgs`` table)."""
+
+    @abstractmethod
+    def upsert(self, entry: "DynamicTgEntry") -> None:
+        ...
+
+    @abstractmethod
+    def replace_single_slot(self, entry: "DynamicTgEntry") -> None:
+        ...
+
+    @abstractmethod
+    def delete_peer_slot(self, int_id: int, system_name: str, slot: int) -> None:
+        ...
+
+    @abstractmethod
+    def load_peer(self, int_id: int, system_name: str) -> Any:
+        """Returns Deferred firing with ``list[DynamicTgEntry]``."""
+
+    @abstractmethod
+    def purge_expired(self, now: float) -> None:
+        ...
+
+
 class ProxyIpBlacklist(ABC):
     """Temporary IP blocks (legacy proxy ``ip_black_list`` / PRBL)."""
 

src/adn_server/application/routing/dynamic_tg_restore.py

@@ -0,0 +1,73 @@
+# ADN DMR Peer Server - application routing dynamic tg restore
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""Bridge/subscription sync after dynamic TG rows are restored from persistence."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from adn_server.application.ports import SubscriptionStore
+from adn_server.application.subscription.routing_table_export import _legacy_to_type
+from adn_server.application.subscription.subscription_queries import store_has_table
+from adn_server.domain import bytes_3
+from adn_server.domain.dynamic_tg import DynamicTgEntry
+from adn_server.domain.subscription import SubscriptionPhase
+
+
+def sync_restored_dynamic_bridges(
+    entries: list[DynamicTgEntry],
+    *,
+    system_name: str,
+    peer_id: bytes,
+    sys_cfg: dict[str, Any],
+    sub_store: SubscriptionStore,
+    ensure_dynamic_relay: Callable[[bytes, str, int, float], None],
+    ua_timer_minutes_for_peer: Callable[[str, bytes], float] | None,
+    now: float,
+) -> None:
+    """Align bridge timers and create missing relay tables for restored dynamics."""
+    for entry in entries:
+        if not entry.single_mode or entry.expires_at is None:
+            continue
+        exp = float(entry.expires_at)
+        if exp <= now:
+            continue
+        for sub in sub_store.snapshot():
+            if (
+                sub.table_key() == str(entry.tgid)
+                and sub.system.value == system_name
+                and int(sub.channel.slot) == int(entry.slot)
+            ):
+                sub.state.timer_expires_at = exp
+                if not sub.is_active() and _legacy_to_type(sub) == "ON":
+                    sub.state.phase = SubscriptionPhase.ACTIVE
+                sub_store.upsert(sub)
+    tmout = float(sys_cfg.get("DEFAULT_UA_TIMER", 10))
+    if ua_timer_minutes_for_peer is not None:
+        tmout = float(ua_timer_minutes_for_peer(system_name, peer_id))
+    seen: set[tuple[int, int]] = set()
+    for entry in entries:
+        key = (int(entry.tgid), int(entry.slot))
+        if key in seen or store_has_table(sub_store, str(entry.tgid)):
+            continue
+        seen.add(key)
+        ensure_dynamic_relay(bytes_3(entry.tgid), system_name, int(entry.slot), tmout)

src/adn_server/application/routing/helpers.py

@@ -445,6 +445,58 @@ def export_peer_ua_sessions(
     return out
 
 
+def restore_peer_ua_entries_to_memory(
+    sys_cfg: dict[str, Any],
+    peer_id: bytes,
+    entries: list[Any],
+    *,
+    now: float | None = None,
+) -> list[int]:
+    """Apply persisted dynamic TG rows to ``_PEER_UA_SESSIONS`` / ``_PEER_UA_MULTI_TGS``."""
+    pkt_time = time.time() if now is None else now
+    pk = bytes_4(int_id(peer_id))
+    restored: list[int] = []
+    for entry in entries:
+        tgid = int(entry.tgid)
+        slot = int(entry.slot)
+        if entry.single_mode:
+            expires = entry.expires_at
+            if expires is not None and float(expires) <= pkt_time:
+                continue
+            per_peer = sys_cfg.setdefault("_PEER_UA_SESSIONS", {}).setdefault(pk, {})
+            per_peer[slot] = {
+                "tgid": tgid,
+                "expires": float(expires) if expires is not None else 0.0,
+            }
+            restored.append(tgid)
+        else:
+            multi = sys_cfg.setdefault("_PEER_UA_MULTI_TGS", {}).setdefault(pk, {})
+            multi.setdefault(slot, set()).add(tgid)
+            restored.append(tgid)
+    return restored
+
+
+def purge_expired_peer_ua_sessions(sys_cfg: dict[str, Any], *, now: float | None = None) -> None:
+    """Drop expired SINGLE=1 sessions from in-memory store."""
+    pkt_time = time.time() if now is None else now
+    store = sys_cfg.get("_PEER_UA_SESSIONS")
+    if not isinstance(store, dict):
+        return
+    for pk in list(store.keys()):
+        per_peer = store.get(pk)
+        if not isinstance(per_peer, dict):
+            continue
+        for slot in list(per_peer.keys()):
+            entry = per_peer.get(slot)
+            if not isinstance(entry, dict):
+                continue
+            exp = float(entry.get("expires", 0) or 0)
+            if exp > 0 and pkt_time >= exp:
+                per_peer.pop(slot, None)
+        if not per_peer:
+            store.pop(pk, None)
+
+
 def clear_peer_ua_sessions(
     peer: dict[str, Any],
     sys_cfg: dict[str, Any],

src/adn_server/application/routing/subscription_table.py

@@ -56,6 +56,29 @@ logger = logging.getLogger(__name__)
 class SubscriptionTableMixin:
     """ensure_dynamic_relay, stat/static TG, OPTIONS refresh (RPTO / startup / dmrd)."""
 
+    def sync_restored_dynamic_tgs(
+        self,
+        peer_id: bytes,
+        system_name: str,
+        sys_cfg: dict[str, Any],
+        entries: list["DynamicTgEntry"],
+        *,
+        now: float,
+    ) -> None:
+        """After DB restore: bridge timers and missing relay tables (routing concern)."""
+        from .dynamic_tg_restore import sync_restored_dynamic_bridges
+
+        sync_restored_dynamic_bridges(
+            entries,
+            system_name=system_name,
+            peer_id=peer_id,
+            sys_cfg=sys_cfg,
+            sub_store=self._subscription_store,
+            ensure_dynamic_relay=self.ensure_dynamic_relay,
+            ua_timer_minutes_for_peer=self._ua_timer_minutes_for_peer,
+            now=now,
+        )
+
     def ensure_dynamic_relay(
         self,
         _tgid: bytes | int,

src/adn_server/domain/dynamic_tg.py

@@ -0,0 +1,38 @@
+# ADN DMR Peer Server - domain dynamic tg
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""Per-peer user-activated dynamic TG persistence."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class DynamicTgEntry:
+    """One persisted dynamic TG row for a hotspot peer."""
+
+    int_id: int
+    system_name: str
+    slot: int
+    tgid: int
+    single_mode: bool
+    expires_at: float | None
+    updated_at: float

src/adn_server/infrastructure/bootstrap/peer_server.py

@@ -30,6 +30,7 @@ from typing import Any
 
 from twisted.internet import reactor, task, threads
 
+from adn_server.application.dynamic_tg_use_cases import DynamicTgUseCases
 from adn_server.application import (
     RoutingUseCases,
     IdentUseCases,
@@ -70,6 +71,9 @@ from adn_server.infrastructure.config_normalizer import (
 from adn_server.infrastructure.config_reload import BindSpec, reload_server_config
 from adn_server.infrastructure.logging_config import reopen_file_handlers
 from adn_server.infrastructure.persistence import PickleSubMapStore
+from adn_server.infrastructure.persistence.database_config import database_settings
+from adn_server.infrastructure.persistence.dynamic_tg_repository import MysqlDynamicTgRepository
+from adn_server.infrastructure.persistence.mysql_pool import create_mysql_pool, verify_database_sync
 from adn_server.infrastructure.persistence.alias_loader import DefaultAliasLoader
 from adn_server.infrastructure.persistence.keys_store import JsonKeysStore
 from adn_server.infrastructure.proxy import apply_proxy_config_reload, start_proxy_service
@@ -229,6 +233,24 @@ def run_peer_server(
     runtime_holder = RuntimeContextHolder(RuntimeContext(config=config, config_path=config_path))
     config = ConfigProxy(runtime_holder)
 
+    db_settings = database_settings(config)
+    if not verify_database_sync(
+        db_settings["db_server"],
+        db_settings["db_username"],
+        db_settings["db_password"],
+        db_settings["db_name"],
+        db_settings["db_port"],
+    ):
+        raise SystemExit(1)
+    mysql_pool = create_mysql_pool(
+        db_settings["db_server"],
+        db_settings["db_username"],
+        db_settings["db_password"],
+        db_settings["db_name"],
+        db_settings["db_port"],
+    )
+    dynamic_tg_store = MysqlDynamicTgRepository(mysql_pool)
+
     # Subscription store is runtime routing authority; acl_router is ACL-only (D-08).
     acl_router = InMemoryAclRouter()
     systems_cfg = config.get("SYSTEMS", {})
@@ -345,6 +367,10 @@ def run_peer_server(
         ta_emblc_encoder=default_ta_emblc_encoder,
     )
     routing_use_cases.apply_startup_subscriptions()
+    dynamic_tg_uc = DynamicTgUseCases(
+        dynamic_tg_store,
+        on_restored=routing_use_cases.sync_restored_dynamic_tgs,
+    )
     routing_table_for_report = routing_use_cases.routing_table_for_report
     voice_use_cases._routing_table_for_report = routing_table_for_report
     report_factory.set_routing_table(routing_table_for_report())
@@ -519,6 +545,7 @@ def run_peer_server(
             on_dmra_fragment_stored=routing_use_cases.on_dmra_fragment_stored,
             routing_table_for_report=routing_table_for_report,
             get_subscription_store=lambda: subscription_store,
+            dynamic_tg_uc=dynamic_tg_uc,
         )
 
     def _listen_system(_name: str, bind: BindSpec, protocol: Any) -> Any:
@@ -633,6 +660,12 @@ def run_peer_server(
     task.LoopingCall(routing_use_cases.log_connected_systems_and_tgs).start(60).addErrback(_looping_errback, logger)
     task.LoopingCall(lambda: logger.debug("(ROUTER) KeepAlive reporting loop started")).start(60).addErrback(_looping_errback, logger)
 
+    def dynamic_tg_purge_loop() -> None:
+        dynamic_tg_uc.purge_expired(config)
+
+    task.LoopingCall(dynamic_tg_purge_loop).start(60).addErrback(_looping_errback, logger)
+    logger.info("(DYNAMIC_TG) Expired SINGLE=1 purge loop started (every 60 seconds)")
+
     # UDP listeners per system (same order as legacy: SYSTEM STARTING then instance created per system)
     logger.info("(GLOBAL) ADN DMR Peer Server -- SYSTEM STARTING...")
     for system_name, sys_cfg in systems_cfg.items():
@@ -654,7 +687,9 @@ def run_peer_server(
 
     if proxy_enabled:
         try:
-            proxy_state = start_proxy_service(config, protocols, logger=logger)
+            proxy_state = start_proxy_service(
+                config, protocols, logger=logger, mysql_pool=mysql_pool,
+            )
         except Exception as exc:
             logger.error("(PROXY) failed to start integrated proxy: %s", exc)
             raise

src/adn_server/infrastructure/config_loader.py

@@ -107,6 +107,7 @@ class YamlConfigLoader:
             "ALIASES": data.get("ALIASES", {}),
             "SYSTEMS": data.get("SYSTEMS", {}),
             "PROXY": data.get("PROXY", {}),
+            "DATABASE": data.get("DATABASE", {}),
             "SELF_SERVICE": data.get("SELF_SERVICE", {}),
         }
         apply_proxy_env_overrides(config)

src/adn_server/infrastructure/config_reload.py

@@ -118,7 +118,7 @@ def merge_system_config(old_cfg: dict[str, Any], new_cfg: dict[str, Any]) -> dic
 def merge_top_level_config(config: dict[str, Any], incoming: dict[str, Any]) -> None:
     """Update GLOBAL / REPORTS / ALIASES / LOGGER in the live config dict."""
     kill_flag = config.get("GLOBAL", {}).get("_KILL_SERVER")
-    for key in ("GLOBAL", "REPORTS", "ALIASES", "LOGGER", "PROXY", "SELF_SERVICE"):
+    for key in ("GLOBAL", "REPORTS", "ALIASES", "LOGGER", "PROXY", "DATABASE", "SELF_SERVICE"):
         if key not in incoming:
             continue
         config[key] = copy.deepcopy(incoming[key])

src/adn_server/infrastructure/persistence/database_config.py

@@ -0,0 +1,42 @@
+# ADN DMR Peer Server - infrastructure persistence database config
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""DATABASE block from config (MariaDB for dynamic TG persistence)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def _block(config: dict[str, Any], key: str) -> dict[str, Any]:
+    raw = config.get(key) or {}
+    return raw if isinstance(raw, dict) else {}
+
+
+def database_settings(config: dict[str, Any]) -> dict[str, Any]:
+    """Resolved MariaDB settings from ``DATABASE``."""
+    block = _block(config, "DATABASE")
+    return {
+        "db_server": str(block.get("DB_SERVER", "localhost")),
+        "db_username": str(block.get("DB_USERNAME", "")),
+        "db_password": str(block.get("DB_PASSWORD", "")),
+        "db_name": str(block.get("DB_NAME", "")),
+        "db_port": int(block.get("DB_PORT", 3306)),
+    }

src/adn_server/infrastructure/persistence/dynamic_tg_repository.py

@@ -0,0 +1,109 @@
+# ADN DMR Peer Server - infrastructure persistence dynamic tg repository
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""MySQL ``peer_dynamic_tgs`` table adapter."""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any
+
+from twisted.enterprise import adbapi
+from twisted.internet.defer import inlineCallbacks, returnValue
+
+from adn_server.application.ports import DynamicTgStore
+from adn_server.domain.dynamic_tg import DynamicTgEntry
+
+logger = logging.getLogger(__name__)
+
+
+def _row_to_entry(row: tuple[Any, ...]) -> DynamicTgEntry:
+    return DynamicTgEntry(
+        int_id=int(row[0]),
+        system_name=str(row[1]),
+        slot=int(row[2]),
+        tgid=int(row[3]),
+        single_mode=bool(int(row[4])),
+        expires_at=float(row[5]) if row[5] is not None else None,
+        updated_at=float(row[6]),
+    )
+
+
+class MysqlDynamicTgRepository(DynamicTgStore):
+    """``DynamicTgStore`` via Twisted adbapi."""
+
+    def __init__(self, pool: adbapi.ConnectionPool) -> None:
+        self._pool = pool
+
+    def upsert(self, entry: DynamicTgEntry) -> None:
+        updated = int(entry.updated_at or time.time())
+        expires = int(entry.expires_at) if entry.expires_at is not None else None
+        self._pool.runOperation(
+            """INSERT INTO peer_dynamic_tgs
+               (int_id, system_name, slot, tgid, single_mode, expires_at, updated_at)
+               VALUES (%s, %s, %s, %s, %s, %s, %s)
+               ON DUPLICATE KEY UPDATE
+                 single_mode=VALUES(single_mode),
+                 expires_at=VALUES(expires_at),
+                 updated_at=VALUES(updated_at)""",
+            (
+                entry.int_id,
+                entry.system_name,
+                entry.slot,
+                entry.tgid,
+                int(entry.single_mode),
+                expires,
+                updated,
+            ),
+        ).addErrback(lambda f: logger.error("(DYNAMIC_TG) upsert: %s", f.getTraceback()))
+
+    def replace_single_slot(self, entry: DynamicTgEntry) -> None:
+        self._pool.runOperation(
+            "DELETE FROM peer_dynamic_tgs WHERE int_id=%s AND system_name=%s AND slot=%s AND single_mode=1",
+            (entry.int_id, entry.system_name, entry.slot),
+        ).addCallback(lambda _: self.upsert(entry)).addErrback(
+            lambda f: logger.error("(DYNAMIC_TG) replace_single_slot: %s", f.getTraceback())
+        )
+
+    def delete_peer_slot(self, int_id: int, system_name: str, slot: int) -> None:
+        self._pool.runOperation(
+            "DELETE FROM peer_dynamic_tgs WHERE int_id=%s AND system_name=%s AND slot=%s",
+            (int_id, system_name, slot),
+        ).addErrback(
+            lambda f: logger.error("(DYNAMIC_TG) delete_peer_slot: %s", f.getTraceback())
+        )
+
+    @inlineCallbacks
+    def load_peer(self, int_id: int, system_name: str) -> Any:
+        rows = yield self._pool.runQuery(
+            """SELECT int_id, system_name, slot, tgid, single_mode, expires_at, updated_at
+               FROM peer_dynamic_tgs
+               WHERE int_id=%s AND system_name=%s""",
+            (int_id, system_name),
+        )
+        returnValue([_row_to_entry(row) for row in (rows or [])])
+
+    def purge_expired(self, now: float) -> None:
+        cutoff = int(now)
+        self._pool.runOperation(
+            "DELETE FROM peer_dynamic_tgs WHERE single_mode=1 AND expires_at IS NOT NULL AND expires_at <= %s",
+            (cutoff,),
+        ).addErrback(lambda f: logger.error("(DYNAMIC_TG) purge_expired: %s", f.getTraceback()))

src/adn_server/infrastructure/persistence/mysql_pool.py

@@ -0,0 +1,88 @@
+# ADN DMR Peer Server - infrastructure persistence mysql pool helpers
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""Shared Twisted adbapi MySQL pool helpers."""
+
+from __future__ import annotations
+
+import logging
+
+from twisted.enterprise import adbapi
+
+logger = logging.getLogger(__name__)
+
+
+def create_mysql_pool(
+    host: str,
+    user: str,
+    password: str,
+    db_name: str,
+    port: int,
+) -> adbapi.ConnectionPool:
+    """Create Twisted adbapi pool using ``MySQLdb`` (``mysqlclient`` package)."""
+    return adbapi.ConnectionPool(
+        "MySQLdb",
+        host=host,
+        user=user,
+        passwd=password,
+        db=db_name,
+        port=port,
+        charset="utf8mb4",
+    )
+
+
+def verify_database_sync(
+    host: str,
+    user: str,
+    password: str,
+    db_name: str,
+    port: int,
+) -> bool:
+    """Blocking startup check: MariaDB reachable and ``peer_dynamic_tgs`` exists."""
+    try:
+        import MySQLdb
+    except ImportError as err:
+        logger.critical(
+            "(DATABASE) mysqlclient required for dynamic TG persistence: %s",
+            err,
+        )
+        return False
+    try:
+        conn = MySQLdb.connect(
+            host=host,
+            user=user,
+            passwd=password,
+            db=db_name,
+            port=port,
+            charset="utf8mb4",
+        )
+        cur = conn.cursor()
+        cur.execute("SELECT 1 FROM peer_dynamic_tgs LIMIT 1")
+        cur.close()
+        conn.close()
+        logger.info("(DATABASE) peer_dynamic_tgs table: OK")
+        return True
+    except Exception as err:
+        logger.critical(
+            "(DATABASE) startup check failed: %s "
+            "(configure DATABASE in adn-server.yaml and run adn-monitor db_bootstrap --update)",
+            err,
+        )
+        return False

src/adn_server/infrastructure/proxy/runtime.py

@@ -138,28 +138,30 @@ def _build_self_service(
     client_sender: FanInClientSender,
     *,
     logger: logging.Logger,
+    mysql_pool: Any | None = None,
 ) -> ProxySelfServiceBridge | None:
     ss = self_service_settings(config)
     if not ss["enabled"]:
         return None
     try:
-        from adn_server.infrastructure.proxy.persistence import (
-            ProxySelfServiceRepository,
-            create_pool,
-        )
+        from adn_server.infrastructure.proxy.persistence import ProxySelfServiceRepository
+        if mysql_pool is None:
+            from adn_server.infrastructure.persistence.database_config import database_settings
+            from adn_server.infrastructure.proxy.persistence import create_pool
+            db = database_settings(config)
+            mysql_pool = create_pool(
+                db["db_server"],
+                db["db_username"],
+                db["db_password"],
+                db["db_name"],
+                db["db_port"],
+            )
     except ImportError as err:
         raise RuntimeError(
             "(SELF_SERVICE) USE_SELFSERVICE requires mysqlclient "
             "(pip install mysqlclient)"
         ) from err
-    pool = create_pool(
-        ss["db_server"],
-        ss["db_username"],
-        ss["db_password"],
-        ss["db_name"],
-        ss["db_port"],
-    )
-    store = ProxySelfServiceRepository(pool)
+    store = ProxySelfServiceRepository(mysql_pool)
     bridge = ProxySelfServiceBridge(
         store,
         use_cases,
@@ -177,6 +179,7 @@ def start_proxy_service(
     protocols: dict[str, Any],
     *,
     logger: logging.Logger,
+    mysql_pool: Any | None = None,
 ) -> ProxyServiceState:
     """Start LISTEN_PORT fan-in and inject into ``PROXY.TARGET_SYSTEM`` MASTER."""
     runtime = _proxy_runtime_snapshot(config)
@@ -292,6 +295,7 @@ def start_proxy_service(
             master_sink,
             state.client_sender,
             logger=logger,
+            mysql_pool=mysql_pool,
         )
         if self_service_bridge is not None:
             fanin._self_service = self_service_bridge  # noqa: SLF001

src/adn_server/infrastructure/proxy/self_service_config.py

@@ -18,12 +18,14 @@
 #   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 ###############################################################################
 
-"""SELF_SERVICE block from config (same keys as adn-monitor / adn-proxy)."""
+"""SELF_SERVICE feature flags; MariaDB connection comes from ``DATABASE``."""
 
 from __future__ import annotations
 
 from typing import Any
 
+from adn_server.infrastructure.persistence.database_config import database_settings
+
 
 def self_service_settings(config: dict[str, Any]) -> dict[str, Any]:
     """Resolved self-service settings; disabled when block missing or USE_SELFSERVICE false."""
@@ -31,13 +33,10 @@ def self_service_settings(config: dict[str, Any]) -> dict[str, Any]:
     if not isinstance(block, dict):
         block = {}
     enabled = bool(block.get("USE_SELFSERVICE", block.get("ENABLED", False)))
+    db = database_settings(config)
     return {
         "enabled": enabled,
-        "db_server": str(block.get("DB_SERVER", "localhost")),
-        "db_username": str(block.get("DB_USERNAME", "")),
-        "db_password": str(block.get("DB_PASSWORD", "")),
-        "db_name": str(block.get("DB_NAME", "")),
-        "db_port": int(block.get("DB_PORT", 3306)),
+        **db,
         "pbkdf2_salt": str(block.get("PBKDF2_SALT", "ADN")),
         "pbkdf2_iterations": int(block.get("PBKDF2_ITERATIONS", 2000)),
     }

src/adn_server/infrastructure/twisted_adapters/udp_hbp.py

@@ -190,6 +190,7 @@ class HBPProtocol(DatagramProtocol):
         routing_table_for_report: Callable[[], dict[str, Any]] | None = None,
         get_subscription_store: Callable[[], Any] | None = None,
         mesh_registry: MeshCodecRegistry | None = None,
+        dynamic_tg_uc: Any = None,
     ) -> None:
         self._CONFIG = config
         self._system = system_name
@@ -211,6 +212,7 @@ class HBPProtocol(DatagramProtocol):
         self._on_talker_alias_stream_end = on_talker_alias_stream_end
         self._on_dmra_fragment_stored = on_dmra_fragment_stored
         self._mesh_registry = mesh_registry if mesh_registry is not None else _DEFAULT_MESH_REGISTRY
+        self._dynamic_tg_uc = dynamic_tg_uc
         self._config = config.get("SYSTEMS", {}).get(system_name, {})
         if self._config.get("MODE") == "OPENBRIDGE":
             self._laststrid = deque([], 20)
@@ -401,6 +403,8 @@ class HBPProtocol(DatagramProtocol):
         """Clear per-peer UA dynamics; legacy bridge reset only outside inject-only."""
         peer = self._peers.get(peer_id, {})
         clear_peer_ua_sessions(peer, self._config, peer_id, slot=slot)
+        if self._dynamic_tg_uc is not None:
+            self._dynamic_tg_uc.delete_peer_slot(peer_id, self._system, slot)
         _kind = "Private call to ID" if call_type == "unit" else "Group call to TG"
         if self._inject_multi_peer_options_filter():
             self._push_config_to_monitor()
@@ -799,13 +803,12 @@ class HBPProtocol(DatagramProtocol):
             self._push_config_to_monitor()
 
     def _on_peer_disconnected(self, peer_id: bytes) -> None:
-        """Drop SINGLE/UA session and stale slot STATUS when a hotspot leaves or re-logs in."""
+        """Drop stale slot STATUS when a hotspot leaves; keep persisted UA state in sys_cfg."""
         peer = self._peers.get(peer_id)
-        sys_cfg = self._config
         if peer is not None:
-            clear_peer_ua_sessions(peer, sys_cfg, peer_id)
-        else:
-            clear_peer_ua_sessions({}, sys_cfg, peer_id)
+            sessions = peer.get("_UA_SESSION")
+            if isinstance(sessions, dict):
+                sessions.clear()
         clear_peer_rx_status_slots(self.STATUS, peer_id)
 
     def _remove_peer(self, peer_id: bytes) -> None:
@@ -915,6 +918,15 @@ class HBPProtocol(DatagramProtocol):
                             self._config,
                             now=pkt_time,
                         )
+                        if self._dynamic_tg_uc is not None:
+                            self._dynamic_tg_uc.persist_after_register(
+                                self._peers[_peer_id],
+                                _peer_id,
+                                _slot,
+                                _int_dst_id,
+                                self._config,
+                                system_name=self._system,
+                            )
                         self._mark_downlink_index_dirty()
                         if _prev_single_tg != _int_dst_id:
                             self._push_config_to_monitor()
@@ -1176,6 +1188,13 @@ class HBPProtocol(DatagramProtocol):
                         self._mark_downlink_index_dirty()
                         self._config_push_throttle.note_peer_connected()
                         self._push_config_to_monitor()
+                        if self._dynamic_tg_uc is not None:
+                            _sys_cfg = self._CONFIG.get("SYSTEMS", {}).get(self._system, {})
+                            self._dynamic_tg_uc.restore_peer(
+                                _peer_id, self._system, _sys_cfg,
+                            ).addCallback(
+                                lambda tgids: self._mark_downlink_index_dirty() if tgids else tgids
+                            )
                 else:
                     self.transport.write(b"".join([MSTNAK, _peer_id]), _sockaddr)
                     logger.info("(%s) Peer info from Radio ID that has not logged in: %s", self._system, int_id(_peer_id))
@@ -1189,16 +1208,6 @@ class HBPProtocol(DatagramProtocol):
                 self._mark_downlink_index_dirty()
                 self.send_peer(_peer_id, b"".join([RPTACK, _peer_id]))
                 logger.info("(%s) Peer %s has sent options %s", self._system, _this_peer["CALLSIGN"], _this_peer["OPTIONS"])
-                if is_proxy_inject_only(self._CONFIG, self._system):
-                    for _slot in (1, 2):
-                        if _slot in self.STATUS:
-                            seed_peer_ua_session_from_status(
-                                _this_peer,
-                                _peer_id,
-                                _slot,
-                                self.STATUS[_slot],
-                                self._config,
-                            )
                 # Inject-only multi-hotspot: OPTIONS live on each peer; do not let last RPTO
                 # overwrite the shared SYSTEM row (legacy had one peer per virtual master).
                 if not is_proxy_inject_only(self._CONFIG, self._system):
@@ -1210,6 +1219,16 @@ class HBPProtocol(DatagramProtocol):
                         self._on_options_received(self._system, _this_peer["OPTIONS"])
                     except Exception:
                         pass
+                if is_proxy_inject_only(self._CONFIG, self._system):
+                    for _slot in (1, 2):
+                        if _slot in self.STATUS:
+                            seed_peer_ua_session_from_status(
+                                _this_peer,
+                                _peer_id,
+                                _slot,
+                                self.STATUS[_slot],
+                                self._config,
+                            )
                 self._push_config_to_monitor()
             else:
                 self.transport.write(b"".join([MSTNAK, _peer_id]), _sockaddr)
@@ -1978,6 +1997,7 @@ def HBPProtocolFactory(
     routing_table_for_report: Callable[[], dict[str, Any]] | None = None,
     get_subscription_store: Callable[[], Any] | None = None,
     mesh_registry: MeshCodecRegistry | None = None,
+    dynamic_tg_uc: Any = None,
 ) -> HBPProtocol:
     """Create HBP protocol instance (legacy: one HBSYSTEM per system)."""
     return HBPProtocol(
@@ -2001,4 +2021,5 @@ def HBPProtocolFactory(
         routing_table_for_report=routing_table_for_report,
         get_subscription_store=get_subscription_store,
         mesh_registry=mesh_registry,
+        dynamic_tg_uc=dynamic_tg_uc,
     )

tests/application/test_dynamic_tg_persist.py

@@ -0,0 +1,65 @@
+# ADN DMR Peer Server - tests application dynamic tg persist
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""Persist SINGLE=1 absolute expiry to MariaDB."""
+
+from __future__ import annotations
+
+from adn_server.application.dynamic_tg_use_cases import DynamicTgUseCases
+from adn_server.application.routing.helpers import register_peer_ua_session
+from adn_server.domain.dynamic_tg import DynamicTgEntry
+from tests.application.test_peer_single_downlink import _peer_id, _sys_cfg
+
+
+class _CaptureStore:
+    def __init__(self) -> None:
+        self.replaced: list[DynamicTgEntry] = []
+
+    def upsert(self, entry: DynamicTgEntry) -> None:
+        pass
+
+    def replace_single_slot(self, entry: DynamicTgEntry) -> None:
+        self.replaced.append(entry)
+
+    def delete_peer_slot(self, int_id: int, system_name: str, slot: int) -> None:
+        pass
+
+    def load_peer(self, int_id: int, system_name: str):
+        return []
+
+    def purge_expired(self, now: float) -> None:
+        pass
+
+
+def test_persist_after_register_stores_absolute_expires_from_memory() -> None:
+    peer_id = _peer_id()
+    sys_cfg = _sys_cfg()
+    peer = {"OPTIONS": b"TS2=730,7305;SINGLE=1;TIMER=60;"}
+    now = 1_700_000_000.0
+    register_peer_ua_session(peer, peer_id, 2, 7304, sys_cfg, now=now)
+    store = _CaptureStore()
+    uc = DynamicTgUseCases(store)
+    uc.persist_after_register(
+        peer, peer_id, 2, 7304, sys_cfg, system_name="MASTER-A",
+    )
+    assert len(store.replaced) == 1
+    entry = store.replaced[0]
+    assert entry.expires_at == now + 60.0 * 60.0
+    assert entry.single_mode is True

tests/application/test_dynamic_tg_restore.py

@@ -0,0 +1,151 @@
+# ADN DMR Peer Server - tests application dynamic tg restore
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""Restore persisted dynamic TG rows into in-memory UA stores."""
+
+from __future__ import annotations
+
+import pytest
+
+from adn_server.application.routing.helpers import (
+    peer_should_receive_group_voice,
+    purge_expired_peer_ua_sessions,
+    restore_peer_ua_entries_to_memory,
+)
+from adn_server.domain.dynamic_tg import DynamicTgEntry
+from tests.application.test_peer_single_downlink import _peer_id, _sys_cfg
+
+
+def test_restore_single_mode_respects_expiry() -> None:
+    peer_id = _peer_id()
+    sys_cfg = _sys_cfg()
+    now = 1_000_000.0
+    entries = [
+        DynamicTgEntry(
+            int_id=730039101,
+            system_name="MASTER-A",
+            slot=2,
+            tgid=7305,
+            single_mode=True,
+            expires_at=now + 300.0,
+            updated_at=now,
+        ),
+        DynamicTgEntry(
+            int_id=730039101,
+            system_name="MASTER-A",
+            slot=2,
+            tgid=730,
+            single_mode=True,
+            expires_at=now - 1.0,
+            updated_at=now - 60.0,
+        ),
+    ]
+    restored = restore_peer_ua_entries_to_memory(sys_cfg, peer_id, entries, now=now + 10)
+    assert restored == [7305]
+    peer = {"OPTIONS": b"TS2=730,7305;SINGLE=1;TIMER=5;"}
+    assert peer_should_receive_group_voice(
+        peer, 2, 7305, peer_id=peer_id, connected_count=2, sys_cfg=sys_cfg, now=now + 20
+    )
+    assert not peer_should_receive_group_voice(
+        peer, 2, 730, peer_id=peer_id, connected_count=2, sys_cfg=sys_cfg, now=now + 20
+    )
+
+
+def test_restore_single_zero_multi_dynamic() -> None:
+    peer_id = _peer_id()
+    sys_cfg = _sys_cfg()
+    now = 1_000_000.0
+    entries = [
+        DynamicTgEntry(
+            int_id=730039101,
+            system_name="MASTER-A",
+            slot=2,
+            tgid=7304,
+            single_mode=False,
+            expires_at=None,
+            updated_at=now,
+        ),
+        DynamicTgEntry(
+            int_id=730039101,
+            system_name="MASTER-A",
+            slot=2,
+            tgid=7306,
+            single_mode=False,
+            expires_at=None,
+            updated_at=now,
+        ),
+    ]
+    restored = restore_peer_ua_entries_to_memory(sys_cfg, peer_id, entries, now=now)
+    assert sorted(restored) == [7304, 7306]
+    peer = {"OPTIONS": b"TS2=730,7305;SINGLE=0;"}
+    assert peer_should_receive_group_voice(
+        peer, 2, 7304, peer_id=peer_id, connected_count=2, sys_cfg=sys_cfg, now=now + 10
+    )
+    assert peer_should_receive_group_voice(
+        peer, 2, 7306, peer_id=peer_id, connected_count=2, sys_cfg=sys_cfg, now=now + 10
+    )
+
+
+def test_restore_single_preserves_absolute_expires_timestamp() -> None:
+    """SINGLE=1: reconnect restores the same wall-clock expiry, not a fresh TIMER."""
+    peer_id = _peer_id()
+    sys_cfg: dict = {}
+    activated_at = 1_700_000_000.0
+    expires_at = activated_at + 60.0 * 60.0  # TIMER=60 at activation
+    reconnect_at = activated_at + 34.0 * 60.0  # 26 minutes remaining
+    entries = [
+        DynamicTgEntry(
+            int_id=730039101,
+            system_name="MASTER-A",
+            slot=2,
+            tgid=7304,
+            single_mode=True,
+            expires_at=expires_at,
+            updated_at=activated_at,
+        ),
+    ]
+    restore_peer_ua_entries_to_memory(sys_cfg, peer_id, entries, now=reconnect_at)
+    pk_sessions = sys_cfg["_PEER_UA_SESSIONS"]
+    per_peer = next(iter(pk_sessions.values()))
+    assert per_peer[2]["expires"] == expires_at
+    assert per_peer[2]["expires"] - reconnect_at == pytest.approx(26.0 * 60.0, rel=0.01)
+
+
+def test_purge_expired_peer_ua_sessions() -> None:
+    peer_id = _peer_id()
+    sys_cfg = _sys_cfg()
+    now = 1_000_000.0
+    entries = [
+        DynamicTgEntry(
+            int_id=730039101,
+            system_name="MASTER-A",
+            slot=2,
+            tgid=7304,
+            single_mode=True,
+            expires_at=now + 60.0,
+            updated_at=now,
+        ),
+    ]
+    restore_peer_ua_entries_to_memory(sys_cfg, peer_id, entries, now=now)
+    purge_expired_peer_ua_sessions(sys_cfg, now=now + 120)
+    peer = {"OPTIONS": b"TS2=730,7305;SINGLE=1;TIMER=5;"}
+    assert not peer_should_receive_group_voice(
+        peer, 2, 7304, peer_id=peer_id, connected_count=2, sys_cfg=sys_cfg, now=now + 130
+    )

tests/infrastructure/test_dynamic_tg_repository.py

@@ -0,0 +1,96 @@
+# ADN DMR Peer Server - tests infrastructure dynamic tg repository
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+"""DynamicTgStore SQL operations (mock pool)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from twisted.internet.defer import succeed
+
+from adn_server.domain.dynamic_tg import DynamicTgEntry
+from adn_server.infrastructure.persistence.dynamic_tg_repository import MysqlDynamicTgRepository
+
+
+def _entry(**kwargs) -> DynamicTgEntry:
+    defaults = dict(
+        int_id=730039101,
+        system_name="MASTER-A",
+        slot=2,
+        tgid=7305,
+        single_mode=True,
+        expires_at=9_999_999.0,
+        updated_at=1_000_000.0,
+    )
+    defaults.update(kwargs)
+    return DynamicTgEntry(**defaults)
+
+
+@pytest.fixture
+def repo() -> MysqlDynamicTgRepository:
+    pool = MagicMock()
+    pool.runOperation.return_value = succeed(None)
+    pool.runQuery.return_value = succeed([])
+    return MysqlDynamicTgRepository(pool)
+
+
+def test_upsert_issues_insert(repo: MysqlDynamicTgRepository) -> None:
+    repo.upsert(_entry())
+    sql, args = repo._pool.runOperation.call_args[0]  # noqa: SLF001
+    assert "INSERT INTO peer_dynamic_tgs" in sql
+    assert args[0] == 730039101
+    assert args[3] == 7305
+
+
+def test_replace_single_slot_deletes_then_upserts(repo: MysqlDynamicTgRepository) -> None:
+    repo.replace_single_slot(_entry())
+    delete_sql = repo._pool.runOperation.call_args_list[0][0][0]  # noqa: SLF001
+    assert "DELETE FROM peer_dynamic_tgs" in delete_sql
+
+
+def test_delete_peer_slot(repo: MysqlDynamicTgRepository) -> None:
+    repo.delete_peer_slot(730039101, "MASTER-A", 2)
+    sql, args = repo._pool.runOperation.call_args[0]  # noqa: SLF001
+    assert "DELETE FROM peer_dynamic_tgs" in sql
+    assert args == (730039101, "MASTER-A", 2)
+
+
+def test_purge_expired(repo: MysqlDynamicTgRepository) -> None:
+    repo.purge_expired(1_000_000.0)
+    sql, args = repo._pool.runOperation.call_args[0]  # noqa: SLF001
+    assert "expires_at <= %s" in sql
+    assert args == (1_000_000,)
+
+
+def test_load_peer_maps_rows(repo: MysqlDynamicTgRepository) -> None:
+    repo._pool.runQuery.return_value = succeed(  # noqa: SLF001
+        [(730039101, "MASTER-A", 2, 7304, 0, None, 1_000_000.0)]
+    )
+    result: list = []
+    d = repo.load_peer(730039101, "MASTER-A")
+    d.addCallback(result.append)
+    from twisted.internet import reactor
+
+    reactor.runUntilCurrent()
+    assert len(result) == 1
+    assert result[0][0].tgid == 7304
+    assert result[0][0].single_mode is False

tests/infrastructure/test_echo_cli.py

@@ -70,12 +70,22 @@ def test_no_proxy_skips_proxy_startup(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(peer_mod, "RecordingHandler", lambda *_: None)
     monkeypatch.setattr(peer_mod, "IdentUseCases", lambda *_: _FakeIdent())
     monkeypatch.setattr(peer_mod, "HBPProtocolFactory", lambda *a, **k: object())
+    monkeypatch.setattr(peer_mod, "verify_database_sync", lambda *_a, **_k: True)
+    monkeypatch.setattr(peer_mod, "create_mysql_pool", lambda *_a, **_k: object())
+    monkeypatch.setattr(peer_mod, "MysqlDynamicTgRepository", lambda *_a, **_k: object())
 
     config = {
         "GLOBAL": {"SERVER_ID": 1},
         "LOGGER": {},
         "ALIASES": {"PATH": "."},
         "REPORTS": {"REPORT": False},
+        "DATABASE": {
+            "DB_SERVER": "localhost",
+            "DB_USERNAME": "hbmon",
+            "DB_PASSWORD": "x",
+            "DB_NAME": "hbmon",
+            "DB_PORT": 3306,
+        },
         "PROXY": {"LISTEN_PORT": 62031, "TARGET_SYSTEM": "SYSTEM"},
         "SYSTEMS": {
             "SYSTEM": {"MODE": "MASTER", "ENABLED": True, "IP": "127.0.0.1", "PORT": 62030, "MAX_PEERS": 4},

tests/infrastructure/test_peer_disconnect_report.py

@@ -55,7 +55,7 @@ def _master_protocol(*, report: MagicMock | None = None) -> HBPProtocol:
     return proto
 
 
-def test_disconnect_clears_internal_sessions_without_bridge_event() -> None:
+def test_disconnect_keeps_sys_cfg_sessions() -> None:
     report = MagicMock()
     proto = _master_protocol(report=report)
     peer_id = bytes_4(730039101)
@@ -68,7 +68,7 @@ def test_disconnect_clears_internal_sessions_without_bridge_event() -> None:
     proto._on_peer_disconnected(peer_id)
 
     report.send_routing_event.assert_not_called()
-    assert export_peer_ua_sessions(sys_cfg, peer_id, now=1_000_100.0) == {}
+    assert export_peer_ua_sessions(sys_cfg, peer_id, now=1_000_100.0)["2"]["tgid"] == 7305
 
 
 def test_export_peer_ua_sessions_omits_expired() -> None:

tests/infrastructure/test_proxy_self_service.py

@@ -114,18 +114,20 @@ def _run_deferred(d: Deferred) -> None:
     assert results, "deferred did not fire synchronously"
 
 
-def test_self_service_settings_reads_monitor_keys() -> None:
+def test_self_service_settings_reads_database_and_self_service_keys() -> None:
     cfg = {
-        "SELF_SERVICE": {
-            "USE_SELFSERVICE": True,
+        "DATABASE": {
             "DB_SERVER": "localhost",
             "DB_USERNAME": "hbmon",
             "DB_PASSWORD": "secret",
             "DB_NAME": "hbmon",
             "DB_PORT": 3306,
+        },
+        "SELF_SERVICE": {
+            "USE_SELFSERVICE": True,
             "PBKDF2_SALT": "ADN",
             "PBKDF2_ITERATIONS": 2000,
-        }
+        },
     }
     ss = self_service_settings(cfg)
     assert ss["enabled"] is True
@@ -231,9 +233,14 @@ GLOBAL: {}
 PROXY:
   LISTEN_PORT: 62031
   TARGET_SYSTEM: SYSTEM
+DATABASE:
+  DB_SERVER: localhost
+  DB_USERNAME: hbmon
+  DB_PASSWORD: secret
+  DB_NAME: hbmon
+  DB_PORT: 3306
 SELF_SERVICE:
   USE_SELFSERVICE: true
-  DB_SERVER: localhost
 SYSTEMS:
   SYSTEM:
     MODE: MASTER