daemon.py (27712B)
1 """ 2 Obsidian Sync daemon - manages vault sync as a system service. 3 4 This module handles: 5 - File system watching (using inotify / polling) 6 - Local file state tracking (path, mtime, ctime, size, hash) 7 - Bidirectional sync with the Obsidian Sync server 8 - Conflict detection and resolution 9 - Automatic reconnect with exponential backoff 10 11 Usage: 12 daemon = SyncDaemon(config) 13 asyncio.run(daemon.run()) 14 """ 15 16 from __future__ import annotations 17 18 import asyncio 19 import hashlib 20 import json 21 import logging 22 import os 23 import stat 24 import time 25 from dataclasses import dataclass, field, asdict 26 from pathlib import Path 27 from typing import Optional, Callable 28 29 from .api import list_vaults, Vault 30 from .crypto import create_encryption_provider, EncryptionV0, EncryptionV2 31 from .websocket_client import ObsidianSyncClient, SyncConnectionError, SyncAuthError 32 33 logger = logging.getLogger(__name__) 34 35 # File types allowed by default (matching Obsidian defaults) 36 DEFAULT_ALLOWED_TYPES = {"md", "canvas", "base"} 37 38 # Image/audio/video/pdf types (user-configurable) 39 IMAGE_TYPES = { 40 "avif", "bmp", "gif", "jpeg", "jpg", "png", "svg", "webp", 41 "tiff", "tif", "ico", 42 } 43 AUDIO_TYPES = {"flac", "m4a", "mp3", "ogg", "wav", "webm", "3gp"} 44 VIDEO_TYPES = {"mkv", "mov", "mp4", "ogv", "webm"} 45 PDF_TYPES = {"pdf"} 46 47 48 def compute_file_hash(data: bytes) -> str: 49 """Compute SHA-256 hash of file content as hex string.""" 50 return hashlib.sha256(data).hexdigest() 51 52 53 @dataclass 54 class LocalFileInfo: 55 path: str 56 mtime: int = 0 57 ctime: int = 0 58 size: int = 0 59 folder: bool = False 60 hash: str = "" 61 sync_time: int = 0 62 sync_hash: str = "" 63 previous_path: str = "" 64 65 66 @dataclass 67 class ServerFileInfo: 68 path: str 69 uid: int = 0 70 mtime: int = 0 71 ctime: int = 0 72 size: int = 0 73 folder: bool = False 74 hash: str = "" 75 deleted: bool = False 76 device: str = "" 77 user: int = 0 78 79 80 @dataclass 81 class SyncConfig: 82 """Configuration for the sync daemon.""" 83 # Authentication 84 token: str 85 86 # Vault settings 87 vault_id: str 88 vault_host: str 89 vault_password: str 90 vault_salt: str 91 encryption_version: int 92 93 # Local vault path 94 vault_path: str 95 96 # Device identification 97 device_name: str = "obsidian-sync-daemon" 98 99 # Sync options 100 allowed_types: set = field(default_factory=lambda: set(DEFAULT_ALLOWED_TYPES)) 101 sync_images: bool = False 102 sync_audio: bool = False 103 sync_video: bool = False 104 sync_pdf: bool = False 105 sync_app_settings: bool = True 106 107 # Ignored folders (list of path prefixes) 108 ignore_folders: list = field(default_factory=list) 109 110 # State file path (for persisting sync state) 111 state_file: str = "" 112 113 @property 114 def effective_allowed_types(self) -> set: 115 types = set(DEFAULT_ALLOWED_TYPES) 116 if self.sync_images: 117 types |= IMAGE_TYPES 118 if self.sync_audio: 119 types |= AUDIO_TYPES 120 if self.sync_video: 121 types |= VIDEO_TYPES 122 if self.sync_pdf: 123 types |= PDF_TYPES 124 return types 125 126 @classmethod 127 def from_file(cls, path: str) -> "SyncConfig": 128 """Load config from a JSON file.""" 129 with open(path) as f: 130 data = json.load(f) 131 config = cls( 132 token=data["token"], 133 vault_id=data["vault_id"], 134 vault_host=data["vault_host"], 135 vault_password=data["vault_password"], 136 vault_salt=data["vault_salt"], 137 encryption_version=data.get("encryption_version", 2), 138 vault_path=data["vault_path"], 139 device_name=data.get("device_name", "obsidian-sync-daemon"), 140 sync_images=data.get("sync_images", False), 141 sync_audio=data.get("sync_audio", False), 142 sync_video=data.get("sync_video", False), 143 sync_pdf=data.get("sync_pdf", False), 144 sync_app_settings=data.get("sync_app_settings", True), 145 ignore_folders=data.get("ignore_folders", []), 146 state_file=data.get("state_file", ""), 147 ) 148 return config 149 150 def to_file(self, path: str) -> None: 151 """Save config to a JSON file.""" 152 data = { 153 "token": self.token, 154 "vault_id": self.vault_id, 155 "vault_host": self.vault_host, 156 "vault_password": self.vault_password, 157 "vault_salt": self.vault_salt, 158 "encryption_version": self.encryption_version, 159 "vault_path": self.vault_path, 160 "device_name": self.device_name, 161 "sync_images": self.sync_images, 162 "sync_audio": self.sync_audio, 163 "sync_video": self.sync_video, 164 "sync_pdf": self.sync_pdf, 165 "sync_app_settings": self.sync_app_settings, 166 "ignore_folders": self.ignore_folders, 167 "state_file": self.state_file, 168 } 169 with open(path, "w") as f: 170 json.dump(data, f, indent=2) 171 172 173 @dataclass 174 class SyncState: 175 """Persisted sync state.""" 176 version: int = 0 177 initial: bool = True 178 local_files: dict = field(default_factory=dict) 179 server_files: dict = field(default_factory=dict) 180 181 def save(self, path: str) -> None: 182 data = { 183 "version": self.version, 184 "initial": self.initial, 185 "local_files": self.local_files, 186 "server_files": self.server_files, 187 } 188 with open(path, "w") as f: 189 json.dump(data, f) 190 191 @classmethod 192 def load(cls, path: str) -> "SyncState": 193 if not os.path.exists(path): 194 return cls() 195 try: 196 with open(path) as f: 197 data = json.load(f) 198 return cls( 199 version=data.get("version", 0), 200 initial=data.get("initial", True), 201 local_files=data.get("local_files", {}), 202 server_files=data.get("server_files", {}), 203 ) 204 except Exception: 205 return cls() 206 207 208 class BackoffTimer: 209 """Exponential backoff timer for reconnection.""" 210 211 def __init__(self, base: float = 5.0, max_delay: float = 300.0): 212 self.base = base 213 self.max_delay = max_delay 214 self._count = 0 215 216 def fail(self) -> None: 217 self._count += 1 218 219 def success(self) -> None: 220 self._count = 0 221 222 def is_ready(self) -> bool: 223 return True # We handle sleep in the main loop 224 225 async def wait(self) -> None: 226 if self._count > 0: 227 delay = min(self.base * (2 ** self._count), self.max_delay) 228 logger.info(f"Waiting {delay:.0f}s before reconnect (attempt {self._count})") 229 await asyncio.sleep(delay) 230 231 232 class SyncDaemon: 233 """ 234 Main sync daemon that manages bidirectional sync with Obsidian Sync. 235 236 Designed to run as a systemd service or background process. 237 """ 238 239 def __init__(self, config: SyncConfig): 240 self.config = config 241 self.vault_path = Path(config.vault_path) 242 self.state_file = config.state_file or str(self.vault_path / ".obsidian" / "sync-state.json") 243 244 self.state = SyncState.load(self.state_file) 245 self.backoff = BackoffTimer() 246 self._client: Optional[ObsidianSyncClient] = None 247 self._running = False 248 self._pending_server_files: list[dict] = [] 249 self._sync_lock = asyncio.Lock() 250 251 # Setup encryption provider 252 self._enc: EncryptionV0 | EncryptionV2 = create_encryption_provider( 253 config.vault_password, 254 config.vault_salt, 255 config.encryption_version, 256 ) 257 258 # ------------------------------------------------------------------ 259 # File system helpers 260 # ------------------------------------------------------------------ 261 262 def _should_sync(self, path: str, is_folder: bool = False) -> bool: 263 """Check if a file/folder should be synced. Matches _allowSyncFile() in app.js.""" 264 if path == "/": 265 return False 266 267 # Check ignored folders 268 for folder in self.config.ignore_folders: 269 if (is_folder and path == folder) or path.startswith(folder + "/"): 270 return False 271 272 # Skip hidden files and special dirs 273 parts = path.split("/") 274 if any(p.startswith(".") or p == "node_modules" for p in parts[:-1]): 275 return False 276 277 if is_folder: 278 return True 279 280 # Skip files starting with '.' 281 filename = parts[-1] 282 if filename.startswith("."): 283 return False 284 285 # Check config directory files 286 config_dir = ".obsidian" 287 if path.startswith(config_dir + "/"): 288 if not self.config.sync_app_settings: 289 return False 290 relative = path[len(config_dir) + 1:] 291 # Skip workspace files 292 if relative in ("workspace.json", "workspace-mobile.json"): 293 return False 294 return True 295 296 # Regular files: check extension 297 ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" 298 return ext in self.config.effective_allowed_types 299 300 async def _scan_local_files(self) -> dict[str, LocalFileInfo]: 301 """Scan local vault for all syncable files.""" 302 files: dict[str, LocalFileInfo] = {} 303 304 def scan_dir(dir_path: Path, rel_prefix: str = "") -> None: 305 try: 306 entries = list(dir_path.iterdir()) 307 except PermissionError: 308 return 309 310 for entry in sorted(entries): 311 rel = (rel_prefix + "/" + entry.name).lstrip("/") if rel_prefix else entry.name 312 313 if entry.name.startswith(".") and rel_prefix == "": 314 # Allow .obsidian but skip other hidden top-level dirs 315 if entry.name != ".obsidian": 316 continue 317 318 if entry.is_dir(follow_symlinks=False): 319 if self._should_sync(rel, is_folder=True): 320 files[rel] = LocalFileInfo( 321 path=rel, 322 folder=True, 323 mtime=0, 324 ctime=0, 325 size=0, 326 ) 327 scan_dir(entry, rel) 328 elif entry.is_file(follow_symlinks=False): 329 if self._should_sync(rel, is_folder=False): 330 try: 331 st = entry.stat() 332 info = files.get(rel) or LocalFileInfo(path=rel) 333 mtime = int(st.st_mtime * 1000) 334 ctime = int(st.st_ctime * 1000) 335 size = st.st_size 336 # Check if file changed 337 if info.mtime != mtime or info.size != size: 338 info.hash = "" # Will be computed on upload 339 info.mtime = mtime 340 info.ctime = ctime 341 info.size = size 342 info.folder = False 343 files[rel] = info 344 except (OSError, PermissionError): 345 pass 346 347 # Restore existing local file info 348 for path, data in self.state.local_files.items(): 349 if isinstance(data, dict): 350 files[path] = LocalFileInfo(**data) 351 352 # Scan actual filesystem 353 await asyncio.get_event_loop().run_in_executor(None, scan_dir, self.vault_path) 354 return files 355 356 def _get_file_ext(self, path: str) -> str: 357 filename = path.split("/")[-1] 358 if "." in filename: 359 return filename.rsplit(".", 1)[-1].lower() 360 return "" 361 362 async def _read_file(self, path: str) -> bytes: 363 """Read a local file.""" 364 full_path = self.vault_path / path 365 loop = asyncio.get_event_loop() 366 367 def _read(): 368 with open(full_path, "rb") as f: 369 return f.read() 370 371 return await loop.run_in_executor(None, _read) 372 373 async def _write_file(self, path: str, data: bytes, mtime: int = 0, ctime: int = 0) -> None: 374 """Write a local file, creating parent directories as needed.""" 375 full_path = self.vault_path / path 376 loop = asyncio.get_event_loop() 377 378 def _write(): 379 full_path.parent.mkdir(parents=True, exist_ok=True) 380 with open(full_path, "wb") as f: 381 f.write(data) 382 if mtime: 383 atime = time.time() 384 mtime_s = mtime / 1000.0 385 os.utime(full_path, (atime, mtime_s)) 386 387 await loop.run_in_executor(None, _write) 388 389 async def _delete_file(self, path: str) -> None: 390 """Delete a local file.""" 391 full_path = self.vault_path / path 392 loop = asyncio.get_event_loop() 393 394 def _delete(): 395 if full_path.is_file(): 396 full_path.unlink() 397 elif full_path.is_dir(): 398 import shutil 399 shutil.rmtree(full_path, ignore_errors=True) 400 401 await loop.run_in_executor(None, _delete) 402 403 async def _create_dir(self, path: str) -> None: 404 """Create a local directory.""" 405 full_path = self.vault_path / path 406 await asyncio.get_event_loop().run_in_executor( 407 None, lambda: full_path.mkdir(parents=True, exist_ok=True) 408 ) 409 410 # ------------------------------------------------------------------ 411 # Sync logic 412 # ------------------------------------------------------------------ 413 414 async def _on_server_push(self, msg: dict) -> None: 415 """Callback for server-pushed file updates.""" 416 self.state.version = msg.get("uid", self.state.version) 417 file_info = { 418 "path": msg.get("path", ""), 419 "uid": msg.get("uid", 0), 420 "size": msg.get("size", 0), 421 "hash": msg.get("hash", ""), 422 "mtime": msg.get("mtime", 0), 423 "ctime": msg.get("ctime", 0), 424 "folder": msg.get("folder", False), 425 "deleted": msg.get("deleted", False), 426 "device": msg.get("device", ""), 427 "user": msg.get("user", 0), 428 } 429 logger.debug(f"Server push: {file_info['path']} (deleted={file_info['deleted']})") 430 self._pending_server_files.append(file_info) 431 432 async def _on_ready(self, version: int) -> None: 433 """Called when server sends 'ready' after initial handshake.""" 434 if version > self.state.version: 435 self.state.version = version 436 self.state.initial = False 437 logger.info(f"Server ready. Version: {version}") 438 self._save_state() 439 440 async def _sync_iteration(self) -> bool: 441 """ 442 Perform one sync iteration. Returns True if more work to do. 443 444 Process order: 445 1. Apply pending server-pushed files 446 2. Delete local files that server removed 447 3. Upload new/changed local files 448 """ 449 if not self._client or not self._client.is_connected(): 450 return False 451 452 local_files = await self._scan_local_files() 453 server_files = {k: v for k, v in self.state.server_files.items()} 454 455 # -- Step 1: Apply pending server pushes -- 456 while self._pending_server_files: 457 pushed = self._pending_server_files.pop(0) 458 path = pushed["path"] 459 460 if not path or not self._should_sync(path, pushed.get("folder", False)): 461 continue 462 463 uid = pushed.get("uid", 0) 464 if uid: 465 self.state.version = max(self.state.version, uid) 466 467 server_files[path] = pushed 468 self.state.server_files[path] = pushed 469 470 local = local_files.get(path) 471 472 if pushed.get("deleted"): 473 if local: 474 logger.info(f"Deleting local file (server deleted): {path}") 475 await self._delete_file(path) 476 del local_files[path] 477 self._save_state() 478 continue 479 480 if pushed.get("folder"): 481 if not (self.vault_path / path).is_dir(): 482 logger.info(f"Creating local folder: {path}") 483 await self._create_dir(path) 484 continue 485 486 # Download file 487 if not local or local.hash != pushed.get("hash", ""): 488 logger.info(f"Downloading: {path}") 489 try: 490 data = await self._client.pull(uid) 491 if data is not None: 492 await self._write_file( 493 path, data, 494 mtime=pushed.get("mtime", 0), 495 ctime=pushed.get("ctime", 0), 496 ) 497 file_hash = compute_file_hash(data) 498 info = local_files.get(path, LocalFileInfo(path=path)) 499 info.hash = file_hash 500 info.sync_hash = file_hash 501 info.sync_time = int(time.time() * 1000) 502 local_files[path] = info 503 self._save_state_with_files(local_files, server_files) 504 except Exception as e: 505 logger.error(f"Failed to download {path}: {e}") 506 507 # -- Step 2: Find local files that server removed -- 508 for path, server_info in list(server_files.items()): 509 if server_info.get("deleted"): 510 continue 511 if path not in local_files: 512 logger.info(f"Deleting remote (no longer exists locally): {path}") 513 try: 514 folder = server_info.get("folder", False) 515 await self._client.push(path, None, folder=folder, deleted=True) 516 server_info["deleted"] = True 517 self.state.server_files[path] = server_info 518 self._save_state() 519 return True 520 except Exception as e: 521 logger.error(f"Failed to delete remote {path}: {e}") 522 523 # -- Step 3: Upload new or changed local files -- 524 # Find the best candidate to upload (smallest changed file first) 525 best: Optional[tuple[str, LocalFileInfo]] = None 526 527 for path, local in local_files.items(): 528 if not self._should_sync(path, local.folder): 529 continue 530 531 server = server_files.get(path) 532 needs_upload = False 533 534 if server is None or server.get("deleted"): 535 needs_upload = True 536 elif local.folder != server.get("folder", False): 537 needs_upload = True 538 elif not local.folder: 539 if not local.hash: 540 # Compute hash 541 try: 542 data = await self._read_file(path) 543 local.hash = compute_file_hash(data) 544 local_files[path] = local 545 except Exception: 546 continue 547 if local.hash != server.get("hash", ""): 548 needs_upload = True 549 550 if needs_upload: 551 if best is None or (not local.folder and local.size < (best[1].size if best else 0)): 552 best = (path, local) 553 554 if best: 555 path, local = best 556 logger.info(f"Uploading: {path} (folder={local.folder})") 557 try: 558 if local.folder: 559 await self._client.push(path, None, folder=True) 560 sv = server_files.get(path, {}) 561 sv.update({"path": path, "folder": True, "deleted": False, "uid": 0, 562 "size": 0, "hash": "", "ctime": 0, "mtime": 0}) 563 server_files[path] = sv 564 self.state.server_files[path] = sv 565 else: 566 data = await self._read_file(path) 567 local.hash = compute_file_hash(data) 568 await self._client.push( 569 path, data, 570 related_path=local.previous_path or None, 571 ctime=local.ctime, 572 mtime=local.mtime, 573 file_hash=local.hash, 574 ) 575 local.sync_hash = local.hash 576 local.sync_time = int(time.time() * 1000) 577 local.previous_path = "" 578 local_files[path] = local 579 sv = server_files.get(path, {}) 580 sv.update({"path": path, "folder": False, "deleted": False, 581 "hash": local.hash, "size": local.size, 582 "mtime": local.mtime, "ctime": local.ctime}) 583 server_files[path] = sv 584 self.state.server_files[path] = sv 585 586 self._save_state_with_files(local_files, server_files) 587 return True 588 except Exception as e: 589 logger.error(f"Failed to upload {path}: {e}") 590 591 # All synced 592 logger.debug("Fully synced") 593 self._save_state_with_files(local_files, server_files) 594 return False 595 596 def _save_state(self) -> None: 597 """Save sync state to disk.""" 598 try: 599 os.makedirs(os.path.dirname(self.state_file), exist_ok=True) 600 self.state.save(self.state_file) 601 except Exception as e: 602 logger.warning(f"Failed to save state: {e}") 603 604 def _save_state_with_files( 605 self, local_files: dict, server_files: dict 606 ) -> None: 607 self.state.local_files = { 608 k: asdict(v) if isinstance(v, LocalFileInfo) else v 609 for k, v in local_files.items() 610 } 611 self.state.server_files = server_files 612 self._save_state() 613 614 # ------------------------------------------------------------------ 615 # File watching (simple polling) 616 # ------------------------------------------------------------------ 617 618 async def _watch_files(self) -> None: 619 """ 620 Watch for local file changes using polling. 621 Triggers sync when changes are detected. 622 623 For production use, replace with inotify-based watcher. 624 """ 625 last_scan: dict[str, tuple[int, int]] = {} # path -> (mtime, size) 626 627 while self._running: 628 await asyncio.sleep(5) # Poll every 5 seconds 629 630 try: 631 changed = False 632 for root, dirs, files in os.walk(self.vault_path): 633 # Skip hidden directories (except .obsidian) 634 dirs[:] = [d for d in dirs if not d.startswith(".") or d == ".obsidian"] 635 636 for fname in files: 637 full = os.path.join(root, fname) 638 rel = os.path.relpath(full, self.vault_path) 639 640 if not self._should_sync(rel): 641 continue 642 643 try: 644 st = os.stat(full) 645 key = (int(st.st_mtime * 1000), st.st_size) 646 if last_scan.get(rel) != key: 647 last_scan[rel] = key 648 changed = True 649 except OSError: 650 pass 651 652 if changed: 653 logger.debug("Local changes detected, triggering sync") 654 # Sync will happen in the main loop 655 except Exception as e: 656 logger.debug(f"File watch error: {e}") 657 658 # ------------------------------------------------------------------ 659 # Main run loop 660 # ------------------------------------------------------------------ 661 662 async def run(self) -> None: 663 """Main run loop. Connects, syncs, and reconnects on failure.""" 664 self._running = True 665 self.vault_path.mkdir(parents=True, exist_ok=True) 666 667 logger.info(f"Starting Obsidian sync daemon for vault at {self.vault_path}") 668 logger.info(f"Vault ID: {self.config.vault_id}, Host: {self.config.vault_host}") 669 logger.info(f"Encryption version: {self.config.encryption_version}") 670 671 # Start file watcher 672 watch_task = asyncio.create_task(self._watch_files()) 673 674 try: 675 while self._running: 676 await self.backoff.wait() 677 678 try: 679 logger.info("Connecting to sync server...") 680 self._client = ObsidianSyncClient( 681 host=self.config.vault_host, 682 token=self.config.token, 683 vault_id=self.config.vault_id, 684 encryption_provider=self._enc, 685 device_name=self.config.device_name, 686 last_version=self.state.version, 687 initial=self.state.initial, 688 ) 689 690 self._client.on_push(self._on_server_push) 691 self._client.on_ready(self._on_ready) 692 693 await self._client.connect() 694 self.backoff.success() 695 logger.info("Connected to sync server") 696 697 # Start heartbeat 698 heartbeat_task = asyncio.create_task(self._client.heartbeat_loop()) 699 700 try: 701 # Main sync loop 702 while self._running and self._client.is_connected(): 703 more_work = await self._sync_iteration() 704 if more_work: 705 await asyncio.sleep(0.05) # Small delay between iterations 706 else: 707 await asyncio.sleep(1.0) # Wait 1s when fully synced 708 finally: 709 heartbeat_task.cancel() 710 711 except SyncAuthError as e: 712 logger.error(f"Authentication error: {e}") 713 self.backoff.fail() 714 if "expired" in str(e).lower(): 715 logger.error("Subscription expired. Stopping.") 716 break 717 718 except SyncConnectionError as e: 719 logger.warning(f"Connection error: {e}") 720 self.backoff.fail() 721 722 except Exception as e: 723 logger.error(f"Sync error: {e}", exc_info=True) 724 self.backoff.fail() 725 726 finally: 727 if self._client: 728 await self._client.disconnect() 729 self._client = None 730 731 finally: 732 watch_task.cancel() 733 self._running = False 734 logger.info("Sync daemon stopped") 735 736 async def stop(self) -> None: 737 """Gracefully stop the daemon.""" 738 self._running = False 739 if self._client: 740 await self._client.disconnect()