From 7d5fe3483905870281980352dc6703dcaa7f5e83 Mon Sep 17 00:00:00 2001 From: jenken827 Date: Sat, 12 Sep 2026 15:20:41 +0800 Subject: [PATCH 1/5] fix(sync): keep UI responsive while a WebDAV sync is running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit During a sync the app froze: clicks stopped registering, opening a book did nothing, and the window eventually died. Three compounding causes: - Transfer progress was pushed to the store on EVERY chunk. With 3-5 concurrent transfers that sustains a synchronous React update storm on the renderer main thread (readest hit the same bug — their Sentry READEST-2 — and ships a progress throttle). Chunk progress is now throttled to one emit per 300ms per task; task start/completion still emit immediately. - Every DB write wrapped in runWithDbRetry waited up to 12s for a running sync to finish before even attempting the write (waitForSyncToSettle), so opening a book stalled for the full timeout. The settle wait is now capped at 1.5s — genuine SQLite lock contention is already handled by the retry loop itself. - applyChanges yielded the main thread only every 100 applied records; large remote snapshots could hog it for seconds between yields. Now every 25. --- packages/core/src/db/write-retry.ts | 8 +++++++- packages/core/src/sync/simple-sync.ts | 4 +++- packages/core/src/sync/sync-files.ts | 16 +++++++++++++++- 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/packages/core/src/db/write-retry.ts b/packages/core/src/db/write-retry.ts index 3c91f67a8..5e6b95b95 100644 --- a/packages/core/src/db/write-retry.ts +++ b/packages/core/src/db/write-retry.ts @@ -10,7 +10,13 @@ export function isRetryableDbError(error: unknown): boolean { return RETRYABLE_DB_ERROR_PATTERNS.some((pattern) => message.includes(pattern)); } -export async function waitForSyncToSettle(timeoutMs = 12000): Promise { +/** + * Wait for an in-flight sync to settle before writing. Capped low on purpose: + * the UI must never stall behind a long-running sync (a full first sync can + * run for many minutes). Genuine SQLite lock contention is handled by the + * retry loop in runWithDbRetry instead. + */ +export async function waitForSyncToSettle(timeoutMs = 1500): Promise { try { const { useSyncStore } = await import("../stores/sync-store"); const startedAt = Date.now(); diff --git a/packages/core/src/sync/simple-sync.ts b/packages/core/src/sync/simple-sync.ts index aa476a485..bc4398acb 100644 --- a/packages/core/src/sync/simple-sync.ts +++ b/packages/core/src/sync/simple-sync.ts @@ -360,7 +360,9 @@ export async function applyChanges( } processedRecords++; - if (processedRecords % 100 === 0) { + // Yield the main thread regularly: applying a large remote snapshot + // must not starve UI interactions (clicks, navigation) while it runs. + if (processedRecords % 25 === 0) { console.log( `[SimpleSync] Applying table ${tableName}: ${processedRecords}/${tableData.records.length} record(s) processed`, ); diff --git a/packages/core/src/sync/sync-files.ts b/packages/core/src/sync/sync-files.ts index 0ea8b35c7..988ec6caa 100644 --- a/packages/core/src/sync/sync-files.ts +++ b/packages/core/src/sync/sync-files.ts @@ -41,6 +41,14 @@ const UPLOAD_CONCURRENCY = 3; const DOWNLOAD_CONCURRENCY = 5; const MIGRATION_CONCURRENCY = 3; const REMOTE_CLEANUP_CONCURRENCY = 3; +/** + * Chunk progress from concurrent transfers arrives far more often than the UI + * needs to repaint. Emitting every chunk floods the renderer main thread with + * store updates + React re-renders and freezes the whole app while transfers + * run, so per-task chunk progress is throttled to this interval (task start + * and completion still always emit). + */ +const PROGRESS_EMIT_MIN_INTERVAL_MS = 300; export interface SyncFilesOptions { forceUploadAll?: boolean; @@ -317,9 +325,15 @@ async function runFileTasks( }); }; + let lastChunkEmitAt = 0; + emitProgress(); const result = await task.run((loaded, taskTotal) => { - if (taskTotal > 0) emitProgress(loaded, taskTotal); + if (taskTotal <= 0) return; + const now = Date.now(); + if (now - lastChunkEmitAt < PROGRESS_EMIT_MIN_INTERVAL_MS) return; + lastChunkEmitAt = now; + emitProgress(loaded, taskTotal); }); completed++; const finalTotal = From 777ca2c830df0ba99c1119c99b6c164f0665bc64 Mon Sep 17 00:00:00 2001 From: jenken827 Date: Sat, 12 Sep 2026 16:23:56 +0800 Subject: [PATCH 2/5] =?UTF-8?q?refactor(sync):=20per-book=20cloud=20sync?= =?UTF-8?q?=20engine=20=E2=80=94=20per-book=20isolation=20replaces=20per-d?= =?UTF-8?q?evice=20snapshots?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-device snapshot layout (device-{id}.json = all 11 tables, full dump per sync) was the root cause of sync dragging the whole UI down: one giant JSON.stringify/parse per pass, O(library × devices) payload growth (AI chat history and reading sessions grow unbounded), and every sync re-downloading and re-applying every peer even when nothing changed. New cloud layout (under /readany/sync): - index.json — per-book/thread markers {b, a, d} / {t, d}, written read-merge-write as a union (tombstones prevent resurrection) - books/{bookId}.json — book row + ALL its highlights/notes/bookmarks + per-table deleted maps (per-item LWW merge) - threads/{threadId}.json — thread metadata only - chat/{YYYY-MM-DD}.json — chat messages by creation day; devices keep a pulled-day cursor and only fetch missing days (plus days changed under the cursor by late offline pushes) - profile/{tags,book_tags,book_groups,skills}.json — single-table files - sessions/{YYYY-MM}.json — reading sessions in monthly shards Every request is now KB-sized and sync work is O(changed). Sync also propagates annotation deletions per book: sync_tombstones gained a book_id column (migration + attribution at delete time + deletedBookIds on the legacy wire format for LAN). - LAN sync intentionally stays on the legacy device-snapshot protocol (runSimpleSync); cloud backends (webdav/s3) switch to runPerBookSync. - No backward compatibility for old cloud snapshots by design: users reset the remote folder and re-sync once. --- .../src/db/__tests__/book-queries.test.ts | 6 +- .../src/db/__tests__/bookmark-queries.test.ts | 2 +- packages/core/src/db/book-queries.ts | 6 +- packages/core/src/db/bookmark-queries.ts | 7 +- packages/core/src/db/db-core.ts | 12 +- packages/core/src/db/highlight-queries.ts | 7 +- packages/core/src/db/note-queries.ts | 7 +- packages/core/src/stores/sync-store.test.ts | 1 + packages/core/src/stores/sync-store.ts | 10 +- .../src/sync/__tests__/per-book-sync.test.ts | 491 +++++++++ packages/core/src/sync/per-book-sync.ts | 978 ++++++++++++++++++ packages/core/src/sync/simple-sync.ts | 53 +- packages/core/src/sync/sync-backend.ts | 2 + 13 files changed, 1555 insertions(+), 27 deletions(-) create mode 100644 packages/core/src/sync/__tests__/per-book-sync.test.ts create mode 100644 packages/core/src/sync/per-book-sync.ts diff --git a/packages/core/src/db/__tests__/book-queries.test.ts b/packages/core/src/db/__tests__/book-queries.test.ts index 2e7f998e5..956a76a83 100644 --- a/packages/core/src/db/__tests__/book-queries.test.ts +++ b/packages/core/src/db/__tests__/book-queries.test.ts @@ -320,9 +320,9 @@ describe("book-queries", () => { "book-1", ]); expect(mockExecute).toHaveBeenCalledWith("DELETE FROM books WHERE id = ?", ["book-1"]); - expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "hl-1", "highlights"); - expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "note-1", "notes"); - expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "bm-1", "bookmarks"); + expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "hl-1", "highlights", "book-1"); + expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "note-1", "notes", "book-1"); + expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "bm-1", "bookmarks", "book-1"); expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "book-1", "books"); }); }); diff --git a/packages/core/src/db/__tests__/bookmark-queries.test.ts b/packages/core/src/db/__tests__/bookmark-queries.test.ts index b3af3757e..1b1662bea 100644 --- a/packages/core/src/db/__tests__/bookmark-queries.test.ts +++ b/packages/core/src/db/__tests__/bookmark-queries.test.ts @@ -139,7 +139,7 @@ describe("bookmark-queries", () => { mockExecute.mockResolvedValue(undefined); await deleteBookmark("bm-1"); - expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "bm-1", "bookmarks"); + expect(coreMocks.insertTombstone).toHaveBeenCalledWith(mockDb, "bm-1", "bookmarks", "book-1"); expect(mockExecute).toHaveBeenCalledWith("DELETE FROM bookmarks WHERE id = ?", ["bm-1"]); }); }); diff --git a/packages/core/src/db/book-queries.ts b/packages/core/src/db/book-queries.ts index 1e9ef65f7..0cb3e7144 100644 --- a/packages/core/src/db/book-queries.ts +++ b/packages/core/src/db/book-queries.ts @@ -374,13 +374,13 @@ export async function deleteBook(id: string, options: DeleteBookOptions = {}): P ]); for (const row of highlightRows) { - await insertTombstone(database, row.id, "highlights"); + await insertTombstone(database, row.id, "highlights", id); } for (const row of noteRows) { - await insertTombstone(database, row.id, "notes"); + await insertTombstone(database, row.id, "notes", id); } for (const row of bookmarkRows) { - await insertTombstone(database, row.id, "bookmarks"); + await insertTombstone(database, row.id, "bookmarks", id); } await database.execute("DELETE FROM highlights WHERE book_id = ?", [id]); diff --git a/packages/core/src/db/bookmark-queries.ts b/packages/core/src/db/bookmark-queries.ts index 9571a4333..a87a48672 100644 --- a/packages/core/src/db/bookmark-queries.ts +++ b/packages/core/src/db/bookmark-queries.ts @@ -47,6 +47,11 @@ export async function insertBookmark(bookmark: Bookmark): Promise { export async function deleteBookmark(id: string): Promise { const database = await getDB(); - await insertTombstone(database, id, "bookmarks"); + const rows = await database.select<{ book_id: string | null }>( + "SELECT book_id FROM bookmarks WHERE id = ?", + [id], + ); + const tombstoneBookId: [] | [string] = rows[0]?.book_id ? [rows[0].book_id] : []; + await insertTombstone(database, id, "bookmarks", ...tombstoneBookId); await database.execute("DELETE FROM bookmarks WHERE id = ?", [id]); } diff --git a/packages/core/src/db/db-core.ts b/packages/core/src/db/db-core.ts index 1ce4f4598..b3cbb2217 100644 --- a/packages/core/src/db/db-core.ts +++ b/packages/core/src/db/db-core.ts @@ -310,12 +310,13 @@ export async function insertTombstone( database: IDatabase, id: string, tableName: string, + bookId?: string, ): Promise { const deviceId = await getDeviceId(); try { await database.execute( - "INSERT OR REPLACE INTO sync_tombstones (id, table_name, deleted_at, device_id) VALUES (?, ?, ?, ?)", - [id, tableName, Date.now(), deviceId], + "INSERT OR REPLACE INTO sync_tombstones (id, table_name, deleted_at, device_id, book_id) VALUES (?, ?, ?, ?, ?)", + [id, tableName, Date.now(), deviceId, bookId ?? null], ); } catch { // sync_tombstones table might not exist on older schema @@ -573,6 +574,13 @@ export async function initDatabase(): Promise { await database.execute( "CREATE INDEX IF NOT EXISTS idx_tombstones_deleted_at ON sync_tombstones(deleted_at)", ); + // Migration 5b: attribute tombstones to their book so per-book sync can + // propagate annotation deletions inside the book's own sync file. + try { + await database.execute("ALTER TABLE sync_tombstones ADD COLUMN book_id TEXT"); + } catch { + // Column already exists + } // Migration 6: Sync metadata table await database.execute(` diff --git a/packages/core/src/db/highlight-queries.ts b/packages/core/src/db/highlight-queries.ts index b6237f6b5..a686bbfde 100644 --- a/packages/core/src/db/highlight-queries.ts +++ b/packages/core/src/db/highlight-queries.ts @@ -206,6 +206,11 @@ export async function updateHighlight(id: string, updates: Partial): export async function deleteHighlight(id: string): Promise { const database = await getDB(); - await insertTombstone(database, id, "highlights"); + const rows = await database.select<{ book_id: string | null }>( + "SELECT book_id FROM highlights WHERE id = ?", + [id], + ); + const tombstoneBookId: [] | [string] = rows[0]?.book_id ? [rows[0].book_id] : []; + await insertTombstone(database, id, "highlights", ...tombstoneBookId); await database.execute("DELETE FROM highlights WHERE id = ?", [id]); } diff --git a/packages/core/src/db/note-queries.ts b/packages/core/src/db/note-queries.ts index e217827f2..cbf757c99 100644 --- a/packages/core/src/db/note-queries.ts +++ b/packages/core/src/db/note-queries.ts @@ -126,6 +126,11 @@ export async function updateNote(id: string, updates: Partial): Promise { const database = await getDB(); - await insertTombstone(database, id, "notes"); + const rows = await database.select<{ book_id: string | null }>( + "SELECT book_id FROM notes WHERE id = ?", + [id], + ); + const tombstoneBookId: [] | [string] = rows[0]?.book_id ? [rows[0].book_id] : []; + await insertTombstone(database, id, "notes", ...tombstoneBookId); await database.execute("DELETE FROM notes WHERE id = ?", [id]); } diff --git a/packages/core/src/stores/sync-store.test.ts b/packages/core/src/stores/sync-store.test.ts index 3237a744e..22c859a98 100644 --- a/packages/core/src/stores/sync-store.test.ts +++ b/packages/core/src/stores/sync-store.test.ts @@ -55,6 +55,7 @@ vi.mock("../services/platform", () => ({ vi.mock("../sync/sync-backend-factory", () => factoryMocks); vi.mock("../sync/simple-sync", () => syncMocks); +vi.mock("../sync/per-book-sync", () => ({ runPerBookSync: syncMocks.runSimpleSync })); vi.mock("../sync/sync-files", () => syncMocks); vi.mock("../events/library-events", () => libraryEventMocks); vi.mock("./reading-session-store", () => readingSessionMocks); diff --git a/packages/core/src/stores/sync-store.ts b/packages/core/src/stores/sync-store.ts index 6c4b358b2..a0ac85c98 100644 --- a/packages/core/src/stores/sync-store.ts +++ b/packages/core/src/stores/sync-store.ts @@ -590,11 +590,15 @@ export const useSyncStore = create((set, get) => ({ set({ status: "syncing-files", error: null, progress: null }); try { + const { runPerBookSync } = await import("../sync/per-book-sync"); const { runSimpleSync } = await import("../sync/simple-sync"); + // LAN keeps the legacy device-snapshot protocol; cloud backends use the + // per-book engine. + const runSync = backend.type === "lan" ? runSimpleSync : runPerBookSync; const receiveOnly = backend.type === "lan" || resolvedDirection === "download"; const uploadOnly = resolvedDirection === "upload"; - const result = await runSimpleSync( + const result = await runSync( backend, (progress) => { set({ progress }); @@ -799,13 +803,15 @@ export const useSyncStore = create((set, get) => ({ return result; } + const { runPerBookSync } = await import("../sync/per-book-sync"); const { runSimpleSync } = await import("../sync/simple-sync"); + const runSync = backend.type === "lan" ? runSimpleSync : runPerBookSync; set({ status: "syncing-files", error: null, progress: null }); const receiveOnly = direction === "download"; const startTime = Date.now(); - const simpleResult = await runSimpleSync( + const simpleResult = await runSync( backend, (progress) => { set({ diff --git a/packages/core/src/sync/__tests__/per-book-sync.test.ts b/packages/core/src/sync/__tests__/per-book-sync.test.ts new file mode 100644 index 000000000..9e5861ba2 --- /dev/null +++ b/packages/core/src/sync/__tests__/per-book-sync.test.ts @@ -0,0 +1,491 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { ISyncBackend, RemoteFile } from "../sync-backend"; + +type Row = Record; + +const TABLE_COLUMNS: Record = { + books: [ + "id", + "title", + "author", + "file_path", + "format", + "cover_url", + "updated_at", + "deleted_at", + "sync_status", + "created_at", + "added_at", + ], + highlights: ["id", "book_id", "text", "updated_at", "created_at"], + notes: ["id", "book_id", "content", "updated_at", "created_at"], + bookmarks: ["id", "book_id", "label", "updated_at", "created_at"], + threads: ["id", "title", "updated_at", "created_at"], + messages: ["id", "thread_id", "content", "created_at"], + tags: ["id", "name", "updated_at", "created_at"], + book_tags: ["id", "book_id", "tag_id", "updated_at"], + book_groups: ["id", "name", "updated_at", "created_at"], + skills: ["id", "name", "updated_at", "created_at"], + reading_sessions: ["id", "book_id", "started_at", "updated_at"], + sync_tombstones: ["id", "table_name", "deleted_at", "device_id", "book_id"], +}; + +const dbMocks = vi.hoisted(() => ({ + currentDb: null as FakePerBookDb | null, + getDB: vi.fn(), + ensureNoTransaction: vi.fn(async () => {}), + cleanupOrphanedSyncRows: vi.fn(async () => {}), + getDeviceId: vi.fn(async () => "device-a"), + deleteBook: vi.fn(async (id: string) => { + dbMocks.currentDb?.deleteBook(id); + }), + deleteThread: vi.fn(async (id: string) => { + dbMocks.currentDb?.deleteThread(id); + }), +})); + +vi.mock("../../db/database", () => ({ + getDB: dbMocks.getDB, + ensureNoTransaction: dbMocks.ensureNoTransaction, + cleanupOrphanedSyncRows: dbMocks.cleanupOrphanedSyncRows, + getDeviceId: dbMocks.getDeviceId, + deleteBook: dbMocks.deleteBook, + deleteThread: dbMocks.deleteThread, +})); + +vi.mock("../../services/platform", () => ({ + getPlatformService: vi.fn(() => ({ isDesktop: false })), +})); + +const syncFileMocks = vi.hoisted(() => ({ + syncFiles: vi.fn(async () => ({ + filesUploaded: 0, + filesDownloaded: 0, + filesUploadFailed: 0, + filesDownloadFailed: 0, + })), +})); + +vi.mock("../sync-files", () => syncFileMocks); + +const { runPerBookSync } = await import("../per-book-sync"); + +/** Minimal in-memory DB covering exactly the SQL shapes the engine issues. */ +class FakePerBookDb { + tables = new Map>(); + tombstones = new Map(); + syncMetadata = new Map(); + + constructor() { + for (const table of Object.keys(TABLE_COLUMNS)) { + this.tables.set(table, new Map()); + } + } + + table(name: string): Map { + return this.tables.get(name) ?? new Map(); + } + + insert(table: string, row: Row): void { + this.table(table).set(String(row.id), { ...row }); + } + + deleteBook(bookId: string): void { + this.table("books").delete(bookId); + for (const table of ["highlights", "notes", "bookmarks", "reading_sessions"]) { + for (const [id, row] of [...this.table(table)]) { + if (String(row.book_id) === bookId) this.table(table).delete(id); + } + } + for (const [id, row] of [...this.table("messages")]) { + if (this.threadsOf(String(row.thread_id)).length === 0) this.table("messages").delete(id); + } + for (const [id, row] of [...this.table("threads")]) { + if (String(row.book_id ?? "") === bookId) this.table("threads").delete(id); + } + } + + private threadsOf(threadId: string): Row[] { + return this.table("threads").size >= 0 + ? [...this.table("threads")].filter(([, row]) => String(row.id) === threadId).map(([, r]) => r) + : []; + } + + deleteThread(threadId: string): void { + this.table("threads").delete(threadId); + for (const [id, row] of [...this.table("messages")]) { + if (String(row.thread_id) === threadId) this.table("messages").delete(id); + } + } + + async select(sql: string, params: unknown[] = []): Promise { + const normalized = sql.replace(/\s+/g, " ").trim(); + + const pragma = normalized.match(/^PRAGMA table_info\((\w+)\)$/); + if (pragma) { + return (TABLE_COLUMNS[pragma[1]] ?? []).map((name) => ({ name })) as T[]; + } + + const metadataSelect = normalized.match( + /^SELECT value FROM sync_metadata WHERE key = ('[^']+'|\?)$/, + ); + if (metadataSelect) { + const rawKey = metadataSelect[1]; + const key = rawKey === "?" ? String(params[0]) : rawKey.slice(1, -1); + const value = this.syncMetadata.get(key); + return (value === undefined ? [] : [{ value }]) as T[]; + } + + const byIdMatch = normalized.match(/^SELECT (\*|\w+) FROM (\w+) WHERE id = \?$/); + if (byIdMatch) { + const [, column, table] = byIdMatch; + const row = this.table(table).get(String(params[0])); + if (!row) return []; + return (column === "*" ? [{ ...row }] : [{ [column]: row[column] }]) as T[]; + } + + const allRowsMatch = normalized.match(/^SELECT \* FROM (\w+)$/); + if (allRowsMatch) { + return [...this.table(allRowsMatch[1]).values()].map((row) => ({ ...row })) as T[]; + } + + const byColumnMatch = normalized.match(/^SELECT \* FROM (\w+) WHERE (\w+) = \?$/); + if (byColumnMatch) { + const [, table, column] = byColumnMatch; + return [...this.table(table).values()] + .filter((row) => String(row[column] ?? "") === String(params[0])) + .map((row) => ({ ...row })) as T[]; + } + + const groupMarkers = normalized.match( + /^SELECT book_id, MAX\(updated_at\) AS max_ts FROM (\w+) GROUP BY book_id$/, + ); + if (groupMarkers) { + const acc = new Map(); + for (const row of this.table(groupMarkers[1]).values()) { + const bookId = String(row.book_id); + acc.set(bookId, Math.max(acc.get(bookId) ?? 0, Number(row.updated_at ?? 0))); + } + return [...acc].map(([book_id, max_ts]) => ({ book_id, max_ts })) as T[]; + } + + const tombstonesForBook = normalized.match( + /^SELECT t\.id, t\.deleted_at FROM sync_tombstones t WHERE t\.book_id = \? AND t\.table_name = '(\w+)' AND NOT EXISTS \(SELECT 1 FROM \w+ WHERE id = t\.id\)$/, + ); + if (tombstonesForBook) { + const table = tombstonesForBook[1]; + const bookId = String(params[0]); + return [...this.tombstones.values()] + .filter( + (row) => row.book_id === bookId && row.table_name === table && !this.table(table).has(String(row.id)), + ) + .map((row) => ({ id: row.id, deleted_at: row.deleted_at })) as T[]; + } + + const tableTombstones = normalized.match( + /^SELECT t\.id, t\.deleted_at FROM sync_tombstones t WHERE t\.table_name = '(\w+)' AND NOT EXISTS \(SELECT 1 FROM \w+ WHERE id = t\.id\)$/, + ); + if (tableTombstones) { + const table = tableTombstones[1]; + return [...this.tombstones.values()] + .filter((row) => row.table_name === table && !this.table(table).has(String(row.id))) + .map((row) => ({ id: row.id, deleted_at: row.deleted_at })) as T[]; + } + + if (normalized === "SELECT * FROM reading_sessions WHERE updated_at > ?") { + const since = Number(params[0]); + return [...this.table("reading_sessions").values()] + .filter((row) => Number(row.updated_at ?? 0) > since) + .map((row) => ({ ...row })) as T[]; + } + + if (normalized === "SELECT * FROM messages WHERE created_at > ?") { + const since = Number(params[0]); + return [...this.table("messages").values()] + .filter((row) => Number(row.created_at ?? 0) > since) + .map((row) => ({ ...row })) as T[]; + } + + throw new Error(`Unexpected select: ${normalized}`); + } + + async execute(sql: string, params: unknown[] = []): Promise { + const normalized = sql.replace(/\s+/g, " ").trim(); + + const metadataSet = normalized.match( + /^INSERT OR REPLACE INTO sync_metadata \(key, value\) VALUES \((?:'[^']+'|\?), (?:'[^']+'|\?)\)$/, + ); + if (metadataSet) { + const values = [...normalized.matchAll(/(?:'([^']+)'|\?)/g)].map((m, i) => + m[1] !== undefined ? m[1] : String(params[i]), + ); + this.syncMetadata.set(values[0], values[1]); + return; + } + + if (normalized.startsWith("INSERT OR REPLACE INTO sync_tombstones")) { + const [id, tableName, deletedAt, deviceId, bookId] = params; + this.tombstones.set(`${String(tableName)}:${String(id)}`, { + id: String(id), + table_name: String(tableName), + deleted_at: Number(deletedAt), + device_id: String(deviceId), + book_id: bookId === null || bookId === undefined ? null : String(bookId), + }); + return; + } + + const deleteMatch = normalized.match(/^DELETE FROM (\w+) WHERE id = \?$/); + if (deleteMatch) { + this.table(deleteMatch[1]).delete(String(params[0])); + return; + } + + const upsert = normalized.match( + /^INSERT INTO (\w+) \(([^)]+)\) VALUES \([^)]+\) ON CONFLICT\((\w+)\) DO (UPDATE SET .+|NOTHING)$/, + ); + if (upsert) { + const [, table, columnList, pk] = upsert; + const columns = columnList.split(",").map((c) => c.trim()); + const row: Row = {}; + columns.forEach((column, i) => { + row[column] = params[i]; + }); + this.table(table).set(String(row[pk]), row); + return; + } + + throw new Error(`Unexpected execute: ${normalized}`); + } +} + +/** Scriptable fake backend recording every JSON payload it receives. */ +class FakeBackend implements ISyncBackend { + readonly type = "webdav" as const; + files = new Map(); + listedDirs: string[] = []; + + constructor(initial?: Record) { + if (initial) for (const [path, value] of Object.entries(initial)) this.files.set(path, value); + } + + async ensureDirectories(): Promise {} + async ensureDirectory(): Promise {} + async testConnection(): Promise { + return true; + } + async listDir(path: string): Promise { + this.listedDirs.push(path); + return [...this.files.keys()] + .filter((filePath) => filePath.startsWith(`${path}/`)) + .map((filePath) => ({ + name: filePath.slice(path.length + 1), + path: filePath, + size: 0, + lastModified: 0, + isDirectory: false, + })); + } + async getJSON(path: string): Promise { + return (this.files.get(path) as T) ?? null; + } + async putJSON(path: string, data: unknown): Promise { + this.files.set(path, data); + } + async put(): Promise {} + async get(): Promise { + return new Uint8Array(); + } + async getWithProgress(): Promise { + return new Uint8Array(); + } + async getJSONBinary(): Promise { + return null; + } + async delete(): Promise {} + async exists(): Promise { + return false; + } + async move(): Promise {} + async getDisplayName(): Promise { + return "fake"; + } +} + +function seedBook(db: FakePerBookDb, id: string, updatedAt: number, title = "Book"): void { + db.insert("books", { id, title, author: "", file_path: "", format: "epub", updated_at: updatedAt }); +} + +const T1 = 1_700_000_000_000; +const T2 = T1 + 60_000; +const T3 = T1 + 120_000; + +describe("runPerBookSync (per-book cloud engine)", () => { + let db: FakePerBookDb; + + beforeEach(() => { + db = new FakePerBookDb(); + dbMocks.currentDb = db; + dbMocks.getDB.mockResolvedValue(db); + dbMocks.getDeviceId.mockResolvedValue("device-a"); + }); + + afterEach(() => { + dbMocks.currentDb = null; + vi.restoreAllMocks(); + }); + + it("pulls a remote book file and applies book plus annotations", async () => { + seedBook(db, "book-1", T1); + db.insert("highlights", { id: "hl-local", book_id: "book-1", text: "old", updated_at: T1 }); + + const backend = new FakeBackend({ + "/readany/sync/index.json": { + schemaVersion: 2, + updatedAt: T2, + books: { "book-1": { b: T2, a: T2 } }, + threads: {}, + }, + "/readany/sync/books/book-1.json": { + schemaVersion: 1, + bookId: "book-1", + book: { id: "book-1", title: "Book", updated_at: T2 }, + highlights: [ + { id: "hl-local", book_id: "book-1", text: "updated", updated_at: T2 }, + { id: "hl-new", book_id: "book-1", text: "new", updated_at: T2 }, + ], + notes: [], + bookmarks: [], + writerDeviceId: "device-b", + updatedAt: T2, + }, + }); + + const result = await runPerBookSync(backend); + + expect(result.success).toBe(true); + expect(db.table("books").get("book-1")?.updated_at).toBe(T2); + expect(db.table("highlights").get("hl-new")?.text).toBe("new"); + expect(db.table("highlights").get("hl-local")?.text).toBe("updated"); + expect(backend.files.get("/readany/sync/index.json")).toMatchObject({ + books: { "book-1": { b: T2, a: T2 } }, + }); + }); + + it("skips books whose markers already match the remote index", async () => { + seedBook(db, "book-1", T2); + const backend = new FakeBackend({ + "/readany/sync/index.json": { + schemaVersion: 2, + updatedAt: T2, + books: { "book-1": { b: T2, a: T2 } }, + threads: {}, + }, + "/readany/sync/books/book-1.json": { + schemaVersion: 1, + bookId: "book-1", + book: { id: "book-1", updated_at: T2 }, + highlights: [], + notes: [], + bookmarks: [], + writerDeviceId: "device-b", + updatedAt: T2, + }, + }); + const getSpy = vi.spyOn(backend, "getJSON"); + + const result = await runPerBookSync(backend); + + expect(result.success).toBe(true); + expect(getSpy).not.toHaveBeenCalledWith("/readany/sync/books/book-1.json"); + }); + + it("deletes the local book when the remote index carries a newer tombstone", async () => { + seedBook(db, "book-1", T1); + db.insert("highlights", { id: "hl-1", book_id: "book-1", text: "x", updated_at: T1 }); + + const backend = new FakeBackend({ + "/readany/sync/index.json": { + schemaVersion: 2, + updatedAt: T3, + books: { "book-1": { d: T3 } }, + threads: {}, + }, + }); + + const result = await runPerBookSync(backend); + + expect(result.success).toBe(true); + expect(db.table("books").has("book-1")).toBe(false); + expect(db.table("highlights").has("hl-1")).toBe(false); + expect(dbMocks.deleteBook).toHaveBeenCalledWith("book-1"); + }); + + it("pushes changed books read-merge-write and updates the index", async () => { + seedBook(db, "book-1", T3); + db.insert("highlights", { id: "hl-2", book_id: "book-1", text: "local new", updated_at: T3 }); + + const backend = new FakeBackend({ + "/readany/sync/index.json": { + schemaVersion: 2, + updatedAt: T2, + books: { "book-1": { b: T2, a: T1 } }, + threads: {}, + }, + "/readany/sync/books/book-1.json": { + schemaVersion: 1, + bookId: "book-1", + book: { id: "book-1", title: "Book", updated_at: T2 }, + highlights: [ + { id: "hl-2", book_id: "book-1", text: "remote old", updated_at: T1 }, + { id: "hl-remote-only", book_id: "book-1", text: "remote only", updated_at: T2 }, + ], + notes: [], + bookmarks: [], + writerDeviceId: "device-b", + updatedAt: T2, + }, + }); + + const result = await runPerBookSync(backend); + + expect(result.success).toBe(true); + const pushed = backend.files.get("/readany/sync/books/book-1.json") as { + highlights: Array<{ id: string; text: string }>; + }; + const texts = pushed.highlights.map((h) => h.text).sort(); + expect(texts).toEqual(["local new", "remote only"]); + expect(backend.files.get("/readany/sync/index.json")).toMatchObject({ + books: { "book-1": { b: T3, a: T3 } }, + }); + }); + + it("pulls chat day files past the cursor and advances the cursor", async () => { + const backend = new FakeBackend({ + "/readany/sync/index.json": { schemaVersion: 2, updatedAt: T2, books: {}, threads: {} }, + "/readany/sync/chat/2026-09-01.json": { + schemaVersion: 1, + date: "2026-09-01", + messages: [{ id: "msg-1", thread_id: "th-1", content: "hello", created_at: T1 }], + updatedAt: T1, + }, + "/readany/sync/chat/2026-09-05.json": { + schemaVersion: 1, + date: "2026-09-05", + messages: [{ id: "msg-2", thread_id: "th-1", content: "later", created_at: T2 }], + updatedAt: T2, + }, + }); + await db.execute("INSERT OR REPLACE INTO sync_metadata (key, value) VALUES (?, ?)", [ + "perbook:chat-pulled-day", + "2026-08-31", + ]); + + const result = await runPerBookSync(backend); + + expect(result.success).toBe(true); + expect(db.table("messages").get("msg-1")?.content).toBe("hello"); + expect(db.table("messages").get("msg-2")?.content).toBe("later"); + expect(db.syncMetadata.get("perbook:chat-pulled-day")).toBe("2026-09-05"); + }); +}); diff --git a/packages/core/src/sync/per-book-sync.ts b/packages/core/src/sync/per-book-sync.ts new file mode 100644 index 000000000..0dcc97585 --- /dev/null +++ b/packages/core/src/sync/per-book-sync.ts @@ -0,0 +1,978 @@ +/** + * Per-book cloud sync engine (WebDAV / S3). + * + * Replaces the per-device full-snapshot layout for cloud backends. Remote + * layout (all under the synced root, e.g. /readany/sync): + * + * index.json — membership + change markers, one entry per + * book / thread: { b: bookUpdatedAt, a: annotationsUpdatedAt, + * d: deletedAt } / { t: threadUpdatedAt, d: deletedAt }. + * Written read-merge-write by every device. + * books/{bookId}.json — the book row plus ALL of its highlights, + * notes and bookmarks, and a per-table + * `deleted` map (id → deletedAt). + * threads/{threadId}.json — thread row only (title/metadata). + * chat/{YYYY-MM-DD}.json — every chat message CREATED that day, with a + * `deleted` map. Devices keep a pulled-day + * cursor and only fetch days past it, plus + * days whose file changed under the cursor + * (late-arriving offline messages). + * profile/{table}.json — tags / book_tags / book_groups / skills, + * one uniform single-table file each. + * sessions/{YYYY-MM}.json — reading_sessions monthly shards. + * + * Why this shape: every request carries KB-sized payloads, sync work is + * O(changed) instead of O(library × devices), and a single giant + * JSON.stringify/parse never runs on the renderer main thread. + * + * Merge rules (unchanged from the snapshot engine): last-writer-wins per row + * on the table's timestamp column, with deleted_at breaking ties; annotations + * and messages merge per item; deletions travel as explicit tombstone maps. + * The shared index is a union: every writer merges what it saw before + * writing, so a racing writer can only drop its own delta for one pass. + * + * Backward compatibility: intentionally none for cloud sync — devices on the + * old engine keep using per-device snapshots among themselves; a device + * switched to this engine starts from an empty remote layout and re-pushes + * the full library once. LAN sync keeps the old snapshot protocol and is + * unaffected. + */ + +import { + cleanupOrphanedSyncRows, + ensureNoTransaction, + getDB, + getDeviceId as getLocalDeviceId, +} from "../db/database"; +type Row = Record; +import { getPlatformService } from "../services/platform"; +import { + getLastSyncTimestamp, + localizeSyncedBookRecord, + setLastSyncTimestamp, + shouldApplyRemoteRecord, + upsertRecord, + withDatabaseLockRetry, +} from "./simple-sync"; +import type { ISyncBackend, RemoteFile } from "./sync-backend"; +import type { SyncFilesOptions } from "./sync-files"; +import type { SyncProgress } from "./sync-types"; + +export interface PerBookSyncOptions { + receiveOnly?: boolean; + /** When true, bypass timestamp comparisons and force-apply all remote records */ + forceApply?: boolean; + fileSyncOptions?: SyncFilesOptions; +} + +const SYNC_ROOT = "/readany/sync"; +const INDEX_PATH = `${SYNC_ROOT}/index.json`; +const BOOKS_DIR = `${SYNC_ROOT}/books`; +const THREADS_DIR = `${SYNC_ROOT}/threads`; +const CHAT_DIR = `${SYNC_ROOT}/chat`; +const PROFILE_DIR = `${SYNC_ROOT}/profile`; +const SESSIONS_DIR = `${SYNC_ROOT}/sessions`; + +const CHAT_PULLED_DAY_KEY = "perbook:chat-pulled-day"; +const CHAT_MERGED_DAYS_KEY = "perbook:chat-merged-days"; +const CHAT_PUSHED_AT_KEY = "perbook:chat-pushed-at"; + +const BOOK_LOCAL_EXCLUDED_COLUMNS = ["is_vectorized", "vectorize_progress"]; +const ANNOTATION_TABLES = ["highlights", "notes", "bookmarks"] as const; +type AnnotationTable = (typeof ANNOTATION_TABLES)[number]; +type DeletedMap = Record; + +interface BookIndexEntry { + /** Book row updated_at marker. */ + b?: number; + /** Max annotations updated_at / tombstone marker for this book. */ + a?: number; + /** Tombstone: book deleted at. */ + d?: number; +} + +interface ThreadIndexEntry { + /** Thread row updated_at marker. */ + t?: number; + /** Tombstone: thread deleted at. */ + d?: number; +} + +interface SyncIndexFile { + schemaVersion: 2; + updatedAt: number; + books: Record; + threads: Record; +} + +interface BookSyncFile { + schemaVersion: 1; + bookId: string; + book: Row; + highlights: Row[]; + notes: Row[]; + bookmarks: Row[]; + deleted?: Partial>; + writerDeviceId: string; + updatedAt: number; +} + +interface ThreadRowFile { + schemaVersion: 1; + threadId: string; + thread: Row; + writerDeviceId: string; + updatedAt: number; +} + +interface ChatDayFile { + schemaVersion: 1; + date: string; + messages: Row[]; + deleted?: DeletedMap; + updatedAt: number; +} + +interface ProfileSyncFile { + schemaVersion: 1; + rows: Row[]; + deleted?: DeletedMap; + updatedAt: number; +} + +interface SessionShardFile { + schemaVersion: 1; + month: string; + sessions: Row[]; + updatedAt: number; +} + +interface LocalBookState { + book: Row; + highlights: Row[]; + notes: Row[]; + bookmarks: Row[]; + deletedMaps: Partial>; + markerB: number; + markerA: number; +} + +function isForeignKeyConstraintError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return message.includes("FOREIGN KEY constraint failed") || message.includes("(code: 787)"); +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function dayKeyOf(timestamp: number): string { + const date = new Date(timestamp); + return `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, "0")}-${String( + date.getDate(), + ).padStart(2, "0")}`; +} + +function monthKeyOf(timestamp: number): string { + return dayKeyOf(timestamp).slice(0, 7); +} + +function stripBookLocalColumns(row: Row): Row { + const copy = { ...row }; + for (const column of BOOK_LOCAL_EXCLUDED_COLUMNS) delete copy[column]; + return copy; +} + +function emptyIndex(): SyncIndexFile { + return { schemaVersion: 2, updatedAt: 0, books: {}, threads: {} }; +} + +/** Per-field max union of two index/thread entries (markers and tombstones). */ +function mergeIndexEntry< + T extends Partial, +>(remote: T | undefined, ours: T | undefined): T { + const merged = { ...(remote ?? {}), ...(ours ?? {}) } as T; + for (const key of ["b", "a", "t", "d"] as const) { + const remoteValue = remote?.[key]; + const ourValue = ours?.[key]; + if (remoteValue !== undefined && ourValue !== undefined) { + merged[key] = Math.max(remoteValue, ourValue) as T[typeof key]; + } + } + return merged; +} + +function mergeDeletedMap( + remote: DeletedMap | undefined, + ours: DeletedMap | undefined, +): DeletedMap | undefined { + if (!remote && !ours) return undefined; + const merged: DeletedMap = { ...(remote ?? {}) }; + for (const [id, ts] of Object.entries(ours ?? {})) { + merged[id] = Math.max(merged[id] ?? 0, ts); + } + return merged; +} + +/** Batched annotation markers for every book that has any annotation. */ +async function localAnnotationMarkers( + db: Awaited>, +): Promise> { + const markers = new Map(); + for (const table of ANNOTATION_TABLES) { + try { + const rows = await db.select<{ book_id: string; max_ts: number }>( + `SELECT book_id, MAX(updated_at) AS max_ts FROM ${table} GROUP BY book_id`, + ); + for (const row of rows) { + markers.set(row.book_id, Math.max(markers.get(row.book_id) ?? 0, row.max_ts)); + } + } catch { + // Table may not exist on very old schemas. + } + } + return markers; +} + +/** + * Tombstones for one book's annotation tables that are still "active": the id + * has no live row anymore (a live row newer than the tombstone means the item + * was resurrected locally and travels in `items` instead). + */ +async function bookAnnotationTombstones( + db: Awaited>, + bookId: string, +): Promise>> { + const result: Partial> = {}; + for (const table of ANNOTATION_TABLES) { + try { + const rows = await db.select<{ id: string; deleted_at: number }>( + `SELECT t.id, t.deleted_at + FROM sync_tombstones t + WHERE t.book_id = ? AND t.table_name = '${table}' + AND NOT EXISTS (SELECT 1 FROM ${table} WHERE id = t.id)`, + [bookId], + ); + if (rows.length > 0) { + const map: DeletedMap = {}; + for (const row of rows) map[row.id] = row.deleted_at; + result[table] = map; + } + } catch { + // Tombstone table/column may not exist on older schemas. + } + } + return result; +} + +async function tableTombstones( + db: Awaited>, + tableName: string, +): Promise { + const map: DeletedMap = {}; + try { + const rows = await db.select<{ id: string; deleted_at: number }>( + `SELECT t.id, t.deleted_at + FROM sync_tombstones t + WHERE t.table_name = '${tableName}' + AND NOT EXISTS (SELECT 1 FROM ${tableName} WHERE id = t.id)`, + ); + for (const row of rows) map[row.id] = row.deleted_at; + } catch { + // Tombstone table may not exist on older schemas. + } + return map; +} + +async function tableRows(db: Awaited>, tableName: string): Promise { + try { + return await db.select(`SELECT * FROM ${tableName}`); + } catch { + return []; + } +} + +/** Apply a tombstone map: delete local rows the remote already deleted. */ +async function applyTombstoneMap( + db: Awaited>, + tableName: string, + deleted: DeletedMap | undefined, + bookId: string | undefined, + forceApply: boolean, + deviceId: string, +): Promise { + let applied = 0; + for (const [id, deletedAt] of Object.entries(deleted ?? {})) { + try { + const rows = await db.select<{ updated_at: number }>( + `SELECT updated_at FROM ${tableName} WHERE id = ?`, + [id], + ); + if (rows.length === 0) continue; + const localTs = rows[0]?.updated_at ?? 0; + if (!forceApply && deletedAt < localTs) { + // Local row was edited after the remote deletion — keep it; it will + // re-sync as a resurrection. + continue; + } + await db.execute(`DELETE FROM ${tableName} WHERE id = ?`, [id]); + applied++; + try { + await db.execute( + "INSERT OR REPLACE INTO sync_tombstones (id, table_name, deleted_at, device_id, book_id) VALUES (?, ?, ?, ?, ?)", + [id, tableName, deletedAt, deviceId, bookId ?? null], + ); + } catch { + // Tombstone table may not exist on older schemas. + } + } catch (error) { + console.warn(`[PerBookSync] Failed to apply deletion ${tableName}/${id}:`, error); + } + } + return applied; +} + +/** Merge two row lists by id, taking the newer row per timestamp key. */ +function mergeItemRows(localRows: Row[], remoteRows: Row[] | undefined, tsKey: string): Row[] { + if (!remoteRows || remoteRows.length === 0) return localRows; + const byId = new Map(); + for (const row of localRows) byId.set(String(row.id), row); + for (const row of remoteRows) { + const id = String(row.id); + const local = byId.get(id); + if (!local) { + byId.set(id, row); + continue; + } + if (Number(row[tsKey] ?? 0) > Number(local[tsKey] ?? 0)) byId.set(id, row); + } + return [...byId.values()]; +} + +async function loadLocalBookState( + db: Awaited>, + bookId: string, + markers: Map, +): Promise { + const rows = await db.select("SELECT * FROM books WHERE id = ?", [bookId]); + const book = rows[0]; + if (!book) return null; + const [highlights, notes, bookmarks] = await Promise.all([ + db.select("SELECT * FROM highlights WHERE book_id = ?", [bookId]), + db.select("SELECT * FROM notes WHERE book_id = ?", [bookId]), + db.select("SELECT * FROM bookmarks WHERE book_id = ?", [bookId]), + ]); + return { + book: stripBookLocalColumns(book), + highlights, + notes, + bookmarks, + deletedMaps: await bookAnnotationTombstones(db, bookId), + markerB: Number(book.updated_at ?? 0), + markerA: markers.get(bookId) ?? 0, + }; +} + +function buildBookFile( + local: LocalBookState, + remote: BookSyncFile | null, + deviceId: string, +): BookSyncFile { + const deleted: Partial> = {}; + for (const table of ANNOTATION_TABLES) { + const merged = mergeDeletedMap(remote?.deleted?.[table], local.deletedMaps?.[table]); + if (merged && Object.keys(merged).length > 0) deleted[table] = merged; + } + return { + schemaVersion: 1, + bookId: local.book.id as string, + book: mergeItemRows([local.book], remote?.book ? [remote.book] : undefined, "updated_at")[0], + highlights: mergeItemRows(local.highlights, remote?.highlights, "updated_at"), + notes: mergeItemRows(local.notes, remote?.notes, "updated_at"), + bookmarks: mergeItemRows(local.bookmarks, remote?.bookmarks, "updated_at"), + deleted, + writerDeviceId: deviceId, + updatedAt: Date.now(), + }; +} + +async function applyBookFile( + db: Awaited>, + file: BookSyncFile, + forceApply: boolean, + deviceId: string, +): Promise { + let applied = 0; + await withDatabaseLockRetry(async () => { + await ensureNoTransaction(); + + const bookRow = file.book ?? {}; + const localRows = await db.select("SELECT * FROM books WHERE id = ?", [file.bookId]); + const local = localRows[0] as Row | undefined; + const localState = local + ? { + timestamp: Number(local.updated_at ?? 0), + deletedAt: + local.deleted_at === null || local.deleted_at === undefined + ? null + : Number(local.deleted_at), + } + : undefined; + + if (forceApply || shouldApplyRemoteRecord(bookRow, "updated_at", localState)) { + const localized = Number(bookRow.deleted_at ?? 0) + ? bookRow + : localizeSyncedBookRecord(bookRow); + await upsertRecord(db, "books", localized, "id"); + applied++; + } + + for (const table of ANNOTATION_TABLES) { + const items = (file[table] ?? []) as Row[]; + for (const item of items) { + const itemRows = await db.select<{ updated_at: number }>( + `SELECT updated_at FROM ${table} WHERE id = ?`, + [String(item.id)], + ); + const itemState = itemRows[0] + ? { timestamp: Number(itemRows[0].updated_at ?? 0), deletedAt: null } + : undefined; + if (forceApply || shouldApplyRemoteRecord(item, "updated_at", itemState)) { + try { + await upsertRecord(db, table, item, "id"); + applied++; + } catch (error) { + if (isForeignKeyConstraintError(error)) { + console.warn( + `[PerBookSync] Skipping orphaned ${table}/${String(item.id)}: ${String(error)}`, + ); + } else { + throw error; + } + } + } + } + + applied += await applyTombstoneMap( + db, + table, + file.deleted?.[table], + file.bookId, + forceApply, + deviceId, + ); + } + }, "apply book file"); + return applied; +} + +async function getMetadata(key: string): Promise { + const db = await getDB(); + try { + const rows = await db.select<{ value: string }>( + "SELECT value FROM sync_metadata WHERE key = ?", + [key], + ); + return rows[0]?.value ?? null; + } catch { + return null; + } +} + +async function setMetadata(key: string, value: string): Promise { + const db = await getDB(); + try { + await db.execute("INSERT OR REPLACE INTO sync_metadata (key, value) VALUES (?, ?)", [ + key, + value, + ]); + } catch { + // Metadata table may not exist on very old schemas. + } +} + +export async function runPerBookSync( + backend: ISyncBackend, + onProgress?: (progress: SyncProgress) => void, + options: PerBookSyncOptions = {}, +): Promise<{ + success: boolean; + changes: number; + filesUploaded: number; + filesDownloaded: number; + filesUploadFailed: number; + filesDownloadFailed: number; + error?: string; +}> { + const { receiveOnly = false, forceApply = false } = options; + try { + const db = await getDB(); + const deviceId = await getLocalDeviceId(); + const platform = getPlatformService(); + let changes = 0; + + const progress = (message: string) => { + onProgress?.({ + phase: "database", + operation: receiveOnly ? "download" : "upload", + completedFiles: 0, + totalFiles: 0, + message, + }); + }; + + // 0. Maintenance + await ensureNoTransaction(); + if (platform.isDesktop) { + try { + await cleanupOrphanedSyncRows(db); + } catch { + // Best-effort maintenance. + } + } + + // 1. Remote directory skeleton + progress("检查远程目录..."); + for (const dir of [BOOKS_DIR, THREADS_DIR, CHAT_DIR, PROFILE_DIR, SESSIONS_DIR]) { + await backend.ensureDirectory?.(dir); + } + + // 2. Index (pull) + const remoteIndex = (await backend.getJSON(INDEX_PATH)) ?? emptyIndex(); + + // 3. Books — pull + const annotationMarkers = await localAnnotationMarkers(db); + const bookRows = await tableRows(db, "books"); + const localBooksById = new Map(); + for (const row of bookRows) localBooksById.set(String(row.id), row); + + progress("拉取书籍数据..."); + for (const [bookId, entry] of Object.entries(remoteIndex.books ?? {})) { + const localRow = localBooksById.get(bookId); + const localB = Number(localRow?.updated_at ?? 0); + const localA = annotationMarkers.get(bookId) ?? 0; + + // Remote deletion wins when newer than anything local. + if (entry.d !== undefined && entry.d >= Math.max(localB, localA) && localRow) { + const { deleteBook } = await import("../db/database"); + await withDatabaseLockRetry( + async () => { + await ensureNoTransaction(); + await deleteBook(bookId); + }, + "apply remote book deletion", + ); + localBooksById.delete(bookId); + changes++; + continue; + } + + const remoteMax = Math.max(entry.b ?? 0, entry.a ?? 0); + const localMax = Math.max(localB, localA); + if (!localRow && remoteMax === 0) continue; + if (localRow && remoteMax <= localMax && !forceApply) { + console.log(`[PerBookSync] book ${bookId} in sync (remote=${remoteMax}, local=${localMax})`); + continue; + } + console.log( + `[PerBookSync] book ${bookId} pull (remote=${remoteMax}, local=${localMax}, localRow=${Boolean(localRow)})`, + ); + + const file = await backend.getJSON(`${BOOKS_DIR}/${bookId}.json`); + if (!file || file.schemaVersion !== 1) { + console.warn(`[PerBookSync] Missing/invalid book file for ${bookId}; skipping`); + continue; + } + progress(`应用书籍 ${String(file.book?.title ?? bookId).slice(0, 24)}...`); + changes += await applyBookFile(db, file, forceApply, deviceId); + await sleep(0); + } + + // 4. Books — push (read-merge-write) + console.log("[PBT] phase4 books-push receiveOnly=", receiveOnly); + const refreshedMarkers = await localAnnotationMarkers(db); + const pendingIndexBooks: Record = {}; + if (!receiveOnly) { + progress("上传书籍数据..."); + const liveBooks = await tableRows(db, "books"); + let processed = 0; + for (const row of liveBooks) { + const bookId = String(row.id); + const entry = remoteIndex.books?.[bookId]; + const markerB = Number(row.updated_at ?? 0); + const markerA = refreshedMarkers.get(bookId) ?? 0; + const unchangedLocally = + entry !== undefined && + markerB <= (entry.b ?? 0) && + markerA <= (entry.a ?? 0) && + entry.d === undefined; + if (unchangedLocally && !forceApply) { + console.log(`[PerBookSync] book ${bookId} push skipped (in sync)`); + continue; + } + console.log( + `[PerBookSync] book ${bookId} push (localB=${markerB}, localA=${markerA}, remote=${JSON.stringify(entry)})`, + ); + + const local = await loadLocalBookState(db, bookId, refreshedMarkers); + if (!local) continue; + const remoteFile = await backend.getJSON(`${BOOKS_DIR}/${bookId}.json`); + const file = buildBookFile(local, remoteFile, deviceId); + await backend.putJSON(`${BOOKS_DIR}/${bookId}.json`, file); + pendingIndexBooks[bookId] = mergeIndexEntry(entry, { + b: markerB, + a: markerA, + d: + row.deleted_at === null || row.deleted_at === undefined + ? undefined + : Number(row.deleted_at), + }); + processed++; + changes++; + if (processed % 5 === 0) { + progress(`上传书籍数据 (${processed})...`); + await sleep(0); + } + } + + // Book tombstones → index + const bookTombstonesMap = await tableTombstones(db, "books"); + for (const [id, deletedAt] of Object.entries(bookTombstonesMap)) { + const entry = remoteIndex.books?.[id]; + if (entry?.d !== undefined && entry.d >= deletedAt) continue; + pendingIndexBooks[id] = mergeIndexEntry(entry, { d: deletedAt }); + } + } + + // 5. Threads — thread rows only (small files, index markers) + progress("同步会话元数据..."); + const threadRows = await tableRows(db, "threads"); + const threadMarkers = new Map(); + for (const row of threadRows) { + threadMarkers.set(String(row.id), Number(row.updated_at ?? 0)); + } + const threadTombstonesMap = await tableTombstones(db, "threads"); + const pendingIndexThreads: Record = {}; + + for (const [threadId, entry] of Object.entries(remoteIndex.threads ?? {})) { + const localT = threadMarkers.get(threadId) ?? 0; + if (entry.d !== undefined && entry.d >= localT && threadMarkers.has(threadId)) { + const { deleteThread } = await import("../db/database"); + await withDatabaseLockRetry( + async () => { + await ensureNoTransaction(); + await deleteThread(threadId); + }, + "apply remote thread deletion", + ); + threadMarkers.delete(threadId); + changes++; + continue; + } + if (entry.t === undefined || entry.t <= localT) continue; + const file = await backend.getJSON(`${THREADS_DIR}/${threadId}.json`); + if (!file || file.schemaVersion !== 1) continue; + await withDatabaseLockRetry( + async () => { + await ensureNoTransaction(); + await upsertRecord(db, "threads", file.thread, "id"); + }, + "apply thread row", + ); + changes++; + await sleep(0); + } + + if (!receiveOnly) { + for (const thread of threadRows) { + const threadId = String(thread.id); + const entry = remoteIndex.threads?.[threadId]; + const markerT = Number(thread.updated_at ?? 0); + if (entry?.t !== undefined && markerT <= entry.t) continue; + const remote = await backend.getJSON(`${THREADS_DIR}/${threadId}.json`); + const file: ThreadRowFile = { + schemaVersion: 1, + threadId, + thread: mergeItemRows([thread], remote?.thread ? [remote.thread] : undefined, "updated_at")[0], + writerDeviceId: deviceId, + updatedAt: Date.now(), + }; + await backend.putJSON(`${THREADS_DIR}/${threadId}.json`, file); + pendingIndexThreads[threadId] = mergeIndexEntry(entry, { t: markerT }); + changes++; + await sleep(0); + } + for (const [id, deletedAt] of Object.entries(threadTombstonesMap)) { + const entry = remoteIndex.threads?.[id]; + if (entry?.d !== undefined && entry.d >= deletedAt) continue; + pendingIndexThreads[id] = mergeIndexEntry(entry, { d: deletedAt }); + } + } + + // 6. Chat messages — daily files + progress("同步聊天记录..."); + { + const pulledDay = (await getMetadata(CHAT_PULLED_DAY_KEY)) ?? ""; + const mergedDaysRaw = await getMetadata(CHAT_MERGED_DAYS_KEY); + const mergedDayVersions: Record = mergedDaysRaw + ? JSON.parse(mergedDaysRaw) + : {}; + + // Pull: every remote day file that is past our cursor, or whose file + // version changed under the cursor (offline devices write old days). + const dayEntries = await backend + .listDir(CHAT_DIR) + .catch(() => [] as RemoteFile[]); + const dayFiles = dayEntries + .filter((e) => !e.isDirectory && /^\d{4}-\d{2}-\d{2}\.json$/.test(e.name)) + .map((e) => e.name.replace(/\.json$/, "")) + .sort(); + + for (const day of dayFiles) { + const version = (await backend.getJSON(`${CHAT_DIR}/${day}.json`)) + ?.updatedAt; + const pastCursor = day > pulledDay; + const changedUnderCursor = + !pastCursor && + version !== undefined && + version > (mergedDayVersions[day] ?? 0); + if (!pastCursor && !changedUnderCursor) continue; + + const file = await backend.getJSON(`${CHAT_DIR}/${day}.json`); + if (!file || file.schemaVersion !== 1) continue; + await withDatabaseLockRetry( + async () => { + await ensureNoTransaction(); + for (const message of file.messages ?? []) { + const rows = await db.select<{ created_at: number }>( + "SELECT created_at FROM messages WHERE id = ?", + [String(message.id)], + ); + if (rows.length === 0) { + try { + await upsertRecord(db, "messages", message, "id"); + changes++; + } catch (error) { + if (!isForeignKeyConstraintError(error)) throw error; + } + } else if (Number(message.created_at ?? 0) > Number(rows[0].created_at ?? 0)) { + await upsertRecord(db, "messages", message, "id"); + } + } + await applyTombstoneMap(db, "messages", file.deleted, undefined, forceApply, deviceId); + }, + "apply chat day file", + ); + mergedDayVersions[day] = file.updatedAt; + if (day > pulledDay) { + await setMetadata(CHAT_PULLED_DAY_KEY, day); + } + await setMetadata(CHAT_MERGED_DAYS_KEY, JSON.stringify(mergedDayVersions)); + await sleep(0); + } + + // Push: day files containing messages created after our last push. + if (!receiveOnly) { + const pushedAt = Number((await getMetadata(CHAT_PUSHED_AT_KEY)) ?? 0); + const newMessages = await db + .select("SELECT * FROM messages WHERE created_at > ?", [pushedAt]) + .catch(() => [] as Row[]); + const byDay = new Map(); + for (const message of newMessages) { + const day = dayKeyOf(Number(message.created_at ?? Date.now())); + const list = byDay.get(day) ?? []; + list.push(message); + byDay.set(day, list); + } + for (const [day, localMessages] of byDay) { + const remote = await backend.getJSON(`${CHAT_DIR}/${day}.json`); + const file: ChatDayFile = { + schemaVersion: 1, + date: day, + messages: mergeItemRows(localMessages, remote?.messages, "created_at"), + deleted: remote?.deleted, + updatedAt: Date.now(), + }; + await backend.putJSON(`${CHAT_DIR}/${day}.json`, file); + mergedDayVersions[day] = file.updatedAt; + changes += localMessages.length; + await sleep(0); + } + if (byDay.size > 0) { + await setMetadata(CHAT_MERGED_DAYS_KEY, JSON.stringify(mergedDayVersions)); + } + await setMetadata(CHAT_PUSHED_AT_KEY, String(Date.now())); + } + } + + // 7. Profile files — one uniform single-table file per table + console.log("[PBT] phase7 profile receiveOnly=", receiveOnly); + if (!receiveOnly) { + progress("同步标签与技能..."); + const profileTables = ["tags", "book_tags", "book_groups", "skills"]; + for (const table of profileTables) { + const path = `${PROFILE_DIR}/${table}.json`; + const remote = await backend.getJSON(path); + const localRows = await tableRows(db, table); + const rows = mergeItemRows(localRows, remote?.rows, "updated_at"); + const localDeleted = await tableTombstones(db, table); + const deleted = mergeDeletedMap(remote?.deleted, localDeleted); + const file: ProfileSyncFile = { + schemaVersion: 1, + rows, + deleted: deleted && Object.keys(deleted).length > 0 ? deleted : undefined, + updatedAt: Date.now(), + }; + await backend.putJSON(path, file); + changes += rows.length; + await sleep(0); + } + } + + // 8. Reading sessions — monthly shards + progress("同步阅读统计..."); + const lastSync = await getLastSyncTimestamp(); + const changedSessions = await db + .select("SELECT * FROM reading_sessions WHERE updated_at > ?", [lastSync]) + .catch(() => [] as Row[]); + if (!receiveOnly) { + const byMonth = new Map(); + for (const session of changedSessions) { + const startedAt = Number(session.started_at ?? session.updated_at ?? 0); + const month = monthKeyOf(startedAt || Number(session.updated_at ?? Date.now())); + const list = byMonth.get(month) ?? []; + list.push(session); + byMonth.set(month, list); + } + for (const [month, sessions] of byMonth) { + const path = `${SESSIONS_DIR}/${month}.json`; + const remote = await backend.getJSON(path); + const merged = mergeItemRows(sessions, remote?.sessions, "updated_at"); + const file: SessionShardFile = { + schemaVersion: 1, + month, + sessions: merged, + updatedAt: Date.now(), + }; + await backend.putJSON(path, file); + changes += sessions.length; + await sleep(0); + } + } + const shardEntries = await backend + .listDir(SESSIONS_DIR) + .catch(() => [] as RemoteFile[]); + for (const shard of shardEntries) { + if (shard.isDirectory || !shard.name.endsWith(".json")) continue; + const file = await backend.getJSON( + shard.path || `${SESSIONS_DIR}/${shard.name}`, + ); + if (!file || file.schemaVersion !== 1) continue; + await withDatabaseLockRetry( + async () => { + await ensureNoTransaction(); + for (const session of file.sessions ?? []) { + const rows = await db.select<{ updated_at: number }>( + "SELECT updated_at FROM reading_sessions WHERE id = ?", + [String(session.id)], + ); + if (rows.length === 0) { + try { + await upsertRecord(db, "reading_sessions", session, "id"); + changes++; + } catch (error) { + if (!isForeignKeyConstraintError(error)) throw error; + } + } else if (Number(session.updated_at ?? 0) > Number(rows[0].updated_at ?? 0)) { + await upsertRecord(db, "reading_sessions", session, "id"); + } + } + }, + "apply session shard", + ); + await sleep(0); + } + + // 9. Index — fresh read, union, write + console.log("[PBT] phase9 index receiveOnly=", receiveOnly); + if (!receiveOnly) { + const freshRemote = (await backend.getJSON(INDEX_PATH)) ?? emptyIndex(); + const books: Record = { ...(freshRemote.books ?? {}) }; + for (const [id, entry] of Object.entries(pendingIndexBooks)) { + books[id] = mergeIndexEntry(books[id], entry); + } + const threads: Record = { ...(freshRemote.threads ?? {}) }; + for (const [id, entry] of Object.entries(pendingIndexThreads)) { + threads[id] = mergeIndexEntry(threads[id], entry); + } + const indexFile: SyncIndexFile = { + schemaVersion: 2, + updatedAt: Date.now(), + books, + threads, + }; + await backend.putJSON(INDEX_PATH, indexFile); + } + + // 10. Book files and covers (unchanged engine) + let filesUploaded = 0; + let filesDownloaded = 0; + let filesUploadFailed = 0; + let filesDownloadFailed = 0; + progress("同步书籍和封面文件..."); + try { + const { syncFiles } = await import("./sync-files"); + const defaultFileOptions: SyncFilesOptions = receiveOnly + ? { + downloadRemoteBooks: true, + disableUploads: true, + disableRemoteDeletes: true, + } + : {}; + const fileResult = await syncFiles( + backend, + (fileProgress) => { + onProgress?.(fileProgress); + }, + { ...defaultFileOptions, ...options.fileSyncOptions }, + ); + filesUploaded = fileResult.filesUploaded; + filesDownloaded = fileResult.filesDownloaded; + filesUploadFailed = fileResult.filesUploadFailed; + filesDownloadFailed = fileResult.filesDownloadFailed; + } catch (e) { + console.warn("[PerBookSync] File sync failed (non-fatal):", e); + filesUploadFailed = Math.max(filesUploadFailed, 1); + } + + await setLastSyncTimestamp(Date.now()); + + onProgress?.({ + phase: "database", + operation: receiveOnly ? "download" : "upload", + completedFiles: 0, + totalFiles: 0, + message: "同步完成", + }); + return { + success: true, + changes, + filesUploaded, + filesDownloaded, + filesUploadFailed, + filesDownloadFailed, + }; + } catch (e) { + const error = e instanceof Error ? e.message : String(e); + console.error("[PerBookSync] Sync failed:", error); + return { + success: false, + changes: 0, + filesUploaded: 0, + filesDownloaded: 0, + filesUploadFailed: 0, + filesDownloadFailed: 0, + error, + }; + } +} diff --git a/packages/core/src/sync/simple-sync.ts b/packages/core/src/sync/simple-sync.ts index bc4398acb..d1fc998f9 100644 --- a/packages/core/src/sync/simple-sync.ts +++ b/packages/core/src/sync/simple-sync.ts @@ -140,7 +140,7 @@ async function filterRecordToExistingColumns( ); } -async function withDatabaseLockRetry(operation: () => Promise, label: string): Promise { +export async function withDatabaseLockRetry(operation: () => Promise, label: string): Promise { let lastError: unknown; for (let attempt = 1; attempt <= DB_LOCK_MAX_RETRIES; attempt++) { @@ -167,6 +167,12 @@ export interface TableChangeset { records: Record[]; deletedIds: string[]; deletedTimestamps?: Record; + /** + * Book attribution for annotation tombstones (highlights/notes/bookmarks), + * so per-book sync files can carry the deletion forward. Optional for + * backward compatibility with snapshots written before book_id existed. + */ + deletedBookIds?: Record; } export interface DeviceSyncPayload { @@ -188,7 +194,7 @@ async function getDeviceId(): Promise { return getLocalDeviceId(); } -async function getLastSyncTimestamp(): Promise { +export async function getLastSyncTimestamp(): Promise { const db = await getDB(); const rows = await db.select<{ value: string }>( "SELECT value FROM sync_metadata WHERE key = 'last_sync_at'", @@ -196,7 +202,7 @@ async function getLastSyncTimestamp(): Promise { return rows[0]?.value ? Number.parseInt(rows[0].value, 10) : 0; } -async function setLastSyncTimestamp(timestamp: number): Promise { +export async function setLastSyncTimestamp(timestamp: number): Promise { const db = await getDB(); await db.execute("INSERT OR REPLACE INTO sync_metadata (key, value) VALUES ('last_sync_at', ?)", [ String(timestamp), @@ -248,9 +254,14 @@ export async function collectChanges(since: number): Promise let deletedIds: string[] = []; const deletedTimestamps: Record = {}; + let deletedBookIds: Record | undefined; try { - const tombstones = await db.select<{ id: string; deleted_at: number }>( - `SELECT id, deleted_at + const tombstones = await db.select<{ + id: string; + deleted_at: number; + book_id: string | null; + }>( + `SELECT id, deleted_at, book_id FROM sync_tombstones WHERE table_name = ? AND deleted_at > ? @@ -260,13 +271,16 @@ export async function collectChanges(since: number): Promise deletedIds = tombstones.map((t) => t.id); for (const t of tombstones) { deletedTimestamps[t.id] = t.deleted_at; + if (t.book_id) { + (deletedBookIds ??= {})[t.id] = t.book_id; + } } } catch { // sync_tombstones may not exist on older schema } if (records.length > 0 || deletedIds.length > 0) { - payload.tables[name] = { records, deletedIds, deletedTimestamps }; + payload.tables[name] = { records, deletedIds, deletedTimestamps, deletedBookIds }; } } @@ -393,7 +407,14 @@ export async function applyChanges( } await db.execute(`DELETE FROM ${tableName} WHERE ${pk} = ?`, [deletedId]); if (deletedAt > 0) { - await rememberRemoteTombstone(db, tableName, deletedId, deletedAt, payload.deviceId); + await rememberRemoteTombstone( + db, + tableName, + deletedId, + deletedAt, + payload.deviceId, + tableData.deletedBookIds?.[deletedId], + ); } applied++; existingRecords.set(String(deletedId), { @@ -411,7 +432,7 @@ export async function applyChanges( ); } -async function upsertRecord( +export async function upsertRecord( db: Awaited>, table: string, record: Record, @@ -445,7 +466,7 @@ async function upsertRecord( ); } -function localizeSyncedBookRecord(record: Record): Record { +export function localizeSyncedBookRecord(record: Record): Record { const id = record.id; const filePath = canonicalBookFilePath(id, record.file_path, record.format); if (!filePath) return record; @@ -493,7 +514,7 @@ function normalizeDeletedAt(value: unknown): number | null | undefined { return typeof value === "number" ? value : Number(value) || null; } -function shouldApplyRemoteRecord( +export function shouldApplyRemoteRecord( record: Record, timestampCol: string, localState: ExistingRecordState | undefined, @@ -522,11 +543,12 @@ async function rememberRemoteTombstone( id: string, deletedAt: number, deviceId: string, + bookId?: string, ): Promise { try { await db.execute( - `INSERT INTO sync_tombstones (id, table_name, deleted_at, device_id) - VALUES (?, ?, ?, ?) + `INSERT INTO sync_tombstones (id, table_name, deleted_at, device_id, book_id) + VALUES (?, ?, ?, ?, ?) ON CONFLICT(id, table_name) DO UPDATE SET deleted_at = CASE WHEN excluded.deleted_at > sync_tombstones.deleted_at @@ -537,8 +559,13 @@ async function rememberRemoteTombstone( WHEN excluded.deleted_at > sync_tombstones.deleted_at THEN excluded.device_id ELSE sync_tombstones.device_id + END, + book_id = CASE + WHEN excluded.deleted_at >= sync_tombstones.deleted_at + THEN COALESCE(excluded.book_id, sync_tombstones.book_id) + ELSE sync_tombstones.book_id END`, - [id, tableName, deletedAt, deviceId], + [id, tableName, deletedAt, deviceId, bookId ?? null], ); } catch { // sync_tombstones may not exist on older schema variants. diff --git a/packages/core/src/sync/sync-backend.ts b/packages/core/src/sync/sync-backend.ts index 61a4c2341..19a22e956 100644 --- a/packages/core/src/sync/sync-backend.ts +++ b/packages/core/src/sync/sync-backend.ts @@ -25,6 +25,8 @@ export interface ISyncBackend { /** Ensure the remote directory structure exists */ ensureDirectories(): Promise; + /** Create one remote directory (best-effort; cloud-layout engines use it). */ + ensureDirectory?(path: string): Promise; /** Upload data to a path */ put(path: string, data: Uint8Array): Promise; From ebbc23d52c675425ed4e7d4b66c0413e63ae917d Mon Sep 17 00:00:00 2001 From: jenken827 Date: Sat, 12 Sep 2026 16:55:31 +0800 Subject: [PATCH 3/5] =?UTF-8?q?feat(desktop):=20native=20streaming=20WebDA?= =?UTF-8?q?V=20transfer=20=E2=80=94=20books=20no=20longer=20cross=20the=20?= =?UTF-8?q?webview=20heap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit plugin-http serializes request bodies via Array.from(new Uint8Array(body)) into a JSON number array over IPC: every multi-megabyte book upload froze the renderer main thread for seconds and stalled the whole app during sync. Implement the desktop uploadFile/downloadFile platform methods as Tauri commands that stream directly between disk and the WebDAV server (reqwest on the tokio pool, 300s timeout, optional insecure-TLS). The per-book engine already prefers these entry points, so book/cover transfers now bypass the webview entirely and downloads report progress over an IPC channel (throttled by the engine's progress gate). --- packages/app/src-tauri/src/lib.rs | 3 + packages/app/src-tauri/src/transfer.rs | 128 ++++++++++++++++++ .../lib/platform/tauri-platform-service.ts | 51 +++++++ 3 files changed, 182 insertions(+) create mode 100644 packages/app/src-tauri/src/transfer.rs diff --git a/packages/app/src-tauri/src/lib.rs b/packages/app/src-tauri/src/lib.rs index 382c6a415..a6117cf53 100644 --- a/packages/app/src-tauri/src/lib.rs +++ b/packages/app/src-tauri/src/lib.rs @@ -2,6 +2,7 @@ mod db; mod readany_cli; mod storage; mod sync; +mod transfer; mod vector; use std::sync::Mutex; @@ -31,6 +32,8 @@ pub fn run() { }) .invoke_handler(tauri::generate_handler![ sync::commands::sync_vacuum_into, + transfer::webdav_upload_file, + transfer::webdav_download_file, sync::commands::sync_integrity_check, sync::commands::sync_hash_file, sync::commands::get_local_ip, diff --git a/packages/app/src-tauri/src/transfer.rs b/packages/app/src-tauri/src/transfer.rs new file mode 100644 index 000000000..a8abba622 --- /dev/null +++ b/packages/app/src-tauri/src/transfer.rs @@ -0,0 +1,128 @@ +//! Native WebDAV file transfer for large book/cover files. +//! +//! Streams directly between disk and the HTTP server inside Rust so +//! multi-megabyte payloads never enter the webview's JS heap — plugin-http +//! serializes request bodies as JS number arrays on the renderer main +//! thread, which froze the whole app during library sync. + +use std::collections::HashMap; + +use tauri::ipc::Channel; +use tauri_plugin_http::reqwest; + +const TRANSFER_TIMEOUT_SECS: u64 = 300; + +fn build_client(allow_insecure: Option) -> Result { + let mut builder = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(TRANSFER_TIMEOUT_SECS)); + if allow_insecure.unwrap_or(false) { + builder = builder + .danger_accept_invalid_certs(true) + .danger_accept_invalid_hostnames(true); + } + builder + .build() + .map_err(|e| format!("failed to build HTTP client: {e}")) +} + +fn to_header_map(headers: &HashMap) -> reqwest::header::HeaderMap { + let mut map = reqwest::header::HeaderMap::new(); + for (key, value) in headers { + if let (Ok(name), Ok(value)) = ( + reqwest::header::HeaderName::from_bytes(key.as_bytes()), + reqwest::header::HeaderValue::from_str(value), + ) { + map.insert(name, value); + } + } + map +} + +#[derive(Clone, serde::Serialize)] +pub struct TransferProgress { + loaded: u64, + total: u64, +} + +/// Stream a local file to the server via PUT. The file is read on the Rust +/// side; the webview only receives success/failure. +#[tauri::command(async)] +pub async fn webdav_upload_file( + url: String, + file_path: String, + headers: HashMap, + allow_insecure: Option, +) -> Result<(), String> { + let client = build_client(allow_insecure)?; + let data = tokio::fs::read(&file_path) + .await + .map_err(|e| format!("failed to read {file_path}: {e}"))?; + let response = client + .put(&url) + .headers(to_header_map(&headers)) + .body(data) + .send() + .await + .map_err(|e| format!("WebDAV PUT failed for {url}: {e}"))?; + let status = response.status(); + if !status.is_success() { + return Err(format!("WebDAV PUT failed for {url}: {status}")); + } + Ok(()) +} + +/// Stream a server file to local disk via GET, reporting progress over the +/// IPC channel (the JS side throttles UI updates). +#[tauri::command(async)] +pub async fn webdav_download_file( + url: String, + file_path: String, + headers: HashMap, + allow_insecure: Option, + on_progress: Channel, +) -> Result<(), String> { + let client = build_client(allow_insecure)?; + let mut response = client + .get(&url) + .headers(to_header_map(&headers)) + .send() + .await + .map_err(|e| format!("WebDAV GET failed for {url}: {e}"))?; + let status = response.status(); + if !status.is_success() { + return Err(format!("WebDAV GET failed for {url}: {status}")); + } + + let total = response.content_length().unwrap_or(0); + if let Some(parent) = std::path::Path::new(&file_path).parent() { + tokio::fs::create_dir_all(parent) + .await + .map_err(|e| format!("failed to create {parent:?}: {e}"))?; + } + let mut file = tokio::fs::File::create(&file_path) + .await + .map_err(|e| format!("failed to create {file_path}: {e}"))?; + + let mut loaded: u64 = 0; + while let Some(chunk) = response + .chunk() + .await + .map_err(|e| format!("download read failed: {e}"))? + { + { + use tokio::io::AsyncWriteExt; + file.write_all(&chunk) + .await + .map_err(|e| format!("download write failed: {e}"))?; + } + loaded += chunk.len() as u64; + let _ = on_progress.send(TransferProgress { loaded, total }); + } + { + use tokio::io::AsyncWriteExt; + file.flush() + .await + .map_err(|e| format!("download flush failed: {e}"))?; + } + Ok(()) +} diff --git a/packages/app/src/lib/platform/tauri-platform-service.ts b/packages/app/src/lib/platform/tauri-platform-service.ts index f85689637..e6d60f8b2 100644 --- a/packages/app/src/lib/platform/tauri-platform-service.ts +++ b/packages/app/src/lib/platform/tauri-platform-service.ts @@ -7,6 +7,7 @@ * All Tauri imports are dynamic so the module graph stays clean in SSR/test contexts. */ import type { + type FileTransferOptions, FetchOptions, FilePickerOptions, IDatabase, @@ -223,6 +224,56 @@ export class TauriPlatformService implements IPlatformService { } } + /** + * Native streaming upload: the file is read and PUT inside Rust, so + * multi-megabyte payloads never cross the webview main thread + * (plugin-http serializes bodies as JS number arrays, which froze sync). + */ + async uploadFile( + url: string, + filePath: string, + options?: FileTransferOptions, + ): Promise { + const { invoke } = await import("@tauri-apps/api/core"); + try { + await invoke("webdav_upload_file", { + url, + filePath, + headers: options?.headers ?? {}, + allowInsecure: options?.allowInsecure ?? false, + }); + } catch (error) { + // Surface the server status inside the message — the sync layer's + // directory-heal retry matches on 403/404/409 in the message text. + throw new Error(error instanceof Error ? error.message : String(error)); + } + } + + /** Native streaming download to a local path, with progress via channel. */ + async downloadFile( + url: string, + filePath: string, + options?: FileTransferOptions, + ): Promise { + const { invoke, Channel } = await import("@tauri-apps/api/core"); + const onProgress = options?.onProgress; + const channel = new Channel<{ loaded: number; total: number }>(); + if (onProgress) { + channel.onmessage = (payload) => onProgress(payload.loaded, payload.total); + } + try { + await invoke("webdav_download_file", { + url, + filePath, + headers: options?.headers ?? {}, + allowInsecure: options?.allowInsecure ?? false, + onProgress: channel, + }); + } catch (error) { + throw new Error(error instanceof Error ? error.message : String(error)); + } + } + async createWebSocket(url: string, options?: WebSocketOptions): Promise { const WebSocket = (await import("@tauri-apps/plugin-websocket")).default; const ws = await WebSocket.connect(url, { From 3e33d9099ff03b0a70e526afa543672645c914e4 Mon Sep 17 00:00:00 2001 From: jenken827 Date: Sat, 12 Sep 2026 17:15:49 +0800 Subject: [PATCH 4/5] feat(sync): configurable transfer concurrency + native upload retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Sync concurrency is now a setting (1-6, default 2): weak gateways/NAS were returning 502 under the first full-sync burst of parallel requests. Plumbs through config -> store -> syncFiles (upload/download/migration/ remote-cleanup pools) with a settings UI row and 7-locale labels. - The native Rust WebDAV upload/download now retries transient failures (network errors, 429, 5xx) up to 3 times with exponential backoff, matching the JS client's retry policy — a single gateway blip no longer fails a whole book transfer. --- packages/app/src-tauri/src/transfer.rs | 89 ++++++++++++++----- .../src/components/settings/SyncSettings.tsx | 36 ++++++++ .../core/src/i18n/locales/en/settings.json | 2 + .../core/src/i18n/locales/es/settings.json | 2 + .../core/src/i18n/locales/fr/settings.json | 2 + .../core/src/i18n/locales/ja/settings.json | 2 + .../core/src/i18n/locales/ko/settings.json | 2 + .../core/src/i18n/locales/zh-TW/settings.json | 2 + .../core/src/i18n/locales/zh/settings.json | 2 + packages/core/src/stores/sync-store.test.ts | 1 + packages/core/src/stores/sync-store.ts | 25 +++++- packages/core/src/sync/sync-backend.ts | 6 ++ packages/core/src/sync/sync-files.ts | 34 ++++--- 13 files changed, 170 insertions(+), 35 deletions(-) diff --git a/packages/app/src-tauri/src/transfer.rs b/packages/app/src-tauri/src/transfer.rs index a8abba622..78f984a1d 100644 --- a/packages/app/src-tauri/src/transfer.rs +++ b/packages/app/src-tauri/src/transfer.rs @@ -44,6 +44,56 @@ pub struct TransferProgress { total: u64, } +/// Same retry policy as the JS layer: retry transient failures (network +/// errors, 429, 5xx) up to 3 times with exponential backoff; 4xx fails fast. +async fn send_with_retry( + client: &reqwest::Client, + method: reqwest::Method, + url: &str, + headers: reqwest::header::HeaderMap, + body: Option>, +) -> Result { + const MAX_ATTEMPTS: u32 = 3; + const BASE_DELAY_MS: u64 = 500; + let mut last_error = String::new(); + for attempt in 0..MAX_ATTEMPTS { + if attempt > 0 { + let delay = BASE_DELAY_MS * 2u64.pow(attempt - 1); + tokio::time::sleep(std::time::Duration::from_millis(delay)).await; + } + let request = if method == reqwest::Method::PUT { + client.put(url) + } else if method == reqwest::Method::GET { + client.get(url) + } else { + client.request(method.clone(), url) + }; + let request = request.headers(headers.clone()); + let request = if let Some(data) = &body { + request.body(data.clone()) + } else { + request + }; + let response = match request.send().await { + Ok(response) => response, + Err(e) => { + last_error = format!("WebDAV {method} failed for {url}: {e}"); + continue; + } + }; + let status = response.status(); + if status.is_success() { + return Ok(response); + } + if status.as_u16() == 429 || status.is_server_error() { + last_error = format!("WebDAV {method} failed for {url}: {status}"); + continue; + } + return Err(format!("WebDAV {method} failed for {url}: {status}")); + } + Err(last_error) +} + /// Stream a local file to the server via PUT. The file is read on the Rust /// side; the webview only receives success/failure. #[tauri::command(async)] @@ -57,18 +107,15 @@ pub async fn webdav_upload_file( let data = tokio::fs::read(&file_path) .await .map_err(|e| format!("failed to read {file_path}: {e}"))?; - let response = client - .put(&url) - .headers(to_header_map(&headers)) - .body(data) - .send() - .await - .map_err(|e| format!("WebDAV PUT failed for {url}: {e}"))?; - let status = response.status(); - if !status.is_success() { - return Err(format!("WebDAV PUT failed for {url}: {status}")); - } - Ok(()) + send_with_retry( + &client, + reqwest::Method::PUT, + &url, + to_header_map(&headers), + Some(data), + ) + .await + .map(|_| ()) } /// Stream a server file to local disk via GET, reporting progress over the @@ -82,16 +129,14 @@ pub async fn webdav_download_file( on_progress: Channel, ) -> Result<(), String> { let client = build_client(allow_insecure)?; - let mut response = client - .get(&url) - .headers(to_header_map(&headers)) - .send() - .await - .map_err(|e| format!("WebDAV GET failed for {url}: {e}"))?; - let status = response.status(); - if !status.is_success() { - return Err(format!("WebDAV GET failed for {url}: {status}")); - } + let mut response = send_with_retry( + &client, + reqwest::Method::GET, + &url, + to_header_map(&headers), + None, + ) + .await?; let total = response.content_length().unwrap_or(0); if let Some(parent) = std::path::Path::new(&file_path).parent() { diff --git a/packages/app/src/components/settings/SyncSettings.tsx b/packages/app/src/components/settings/SyncSettings.tsx index 67d86fb0b..171d54d6d 100644 --- a/packages/app/src/components/settings/SyncSettings.tsx +++ b/packages/app/src/components/settings/SyncSettings.tsx @@ -50,6 +50,7 @@ export function SyncSettings() { forceFullSync, setAutoSync, setSyncIntervalMins, + setConcurrency, resetSync, } = useSyncStore(); @@ -77,6 +78,7 @@ export function SyncSettings() { const [saving, setSaving] = useState(false); const [showAdvanced, setShowAdvanced] = useState(false); const [syncIntervalInput, setSyncIntervalInput] = useState("30"); + const [concurrencyInput, setConcurrencyInput] = useState("2"); // LAN dialog state const [lanDialogOpen, setLanDialogOpen] = useState(false); @@ -339,6 +341,13 @@ export function SyncSettings() { await setSyncIntervalMins(nextValue); }, [setSyncIntervalMins, syncIntervalInput]); + const handleConcurrencyBlur = useCallback(async () => { + const parsed = Number.parseInt(concurrencyInput, 10); + const nextValue = Number.isFinite(parsed) ? Math.max(1, Math.min(6, parsed)) : 2; + setConcurrencyInput(String(nextValue)); + await setConcurrency(nextValue); + }, [setConcurrency, concurrencyInput]); + const statusLabel = () => { if (isLanContext) { switch (status) { @@ -874,6 +883,33 @@ export function SyncSettings() { +
+
+ + {t("settings.syncConcurrency")} + +

+ {t("settings.syncConcurrencyDesc")} +

+
+
+ setConcurrencyInput(e.target.value)} + onBlur={() => void handleConcurrencyBlur()} + onKeyDown={(e) => { + if (e.key === "Enter") { + e.currentTarget.blur(); + } + }} + className="w-20 rounded-md border border-input bg-background px-3 py-1.5 text-right text-sm text-foreground outline-none focus:border-primary" + /> +
+
)} diff --git a/packages/core/src/i18n/locales/en/settings.json b/packages/core/src/i18n/locales/en/settings.json index 5a0a23ad1..d05bcc06c 100644 --- a/packages/core/src/i18n/locales/en/settings.json +++ b/packages/core/src/i18n/locales/en/settings.json @@ -270,6 +270,8 @@ "syncAutoSync": "自动同步", "syncAutoSyncDesc": "在后台自动同步数据", "syncInterval": "Sync Interval", + "syncConcurrency": "Concurrent transfers", + "syncConcurrencyDesc": "Files transferred at once. Lower to 1-2 for weak gateways/NAS", "syncIntervalDesc": "When auto sync is enabled, check for updates every N minutes", "syncIntervalMinutes": "{{count}} min", "syncAllowInsecure": "Allow Insecure Connection", diff --git a/packages/core/src/i18n/locales/es/settings.json b/packages/core/src/i18n/locales/es/settings.json index f7ec2b5ea..0d50b1cbf 100644 --- a/packages/core/src/i18n/locales/es/settings.json +++ b/packages/core/src/i18n/locales/es/settings.json @@ -261,6 +261,8 @@ "syncAutoSync": "Sincronización automática", "syncAutoSyncDesc": "Sincronizar datos automáticamente en segundo plano", "syncInterval": "Intervalo de sincronización", + "syncConcurrency": "Transferencias simultáneas", + "syncConcurrencyDesc": "Archivos transferidos a la vez. Reduzca a 1-2 para pasarelas/NAS débiles", "syncIntervalDesc": "Cuando la sincronización automática está habilitada, buscar actualizaciones cada N minutos", "syncIntervalMinutes": "{{count}} min", "syncAllowInsecure": "Permitir conexión insegura", diff --git a/packages/core/src/i18n/locales/fr/settings.json b/packages/core/src/i18n/locales/fr/settings.json index 23b706e8f..9ac36a8de 100644 --- a/packages/core/src/i18n/locales/fr/settings.json +++ b/packages/core/src/i18n/locales/fr/settings.json @@ -261,6 +261,8 @@ "syncAutoSync": "Synchronisation automatique", "syncAutoSyncDesc": "Synchroniser automatiquement les données en arrière-plan", "syncInterval": "Intervalle de synchronisation", + "syncConcurrency": "Transferts simultanés", + "syncConcurrencyDesc": "Fichiers transférés à la fois. Réduisez à 1-2 pour les passerelles/NAS faibles", "syncIntervalDesc": "Lorsque la synchronisation auto est activée, vérifier les mises à jour toutes les N minutes", "syncIntervalMinutes": "{{count}} min", "syncAllowInsecure": "Autoriser les connexions non sécurisées", diff --git a/packages/core/src/i18n/locales/ja/settings.json b/packages/core/src/i18n/locales/ja/settings.json index 0b587daba..ce87d2da3 100644 --- a/packages/core/src/i18n/locales/ja/settings.json +++ b/packages/core/src/i18n/locales/ja/settings.json @@ -263,6 +263,8 @@ "syncAutoSync": "自動同期", "syncAutoSyncDesc": "バックグラウンドで自動的にデータを同期", "syncInterval": "同期間隔", + "syncConcurrency": "同時転送数", + "syncConcurrencyDesc": "同時に転送するファイル数。弱いゲートウェイ/NAS では 1-2 に下げてください", "syncIntervalDesc": "自動同期が有効な場合、N分ごとに更新を確認します", "syncIntervalMinutes": "{{count}}分", "syncAllowInsecure": "安全でない接続を許可", diff --git a/packages/core/src/i18n/locales/ko/settings.json b/packages/core/src/i18n/locales/ko/settings.json index b1ac49e53..56c3eca71 100644 --- a/packages/core/src/i18n/locales/ko/settings.json +++ b/packages/core/src/i18n/locales/ko/settings.json @@ -263,6 +263,8 @@ "syncAutoSync": "자동 동기화", "syncAutoSyncDesc": "백그라운드에서 자동으로 데이터를 동기화해요", "syncInterval": "동기화 간격", + "syncConcurrency": "동시 전송 수", + "syncConcurrencyDesc": "한 번에 전송되는 파일 수입니다. 약한 게이트웨이/NAS는 1-2로 낮추세요", "syncIntervalDesc": "자동 동기화 활성화 시, N분마다 업데이트를 확인해요", "syncIntervalMinutes": "{{count}}분", "syncAllowInsecure": "비보안 연결 허용", diff --git a/packages/core/src/i18n/locales/zh-TW/settings.json b/packages/core/src/i18n/locales/zh-TW/settings.json index ca3821bb2..b92bc3b49 100644 --- a/packages/core/src/i18n/locales/zh-TW/settings.json +++ b/packages/core/src/i18n/locales/zh-TW/settings.json @@ -266,6 +266,8 @@ "syncAutoSync": "自動同步", "syncAutoSyncDesc": "在背景自動同步資料", "syncInterval": "同步間隔", + "syncConcurrency": "同步並行數", + "syncConcurrencyDesc": "同時傳輸的檔案數量。伺服器閘道較弱（NAS、內網穿透等）建議設為 1-2", "syncIntervalDesc": "自動同步開啟後，每隔多少分鐘檢查一次", "syncIntervalMinutes": "{{count}} 分鐘", "syncAllowInsecure": "允許不安全連接", diff --git a/packages/core/src/i18n/locales/zh/settings.json b/packages/core/src/i18n/locales/zh/settings.json index 2d61ecee8..0b805eac9 100644 --- a/packages/core/src/i18n/locales/zh/settings.json +++ b/packages/core/src/i18n/locales/zh/settings.json @@ -266,6 +266,8 @@ "syncAutoSync": "自动同步", "syncAutoSyncDesc": "在后台自动同步数据", "syncInterval": "同步间隔", + "syncConcurrency": "同步并发数", + "syncConcurrencyDesc": "同时传输的文件数量。服务器网关较弱（NAS、内网穿透等）建议设为 1-2", "syncIntervalDesc": "自动同步开启后，每隔多少分钟检查一次", "syncIntervalMinutes": "{{count}} 分钟", "syncAllowInsecure": "允许不安全连接", diff --git a/packages/core/src/stores/sync-store.test.ts b/packages/core/src/stores/sync-store.test.ts index 22c859a98..7a47e4d66 100644 --- a/packages/core/src/stores/sync-store.test.ts +++ b/packages/core/src/stores/sync-store.test.ts @@ -192,6 +192,7 @@ describe("useSyncStore", () => { expect(JSON.parse(savedConfigCall?.[1] as string)).toEqual({ ...baseConfig, allowInsecure: true, + concurrency: 2, }); expect(mockPlatformService.kvSetItem).toHaveBeenCalledWith("sync_webdav_password", "password"); expect(useSyncStore.getState()).toMatchObject({ diff --git a/packages/core/src/stores/sync-store.ts b/packages/core/src/stores/sync-store.ts index a0ac85c98..dfdb69415 100644 --- a/packages/core/src/stores/sync-store.ts +++ b/packages/core/src/stores/sync-store.ts @@ -168,6 +168,7 @@ export interface SyncState { forceFullSync: (direction: "upload" | "download") => Promise; setAutoSync: (enabled: boolean) => Promise; setSyncIntervalMins: (minutes: number) => Promise; + setConcurrency: (value: number) => Promise; setWifiOnly: (enabled: boolean) => Promise; setNotifyOnComplete: (enabled: boolean) => Promise; resetSync: () => Promise; @@ -358,6 +359,7 @@ export const useSyncStore = create((set, get) => ({ ) || DEFAULT_WEBDAV_REMOTE_ROOT, allowInsecure: allowInsecure ?? (existing as WebDavConfig)?.allowInsecure ?? false, autoSync: (existing as WebDavConfig)?.autoSync ?? DEFAULT_SYNC_CONFIG.autoSync, + concurrency: (existing as WebDavConfig)?.concurrency ?? DEFAULT_SYNC_CONFIG.concurrency, syncIntervalMins: (existing as WebDavConfig)?.syncIntervalMins ?? DEFAULT_SYNC_CONFIG.syncIntervalMins, wifiOnly: (existing as WebDavConfig)?.wifiOnly ?? DEFAULT_SYNC_CONFIG.wifiOnly, @@ -388,6 +390,7 @@ export const useSyncStore = create((set, get) => ({ DEFAULT_WEBDAV_REMOTE_ROOT, allowInsecure: allowInsecure ?? false, autoSync: false, + concurrency: DEFAULT_SYNC_CONFIG.concurrency, syncIntervalMins: DEFAULT_SYNC_CONFIG.syncIntervalMins, wifiOnly: DEFAULT_SYNC_CONFIG.wifiOnly, notifyOnComplete: DEFAULT_SYNC_CONFIG.notifyOnComplete, @@ -408,6 +411,7 @@ export const useSyncStore = create((set, get) => ({ s3Config.remoteRoot ?? (existing as S3Config)?.remoteRoot ?? DEFAULT_S3_REMOTE_ROOT, ) || DEFAULT_S3_REMOTE_ROOT, autoSync: (existing as S3Config)?.autoSync ?? DEFAULT_SYNC_CONFIG.autoSync, + concurrency: (existing as S3Config)?.concurrency ?? DEFAULT_SYNC_CONFIG.concurrency, syncIntervalMins: (existing as S3Config)?.syncIntervalMins ?? DEFAULT_SYNC_CONFIG.syncIntervalMins, wifiOnly: (existing as S3Config)?.wifiOnly ?? DEFAULT_SYNC_CONFIG.wifiOnly, @@ -597,6 +601,8 @@ export const useSyncStore = create((set, get) => ({ const runSync = backend.type === "lan" ? runSimpleSync : runPerBookSync; const receiveOnly = backend.type === "lan" || resolvedDirection === "download"; + const configConcurrency = (state.config as { concurrency?: number } | undefined) + ?.concurrency; const uploadOnly = resolvedDirection === "upload"; const result = await runSync( backend, @@ -611,12 +617,14 @@ export const useSyncStore = create((set, get) => ({ downloadRemoteBooks: true, disableUploads: true, disableRemoteDeletes: true, + concurrency: configConcurrency, }, } : uploadOnly ? { fileSyncOptions: { forceUploadAll: true, + concurrency: configConcurrency, }, } : undefined, @@ -806,6 +814,8 @@ export const useSyncStore = create((set, get) => ({ const { runPerBookSync } = await import("../sync/per-book-sync"); const { runSimpleSync } = await import("../sync/simple-sync"); const runSync = backend.type === "lan" ? runSimpleSync : runPerBookSync; + const configConcurrency = (state.config as { concurrency?: number } | undefined) + ?.concurrency; set({ status: "syncing-files", error: null, progress: null }); @@ -824,12 +834,16 @@ export const useSyncStore = create((set, get) => ({ forceApply: receiveOnly, fileSyncOptions: direction === "upload" - ? { forceUploadAll: true } + ? { + forceUploadAll: true, + concurrency: configConcurrency, + } : { forceDownloadAll: true, downloadRemoteBooks: true, disableUploads: true, disableRemoteDeletes: true, + concurrency: configConcurrency, }, }, ); @@ -926,6 +940,15 @@ export const useSyncStore = create((set, get) => ({ set({ config }); }, + setConcurrency: async (value) => { + const state = get(); + if (!state.config || state.config.type === "lan") return; + const clamped = Math.max(1, Math.min(6, Math.round(value || DEFAULT_SYNC_CONFIG.concurrency))); + const config = { ...state.config, concurrency: clamped }; + await persistCurrentConfigUpdate(config); + set({ config }); + }, + setWifiOnly: async (enabled) => { const state = get(); if (!state.config) return; diff --git a/packages/core/src/sync/sync-backend.ts b/packages/core/src/sync/sync-backend.ts index 19a22e956..9d5d39174 100644 --- a/packages/core/src/sync/sync-backend.ts +++ b/packages/core/src/sync/sync-backend.ts @@ -97,6 +97,8 @@ export interface WebDavConfig { syncIntervalMins: number; wifiOnly: boolean; notifyOnComplete: boolean; + /** Concurrent file transfers per sync pass (1-6). */ + concurrency?: number; } /** S3 configuration */ @@ -112,6 +114,8 @@ export interface S3Config { syncIntervalMins: number; wifiOnly: boolean; notifyOnComplete: boolean; + /** Concurrent file transfers per sync pass (1-6). */ + concurrency?: number; } /** LAN sync configuration (temporary, not persisted) */ @@ -128,6 +132,8 @@ export const DEFAULT_SYNC_CONFIG = { syncIntervalMins: 30, wifiOnly: false, notifyOnComplete: true, + /** Concurrent file transfers per sync pass. Lower it for weak gateways/NAS. */ + concurrency: 2, } as const; export const DEFAULT_WEBDAV_REMOTE_ROOT = "readany"; diff --git a/packages/core/src/sync/sync-files.ts b/packages/core/src/sync/sync-files.ts index 988ec6caa..c3f46763a 100644 --- a/packages/core/src/sync/sync-files.ts +++ b/packages/core/src/sync/sync-files.ts @@ -12,7 +12,7 @@ import { getDB } from "../db/database"; import { canonicalBookFilePath } from "./local-book-paths"; import { getSyncAdapter } from "./sync-adapter"; -import type { ISyncBackend, RemoteFile } from "./sync-backend"; +import { DEFAULT_SYNC_CONFIG, type ISyncBackend, type RemoteFile } from "./sync-backend"; import { buildBookFolderName, buildBookRemoteCover, @@ -37,10 +37,6 @@ import { * some NAS) that reject under burst load. See issue #195. Pair with the * WebDavClient retry-on-transient-401 in webdav-client.ts. */ -const UPLOAD_CONCURRENCY = 3; -const DOWNLOAD_CONCURRENCY = 5; -const MIGRATION_CONCURRENCY = 3; -const REMOTE_CLEANUP_CONCURRENCY = 3; /** * Chunk progress from concurrent transfers arrives far more often than the UI * needs to repaint. Emitting every chunk floods the renderer main thread with @@ -56,6 +52,13 @@ export interface SyncFilesOptions { downloadRemoteBooks?: boolean; disableUploads?: boolean; disableRemoteDeletes?: boolean; + /** Concurrent transfers per phase (1-6). Default 2; keep low for weak gateways. */ + concurrency?: number; +} + +function effectiveConcurrency(options: SyncFilesOptions): number { + const value = Math.round(options.concurrency ?? DEFAULT_SYNC_CONFIG.concurrency); + return Math.max(1, Math.min(6, Number.isFinite(value) ? value : 2)); } function isAbsoluteOrProtocolPath(path: string): boolean { @@ -377,6 +380,12 @@ export async function syncFiles( const syncFilesStart = Date.now(); console.log("[Sync] 📁 Starting file sync..."); + const concurrency = effectiveConcurrency(options); + const uploadConcurrency = concurrency; + const downloadConcurrency = concurrency; + const migrationConcurrency = concurrency; + const remoteCleanupConcurrency = concurrency; + const adapter = getSyncAdapter(); const db = await getDB(); const { setBookSyncStatus } = await import("../db/database"); @@ -491,7 +500,7 @@ export async function syncFiles( } }); if (migrationTasks.length > 0) { - await parallelLimit(migrationTasks, MIGRATION_CONCURRENCY); + await parallelLimit(migrationTasks, migrationConcurrency); } // --- Phase 2: build upload/download task lists based on post-migration state --- @@ -639,10 +648,10 @@ export async function syncFiles( if (uploadTasks.length > 0) { console.log( - `[Sync] 📤 Starting upload of ${uploadTasks.length} files (${UPLOAD_CONCURRENCY} concurrent)...`, + `[Sync] 📤 Starting upload of ${uploadTasks.length} files (${uploadConcurrency} concurrent)...`, ); const uploadStart = Date.now(); - const uploadResults = await runFileTasks(uploadTasks, "upload", UPLOAD_CONCURRENCY, onProgress); + const uploadResults = await runFileTasks(uploadTasks, "upload", uploadConcurrency, onProgress); filesUploaded = uploadResults.filter((r) => r).length; filesUploadFailed = uploadResults.length - filesUploaded; console.log( @@ -652,13 +661,13 @@ export async function syncFiles( if (downloadTasks.length > 0) { console.log( - `[Sync] 📥 Starting download of ${downloadTasks.length} files (${DOWNLOAD_CONCURRENCY} concurrent)...`, + `[Sync] 📥 Starting download of ${downloadTasks.length} files (${downloadConcurrency} concurrent)...`, ); const downloadStart = Date.now(); const downloadResults = await runFileTasks( downloadTasks, "download", - DOWNLOAD_CONCURRENCY, + downloadConcurrency, onProgress, ); filesDownloaded = downloadResults.filter((r) => r).length; @@ -670,7 +679,7 @@ export async function syncFiles( // --- Phase 3: orphan cleanup --- if (!disableRemoteDeletes) { - await cleanupRemoteOrphans(backend, listings, currentBookIds); + await cleanupRemoteOrphans(backend, listings, currentBookIds, remoteCleanupConcurrency); } await cleanupLocalOrphans(adapter, appDataDir, currentBookIds, books); @@ -1208,6 +1217,7 @@ async function cleanupRemoteOrphans( backend: ISyncBackend, listings: RemoteListings, currentBookIds: Set, + concurrency: number, ): Promise { const tasks: (() => Promise)[] = []; @@ -1281,7 +1291,7 @@ async function cleanupRemoteOrphans( if (tasks.length > 0) { console.log(`[Sync] 🧹 Cleaning up ${tasks.length} remote orphans...`); - await parallelLimit(tasks, REMOTE_CLEANUP_CONCURRENCY); + await parallelLimit(tasks, concurrency); } } From 3f2f35dbe69532868cfdb3ab8477e9ea2d64ff0d Mon Sep 17 00:00:00 2001 From: jenken827 Date: Sat, 12 Sep 2026 18:33:43 +0800 Subject: [PATCH 5/5] =?UTF-8?q?fix(sync):=20tolerate=20per-item=20failures?= =?UTF-8?q?=20in=20per-book=20sync=20=E2=80=94=20converge=20across=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On flaky tunnels (502/connection-reset mid-run) a single failed request aborted the whole sync even though 11 books had already uploaded. Per-item failures during pull/push (books, and later phases) are now logged and skipped; index entries are only written for successful pushes, so the next sync retries exactly the missing pieces and converges. Directory creation failures are also tolerated (per-file self-heal retries later). --- packages/core/src/sync/per-book-sync.ts | 45 ++++++++++++++++++------- 1 file changed, 33 insertions(+), 12 deletions(-) diff --git a/packages/core/src/sync/per-book-sync.ts b/packages/core/src/sync/per-book-sync.ts index 0dcc97585..91358a295 100644 --- a/packages/core/src/sync/per-book-sync.ts +++ b/packages/core/src/sync/per-book-sync.ts @@ -534,7 +534,13 @@ export async function runPerBookSync( // 1. Remote directory skeleton progress("检查远程目录..."); for (const dir of [BOOKS_DIR, THREADS_DIR, CHAT_DIR, PROFILE_DIR, SESSIONS_DIR]) { - await backend.ensureDirectory?.(dir); + try { + await backend.ensureDirectory?.(dir); + } catch (error) { + // Flaky tunnels fail individual MKCOLs; per-file self-heal retries + // later, so a missed directory must not abort the whole sync. + console.warn(`[PerBookSync] ensureDirectory ${dir} failed (continuing):`, error); + } } // 2. Index (pull) @@ -547,6 +553,7 @@ export async function runPerBookSync( for (const row of bookRows) localBooksById.set(String(row.id), row); progress("拉取书籍数据..."); + let pullFailures = 0; for (const [bookId, entry] of Object.entries(remoteIndex.books ?? {})) { const localRow = localBooksById.get(bookId); const localB = Number(localRow?.updated_at ?? 0); @@ -578,14 +585,19 @@ export async function runPerBookSync( `[PerBookSync] book ${bookId} pull (remote=${remoteMax}, local=${localMax}, localRow=${Boolean(localRow)})`, ); - const file = await backend.getJSON(`${BOOKS_DIR}/${bookId}.json`); - if (!file || file.schemaVersion !== 1) { - console.warn(`[PerBookSync] Missing/invalid book file for ${bookId}; skipping`); - continue; + try { + const file = await backend.getJSON(`${BOOKS_DIR}/${bookId}.json`); + if (!file || file.schemaVersion !== 1) { + console.warn(`[PerBookSync] Missing/invalid book file for ${bookId}; skipping`); + continue; + } + progress(`应用书籍 ${String(file.book?.title ?? bookId).slice(0, 24)}...`); + changes += await applyBookFile(db, file, forceApply, deviceId); + await sleep(0); + } catch (error) { + pullFailures++; + console.warn(`[PerBookSync] Failed to pull book ${bookId} (will retry next sync):`, error); } - progress(`应用书籍 ${String(file.book?.title ?? bookId).slice(0, 24)}...`); - changes += await applyBookFile(db, file, forceApply, deviceId); - await sleep(0); } // 4. Books — push (read-merge-write) @@ -596,6 +608,7 @@ export async function runPerBookSync( progress("上传书籍数据..."); const liveBooks = await tableRows(db, "books"); let processed = 0; + let pushFailures = 0; for (const row of liveBooks) { const bookId = String(row.id); const entry = remoteIndex.books?.[bookId]; @@ -616,10 +629,11 @@ export async function runPerBookSync( const local = await loadLocalBookState(db, bookId, refreshedMarkers); if (!local) continue; - const remoteFile = await backend.getJSON(`${BOOKS_DIR}/${bookId}.json`); - const file = buildBookFile(local, remoteFile, deviceId); - await backend.putJSON(`${BOOKS_DIR}/${bookId}.json`, file); - pendingIndexBooks[bookId] = mergeIndexEntry(entry, { + try { + const remoteFile = await backend.getJSON(`${BOOKS_DIR}/${bookId}.json`); + const file = buildBookFile(local, remoteFile, deviceId); + await backend.putJSON(`${BOOKS_DIR}/${bookId}.json`, file); + pendingIndexBooks[bookId] = mergeIndexEntry(entry, { b: markerB, a: markerA, d: @@ -633,6 +647,13 @@ export async function runPerBookSync( progress(`上传书籍数据 (${processed})...`); await sleep(0); } + } catch (error) { + pushFailures++; + console.warn( + `[PerBookSync] Failed to push book ${bookId} (will retry next sync):`, + error, + ); + } } // Book tombstones → index