feat(sync): stream Google Drive book uploads/downloads from disk (#4824)

Add uploadStream + downloadStream to the Google Drive provider so book files sync straight from/to disk instead of buffering the whole file in the JS heap. Marshaling a large book across the WebView<->Rust bridge as a single Uint8Array crashes the renderer on mobile, so book sync over Drive was effectively desktop-only; this unlocks it on Android/iOS and keeps the heap flat for gigabyte-scale PDFs on desktop too. - driveRest.ts: resumableCreateUrl / resumableUpdateUrl builders. - GoogleDriveProvider: uploadStream opens a Drive resumable session (POST new / PATCH existing; metadata in the initiation, so no reparent follow-up), then PUTs the bytes to the one-time session URI via the native upload plugin (tauriUpload). downloadStream GETs alt=media to disk via tauriDownload with a bearer token. Attached on Tauri only; web keeps the buffered fallback. Both swallow to false per the provider contract (engine retries once). Reuses @tauri-apps/plugin-upload already shipped for WebDAV streaming; no new native code. Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 22:59:35 +08:00
parent 324bb8a366
commit 531f0b58ae
4 changed files with 326 additions and 4 deletions
@@ -0,0 +1,152 @@
+import { beforeEach, describe, expect, test, vi } from 'vitest';
+
+// Streaming is Tauri-only (it shells the bytes through the native upload plugin
+// off the disk); force the platform probe on so the provider attaches the
+// streaming methods, and stub the native transfer plugin.
+vi.mock('@/services/environment', () => ({ isTauriAppPlatform: () => true }));
+vi.mock('@/utils/transfer', () => ({
+  tauriUpload: vi.fn(async () => '{"id":"NID"}'),
+  tauriDownload: vi.fn(async () => ({})),
+}));
+
+import {
+  createGoogleDriveProvider,
+  type DriveAuth,
+  type FetchFn,
+} from '@/services/sync/providers/gdrive/GoogleDriveProvider';
+import { tauriUpload, tauriDownload } from '@/utils/transfer';
+
+const auth: DriveAuth = { getAccessToken: async () => 'TOKEN' };
+
+const json = (body: unknown, status = 200): Response =>
+  new Response(JSON.stringify(body), { status, headers: { 'content-type': 'application/json' } });
+const folder = (id: string) => ({ id, mimeType: 'application/vnd.google-apps.folder' });
+const session = (location: string | null): Response =>
+  new Response(JSON.stringify({}), {
+    status: 200,
+    headers: location ? { Location: location } : {},
+  });
+
+interface Harness {
+  provider: ReturnType<typeof createGoogleDriveProvider>;
+  fetchMock: ReturnType<typeof vi.fn>;
+  url: (n: number) => string;
+  method: (n: number) => string | undefined;
+  body: (n: number) => unknown;
+}
+
+const makeDrive = (): Harness => {
+  const fetchMock = vi.fn();
+  const provider = createGoogleDriveProvider(auth, fetchMock as unknown as FetchFn, {
+    sleep: async () => {},
+  });
+  return {
+    provider,
+    fetchMock,
+    url: (n) => fetchMock.mock.calls[n]?.[0] as string,
+    method: (n) => (fetchMock.mock.calls[n]?.[1] as RequestInit | undefined)?.method,
+    body: (n) => {
+      const raw = (fetchMock.mock.calls[n]?.[1] as RequestInit | undefined)?.body;
+      return typeof raw === 'string' ? JSON.parse(raw) : undefined;
+    },
+  };
+};
+
+// Resolve the three folder segments of /Readest/books/h to existing folders.
+const stageFolders = (h: Harness) => {
+  h.fetchMock
+    .mockResolvedValueOnce(json({ files: [folder('RID')] })) // Readest
+    .mockResolvedValueOnce(json({ files: [folder('BID')] })) // books
+    .mockResolvedValueOnce(json({ files: [folder('HID')] })); // h
+};
+
+const BOOK = '/Readest/books/h/book.epub';
+
+describe('GoogleDriveProvider — streaming', () => {
+  // The native-transfer mocks are module-level (shared); clear call history
+  // between tests so per-test call-count assertions stand alone.
+  beforeEach(() => {
+    vi.mocked(tauriUpload).mockClear();
+    vi.mocked(tauriDownload).mockClear();
+  });
+
+  test('exposes uploadStream/downloadStream on Tauri', () => {
+    const h = makeDrive();
+    expect(typeof h.provider.uploadStream).toBe('function');
+    expect(typeof h.provider.downloadStream).toBe('function');
+  });
+
+  test('uploadStream creates a new file via a resumable session and streams the bytes', async () => {
+    const h = makeDrive();
+    stageFolders(h);
+    h.fetchMock
+      .mockResolvedValueOnce(json({ files: [] })) // findChild('book.epub') — absent
+      .mockResolvedValueOnce(session('https://upload.example/session/abc')); // initiation
+
+    const ok = await h.provider.uploadStream!(BOOK, '/disk/book.epub');
+    expect(ok).toBe(true);
+
+    // Initiation is a POST to the resumable create endpoint, metadata in the body.
+    expect(h.method(4)).toBe('POST');
+    expect(h.url(4)).toContain('uploadType=resumable');
+    expect(h.body(4)).toEqual({ name: 'book.epub', parents: ['HID'] });
+
+    // The bytes are PUT to the session URI from disk via the native plugin.
+    expect(tauriUpload).toHaveBeenCalledTimes(1);
+    const call = vi.mocked(tauriUpload).mock.calls[0]!;
+    expect(call[0]).toBe('https://upload.example/session/abc');
+    expect(call[1]).toBe('/disk/book.epub');
+    expect(call[2]).toBe('PUT');
+  });
+
+  test('uploadStream overwrites an existing file by id (PATCH session), preserving the id', async () => {
+    const h = makeDrive();
+    stageFolders(h);
+    h.fetchMock
+      .mockResolvedValueOnce(json({ files: [{ id: 'EXIST' }] })) // findChild — exists
+      .mockResolvedValueOnce(session('https://upload.example/session/upd'));
+
+    const ok = await h.provider.uploadStream!(BOOK, '/disk/book.epub');
+    expect(ok).toBe(true);
+    expect(h.method(4)).toBe('PATCH');
+    expect(h.url(4)).toContain('/EXIST?uploadType=resumable');
+    // No parent reparent on overwrite; only the name rides in the body.
+    expect(h.body(4)).toEqual({ name: 'book.epub' });
+  });
+
+  test('uploadStream returns false (no throw) when the session has no Location, and skips the PUT', async () => {
+    const h = makeDrive();
+    stageFolders(h);
+    h.fetchMock
+      .mockResolvedValueOnce(json({ files: [] })) // findChild — absent
+      .mockResolvedValueOnce(session(null)); // initiation without a session URI
+
+    const ok = await h.provider.uploadStream!(BOOK, '/disk/book.epub');
+    expect(ok).toBe(false);
+    expect(tauriUpload).not.toHaveBeenCalled();
+  });
+
+  test('downloadStream resolves the id and streams to disk with a bearer token', async () => {
+    const h = makeDrive();
+    stageFolders(h);
+    h.fetchMock.mockResolvedValueOnce(json({ files: [{ id: 'DID' }] })); // findChild('book.epub')
+
+    const ok = await h.provider.downloadStream!(BOOK, '/disk/dst.epub');
+    expect(ok).toBe(true);
+    expect(tauriDownload).toHaveBeenCalledTimes(1);
+    const call = vi.mocked(tauriDownload).mock.calls[0]!;
+    expect(call[0]).toContain('/DID?alt=media');
+    expect(call[1]).toBe('/disk/dst.epub');
+    expect(call[3]).toEqual({ Authorization: 'Bearer TOKEN' });
+  });
+
+  test('downloadStream returns false when the remote file is absent, and skips the GET', async () => {
+    const h = makeDrive();
+    stageFolders(h);
+    h.fetchMock.mockResolvedValueOnce(json({ files: [] })); // findChild — absent
+
+    const ok = await h.provider.downloadStream!(BOOK, '/disk/dst.epub');
+    expect(ok).toBe(false);
+    expect(tauriDownload).not.toHaveBeenCalled();
+  });
+});
@@ -11,6 +11,8 @@ import {
  mediaUpdateUrl,
  metadataUrl,
  reparentUrl,
+  resumableCreateUrl,
+  resumableUpdateUrl,
  simpleUploadUrl,
 } from '@/services/sync/providers/gdrive/driveRest';

@@ -49,6 +51,15 @@ describe('driveRest', () => {
    expect(mediaUpdateUrl('FID')).toContain('/FID?uploadType=media');
  });

+  test('resumable upload URLs use uploadType=resumable and request the id', () => {
+    expect(resumableCreateUrl()).toBe(
+      'https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable&fields=id',
+    );
+    expect(resumableUpdateUrl('FID')).toBe(
+      'https://www.googleapis.com/upload/drive/v3/files/FID?uploadType=resumable&fields=id',
+    );
+  });
+
  test('metadataUrl + deleteUrl target the file id', () => {
    expect(metadataUrl('FID')).toBe(
      `${FILES_ENDPOINT}/FID?fields=id,name,mimeType,size,modifiedTime,md5Checksum`,
@@ -33,6 +33,8 @@
 * used with the author's explicit permission.
 */

+import { isTauriAppPlatform } from '@/services/environment';
+import { tauriDownload, tauriUpload } from '@/utils/transfer';
 import {
  FileEntry,
  FileHead,
@@ -51,6 +53,8 @@ import {
  mediaUpdateUrl,
  metadataUrl,
  reparentUrl,
+  resumableCreateUrl,
+  resumableUpdateUrl,
  simpleUploadUrl,
 } from './driveRest';

@@ -91,6 +95,8 @@ const HTTP_NO_CONTENT = 204;
 const HTTP_SERVER_ERROR_FLOOR = 500;

 const CONTENT_TYPE_HEADER = 'Content-Type';
+/** Tells a resumable session the content type of the bytes the PUT will carry. */
+const UPLOAD_CONTENT_TYPE_HEADER = 'X-Upload-Content-Type';
 const JSON_CONTENT_TYPE = 'application/json';
 const DEFAULT_TEXT_CONTENT_TYPE = 'application/json';
 const DEFAULT_BINARY_CONTENT_TYPE = 'application/octet-stream';
@@ -564,6 +570,115 @@ class DriveProviderImpl {
    return (await res.json()) as DriveFile;
  }

+  // --- streaming (Tauri only) ----------------------------------------------
+
+  /**
+   * Streaming upload: create-or-overwrite a file by streaming its bytes straight
+   * from disk through a Drive *resumable* session, so a gigabyte-scale book never
+   * lands in the JS heap (buffered {@link writeBinary} marshals the whole file
+   * across the WebView↔Rust bridge, which crashes the renderer on mobile — the
+   * whole reason this path exists).
+   *
+   * Two-step handshake: (1) POST/PATCH the metadata to open a session — Drive
+   * replies with a one-time, already-authenticated session URI in `Location`;
+   * (2) the native upload plugin PUTs the file body to that URI off the disk.
+   * The metadata (name + parent) rides in the initiation, so unlike the buffered
+   * create-then-name path there is no follow-up reparent PATCH.
+   *
+   * Returns `true` on success, `false` on a swallowed failure (matching the
+   * provider contract: the engine re-ensures dirs and retries once, then throws).
+   */
+  async uploadStream(remotePath: string, localPath: string): Promise<boolean> {
+    try {
+      const segments = splitSegments(remotePath);
+      const name = segments.pop();
+      if (name === undefined) return false;
+      const folderId = await this.resolveFolderSegments(segments, true);
+      if (folderId === null) return false;
+
+      const existingId = await this.findChild(name, folderId);
+      const sessionUri = await this.openResumableSession(
+        name,
+        folderId,
+        existingId,
+        DEFAULT_BINARY_CONTENT_TYPE,
+        remotePath,
+      );
+
+      const responseBody = await tauriUpload(sessionUri, localPath, 'PUT', undefined, {
+        [CONTENT_TYPE_HEADER]: DEFAULT_BINARY_CONTENT_TYPE,
+      } as unknown as Map<string, string>);
+
+      // Overwrite preserves the file id; a new file's id is in the completion
+      // body. Cache it so a following head/read skips re-walking the tree;
+      // best-effort, since a missing id just means the next access re-resolves.
+      const id = existingId ?? parseUploadedId(responseBody);
+      if (id) this.idCache.set(remotePath, id);
+      return true;
+    } catch (e) {
+      console.warn('GoogleDriveProvider.uploadStream failed', remotePath, e);
+      return false;
+    }
+  }
+
+  /**
+   * Open a resumable upload session and return its one-time session URI. A new
+   * file POSTs `{name, parents}`; an overwrite PATCHes the known id with `{name}`
+   * (preserving the id and any links). `X-Upload-Content-Type` declares the body
+   * type the PUT will carry.
+   */
+  private async openResumableSession(
+    name: string,
+    folderId: string,
+    existingId: string | null,
+    contentType: string,
+    path: string,
+  ): Promise<string> {
+    const isNew = existingId === null;
+    const res = await this.authedFetch(
+      isNew ? resumableCreateUrl() : resumableUpdateUrl(existingId),
+      isNew ? HTTP_POST : HTTP_PATCH,
+      {
+        headers: {
+          [CONTENT_TYPE_HEADER]: JSON_CONTENT_TYPE,
+          [UPLOAD_CONTENT_TYPE_HEADER]: contentType,
+        },
+        body: JSON.stringify(isNew ? { name, parents: [folderId] } : { name }),
+      },
+    );
+    await this.ensureOk(res, 'upload', path);
+    const location = res.headers.get('Location');
+    if (!location) {
+      throw new DriveHttpError(
+        res.status,
+        undefined,
+        `Drive upload failed: no resumable session URI for ${path}`,
+      );
+    }
+    return location;
+  }
+
+  /**
+   * Streaming download: GET the file's bytes straight to disk through the native
+   * transfer plugin (same heap-safety rationale as {@link uploadStream}). The
+   * media URL needs a bearer token (the session-URI shortcut is upload-only).
+   * Returns `false` when the file is absent or the transport swallows a failure.
+   */
+  async downloadStream(remotePath: string, localPath: string): Promise<boolean> {
+    try {
+      const fileId = await this.resolveFile(remotePath);
+      if (fileId === null) return false;
+      const token = await this.auth.getAccessToken();
+      await tauriDownload(mediaDownloadUrl(fileId), localPath, undefined, {
+        Authorization: `Bearer ${token}`,
+      });
+      return true;
+    } catch (e) {
+      console.warn('GoogleDriveProvider.downloadStream failed', remotePath, e);
+      return false;
+    }
+  }
+
  // --- request plumbing ----------------------------------------------------

  /** Issue an authenticated Drive request, retried on transient failures. */
@@ -621,6 +736,19 @@ class DriveProviderImpl {
  }
 }

+/**
+ * Best-effort parse of a resumable upload's completion body for the new file id
+ * (we request `fields=id`). Undefined on any malformed/empty body — the caller
+ * treats that as "don't cache", and the next access re-resolves the id.
+ */
+const parseUploadedId = (body: string): string | undefined => {
+  try {
+    return (JSON.parse(body) as DriveFile).id;
+  } catch {
+    return undefined;
+  }
+};
+
 /** Best-effort read of Drive's `error.errors[0].reason` for 403 classification. */
 const readDriveErrorReason = async (res: Response): Promise<string | undefined> => {
  try {
@@ -634,9 +762,13 @@ const readDriveErrorReason = async (res: Response): Promise<string | undefined>
 /**
 * Build a Google Drive {@link FileSyncProvider}. Each public method is wrapped so
 * a Drive HTTP failure surfaces as a {@link FileSyncError} the engine can branch
- * on — mirroring `createWebDAVProvider`. Streaming is intentionally omitted
- * (PR1): the engine falls back to buffered read/write, and resumable Drive upload
- * lands in a later phase to unlock large-book sync on mobile.
+ * on — mirroring `createWebDAVProvider`.
+ *
+ * Streaming `uploadStream`/`downloadStream` are attached only on Tauri platforms
+ * (they shell the bytes through the native transfer plugin off the disk); on web
+ * they stay absent and the engine falls back to buffered read/write. They are
+ * deliberately NOT `wrap`ped — they own the provider's boolean contract
+ * (swallow-to-`false`), matching `createWebDAVProvider`.
 */
 export const createGoogleDriveProvider = (
  auth: DriveAuth,
@@ -644,7 +776,7 @@ export const createGoogleDriveProvider = (
  options: GoogleDriveProviderOptions = {},
 ): FileSyncProvider => {
  const impl = new DriveProviderImpl(auth, fetchFn, options.sleep);
-  return {
+  const provider: FileSyncProvider = {
    rootPath: impl.rootPath,
    readText: (path) => wrap(() => impl.readText(path)),
    readBinary: (path) => wrap(() => impl.readBinary(path)),
@@ -655,4 +787,11 @@ export const createGoogleDriveProvider = (
    ensureDir: (paths) => wrap(() => impl.ensureDir(paths)),
    deleteDir: (path) => wrap(() => impl.deleteDir(path)),
  };
+
+  if (isTauriAppPlatform()) {
+    provider.uploadStream = (remotePath, localPath) => impl.uploadStream(remotePath, localPath);
+    provider.downloadStream = (remotePath, localPath) => impl.downloadStream(remotePath, localPath);
+  }
+
+  return provider;
 };
@@ -89,6 +89,26 @@ export const simpleUploadUrl = (): string =>
 export const mediaUpdateUrl = (fileId: string): string =>
  `${UPLOAD_ENDPOINT}/${fileId}?uploadType=media&fields=id,md5Checksum,size`;

+/**
+ * URL to open a *resumable* upload session that creates a new file. Unlike the
+ * simple media upload, the resumable initiation request carries the file
+ * metadata (name + parent) in its body, so no follow-up reparent PATCH is
+ * needed. The POST replies with the one-time session URI in its `Location`
+ * header; the bytes are then streamed to that URI from disk (constant heap,
+ * the whole point for large books on mobile). `fields=id` narrows the
+ * completion response to the new file id.
+ */
+export const resumableCreateUrl = (): string => `${UPLOAD_ENDPOINT}?uploadType=resumable&fields=id`;
+
+/**
+ * URL to open a resumable upload session that overwrites an *existing* file's
+ * bytes (a PATCH to the known id), preserving the file id and any links rather
+ * than orphaning it. The session-URI handshake is identical to
+ * {@link resumableCreateUrl}.
+ */
+export const resumableUpdateUrl = (fileId: string): string =>
+  `${UPLOAD_ENDPOINT}/${fileId}?uploadType=resumable&fields=id`;
+
 /**
 * URL to fetch a single file's metadata, restricted to the {@link FileEntry}
 * fields the provider exposes.