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(
37 default_factory=lambda: list(DEFAULT_SYNC_RETRY_DELAYS)
38 )
39 sync_max_retries: int = DEFAULT_SYNC_MAX_RETRIES
40 cache_retention_days: int = 7
41 base_dir: Path = DEFAULT_BASE_DIR
42
43 @property
44 def captures_dir(self) -> Path:
45 return self.base_dir / "captures"
46
47 @property
48 def config_dir(self) -> Path:
49 return self.base_dir / "config"
50
51 @property
52 def state_dir(self) -> Path:
53 return self.base_dir / "state"
54
55 @property
56 def config_path(self) -> Path:
57 return self.config_dir / "config.json"
58
59 @property
60 def restore_token_path(self) -> Path:
61 return self.config_dir / "restore_token"
62
63 def ensure_dirs(self) -> None:
64 """Create all required directories."""
65 self.captures_dir.mkdir(parents=True, exist_ok=True)
66 self.config_dir.mkdir(parents=True, exist_ok=True)
67 self.state_dir.mkdir(parents=True, exist_ok=True)
68
69
70def load_config(base_dir: Path | None = None) -> Config:
71 """Load config from disk, returning defaults if not found."""
72 config = Config()
73 if base_dir:
74 config.base_dir = base_dir
75
76 config_path = config.config_path
77 if not config_path.exists():
78 return config
79
80 try:
81 with open(config_path, encoding="utf-8") as f:
82 data = json.load(f)
83 except (json.JSONDecodeError, OSError) as e:
84 logger.warning(f"Failed to load config from {config_path}: {e}")
85 return config
86
87 config.server_url = data.get("server_url", "")
88 config.key = data.get("key", "")
89 config.stream = data.get("stream", "")
90 config.segment_interval = data.get("segment_interval", DEFAULT_SEGMENT_INTERVAL)
91 if "sync_retry_delays" in data:
92 config.sync_retry_delays = data["sync_retry_delays"]
93 if "sync_max_retries" in data:
94 config.sync_max_retries = data["sync_max_retries"]
95 try:
96 config.cache_retention_days = int(data.get("cache_retention_days", 7))
97 except (TypeError, ValueError):
98 config.cache_retention_days = 7
99
100 return config
101
102
103def save_config(config: Config) -> None:
104 """Save config to disk with user-only permissions."""
105 config.ensure_dirs()
106
107 data = {
108 "server_url": config.server_url,
109 "key": config.key,
110 "stream": config.stream,
111 "segment_interval": config.segment_interval,
112 "sync_retry_delays": config.sync_retry_delays,
113 "sync_max_retries": config.sync_max_retries,
114 "cache_retention_days": config.cache_retention_days,
115 }
116
117 config_path = config.config_path
118 tmp_path = config_path.with_suffix(f".{os.getpid()}.tmp")
119
120 with open(tmp_path, "w", encoding="utf-8") as f:
121 json.dump(data, f, indent=2)
122 f.write("\n")
123
124 # Set user-only read/write before moving into place
125 os.chmod(tmp_path, stat.S_IRUSR | stat.S_IWUSR)
126 os.rename(str(tmp_path), str(config_path))
127 logger.info(f"Config saved to {config_path}")