# -*- coding: utf-8 -*- """ SyncStore — JSON-per-Entity Persistenz fuer synchronisierbare Daten. Speicherlayout: {base_directory}/{data_type}/{item_id}.json Folgt dem Scheduler-Pattern (./cron/*.json). """ import json import threading from pathlib import Path from typing import Any from trixy_core.sync.models import SyncItem, _now_iso from trixy_core.utils.debug import pdebug, perror class SyncStore: """ Persistenter Speicher fuer SyncItems. Speichert jedes Item als einzelne JSON-Datei. Thread-safe durch Lock. """ def __init__(self, base_directory: str = "./sync_data") -> None: """ Initialisiert den SyncStore. Args: base_directory: Basis-Verzeichnis fuer Sync-Daten """ self._base_dir = Path(base_directory) self._base_dir.mkdir(parents=True, exist_ok=True) self._lock = threading.Lock() @property def base_directory(self) -> Path: """Basis-Verzeichnis.""" return self._base_dir # ========================================================================= # CRUD # ========================================================================= def save(self, item: SyncItem) -> None: """ Speichert ein SyncItem als JSON-Datei. Args: item: Das zu speichernde Item """ with self._lock: item_dir = self._base_dir / item.data_type item_dir.mkdir(parents=True, exist_ok=True) file_path = item_dir / f"{item.item_id}.json" try: file_path.write_text( json.dumps(item.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8", ) except Exception as e: perror(f"SyncStore: Fehler beim Speichern von {item.item_id}: {e}") def get(self, data_type: str, item_id: str) -> SyncItem | None: """ Laedt ein einzelnes SyncItem. Args: data_type: Datentyp (z.B. "notes.note") item_id: Eindeutige ID Returns: SyncItem oder None wenn nicht gefunden """ file_path = self._base_dir / data_type / f"{item_id}.json" if not file_path.exists(): return None try: data = json.loads(file_path.read_text(encoding="utf-8")) return SyncItem.from_dict(data) except Exception as e: perror(f"SyncStore: Fehler beim Laden von {item_id}: {e}") return None def get_all(self, data_type: str) -> list[SyncItem]: """ Laedt alle SyncItems eines Datentyps (ohne Soft-Deleted). Args: data_type: Datentyp Returns: Liste der aktiven Items """ type_dir = self._base_dir / data_type if not type_dir.exists(): return [] items: list[SyncItem] = [] for file_path in type_dir.glob("*.json"): try: data = json.loads(file_path.read_text(encoding="utf-8")) item = SyncItem.from_dict(data) if not item.deleted: items.append(item) except Exception as e: perror(f"SyncStore: Fehler beim Laden von {file_path.name}: {e}") return items def delete(self, data_type: str, item_id: str) -> None: """ Soft-Delete eines Items (setzt deleted=True, aktualisiert updated_at). Args: data_type: Datentyp item_id: Item-ID """ item = self.get(data_type, item_id) if item is None: return item.deleted = True item.updated_at = _now_iso() item.version += 1 self.save(item) # ========================================================================= # Sync-spezifisch # ========================================================================= def get_changes_since(self, since: str) -> list[SyncItem]: """ Gibt alle Items zurueck, die seit 'since' geaendert wurden. Args: since: ISO 8601 Zeitpunkt Returns: Liste geaenderter Items (inkl. Soft-Deleted) """ results: list[SyncItem] = [] if not self._base_dir.exists(): return results for type_dir in self._base_dir.iterdir(): if not type_dir.is_dir() or type_dir.name.startswith("."): continue for file_path in type_dir.glob("*.json"): try: data = json.loads(file_path.read_text(encoding="utf-8")) item = SyncItem.from_dict(data) if item.updated_at > since: results.append(item) except Exception as e: perror(f"SyncStore: Fehler beim Lesen von {file_path}: {e}") return results def merge_incoming(self, item: SyncItem) -> bool: """ Merged ein eingehendes Item (Last-Write-Wins). - Lokal nicht vorhanden → speichern - incoming.updated_at > local.updated_at → ueberschreiben - Sonst → ignorieren Args: item: Das eingehende Item Returns: True wenn lokal aktualisiert wurde """ local = self.get(item.data_type, item.item_id) if local is None: # Lokal nicht vorhanden -> speichern self.save(item) pdebug(f"SyncStore: Neues Item gemerged: {item.item_id}") return True if item.updated_at > local.updated_at: # Eingehendes Item ist neuer -> ueberschreiben self.save(item) pdebug(f"SyncStore: Item aktualisiert: {item.item_id}") return True # Lokale Version ist neuer oder gleich -> ignorieren return False # ========================================================================= # Sync-Zeitpunkt # ========================================================================= def get_last_sync_time(self) -> str: """ Liest den letzten Sync-Zeitpunkt aus .last_sync Datei. Returns: ISO 8601 Zeitpunkt oder leerer String """ sync_file = self._base_dir / ".last_sync" if not sync_file.exists(): return "" try: return sync_file.read_text(encoding="utf-8").strip() except Exception: return "" def set_last_sync_time(self, ts: str) -> None: """ Speichert den letzten Sync-Zeitpunkt. Args: ts: ISO 8601 Zeitpunkt """ sync_file = self._base_dir / ".last_sync" try: sync_file.write_text(ts, encoding="utf-8") except Exception as e: perror(f"SyncStore: Fehler beim Schreiben von .last_sync: {e}")