update
This commit is contained in:
@@ -81,7 +81,7 @@ Ohne diese Datei läuft die Extension rein lokal. Optional `"interval"`: Ruhetak
|
||||
|
||||
Server-API (alle Aufrufe mit `Authorization: Bearer <SYNC_TOKEN>`):
|
||||
|
||||
- `POST /api/sync` – `{"since": <cursor>, "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.
|
||||
- `POST /api/sync` – `{"since": <cursor>, "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.
|
||||
- `GET /api/events`, `/api/todos`, `/api/notes` – aktueller Stand ohne Gelöschte, für andere Anwendungen
|
||||
|
||||
### Server starten
|
||||
|
||||
@@ -12,3 +12,6 @@ SYNC_TOKEN = os.getenv("SYNC_TOKEN", "")
|
||||
# files: Dateien der PCs (Claude-Sitzungen), kommen nur zurück, wenn der Client sie anfragt
|
||||
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))
|
||||
|
||||
@@ -75,6 +75,21 @@ 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,)
|
||||
|
||||
@@ -5,7 +5,7 @@ from fastapi import FastAPI, Header, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from . import db
|
||||
from .config import BASE_COLLECTIONS, COLLECTIONS, SYNC_TOKEN
|
||||
from .config import BASE_COLLECTIONS, COLLECTIONS, FILES_PAGE_BYTES, SYNC_TOKEN
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
log = logging.getLogger("app")
|
||||
@@ -68,11 +68,19 @@ def sync(body: SyncIn, authorization: str = Header(default="")):
|
||||
else:
|
||||
rejected[collection].append(stored)
|
||||
out = {}
|
||||
cursor = db.cursor(conn)
|
||||
for collection in body.requested:
|
||||
changed = db.since(conn, collection, body.since)
|
||||
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)
|
||||
ids = {item["id"] for item in changed}
|
||||
out[collection] = changed + [item for item in rejected[collection] if item["id"] not in ids]
|
||||
out["cursor"] = db.cursor(conn)
|
||||
out["cursor"] = cursor
|
||||
if accepted:
|
||||
log.info("sync: %d Einträge übernommen", accepted)
|
||||
return out
|
||||
|
||||
@@ -9,6 +9,7 @@ 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
|
||||
importlib.reload(config)
|
||||
importlib.reload(db)
|
||||
@@ -105,3 +106,19 @@ def test_dateien_nur_auf_anfrage(client):
|
||||
# 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)
|
||||
|
||||
@@ -56,6 +56,8 @@ export class Sync {
|
||||
this._filesTimerId = null;
|
||||
this._filesBusy = false;
|
||||
this._session = new Soup.Session({timeout: 10});
|
||||
// Dateien sind groß, der Server braucht für ein Paket länger als für Einträge
|
||||
this._filesSession = new Soup.Session({timeout: 120});
|
||||
this._busy = false;
|
||||
this._failing = false;
|
||||
this._timerId = null;
|
||||
@@ -148,6 +150,7 @@ export class Sync {
|
||||
// Laufende Anfrage abbrechen, ohne das als Verbindungsfehler zu melden
|
||||
this._stopped = true;
|
||||
this._session.abort();
|
||||
this._filesSession.abort();
|
||||
}
|
||||
|
||||
async _cycle() {
|
||||
@@ -195,12 +198,12 @@ export class Sync {
|
||||
};
|
||||
}
|
||||
|
||||
_post(payload) {
|
||||
_post(payload, session = this._session) {
|
||||
const message = Soup.Message.new('POST', `${this._config.url}/api/sync`);
|
||||
message.request_headers.append('Authorization', `Bearer ${this._config.token}`);
|
||||
message.set_request_body_from_bytes('application/json', new GLib.Bytes(new TextEncoder().encode(JSON.stringify(payload))));
|
||||
return new Promise((resolve, reject) => {
|
||||
this._session.send_and_read_async(message, GLib.PRIORITY_DEFAULT, null, (session, result) => {
|
||||
session.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)
|
||||
@@ -294,13 +297,18 @@ export class Sync {
|
||||
const nowStamp = new Date(now * 1000).toISOString();
|
||||
for (const id of remove)
|
||||
items.push({id, updated: nowStamp, deleted: true});
|
||||
// Mindestens ein Lauf, auch ohne eigene Änderungen: holt Neues vom Server
|
||||
const packets = items.length ? batches(items, item => (item.content ?? '').length, FILES_BATCH_BYTES) : [[]];
|
||||
for (const files of packets) {
|
||||
const response = await this._post({since: state.cursor, files});
|
||||
// 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;
|
||||
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;
|
||||
result[item.deleted ? 'removed' : 'uploaded']++;
|
||||
|
||||
@@ -46,7 +46,8 @@ trap cleanup EXIT
|
||||
|
||||
echo "=== Server starten"
|
||||
rm -rf "$SERVER_DATA"
|
||||
(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 &)
|
||||
# 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 &)
|
||||
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}'
|
||||
@@ -155,7 +156,19 @@ 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
|
||||
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"]))'; }
|
||||
# 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)))'; }
|
||||
# Wie make sync-files, nur auf dem Bus der Test-Shell. 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"
|
||||
@@ -172,7 +185,8 @@ check "Datei vom Server geschrieben, Änderungszeit übernommen" "$(cat "$FILES/
|
||||
echo '{"a":2}' > "$FILES/-home-x/alt.jsonl"; touch -d '70 minutes 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" "$(api -X POST "$URL/api/sync" -d '{"since": 0, "files": []}' | python3 -c 'import json,sys; print([f["content"].strip() for f in json.load(sys.stdin)["files"] if f["id"] == "-home-x/alt.jsonl"])')" "['{\"a\":2}']"
|
||||
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"
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user