solpbc.org
linux observer
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Configuration loading and persistence for solstone-linux.
5
6Config lives at ~/.local/share/solstone-linux/config/config.json.
7Captures go to ~/.local/share/solstone-linux/captures/.
8Screencast restore token at ~/.local/share/solstone-linux/config/restore_token.
9"""
10
11from __future__ import annotations
12
13import json
14import logging
15import os
16import stat
17from dataclasses import dataclass, field
18from pathlib import Path
19
20logger = logging.getLogger(__name__)
21
22DEFAULT_BASE_DIR = Path.home() / ".local" / "share" / "solstone-linux"
23DEFAULT_SEGMENT_INTERVAL = 300
24DEFAULT_SYNC_RETRY_DELAYS = [5, 30, 120, 300]
25DEFAULT_SYNC_MAX_RETRIES = 10
26
27
28@dataclass
29class Config:
30 """Configuration for the Linux desktop observer."""
31
32 server_url: str = ""
33 key: str = ""
34 stream: str = ""
35 segment_interval: int = DEFAULT_SEGMENT_INTERVAL
36 sync_retry_delays: list[int] = field(default_factory=lambda: list(DEFAULT_SYNC_RETRY_DELAYS))
37 sync_max_retries: int = DEFAULT_SYNC_MAX_RETRIES
38 base_dir: Path = DEFAULT_BASE_DIR
39
40 @property
41 def captures_dir(self) -> Path:
42 return self.base_dir / "captures"
43
44 @property
45 def config_dir(self) -> Path:
46 return self.base_dir / "config"
47
48 @property
49 def state_dir(self) -> Path:
50 return self.base_dir / "state"
51
52 @property
53 def config_path(self) -> Path:
54 return self.config_dir / "config.json"
55
56 @property
57 def restore_token_path(self) -> Path:
58 return self.config_dir / "restore_token"
59
60 def ensure_dirs(self) -> None:
61 """Create all required directories."""
62 self.captures_dir.mkdir(parents=True, exist_ok=True)
63 self.config_dir.mkdir(parents=True, exist_ok=True)
64 self.state_dir.mkdir(parents=True, exist_ok=True)
65
66
67def load_config(base_dir: Path | None = None) -> Config:
68 """Load config from disk, returning defaults if not found."""
69 config = Config()
70 if base_dir:
71 config.base_dir = base_dir
72
73 config_path = config.config_path
74 if not config_path.exists():
75 return config
76
77 try:
78 with open(config_path, encoding="utf-8") as f:
79 data = json.load(f)
80 except (json.JSONDecodeError, OSError) as e:
81 logger.warning(f"Failed to load config from {config_path}: {e}")
82 return config
83
84 config.server_url = data.get("server_url", "")
85 config.key = data.get("key", "")
86 config.stream = data.get("stream", "")
87 config.segment_interval = data.get("segment_interval", DEFAULT_SEGMENT_INTERVAL)
88 if "sync_retry_delays" in data:
89 config.sync_retry_delays = data["sync_retry_delays"]
90 if "sync_max_retries" in data:
91 config.sync_max_retries = data["sync_max_retries"]
92
93 return config
94
95
96def save_config(config: Config) -> None:
97 """Save config to disk with user-only permissions."""
98 config.ensure_dirs()
99
100 data = {
101 "server_url": config.server_url,
102 "key": config.key,
103 "stream": config.stream,
104 "segment_interval": config.segment_interval,
105 "sync_retry_delays": config.sync_retry_delays,
106 "sync_max_retries": config.sync_max_retries,
107 }
108
109 config_path = config.config_path
110 tmp_path = config_path.with_suffix(f".{os.getpid()}.tmp")
111
112 with open(tmp_path, "w", encoding="utf-8") as f:
113 json.dump(data, f, indent=2)
114 f.write("\n")
115
116 # Set user-only read/write before moving into place
117 os.chmod(tmp_path, stat.S_IRUSR | stat.S_IWUSR)
118 os.rename(str(tmp_path), str(config_path))
119 logger.info(f"Config saved to {config_path}")