bridget: use standard settings path and migrate userdata

core/services/bridget/bridget.py

@@ -1,5 +1,4 @@
 import logging
-from pathlib import Path
 from typing import Annotated, Dict, List
 
 import requests
@@ -10,8 +9,6 @@
 from serial.tools.list_ports_linux import SysFS
 from settings import BridgeSettingsSpecV2, SettingsV2
 
-USERDATA = Path("/usr/blueos/userdata/")
-
 
 class BridgeFrontendSpec(BaseModel):
     """Basic interface for 'bridges' links."""
@@ -46,9 +43,7 @@ class Bridget:
 
     def __init__(self) -> None:
         self._bridges: Dict[BridgeFrontendSpec, Bridge] = {}
-        # We use userdata because our regular settings folder is under /root, which regular users
-        # don't have access to.
-        self._settings_manager = PydanticManager("bridget", SettingsV2, USERDATA / "settings" / "bridget")
+        self._settings_manager = PydanticManager("bridget", SettingsV2)
         self._settings_manager.load()
         for bridge_settings_spec in self._settings_manager.settings.specsv2:
             try:

core/services/bridget/settings.py

@@ -1,8 +1,12 @@
+import json
+import pathlib
 from typing import Any, Dict, List
 
 from commonwealth.settings.settings import PydanticSettings
 from pydantic import BaseModel, Field
 
+OLD_SETTINGS_DIR = pathlib.Path("/usr/blueos/userdata/settings/bridget/bridget")
+
 
 class BridgeSettingsSpecV1(BaseModel):
     serial_path: str
@@ -25,19 +29,6 @@ def __eq__(self, other: object) -> Any:
         return False
 
 
-class SettingsV1(PydanticSettings):
-    specs: List[BridgeSettingsSpecV1] = Field(default_factory=list)
-
-    def migrate(self, data: Dict[str, Any]) -> None:
-        if data["VERSION"] == SettingsV1.STATIC_VERSION:
-            return
-
-        if data["VERSION"] < SettingsV1.STATIC_VERSION:
-            super().migrate(data)
-
-        data["VERSION"] = SettingsV1.STATIC_VERSION
-
-
 class BridgeSettingsSpecV2(BaseModel):
     udp_target_port: int
     udp_listen_port: int
@@ -61,6 +52,53 @@ def __eq__(self, other: object) -> Any:
         return False
 
 
+def migrate_from_old_settings(
+    version: int,
+    target_settings: "SettingsV1",
+    old_dir: pathlib.Path | None = None,
+) -> None:
+    old_settings_file_path = (old_dir or OLD_SETTINGS_DIR) / f"settings-{version}.json"
+    if not old_settings_file_path.exists():
+        return
+
+    try:
+        with open(old_settings_file_path, "r", encoding="utf-8") as file:
+            old_data = json.load(file)
+
+        if version == 1 and "specs" in old_data:
+            target_settings.specs = [BridgeSettingsSpecV1.model_validate(spec) for spec in old_data["specs"]]
+        elif version == 2 and "specsv2" in old_data and isinstance(target_settings, SettingsV2):
+            target_settings.specsv2 = [BridgeSettingsSpecV2.model_validate(spec) for spec in old_data["specsv2"]]
+    except Exception:
+        # If migration fails, continue with empty settings
+        pass
+
+
+class SettingsV1(PydanticSettings):
+    specs: List[BridgeSettingsSpecV1] = Field(default_factory=list)
+
+    def migrate(self, data: Dict[str, Any]) -> None:
+        if data["VERSION"] == SettingsV1.STATIC_VERSION:
+            return
+
+        if data["VERSION"] < SettingsV1.STATIC_VERSION:
+            super().migrate(data)
+
+        data["VERSION"] = SettingsV1.STATIC_VERSION
+
+    def on_settings_created(self, file_path: pathlib.Path) -> None:
+        if self.VERSION not in (
+            SettingsV1.STATIC_VERSION,
+            SettingsV1.STATIC_VERSION + 1,
+        ):
+            return
+
+        if (file_path.parent / "settings-1.json").exists():
+            return
+
+        migrate_from_old_settings(1, self)
+
+
 class SettingsV2(SettingsV1):
     specsv2: List[BridgeSettingsSpecV2] = Field(default_factory=list)
 
@@ -84,3 +122,15 @@ def migrate(self, data: Dict[str, Any]) -> None:
                         "udp_listen_port": spec["udp_port"] if server else 0,
                     }
                 )
+
+    def on_settings_created(self, file_path: pathlib.Path) -> None:
+        if self.VERSION not in (
+            SettingsV2.STATIC_VERSION,
+            SettingsV2.STATIC_VERSION + 1,
+        ):
+            return
+
+        if (file_path.parent / "settings-2.json").exists():
+            return
+
+        migrate_from_old_settings(2, self)