diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b499525a..8237ee5e 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -49,6 +49,20 @@ jobs: - name: Setup Rust toolchain uses: dtolnay/rust-toolchain@stable + with: + targets: wasm32-unknown-unknown + + - name: Install wasm-pack + shell: bash + run: | + archive="$RUNNER_TEMP/wasm-pack.tar.gz" + directory="$RUNNER_TEMP/wasm-pack-v0.15.0-x86_64-unknown-linux-musl" + curl --proto '=https' --tlsv1.2 -LsSf \ + https://github.com/wasm-bindgen/wasm-pack/releases/download/v0.15.0/wasm-pack-v0.15.0-x86_64-unknown-linux-musl.tar.gz \ + -o "$archive" + echo "c09f971ecaed9a2efc80fdcea7a00ef6b53c7fadc8c57d1f61b53a6aa66b668a $archive" | sha256sum -c - + tar -xzf "$archive" -C "$RUNNER_TEMP" + install -m 755 "$directory/wasm-pack" "$CARGO_HOME/bin/wasm-pack" - name: Setup Rust cache uses: Swatinem/rust-cache@v2 @@ -73,6 +87,9 @@ jobs: working-directory: packages/parser-native run: cargo test + - name: Build browser parser WASM + run: pnpm run build:parser-wasm + # Debug 构建足够触发 TS 侧 parity 测试（native 缺失时这些测试会自动 skip， # 构建后即可在 CI 上真实覆盖 Rust/TS 双路径一致性）。 - name: Build native parser (debug) diff --git a/package.json b/package.json index f0d7f443..614b6052 100644 --- a/package.json +++ b/package.json @@ -28,6 +28,7 @@ "dev": "node scripts/dev-select.mjs", "dev:desktop": "pnpm --filter @openchatlab/desktop dev", "build:native": "pnpm --filter @openchatlab/parser-native build", + "build:parser-wasm": "pnpm --filter @openchatlab/parser-native build:wasm", "build:desktop": "pnpm --filter @openchatlab/desktop build", "build:mac": "pnpm --filter @openchatlab/desktop build:mac", "build:win": "pnpm --filter @openchatlab/desktop build:win", diff --git a/packages/parser-native/Cargo.lock b/packages/parser-native/Cargo.lock index 589971ad..fe36d53b 100644 --- a/packages/parser-native/Cargo.lock +++ b/packages/parser-native/Cargo.lock @@ -8,6 +8,12 @@ version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + [[package]] name = "cfg-if" version = "1.0.4" @@ -24,6 +30,7 @@ dependencies = [ "napi-derive", "serde", "serde_json", + "wasm-bindgen", ] [[package]] @@ -236,6 +243,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2bf50223579dc7cdcfb3bfcacf7069ff68243f8c363f62ffa99cf000a6b9c451" +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -266,6 +279,12 @@ version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + [[package]] name = "semver" version = "1.0.28" @@ -345,6 +364,51 @@ version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" +[[package]] +name = "wasm-bindgen" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" +dependencies = [ + "unicode-ident", +] + [[package]] name = "windows-link" version = "0.2.1" diff --git a/packages/parser-native/Cargo.toml b/packages/parser-native/Cargo.toml index 77da84ff..ab8a22e5 100644 --- a/packages/parser-native/Cargo.toml +++ b/packages/parser-native/Cargo.toml @@ -3,21 +3,30 @@ name = "chatlab-parser-native" version = "0.1.0" edition = "2021" license = "AGPL-3.0" -description = "ChatLab native chat-export parsing kernels (napi-rs)" +description = "ChatLab Rust chat-export parsing kernels for N-API and browser WASM" [lib] crate-type = ["cdylib"] +[features] +default = ["napi"] +napi = ["dep:napi", "dep:napi-derive", "dep:napi-build"] +wasm = ["dep:wasm-bindgen"] + [dependencies] -napi = { version = "3", features = ["napi8"] } -napi-derive = "3" +napi = { version = "3", features = ["napi8"], optional = true } +napi-derive = { version = "3", optional = true } serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order", "raw_value"] } memchr = "2" +wasm-bindgen = { version = "0.2.126", optional = true } [build-dependencies] -napi-build = "2" +napi-build = { version = "2", optional = true } [profile.release] lto = true strip = "symbols" + +[package.metadata.wasm-pack.profile.release] +wasm-opt = ["-O", "--enable-bulk-memory"] diff --git a/packages/parser-native/build.rs b/packages/parser-native/build.rs index 0f1b0100..f2be9938 100644 --- a/packages/parser-native/build.rs +++ b/packages/parser-native/build.rs @@ -1,3 +1,4 @@ fn main() { + #[cfg(feature = "napi")] napi_build::setup(); } diff --git a/packages/parser-native/package.json b/packages/parser-native/package.json index eddabcdd..4b8b7f46 100644 --- a/packages/parser-native/package.json +++ b/packages/parser-native/package.json @@ -2,7 +2,7 @@ "name": "@openchatlab/parser-native", "version": "0.0.0", "private": true, - "description": "ChatLab Rust 原生解析内核（napi-rs，一期覆盖 WeFlow 格式）", + "description": "ChatLab Rust 解析内核（N-API 与浏览器 WASM）", "main": "index.js", "types": "index.d.ts", "napi": { @@ -11,6 +11,7 @@ "scripts": { "build": "napi build --platform --release", "build:debug": "napi build --platform", + "build:wasm": "node scripts/build-wasm.mjs", "test": "cargo test" }, "devDependencies": { diff --git a/packages/parser-native/scripts/build-wasm.mjs b/packages/parser-native/scripts/build-wasm.mjs new file mode 100644 index 00000000..70797b9d --- /dev/null +++ b/packages/parser-native/scripts/build-wasm.mjs @@ -0,0 +1,40 @@ +import { spawnSync } from 'node:child_process' +import { rmSync } from 'node:fs' +import { resolve } from 'node:path' + +const crateDirectory = resolve(import.meta.dirname, '..') +const outputDirectory = resolve(crateDirectory, '../web-runtime/src/wasm/generated') +const version = spawnSync('wasm-pack', ['--version'], { encoding: 'utf8' }) + +if (version.error) throw version.error +if (version.status !== 0) process.exit(version.status ?? 1) +if (version.stdout.trim() !== 'wasm-pack 0.15.0') { + throw new Error(`Expected wasm-pack 0.15.0, received: ${version.stdout.trim() || 'unknown version'}`) +} + +const result = spawnSync( + 'wasm-pack', + [ + 'build', + '--target', + 'web', + '--release', + '--out-dir', + outputDirectory, + '--out-name', + 'parser_native', + '--', + '--locked', + '--no-default-features', + '--features', + 'wasm', + ], + { cwd: crateDirectory, stdio: 'inherit' } +) + +if (result.error) throw result.error +if (result.status !== 0) process.exit(result.status ?? 1) + +// wasm-pack emits npm metadata that would be detected as a nested pnpm workspace. +rmSync(resolve(outputDirectory, 'package.json'), { force: true }) +rmSync(resolve(outputDirectory, '.gitignore'), { force: true }) diff --git a/packages/parser-native/src/input.rs b/packages/parser-native/src/input.rs index 81bda2d0..27c0fad5 100644 --- a/packages/parser-native/src/input.rs +++ b/packages/parser-native/src/input.rs @@ -4,6 +4,7 @@ //! passed to the JS constructor. Multi-file formats (QQ chunked, Google Chat //! Takeout) will extend this struct with base-dir scoped file access. +#[cfg(feature = "napi")] use std::io; #[derive(Clone)] @@ -17,6 +18,7 @@ pub struct KernelInput { } impl KernelInput { + #[cfg(feature = "napi")] pub fn read_primary(&self) -> io::Result> { std::fs::read(&self.primary_path) } diff --git a/packages/parser-native/src/lib.rs b/packages/parser-native/src/lib.rs index 87dbf847..1b71eb59 100644 --- a/packages/parser-native/src/lib.rs +++ b/packages/parser-native/src/lib.rs @@ -5,16 +5,24 @@ mod input; mod jsutil; mod protocol; mod scanner; +#[cfg(feature = "wasm")] +mod wasm; mod weflow; +#[cfg(feature = "napi")] use std::sync::atomic::{AtomicU64, Ordering}; +#[cfg(feature = "napi")] use std::sync::{Arc, Mutex, MutexGuard}; +#[cfg(feature = "napi")] use napi::bindgen_prelude::*; +#[cfg(feature = "napi")] use napi_derive::napi; use input::KernelInput; -use protocol::{KernelOutput, NativeMember, NativeMessage, NativeParseProgress}; +use protocol::{KernelOutput, NativeMessage}; +#[cfg(feature = "napi")] +use protocol::{NativeMember, NativeParseProgress}; // ==================== Format registry ==================== @@ -27,7 +35,7 @@ enum FormatKind { } impl FormatKind { - fn from_id(id: &str) -> Option { + pub(crate) fn from_id(id: &str) -> Option { match id { "weflow" => Some(FormatKind::Weflow), "chatlab" => Some(FormatKind::Chatlab), @@ -35,7 +43,7 @@ impl FormatKind { } } - fn run( + pub(crate) fn run( &self, buf: &[u8], input: &KernelInput, @@ -50,7 +58,7 @@ impl FormatKind { // ==================== Shared parser state ==================== -struct ParsedState { +pub(crate) struct ParsedState { output: KernelOutput, cursor: usize, /// Totals captured at parse completion; `take_*` drains the vectors, so @@ -59,6 +67,7 @@ struct ParsedState { member_total: usize, } +#[cfg(feature = "napi")] struct Shared { bytes_read: AtomicU64, total_bytes: AtomicU64, @@ -66,6 +75,7 @@ struct Shared { parsed: Mutex>, } +#[cfg(feature = "napi")] impl Shared { fn new() -> Arc { Arc::new(Shared { @@ -118,7 +128,7 @@ impl Shared { } /// Take the next `size` messages, advancing the cursor; None when exhausted. -fn take_batch_from(state: &mut ParsedState, size: u32) -> Option> { +pub(crate) fn take_batch_from(state: &mut ParsedState, size: u32) -> Option> { let cursor = state.cursor; let total = state.output.messages.len(); if cursor >= total { @@ -135,12 +145,14 @@ fn take_batch_from(state: &mut ParsedState, size: u32) -> Option, } +#[cfg(feature = "napi")] impl Task for ParseFormatTask { type Output = (); type JsValue = (); @@ -158,6 +170,7 @@ impl Task for ParseFormatTask { /// `parse()` → `metaJson()` → `takeMembers()` → `takeBatch()` until null → /// `summaryJson()`. Constructing with an unknown format id throws, which the /// TS wrapper treats like any other native failure (falls back to TS). +#[cfg(feature = "napi")] #[napi] pub struct NativeParser { format: FormatKind, @@ -165,6 +178,7 @@ pub struct NativeParser { shared: Arc, } +#[cfg(feature = "napi")] #[napi] impl NativeParser { #[napi(constructor)] diff --git a/packages/parser-native/src/protocol.rs b/packages/parser-native/src/protocol.rs index 2e18e5ea..6eeb93c1 100644 --- a/packages/parser-native/src/protocol.rs +++ b/packages/parser-native/src/protocol.rs @@ -4,23 +4,30 @@ //! not produce stay `None`); the format-specific meta travels as a JSON string //! so adding formats never changes the N-API surface. +#[cfg(feature = "napi")] use napi_derive::napi; +use serde::Serialize; -#[napi(object)] +#[cfg(feature = "napi")] +#[cfg_attr(feature = "napi", napi(object))] pub struct NativeParseProgress { pub bytes_read: f64, pub total_bytes: f64, pub messages_processed: f64, } -#[napi(object)] +#[cfg_attr(feature = "napi", napi(object))] +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] pub struct NativeMemberRole { pub id: String, /// Present only when the source role object had a `name` key. pub name: Option, } -#[napi(object)] +#[cfg_attr(feature = "napi", napi(object))] +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] pub struct NativeMember { pub platform_id: String, pub account_name: String, @@ -30,8 +37,9 @@ pub struct NativeMember { pub roles: Option>, } -#[napi(object)] -#[derive(Default)] +#[cfg_attr(feature = "napi", napi(object))] +#[derive(Default, Serialize)] +#[serde(rename_all = "camelCase")] pub struct NativeMessage { pub platform_message_id: Option, pub sender_platform_id: String, diff --git a/packages/parser-native/src/wasm.rs b/packages/parser-native/src/wasm.rs new file mode 100644 index 00000000..b397b703 --- /dev/null +++ b/packages/parser-native/src/wasm.rs @@ -0,0 +1,131 @@ +//! Browser Worker binding for the shared Rust parser kernels. +//! +//! Parsing is synchronous because the binding runs inside the dedicated Web +//! Worker. Results remain batched at the JS boundary to avoid creating one +//! large object graph in a single transfer. + +use wasm_bindgen::prelude::*; + +use crate::input::KernelInput; +use crate::{take_batch_from, FormatKind, ParsedState}; + +#[wasm_bindgen] +pub struct WasmParser { + state: ParsedState, +} + +#[wasm_bindgen] +impl WasmParser { + #[wasm_bindgen(constructor)] + pub fn new( + format_id: String, + bytes: &[u8], + file_name: String, + options_json: Option, + ) -> Result { + let format = FormatKind::from_id(&format_id) + .ok_or_else(|| js_error(format!("unknown WASM format id: {format_id}")))?; + let input = KernelInput { + primary_path: file_name, + options_json, + }; + let output = format + .run(bytes, &input, &mut |_, _| {}) + .map_err(|error| js_error(format!("parse failed: {error}")))?; + + Ok(WasmParser { + state: ParsedState { + message_total: output.messages.len(), + member_total: output.members.len(), + output, + cursor: 0, + }, + }) + } + + pub fn meta_json(&self) -> String { + self.state.output.meta_json.clone() + } + + pub fn take_members_json(&mut self) -> Result { + serialize(&std::mem::take(&mut self.state.output.members), "members") + } + + pub fn take_batch_json(&mut self, size: u32) -> Result, JsValue> { + take_batch_from(&mut self.state, size) + .map(|batch| serialize(&batch, "message batch")) + .transpose() + } + + pub fn summary_json(&self) -> String { + format!( + "{{\"messageCount\":{},\"memberCount\":{}}}", + self.state.message_total, self.state.member_total + ) + } +} + +fn serialize(value: &T, label: &str) -> Result { + serde_json::to_string(value) + .map_err(|error| js_error(format!("failed to serialize {label}: {error}"))) +} + +fn js_error(message: String) -> JsValue { + JsValue::from_str(&message) +} + +#[cfg(test)] +mod tests { + use serde_json::Value; + + use super::*; + + #[test] + fn exposes_chatlab_results_in_bounded_json_batches() { + let doc = br#"{ + "chatlab": {"version": "1"}, + "meta": {"name": "Browser", "platform": "wechat", "type": "group"}, + "members": [{"platformId": "alice", "accountName": "Alice"}], + "messages": [ + {"sender": "alice", "accountName": "Alice", "timestamp": 1, "type": 0, "content": "one"}, + {"sender": "alice", "accountName": "Alice", "timestamp": 2, "type": 0, "content": "two"} + ] + }"#; + let mut parser = + WasmParser::new("chatlab".to_string(), doc, "browser.json".to_string(), None) + .expect("WASM parser should accept a valid ChatLab export"); + + let meta: Value = + serde_json::from_str(&parser.meta_json()).expect("meta JSON should be valid"); + assert_eq!(meta["name"], "Browser"); + + let members: Value = serde_json::from_str( + &parser + .take_members_json() + .expect("members should serialize"), + ) + .expect("members JSON should be valid"); + assert_eq!(members[0]["platformId"], "alice"); + + let first: Value = serde_json::from_str( + &parser + .take_batch_json(1) + .expect("batch should serialize") + .expect("first batch should exist"), + ) + .expect("batch JSON should be valid"); + assert_eq!(first[0]["content"], "one"); + assert!(parser + .take_batch_json(10) + .expect("batch should serialize") + .is_some()); + assert!(parser + .take_batch_json(10) + .expect("empty batch should succeed") + .is_none()); + assert_eq!( + parser.summary_json(), + r#"{"messageCount":2,"memberCount":1}"# + ); + } +} diff --git a/packages/web-runtime/scripts/bench-import.mts b/packages/web-runtime/scripts/bench-import.mts new file mode 100644 index 00000000..6135325e --- /dev/null +++ b/packages/web-runtime/scripts/bench-import.mts @@ -0,0 +1,451 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import sqlite3InitModule, { type Database, type Sqlite3Static } from '@sqlite.org/sqlite-wasm' +import type { DatabaseAdapter } from '@openchatlab/core' +import { initSync as initParserWasm } from '../src/wasm/generated/parser_native.js' +import { SqliteWasmDatabaseAdapter } from '../src/sqlite/adapter' +import { + parseBrowserImportSource, + type BrowserImportFormatId, + type BrowserParseSource, +} from '../src/import/browser-parser' +import { + BrowserSessionRuntime, + sessionDatabaseFilename, + type WorkspaceDatabasePort, +} from '../src/import/session-runtime' + +const MESSAGE_COUNT = 100_000 +const MEMBER_COUNT = 50 +const BASE_TIMESTAMP = 1_700_000_000 + +initParserWasm({ + module: new Uint8Array(readFileSync(new URL('../src/wasm/generated/parser_native_bg.wasm', import.meta.url))), +}) + +interface BenchmarkFixture { + label: string + formatId: BrowserImportFormatId + source: BrowserParseSource + firstContent: string + lastContent: string + firstTimestamp: number + lastTimestamp: number + chatIndex?: number +} + +class MemoryWorkspaceDatabase implements WorkspaceDatabasePort { + private readonly databases = new Map() + + constructor(private readonly sqlite3: Sqlite3Static) {} + + async withDatabase(filename: string, schemaSql: string, operation: (db: DatabaseAdapter) => T): Promise { + let entry = this.databases.get(filename) + if (!entry) { + const raw = new this.sqlite3.oo1.DB(':memory:', 'c') + entry = { raw, adapter: new SqliteWasmDatabaseAdapter(this.sqlite3, raw) } + this.databases.set(filename, entry) + } + entry.adapter.exec(schemaSql) + return operation(entry.adapter) + } + + async deleteDatabase(filename: string): Promise { + const entry = this.databases.get(filename) + if (!entry) return false + entry.adapter.close() + this.databases.delete(filename) + return true + } + + async ensureCapacity(_minimum: number): Promise { + return 32 + } + + async getDatabaseFilenames(): Promise { + return [...this.databases.keys()] + } + + getDatabase(filename: string): DatabaseAdapter | undefined { + return this.databases.get(filename)?.adapter + } + + dispose(): void { + for (const entry of this.databases.values()) entry.adapter.close() + this.databases.clear() + } +} + +function createChatLabFixture(): BenchmarkFixture { + const lines: string[] = [ + JSON.stringify({ + _type: 'header', + chatlab: { version: '1', exportedAt: BASE_TIMESTAMP }, + meta: { name: 'Web runtime large fixture', platform: 'wechat', type: 'group', ownerId: 'member-0' }, + }), + ] + for (let index = 0; index < MEMBER_COUNT; index += 1) { + lines.push( + JSON.stringify({ + _type: 'member', + platformId: `member-${index}`, + accountName: `Member ${index}`, + }) + ) + } + for (let index = 0; index < MESSAGE_COUNT; index += 1) { + const member = index % MEMBER_COUNT + lines.push( + JSON.stringify({ + _type: 'message', + sender: `member-${member}`, + accountName: `Member ${member}`, + timestamp: BASE_TIMESTAMP + index, + type: 0, + content: `fixture-message-${index}`, + platformMessageId: `fixture-${index}`, + }) + ) + } + + const blob = new Blob([lines.join('\n')], { type: 'application/x-ndjson' }) + return { + label: 'ChatLab JSONL', + formatId: 'chatlab-jsonl', + source: blobSource('web-runtime-large-fixture.jsonl', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp: BASE_TIMESTAMP, + lastTimestamp: BASE_TIMESTAMP + MESSAGE_COUNT - 1, + } +} + +function createChatLabJsonFixture(): BenchmarkFixture { + const members = Array.from({ length: MEMBER_COUNT }, (_, index) => ({ + platformId: `member-${index}`, + accountName: `Member ${index}`, + })) + const messages = Array.from({ length: MESSAGE_COUNT }, (_, index) => { + const member = index % MEMBER_COUNT + return { + sender: `member-${member}`, + accountName: `Member ${member}`, + timestamp: BASE_TIMESTAMP + index, + type: 0, + content: `fixture-message-${index}`, + platformMessageId: `fixture-${index}`, + } + }) + const blob = new Blob([ + JSON.stringify({ + chatlab: { version: '1', exportedAt: BASE_TIMESTAMP }, + meta: { name: 'Web runtime ChatLab JSON fixture', platform: 'wechat', type: 'group', ownerId: 'member-0' }, + members, + messages, + }), + ]) + return { + label: 'ChatLab JSON (Rust WASM)', + formatId: 'chatlab', + source: blobSource('web-runtime-chatlab-wasm.json', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp: BASE_TIMESTAMP, + lastTimestamp: BASE_TIMESTAMP + MESSAGE_COUNT - 1, + } +} + +function createWeFlowFixture(): BenchmarkFixture { + const messages = Array.from({ length: MESSAGE_COUNT }, (_, index) => { + const member = index % MEMBER_COUNT + return { + localId: index + 1, + createTime: BASE_TIMESTAMP + index, + type: '文本消息', + content: `fixture-message-${index}`, + isSend: index === 0 ? 1 : 0, + senderUsername: `member-${member}`, + senderDisplayName: `Member ${member}`, + senderAvatarKey: `member-${member}`, + } + }) + const blob = new Blob([ + JSON.stringify({ + weflow: { version: '1', exportedAt: BASE_TIMESTAMP }, + session: { + wxid: 'benchmark@chatroom', + displayName: 'Web runtime WeFlow fixture', + type: '群聊', + }, + messages, + }), + ]) + return { + label: 'WeFlow JSON (Rust WASM)', + formatId: 'weflow', + source: blobSource('web-runtime-weflow-wasm.json', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp: BASE_TIMESTAMP, + lastTimestamp: BASE_TIMESTAMP + MESSAGE_COUNT - 1, + } +} + +function createWhatsAppFixture(): BenchmarkFixture { + const baseDate = new Date(2024, 0, 1, 0, 0, 0) + const firstTimestamp = Math.floor(baseDate.getTime() / 1000) + const lines = ['Messages and calls are end-to-end encrypted.'] + for (let index = 0; index < MESSAGE_COUNT; index += 1) { + const timestamp = new Date((firstTimestamp + index) * 1000) + lines.push(`${formatWhatsAppTimestamp(timestamp)} - Member ${index % MEMBER_COUNT}: fixture-message-${index}`) + } + + const blob = new Blob([lines.join('\n')], { type: 'text/plain' }) + return { + label: 'WhatsApp TXT', + formatId: 'whatsapp-native-txt', + source: blobSource('WhatsApp benchmark.txt', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp, + lastTimestamp: firstTimestamp + MESSAGE_COUNT - 1, + } +} + +function createLineFixture(): BenchmarkFixture { + const baseDate = new Date(2024, 0, 2, 0, 0, 0) + const firstTimestamp = Math.floor(baseDate.getTime() / 1000) + const lastTimestamp = firstTimestamp + (24 * 60 - 1) * 60 + const lines = ['[LINE] Chat history in Benchmark Team', 'Saved on: 2024/01/03 09:00', '', '2024.01.02 Tuesday'] + for (let index = 0; index < MESSAGE_COUNT; index += 1) { + const minute = index === MESSAGE_COUNT - 1 ? 24 * 60 - 1 : index % (24 * 60 - 1) + lines.push(`${formatLineTime(minute)}\tMember ${index % MEMBER_COUNT}\tfixture-message-${index}`) + } + + const blob = new Blob([lines.join('\n')], { type: 'text/plain' }) + return { + label: 'LINE TXT', + formatId: 'line-native-txt', + source: blobSource('[LINE] Benchmark Team.txt', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp, + lastTimestamp, + } +} + +function createQqFixture(): BenchmarkFixture { + const baseDate = new Date(2024, 0, 3, 0, 0, 0) + const firstTimestamp = Math.floor(baseDate.getTime() / 1000) + const lines = ['消息记录（此消息记录为文本格式，不支持重新导入）', '消息对象:Benchmark Team'] + for (let index = 0; index < MESSAGE_COUNT; index += 1) { + const timestamp = new Date((firstTimestamp + index) * 1000) + const member = index % MEMBER_COUNT + lines.push(`${formatQqTimestamp(timestamp)} Member ${member}(${20_000 + member})`, `fixture-message-${index}`) + } + + const blob = new Blob([lines.join('\n')], { type: 'text/plain' }) + return { + label: 'QQ TXT', + formatId: 'qq-native-txt', + source: blobSource('qq-benchmark.txt', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp, + lastTimestamp: firstTimestamp + MESSAGE_COUNT - 1, + } +} + +function createTelegramFixture(): BenchmarkFixture { + const messages = Array.from({ length: MESSAGE_COUNT }, (_, index) => { + const member = index % MEMBER_COUNT + return { + id: index + 1, + type: 'message', + date: '2024-01-04T00:00:00', + date_unixtime: String(BASE_TIMESTAMP + index), + from: `Member ${member}`, + from_id: `user${20_000 + member}`, + text: `fixture-message-${index}`, + } + }) + const blob = new Blob( + [ + JSON.stringify({ + name: 'Benchmark Team', + type: 'private_group', + id: 4242, + messages, + }), + ], + { type: 'application/json' } + ) + return { + label: 'Telegram single-chat JSON', + formatId: 'telegram-native-single', + source: blobSource('telegram-benchmark.json', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp: BASE_TIMESTAMP, + lastTimestamp: BASE_TIMESTAMP + MESSAGE_COUNT - 1, + } +} + +function createTelegramFullFixture(): BenchmarkFixture { + const messages = Array.from({ length: MESSAGE_COUNT }, (_, index) => { + const member = index % MEMBER_COUNT + return { + id: index + 1, + type: 'message', + date: '2024-01-04T00:00:00', + date_unixtime: String(BASE_TIMESTAMP + index), + from: `Member ${member}`, + from_id: `user${20_000 + member}`, + text: `fixture-message-${index}`, + } + }) + const blob = new Blob( + [ + JSON.stringify({ + about: 'This file was exported by Telegram Desktop.', + chats: { + list: [ + { name: 'Empty Chat', type: 'personal_chat', id: 1, messages: [] }, + { name: 'Benchmark Team', type: 'private_group', id: 4242, messages }, + ], + }, + }), + ], + { type: 'application/json' } + ) + return { + label: 'Telegram full export JSON', + formatId: 'telegram-native', + source: blobSource('telegram-full-benchmark.json', blob), + firstContent: 'fixture-message-0', + lastContent: `fixture-message-${MESSAGE_COUNT - 1}`, + firstTimestamp: BASE_TIMESTAMP, + lastTimestamp: BASE_TIMESTAMP + MESSAGE_COUNT - 1, + chatIndex: 1, + } +} + +function blobSource(name: string, blob: Blob): BrowserParseSource { + return { + name, + size: blob.size, + type: blob.type, + text: () => blob.text(), + arrayBuffer: () => blob.arrayBuffer(), + slice: (start?: number, end?: number) => blob.slice(start, end), + } +} + +function formatWhatsAppTimestamp(value: Date): string { + const pad = (part: number) => String(part).padStart(2, '0') + return `${value.getFullYear()}/${pad(value.getMonth() + 1)}/${pad(value.getDate())} ${pad(value.getHours())}:${pad(value.getMinutes())}:${pad(value.getSeconds())}` +} + +function formatLineTime(minute: number): string { + const hours = Math.floor(minute / 60) + const minutes = minute % 60 + return `${String(hours).padStart(2, '0')}:${String(minutes).padStart(2, '0')}` +} + +function formatQqTimestamp(value: Date): string { + const pad = (part: number) => String(part).padStart(2, '0') + return `${value.getFullYear()}-${pad(value.getMonth() + 1)}-${pad(value.getDate())} ${pad(value.getHours())}:${pad(value.getMinutes())}:${pad(value.getSeconds())}` +} + +function createFixture(kind: string | undefined): BenchmarkFixture { + if (kind === 'chatlab-json') return createChatLabJsonFixture() + if (kind === 'weflow') return createWeFlowFixture() + if (kind === 'whatsapp') return createWhatsAppFixture() + if (kind === 'line') return createLineFixture() + if (kind === 'qq') return createQqFixture() + if (kind === 'telegram') return createTelegramFixture() + if (kind === 'telegram-full') return createTelegramFullFixture() + return createChatLabFixture() +} + +const sqlite3 = await sqlite3InitModule() +const database = new MemoryWorkspaceDatabase(sqlite3) +const runtime = new BrowserSessionRuntime(database, { + createSessionId: () => 'large-fixture-session', + now: () => BASE_TIMESTAMP, +}) +const fixture = createFixture(process.argv[2]) + +try { + let scanElapsedMs: number | undefined + let parseElapsedMs: number | undefined + if (fixture.formatId === 'telegram-native') { + const scanStartedAt = performance.now() + const chats = await runtime.scanMultiChatSource(fixture.source) + scanElapsedMs = performance.now() - scanStartedAt + assert.deepEqual( + chats.map((chat) => ({ index: chat.index, messageCount: chat.messageCount })), + [ + { index: 0, messageCount: 0 }, + { index: 1, messageCount: MESSAGE_COUNT }, + ] + ) + } + if (fixture.formatId === 'chatlab' || fixture.formatId === 'weflow') { + const parseStartedAt = performance.now() + const parsed = await parseBrowserImportSource(fixture.source, { formatId: fixture.formatId }) + parseElapsedMs = performance.now() - parseStartedAt + assert.equal(parsed.messages.length, MESSAGE_COUNT) + assert.equal(parsed.members.length, MEMBER_COUNT) + assert.equal(parsed.messages[0]?.content, fixture.firstContent) + assert.equal(parsed.messages.at(-1)?.content, fixture.lastContent) + } + const startedAt = performance.now() + const result = await runtime.importSource(fixture.source, { + formatId: fixture.formatId, + chatIndex: fixture.chatIndex, + }) + const elapsedMs = performance.now() - startedAt + const sessions = await runtime.listSessions() + const sessionDb = database.getDatabase(sessionDatabaseFilename(result.sessionId)) + assert.ok(sessionDb) + + const counts = sessionDb.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } + const members = sessionDb.prepare('SELECT COUNT(*) AS count FROM member').get() as { count: number } + const first = sessionDb.prepare('SELECT content, ts FROM message ORDER BY id ASC LIMIT 1').get() + const last = sessionDb.prepare('SELECT content, ts FROM message ORDER BY id DESC LIMIT 1').get() + + assert.equal(result.messageCount, MESSAGE_COUNT) + assert.equal(result.memberCount, MEMBER_COUNT) + assert.equal(counts.count, MESSAGE_COUNT) + assert.equal(members.count, MEMBER_COUNT) + assert.deepEqual(first, { content: fixture.firstContent, ts: fixture.firstTimestamp }) + assert.deepEqual(last, { + content: fixture.lastContent, + ts: fixture.lastTimestamp, + }) + assert.equal(sessions.length, 1) + assert.equal(sessions[0]?.messageCount, MESSAGE_COUNT) + assert.equal(sessions[0]?.lastMessageTs, fixture.lastTimestamp) + + console.log( + JSON.stringify( + { + fixture: fixture.label, + fixtureBytes: fixture.source.size, + messages: MESSAGE_COUNT, + members: MEMBER_COUNT, + elapsedMs: Number(elapsedMs.toFixed(1)), + parseElapsedMs: parseElapsedMs === undefined ? undefined : Number(parseElapsedMs.toFixed(1)), + scanElapsedMs: scanElapsedMs === undefined ? undefined : Number(scanElapsedMs.toFixed(1)), + rowsPerSecond: Math.round(MESSAGE_COUNT / (elapsedMs / 1000)), + correctness: 'passed', + persistence: 'not measured (in-memory VFS)', + }, + null, + 2 + ) + ) +} finally { + database.dispose() +} diff --git a/packages/web-runtime/src/import/browser-parser.ts b/packages/web-runtime/src/import/browser-parser.ts index 250a56ef..b7b5ba61 100644 --- a/packages/web-runtime/src/import/browser-parser.ts +++ b/packages/web-runtime/src/import/browser-parser.ts @@ -24,12 +24,7 @@ import { type BrowserParseSource, type ChatLabBrowserFormatId, } from './chatlab-parser' - -export interface BrowserImportLogEvent { - level: 'debug' | 'info' | 'error' - message: string - data?: Record -} +import { parseWithWasm, type BrowserImportLogEvent, type BrowserWasmParserLoader } from './wasm-parser' export type BrowserImportFormatId = | ChatLabBrowserFormatId @@ -46,6 +41,7 @@ export interface ParseBrowserSourceOptions { checkCancelled?: () => void onProgress?: (progress: BrowserChatParseProgress) => void yieldEvery?: number + wasmLoader?: BrowserWasmParserLoader onLog?: (event: BrowserImportLogEvent) => void } @@ -97,6 +93,16 @@ export async function parseBrowserImportSource( ) } + if (formatId === 'chatlab' || formatId === 'weflow') { + const wasmResult = await parseWithWasm(source, formatId, { + checkCancelled: options.checkCancelled, + onProgress: options.onProgress, + onLog: options.onLog, + loader: options.wasmLoader, + }) + if (wasmResult) return wasmResult + } + if (formatId === 'whatsapp-native-txt') { options.checkCancelled?.() const content = await source.text() @@ -211,3 +217,4 @@ export async function parseBrowserImportSource( } export type { BrowserParseSource } from './chatlab-parser' +export type { BrowserImportLogEvent, BrowserWasmParserLoader } from './wasm-parser' diff --git a/packages/web-runtime/src/import/session-runtime.test.ts b/packages/web-runtime/src/import/session-runtime.test.ts index 2bf56db4..0057b67e 100644 --- a/packages/web-runtime/src/import/session-runtime.test.ts +++ b/packages/web-runtime/src/import/session-runtime.test.ts @@ -152,6 +152,7 @@ describe('BrowserSessionRuntime', () => { 'Messages and calls are end-to-end encrypted.', '2024/01/02 03:04 - Alice: hello whatsapp', '2024/01/02 03:05 - Bob: second message', + '2024/01/02 03:06 - Alice changed the group description', '', ].join('\n') ) @@ -172,8 +173,8 @@ describe('BrowserSessionRuntime', () => { assert.deepEqual(result, { sessionId: 'whatsapp-session', formatId: 'whatsapp-native-txt', - messageCount: 2, - memberCount: 2, + messageCount: 3, + memberCount: 3, skippedCount: 0, }) assert.deepEqual(await runtime.listSessions(), [ @@ -183,12 +184,12 @@ describe('BrowserSessionRuntime', () => { platform: 'whatsapp', type: 'private', importedAt: 100, - messageCount: 2, - memberCount: 2, + messageCount: 3, + memberCount: 3, groupId: null, groupAvatar: null, ownerId: null, - lastMessageTs: Math.floor(new Date('2024-01-02T03:05:00').getTime() / 1000), + lastMessageTs: Math.floor(new Date('2024-01-02T03:06:00').getTime() / 1000), formatId: 'whatsapp-native-txt', }, ]) @@ -196,6 +197,7 @@ describe('BrowserSessionRuntime', () => { assert.deepEqual(sessionDb?.prepare('SELECT sender_account_name, content FROM message ORDER BY id').all(), [ { sender_account_name: 'Alice', content: 'hello whatsapp' }, { sender_account_name: 'Bob', content: 'second message' }, + { sender_account_name: '系统消息', content: 'Alice changed the group description' }, ]) database.dispose() }) @@ -231,7 +233,7 @@ describe('BrowserSessionRuntime', () => { }, { localId: 2, - createTime: 1_704_164_646, + createTime: '1704164646', type: '图片消息', content: '[图片]', isSend: 0, @@ -239,6 +241,15 @@ describe('BrowserSessionRuntime', () => { senderDisplayName: 'Bob', senderAvatarKey: 'bob', }, + { + localId: 3, + createTime: null, + type: '文本消息', + content: 'invalid timestamp', + isSend: 0, + senderUsername: 'alice', + senderDisplayName: 'Alice', + }, ], }) @@ -260,7 +271,7 @@ describe('BrowserSessionRuntime', () => { formatId: 'weflow', messageCount: 2, memberCount: 2, - skippedCount: 0, + skippedCount: 1, }) assert.deepEqual( (await runtime.listSessions()).map((session) => ({ @@ -304,6 +315,7 @@ describe('BrowserSessionRuntime', () => { '2024.01.02 Tuesday', '03:04\tAlice\thello line', '03:05\tBob\t[Sticker]', + '03:06\t\tAlice joined the group', '', ].join('\n') ) @@ -324,8 +336,8 @@ describe('BrowserSessionRuntime', () => { assert.deepEqual(result, { sessionId: 'line-session', formatId: 'line-native-txt', - messageCount: 2, - memberCount: 2, + messageCount: 3, + memberCount: 3, skippedCount: 0, }) assert.deepEqual( @@ -341,6 +353,7 @@ describe('BrowserSessionRuntime', () => { assert.deepEqual(sessionDb?.prepare('SELECT sender_account_name, type, content FROM message ORDER BY id').all(), [ { sender_account_name: 'Alice', type: 0, content: 'hello line' }, { sender_account_name: 'Bob', type: 5, content: '[Sticker]' }, + { sender_account_name: '系統', type: 80, content: 'Alice joined the group' }, ]) database.dispose() }) diff --git a/packages/web-runtime/src/import/session-runtime.ts b/packages/web-runtime/src/import/session-runtime.ts index fe4c4c80..5b6e7759 100644 --- a/packages/web-runtime/src/import/session-runtime.ts +++ b/packages/web-runtime/src/import/session-runtime.ts @@ -13,6 +13,7 @@ import { scanBrowserMultiChatSource, type BrowserImportFormatId, type BrowserImportLogEvent, + type BrowserImportParseResult, type BrowserParseSource, } from './browser-parser' import { BrowserSessionCatalog, type BrowserSessionCatalogItem } from './session-catalog' @@ -139,10 +140,25 @@ export class BrowserSessionRuntime { onProgress: (progress) => options.onProgress?.(progress), onLog: options.onLog, }) - if (parsed.messages.length === 0) { + options.checkCancelled?.() + const messages: BrowserImportParseResult['messages'] = [] + for (const message of parsed.messages) { + const timestamp = normalizeImportTimestamp(message.timestamp) + if (timestamp === null) continue + messages.push(timestamp === message.timestamp ? message : { ...message, timestamp }) + } + const skippedInvalidTimestampCount = parsed.messages.length - messages.length + if (messages.length === 0) { throw new WebRuntimeError('EMPTY_IMPORT_FILE', 'The import file does not contain any messages') } - options.checkCancelled?.() + if (skippedInvalidTimestampCount > 0) { + options.onLog?.({ + level: 'info', + message: 'Invalid browser import timestamps skipped', + data: { formatId, skippedCount: skippedInvalidTimestampCount }, + }) + } + const members = mergeInferredMembers(parsed.members, messages) const sessionId = this.createSessionId() validateSessionId(sessionId) @@ -154,12 +170,12 @@ export class BrowserSessionRuntime { platform: parsed.meta.platform, type: parsed.meta.type, importedAt, - messageCount: parsed.messages.length, - memberCount: parsed.members.length, + messageCount: messages.length, + memberCount: members.length, groupId: parsed.meta.groupId ?? null, groupAvatar: parsed.meta.groupAvatar ?? null, ownerId: parsed.meta.ownerId ?? null, - lastMessageTs: parsed.messages.reduce( + lastMessageTs: messages.reduce( (latest, message) => (latest === null || message.timestamp > latest ? message.timestamp : latest), null ), @@ -169,15 +185,18 @@ export class BrowserSessionRuntime { const existingCount = await this.catalog.count() await this.database.ensureCapacity(Math.max(8, (existingCount + 2) * 3)) await this.catalog.beginImport(item) - options.onProgress?.({ stage: 'catalog', progress: 0.55, messagesProcessed: parsed.messages.length }) + options.onProgress?.({ stage: 'catalog', progress: 0.55, messagesProcessed: messages.length }) try { options.checkCancelled?.() options.onProgress?.({ stage: 'saving', progress: 0.6, messagesProcessed: 0 }) const stats = await this.database.withDatabase(filename, CHAT_DB_TABLES, (db) => { - const result = writeParseResultToDb(db, parsed.meta, parsed.members, parsed.messages) + const result = writeParseResultToDb(db, parsed.meta, members, messages) db.exec(CHAT_DB_INDEXES) - return result + return { + ...result, + skippedCount: result.skippedCount + skippedInvalidTimestampCount, + } }) options.checkCancelled?.() await this.catalog.completeImport(sessionId, stats) @@ -226,6 +245,29 @@ export class BrowserSessionRuntime { } } +function normalizeImportTimestamp(value: unknown): number | null { + if (typeof value === 'number') return Number.isFinite(value) ? value : null + if (typeof value !== 'string' || value.trim() === '') return null + const timestamp = Number(value) + return Number.isFinite(timestamp) ? timestamp : null +} + +function mergeInferredMembers( + members: BrowserImportParseResult['members'], + messages: BrowserImportParseResult['messages'] +): BrowserImportParseResult['members'] { + const memberMap = new Map(members.map((member) => [member.platformId, member])) + for (const message of messages) { + if (memberMap.has(message.senderPlatformId)) continue + memberMap.set(message.senderPlatformId, { + platformId: message.senderPlatformId, + accountName: message.senderAccountName, + groupNickname: message.senderGroupNickname, + }) + } + return Array.from(memberMap.values()) +} + function defaultSessionId(): string { if (typeof crypto?.randomUUID !== 'function') { throw new WebRuntimeError('RANDOM_UUID_UNAVAILABLE', 'crypto.randomUUID is required to create a browser session') diff --git a/packages/web-runtime/src/import/wasm-parser.test.ts b/packages/web-runtime/src/import/wasm-parser.test.ts new file mode 100644 index 00000000..21611845 --- /dev/null +++ b/packages/web-runtime/src/import/wasm-parser.test.ts @@ -0,0 +1,152 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import { describe, it } from 'node:test' +import initWasm, { initSync, WasmParser } from '../wasm/generated/parser_native.js' +import { WebRuntimeError } from '../runtime-error' +import { parseBrowserImportSource } from './browser-parser' +import type { BrowserParseSource } from './chatlab-parser' +import type { BrowserWasmParserLoader } from './wasm-parser' + +const wasmBytes = new Uint8Array(readFileSync(new URL('../wasm/generated/parser_native_bg.wasm', import.meta.url))) +initSync({ module: wasmBytes }) + +const wasmLoader: BrowserWasmParserLoader = async () => ({ default: initWasm, WasmParser }) + +function source(name: string, value: unknown): BrowserParseSource { + const blob = new Blob([typeof value === 'string' ? value : JSON.stringify(value)]) + return { + name, + size: blob.size, + type: 'application/json', + text: () => blob.text(), + arrayBuffer: () => blob.arrayBuffer(), + slice: (start, end) => blob.slice(start, end), + } +} + +describe('browser Rust WASM parser', () => { + it('parses ChatLab JSON and WeFlow JSON through the generated WASM module', async () => { + const chatlabSource = source('chatlab.json', { + chatlab: { version: '1' }, + meta: { name: 'ChatLab', platform: 'wechat', type: 'group' }, + members: [{ platformId: 'alice', accountName: 'Alice' }], + messages: [ + { + sender: 'alice', + accountName: 'Alice', + timestamp: 1, + type: 0, + content: 'hello chatlab', + }, + ], + }) + const chatlab = await parseBrowserImportSource(chatlabSource, { formatId: 'chatlab', wasmLoader }) + const chatlabTs = await parseBrowserImportSource(chatlabSource, { + formatId: 'chatlab', + wasmLoader: async () => null, + }) + assert.equal(chatlab.meta.name, 'ChatLab') + assert.deepEqual( + chatlab.members.map((member) => member.platformId), + ['alice'] + ) + assert.deepEqual( + chatlab.messages.map((message) => message.content), + ['hello chatlab'] + ) + assert.deepEqual(chatlab, chatlabTs) + + const weflowSource = source('weflow.json', { + weflow: { version: '1' }, + session: { wxid: 'room@chatroom', displayName: 'WeFlow', type: '群聊' }, + messages: [ + { + localId: 1, + createTime: 2, + type: '文本消息', + content: 'hello weflow', + isSend: 1, + senderUsername: 'bob', + senderDisplayName: 'Bob', + }, + ], + }) + const weflow = await parseBrowserImportSource(weflowSource, { formatId: 'weflow', wasmLoader }) + const weflowTs = await parseBrowserImportSource(weflowSource, { + formatId: 'weflow', + wasmLoader: async () => null, + }) + assert.deepEqual(weflow.meta, { + name: 'WeFlow', + platform: 'weixin', + type: 'group', + groupId: 'room@chatroom', + groupAvatar: undefined, + ownerId: 'bob', + }) + assert.deepEqual( + weflow.messages.map((message) => message.content), + ['hello weflow'] + ) + assert.deepEqual(weflow, weflowTs) + }) + + it('falls back to TS when the strict Rust kernel rejects an off-spec ChatLab message', async () => { + const logs: string[] = [] + const result = await parseBrowserImportSource( + source('fallback.json', { + chatlab: { version: '1' }, + meta: { name: 'Fallback', platform: 'wechat', type: 'group' }, + messages: [{ sender: 'alice', timestamp: 1, type: 0, content: 'TS handles the missing account name' }], + }), + { + formatId: 'chatlab', + wasmLoader, + onLog: (event) => logs.push(`${event.level}:${event.message}`), + } + ) + + assert.equal(result.messages[0]?.senderAccountName, 'alice') + assert.ok(logs.some((message) => message.includes('info:Rust WASM parse failed; falling back to TS'))) + }) + + it('falls back to TS when the WASM module cannot initialize', async () => { + const result = await parseBrowserImportSource( + source('init-fallback.json', { + chatlab: { version: '1' }, + meta: { name: 'Init fallback', platform: 'wechat', type: 'group' }, + messages: [{ sender: 'alice', accountName: 'Alice', timestamp: 1, type: 0, content: 'fallback' }], + }), + { + formatId: 'chatlab', + wasmLoader: async () => { + throw new Error('WASM initialization failed') + }, + } + ) + + assert.equal(result.messages[0]?.content, 'fallback') + }) + + it('propagates cancellation after synchronous WASM parsing instead of falling back', async () => { + let checks = 0 + await assert.rejects( + parseBrowserImportSource( + source('cancelled.json', { + chatlab: { version: '1' }, + meta: { name: 'Cancelled', platform: 'wechat', type: 'group' }, + messages: [{ sender: 'alice', accountName: 'Alice', timestamp: 1, type: 0, content: 'stop' }], + }), + { + formatId: 'chatlab', + wasmLoader, + checkCancelled: () => { + checks += 1 + if (checks >= 3) throw new WebRuntimeError('REQUEST_CANCELLED', 'cancelled') + }, + } + ), + (error: unknown) => error instanceof WebRuntimeError && error.code === 'REQUEST_CANCELLED' + ) + }) +}) diff --git a/packages/web-runtime/src/import/wasm-parser.ts b/packages/web-runtime/src/import/wasm-parser.ts new file mode 100644 index 00000000..40bf17b8 --- /dev/null +++ b/packages/web-runtime/src/import/wasm-parser.ts @@ -0,0 +1,248 @@ +import type { + BrowserChatParseProgress, + BrowserParsedMember, + BrowserParsedMessage, + BrowserParseSource, +} from './chatlab-parser' +import type { BrowserImportParseResult } from './browser-parser' + +type BrowserWasmFormatId = 'chatlab' | 'weflow' + +interface WasmNativeRole { + id: string + name?: string | null +} + +interface WasmNativeMember { + platformId: string + accountName: string + groupNickname?: string | null + aliases?: string[] | null + avatar?: string | null + roles?: WasmNativeRole[] | null +} + +interface WasmNativeMessage { + platformMessageId?: string | null + senderPlatformId: string + senderAccountName: string + senderGroupNickname?: string | null + timestamp?: number | null + messageType: number + content?: string | null + replyToMessageId?: string | null +} + +interface BrowserWasmParser { + free(): void + meta_json(): string + take_members_json(): string + take_batch_json(size: number): string | undefined + summary_json(): string +} + +export interface BrowserWasmParserModule { + default(moduleOrPath?: unknown): Promise + WasmParser: new ( + formatId: string, + bytes: Uint8Array, + fileName: string, + optionsJson?: string | null + ) => BrowserWasmParser +} + +export type BrowserWasmParserLoader = () => Promise + +export interface BrowserImportLogEvent { + level: 'debug' | 'info' | 'error' + message: string + data?: Record +} + +export interface ParseWithWasmOptions { + checkCancelled?: () => void + onProgress?: (progress: BrowserChatParseProgress) => void + onLog?: (event: BrowserImportLogEvent) => void + loader?: BrowserWasmParserLoader + batchSize?: number +} + +let modulePromise: Promise | undefined + +export async function parseWithWasm( + source: BrowserParseSource, + formatId: BrowserWasmFormatId, + options: ParseWithWasmOptions = {} +): Promise { + const startedAt = performance.now() + let parser: BrowserWasmParser | undefined + try { + options.checkCancelled?.() + const module = await (options.loader ?? loadDefaultModule)() + if (!module) return null + await module.default() + const bytes = new Uint8Array(await source.arrayBuffer()) + options.checkCancelled?.() + options.onProgress?.({ stage: 'parsing', progress: 0, messagesProcessed: 0 }) + parser = new module.WasmParser(formatId, bytes, source.name) + + await yieldToWorkerQueue() + options.checkCancelled?.() + + const meta = JSON.parse(parser.meta_json()) as Record + const members = mapMembers(JSON.parse(parser.take_members_json()) as WasmNativeMember[], formatId, meta) + const summary = JSON.parse(parser.summary_json()) as { messageCount: number } + const messages: BrowserParsedMessage[] = [] + const batchSize = Math.max(1, options.batchSize ?? 5000) + + while (true) { + const batchJson = parser.take_batch_json(batchSize) + if (batchJson === undefined) break + const batch = JSON.parse(batchJson) as WasmNativeMessage[] + messages.push(...batch.map((message) => mapMessage(message, formatId))) + options.onProgress?.({ + stage: 'parsing', + progress: summary.messageCount === 0 ? 1 : messages.length / summary.messageCount, + messagesProcessed: messages.length, + }) + await yieldToWorkerQueue() + options.checkCancelled?.() + } + + const result = mapResult(formatId, meta, members, messages) + options.onLog?.({ + level: 'info', + message: 'Rust WASM parse completed', + data: { + formatId, + size: source.size, + durationMs: Math.round(performance.now() - startedAt), + messageCount: messages.length, + memberCount: members.length, + }, + }) + return result + } catch (error) { + options.checkCancelled?.() + options.onLog?.({ + level: 'info', + message: 'Rust WASM parse failed; falling back to TS', + data: { + formatId, + error: error instanceof Error ? error.message : String(error), + }, + }) + return null + } finally { + parser?.free() + } +} + +function loadDefaultModule(): Promise { + modulePromise ??= import('../wasm/generated/parser_native.js').catch((error: unknown) => { + modulePromise = undefined + throw error + }) + return modulePromise +} + +function mapResult( + formatId: BrowserWasmFormatId, + meta: Record, + members: BrowserParsedMember[], + messages: BrowserParsedMessage[] +): BrowserImportParseResult { + if (formatId === 'weflow') { + return { + formatId, + meta: { + name: meta.name as string, + platform: 'weixin', + type: meta.chatType === 'private' ? 'private' : 'group', + groupId: optionalString(meta.groupId), + groupAvatar: optionalString(meta.groupAvatar), + ownerId: optionalString(meta.ownerId), + }, + members, + messages, + } + } + + return { + formatId, + meta: compact({ + name: meta.name as string, + platform: meta.platform as string, + type: meta.chatType as string, + groupId: optionalString(meta.groupId), + groupAvatar: optionalString(meta.groupAvatar), + ownerId: optionalString(meta.ownerId), + sourceSessionId: optionalString(meta.sourceSessionId), + }), + members, + messages, + } +} + +function mapMembers( + members: WasmNativeMember[], + formatId: BrowserWasmFormatId, + meta: Record +): BrowserParsedMember[] { + if (formatId === 'weflow') { + return members.map((member) => ({ + platformId: member.platformId, + accountName: member.accountName, + avatar: optionalString(member.avatar), + })) + } + + const fromHead = meta.membersFromHead === true + return members.map((member) => { + if (!fromHead) { + return { + platformId: member.platformId, + accountName: member.accountName, + groupNickname: optionalString(member.groupNickname), + } + } + return compact({ + platformId: member.platformId, + accountName: member.accountName, + groupNickname: optionalString(member.groupNickname), + aliases: member.aliases ?? undefined, + avatar: optionalString(member.avatar), + roles: member.roles?.map((role) => ({ id: role.id, name: optionalString(role.name) })) ?? undefined, + }) + }) +} + +function mapMessage(message: WasmNativeMessage, formatId: BrowserWasmFormatId): BrowserParsedMessage { + const mapped = { + platformMessageId: optionalString(message.platformMessageId), + senderPlatformId: message.senderPlatformId, + senderAccountName: message.senderAccountName, + senderGroupNickname: optionalString(message.senderGroupNickname), + timestamp: message.timestamp as number, + type: message.messageType, + content: message.content ?? null, + replyToMessageId: optionalString(message.replyToMessageId), + } + if (formatId === 'weflow') { + const { replyToMessageId: _replyToMessageId, ...weflowMessage } = mapped + return weflowMessage + } + return compact(mapped) +} + +function optionalString(value: unknown): string | undefined { + return typeof value === 'string' ? value : undefined +} + +function compact>(value: T): T { + return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined)) as T +} + +function yieldToWorkerQueue(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)) +} diff --git a/packages/web-runtime/src/rpc/client.test.ts b/packages/web-runtime/src/rpc/client.test.ts index 4d8078e1..6a6c8799 100644 --- a/packages/web-runtime/src/rpc/client.test.ts +++ b/packages/web-runtime/src/rpc/client.test.ts @@ -117,12 +117,21 @@ describe('WebRuntimeRpcClient', () => { client.dispose() }) - it('sends cancellation through the same channel and rejects locally', async () => { + it('waits for worker cancellation acknowledgement before rejecting', async () => { const worker = new FakeWorker() const client = new WebRuntimeRpcClient(() => worker) const controller = new AbortController() const resultPromise = client.request('db.open', { filename: '/cancelled.db' }, { signal: controller.signal }) const request = worker.messages[0] as { id: string } + let settled = false + void resultPromise.then( + () => { + settled = true + }, + () => { + settled = true + } + ) controller.abort('user cancelled') @@ -131,10 +140,43 @@ describe('WebRuntimeRpcClient', () => { type: 'cancel', payload: { reason: 'user cancelled' }, }) + await Promise.resolve() + assert.equal(settled, false) + + worker.emitMessage({ + id: request.id, + type: 'error', + payload: { + taskType: 'db.open', + error: { name: 'WebRuntimeError', code: 'REQUEST_CANCELLED', message: 'The Worker task was cancelled' }, + }, + }) await assert.rejects(resultPromise, (error: unknown) => (error as Error).name === 'AbortError') client.dispose() }) + it('returns a mutating task result when it completes before cancellation acknowledgement', async () => { + const worker = new FakeWorker() + const client = new WebRuntimeRpcClient(() => worker) + const controller = new AbortController() + const resultPromise = client.request( + 'session.rename', + { sessionId: 'session-one', name: 'Renamed' }, + { signal: controller.signal } + ) + const request = worker.messages[0] as { id: string } + + controller.abort('user cancelled') + worker.emitMessage({ + id: request.id, + type: 'result', + payload: { taskType: 'session.rename', result: { renamed: true } }, + }) + + assert.deepEqual(await resultPromise, { renamed: true }) + client.dispose() + }) + it('rejects pending work after a crash and creates a fresh worker for the next request', async () => { const workers: FakeWorker[] = [] const client = new WebRuntimeRpcClient(() => { diff --git a/packages/web-runtime/src/rpc/client.ts b/packages/web-runtime/src/rpc/client.ts index d77bb01b..7d6de50c 100644 --- a/packages/web-runtime/src/rpc/client.ts +++ b/packages/web-runtime/src/rpc/client.ts @@ -31,6 +31,8 @@ interface PendingRequest { reject(error: Error): void onProgress?: (event: RpcProgressPayload) => void removeAbortListener?: () => void + abortRequested?: boolean + abortReason?: unknown } export class WebRuntimeRpcError extends Error { @@ -80,12 +82,17 @@ export class WebRuntimeRpcClient { if (options.signal) { const handleAbort = () => { if (!this.pending.has(id)) return - worker.postMessage({ - id, - type: 'cancel', - payload: { reason: normalizeAbortReason(options.signal?.reason) }, - }) - this.settleRequest(id, 'reject', createAbortError(options.signal?.reason)) + pending.abortRequested = true + pending.abortReason = options.signal?.reason + try { + worker.postMessage({ + id, + type: 'cancel', + payload: { reason: normalizeAbortReason(pending.abortReason) }, + }) + } catch (error) { + this.settleRequest(id, 'reject', normalizeError(error, 'POST_MESSAGE_FAILED')) + } } options.signal.addEventListener('abort', handleAbort, { once: true }) pending.removeAbortListener = () => options.signal?.removeEventListener('abort', handleAbort) @@ -126,13 +133,17 @@ export class WebRuntimeRpcClient { if (!pending) return if (envelope.type === 'progress') { - pending.onProgress?.(envelope.payload) + if (!pending.abortRequested) pending.onProgress?.(envelope.payload) return } if (envelope.type === 'result') { this.settleRequest(envelope.id, 'resolve', envelope.payload.result) return } + if (pending.abortRequested && envelope.payload.error.code === 'REQUEST_CANCELLED') { + this.settleRequest(envelope.id, 'reject', createAbortError(pending.abortReason)) + return + } this.settleRequest(envelope.id, 'reject', new WebRuntimeRpcError(envelope.payload.error)) } diff --git a/packages/web-runtime/src/sqlite/database-runtime.test.ts b/packages/web-runtime/src/sqlite/database-runtime.test.ts index 4ec14880..6e0aba0e 100644 --- a/packages/web-runtime/src/sqlite/database-runtime.test.ts +++ b/packages/web-runtime/src/sqlite/database-runtime.test.ts @@ -4,7 +4,7 @@ import sqlite3InitModule, { type Database, type SAHPoolUtil } from '@sqlite.org/ import { CURRENT_SCHEMA_VERSION, getHourlyActivity } from '@openchatlab/core' import { BrowserDatabaseRuntime } from './database-runtime' -async function createMemoryRuntime(): Promise<{ +async function createMemoryRuntime(options: { initializationFailures?: number } = {}): Promise<{ runtime: BrowserDatabaseRuntime openedDatabases: Database[] filenames: Set @@ -28,8 +28,13 @@ async function createMemoryRuntime(): Promise<{ reserveMinimumCapacity: async (minimum: number) => minimum, getFileNames: () => [...filenames], } as unknown as SAHPoolUtil + let remainingInitializationFailures = options.initializationFailures ?? 0 return { runtime: new BrowserDatabaseRuntime(async (onStage) => { + if (remainingInitializationFailures > 0) { + remainingInitializationFailures -= 1 + throw new Error('simulated sqlite initialization failure') + } onStage?.('sqlite-initializing') onStage?.('sqlite-ready') onStage?.('opfs-pool-initializing') @@ -42,6 +47,14 @@ async function createMemoryRuntime(): Promise<{ } describe('BrowserDatabaseRuntime', () => { + it('retries SQLite initialization after a previous attempt fails', async () => { + const { runtime } = await createMemoryRuntime({ initializationFailures: 1 }) + + await assert.rejects(runtime.open('/session.db'), /simulated sqlite initialization failure/) + await assert.doesNotReject(runtime.open('/session.db')) + await runtime.close() + }) + it('opens one absolute database, initializes the core schema, and closes it', async () => { const { runtime, openedDatabases } = await createMemoryRuntime() diff --git a/packages/web-runtime/src/sqlite/database-runtime.ts b/packages/web-runtime/src/sqlite/database-runtime.ts index 4add31c3..9ac88292 100644 --- a/packages/web-runtime/src/sqlite/database-runtime.ts +++ b/packages/web-runtime/src/sqlite/database-runtime.ts @@ -133,11 +133,16 @@ export class BrowserDatabaseRuntime { return this.database } - private getInitializedRuntime( + private async getInitializedRuntime( onStage?: (stage: SqliteInitializationStage) => void ): Promise { - this.initialized ??= this.initializeSqlite(onStage) - return this.initialized + const initialized = (this.initialized ??= this.initializeSqlite(onStage)) + try { + return await initialized + } catch (error) { + if (this.initialized === initialized) this.initialized = undefined + throw error + } } private buildOpenResult(filename: string, runtime: InitializedSqliteRuntime): OpenDatabaseResult { diff --git a/packages/web-runtime/src/sqlite/opfs.ts b/packages/web-runtime/src/sqlite/opfs.ts index 8a7bd966..f5e32ff5 100644 --- a/packages/web-runtime/src/sqlite/opfs.ts +++ b/packages/web-runtime/src/sqlite/opfs.ts @@ -14,6 +14,10 @@ export type SqliteInitializationStage = | 'opfs-pool-initializing' | 'opfs-pool-ready' +type OpfsSahPoolOptions = Parameters[0] & { + forceReinitIfPreviouslyFailed?: boolean +} + export async function initializeOpfsSqlite( onStage?: (stage: SqliteInitializationStage) => void ): Promise { @@ -26,10 +30,13 @@ export async function initializeOpfsSqlite( onStage?.('sqlite-ready') onStage?.('opfs-pool-initializing') - const pool = await sqlite3.installOpfsSAHPoolVfs({ + // sqlite-wasm supports this retry option at runtime, but its published types do not declare it yet. + const poolOptions: OpfsSahPoolOptions = { directory: WEB_RUNTIME_SAHPOOL_DIRECTORY, initialCapacity: 8, - }) + forceReinitIfPreviouslyFailed: true, + } + const pool = await sqlite3.installOpfsSAHPoolVfs(poolOptions) onStage?.('opfs-pool-ready') return { sqlite3, pool } diff --git a/packages/web-runtime/src/wasm/generated/parser_native.d.ts b/packages/web-runtime/src/wasm/generated/parser_native.d.ts new file mode 100644 index 00000000..6021f94c --- /dev/null +++ b/packages/web-runtime/src/wasm/generated/parser_native.d.ts @@ -0,0 +1,50 @@ +/* tslint:disable */ +/* eslint-disable */ + +export class WasmParser { + free(): void; + [Symbol.dispose](): void; + meta_json(): string; + constructor(format_id: string, bytes: Uint8Array, file_name: string, options_json?: string | null); + summary_json(): string; + take_batch_json(size: number): string | undefined; + take_members_json(): string; +} + +export type InitInput = RequestInfo | URL | Response | BufferSource | WebAssembly.Module; + +export interface InitOutput { + readonly memory: WebAssembly.Memory; + readonly __wbg_wasmparser_free: (a: number, b: number) => void; + readonly wasmparser_new: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number) => void; + readonly wasmparser_meta_json: (a: number, b: number) => void; + readonly wasmparser_take_members_json: (a: number, b: number) => void; + readonly wasmparser_take_batch_json: (a: number, b: number, c: number) => void; + readonly wasmparser_summary_json: (a: number, b: number) => void; + readonly __wbindgen_add_to_stack_pointer: (a: number) => number; + readonly __wbindgen_export: (a: number, b: number, c: number) => void; + readonly __wbindgen_export2: (a: number, b: number) => number; + readonly __wbindgen_export3: (a: number, b: number, c: number, d: number) => number; +} + +export type SyncInitInput = BufferSource | WebAssembly.Module; + +/** + * Instantiates the given `module`, which can either be bytes or + * a precompiled `WebAssembly.Module`. + * + * @param {{ module: SyncInitInput }} module - Passing `SyncInitInput` directly is deprecated. + * + * @returns {InitOutput} + */ +export function initSync(module: { module: SyncInitInput } | SyncInitInput): InitOutput; + +/** + * If `module_or_path` is {RequestInfo} or {URL}, makes a request and + * for everything else, calls `WebAssembly.instantiate` directly. + * + * @param {{ module_or_path: InitInput | Promise }} module_or_path - Passing `InitInput` directly is deprecated. + * + * @returns {Promise} + */ +export default function __wbg_init (module_or_path?: { module_or_path: InitInput | Promise } | InitInput | Promise): Promise; diff --git a/packages/web-runtime/src/wasm/generated/parser_native.js b/packages/web-runtime/src/wasm/generated/parser_native.js new file mode 100644 index 00000000..e161487f --- /dev/null +++ b/packages/web-runtime/src/wasm/generated/parser_native.js @@ -0,0 +1,375 @@ +/* @ts-self-types="./parser_native.d.ts" */ + +export class WasmParser { + __destroy_into_raw() { + const ptr = this.__wbg_ptr; + this.__wbg_ptr = 0; + WasmParserFinalization.unregister(this); + return ptr; + } + free() { + const ptr = this.__destroy_into_raw(); + wasm.__wbg_wasmparser_free(ptr, 0); + } + /** + * @returns {string} + */ + meta_json() { + let deferred1_0; + let deferred1_1; + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + wasm.wasmparser_meta_json(retptr, this.__wbg_ptr); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + deferred1_0 = r0; + deferred1_1 = r1; + return getStringFromWasm0(r0, r1); + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + wasm.__wbindgen_export(deferred1_0, deferred1_1, 1); + } + } + /** + * @param {string} format_id + * @param {Uint8Array} bytes + * @param {string} file_name + * @param {string | null} [options_json] + */ + constructor(format_id, bytes, file_name, options_json) { + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + const ptr0 = passStringToWasm0(format_id, wasm.__wbindgen_export2, wasm.__wbindgen_export3); + const len0 = WASM_VECTOR_LEN; + const ptr1 = passArray8ToWasm0(bytes, wasm.__wbindgen_export2); + const len1 = WASM_VECTOR_LEN; + const ptr2 = passStringToWasm0(file_name, wasm.__wbindgen_export2, wasm.__wbindgen_export3); + const len2 = WASM_VECTOR_LEN; + var ptr3 = isLikeNone(options_json) ? 0 : passStringToWasm0(options_json, wasm.__wbindgen_export2, wasm.__wbindgen_export3); + var len3 = WASM_VECTOR_LEN; + wasm.wasmparser_new(retptr, ptr0, len0, ptr1, len1, ptr2, len2, ptr3, len3); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + var r2 = getDataViewMemory0().getInt32(retptr + 4 * 2, true); + if (r2) { + throw takeObject(r1); + } + this.__wbg_ptr = r0; + WasmParserFinalization.register(this, this.__wbg_ptr, this); + return this; + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + } + } + /** + * @returns {string} + */ + summary_json() { + let deferred1_0; + let deferred1_1; + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + wasm.wasmparser_summary_json(retptr, this.__wbg_ptr); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + deferred1_0 = r0; + deferred1_1 = r1; + return getStringFromWasm0(r0, r1); + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + wasm.__wbindgen_export(deferred1_0, deferred1_1, 1); + } + } + /** + * @param {number} size + * @returns {string | undefined} + */ + take_batch_json(size) { + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + wasm.wasmparser_take_batch_json(retptr, this.__wbg_ptr, size); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + var r2 = getDataViewMemory0().getInt32(retptr + 4 * 2, true); + var r3 = getDataViewMemory0().getInt32(retptr + 4 * 3, true); + if (r3) { + throw takeObject(r2); + } + let v1; + if (r0 !== 0) { + v1 = getStringFromWasm0(r0, r1).slice(); + wasm.__wbindgen_export(r0, r1 * 1, 1); + } + return v1; + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + } + } + /** + * @returns {string} + */ + take_members_json() { + let deferred2_0; + let deferred2_1; + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + wasm.wasmparser_take_members_json(retptr, this.__wbg_ptr); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + var r2 = getDataViewMemory0().getInt32(retptr + 4 * 2, true); + var r3 = getDataViewMemory0().getInt32(retptr + 4 * 3, true); + var ptr1 = r0; + var len1 = r1; + if (r3) { + ptr1 = 0; len1 = 0; + throw takeObject(r2); + } + deferred2_0 = ptr1; + deferred2_1 = len1; + return getStringFromWasm0(ptr1, len1); + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + wasm.__wbindgen_export(deferred2_0, deferred2_1, 1); + } + } +} +if (Symbol.dispose) WasmParser.prototype[Symbol.dispose] = WasmParser.prototype.free; +function __wbg_get_imports() { + const import0 = { + __proto__: null, + __wbg___wbindgen_throw_344f42d3211c4765: function(arg0, arg1) { + throw new Error(getStringFromWasm0(arg0, arg1)); + }, + __wbindgen_cast_0000000000000001: function(arg0, arg1) { + // Cast intrinsic for `Ref(String) -> Externref`. + const ret = getStringFromWasm0(arg0, arg1); + return addHeapObject(ret); + }, + }; + return { + __proto__: null, + "./parser_native_bg.js": import0, + }; +} + +const WasmParserFinalization = (typeof FinalizationRegistry === 'undefined') + ? { register: () => {}, unregister: () => {} } + : new FinalizationRegistry(ptr => wasm.__wbg_wasmparser_free(ptr, 1)); + +function addHeapObject(obj) { + if (heap_next === heap.length) heap.push(heap.length + 1); + const idx = heap_next; + heap_next = heap[idx]; + + heap[idx] = obj; + return idx; +} + +function dropObject(idx) { + if (idx < 1028) return; + heap[idx] = heap_next; + heap_next = idx; +} + +let cachedDataViewMemory0 = null; +function getDataViewMemory0() { + if (cachedDataViewMemory0 === null || cachedDataViewMemory0.buffer.detached === true || (cachedDataViewMemory0.buffer.detached === undefined && cachedDataViewMemory0.buffer !== wasm.memory.buffer)) { + cachedDataViewMemory0 = new DataView(wasm.memory.buffer); + } + return cachedDataViewMemory0; +} + +function getStringFromWasm0(ptr, len) { + return decodeText(ptr >>> 0, len); +} + +let cachedUint8ArrayMemory0 = null; +function getUint8ArrayMemory0() { + if (cachedUint8ArrayMemory0 === null || cachedUint8ArrayMemory0.byteLength === 0) { + cachedUint8ArrayMemory0 = new Uint8Array(wasm.memory.buffer); + } + return cachedUint8ArrayMemory0; +} + +function getObject(idx) { return heap[idx]; } + +let heap = new Array(1024).fill(undefined); +heap.push(undefined, null, true, false); + +let heap_next = heap.length; + +function isLikeNone(x) { + return x === undefined || x === null; +} + +function passArray8ToWasm0(arg, malloc) { + const ptr = malloc(arg.length * 1, 1) >>> 0; + getUint8ArrayMemory0().set(arg, ptr / 1); + WASM_VECTOR_LEN = arg.length; + return ptr; +} + +function passStringToWasm0(arg, malloc, realloc) { + if (realloc === undefined) { + const buf = cachedTextEncoder.encode(arg); + const ptr = malloc(buf.length, 1) >>> 0; + getUint8ArrayMemory0().subarray(ptr, ptr + buf.length).set(buf); + WASM_VECTOR_LEN = buf.length; + return ptr; + } + + let len = arg.length; + let ptr = malloc(len, 1) >>> 0; + + const mem = getUint8ArrayMemory0(); + + let offset = 0; + + for (; offset < len; offset++) { + const code = arg.charCodeAt(offset); + if (code > 0x7F) break; + mem[ptr + offset] = code; + } + if (offset !== len) { + if (offset !== 0) { + arg = arg.slice(offset); + } + ptr = realloc(ptr, len, len = offset + arg.length * 3, 1) >>> 0; + const view = getUint8ArrayMemory0().subarray(ptr + offset, ptr + len); + const ret = cachedTextEncoder.encodeInto(arg, view); + + offset += ret.written; + ptr = realloc(ptr, len, offset, 1) >>> 0; + } + + WASM_VECTOR_LEN = offset; + return ptr; +} + +function takeObject(idx) { + const ret = getObject(idx); + dropObject(idx); + return ret; +} + +let cachedTextDecoder = new TextDecoder('utf-8', { ignoreBOM: true, fatal: true }); +cachedTextDecoder.decode(); +const MAX_SAFARI_DECODE_BYTES = 2146435072; +let numBytesDecoded = 0; +function decodeText(ptr, len) { + numBytesDecoded += len; + if (numBytesDecoded >= MAX_SAFARI_DECODE_BYTES) { + cachedTextDecoder = new TextDecoder('utf-8', { ignoreBOM: true, fatal: true }); + cachedTextDecoder.decode(); + numBytesDecoded = len; + } + return cachedTextDecoder.decode(getUint8ArrayMemory0().subarray(ptr, ptr + len)); +} + +const cachedTextEncoder = new TextEncoder(); + +if (!('encodeInto' in cachedTextEncoder)) { + cachedTextEncoder.encodeInto = function (arg, view) { + const buf = cachedTextEncoder.encode(arg); + view.set(buf); + return { + read: arg.length, + written: buf.length + }; + }; +} + +let WASM_VECTOR_LEN = 0; + +let wasmModule, wasmInstance, wasm; +function __wbg_finalize_init(instance, module) { + wasmInstance = instance; + wasm = instance.exports; + wasmModule = module; + cachedDataViewMemory0 = null; + cachedUint8ArrayMemory0 = null; + return wasm; +} + +async function __wbg_load(module, imports) { + if (typeof Response === 'function' && module instanceof Response) { + if (typeof WebAssembly.instantiateStreaming === 'function') { + try { + return await WebAssembly.instantiateStreaming(module, imports); + } catch (e) { + const validResponse = module.ok && expectedResponseType(module.type); + + if (validResponse && module.headers.get('Content-Type') !== 'application/wasm') { + console.warn("`WebAssembly.instantiateStreaming` failed because your server does not serve Wasm with `application/wasm` MIME type. Falling back to `WebAssembly.instantiate` which is slower. Original error:\n", e); + + } else { throw e; } + } + } + + const bytes = await module.arrayBuffer(); + return await WebAssembly.instantiate(bytes, imports); + } else { + const instance = await WebAssembly.instantiate(module, imports); + + if (instance instanceof WebAssembly.Instance) { + return { instance, module }; + } else { + return instance; + } + } + + function expectedResponseType(type) { + switch (type) { + case 'basic': case 'cors': case 'default': return true; + } + return false; + } +} + +function initSync(module) { + if (wasm !== undefined) return wasm; + + + if (module !== undefined) { + if (Object.getPrototypeOf(module) === Object.prototype) { + ({module} = module) + } else { + console.warn('using deprecated parameters for `initSync()`; pass a single object instead') + } + } + + const imports = __wbg_get_imports(); + if (!(module instanceof WebAssembly.Module)) { + module = new WebAssembly.Module(module); + } + const instance = new WebAssembly.Instance(module, imports); + return __wbg_finalize_init(instance, module); +} + +async function __wbg_init(module_or_path) { + if (wasm !== undefined) return wasm; + + + if (module_or_path !== undefined) { + if (Object.getPrototypeOf(module_or_path) === Object.prototype) { + ({module_or_path} = module_or_path) + } else { + console.warn('using deprecated parameters for the initialization function; pass a single object instead') + } + } + + if (module_or_path === undefined) { + module_or_path = new URL('parser_native_bg.wasm', import.meta.url); + } + const imports = __wbg_get_imports(); + + if (typeof module_or_path === 'string' || (typeof Request === 'function' && module_or_path instanceof Request) || (typeof URL === 'function' && module_or_path instanceof URL)) { + module_or_path = fetch(module_or_path); + } + + const { instance, module } = await __wbg_load(await module_or_path, imports); + + return __wbg_finalize_init(instance, module); +} + +export { initSync, __wbg_init as default }; diff --git a/packages/web-runtime/src/wasm/generated/parser_native_bg.wasm b/packages/web-runtime/src/wasm/generated/parser_native_bg.wasm new file mode 100644 index 00000000..465793d4 Binary files /dev/null and b/packages/web-runtime/src/wasm/generated/parser_native_bg.wasm differ diff --git a/packages/web-runtime/src/wasm/generated/parser_native_bg.wasm.d.ts b/packages/web-runtime/src/wasm/generated/parser_native_bg.wasm.d.ts new file mode 100644 index 00000000..4065507c --- /dev/null +++ b/packages/web-runtime/src/wasm/generated/parser_native_bg.wasm.d.ts @@ -0,0 +1,13 @@ +/* tslint:disable */ +/* eslint-disable */ +export const memory: WebAssembly.Memory; +export const __wbg_wasmparser_free: (a: number, b: number) => void; +export const wasmparser_new: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number) => void; +export const wasmparser_meta_json: (a: number, b: number) => void; +export const wasmparser_take_members_json: (a: number, b: number) => void; +export const wasmparser_take_batch_json: (a: number, b: number, c: number) => void; +export const wasmparser_summary_json: (a: number, b: number) => void; +export const __wbindgen_add_to_stack_pointer: (a: number) => number; +export const __wbindgen_export: (a: number, b: number, c: number) => void; +export const __wbindgen_export2: (a: number, b: number) => number; +export const __wbindgen_export3: (a: number, b: number, c: number, d: number) => number; diff --git a/packages/web-runtime/src/worker/controller.test.ts b/packages/web-runtime/src/worker/controller.test.ts index 903a57e4..d354dd14 100644 --- a/packages/web-runtime/src/worker/controller.test.ts +++ b/packages/web-runtime/src/worker/controller.test.ts @@ -214,6 +214,46 @@ describe('WebRuntimeWorkerController', () => { assert.equal(failed.type === 'error' ? failed.payload.error.code : '', 'DB_CLOSE_FAILED') }) + it('returns the result when a catalog mutation finishes after cancellation is requested', async () => { + const sink = new CapturingSink() + const database = new FakeDatabaseRuntime() + const sessions = new FakeSessionRuntime() + let markRenameStarted!: () => void + let releaseRename!: () => void + const renameStarted = new Promise((resolve) => { + markRenameStarted = resolve + }) + const renameGate = new Promise((resolve) => { + releaseRename = resolve + }) + sessions.renameSession = async () => { + markRenameStarted() + await renameGate + return true + } + const controller = new WebRuntimeWorkerController(sink, database, () => supportedCapabilities, sessions) + + controller.handleMessage({ + id: 'rename-cancelled-after-write', + type: 'session.rename', + payload: { sessionId: 'session-one', name: 'Renamed' }, + }) + await renameStarted + controller.handleMessage({ + id: 'rename-cancelled-after-write', + type: 'cancel', + payload: { reason: 'cancelled' }, + }) + releaseRename() + + const completed = await waitForMessage(sink, 'rename-cancelled-after-write', 'result') + assert.deepEqual(completed.type === 'result' ? completed.payload.result : null, { renamed: true }) + assert.equal( + sink.messages.some((message) => message.id === 'rename-cancelled-after-write' && message.type === 'error'), + false + ) + }) + it('routes import progress and session catalog tasks through the session runtime', async () => { const sink = new CapturingSink() const database = new FakeDatabaseRuntime() diff --git a/packages/web-runtime/src/worker/controller.ts b/packages/web-runtime/src/worker/controller.ts index 0bf3bf36..449cfba4 100644 --- a/packages/web-runtime/src/worker/controller.ts +++ b/packages/web-runtime/src/worker/controller.ts @@ -64,7 +64,6 @@ export class WebRuntimeWorkerController { try { this.throwIfCancelled(request.id) const result = await this.dispatch(request) - this.throwIfCancelled(request.id) this.sink.postMessage({ id: request.id, type: 'result',