From 2b412f098e1d6e82e5646265f73779acb1ba6a68 Mon Sep 17 00:00:00 2001
From: rawan10101 <rawan_khalid@aucegypt.edu>
Date: Tue, 21 Jul 2026 01:57:57 +0300
Subject: [PATCH] feat: initial WASM RPC implementation (RPC handshake
 incomplete)

---
 src/wasm/msgpack.js     |  25 ++++
 src/wasm/nvim-worker.js | 303 ++++++++++++++++++++++++++++++++++++++++
 src/wasm/protocol.js    |  34 +++++
 src/wasm/rpc.js         | 190 +++++++++++++++++++++++++
 4 files changed, 552 insertions(+)

diff --git a/src/wasm/msgpack.js b/src/wasm/msgpack.js
index e69de29bb2..52291cb566 100644
--- a/src/wasm/msgpack.js
+++ b/src/wasm/msgpack.js
@@ -0,0 +1,25 @@
+// Check that the library is actually loaded.
+if (typeof MessagePack === 'undefined') {
+  throw new Error(
+    'MessagePack library not loaded. ' +
+    'Check that the CDN script tag for @msgpack/msgpack is present and loads correctly.'
+  );
+}
+
+const MsgpackCodec = {
+  encode(value) {
+    const encoded = MessagePack.encode(value);
+
+    // Debug: log the encoded bytes (first few bytes) to verify.
+    const prefix = Array.from(encoded.slice(0, Math.min(encoded.length, 8)))
+      .map(b => b.toString(16).padStart(2, '0'))
+      .join(' ');
+    console.log(`[MsgpackCodec.encode] ${encoded.length} bytes, prefix: ${prefix}`);
+
+    return encoded;
+  },
+
+  decodeMulti(bytes) {
+    return [...MessagePack.decodeMulti(bytes)];  // ✅ correct global
+  },
+};
diff --git a/src/wasm/nvim-worker.js b/src/wasm/nvim-worker.js
index e69de29bb2..07b9cbbcf9 100644
--- a/src/wasm/nvim-worker.js
+++ b/src/wasm/nvim-worker.js
@@ -0,0 +1,303 @@
+const CAP = 1 << 16; // must match the main thread's ring buffer size
+let state, ringData; // state is now Int32Array(3): [head, tail, closed]
+
+let stdinPollCount = 0;
+let totalBytesRead = 0;
+const consumedBytes = [];
+
+let pendingPollCallbacks = [];
+function wakePendingPolls() {
+    const cbs = pendingPollCallbacks;
+    pendingPollCallbacks = [];
+    for (const cb of cbs) cb(65);
+}
+let lastCheckedTail = -1;
+
+function checkForNewDataAndWake() {
+    if (!state) return; // not initialized yet
+    const head = Atomics.load(state, 0);
+    const tail = Atomics.load(state, 1);
+    if (head !== tail && pendingPollCallbacks.length > 0) {
+        wakePendingPolls();
+    }
+}
+
+setInterval(checkForNewDataAndWake, 20); // check every 20ms
+
+function popNonBlocking() {
+    const head = Atomics.load(state, 0);
+    const tail = Atomics.load(state, 1);
+    if (head !== tail) {
+        const b = ringData[tail];
+        Atomics.store(state, 1, (tail + 1) % CAP);
+        totalBytesRead++;
+        consumedBytes.push(b);
+        return b;
+    }
+    return -1;
+}
+
+// blocks the Worker thread until a byte is available or shutdown fires.
+function popBlocking() {
+    while (true) { //producer-consumer
+        const head = Atomics.load(state, 0);
+        const tail = Atomics.load(state, 1);
+
+        if (head !== tail) {
+            const b = ringData[tail];
+            Atomics.store(state, 1, (tail + 1) % CAP);
+            totalBytesRead++;
+            consumedBytes.push(b);
+            return b;
+        }
+
+        if (Atomics.load(state, 2) === 1) {
+            return -1; 
+        }
+
+        // Blocks here until Atomics.notify(state, 0) fires, or 1s elapses
+        Atomics.wait(state, 0, head, 1000);
+    }
+}
+
+let stdoutBuf = [];
+let stderrBuf = [];
+
+function flushStdout() {
+  if (stdoutBuf.length) {
+    postMessage({ type: 'stdout', bytes: stdoutBuf });
+    stdoutBuf = [];
+  }
+}
+
+function flushStderr() {
+  if (stderrBuf.length) {
+    postMessage({ type: 'stderr', bytes: stderrBuf });
+    stderrBuf = [];
+  }
+}
+
+function flushAll() {
+  flushStdout();
+  flushStderr();
+}
+
+function writeStdout(c) {
+  stdoutBuf.push(c);
+  flushStdout();
+}
+
+function writeStderr(c) {
+  stderrBuf.push(c);
+  if (stderrBuf.length > 80) {
+    flushStderr();
+  }
+}
+
+function makeArgv(M, args) {
+  const ptrs = args.map(s => {
+    const len = M.lengthBytesUTF8(s) + 1, p = M._malloc(len);
+    M.stringToUTF8(s, p, len);
+    return p;
+  });
+  const argv = M._malloc((ptrs.length + 1) * 4);
+  ptrs.forEach((p, i) => M.setValue(argv + i * 4, p, '*'));
+  M.setValue(argv + ptrs.length * 4, 0, '*');
+  return { argc: ptrs.length, argv };
+}
+
+let moduleRef = null;
+
+self.onerror = (e) => {
+  postMessage({ type: 'status', text: 'WORKER ERROR: ' + e.message });
+};
+
+self.onunhandledrejection = (e) => {
+  postMessage({ type: 'status', text: 'WORKER REJECTION: ' + e });
+};
+
+function makeBridgeReadOps(origOps, label) {
+  const newOps = Object.assign({}, origOps);
+  newOps.read = function (stream, buffer, offset, length, position) {
+      let n = 0;
+      while (n < length) {
+          const b = popNonBlocking();
+          if (b === -1) break;
+          buffer[offset + n] = b;
+          n++;
+      }
+      if (n === 0) {
+          console.log(`[stream_ops:${label}] no data available, returning 0 (temporary — not correct long-term)`);
+          return 0; // TEMP: fake EOF
+      }
+      console.log(`[stream_ops:${label}] read ${n} bytes`);
+      return n;
+  };
+newOps.poll = function (stream, timeout, notifyCallback) {
+    const head = Atomics.load(state, 0);
+    const tail = Atomics.load(state, 1);
+
+    if (head !== tail) {
+        return 65;
+    }
+
+    if (notifyCallback) {
+        pendingPollCallbacks.push(notifyCallback);
+    }
+
+    return 0;
+};
+  const origGetattr = origOps.getattr;
+  newOps.getattr = function (stream) {
+    const attr = origGetattr ? origGetattr.call(this, stream) : { mode: 0 };
+    attr.mode = (attr.mode & ~0xF000) | 0x1000;
+    return attr;
+  };
+  return newOps;
+}
+
+self.onmessage = async (ev) => {
+  const msg = ev.data;
+
+  if (msg.type === 'init') {
+    state = new Int32Array(msg.sab, 0, 3);   // [head, tail, closed]
+    ringData = new Uint8Array(msg.sab, 12, CAP); // offset moved from 8 → 12
+setInterval(checkForNewDataAndWake, 20); 
+    postMessage({ type: 'status', text: 'loading wasm...' });
+    importScripts('../../zig-out/bin/nvim.js');
+
+    const m = await createNvim({
+      locateFile: (p) => p.endsWith('.data') ? '../../zig-out/bin/nvim.data' : '../../zig-out/bin/' + p,
+      noInitialRun: true,
+      stdin: () => {
+      console.log('[stdin callback] fired');
+      const b = popNonBlocking();
+      return b === -1 ? null : b;
+      },
+      stdout: c => writeStdout(c),
+      stderr: c => writeStderr(c),
+      print: t => postMessage({ type: 'status', text: '[print] ' + t }),
+      printErr: t => postMessage({ type: 'status', text: '[printErr] ' + t }),
+      preRun: [m => {
+        m.ENV.TERM = "xterm-256color";
+        m.ENV.HOME = "/home/user";
+        m.ENV.VIMRUNTIME = "/runtime";
+        m.ENV.COLUMNS = String(msg.cols);
+        m.ENV.LINES = String(msg.rows);
+        try { m.FS.mkdir('/tmp'); } catch (e) {}
+        m.FS.mkdir('/home/user');
+        m.FS.mkdir('/home/user/.config');
+        m.FS.mkdir('/home/user/.local');
+        m.FS.mkdir('/home/user/.local/share');
+        m.FS.mount(m.IDBFS, {}, '/home/user/.config');
+        m.FS.mount(m.IDBFS, {}, '/home/user/.local/share');
+      }],
+    });
+
+    await new Promise((res, rej) => m.FS.syncfs(true, e => e ? rej(e) : res()));
+    try { m.FS.mkdir('/home/user/.config/nvim'); } catch (e) {}
+    try { m.FS.mkdir('/home/user/.local/share/nvim'); } catch (e) {}
+    try { m.FS.mkdir('/runtime/parser'); } catch (e) {}
+    ['lua', 'c', 'vim', 'vimdoc', 'query', 'markdown', 'markdown_inline'].forEach(p => {
+      try { m.FS.writeFile(`/runtime/parser/${p}.so`, ''); } catch (e) {}
+    });
+
+    moduleRef = m;
+
+    function sanitizeAttr(attr) {
+      attr.dev = attr.dev ?? 1;
+      attr.ino = attr.ino ?? 1;
+      attr.mode = attr.mode ?? 0o666;
+      attr.nlink = attr.nlink ?? 1;
+      attr.uid = attr.uid ?? 0;
+      attr.gid = attr.gid ?? 0;
+      attr.rdev = attr.rdev ?? 0;
+      attr.size = (typeof attr.size === 'number' && !isNaN(attr.size)) ? attr.size : 0;
+      attr.blksize = attr.blksize ?? 4096;
+      attr.blocks = attr.blocks ?? 0;
+      const validDate = d => d instanceof Date && !isNaN(d.getTime());
+      attr.atime = validDate(attr.atime) ? attr.atime : new Date(0);
+      attr.mtime = validDate(attr.mtime) ? attr.mtime : new Date(0);
+      attr.ctime = validDate(attr.ctime) ? attr.ctime : new Date(0);
+      return attr;
+    }
+
+    function isStdinLike(path) {
+      const p = path || '';
+      return p.startsWith('pipe[') || p.includes('my_stdin') || p === '/dev/stdin';
+    }
+
+    ['fstat', 'stat', 'lstat'].forEach(name => {
+      if (typeof m.FS[name] !== 'function') return;
+      const orig = m.FS[name].bind(m.FS);
+      m.FS[name] = function (...args) {
+        const attr = orig(...args);
+        const path = (name === 'fstat')
+          ? (m.FS.streams[args[0]] && m.FS.streams[args[0]].path)
+          : args[0];
+        if (isStdinLike(path)) {
+          attr.mode = 0o010666;
+        }
+        return sanitizeAttr(attr);
+      };
+    });
+
+    const origCreateStream = m.FS.createStream;
+    m.FS.createStream = function (stream, fd) {
+      const s = origCreateStream.call(this, stream, fd);
+      const p = s.path || '';
+      const looksLikeStdin = p.startsWith('pipe[') || p.includes('my_stdin') || p === '/dev/stdin';
+      if (looksLikeStdin && !s._patched) {
+        s.stream_ops = makeBridgeReadOps(s.stream_ops, 'createStream:' + p);
+        s._patched = true;
+      }
+      return s;
+    };
+
+    const origFSRead = m.FS.read;
+    m.FS.read = function (stream, buffer, offset, length, position) {
+      if (stream.path && stream.path.startsWith('pipe[') && !stream._patched) {
+        stream.stream_ops = makeBridgeReadOps(stream.stream_ops, 'fallback:' + stream.path);
+        stream._patched = true;
+      }
+      return origFSRead.call(this, stream, buffer, offset, length, position);
+    };
+
+    const origFSWrite = m.FS.write;
+    m.FS.write = function (stream, buffer, offset, length, position, canOwn) {
+      return origFSWrite.call(this, stream, buffer, offset, length, position, canOwn);
+    };
+
+    const { argc, argv } = makeArgv(m, ["nvim", "--embed"]);
+    let ret;
+    try {
+      postMessage({ type: 'status', text: 'Starting Neovim...' });
+      ret = await m._nvim_main(argc, argv);
+      const headAtExit = Atomics.load(state, 0);
+      const tailAtExit = Atomics.load(state, 1);
+      const unread = (headAtExit - tailAtExit + CAP) % CAP;
+      postMessage({ type: 'status', text: `_nvim_main RETURNED ret=${ret}, unread bytes still in buffer=${unread} (head=${headAtExit}, tail=${tailAtExit})` });
+
+    } catch (e) {
+      flushAll();
+      postMessage({ type: 'status', text: 'EXCEPTION: ' + e.message });
+      throw e;
+    }
+
+    flushAll();
+
+    const hexStr = consumedBytes.slice(0, 50).map(x => x.toString(16).padStart(2, '0')).join(' ');
+    const remaining = consumedBytes.length > 50 ? `... (${consumedBytes.length - 50} more bytes)` : '';
+    postMessage({ type: 'status', text: `EXIT CODE ${ret}. Read ${totalBytesRead} bytes: ${hexStr}${remaining}` });
+  }
+  if (msg.type === 'persist') {
+    if (!moduleRef) { postMessage({ type: 'persisted', error: 'module not ready' }); return; }
+    moduleRef.FS.syncfs(false, e => postMessage({ type: 'persisted', error: e ? String(e) : null }));
+  }
+    if (msg.type === 'shutdown') {
+    if (state) {
+        Atomics.store(state, 2, 1);
+        Atomics.notify(state, 0); // wake up any pending popBlocking()
+    }
+  }
+};
diff --git a/src/wasm/protocol.js b/src/wasm/protocol.js
index e69de29bb2..b8f1fede9f 100644
--- a/src/wasm/protocol.js
+++ b/src/wasm/protocol.js
@@ -0,0 +1,34 @@
+// msgpack-rpc message shapes:
+//   request:      [0, msgid, method, params]
+//   response:     [1, msgid, error, result]
+//   notification: [2, method, params]
+
+const Protocol = {
+  encodeRequest(msgid, method, params) {
+    return MsgpackCodec.encode([0, msgid, method, params]);
+  },
+
+  encodeNotification(method, params) {
+    return MsgpackCodec.encode([2, method, params]);
+  },
+
+  // classifies a decoded msgpack-rpc array into a tagged object.
+  // throws if `msg` doesn't look like a valid rpc message.
+  parseMessage(msg) {
+    if (!Array.isArray(msg)) throw new Error('rpc message is not an array');
+    const [type, ...rest] = msg;
+    if (type === 0) {
+      const [msgid, method, params] = rest;
+      return { kind: 'request', msgid, method, params };
+    }
+    if (type === 1) {
+      const [msgid, error, result] = rest;
+      return { kind: 'response', msgid, error, result };
+    }
+    if (type === 2) {
+      const [method, params] = rest;
+      return { kind: 'notification', method, params };
+    }
+    throw new Error('unknown rpc message type: ' + type);
+  },
+};
diff --git a/src/wasm/rpc.js b/src/wasm/rpc.js
index e69de29bb2..2aa220f434 100644
--- a/src/wasm/rpc.js
+++ b/src/wasm/rpc.js
@@ -0,0 +1,190 @@
+// Transport: owns the worker + SharedArrayBuffer ring buffer
+class WorkerTransport {
+  constructor(workerPath, { cols = 80, rows = 24, cap = 1 << 16 } = {}) {
+    this.CAP = cap;
+    this.sab = new SharedArrayBuffer(12 + this.CAP); 
+    this.state = new Int32Array(this.sab, 0, 3);     // [head, tail, closed]
+    this.ringData = new Uint8Array(this.sab, 12, this.CAP); // offset 12
+
+    this._bytesHandlers = [];
+    this._statusHandlers = [];
+
+    this.worker = new Worker(workerPath);
+    this.worker.onmessage = (ev) => {
+      console.log('[main] Worker message:', ev.data.type, ev.data);
+      this._onWorkerMessage(ev.data);
+    };
+    this.worker.onerror = (e) => this._emitStatus('worker error: ' + e.message);
+    this.worker.postMessage({ type: 'init', sab: this.sab, cols, rows });
+  }
+  shutdown() {
+    Atomics.store(this.state, 2, 1);
+    Atomics.notify(this.state, 0);
+    this.worker.postMessage({ type: 'shutdown' });
+    }
+
+  _onWorkerMessage(msg) {
+    console.log('[main] _onWorkermessage:', msg.type);
+    if (msg.type === 'stdout') {
+      console.log(`[main] stdout bytes:`, msg.bytes);
+      this._bytesHandlers.forEach(h => h(msg.bytes));
+    } else if (msg.type === 'stderr') {
+      const text = new TextDecoder().decode(new Uint8Array(msg.bytes));
+      this._emitStatus('[stderr] ' + text);
+    } else if (msg.type === 'status') {
+      this._emitStatus(msg.text);
+    }
+  }
+
+  _emitStatus(text) {
+    console.log('[main] status:', text);
+    this._statusHandlers.forEach(h => h(text));
+  }
+
+  send(bytes) {
+    console.log("[main] SEND: writing", bytes.length, "bytes");
+    
+// This ensures all data is in the buffer before Nvim wakes up
+const hexStr = Array.from(bytes)
+        .map(b => b.toString(16).padStart(2, '0'))
+        .join(' ');
+    console.log(`[main] Send: ${bytes.length} bytes: ${hexStr}`);
+    
+    for (let i = 0; i < bytes.length; i++) {
+      const b = bytes[i];
+      const head = Atomics.load(this.state, 0);
+      const tail = Atomics.load(this.state, 1);
+      
+      const next = (head + 1) % this.CAP;
+      if (next === tail) {
+        console.error('[main] Send: ring buffer full');
+        return false;
+      }
+      
+      this.ringData[head] = b;
+      Atomics.store(this.state, 0, next);
+      // wait until all bytes are written
+    }
+    
+    // notify once after all bytes are in the buffer
+    Atomics.notify(this.state, 0);
+    
+    console.log("[main] SEND DONE, notified worker");
+    return true;
+  }
+
+  onBytes(cb) { this._bytesHandlers.push(cb); }
+  onStatus(cb) { this._statusHandlers.push(cb); }
+  persist() { this.worker.postMessage({ type: 'persist' }); }
+}
+
+class RpcClient {
+  constructor(transport) {
+    this.transport = transport;
+    this.nextMsgId = 1;
+    this.pending = new Map();
+    this.notificationHandlers = new Map();
+
+    this._buffer = new Uint8Array(0);
+    this._decodedMessages = [];
+    
+    transport.onBytes(bytes => this._handleBytes(bytes));
+  }
+
+  _handleBytes(newBytes) {
+    const combined = new Uint8Array(this._buffer.length + newBytes.length);
+    combined.set(this._buffer, 0);
+    combined.set(newBytes, this._buffer.length);
+    this._buffer = combined;
+
+    const decoder = new MessagePack.Decoder();
+    let offset = 0;
+
+    while (offset < this._buffer.length) {
+      let msg;
+      try {
+        // decodes exactly one value starting at `offset` tells us how much it consumed
+        decoder.setBuffer(this._buffer.subarray(offset));
+        msg = decoder.decode();
+      } catch (e) {
+        break; // incomplete trailing message — stop, wait for more bytes
+      }
+      offset += decoder.bytesConsumed ?? decoder.pos; // depends on library version
+      this._dispatch(msg);
+    }
+
+    // only drop what was actually consumed and keep any incomplete tail
+    this._buffer = this._buffer.slice(offset);
+  }
+
+  _dispatch(rawMsg) {
+    console.log('[RpcClient] dispatching:', rawMsg);
+    
+    let msg;
+    try {
+      msg = Protocol.parseMessage(rawMsg);
+    } catch (e) {
+      console.error('[RpcClient] parse failed', rawMsg, e);
+      return;
+    }
+
+    console.log('[RpcClient] parsed:', msg);
+
+    if (msg.kind === 'response') {
+      const p = this.pending.get(msg.msgid);
+      if (!p) {
+        console.warn('[RpcClient] unknown msgid', msg.msgid);
+        return;
+      }
+      this.pending.delete(msg.msgid);
+      clearTimeout(p.timeoutId);
+      console.log('[RpcClient] resolved msgid', msg.msgid);
+      if (msg.error) {
+        p.reject(new Error(msg.error));
+      } else {
+        p.resolve(msg.result);
+      }
+    } else if (msg.kind === 'notification') {
+      console.log('[RpcClient] notification:', msg.method);
+      const handlers = this.notificationHandlers.get(msg.method) || [];
+      handlers.forEach(h => h(msg.params));
+    } else if (msg.kind === 'request') {
+      console.warn('[RpcClient] unhandled request', msg);
+    }
+  }
+
+  request(method, params = []) {
+    const msgid = this.nextMsgId++;
+    const bytes = Protocol.encodeRequest(msgid, method, params);
+    console.log(`[RpcClient] sending request ${msgid} (${method}), ${bytes.length} bytes`);
+    
+    return new Promise((resolve, reject) => {
+      this.pending.set(msgid, { resolve, reject, method, timestamp: Date.now() });
+      
+      const timeoutId = setTimeout(() => {
+        if (this.pending.has(msgid)) {
+          this.pending.delete(msgid);
+          reject(new Error(`Request ${msgid} (${method}) timed out after 30s`));
+        }
+      }, 30000);
+      
+      this.pending.get(msgid).timeoutId = timeoutId;
+      
+      //  This batches all bytes before notifying
+      this.transport.send(bytes);
+    });
+  }
+
+  notify(method, params = []) {
+    const bytes = Protocol.encodeNotification(method, params);
+    console.log(`[RpcClient] sending notification ${method}, ${bytes.length} bytes`);
+    this.transport.send(bytes);
+  }
+
+  on(method, handler) {
+    if (!this.notificationHandlers.has(method)) {
+      this.notificationHandlers.set(method, []);
+    }
+    this.notificationHandlers.get(method).push(handler);
+  }
+}