feat(core): import progress & perf (#15197)

#### PR Dependency Tree


* **PR #15197** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Added a new import pipeline with “plan then commit” batch handling for
Markdown, Notion HTML, Obsidian, and Bear backups.
* Enabled native import sessions with progress, cancellation, and
batch-by-batch committing (including assets, folders, icons, and tags).
  * Added web preflight limits for ZIP and multi-file imports.
* **Bug Fixes**
* Improved import error/warning reporting and continued processing when
some items fail.
* Strengthened snapshot-based file/directory picking to preserve paths.
* **Chores**
  * Updated project packaging/configuration for the new import workflow.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/frontend/native/Cargo.toml

@@ -9,12 +9,14 @@ crate-type = ["cdylib", "rlib"]
 
 [dependencies]
 affine_common = { workspace = true, features = ["hashcash"] }
+affine_importer = { workspace = true }
 affine_media_capture = { path = "./media_capture" }
 affine_nbstore = { workspace = true, features = ["napi"] }
 affine_sqlite_v1 = { path = "./sqlite_v1" }
 napi = { workspace = true }
 napi-derive = { workspace = true }
 once_cell = { workspace = true }
+serde_json = { workspace = true }
 sqlx = { workspace = true, default-features = false, features = [
   "chrono",
   "macros",
@@ -38,9 +40,9 @@ mimalloc = { workspace = true }
 mimalloc = { workspace = true, features = ["local_dynamic_tls"] }
 
 [dev-dependencies]
-chrono     = { workspace = true }
-serde_json = { workspace = true }
-uuid       = { workspace = true }
+chrono = { workspace = true }
+uuid   = { workspace = true }
+zip    = { workspace = true }
 
 [build-dependencies]
 napi-build = { workspace = true }

packages/frontend/native/__tests__/db.spec.mts

@@ -2,7 +2,7 @@ import { fileURLToPath } from 'node:url';
 
 import test from 'ava';
 
-import { SqliteConnection, ValidationResult } from '../index';
+import { SqliteConnection, ValidationResult } from '../index.js';
 
 test('db validate', async t => {
   const path = fileURLToPath(

packages/frontend/native/__tests__/pool.spec.mts

@@ -1,6 +1,6 @@
 import test from 'ava';
 
-import { DocStoragePool } from '../index';
+import { DocStoragePool } from '../index.js';
 
 test('can batch read/write pool', async t => {
   const pool = new DocStoragePool();

packages/frontend/native/index.d.ts

@@ -18,13 +18,6 @@ export declare class ApplicationStateChangedSubscriber {
   unsubscribe(): void
 }
 
-export declare class AudioCaptureSession {
-  stop(): void
-  get sampleRate(): number
-  get channels(): number
-  get actualSampleRate(): number
-}
-
 export declare class ShareableContent {
   static onApplicationListChanged(callback: ((err: Error | null, ) => void)): ApplicationListChangedSubscriber
   static onAppStateChanged(app: ApplicationInfo, callback: ((err: Error | null, ) => void)): ApplicationStateChangedSubscriber
@@ -32,8 +25,6 @@ export declare class ShareableContent {
   static applications(): Array<ApplicationInfo>
   static applicationWithProcessId(processId: number): ApplicationInfo | null
   static isUsingMicrophone(processId: number): boolean
-  static tapAudio(processId: number, audioStreamCallback: ((err: Error | null, arg: Float32Array) => void)): AudioCaptureSession
-  static tapGlobalAudio(excludedProcesses: Array<ApplicationInfo> | undefined | null, audioStreamCallback: ((err: Error | null, arg: Float32Array) => void)): AudioCaptureSession
 }
 
 export declare function abortRecording(id: string): Promise<void>
@@ -75,6 +66,29 @@ export interface RecordingStartOptions {
 export declare function startRecording(opts: RecordingStartOptions): Promise<RecordingSessionMeta>
 
 export declare function stopRecording(id: string): Promise<RecordingArtifact>
+export declare function cancelImportSession(sessionId: string): void
+
+export interface CreateImportBatchLimits {
+  maxDocs?: number
+  maxBlobs?: number
+  maxBlobBytes?: number
+}
+
+export declare function createImportSession(options: CreateImportSessionOptions): string
+
+export interface CreateImportSessionOptions {
+  format: string
+  source: CreateImportSessionSource
+  batchLimits?: CreateImportBatchLimits
+}
+
+export interface CreateImportSessionSource {
+  kind: string
+  path: string
+}
+
+export declare function disposeImportSession(sessionId: string): void
+
 export interface MermaidRenderOptions {
   theme?: string
   fontFamily?: string
@@ -92,6 +106,8 @@ export interface MermaidRenderResult {
 
 export declare function mintChallengeResponse(resource: string, bits?: number | undefined | null): Promise<string>
 
+export declare function nextImportBatch(sessionId: string): string | null
+
 export declare function renderMermaidSvg(request: MermaidRenderRequest): MermaidRenderResult
 
 export declare function renderTypstSvg(request: TypstRenderRequest): TypstRenderResult

packages/frontend/native/index.js

@@ -575,14 +575,17 @@ module.exports = nativeBinding
 module.exports.ApplicationInfo = nativeBinding.ApplicationInfo
 module.exports.ApplicationListChangedSubscriber = nativeBinding.ApplicationListChangedSubscriber
 module.exports.ApplicationStateChangedSubscriber = nativeBinding.ApplicationStateChangedSubscriber
-module.exports.AudioCaptureSession = nativeBinding.AudioCaptureSession
 module.exports.ShareableContent = nativeBinding.ShareableContent
 module.exports.abortRecording = nativeBinding.abortRecording
 module.exports.decodeAudio = nativeBinding.decodeAudio
 module.exports.decodeAudioSync = nativeBinding.decodeAudioSync
 module.exports.startRecording = nativeBinding.startRecording
 module.exports.stopRecording = nativeBinding.stopRecording
+module.exports.cancelImportSession = nativeBinding.cancelImportSession
+module.exports.createImportSession = nativeBinding.createImportSession
+module.exports.disposeImportSession = nativeBinding.disposeImportSession
 module.exports.mintChallengeResponse = nativeBinding.mintChallengeResponse
+module.exports.nextImportBatch = nativeBinding.nextImportBatch
 module.exports.renderMermaidSvg = nativeBinding.renderMermaidSvg
 module.exports.renderTypstSvg = nativeBinding.renderTypstSvg
 module.exports.verifyChallengeResponse = nativeBinding.verifyChallengeResponse

packages/frontend/native/media-capture-exapmle.ts

@@ -1,149 +0,0 @@
-import { join } from 'node:path';
-import { fileURLToPath } from 'node:url';
-
-import {
-  Whisper,
-  WhisperFullParams,
-  WhisperSamplingStrategy,
-} from '@napi-rs/whisper';
-import { BehaviorSubject, EMPTY, Observable } from 'rxjs';
-import {
-  distinctUntilChanged,
-  exhaustMap,
-  groupBy,
-  mergeMap,
-  switchMap,
-  tap,
-} from 'rxjs/operators';
-
-import { type Application, ShareableContent } from './index.js';
-
-const rootDir = join(fileURLToPath(import.meta.url), '..');
-
-const shareableContent = new ShareableContent();
-
-const appList = new Set([
-  'com.tinyspeck.slackmacgap.helper',
-  'us.zoom.xos',
-  'org.mozilla.firefoxdeveloperedition',
-]);
-
-console.info(shareableContent.applications().map(app => app.bundleIdentifier));
-
-const GGLM_LARGE = join(rootDir, 'ggml-large-v3-turbo.bin');
-
-const whisper = new Whisper(GGLM_LARGE, {
-  useGpu: true,
-  gpuDevice: 1,
-});
-
-const whisperParams = new WhisperFullParams(WhisperSamplingStrategy.Greedy);
-
-const SAMPLE_WINDOW_MS = 3000; // 3 seconds, similar to stream.cpp's step_ms
-const SAMPLES_PER_WINDOW = (SAMPLE_WINDOW_MS / 1000) * 16000; // 16kHz sample rate
-
-// eslint-disable-next-line rxjs/finnish
-const runningApplications = new BehaviorSubject(
-  shareableContent.applications()
-);
-
-const applicationListChangedSubscriber =
-  ShareableContent.onApplicationListChanged(() => {
-    runningApplications.next(shareableContent.applications());
-  });
-
-runningApplications
-  .pipe(
-    mergeMap(apps => apps.filter(app => appList.has(app.bundleIdentifier))),
-    groupBy(app => app.bundleIdentifier),
-    mergeMap(app$ =>
-      app$.pipe(
-        exhaustMap(app =>
-          new Observable<[Application, boolean]>(subscriber => {
-            const stateSubscriber = ShareableContent.onAppStateChanged(
-              app,
-              err => {
-                if (err) {
-                  subscriber.error(err);
-                  return;
-                }
-                subscriber.next([app, app.isRunning]);
-              }
-            );
-            return () => {
-              stateSubscriber.unsubscribe();
-            };
-          }).pipe(
-            distinctUntilChanged(
-              ([_, isRunningA], [__, isRunningB]) => isRunningA === isRunningB
-            ),
-            switchMap(([app]) =>
-              !app.isRunning
-                ? EMPTY
-                : new Observable(observer => {
-                    const buffers: Float32Array[] = [];
-                    const audioStream = app.tapAudio((err, samples) => {
-                      if (err) {
-                        observer.error(err);
-                        return;
-                      }
-
-                      if (samples) {
-                        buffers.push(samples);
-                        observer.next(samples);
-
-                        // Calculate total samples in buffer
-                        const totalSamples = buffers.reduce(
-                          (acc, buf) => acc + buf.length,
-                          0
-                        );
-
-                        // Process when we have enough samples for our window
-                        if (totalSamples >= SAMPLES_PER_WINDOW) {
-                          // Concatenate all buffers
-                          const concatenated = new Float32Array(totalSamples);
-                          let offset = 0;
-                          buffers.forEach(buf => {
-                            concatenated.set(buf, offset);
-                            offset += buf.length;
-                          });
-
-                          // Transcribe the audio
-                          const result = whisper.full(
-                            whisperParams,
-                            concatenated
-                          );
-
-                          // Print results
-                          console.info(result);
-
-                          // Keep any remaining samples for next window
-                          const remainingSamples =
-                            totalSamples - SAMPLES_PER_WINDOW;
-                          if (remainingSamples > 0) {
-                            const lastBuffer = buffers[buffers.length - 1];
-                            buffers.length = 0;
-                            buffers.push(lastBuffer.slice(-remainingSamples));
-                          } else {
-                            buffers.length = 0;
-                          }
-                        }
-                      }
-                    });
-
-                    return () => {
-                      audioStream.stop();
-                    };
-                  })
-            )
-          )
-        )
-      )
-    ),
-    tap({
-      finalize: () => {
-        applicationListChangedSubscriber.unsubscribe();
-      },
-    })
-  )
-  .subscribe();

packages/frontend/native/media_capture/src/macos/screen_capture_kit.rs

@@ -682,14 +682,6 @@ impl ShareableContent {
     }
   }
 
-  #[napi]
-  pub fn tap_audio(
-    process_id: u32,
-    audio_stream_callback: ThreadsafeFunction<napi::bindgen_prelude::Float32Array, ()>,
-  ) -> Result<AudioCaptureSession> {
-    ShareableContent::tap_audio_with_callback(process_id, AudioCallback::Js(Arc::new(audio_stream_callback)))
-  }
-
   pub(crate) fn tap_global_audio_with_callback(
     excluded_processes: Option<Vec<&ApplicationInfo>>,
     audio_stream_callback: AudioCallback,
@@ -706,15 +698,4 @@ impl ShareableContent {
     let boxed_manager = Box::new(device_manager);
     Ok(AudioCaptureSession::new(boxed_manager))
   }
-
-  #[napi]
-  pub fn tap_global_audio(
-    excluded_processes: Option<Vec<&ApplicationInfo>>,
-    audio_stream_callback: ThreadsafeFunction<napi::bindgen_prelude::Float32Array, ()>,
-  ) -> Result<AudioCaptureSession> {
-    ShareableContent::tap_global_audio_with_callback(
-      excluded_processes,
-      AudioCallback::Js(Arc::new(audio_stream_callback)),
-    )
-  }
 }

packages/frontend/native/media_capture/src/macos/tap_audio.rs

@@ -21,7 +21,6 @@ use coreaudio::sys::{
   kAudioSubTapUIDKey,
 };
 use napi::bindgen_prelude::Result;
-use napi_derive::napi;
 use objc2::runtime::AnyObject;
 
 use crate::{
@@ -887,8 +886,6 @@ impl Drop for AggregateDeviceManager {
   }
 }
 
-// NEW NAPI Struct: AudioCaptureSession
-#[napi]
 pub struct AudioCaptureSession {
   // Use Option<Box<...>> to allow taking ownership in stop()
   manager: Option<Box<AggregateDeviceManager>>,
@@ -896,9 +893,7 @@ pub struct AudioCaptureSession {
   channels: Option<u32>,
 }
 
-#[napi]
 impl AudioCaptureSession {
-  // Constructor called internally, not directly via NAPI
   pub(crate) fn new(manager: Box<AggregateDeviceManager>) -> Self {
     Self {
       manager: Some(manager),
@@ -907,7 +902,6 @@ impl AudioCaptureSession {
     }
   }
 
-  #[napi]
   pub fn stop(&mut self) -> Result<()> {
     if let Some(manager) = self.manager.take() {
       // Cache the stats before dropping
@@ -926,7 +920,6 @@ impl AudioCaptureSession {
     }
   }
 
-  #[napi(getter)]
   pub fn get_sample_rate(&self) -> Result<f64> {
     if let Some(manager) = &self.manager {
       manager
@@ -943,7 +936,6 @@ impl AudioCaptureSession {
     }
   }
 
-  #[napi(getter)]
   pub fn get_channels(&self) -> Result<u32> {
     if let Some(manager) = &self.manager {
       manager
@@ -960,7 +952,6 @@ impl AudioCaptureSession {
     }
   }
 
-  #[napi(getter)]
   pub fn get_actual_sample_rate(&self) -> Result<f64> {
     if let Some(manager) = &self.manager {
       manager

packages/frontend/native/media_capture/src/windows/audio_capture.rs

@@ -200,7 +200,6 @@ fn emit_mixed_frames(
   }
 }
 
-#[napi]
 pub struct AudioCaptureSession {
   mic_stream: Option<cpal::Stream>,
   lb_stream: Option<cpal::Stream>,
@@ -258,25 +257,20 @@ where
   }
 }
 
-#[napi]
 impl AudioCaptureSession {
-  #[napi(getter)]
   pub fn get_sample_rate(&self) -> f64 {
     self.sample_rate.0 as f64
   }
 
-  #[napi(getter)]
   pub fn get_channels(&self) -> u32 {
     self.channels
   }
 
-  #[napi(getter)]
   pub fn get_actual_sample_rate(&self) -> f64 {
     // For CPAL we always operate at the target rate which is sample_rate
     self.sample_rate.0 as f64
   }
 
-  #[napi]
   pub fn stop(&mut self) -> Result<()> {
     teardown_audio_capture_resources(
       &mut self.mic_stream,

packages/frontend/native/media_capture/src/windows/screen_capture_kit.rs

@@ -225,16 +225,6 @@ impl ShareableContent {
     crate::windows::audio_capture::start_recording(audio_stream_callback, target)
   }
 
-  #[napi]
-  pub fn tap_audio(
-    _process_id: u32, // Currently unused - Windows captures global audio
-    audio_stream_callback: ThreadsafeFunction<napi::bindgen_prelude::Float32Array, ()>,
-  ) -> Result<AudioCaptureSession> {
-    // On Windows with CPAL, we capture global audio (mic + loopback)
-    // since per-application audio tapping isn't supported the same way as macOS
-    ShareableContent::tap_audio_with_callback(_process_id, AudioCallback::Js(Arc::new(audio_stream_callback)), None)
-  }
-
   pub(crate) fn tap_global_audio_with_callback(
     _excluded_processes: Option<Vec<&ApplicationInfo>>,
     audio_stream_callback: AudioCallback,
@@ -246,18 +236,6 @@ impl ShareableContent {
     crate::windows::audio_capture::start_recording(audio_stream_callback, target)
   }
 
-  #[napi]
-  pub fn tap_global_audio(
-    _excluded_processes: Option<Vec<&ApplicationInfo>>,
-    audio_stream_callback: ThreadsafeFunction<napi::bindgen_prelude::Float32Array, ()>,
-  ) -> Result<AudioCaptureSession> {
-    ShareableContent::tap_global_audio_with_callback(
-      _excluded_processes,
-      AudioCallback::Js(Arc::new(audio_stream_callback)),
-      None,
-    )
-  }
-
   #[napi]
   pub fn is_using_microphone(process_id: u32) -> Result<bool> {
     is_process_actively_using_microphone(process_id)

packages/frontend/native/nbstore/Cargo.toml

@@ -13,7 +13,8 @@ napi       = ["affine_common/napi"]
 use-as-lib = ["napi-derive/noop", "napi/noop"]
 
 [dependencies]
-affine_common = { workspace = true, features = ["ydoc-loader"] }
+affine_common = { workspace = true, features = ["napi"] }
+affine_doc_loader = { workspace = true }
 affine_schema = { path = "../schema" }
 anyhow = { workspace = true }
 bincode = { version = "2.0.1", features = ["serde"] }

packages/frontend/native/nbstore/src/error.rs

@@ -1,4 +1,4 @@
-use affine_common::doc_parser::ParseError;
+use affine_doc_loader::ParseError;
 
 pub type Result<T> = std::result::Result<T, Error>;
 

packages/frontend/native/nbstore/src/indexer.rs

@@ -1,4 +1,4 @@
-use affine_common::doc_parser::{BlockInfo, CrawlResult, ParseError, parse_doc_from_binary};
+use affine_doc_loader::{BlockInfo, CrawlResult, ParseError, parse_doc_from_binary};
 use memory_indexer::{SearchHit, SnapshotData};
 use napi_derive::napi;
 use serde::Serialize;
@@ -276,7 +276,7 @@ fn merge_updates(mut segments: Vec<Vec<u8>>, guid: &str) -> Result<Vec<u8>> {
 mod tests {
   use std::path::{Path, PathBuf};
 
-  use affine_common::doc_parser::ParseError;
+  use affine_doc_loader::ParseError;
   use chrono::Utc;
   use serde_json::Value;
   use tokio::fs;

packages/frontend/native/nbstore/src/storage.rs

@@ -311,7 +311,7 @@ mod tests {
     storage
       .set_blob(crate::SetBlob {
         key: "large-blob".to_string(),
-        data: vec![7; 1024 * 1024],
+        data: Into::<crate::Data>::into(vec![7; 1024 * 1024]),
         mime: "application/octet-stream".to_string(),
       })
       .await

packages/frontend/native/src/import_session.rs

@@ -0,0 +1,270 @@
+use std::{
+  collections::HashMap,
+  path::PathBuf,
+  sync::{
+    Mutex,
+    atomic::{AtomicU64, Ordering},
+  },
+};
+
+use affine_importer::{
+  ImportBatchLimits, ImportFormat, ImportOptions, ImportSession, ImportSessionOptions, ImportSessionSource,
+};
+use napi::{Status, bindgen_prelude::*};
+use napi_derive::napi;
+use once_cell::sync::Lazy;
+
+static IMPORT_SESSIONS: Lazy<Mutex<HashMap<String, ImportSession>>> = Lazy::new(|| Mutex::new(HashMap::new()));
+static NEXT_SESSION_ID: AtomicU64 = AtomicU64::new(1);
+
+#[napi(object)]
+pub struct CreateImportSessionOptions {
+  pub format: String,
+  pub source: CreateImportSessionSource,
+  pub batch_limits: Option<CreateImportBatchLimits>,
+}
+
+#[napi(object)]
+pub struct CreateImportSessionSource {
+  pub kind: String,
+  pub path: String,
+}
+
+#[napi(object)]
+pub struct CreateImportBatchLimits {
+  pub max_docs: Option<u32>,
+  pub max_blobs: Option<u32>,
+  pub max_blob_bytes: Option<i64>,
+}
+
+fn parse_format(format: &str) -> Result<ImportFormat> {
+  match format {
+    "markdownZip" => Ok(ImportFormat::MarkdownZip),
+    "notionZip" => Ok(ImportFormat::NotionZip),
+    "notionMarkdownZip" => Ok(ImportFormat::NotionMarkdownZip),
+    "notionHtmlZip" => Ok(ImportFormat::NotionHtmlZip),
+    "obsidian" => Ok(ImportFormat::Obsidian),
+    "bearZip" => Ok(ImportFormat::BearZip),
+    _ => Err(Error::new(
+      Status::InvalidArg,
+      format!("unsupported import format: {format}"),
+    )),
+  }
+}
+
+fn map_import_error(error: affine_importer::ImportError) -> Error {
+  let status = match error {
+    affine_importer::ImportError::Cancelled => Status::Cancelled,
+    affine_importer::ImportError::UnsupportedFormat | affine_importer::ImportError::InvalidSource(_) => {
+      Status::InvalidArg
+    }
+    affine_importer::ImportError::Zip(_)
+    | affine_importer::ImportError::Io(_)
+    | affine_importer::ImportError::Document(_) => Status::GenericFailure,
+  };
+  Error::new(status, error.to_string())
+}
+
+#[napi]
+pub fn create_import_session(options: CreateImportSessionOptions) -> Result<String> {
+  let format = parse_format(&options.format)?;
+  let source = match options.source.kind.as_str() {
+    "filePath" => ImportSessionSource::FilePath(PathBuf::from(options.source.path)),
+    "directoryPath" => ImportSessionSource::DirectoryPath(PathBuf::from(options.source.path)),
+    kind => {
+      return Err(Error::new(
+        Status::InvalidArg,
+        format!("unsupported import source kind: {kind}"),
+      ));
+    }
+  };
+  let mut batch_limits = ImportBatchLimits::default();
+  if let Some(limits) = options.batch_limits {
+    if let Some(max_docs) = limits.max_docs {
+      batch_limits.max_docs = max_docs as usize;
+    }
+    if let Some(max_blobs) = limits.max_blobs {
+      batch_limits.max_blobs = max_blobs as usize;
+    }
+    if let Some(max_blob_bytes) = limits.max_blob_bytes {
+      if !(0..=100 * 1024 * 1024).contains(&max_blob_bytes) {
+        return Err(Error::new(Status::InvalidArg, "maxBlobBytes not valid"));
+      }
+      batch_limits.max_blob_bytes = max_blob_bytes as u64;
+    }
+  }
+  let session = ImportSession::create(ImportSessionOptions {
+    format,
+    source,
+    import_options: ImportOptions::default(),
+    batch_limits,
+  })
+  .map_err(map_import_error)?;
+  let id = format!("import-session-{}", NEXT_SESSION_ID.fetch_add(1, Ordering::Relaxed));
+  IMPORT_SESSIONS
+    .lock()
+    .map_err(|_| Error::new(Status::GenericFailure, "import session registry poisoned"))?
+    .insert(id.clone(), session);
+  Ok(id)
+}
+
+#[napi]
+pub fn next_import_batch(session_id: String) -> Result<Option<String>> {
+  let mut sessions = IMPORT_SESSIONS
+    .lock()
+    .map_err(|_| Error::new(Status::GenericFailure, "import session registry poisoned"))?;
+  let session = sessions
+    .get_mut(&session_id)
+    .ok_or_else(|| Error::new(Status::InvalidArg, format!("unknown import session: {session_id}")))?;
+  session
+    .next_batch()
+    .map_err(map_import_error)?
+    .map(|batch| serde_json::to_string(&batch).map_err(|err| Error::new(Status::GenericFailure, err.to_string())))
+    .transpose()
+}
+
+#[napi]
+pub fn cancel_import_session(session_id: String) -> Result<()> {
+  let mut sessions = IMPORT_SESSIONS
+    .lock()
+    .map_err(|_| Error::new(Status::GenericFailure, "import session registry poisoned"))?;
+  let session = sessions
+    .get_mut(&session_id)
+    .ok_or_else(|| Error::new(Status::InvalidArg, format!("unknown import session: {session_id}")))?;
+  session.cancel();
+  Ok(())
+}
+
+#[napi]
+pub fn dispose_import_session(session_id: String) -> Result<()> {
+  IMPORT_SESSIONS
+    .lock()
+    .map_err(|_| Error::new(Status::GenericFailure, "import session registry poisoned"))?
+    .remove(&session_id);
+  Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+  use std::{
+    fs,
+    io::Write,
+    path::PathBuf,
+    sync::atomic::{AtomicU64, Ordering},
+  };
+
+  use zip::{ZipWriter, write::SimpleFileOptions};
+
+  use super::*;
+
+  static NEXT_TEST_ARCHIVE_ID: AtomicU64 = AtomicU64::new(1);
+
+  fn archive_path(entries: &[(&str, &[u8])]) -> PathBuf {
+    let path = std::env::temp_dir().join(format!(
+      "affine-native-import-{}-{}.zip",
+      std::process::id(),
+      NEXT_TEST_ARCHIVE_ID.fetch_add(1, Ordering::Relaxed)
+    ));
+    let file = std::fs::File::create(&path).unwrap();
+    let mut writer = ZipWriter::new(file);
+    for (path, bytes) in entries {
+      writer.start_file(path, SimpleFileOptions::default()).unwrap();
+      writer.write_all(bytes).unwrap();
+    }
+    writer.finish().unwrap();
+    path
+  }
+
+  fn directory_path(entries: &[(&str, &[u8])]) -> PathBuf {
+    let root = std::env::temp_dir().join(format!(
+      "affine-native-import-dir-{}-{}",
+      std::process::id(),
+      NEXT_TEST_ARCHIVE_ID.fetch_add(1, Ordering::Relaxed)
+    ));
+    for (path, bytes) in entries {
+      let path = root.join(path);
+      fs::create_dir_all(path.parent().unwrap()).unwrap();
+      fs::write(path, bytes).unwrap();
+    }
+    root
+  }
+
+  #[test]
+  fn import_session_returns_one_serialized_batch() {
+    let path = archive_path(&[("entry.md", b"entry")]);
+    let id = create_import_session(CreateImportSessionOptions {
+      format: "markdownZip".to_string(),
+      source: CreateImportSessionSource {
+        kind: "filePath".to_string(),
+        path: path.to_string_lossy().to_string(),
+      },
+      batch_limits: None,
+    })
+    .unwrap();
+
+    let batch = next_import_batch(id.clone()).unwrap().unwrap();
+    assert!(batch.contains("\"docs\""));
+    assert!(next_import_batch(id.clone()).unwrap().is_none());
+    dispose_import_session(id).unwrap();
+    let _ = std::fs::remove_file(path);
+  }
+
+  #[test]
+  fn import_session_cancel_blocks_next_batch() {
+    let path = archive_path(&[("entry.md", b"entry")]);
+    let id = create_import_session(CreateImportSessionOptions {
+      format: "markdownZip".to_string(),
+      source: CreateImportSessionSource {
+        kind: "filePath".to_string(),
+        path: path.to_string_lossy().to_string(),
+      },
+      batch_limits: None,
+    })
+    .unwrap();
+
+    cancel_import_session(id.clone()).unwrap();
+    assert!(next_import_batch(id.clone()).is_err());
+    dispose_import_session(id).unwrap();
+    let _ = std::fs::remove_file(path);
+  }
+
+  #[test]
+  fn import_session_accepts_directory_source() {
+    let path = directory_path(&[("entry.md", b"entry")]);
+    let id = create_import_session(CreateImportSessionOptions {
+      format: "obsidian".to_string(),
+      source: CreateImportSessionSource {
+        kind: "directoryPath".to_string(),
+        path: path.to_string_lossy().to_string(),
+      },
+      batch_limits: None,
+    })
+    .unwrap();
+
+    let batch = next_import_batch(id.clone()).unwrap().unwrap();
+    assert!(batch.contains("\"docs\""));
+    assert!(next_import_batch(id.clone()).unwrap().is_none());
+    dispose_import_session(id).unwrap();
+    let _ = fs::remove_dir_all(path);
+  }
+
+  #[test]
+  fn import_session_rejects_negative_blob_byte_limit() {
+    let path = archive_path(&[("entry.md", b"entry")]);
+    let result = create_import_session(CreateImportSessionOptions {
+      format: "markdownZip".to_string(),
+      source: CreateImportSessionSource {
+        kind: "filePath".to_string(),
+        path: path.to_string_lossy().to_string(),
+      },
+      batch_limits: Some(CreateImportBatchLimits {
+        max_docs: None,
+        max_blobs: None,
+        max_blob_bytes: Some(-1),
+      }),
+    });
+
+    assert!(result.is_err());
+    let _ = std::fs::remove_file(path);
+  }
+}

packages/frontend/native/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod hashcash;
+mod import_session;
 #[cfg(not(any(target_os = "android", target_os = "ios")))]
 pub mod preview;
 
@@ -10,3 +11,4 @@ static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
 pub use affine_media_capture::*;
 pub use affine_nbstore::*;
 pub use affine_sqlite_v1::*;
+pub use import_session::*;