ziex-dev-mirror/ziex
1
0
Fork 0
mirror of https://github.com/ziex-dev/ziex.git synced 2026-07-20 02:29:36 -06:00
Code Issues Packages Projects Releases Wiki Activity

feat(wasm): add blocking fetch using jspi

Browse Source
This commit is contained in:
Nurul Huda (Apon) 2026-07-11 00:54:41 +06:00
parent 5736b807fc
commit 17e5f9748e
No known key found for this signature in database
GPG Key ID: 5D3F1DE2855A2F79
14 changed files with 295 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/ziex/src/cloudflare/do.ts
View File

@ -1,5 +1,6 @@
import { ZxBridge } from "../wasm";
import { createKVImports, createMemoryKV } from "../kv";
import { createFetchImports } from "../fetch";
import { createD1Imports } from "../db";
import { createWasiImports } from "../wasi";
import { buildWsImports, attachWebSocket } from "../runtime";
@ -128,6 +129,7 @@ export function createWebSocketDO(
__zx_ws: wsImports,
__zx_kv: createKVImports(kvBindings ?? { default: createMemoryKV() }, mem),
__zx_db: createD1Imports(dbBindings ?? {}, mem),
__zx_net: createFetchImports(mem),
...(options?.imports ? options.imports(mem) : {}),
...bridgeImports,
} as WebAssembly.Imports);

100
pkg/ziex/src/fetch.ts Normal file
View File

@ -0,0 +1,100 @@
/**
* Create a `fetch` import object for blocking WASM fetch via JSPI.
* When JSPI is unavailable the import always returns -1 (network error).
*/
export function createFetchImports(
getMemory: () => WebAssembly.Memory,
): Record<string, unknown> {
const encoder = new TextEncoder();
const decoder = new TextDecoder();
function readStr(ptr: number, len: number): string {
return decoder.decode(new Uint8Array(getMemory().buffer, ptr, len));
}
function writeBytes(buf_ptr: number, buf_max: number, data: Uint8Array): number {
if (data.length > buf_max) return -2;
new Uint8Array(getMemory().buffer, buf_ptr, data.length).set(data);
return data.length;
}
function readBytes(ptr: number, len: number): Uint8Array {
return new Uint8Array(getMemory().buffer, ptr, len);
}
function parseHeaders(headersJson: string): Record<string, string> {
let headers: Record<string, string> = {};
try {
headers = JSON.parse(headersJson);
} catch {
for (const line of headersJson.split('\n')) {
const colonIdx = line.indexOf(':');
if (colonIdx > 0) {
headers[line.slice(0, colonIdx)] = line.slice(colonIdx + 1);
}
}
}
return headers;
}
async function doFetch(
url_ptr: number,
url_len: number,
method_ptr: number,
method_len: number,
headers_ptr: number,
headers_len: number,
body_ptr: number,
body_len: number,
timeout_ms: number,
status_out: number,
buf_ptr: number,
buf_max: number,
): Promise<number> {
const url = readStr(url_ptr, url_len);
const method = method_len > 0 ? readStr(method_ptr, method_len) : 'GET';
const headersJson = headers_len > 0 ? readStr(headers_ptr, headers_len) : '{}';
const body = body_len > 0 ? readBytes(body_ptr, body_len) : undefined;
const headers = parseHeaders(headersJson);
const controller = new AbortController();
const timeout = timeout_ms > 0 ? setTimeout(() => controller.abort(), timeout_ms) : null;
try {
const response = await fetch(url, {
method,
headers: Object.keys(headers).length > 0 ? headers : undefined,
body: method !== 'GET' && method !== 'HEAD' ? body : undefined,
signal: controller.signal,
});
if (timeout) clearTimeout(timeout);
new DataView(getMemory().buffer).setUint16(status_out, response.status, true);
const bytes = new Uint8Array(await response.arrayBuffer());
return writeBytes(buf_ptr, buf_max, bytes);
} catch (err) {
if (timeout) clearTimeout(timeout);
const msg = err instanceof Error ? err.message : String(err);
console.error(`wasm fetch failed: ${url} (${msg})`);
return -1;
}
}
const Suspending = (WebAssembly as any).Suspending;
if (typeof Suspending !== 'function') {
return {
fetch: (
_url_ptr: number, _url_len: number,
_method_ptr: number, _method_len: number,
_headers_ptr: number, _headers_len: number,
_body_ptr: number, _body_len: number,
_timeout_ms: number,
_status_out: number,
_buf_ptr: number, _buf_max: number,
): number => -1,
};
}
return {
fetch: new Suspending(doFetch),
};
}

2
pkg/ziex/src/runtime.ts
View File

@ -1,5 +1,6 @@
import { ZxWasiBridge } from "./wasm/wasi";
import { createKVImports, createMemoryKV } from "./kv";
import { createFetchImports } from "./fetch";
import { createD1Imports } from "./db";
import { createWasiImports, ProcExit, mergeUint8Arrays } from "./wasi";
import type { WASI } from "./wasi";
@ -280,6 +281,7 @@ export async function run({
__zx_ws: buildWsImports(jspi ? Suspending : null, mem, new TextDecoder(), wsState),
__zx_kv: createKVImports(kvBindings ?? { default: createMemoryKV() }, mem),
__zx_db: createD1Imports(dbBindings ?? {}, mem),
__zx_net: createFetchImports(mem),
...(imports ? imports(mem) : {}),
...ZxWasiBridge.createImportObject(bridgeRef),
} as WebAssembly.Imports);

1
pkg/ziex/src/wasi.ts
View File

@ -33,6 +33,7 @@ export function createWasiImports({
"--pathname", url.pathname,
"--method", request.method,
"--search", url.search,
"--url", url.href
];
request.headers.forEach((value, name) => {

22
pkg/ziex/src/wasm/core.ts
View File

@ -1,4 +1,5 @@
import { ZigJS } from "../../../../vendor/jsz/js/src";
import { createFetchImports } from "../fetch";
/**
* Core WASM bridge - environment-agnostic (browser + edge).
@ -156,7 +157,7 @@ export class ZxBridgeCore {
const url = readString(urlPtr, urlLen);
const method = methodLen > 0 ? readString(methodPtr, methodLen) : 'GET';
const headersJson = headersLen > 0 ? readString(headersPtr, headersLen) : '{}';
const body = bodyLen > 0 ? readString(bodyPtr, bodyLen) : undefined;
const body = bodyLen > 0 ? getMemoryView().subarray(bodyPtr, bodyPtr + bodyLen) : undefined;
let headers: Record<string, string> = {};
try {
@ -183,8 +184,8 @@ export class ZxBridgeCore {
fetch(url, fetchOptions)
.then(async (response) => {
if (timeout) clearTimeout(timeout);
const text = await response.text();
this._notifyFetchComplete(fetchId, response.status, text, false);
const bytes = new Uint8Array(await response.arrayBuffer());
this._notifyFetchComplete(fetchId, response.status, bytes, false);
})
.catch((error) => {
if (timeout) clearTimeout(timeout);
@ -194,10 +195,15 @@ export class ZxBridgeCore {
});
}
/** Notify WASM that a fetch completed */
protected _notifyFetchComplete(fetchId: bigint, statusCode: number, body: string, isError: boolean): void {
/** Notify WASM that a fetch completed. `body` is raw bytes on success, text on error. */
protected _notifyFetchComplete(
fetchId: bigint,
statusCode: number,
body: string | Uint8Array,
isError: boolean,
): void {
const handler = this.#fetchCompleteHandler;
const encoded = textEncoder.encode(body);
const encoded = typeof body === 'string' ? textEncoder.encode(body) : body;
const ptr = this._alloc(encoded.length);
writeBytes(ptr, encoded);
invokeWasmExport(handler, fetchId, statusCode, ptr, encoded.length, isError ? 1 : 0);
@ -265,6 +271,10 @@ export class ZxBridgeCore {
static createImportObject(bridgeRef: { current: ZxBridgeCore | null }): WebAssembly.Imports {
return {
...jsz.importObject(),
__zx_net: createFetchImports(() => {
if (!jsz.memory) throw new Error("WASM memory is not ready");
return jsz.memory;
}) as WebAssembly.ModuleImports,
__zx: {
_log: (level: number, ptr: number, len: number) => ZxBridgeCore.log(level, ptr, len),
_fetchAsync: (

5
pkg/ziex/src/wasm/index.ts
View File

@ -24,6 +24,7 @@ import {
getMemoryView,
} from "./core";
import { createKVImports, type KVNamespace } from "../kv";
import { createFetchImports } from "../fetch";
import { createBrowserKVBindings } from "../browser/kv";
import type {
WsOnOpenHandler,
@ -216,6 +217,10 @@ export class ZxBridge extends ZxBridgeCore {
static override createImportObject(bridgeRef: { current: ZxBridge | null }): WebAssembly.Imports {
return {
...jsz.importObject(),
__zx_net: createFetchImports(() => {
if (!jsz.memory) throw new Error("WASM memory is not ready");
return jsz.memory;
}) as WebAssembly.ModuleImports,
__zx: {
_log: (level: number, ptr: number, len: number) => ZxBridgeCore.log(level, ptr, len),
_setEventHandlerMode: (vnodeId: bigint, eventTypeId: number, maySuspend: number) => {

9
pkg/ziex/src/wasm/wasi.ts
View File

@ -68,7 +68,7 @@ export class ZxWasiBridge {
const url = this.#readString(urlPtr, urlLen);
const method = methodLen > 0 ? this.#readString(methodPtr, methodLen) : 'GET';
const headersJson = headersLen > 0 ? this.#readString(headersPtr, headersLen) : '{}';
const body = bodyLen > 0 ? this.#readString(bodyPtr, bodyLen) : undefined;
const body = bodyLen > 0 ? this.#view().subarray(bodyPtr, bodyPtr + bodyLen) : undefined;
let headers: Record<string, string> = {};
try {
@ -91,7 +91,8 @@ export class ZxWasiBridge {
})
.then(async (res) => {
if (timeout) clearTimeout(timeout);
this.#notifyFetchComplete(fetchId, res.status, await res.text(), false);
const bytes = new Uint8Array(await res.arrayBuffer());
this.#notifyFetchComplete(fetchId, res.status, bytes, false);
})
.catch((err: Error) => {
if (timeout) clearTimeout(timeout);
@ -100,8 +101,8 @@ export class ZxWasiBridge {
});
}
#notifyFetchComplete(fetchId: bigint, status: number, body: string, isError: boolean): void {
const encoded = encoder.encode(body);
#notifyFetchComplete(fetchId: bigint, status: number, body: string | Uint8Array, isError: boolean): void {
const encoded = typeof body === 'string' ? encoder.encode(body) : body;
const ptr = this.#alloc(encoded.length);
this.#writeBytes(ptr, encoded);
this.#fetchCompleteHandler(fetchId, status, ptr, encoded.length, isError ? 1 : 0);

5
src/runtime/core/App/Wasm.zig
View File

@ -38,6 +38,7 @@ pub fn run(process_init: std.process.Init) !void {
var pathname: []const u8 = "/";
var search: []const u8 = "";
var url: []const u8 = "";
var method: zx.server.Request.Method = .GET;
var header_entries = std.ArrayList(HeaderEntry).empty;
defer header_entries.deinit(allocator);
@ -59,6 +60,8 @@ pub fn run(process_init: std.process.Init) !void {
.value = std.mem.trimStart(u8, header_str[sep + 1 ..], " "),
});
}
} else if (std.mem.eql(u8, arg, "--url")) {
url = args.next() orelse return error.MissingUrl;
}
}
@ -93,7 +96,7 @@ pub fn run(process_init: std.process.Init) !void {
backend.cookie_header = cookie_header;
backend.route_match = Router.matchRoute(pathname, .{ .match = .exact });
const request = backend.request(method, pathname);
const request = backend.request(method, pathname, url);
const response = backend.response();
const http = backend.http();

15
src/runtime/core/Fetch.zig
View File

@ -20,6 +20,7 @@ const common = @import("common.zig");
const server_impl = if (!is_wasm) @import("../server/fetch.zig") else struct {};
const client_impl = if (is_wasm) @import("../client/fetch.zig") else struct {};
const wasm_impl = if (is_wasm) @import("Fetch/Wasm.zig") else struct {};
pub const Method = common.Method;
pub const ContentType = common.ContentType;
@ -28,7 +29,7 @@ pub const is_wasm = builtin.cpu.arch == .wasm32 or builtin.cpu.arch == .wasm64;
/// Io determines how fetch operations are executed.
///
/// - `blocking`: Blocks until complete (server-side)
/// - `blocking`: Blocks until complete (server-side native, WASM via JSPI)
/// - `callback`: Calls a callback when complete (client-side WASM)
pub const Io = struct {
mode: Mode,
@ -57,7 +58,7 @@ pub const Io = struct {
/// Check if this Io mode is supported on the current platform.
pub fn isSupported(self: Io) bool {
return switch (self.mode) {
.blocking => !is_wasm, // Blocking only works on server
.blocking => true, // Native on server, JSPI-suspended on WASM
.callback => true, // Callback works everywhere
};
}
@ -121,6 +122,11 @@ pub const Response = struct {
return self._body;
}
/// Returns the raw response body bytes.
pub fn bytes(self: *Response) ![]const u8 {
return self.text();
}
/// Get an `std.Io.Reader` for streaming the response body.
pub fn reader(self: *Response) std.Io.Reader {
return std.Io.Reader.fixed(self._body);
@ -199,7 +205,7 @@ pub const ResponseCallbackCtx = *const fn (ctx: *anyopaque, response: ?*Response
/// Perform an HTTP fetch request.
///
/// The `io` parameter determines the execution model:
/// - `Io.blocking` - Blocks until complete, returns Response (server-only)
/// - `Io.blocking` - Blocks until complete, returns Response (server native, WASM via JSPI)
/// - `Io.wasm(&callback)` - Calls callback when complete (WASM)
///
/// **Server-side (blocking):**
@ -237,9 +243,8 @@ pub fn fetch(
switch (io.mode) {
.blocking => {
// Server-side: blocking fetch
if (is_wasm) {
return error.UnsupportedIoMode;
return try wasm_impl.fetch(allocator, url, init);
}
const response = try server_impl.fetch(allocator, url, init);
return response;

129
src/runtime/core/Fetch/Wasm.zig Normal file
View File

@ -0,0 +1,129 @@
const std = @import("std");
const Fetch = @import("../Fetch.zig");
const Response = Fetch.Response;
const Headers = Fetch.Headers;
const RequestInit = Fetch.RequestInit;
const FetchError = Fetch.FetchError;
const max_response_bytes = 16 * 1024 * 1024;
/// Perform a blocking HTTP fetch via the JS `fetch` import (JSPI).
pub fn fetch(allocator: std.mem.Allocator, url: []const u8, init: RequestInit) FetchError!Response {
const method_str = @tagName(init.method);
var headers_buf: [8192]u8 = undefined;
const headers_json = serializeHeadersJson(init.headers, &headers_buf);
const body = init.body orelse "";
var status: u16 = 0;
var capacity: usize = 65536;
var buf = try allocator.alloc(u8, capacity);
defer allocator.free(buf);
while (true) {
const n = ext.fetch(
url.ptr,
url.len,
method_str.ptr,
method_str.len,
headers_json.ptr,
headers_json.len,
body.ptr,
body.len,
init.timeout_ms,
&status,
buf.ptr,
capacity,
);
if (n == -2) {
capacity *= 2;
if (capacity > max_response_bytes) return error.InvalidResponse;
buf = try allocator.realloc(buf, capacity);
continue;
}
if (n < 0) return error.NetworkError;
const body_data = try allocator.dupe(u8, buf[0..@intCast(n)]);
return Response{
.status = status,
.status_text = statusText(status),
.headers = Headers.init(allocator),
._body = body_data,
._body_used = false,
._allocator = allocator,
._owns_memory = true,
};
}
}
const ext = struct {
pub extern "__zx_net" fn fetch(
url_ptr: [*]const u8,
url_len: usize,
method_ptr: [*]const u8,
method_len: usize,
headers_ptr: [*]const u8,
headers_len: usize,
body_ptr: [*]const u8,
body_len: usize,
timeout_ms: u32,
status_out: *u16,
buf_ptr: [*]u8,
buf_max: usize,
) i32;
};
fn serializeHeadersJson(headers: ?[]const RequestInit.Header, buf: []u8) []const u8 {
var len: usize = 0;
buf[len] = '{';
len += 1;
if (headers) |hdrs| {
for (hdrs, 0..) |h, i| {
if (i > 0) {
buf[len] = ',';
len += 1;
}
buf[len] = '"';
len += 1;
const name_end = @min(len + h.name.len, buf.len - 10);
@memcpy(buf[len..name_end], h.name[0..@min(h.name.len, name_end - len)]);
len = name_end;
buf[len] = '"';
len += 1;
buf[len] = ':';
len += 1;
buf[len] = '"';
len += 1;
const val_end = @min(len + h.value.len, buf.len - 2);
@memcpy(buf[len..val_end], h.value[0..@min(h.value.len, val_end - len)]);
len = val_end;
buf[len] = '"';
len += 1;
}
}
buf[len] = '}';
len += 1;
return buf[0..len];
}
fn statusText(code: u16) []const u8 {
return switch (code) {
200 => "OK",
201 => "Created",
204 => "No Content",
301 => "Moved Permanently",
302 => "Found",
304 => "Not Modified",
400 => "Bad Request",
401 => "Unauthorized",
403 => "Forbidden",
404 => "Not Found",
500 => "Internal Server Error",
else => "Unknown",
};
}

4
src/runtime/core/Http/Wasm.zig
View File

@ -50,9 +50,9 @@ pub const Backend = struct {
}
/// Build an abstract `Request` backed by this instance.
pub fn request(self: *Backend, method: Request.Method, pathname: []const u8) Request {
pub fn request(self: *Backend, method: Request.Method, pathname: []const u8, url: []const u8) Request {
return (Request.Builder{
.url = "",
.url = url,
.method = method,
.method_str = @tagName(method),
.pathname = pathname,

21
src/runtime/core/Kv/Wasm.zig
View File

@ -3,11 +3,24 @@ const Wasm = @This();
const std = @import("std");
const Kv = @import("../Kv.zig");
const max_response_bytes = 16 * 1024 * 1024;
fn get(_: ?*anyopaque, ns: []const u8, allocator: std.mem.Allocator, key: []const u8) !?[]u8 {
var buf: [8192]u8 = undefined;
const n = ext.kv_get(ns.ptr, ns.len, key.ptr, key.len, &buf, buf.len);
if (n < 0) return null;
return try allocator.dupe(u8, buf[0..@intCast(n)]);
var capacity: usize = 65536;
var buf = try allocator.alloc(u8, capacity);
defer allocator.free(buf);
while (true) {
const n = ext.kv_get(ns.ptr, ns.len, key.ptr, key.len, buf.ptr, capacity);
if (n == -2) {
capacity *= 2;
if (capacity > max_response_bytes) return error.InvalidResponse;
buf = try allocator.realloc(buf, capacity);
continue;
}
if (n < 0) return null;
return try allocator.dupe(u8, buf[0..@intCast(n)]);
}
}
fn put(_: ?*anyopaque, ns: []const u8, key: []const u8, value: []const u8, _: Kv.PutOptions) !void {

2
src/runtime/core/Request.zig
View File

@ -1,4 +1,4 @@
pub const Request = @This();
const Request = @This();
const std = @import("std");
const common = @import("common.zig");

2
src/runtime/core/Response.zig
View File

@ -1,4 +1,4 @@
pub const Response = @This();
const Response = @This();
const std = @import("std");
const common = @import("common.zig");