From 16630f0368aa581de13ce8f0509dbe3b9d26cd86 Mon Sep 17 00:00:00 2001 From: team3 Date: Mon, 5 Oct 2026 21:15:41 +0200 Subject: [PATCH] update --- README.md | 20 ++++--- backend/app/config.py | 4 +- backend/app/db.py | 20 ++----- backend/app/files.py | 111 +++++++++++++++++++++++++++++++++++++ backend/app/main.py | 43 ++++++++------ backend/tests/test_sync.py | 67 ++++++++++++++-------- desktop@marha.de/files.js | 52 +++++++++++++++-- desktop@marha.de/sync.js | 111 ++++++++++++++++++++++++++----------- dev/tests/run.js | 14 ++++- dev/tests/sync.sh | 49 ++++++++-------- 10 files changed, 364 insertions(+), 127 deletions(-) create mode 100644 backend/app/files.py diff --git a/README.md b/README.md index ebabbe2..e1bab27 100644 --- a/README.md +++ b/README.md @@ -63,13 +63,18 @@ Wiederverbinden wird nachgeliefert. Eine gerade offene Notiz wird nicht von auß ### Claude-Code-Sitzungen (nur PCs) -Die Extension gleicht auch `~/.claude/projects/` ab, also Sitzungen und Memory von Claude Code. -Alle 10 Minuten prüft sie den Ordner. Hoch geht eine Datei, wenn sie sich geändert hat und seit -einer Stunde nicht angefasst wurde. Vom Server kommt sie, wenn sie dort neuer ist, mit ihrer -Änderungszeit. Gelöschte Dateien verschwinden auch auf dem anderen PC. Die App fragt diese -Dateien nie an. Bedingung: gleicher Pfad auf beiden PCs, der Ordnername enthält ihn. +Die Extension gleicht auch `~/.claude/projects/` ab, also Sitzungen und Memory von Claude Code, +im selben Takt wie die Einträge. Hoch geht eine Datei, wenn sie sich geändert hat und seit einer +Minute nicht angefasst wurde. Sitzungen wachsen nur am Ende, deshalb geht meist nur der neue Rest: +der Client nennt Größe und Hash der letzten 64 KB des Standes, den der Server haben müsste, und der +Server hängt an. Passt das nicht, geht die ganze Datei. Vom Server kommt eine Datei, wenn sie dort +neuer ist, ebenfalls nur ab dem eigenen Stand, mit ihrer Änderungszeit. Gelöschte Dateien +verschwinden auch auf dem anderen PC. Die App fragt diese Dateien nie an. +Auf dem Server liegen die Inhalte unter `data/files/`, in SQLite nur Pfad, Zeitpunkt und Größe. +Eine Sitzung darf nur auf einem PC gleichzeitig laufen, sonst gewinnt die neuere Datei als Ganzes. +Bedingung: gleicher Pfad auf beiden PCs, der Ordnername enthält ihn. `"files": false` in `sync.json` schaltet es ab, ein Pfad statt `false` ersetzt den Ordner. -`make sync-files` stößt den Abgleich sofort an, auch für Dateien, die noch keine Stunde ruhen, und wartet, bis er durch ist. +`make sync-files` stößt den Abgleich sofort an, auch für Dateien, die noch keine Minute ruhen, und wartet, bis er durch ist. Zugang auf jedem Gerät in `~/.local/share/desktop-app/sync.json`: @@ -81,7 +86,8 @@ Ohne diese Datei läuft die Extension rein lokal. Optional `"interval"`: Ruhetak Server-API (alle Aufrufe mit `Authorization: Bearer `): -- `POST /api/sync` – `{"since": , "events": [...], "todos": [...], "notes": [...], "folders": [...], "files": [...]}`; Antwort: Änderungen seit `since` in den angefragten Sammlungen plus neuer `cursor`. Ohne Sammlungen kommen die vier ersten, `files` nur auf Anfrage und in Paketen bis 16 MB: `more: true` heißt, mit dem neuen `cursor` weiter fragen. +- `POST /api/sync` – `{"since": , "events": [...], "todos": [...], "notes": [...], "folders": [...], "files": [...]}`; Antwort: Änderungen seit `since` in den angefragten Sammlungen plus neuer `cursor`. Ohne Sammlungen kommen die vier ersten, `files` nur auf Anfrage. Ein `files`-Eintrag ist `{"id", "updated", "content"}` (ganz), `{"id", "updated", "base": {"size", "hash"}, "append"}` (nur der Rest) oder ein Tombstone; zurück kommt nur der Steckbrief `{"id", "updated", "size"}`, bei Ablehnung mit `hash`. +- `GET /api/files/?from=&hash=` – Inhalt einer Datei. Passt der eigene Stand, kommt nur der Rest ab `from`, sonst alles; `X-From` sagt, ab wo. - `GET /api/events`, `/api/todos`, `/api/notes` – aktueller Stand ohne Gelöschte, für andere Anwendungen ### Server starten diff --git a/backend/app/config.py b/backend/app/config.py index 3d848de..5f654d0 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -13,5 +13,5 @@ SYNC_TOKEN = os.getenv("SYNC_TOKEN", "") COLLECTIONS = ("events", "todos", "notes", "folders", "files") BASE_COLLECTIONS = COLLECTIONS[:-1] -# Dateien kommen höchstens in Paketen dieser Größe zurück, der Client holt mit "more" weiter -FILES_PAGE_BYTES = int(os.getenv("FILES_PAGE_BYTES", 16 * 1024 * 1024)) +# Inhalt der Dateien (Claude-Sitzungen) liegt auf der Platte, in SQLite nur der Steckbrief +FILES_DIR = Path(os.getenv("FILES_DIR", DATA_DIR / "files")) diff --git a/backend/app/db.py b/backend/app/db.py index d844d5b..9fd37a1 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -25,11 +25,14 @@ CREATE INDEX IF NOT EXISTS items_seq ON items(seq); # Große Datei-Uploads halten die Datenbank sekundenlang. WAL lässt Leser derweil durch, # Schreiber warten bis zu 60 s statt mit "database is locked" abzubrechen. @contextmanager -def connect(): +def connect(write=False): DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH, timeout=60) conn.row_factory = sqlite3.Row try: + # Schreiber nacheinander: so greifen zwei Anhänge an dieselbe Datei nie gleichzeitig zu + if write: + conn.execute("BEGIN IMMEDIATE") yield conn conn.commit() finally: @@ -78,21 +81,6 @@ def since(conn, collection, seq): return [json.loads(r["data"]) for r in rows] -def since_paged(conn, collection, seq, max_bytes): - """Wie since, aber höchstens max_bytes Daten. Gibt (Einträge, letzter seq, mehr da) zurück.""" - rows = conn.execute( - "SELECT data, seq FROM items WHERE collection = ? AND seq > ? ORDER BY seq", (collection, seq) - ) - items, size, last = [], 0, seq - for r in rows: - if items and size + len(r["data"]) > max_bytes: - return items, last, True - items.append(json.loads(r["data"])) - size += len(r["data"]) - last = r["seq"] - return items, last, False - - def alive(conn, collection): rows = conn.execute( "SELECT data FROM items WHERE collection = ? AND deleted = 0 ORDER BY id", (collection,) diff --git a/backend/app/files.py b/backend/app/files.py new file mode 100644 index 0000000..dfbf5bc --- /dev/null +++ b/backend/app/files.py @@ -0,0 +1,111 @@ +import hashlib +import json +from pathlib import Path + +from .config import FILES_DIR + +# Dateien der PCs (Claude-Code-Sitzungen und Memory) liegen auf der Platte unter FILES_DIR, +# in SQLite steht nur noch der Steckbrief: id, updated, size, deleted. +# Sitzungen wachsen nur am Ende. Deshalb schickt der Client meist nur den neuen Rest ("append") +# und sagt dazu, welchen Stand er beim Server vermutet (base: size und Hash der letzten 64 KB). +# Stimmt das, wird angehängt. Sonst bekommt er den Server-Stand zurück und schickt beim +# nächsten Mal die ganze Datei oder holt sich die neuere. + +TAIL = 64 * 1024 + + +def path(file_id): + return FILES_DIR / file_id + + +def tail_hash(data: bytes) -> str: + return hashlib.sha256(data[-TAIL:]).hexdigest() + + +# Hash der letzten 64 KB vor Position size, ohne die ganze Datei zu lesen +def tail_hash_at(file_id, size) -> str: + with open(path(file_id), "rb") as f: + start = max(0, size - TAIL) + f.seek(start) + return hashlib.sha256(f.read(size - start)).hexdigest() + + +def meta(conn, file_id): + row = conn.execute("SELECT data FROM items WHERE collection = 'files' AND id = ?", (file_id,)).fetchone() + return json.loads(row["data"]) if row else None + + +def _store(conn, item): + from .db import cursor + + conn.execute( + "INSERT OR REPLACE INTO items (collection, id, updated, deleted, data, seq) VALUES ('files', ?, ?, ?, ?, ?)", + (item["id"], item["updated"], 1 if item.get("deleted") else 0, json.dumps(item, ensure_ascii=False), cursor(conn) + 1), + ) + + +# Server-Stand für den Client: Steckbrief plus Hash, damit er den Anschluss findet +def state(conn, file_id): + m = meta(conn, file_id) + if m and not m.get("deleted") and path(file_id).exists(): + m = {**m, "hash": tail_hash_at(file_id, m["size"])} + return m + + +def merge(conn, item): + """Nimmt eine Datei an: ganz (content), als Anhang (base + append) oder als Tombstone. + + Gibt (True, None) zurück, wenn übernommen, sonst (False, Server-Stand). + """ + file_id = item["id"] + current = meta(conn, file_id) + if current and current["updated"] >= item["updated"]: + return False, state(conn, file_id) + target = path(file_id) + if item.get("deleted"): + target.unlink(missing_ok=True) + _store(conn, {"id": file_id, "updated": item["updated"], "deleted": True}) + return True, None + if "append" in item: + base = item.get("base") or {} + if not current or current.get("deleted") or not target.exists() or current["size"] != base.get("size") \ + or tail_hash_at(file_id, current["size"]) != base.get("hash"): + return False, state(conn, file_id) + data = item["append"].encode() + with open(target, "ab") as f: + f.write(data) + _store(conn, {"id": file_id, "updated": item["updated"], "size": current["size"] + len(data)}) + return True, None + data = item["content"].encode() + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + _store(conn, {"id": file_id, "updated": item["updated"], "size": len(data)}) + return True, None + + +# Inhalt ab Position: passt der Anfang des Clients (size, hash), kommt nur der Rest, sonst alles. +def read_from(file_id, size, hash_) -> tuple[bytes, int]: + target = path(file_id) + total = target.stat().st_size + start = size if 0 < size <= total and tail_hash_at(file_id, size) == hash_ else 0 + with open(target, "rb") as f: + f.seek(start) + return f.read(), start + + +# Ältere Stände hielten den Inhalt in der Datenbank. Einmal auf die Platte schreiben. +def migrate(conn): + rows = conn.execute("SELECT id, data FROM items WHERE collection = 'files' AND deleted = 0").fetchall() + moved = 0 + for row in rows: + item = json.loads(row["data"]) + if "content" not in item: + continue + data = item.pop("content").encode() + target = path(row["id"]) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + item["size"] = len(data) + conn.execute("UPDATE items SET data = ? WHERE collection = 'files' AND id = ?", (json.dumps(item, ensure_ascii=False), row["id"])) + moved += 1 + return moved diff --git a/backend/app/main.py b/backend/app/main.py index 4f5201b..20a7b1b 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,11 +1,12 @@ import logging from contextlib import asynccontextmanager -from fastapi import FastAPI, Header, HTTPException +from fastapi import FastAPI, Header, HTTPException, Query +from fastapi.responses import Response from pydantic import BaseModel -from . import db -from .config import BASE_COLLECTIONS, COLLECTIONS, FILES_PAGE_BYTES, SYNC_TOKEN +from . import db, files +from .config import BASE_COLLECTIONS, COLLECTIONS, SYNC_TOKEN logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("app") @@ -16,6 +17,10 @@ async def lifespan(_app): if not SYNC_TOKEN: log.warning("SYNC_TOKEN ist leer, jede Anfrage wird abgelehnt.") db.init() + with db.connect(write=True) as conn: + moved = files.migrate(conn) + if moved: + log.info("%d Dateien aus der Datenbank nach %s verschoben", moved, files.FILES_DIR) yield @@ -57,30 +62,23 @@ def sync(body: SyncIn, authorization: str = Header(default="")): erkennt sie am gleichen updated und überspringt sie. """ check_token(authorization) - with db.connect() as conn: + with db.connect(write=True) as conn: accepted = 0 rejected = {collection: [] for collection in COLLECTIONS} for collection in COLLECTIONS: + merge = files.merge if collection == "files" else lambda c, item: db.merge(c, collection, item) for item in getattr(body, collection) or []: - ok, stored = db.merge(conn, collection, item) + ok, stored = merge(conn, item) if ok: accepted += 1 - else: + elif stored: rejected[collection].append(stored) out = {} - cursor = db.cursor(conn) for collection in body.requested: - if collection == "files": - # Dateien sind groß: nur ein Paket, der Stand bleibt beim letzten gelieferten Eintrag - changed, last, more = db.since_paged(conn, collection, body.since, FILES_PAGE_BYTES) - if more: - out["more"] = True - cursor = min(cursor, last) - else: - changed = db.since(conn, collection, body.since) + changed = db.since(conn, collection, body.since) ids = {item["id"] for item in changed} out[collection] = changed + [item for item in rejected[collection] if item["id"] not in ids] - out["cursor"] = cursor + out["cursor"] = db.cursor(conn) if accepted: log.info("sync: %d Einträge übernommen", accepted) return out @@ -106,3 +104,16 @@ def notes(authorization: str = Header(default="")): check_token(authorization) with db.connect() as conn: return db.alive(conn, "notes") + + +# Inhalt einer Datei. Mit from und hash (Länge und Hash der letzten 64 KB des eigenen Stands) +# kommt nur der Rest ab dieser Position, sonst die ganze Datei. X-From sagt, ab wo. +@app.get("/api/files/{file_id:path}") +def file_content(file_id: str, from_: int = Query(0, alias="from"), hash: str = "", authorization: str = Header(default="")): + check_token(authorization) + with db.connect() as conn: + meta = files.meta(conn, file_id) + if not meta or meta.get("deleted"): + raise HTTPException(status_code=404, detail="Datei unbekannt") + data, start = files.read_from(file_id, from_, hash) + return Response(content=data, media_type="application/octet-stream", headers={"X-From": str(start), "X-Size": str(start + len(data))}) diff --git a/backend/tests/test_sync.py b/backend/tests/test_sync.py index 7f0a057..9959d7d 100644 --- a/backend/tests/test_sync.py +++ b/backend/tests/test_sync.py @@ -9,10 +9,11 @@ from fastapi.testclient import TestClient def client(tmp_path): os.environ["DB_PATH"] = str(tmp_path / "test.db") os.environ["SYNC_TOKEN"] = "geheim" - os.environ["FILES_PAGE_BYTES"] = "150" - from app import config, db, main + os.environ["FILES_DIR"] = str(tmp_path / "files") + from app import config, db, files, main importlib.reload(config) importlib.reload(db) + importlib.reload(files) importlib.reload(main) with TestClient(main.app) as c: yield c @@ -94,31 +95,53 @@ def test_seit_cursor_liefert_nur_neues(client): def test_dateien_nur_auf_anfrage(client): - sync(client, files=[{"id": "-home-x/s1.jsonl", "updated": "2026-09-22T10:00:00.000Z", "content": "zeile"}]) + sync(client, files=[{"id": "-home-x/s1.jsonl", "updated": "2026-09-22T10:00:00.000Z", "content": "zeile\n"}]) # Handy und Extension fragen ohne files: nichts davon in der Antwort r = sync(client, events=[], todos=[], notes=[], folders=[]) assert "files" not in r and r["events"] == [] assert "files" not in sync(client) - # Nur Dateien angefragt: nur Dateien zurück, Stand zählt trotzdem + # Nur Dateien angefragt: nur der Steckbrief zurück, kein Inhalt r = sync(client, files=[]) - assert [f["id"] for f in r["files"]] == ["-home-x/s1.jsonl"] and "events" not in r + assert r["files"] == [{"id": "-home-x/s1.jsonl", "updated": "2026-09-22T10:00:00.000Z", "size": 6}] and "events" not in r assert r["cursor"] == 1 - # Tombstone gewinnt mit neuerem updated - r = sync(client, files=[{"id": "-home-x/s1.jsonl", "updated": "2026-09-22T11:00:00.000Z", "deleted": True}]) - assert r["files"][-1]["deleted"] is True -def test_dateien_seitenweise(client): - for i in range(3): - sync(client, files=[{"id": f"s{i}.jsonl", "updated": "2026-09-22T10:00:00.000Z", "content": "x" * 60}]) - # Jede Antwort höchstens 150 Bytes: ein Eintrag pro Seite, more zeigt an, dass es weitergeht - seen, since = [], 0 - for _ in range(5): - r = sync(client, since=since, files=[]) - seen += [f["id"] for f in r["files"]] - since = r["cursor"] - if not r.get("more"): - break - assert seen == ["s0.jsonl", "s1.jsonl", "s2.jsonl"] and since == 3 - # Basis-Sammlungen bleiben ungeteilt - assert "more" not in sync(client) +def test_dateien_inhalt_auf_platte_und_anhaengen(client, tmp_path): + import hashlib + sync(client, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T10:00:00.000Z", "content": "a\n"}]) + assert (tmp_path / "files" / "p" / "s.jsonl").read_bytes() == b"a\n" + # Anhang mit passendem Stand + base = {"size": 2, "hash": hashlib.sha256(b"a\n").hexdigest()} + r = sync(client, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T11:00:00.000Z", "base": base, "append": "b\n"}]) + assert (tmp_path / "files" / "p" / "s.jsonl").read_bytes() == b"a\nb\n" + assert r["files"][-1]["size"] == 4 + # Anhang mit falschem Stand: abgelehnt, Server-Stand samt Hash kommt zurück + r = sync(client, since=99, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T12:00:00.000Z", "base": base, "append": "c\n"}]) + assert r["files"] == [{"id": "p/s.jsonl", "updated": "2026-09-22T11:00:00.000Z", "size": 4, "hash": hashlib.sha256(b"a\nb\n").hexdigest()}] + assert (tmp_path / "files" / "p" / "s.jsonl").read_bytes() == b"a\nb\n" + # Inhalt holen: ganz, oder nur den Rest ab bekanntem Stand + r = client.get("/api/files/p/s.jsonl", headers=AUTH) + assert r.content == b"a\nb\n" and r.headers["X-From"] == "0" and r.headers["X-Size"] == "4" + r = client.get("/api/files/p/s.jsonl", params={"from": 2, "hash": base["hash"]}, headers=AUTH) + assert r.content == b"b\n" and r.headers["X-From"] == "2" + r = client.get("/api/files/p/s.jsonl", params={"from": 2, "hash": "falsch"}, headers=AUTH) + assert r.content == b"a\nb\n" and r.headers["X-From"] == "0" + # Löschen räumt die Platte, Tombstone bleibt, Inhalt weg + sync(client, files=[{"id": "p/s.jsonl", "updated": "2026-09-22T13:00:00.000Z", "deleted": True}]) + assert not (tmp_path / "files" / "p" / "s.jsonl").exists() + assert client.get("/api/files/p/s.jsonl", headers=AUTH).status_code == 404 + assert sync(client, files=[])["files"][-1]["deleted"] is True + + +def test_dateien_aus_datenbank_auf_platte(client, tmp_path): + import json + from app import db, files + # Alter Stand: Inhalt in der Datenbank + with db.connect(write=True) as conn: + conn.execute("INSERT INTO items VALUES ('files', 'alt/s.jsonl', '2026-09-01T00:00:00.000Z', 0, ?, 1)", + (json.dumps({"id": "alt/s.jsonl", "updated": "2026-09-01T00:00:00.000Z", "content": "alt\n"}),)) + with db.connect(write=True) as conn: + assert files.migrate(conn) == 1 + assert files.migrate(conn) == 0 + assert (tmp_path / "files" / "alt" / "s.jsonl").read_bytes() == b"alt\n" + assert sync(client, files=[])["files"] == [{"id": "alt/s.jsonl", "updated": "2026-09-01T00:00:00.000Z", "size": 4}] diff --git a/desktop@marha.de/files.js b/desktop@marha.de/files.js index 091d162..e677026 100644 --- a/desktop@marha.de/files.js +++ b/desktop@marha.de/files.js @@ -1,9 +1,14 @@ // Dateiabgleich für ~/.claude/projects (Claude-Code-Sitzungen und Memory), nur zwischen PCs. // Reine Logik ohne Dateisystem, damit sie sich testen lässt. sync.js liest und schreibt die Dateien. // -// Regeln: Eine Datei geht hoch, wenn sie sich geändert hat und seit einer Stunde nicht angefasst wurde. -// Vom Server kommt sie, wenn sie dort neuer ist. Bei Konflikten gewinnt die neuere Änderungszeit. -// Verschwundene Dateien werden als Tombstone gemeldet, gelöschte vom Server lokal entfernt. +// Regeln: Eine Datei geht hoch, wenn sie sich geändert hat und seit einer Minute nicht angefasst wurde. +// Sitzungen wachsen nur am Ende, deshalb geht meist nur der neue Rest ("append"), zusammen mit dem +// Stand, den der Server haben müsste (base: Größe und Hash der letzten 64 KB). Vom Server kommt eine +// Datei, wenn sie dort neuer ist, ebenfalls nur ab dem eigenen Stand. Bei Konflikten gewinnt die neuere +// Änderungszeit. Verschwundene Dateien werden als Tombstone gemeldet, gelöschte vom Server lokal entfernt. +import GLib from 'gi://GLib'; + +export const TAIL = 64 * 1024; // Zeitstempel aus der Änderungszeit der Datei, gleiche Form wie updated bei Terminen export function fileStamp(mtimeSeconds) { @@ -14,7 +19,19 @@ export function stampToSeconds(stamp) { return Math.floor(Date.parse(stamp) / 1000); } -// Was hochgeladen wird. local: Map id → mtime in Sekunden. pushed: id → updated, null = als gelöscht gemeldet. +// Hash der letzten 64 KB, gleiche Rechnung wie auf dem Server +export function tailHash(bytes) { + const tail = bytes.length > TAIL ? bytes.subarray(bytes.length - TAIL) : bytes; + return GLib.compute_checksum_for_bytes(GLib.ChecksumType.SHA256, new GLib.Bytes(tail)); +} + +// pushed[id] ist der Stand auf dem Server: {updated, size, hash}. Ältere Stände sind nur der +// Zeitstempel als String, null heißt als gelöscht gemeldet. +export function knownUpdated(known) { + return known && typeof known === 'object' ? known.updated : known; +} + +// Was hochgeladen wird. local: Map id → mtime in Sekunden. // upload: Einträge ohne Inhalt, remove: IDs, die es lokal nicht mehr gibt. export function planPush(local, pushed, nowSeconds, idleSeconds) { const upload = []; @@ -22,13 +39,31 @@ export function planPush(local, pushed, nowSeconds, idleSeconds) { if (nowSeconds - mtime < idleSeconds) continue; const updated = fileStamp(mtime); - if (pushed[id] !== updated) + if (knownUpdated(pushed[id]) !== updated) upload.push({id, updated}); } const remove = Object.keys(pushed).filter(id => pushed[id] !== null && !local.has(id)); return {upload, remove}; } +// Eintrag für den Server aus den Bytes der Datei: nur der Rest, wenn der bekannte Anfang noch stimmt, +// sonst die ganze Datei. null, wenn sie kein UTF-8 ist. +export function uploadItem(id, updated, bytes, known) { + const decoder = new TextDecoder('utf-8', {fatal: true}); + if (known && typeof known === 'object' && known.size < bytes.length && tailHash(bytes.subarray(0, known.size)) === known.hash) { + try { + return {id, updated, base: {size: known.size, hash: known.hash}, append: decoder.decode(bytes.subarray(known.size))}; + } catch { + // Schnitt mitten im Zeichen: dann eben ganz + } + } + try { + return {id, updated, content: decoder.decode(bytes)}; + } catch { + return null; + } +} + // Was mit Einträgen vom Server passiert: write, delete oder skip. // mine: der lokale Zeitstempel oder null, damit sync.js den Stand nachhalten kann. export function planPull(remote, local) { @@ -63,3 +98,10 @@ export function batches(items, sizeOf, maxBytes) { out.push(current); return out; } + +export function concat(a, b) { + const out = new Uint8Array(a.length + b.length); + out.set(a, 0); + out.set(b, a.length); + return out; +} diff --git a/desktop@marha.de/sync.js b/desktop@marha.de/sync.js index dc452d8..256aa2a 100644 --- a/desktop@marha.de/sync.js +++ b/desktop@marha.de/sync.js @@ -2,7 +2,7 @@ import Gio from 'gi://Gio'; import GLib from 'gi://GLib'; import Soup from 'gi://Soup'; -import {batches, fileStamp, planPull, planPush, stampToSeconds} from './files.js'; +import {batches, concat, planPull, planPush, stampToSeconds, tailHash, uploadItem} from './files.js'; import {readJson, writeJson} from './store.js'; // Abgleich mit dem Server. Zugang steht in sync.json im Datenordner: @@ -23,11 +23,11 @@ const INTERVALS = [ // Kurze Wartezeit nach einer eigenen Änderung, damit mehrere Klicks in einem Lauf landen const SOON_MS = 500; -// Dateien (Claude-Sitzungen und Memory): alle 10 Minuten, hoch geht nur, was eine Stunde ruht. +// Dateien (Claude-Sitzungen und Memory): gleicher Takt wie die Einträge, hoch geht, was eine Minute ruht. +// Meist nur der neue Rest einer Datei, siehe files.js. // "files" in sync.json: false schaltet ab, ein Pfad ersetzt ~/.claude/projects. -const FILES_FIRST_SECONDS = 60; -const FILES_INTERVAL_SECONDS = 10 * 60; -const FILES_IDLE_SECONDS = 60 * 60; +const FILES_FIRST_SECONDS = 10; +const FILES_IDLE_SECONDS = 60; const FILES_BATCH_BYTES = 16 * 1024 * 1024; // Von außen anstoßbar: make sync-files ruft diese Methode über D-Bus auf @@ -267,16 +267,21 @@ export class Sync { } // ── Dateien ────────────────────────────────────────────────────────────── + // Nach jedem Lauf neu planen, im Takt der Einträge (1 s, 5 s, 10 s je nach Aktivität). _scheduleFiles(seconds) { if (this._stopped) return; this._filesTimerId = GLib.timeout_add_seconds(GLib.PRIORITY_DEFAULT, seconds, () => { this._filesTimerId = null; - this._filesCycle().catch(e => this._fail(e.message)).finally(() => this._scheduleFiles(FILES_INTERVAL_SECONDS)); + this._filesCycle().catch(e => this._fail(e.message)).finally(() => this._scheduleFiles(this.interval)); return GLib.SOURCE_REMOVE; }); } + _path(id) { + return GLib.build_filenamev([this._filesRoot, id]); + } + // Eigener Lauf mit eigenem Stand: die Antwort enthält nur Dateien, der Takt der Einträge bleibt unberührt. // Gibt zurück, was übertragen wurde, oder null, wenn gerade ein Lauf aktiv ist. // force: Ruhezeit ignorieren, alles Geänderte geht hoch. @@ -291,50 +296,64 @@ export class Sync { const now = Math.floor(GLib.get_real_time() / 1e6); const {upload, remove} = planPush(local, state.pushed, now, force ? 0 : FILES_IDLE_SECONDS); const items = []; - for (const entry of upload) { - const content = readText(GLib.build_filenamev([this._filesRoot, entry.id])); - if (content !== null) - items.push({...entry, content}); + for (const {id, updated} of upload) { + const bytes = readBytes(this._path(id)); + const item = bytes && uploadItem(id, updated, bytes, state.pushed[id]); + if (item) + items.push(Object.assign(item, {_size: bytes.length, _hash: tailHash(bytes)})); } const nowStamp = new Date(now * 1000).toISOString(); for (const id of remove) items.push({id, updated: nowStamp, deleted: true}); - // Hoch in Paketen, runter in Paketen: solange der Server "more" meldet, weiter fragen - const packets = batches(items, item => (item.content ?? '').length, FILES_BATCH_BYTES); - let index = 0; - let more = true; - while (index < packets.length || more) { - const files = packets[index++] ?? []; - const before = state.cursor; + // Mindestens ein Lauf, auch ohne eigene Änderungen: holt Neues vom Server + const packets = items.length ? batches(items, item => (item.content ?? item.append ?? '').length, FILES_BATCH_BYTES) : [[]]; + for (const packet of packets) { + const files = packet.map(({_size, _hash, ...item}) => item); const response = await this._post({since: state.cursor, files}, this._filesSession); // Ein Server ohne Dateien-Sammlung würde alles verwerfen, der Stand bliebe sonst falsch if (!Array.isArray(response.files)) throw new Error('Server kennt keine Dateien, dort erst make prod ausführen'); - more = Boolean(response.more) && response.cursor !== before; - for (const item of files) { - state.pushed[item.id] = item.deleted ? null : item.updated; + for (const item of packet) { + // Kommt der Eintrag mit anderem updated zurück, hat der Server abgelehnt. Sein Stand gilt dann als bekannt, + // beim nächsten Lauf passt der Anhang oder es geht die ganze Datei. + const echo = response.files.find(f => f.id === item.id && f.updated !== item.updated); + if (echo) { + state.pushed[item.id] = echo.deleted ? null : {updated: echo.updated, size: echo.size, hash: echo.hash}; + continue; + } + state.pushed[item.id] = item.deleted ? null : {updated: item.updated, size: item._size, hash: item._hash}; result[item.deleted ? 'removed' : 'uploaded']++; } - this._applyFiles(response, local, result); + await this._applyFiles(response, local, result); writeJson(this._statePath, this._state); } + if (result.uploaded + result.removed + result.written + result.deleted) + this._touchActivity(); } finally { this._filesBusy = false; } return result; } - _applyFiles(response, local, result) { + async _applyFiles(response, local, result) { const state = this._state.files; - for (const {file, action, mine} of planPull(response.files ?? [], local)) { - const path = GLib.build_filenamev([this._filesRoot, file.id]); + for (const {file, action, mine} of planPull(response.files, local)) { + const path = this._path(file.id); if (action === 'write') { + // Nur den Rest holen, wenn der eigene Anfang noch zum Server passt + const have = readBytes(path) ?? new Uint8Array(0); + const {bytes, from} = await this._download(file.id, have.length, tailHash(have)); + const appended = from > 0 && from === have.length; + const content = appended ? concat(have, bytes) : bytes; GLib.mkdir_with_parents(GLib.path_get_dirname(path), 0o700); - GLib.file_set_contents(path, file.content); + if (appended) + appendBytes(path, bytes); + else + GLib.file_set_contents(path, content); const seconds = stampToSeconds(file.updated); Gio.File.new_for_path(path).set_attribute_uint64('time::modified', seconds, Gio.FileQueryInfoFlags.NONE, null); local.set(file.id, seconds); - state.pushed[file.id] = file.updated; + state.pushed[file.id] = {updated: file.updated, size: content.length, hash: tailHash(content)}; result.written++; } else if (action === 'delete') { GLib.unlink(path); @@ -343,14 +362,36 @@ export class Sync { result.deleted++; } else if (file.deleted && mine === null) { state.pushed[file.id] = null; - } else if (mine === file.updated) { - state.pushed[file.id] = mine; + } else if (mine === file.updated && typeof state.pushed[file.id] !== 'object') { + // Gleicher Stand wie der Server, aber noch ohne Größe und Hash gemerkt + const bytes = readBytes(path); + if (bytes) + state.pushed[file.id] = {updated: mine, size: bytes.length, hash: tailHash(bytes)}; } - // Sonst ist die lokale Datei neuer und geht hoch, sobald sie eine Stunde ruht. + // Sonst ist die lokale Datei neuer und geht hoch, sobald sie eine Minute ruht. } state.cursor = response.cursor; } + // Inhalt vom Server ab dem eigenen Stand. X-From sagt, ab wo er wirklich liefert (0 = ganze Datei). + _download(id, from, hash) { + const path = id.split('/').map(encodeURIComponent).join('/'); + const message = Soup.Message.new('GET', `${this._config.url}/api/files/${path}?from=${from}&hash=${hash}`); + message.request_headers.append('Authorization', `Bearer ${this._config.token}`); + return new Promise((resolve, reject) => { + this._filesSession.send_and_read_async(message, GLib.PRIORITY_DEFAULT, null, (session, result) => { + try { + const bytes = session.send_and_read_finish(result); + if (message.get_status() !== Soup.Status.OK) + throw new Error(`HTTP ${message.get_status()} ${message.get_reason_phrase()}`); + resolve({bytes: bytes.get_data() ?? new Uint8Array(0), from: Number(message.get_response_headers().get_one('X-From'))}); + } catch (e) { + reject(e); + } + }); + }); + } + _fail(message) { if (this._stopped) return; @@ -385,12 +426,18 @@ function scanFiles(root) { return local; } -// Textdatei lesen. null, wenn sie fehlt oder kein UTF-8 ist. -function readText(path) { +// Datei als Bytes lesen. null, wenn sie fehlt. +function readBytes(path) { try { const [, bytes] = GLib.file_get_contents(path); - return new TextDecoder('utf-8', {fatal: true}).decode(bytes); + return bytes; } catch { return null; } } + +function appendBytes(path, bytes) { + const stream = Gio.File.new_for_path(path).append_to(Gio.FileCreateFlags.NONE, null); + stream.write_bytes(new GLib.Bytes(bytes), null); + stream.close(null); +} diff --git a/dev/tests/run.js b/dev/tests/run.js index 5dd0305..acda77e 100644 --- a/dev/tests/run.js +++ b/dev/tests/run.js @@ -1,7 +1,7 @@ // Tests der reinen Module: gjs -m dev/tests/run.js import * as D from '../../desktop@marha.de/dates.js'; import {continueList, renderMarkdown} from '../../desktop@marha.de/markdown.js'; -import {batches, fileStamp, planPull, planPush, stampToSeconds} from '../../desktop@marha.de/files.js'; +import {batches, concat, fileStamp, planPull, planPush, stampToSeconds, tailHash, uploadItem} from '../../desktop@marha.de/files.js'; let failed = 0; function eq(actual, expected, name) { @@ -75,9 +75,19 @@ const H = 3600, NOW = 1_800_000_000; eq(fileStamp(NOW), '2027-01-15T08:00:00.000Z', 'fileStamp'); eq(stampToSeconds('2027-01-15T08:00:00.000Z'), NOW, 'stampToSeconds'); const local = new Map([['a/s.jsonl', NOW - 2 * H], ['a/frisch.jsonl', NOW - 10], ['m/MEMORY.md', NOW - 3 * H]]); -eq(planPush(local, {'m/MEMORY.md': fileStamp(NOW - 3 * H), 'weg.md': fileStamp(NOW - 5 * H), 'alt.md': null}, NOW, H), +eq(planPush(local, {'m/MEMORY.md': {updated: fileStamp(NOW - 3 * H), size: 5, hash: 'x'}, 'weg.md': fileStamp(NOW - 5 * H), 'alt.md': null}, NOW, H), {upload: [{id: 'a/s.jsonl', updated: fileStamp(NOW - 2 * H)}], remove: ['weg.md']}, 'planPush: nur ruhende, geänderte Dateien hoch, Verschwundene als Tombstone, Gemeldete nicht nochmal'); +const enc = s => new TextEncoder().encode(s); +eq(tailHash(enc('abc')), 'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad', 'tailHash = SHA-256 bei kleinen Dateien'); +eq(tailHash(enc('x'.repeat(70000) + 'abc')), tailHash(enc('y'.repeat(70000) + 'x'.repeat(65533) + 'abc')), 'tailHash sieht nur die letzten 64 KB'); +const known = {updated: 'u', size: 4, hash: tailHash(enc('a\nb\n'))}; +eq(uploadItem('s', 'v', enc('a\nb\nc\n'), known), {id: 's', updated: 'v', base: {size: 4, hash: known.hash}, append: 'c\n'}, 'uploadItem: nur der Rest, wenn der Anfang stimmt'); +eq(uploadItem('s', 'v', enc('x\nb\nc\n'), known), {id: 's', updated: 'v', content: 'x\nb\nc\n'}, 'uploadItem: ganze Datei bei anderem Anfang'); +eq(uploadItem('s', 'v', enc('a\n'), known), {id: 's', updated: 'v', content: 'a\n'}, 'uploadItem: ganze Datei, wenn kürzer'); +eq(uploadItem('s', 'v', enc('a\nb\nc\n'), 'u'), {id: 's', updated: 'v', content: 'a\nb\nc\n'}, 'uploadItem: alter Stand ohne Größe → ganz'); +eq(uploadItem('s', 'v', new Uint8Array([0xff, 0xfe]), null), null, 'uploadItem: kein UTF-8 → nichts'); +eq(Array.from(concat(new Uint8Array([1, 2]), new Uint8Array([3]))), [1, 2, 3], 'concat'); eq(planPull([ {id: 'a/s.jsonl', updated: fileStamp(NOW - 1 * H), content: 'x'}, {id: 'a/s.jsonl', updated: fileStamp(NOW - 3 * H), content: 'x'}, diff --git a/dev/tests/sync.sh b/dev/tests/sync.sh index 772c17a..66ff9e7 100755 --- a/dev/tests/sync.sh +++ b/dev/tests/sync.sh @@ -46,8 +46,7 @@ trap cleanup EXIT echo "=== Server starten" rm -rf "$SERVER_DATA" -# FILES_PAGE_BYTES winzig: Dateien kommen einzeln, der Client muss die Seiten nachholen -(cd "$ROOT/backend" && DATA_DIR="$SERVER_DATA" SYNC_TOKEN=test FILES_PAGE_BYTES=10 .venv/bin/uvicorn app.main:app --port $PORT > /tmp/desktop-app-sync-server.log 2>&1 &) +(cd "$ROOT/backend" && DATA_DIR="$SERVER_DATA" SYNC_TOKEN=test .venv/bin/uvicorn app.main:app --port $PORT > /tmp/desktop-app-sync-server.log 2>&1 &) for i in $(seq 1 30); do curl -s "$URL/api/health" >/dev/null && break; sleep 0.5; done SERVER_PID=$(pgrep -f "uvicorn app.main:app --port $PORT") check "Server läuft" "$(curl -s $URL/api/health)" '{"ok":true}' @@ -156,45 +155,45 @@ check "Lokal gelöschter Ordner auf dem Server als gelöscht" "$(folders | pytho echo "=== Dateien: Claude-Sitzungen zwischen PCs" FILES=/tmp/desktop-app-nested/claude-projects -# Alle Dateien vom Server, seitenweise wie der Client -allfiles() { python3 - "$URL" <<'PY' -import json, sys, urllib.request -url, since, out = sys.argv[1], 0, [] -while True: - req = urllib.request.Request(url + "/api/sync", data=json.dumps({"since": since, "files": []}).encode(), headers={"Authorization": "Bearer test", "Content-Type": "application/json"}) - d = json.load(urllib.request.urlopen(req)) - out += d["files"]; since = d["cursor"] - if not d.get("more"): break -print(json.dumps(out)) -PY -} -files() { allfiles | python3 -c 'import json,sys; print(sorted((f["id"], bool(f.get("deleted"))) for f in json.load(sys.stdin)))'; } +SFILES=$SERVER_DATA/files +files() { api -X POST "$URL/api/sync" -d '{"since": 0, "files": []}' | python3 -c 'import json,sys; print(sorted((f["id"], bool(f.get("deleted"))) for f in json.load(sys.stdin)["files"]))'; } +same() { cmp -s "$SFILES/$1" "$FILES/$1" && echo gleich || echo verschieden; } # Lauf wie vom Timer: Ruhezeit gilt timercycle() { "$E" "$PRE ext._sync._filesCycle(); 'ok'" >/dev/null; sleep 3; } # Wie make sync-files, nur auf dem Bus der Test-Shell: erzwungen, wartet, bis der Lauf durch ist. filescycle() { DBUS_SESSION_BUS_ADDRESS=$(cat /tmp/desktop-app-nested/bus) gdbus call --session --dest org.gnome.Shell --object-path /de/marha/desktop --method de.marha.Desktop.SyncFiles | sed -e "s/^('//" -e "s/',)$//"; } mkdir -p "$FILES/-home-x/memory" -echo '{"a":1}' > "$FILES/-home-x/alt.jsonl"; touch -d '2 hours ago' "$FILES/-home-x/alt.jsonl" -echo '# Memo' > "$FILES/-home-x/memory/MEMORY.md"; touch -d '90 minutes ago' "$FILES/-home-x/memory/MEMORY.md" -echo 'läuft noch' > "$FILES/-home-x/frisch.jsonl" +printf '{"a":1}\n' > "$FILES/-home-x/alt.jsonl"; touch -d '2 hours ago' "$FILES/-home-x/alt.jsonl" +printf '# Memo\n' > "$FILES/-home-x/memory/MEMORY.md"; touch -d '90 minutes ago' "$FILES/-home-x/memory/MEMORY.md" +printf 'läuft noch\n' > "$FILES/-home-x/frisch.jsonl" timercycle check "Timer-Lauf: ruhende Dateien auf dem Server, frische nicht" "$(files)" "[('-home-x/alt.jsonl', False), ('-home-x/memory/MEMORY.md', False)]" +check "Inhalt liegt als Datei auf dem Server" "$(same -home-x/alt.jsonl) $(same -home-x/memory/MEMORY.md)" "gleich gleich" +check "Takt nach Übertragung wie beim Kalender" "$(val 'ext._sync.interval')" "1" check "make sync-files erzwingt auch die frische Datei" "$(filescycle)" "hoch 1, gelöscht gemeldet 0, runter 0, lokal gelöscht 0" check "Frische Datei jetzt auf dem Server" "$(files)" "[('-home-x/alt.jsonl', False), ('-home-x/frisch.jsonl', False), ('-home-x/memory/MEMORY.md', False)]" check "Normale Anfrage bekommt keine Dateien" "$(api -X POST "$URL/api/sync" -d '{"since": 0}' | python3 -c 'import json,sys; d=json.load(sys.stdin); print("files" in d, "events" in d)')" "False True" -check "Stand der Dateien getrennt gemerkt" "$(val 'Object.keys(state().files.pushed).sort().join(",") + " " + (state().files.cursor > 0)')" "-home-x/alt.jsonl,-home-x/frisch.jsonl,-home-x/memory/MEMORY.md true" -push '{"files": [{"id": "-home-y/s2.jsonl", "updated": "2026-09-01T00:00:00.000Z", "content": "vom anderen PC"}]}' +check "Stand der Dateien mit Größe gemerkt" "$(val 'Object.keys(state().files.pushed).sort().join(",") + " " + (state().files.cursor > 0) + " " + state().files.pushed["-home-x/alt.jsonl"].size')" "-home-x/alt.jsonl,-home-x/frisch.jsonl,-home-x/memory/MEMORY.md true 8" +printf '{"a":2}\n' >> "$FILES/-home-x/alt.jsonl"; touch -d '2 minutes ago' "$FILES/-home-x/alt.jsonl" +timercycle +check "Angehängte Zeile nach einer Minute Ruhe auf dem Server" "$(same -home-x/alt.jsonl) $(val 'state().files.pushed["-home-x/alt.jsonl"].size')" "gleich 16" +printf 'neu\n' > "$FILES/-home-x/alt.jsonl"; touch -d '2 minutes ago' "$FILES/-home-x/alt.jsonl" +timercycle +check "Umgeschriebene Datei ganz ersetzt" "$(same -home-x/alt.jsonl) $(cat "$SFILES/-home-x/alt.jsonl")" "gleich neu" +push '{"files": [{"id": "-home-y/s2.jsonl", "updated": "2026-09-01T00:00:00.000Z", "content": "vom anderen PC\n"}]}' check "Abgleich holt die Datei" "$(filescycle)" "hoch 0, gelöscht gemeldet 0, runter 1, lokal gelöscht 0" check "Datei vom Server geschrieben, Änderungszeit übernommen" "$(cat "$FILES/-home-y/s2.jsonl") $(stat -c %Y "$FILES/-home-y/s2.jsonl")" "vom anderen PC $(date -u -d 2026-09-01T00:00:00Z +%s)" -echo '{"a":2}' > "$FILES/-home-x/alt.jsonl"; touch -d '70 minutes ago' "$FILES/-home-x/alt.jsonl" +H=$(printf 'vom anderen PC\n' | sha256sum | cut -d' ' -f1) +push '{"files": [{"id": "-home-y/s2.jsonl", "updated": "2026-09-01T01:00:00.000Z", "base": {"size": 15, "hash": "'$H'"}, "append": "zweite Zeile\n"}]}' +check "Anhang vom Server kommt nur als Rest" "$(filescycle) | $(tr '\n' '|' < "$FILES/-home-y/s2.jsonl")" "hoch 0, gelöscht gemeldet 0, runter 1, lokal gelöscht 0 | vom anderen PC|zweite Zeile|" +printf 'meins\n' > "$FILES/-home-x/alt.jsonl"; touch -d '90 seconds ago' "$FILES/-home-x/alt.jsonl" push '{"files": [{"id": "-home-x/alt.jsonl", "updated": "'$(date -u -d '3 hours ago' +%Y-%m-%dT%H:%M:%S.000Z)'", "content": "veraltet"}]}' -filescycle >/dev/null -check "Lokal neuer gewinnt und geht hoch" "$(allfiles | python3 -c 'import json,sys; print([f["content"].strip() for f in json.load(sys.stdin) if f["id"] == "-home-x/alt.jsonl"])')" "['{\"a\":2}']" -check "Seiten: Einzelanfrage liefert nur ein Paket mit more" "$(api -X POST "$URL/api/sync" -d '{"since": 0, "files": []}' | python3 -c 'import json,sys; d=json.load(sys.stdin); print(len(d["files"]), d.get("more"))')" "1 True" +timercycle +check "Lokal neuer gewinnt und geht hoch" "$(cat "$SFILES/-home-x/alt.jsonl")" "meins" rm "$FILES/-home-x/memory/MEMORY.md" push '{"files": [{"id": "-home-y/s2.jsonl", "updated": "'$(now)'", "deleted": true}]}' check "Abgleich meldet Löschungen" "$(filescycle)" "hoch 0, gelöscht gemeldet 1, runter 0, lokal gelöscht 1" -check "Gelöscht in beide Richtungen" "$(files | python3 -c 'import sys; print("MEMORY.md\x27, True" in sys.stdin.read())') $([ -e "$FILES/-home-y/s2.jsonl" ] && echo da || echo weg)" "True weg" +check "Gelöscht in beide Richtungen, auch auf der Server-Platte" "$(files | python3 -c 'import sys; print("MEMORY.md\x27, True" in sys.stdin.read())') $([ -e "$FILES/-home-y/s2.jsonl" ] && echo da || echo weg) $([ -e "$SFILES/-home-x/memory/MEMORY.md" ] && echo da || echo weg)" "True weg weg" check "Gemeldete Löschung nicht nochmal" "$(val 'state().files.pushed["-home-x/memory/MEMORY.md"] === null && state().files.pushed["-home-y/s2.jsonl"] === null')" "true" check "Zweiter Lauf überträgt nichts" "$(filescycle)" "hoch 0, gelöscht gemeldet 0, runter 0, lokal gelöscht 0" check "Zweiter Lauf schickt nichts erneut" "$(files)" "[('-home-x/alt.jsonl', False), ('-home-x/frisch.jsonl', False), ('-home-x/memory/MEMORY.md', True), ('-home-y/s2.jsonl', True)]"