path: root/packages/meshbay-hub/src/meshbay_hub/static/transport.js

/**
 * MeshBay Browser Transport — WebRTC DataChannel client.
 *
 * Connects to a MeshBay node via WebRTC DataChannel (P2P, E2E).
 * The hub is only used for signaling (SDP/ICE relay) — after connection,
 * all data flows directly between browser and node.
 *
 * Wire format: length-prefixed msgpack (4-byte big-endian + msgpack payload).
 * Same format as QUIC and TCP+TLS transports on the node side.
 *
 * Usage:
 *   const transport = new MeshBayTransport(hubUrl, accessToken);
 *   await transport.connect(nodeId, jwtToken, groupId);
 *   const index = await transport.fetchIndex();
 *   const chunk = await transport.fetchChunk(fileId, 0);
 *   transport.close();
 */

async function _pkFromSk(skPkcs8B64) {
  const raw = Uint8Array.from(atob(skPkcs8B64), c => c.charCodeAt(0));
  const sk = await crypto.subtle.importKey('pkcs8', raw, { name: 'X25519' }, true, ['deriveBits']);
  const jwk = await crypto.subtle.exportKey('jwk', sk);
  const b64url = jwk.x;
  const b64 = b64url.replace(/-/g, '+').replace(/_/g, '/');
  const pad = b64.length % 4;
  return pad ? b64 + '='.repeat(4 - pad) : b64;
}

async function _pkEdFromSk(skPkcs8B64) {
  const raw = Uint8Array.from(atob(skPkcs8B64), c => c.charCodeAt(0));
  const sk = await crypto.subtle.importKey('pkcs8', raw, { name: 'Ed25519' }, true, ['sign']);
  const jwk = await crypto.subtle.exportKey('jwk', sk);
  const b64 = jwk.x.replace(/-/g, '+').replace(/_/g, '/');
  const pad = b64.length % 4;
  return pad ? b64 + '='.repeat(4 - pad) : b64;
}

// 48 KB is what fits comfortably in one SCTP message across stacks; the window is
// what makes the rate independent of the round trip. 32 × 48 KB = 1.5 MB in
// flight, which saturates any path up to roughly 100 Mb/s at 100 ms.
const UPLOAD_CHUNK_SIZE = 48 * 1024;
const UPLOAD_WINDOW = 32;
const UPLOAD_BUFFER_HIGH = 1024 * 1024;

// Segments of 256 KB: 24 in flight is 6 MB, enough to keep playback fed over a
// slow link and small enough that nothing accumulates.
const STREAM_CREDITS = 24;

function _aborted() {
  const err = new Error('Cancelled');
  err.name = 'AbortError';
  return err;
}

const JOIN_REFUSALS = {
  code_required: 'This node does not know this browser yet. Ask the node operator '
    + 'for a pairing code (meshbay-node operator pair).',
  code_invalid: 'That pairing code is not valid — it may be mistyped, expired, '
    + 'already used, or issued for a different account.',
  key_changed: 'This account is already paired with a different key on this node. '
    + 'If you reset your keys, the operator must unpin you before pairing again.',
  not_authorized_for_group: 'The node does not list you as a member of this group. '
    + 'Being a member on the hub is not enough — ask the operator for an invite.',
  no_gek: 'This group has no key yet. The node operator must run '
    + '`meshbay-node gek-init` for it.',
  signature_invalid: 'The node rejected the signature over your keys.',
  stale_request: 'Your clock is too far from the node\'s — check the system time.',
  group_mismatch: 'The node refused a request naming a different group.',
};

class MeshBayTransport {
  constructor(hubUrl, accessToken) {
    this._hubUrl = hubUrl;
    this._accessToken = accessToken;
    this._pc = null;
    this._channel = null;
    this._pending = new Map();
    this._seqId = 0;
    this._recvBuf = new Uint8Array(0);
    this._connected = false;
    this._onChat = null;
    this._onStreamInit = null;
    this._onStreamData = null;
    this._onStreamEnd = null;
    this._onIndexSync = null;
    // filename → the uploader waiting on it. Keyed rather than FIFO because
    // several uploads may be in flight at once and their acks interleave; the
    // node names the file in every one.
    this._uploaders = new Map();
  }

  get connected() { return this._connected; }

  set onChat(fn) { this._onChat = fn; }
  set onStreamInit(fn) { this._onStreamInit = fn; }
  set onStreamData(fn) { this._onStreamData = fn; }
  set onStreamEnd(fn) { this._onStreamEnd = fn; }
  set onIndexSync(fn) { this._onIndexSync = fn; }

  get sessionKeys() { return this._sessionKeys; }

  /** Set on a first join: the identity created for this node, still to be left with it. */
  get newNodeBundle() { return this._newNodeBundle || null; }
  set newNodeBundle(v) { this._newNodeBundle = v; }

  async connect(nodeId, jwtToken, groupId, gekRaw, sessionKeys, bundleKey, username,
                userId, joinCode) {
    this._gekRaw = gekRaw || null;
    this._sessionKeys = sessionKeys || null;
    this._bundleKey = bundleKey || null;
    this._username = username || null;
    this._userId = userId || null;
    this._newNodeBundle = null;
    this._joinError = null;
    this._pc = new RTCPeerConnection({
      iceServers: [{ urls: 'stun:stun.l.google.com:19302' }],
    });

    this._channel = this._pc.createDataChannel('mnp', { ordered: true });
    this._channel.binaryType = 'arraybuffer';

    let channelReject = null;
    const channelReady = new Promise((resolve, reject) => {
      channelReject = reject;
      const timeout = setTimeout(() => reject(new Error('DataChannel open timeout')), 30000);
      this._channel.onopen = () => {
        clearTimeout(timeout);
        this._connected = true;
        resolve();
      };
    });

    this._channel.onmessage = (event) => this._onMessage(event.data);
    this._channel.onclose = (ev) => {
      console.warn('[MeshBay] DataChannel closed', this._channel?.readyState, ev);
      this._connected = false;
      if (channelReject) channelReject(new Error('DataChannel closed'));
      for (const [, p] of this._pending) p.reject(new Error('DataChannel closed'));
      this._pending.clear();
    };
    this._channel.onerror = (ev) => {
      console.error('[MeshBay] DataChannel error', ev);
      if (channelReject) channelReject(new Error('DataChannel error'));
    };

    this._pc.onconnectionstatechange = () => {
      console.log('[MeshBay] PC state:', this._pc.connectionState);
    };
    this._pc.oniceconnectionstatechange = () => {
      console.log('[MeshBay] ICE state:', this._pc.iceConnectionState);
    };

    const offer = await this._pc.createOffer();
    await this._pc.setLocalDescription(offer);

    await new Promise((resolve) => {
      if (this._pc.iceGatheringState === 'complete') return resolve();
      this._pc.onicegatheringstatechange = () => {
        if (this._pc.iceGatheringState === 'complete') resolve();
      };
    });

    const resp = await fetch(`${this._hubUrl}/v1/nodes/${nodeId}/webrtc/offer`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this._accessToken}`,
      },
      body: JSON.stringify({
        sdp: this._pc.localDescription.sdp,
        ice_candidates: [],
      }),
    });

    if (!resp.ok) {
      const detail = await resp.json().catch(() => ({}));
      throw new Error(`Signaling failed: ${resp.status} ${detail.detail || ''}`);
    }

    const answer = await resp.json();
    this._rawAnswerSdp = answer.sdp;
    await this._pc.setRemoteDescription({ type: 'answer', sdp: answer.sdp });

    await channelReady;

    // The client nonce is what makes the NODE's proof fresh (C3) — without it a
    // recorded handshake_ack could be replayed by an impersonating peer.
    this._nonceClient = crypto.getRandomValues(new Uint8Array(32));

    const reply = await this._sendAndWait({
      type: 'handshake',
      v: '0.1',
      token: jwtToken,
      group_id: groupId || '',
      nonce: window.MeshBayCrypto.b64encode(this._nonceClient),
    });

    if (reply.type === 'handshake_challenge') {
      if (!window.MeshBayCrypto) {
        throw new Error('Node requires GEK proof but no crypto available');
      }

      // Recorded the moment the challenge arrives, because everything below may
      // need them — joining, in particular, happens before the proof and signs a
      // transcript over both. Reading them further down, next to the proof that
      // also uses them, meant join_request ran with neither.
      //
      // nonce_node ties a join to this connection, so one cannot be lifted onto
      // another. node_pk is announced here because a first-time member has no
      // GEK and so cannot complete the handshake that would prove it; it is
      // unverified at this point and checked against the ack below.
      this._nonceNode = window.MeshBayCrypto.b64decode(reply.nonce);
      this.nodePk = reply.node_pk || null;

      // Our identity for THIS node: fetched from it, or created if this is a
      // first join. Keys are per node, so there is nothing to carry between
      // them — and an operator who cracks the copy on their own disk gets a key
      // that opens nothing anywhere else.
      let fresh = false;
      if (!this._sessionKeys && this._bundleKey && window.MeshBayKeys) {
        const kpResp = await this._sendAndWait({
          type: 'keypair_bundle_fetch', v: '0.1',
        });
        if (kpResp.type === 'keypair_bundle_resp' && kpResp.found) {
          const keys = await window.MeshBayKeys.decryptBundleWithKey(
            kpResp.bundle_enc, this._bundleKey);
          const pkXB64 = await _pkFromSk(keys.skX);
          this._sessionKeys = { skXB64: keys.skX, skEdB64: keys.skEd, pkXB64 };
        } else {
          // This node has never seen us. Generate the identity we will use here
          // and nowhere else; it is stored on this node once the join succeeds,
          // which is what lets another browser become the same person here.
          const id = await window.MeshBayKeys.generateNodeIdentity(this._bundleKey);
          this._sessionKeys = {
            skEdB64: id.skEdB64, skXB64: id.skXB64, pkXB64: id.pkXB64,
          };
          this._newNodeBundle = id.bundleEnc;
          fresh = true;
        }
      }

      // An identity this node already knows still needs its group key, which the
      // node wraps on every connection.
      if (!gekRaw && this._sessionKeys && !fresh) {
        const bundleResp = await this._sendAndWait({
          type: 'gek_bundle_fetch', v: '0.1',
        });
        if (bundleResp.type === 'gek_bundle_resp' && bundleResp.found) {
          const skXRaw = Uint8Array.from(atob(this._sessionKeys.skXB64), c => c.charCodeAt(0));
          const myPkX = Uint8Array.from(atob(this._sessionKeys.pkXB64), c => c.charCodeAt(0));
          try {
            gekRaw = await window.MeshBayCrypto.unwrapGEK(bundleResp, skXRaw, myPkX);
            this._gekRaw = gekRaw;
          } catch (e) {
            console.warn('[MeshBay] stored GEK bundle did not open; joining instead');
          }
        }
      }

      // No stored bundle: ask the node to recognise us and wrap the key itself.
      // This is the normal path for anyone who joined after the invite redesign —
      // no bundle is pre-stored for members any more. A code is needed only the
      // first time this node sees this account.
      if (!gekRaw && this._sessionKeys && userId) {
        try {
          gekRaw = await this.joinGroup(userId, groupId, joinCode);
        } catch (e) {
          // The UI turns this into "ask the operator for an invite code".
          this._joinError = e;
        }
      }

      if (!gekRaw && !this._sessionKeys) {
        // No identity keys in this browser and none recoverable from the node:
        // the keypair bundle is created where you register and only reaches a
        // node after a first successful connection, so a brand-new member opening
        // a second browser has nothing to sign or unwrap with. Say that, rather
        // than blaming the GEK — a code prompt here would be useless, since a
        // code proves who you are and we have no key to bind to.
        const err = new Error(
          'This browser does not hold your keys. Open the group once from the '
          + 'browser where you registered — after that this one can recover them.');
        err.reason = 'no_keys';
        throw err;
      }

      if (!gekRaw) {
        throw this._joinError
          || new Error('Node requires GEK proof but no GEK available');
      }

      const C = window.MeshBayCrypto;
      // Node's answer SDP carries ITS fingerprint; our offer carries ours. Throws
      // if either is missing rather than proceeding with an unbound proof (L4).
      const binding = C.webrtcBinding(
        _extractDtlsFingerprint(this._pc.localDescription.sdp),
        _extractDtlsFingerprint(this._rawAnswerSdp),
      );
      const nonceNode = this._nonceNode;   // captured when the challenge arrived
      const gid = groupId || '';

      const proof = await C.handshakeProof(
        gekRaw, 'client', gid, this._nonceClient, nonceNode, binding);

      const ack = await this._sendAndWait({
        type: 'handshake_response',
        v: '0.1',
        proof: C.b64encode(proof),
      });
      if (ack.type !== 'handshake_ack') {
        throw new Error('GEK proof rejected: ' + (ack.detail || JSON.stringify(ack)));
      }

      // Authenticate the NODE before trusting anything it says (C3). Until this
      // ran, node_pk was decorative: a peer that had hijacked signaling could
      // accept our proof, ignore it, and serve a forged index, chat history and
      // is_node_admin flag.
      const expected = await C.handshakeProof(
        gekRaw, 'node', gid, this._nonceClient, nonceNode, binding);
      if (!ack.proof || !C.constantTimeEqual(C.b64decode(ack.proof), expected)) {
        throw new Error('Node failed to prove GEK possession — refusing connection');
      }
      const transcript = C.handshakeTranscript(
        'node', gid, this._nonceClient, nonceNode, binding);
      if (!ack.node_pk || !ack.sig
          || !await C.verifyNodeSignature(ack.node_pk, ack.sig, transcript)) {
        throw new Error('Node signature invalid — refusing connection');
      }
      // Trust On First Use (11.5.8). With C6 closed, a substituted node already
      // fails the GEK proof — this covers the case where an attacker HAS the GEK
      // (an ex-member, or a leaked key) and swaps the node underneath.
      // Strict refusal: a warning users can click through is decorative.
      // The key announced in the challenge must be the one that just proved
      // itself. A peer that changed identity mid-handshake is not one to trust
      // with anything, including a join we may already have signed for it.
      if (this.nodePk && this.nodePk !== ack.node_pk) {
        throw new Error('Node identity changed during the handshake — refusing');
      }
      _checkNodePin(nodeId, ack.node_pk);
      this.nodePk = ack.node_pk;

      return ack;
    }

    // A node that answers a handshake with anything other than a challenge is not
    // running the mutual protocol. Accepting a bare handshake_ack here would let a
    // peer skip proving GEK possession entirely (C3/C6).
    const rejected = new Error(
      'MNP handshake rejected: ' + (reply.detail || `unexpected ${reply.type}`));
    // `not_a_member` usually means our token predates being added to the group;
    // the caller refreshes it and tries again rather than showing that to someone
    // who was invited thirty seconds ago.
    rejected.reason = reply.code || '';
    throw rejected;
  }

  /**
   * Pair this browser with the node using a one-time code (M3, and the same
   * substitution as H3).
   *
   * The node has no way to know which key belongs to its operator unless someone
   * tells it locally — asking the hub would let the hub name itself node
   * administrator. The code comes from `meshbay-node operator pair`, over SSH, and
   * the hub never sees it.
   */
  async pairOperator(userId, code) {
    if (!this._connected) throw new Error('Not connected to the node');
    if (!userId) throw new Error('Missing user id');
    if (!this._sessionKeys || !this._sessionKeys.skEdB64 || !this._sessionKeys.skXB64) {
      throw new Error('Identity keys unavailable in this browser — sign in again');
    }
    if (!this._nonceNode || !this.nodePk) {
      throw new Error('Handshake incomplete — reconnect and retry');
    }

    const C = window.MeshBayCrypto;
    // Both public keys are derived from OUR OWN secret keys, never read back from
    // the hub: signing a public key the directory handed us would reintroduce the
    // substitution this whole mechanism exists to close.
    const pkEdB64 = await _pkEdFromSk(this._sessionKeys.skEdB64);
    const pkXB64 = await _pkFromSk(this._sessionKeys.skXB64);
    const ts = Math.floor(Date.now() / 1000);

    // group_id is empty: operator authority is node-wide, not per group.
    const transcript = C.joinTranscript(
      this.nodePk, '', userId, pkEdB64, pkXB64, this._nonceNode, ts);
    const sig = await window.MeshBayKeys.signBytes(this._sessionKeys.skEdB64, transcript);

    const resp = await this._sendAndWait({
      type: 'join_request',
      v: '0.1',
      group_id: '',
      pk_ed25519: pkEdB64,
      pk_x25519: pkXB64,
      code: code || '',
      ts,
      sig,
    });

    if (resp.type === 'error') throw new Error(resp.detail || 'Pairing refused');
    if (resp.type !== 'join_result' || !resp.ok) {
      const reason = resp.reason || 'unknown';
      const err = new Error(JOIN_REFUSALS[reason] || `Pairing refused: ${reason}`);
      err.reason = reason;
      throw err;
    }
    this.memberRole = 'operator';
    return resp;
  }

  async fetchIndex() {
    const msg = await this._sendAndWait({ type: 'index_sync', v: '0.1' });
    if (msg.type === 'error') throw new Error(msg.detail);
    return msg;
  }

  async fetchChunk(fileId, chunkIndex) {
    const msg = await this._sendAndWait({
      type: 'file_req',
      v: '0.1',
      file_id: fileId,
      chunk_index: chunkIndex,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    return msg;
  }

  async fetchStreamSegment(fileId, segmentIndex, segmentDuration) {
    const msg = await this._sendAndWait({
      type: 'stream_seg',
      v: '0.1',
      file_id: fileId,
      segment_index: segmentIndex,
      segment_duration: segmentDuration || 4,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    return _b64decode(msg.data_b64);
  }

  async fetchChatHistory(since, limit) {
    const msg = await this._sendAndWait({
      type: 'chat_hist',
      v: '0.1',
      since: since || 0,
      limit: limit || 100,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    return msg.messages || [];
  }

  async sendChat(payload, iteration, threadId, senderName) {
    const msg = await this._sendAndWait({
      type: 'chat_msg',
      v: '0.1',
      payload: payload,
      iteration: iteration || 0,
      thread_id: threadId || null,
      sender_name: senderName || null,
    });
    return msg;
  }

  /**
   * Authorize a privileged node operation with the user's Ed25519 identity key.
   *
   * The client rebuilds the signed transcript from the challenge fields and refuses
   * to sign unless the operation and subject match what the user actually asked for.
   * Previously the node sent 32 opaque random bytes and the client signed them
   * blind, which let any peer obtain a signature over content of its choosing
   * (finding H5).
   */
  async _authorizeAdminOp(challenge, expectedOp, expectedSubject, signFn) {
    if (challenge.op !== expectedOp || challenge.subject !== expectedSubject) {
      throw new Error(
        `Refusing to sign: node asked to authorize "${challenge.op}" on ` +
        `"${challenge.subject}", but the requested action was "${expectedOp}" ` +
        `on "${expectedSubject}"`);
    }
    if (!signFn) throw new Error('Admin challenge received but no signing key available');

    const transcript = window.MeshBayCrypto.adminTranscript(
      challenge.op, challenge.node_pk, challenge.group_id,
      challenge.subject, challenge.nonce, challenge.ts);

    const signature = await signFn(transcript);
    const ack = await this._sendAndWait({
      type: 'admin_response',
      v: '0.1',
      op_id: challenge.op_id,
      signature,
    });
    if (ack.type === 'error') throw new Error(ack.detail);
    return ack;
  }

  async deleteFile(fileId, signFn) {
    const msg = await this._sendAndWait({
      type: 'file_delete',
      v: '0.1',
      file_id: fileId,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    if (msg.type === 'admin_challenge') {
      return this._authorizeAdminOp(msg, 'file_delete', fileId, signFn);
    }
    return msg;
  }

  /**
   * Remove an empty directory. Operator only, and the node checks that — this
   * signs with the identity it pinned for us, exactly like deleting a file.
   */
  async deleteDirectory(dir, signFn) {
    const msg = await this._sendAndWait({
      type: 'dir_delete',
      v: '0.1',
      dir,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    if (msg.type === 'admin_challenge') {
      // `dir`, not msg.subject: comparing the node's answer against itself is
      // no check at all, and the point of this one is that we know what we
      // asked for without being told.
      return this._authorizeAdminOp(msg, 'dir_delete', dir, signFn);
    }
    return msg;
  }

  /**
   * Stop this node serving the group key to someone. Operator only.
   *
   * Only the node can do this: its roster decides who it serves. Removing them
   * on the hub is the other half, and neither implies the other.
   */
  async revokeMember(userId, signFn) {
    const msg = await this._sendAndWait({
      type: 'member_revoke', v: '0.1', user_id: userId,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    if (msg.type === 'admin_challenge') {
      return this._authorizeAdminOp(msg, 'member_revoke', userId, signFn);
    }
    return msg;
  }

  /**
   * Ask for a video stream, and say how much we can take.
   *
   * `credits` bounds what is in flight. Without it the node pushes the whole
   * film as fast as ffmpeg produces it and the browser holds all of it while
   * MediaSource consumes a segment at a time — which is fine for a clip and
   * fatal for anything worth streaming.
   */
  requestStream(fileId, credits = STREAM_CREDITS) {
    this._send({ type: 'stream_req', v: '0.1', file_id: fileId, credits });
  }

  /** Room for `n` more segments. */
  grantStreamCredit(n = 1) {
    if (!this._connected) return;
    this._send({ type: 'stream_more', v: '0.1', n });
  }

  /**
   * Nobody is watching any more.
   *
   * Closing the viewer used to say nothing to the node, which went on
   * transcoding and holding one of its two slots until the credit timeout — so
   * the next video answered "server busy".
   */
  stopStream() {
    if (!this._connected) return;
    try { this._send({ type: 'stream_stop', v: '0.1' }); } catch { /* gone */ }
  }

  /**
   * Push a whole file, several chunks in flight at once.
   *
   * One chunk per round trip is 48 KB of throughput per RTT no matter how much
   * bandwidth there is: 4.8 MB/s on a 10 ms path, 480 KB/s on a 100 ms one, and
   * the sender is idle for almost all of it — which also keeps SCTP's congestion
   * window shut, so the transport never gets a chance to speed up either. A
   * window of chunks makes the rate depend on bandwidth rather than distance.
   *
   * Order is not at risk: a DataChannel is ordered and reliable by default, and
   * the node refuses any chunk that is not the one it expects next.
   *
   * The node decides where this lands (uploads/) and under what name — it finds a
   * free one rather than replacing anything. The ack says which, and that is what
   * this returns.
   */
  async uploadFile(file, { chunkSize, onProgress, signal } = {}) {
    // The same file twice at once would confuse the node, which keys its own
    // upload state by name — and would race for the same destination.
    if (this._uploaders.has(file.name)) {
      throw new Error(`${file.name} is already being uploaded`);
    }
    const size = chunkSize || UPLOAD_CHUNK_SIZE;
    const total = Math.max(1, Math.ceil(file.size / size));
    let acked = 0;
    let stored = null;
    let failure = null;

    const acks = [];
    this._uploaders.set(file.name, (msg) => {
      if (msg.type === 'error') {
        failure = new Error(msg.detail || 'Upload refused');
      } else if (msg.stored_as) {
        stored = msg;
      }
      acked += 1;
      if (onProgress) onProgress(Math.min(file.size, acked * size), file.size);
      const waiter = acks.shift();
      if (waiter) waiter();
    });

    const nextAck = () => new Promise(r => acks.push(r));

    try {
      for (let i = 0; i < total; i++) {
        if (signal && signal.aborted) throw _aborted();
        // Backpressure: without it the whole file lands in the browser's send
        // buffer in seconds and the progress bar becomes a work of fiction.
        while (this._channel && this._channel.bufferedAmount > UPLOAD_BUFFER_HIGH) {
          if (signal && signal.aborted) throw _aborted();
          await new Promise(r => setTimeout(r, 20));
        }
        while (i - acked >= UPLOAD_WINDOW) {
          await nextAck();
          if (failure) throw failure;
        }
        if (failure) throw failure;

        const buf = new Uint8Array(
          await file.slice(i * size, (i + 1) * size).arrayBuffer());
        this._send({
          type: 'file_upload',
          v: '0.1',
          filename: file.name,
          chunk_index: i,
          total_chunks: total,
          data: buf,
        });
      }
      while (acked < total) {
        await nextAck();
        if (failure) throw failure;
      }
    } finally {
      this._uploaders.delete(file.name);
    }
    return stored || {};
  }

  /** Create a directory under the current one. Any member may. */
  async createDirectory(dir, name) {
    const msg = await this._sendAndWait({
      type: 'dir_create', v: '0.1', dir: dir || '', name,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    return msg;
  }

  /**
   * Ask the node for a one-time pairing code admitting `userId` to this group.
   *
   * This replaces wrapping the group key in the browser. We no longer fetch the
   * invitee's public key from the hub, so the hub can no longer answer with its own
   * and be handed the group key (H3). The node wraps the key later, itself, for a
   * key the invitee proves possession of.
   *
   * Returns {code, expires_at} — the code is displayed once and passed to the
   * invitee out of band.
   */
  async createInvite(userId, groupId, username, signFn) {
    const msg = await this._sendAndWait({
      type: 'invite_create',
      v: '0.1',
      user_id: userId,
      group_id: groupId,
      username: username || '',
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    if (msg.type === 'admin_challenge') {
      return this._authorizeAdminOp(msg, 'invite_create', userId, signFn);
    }
    return msg;
  }

  /**
   * Ask the node to recognise us and hand over the group key.
   *
   * Sent when we hold no GEK for a group. `code` is needed only the first time
   * this node sees this account (and not at all in an open-join group).
   */
  async joinGroup(userId, groupId, code) {
    if (!this._sessionKeys || !this._sessionKeys.skEdB64 || !this._sessionKeys.skXB64) {
      throw new Error('Identity keys unavailable in this browser — sign in again');
    }
    if (!this._nonceNode || !this.nodePk) {
      throw new Error('Handshake incomplete — reconnect and retry');
    }

    const C = window.MeshBayCrypto;
    const pkEdB64 = await _pkEdFromSk(this._sessionKeys.skEdB64);
    const pkXB64 = await _pkFromSk(this._sessionKeys.skXB64);
    const ts = Math.floor(Date.now() / 1000);

    const transcript = C.joinTranscript(
      this.nodePk, groupId || '', userId, pkEdB64, pkXB64, this._nonceNode, ts);
    const sig = await window.MeshBayKeys.signBytes(this._sessionKeys.skEdB64, transcript);

    const resp = await this._sendAndWait({
      type: 'join_request',
      v: '0.1',
      group_id: groupId || '',
      pk_ed25519: pkEdB64,
      pk_x25519: pkXB64,
      code: code || '',
      ts,
      sig,
    });

    if (resp.type === 'error') throw new Error(resp.detail || 'Join refused');
    if ((resp.type !== 'join_result' || !resp.ok) || !resp.gek) {
      const reason = resp.reason || 'unknown';
      const err = new Error(JOIN_REFUSALS[reason] || `Join refused: ${reason}`);
      // The UI reacts to `code_required` by asking for one; everything else is
      // shown as-is.
      err.reason = reason;
      throw err;
    }

    // Unwrap with our own secret key — the node wrapped for the public key we
    // just proved we hold, so nobody else can open this.
    const skXRaw = Uint8Array.from(atob(this._sessionKeys.skXB64), c => c.charCodeAt(0));
    const myPkX = Uint8Array.from(atob(pkXB64), c => c.charCodeAt(0));
    const gekRaw = await C.unwrapGEK(resp, skXRaw, myPkX);
    this._gekRaw = gekRaw;
    // What the node's roster says this identity is, which is not what the hub
    // says: `operator` here means this browser's key was paired with the node,
    // not merely that the account owns it.
    this.memberRole = resp.role || '';
    return gekRaw;
  }

  /**
   * Withdraw our key backup from this node.
   *
   * The counterpart of storeKeypairBundle: turning the setting off has to remove
   * what is already stored, not merely stop adding to it — otherwise the blob
   * stays on every node the account has ever joined (C4).
   */
  async deleteKeypairBundle() {
    const msg = await this._sendAndWait({
      type: 'keypair_bundle_delete', v: '0.1',
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    return msg;
  }

  async storeKeypairBundle(bundleEnc) {
    const msg = await this._sendAndWait({
      type: 'keypair_bundle_store',
      v: '0.1',
      bundle_enc: bundleEnc,
    });
    if (msg.type === 'error') throw new Error(msg.detail);
    return msg;
  }

  get gekRaw() { return this._gekRaw; }

  close() {
    if (this._channel) this._channel.close();
    if (this._pc) this._pc.close();
    this._connected = false;
    for (const [, p] of this._pending) p.reject(new Error('Transport closed'));
    this._pending.clear();
  }

  // ── Internal ──────────────────────────────────────────────────────────────

  _sendAndWait(obj) {
    return new Promise((resolve, reject) => {
      const id = this._seqId++;
      const timeout = setTimeout(() => {
        this._pending.delete(id);
        reject(new Error('Response timeout'));
      }, 30000);
      this._pending.set(id, {
        _reqType: obj.type,
        // Chunks are the one request that runs several at a time and can be
        // interleaved with anything else on the channel. Matching them by
        // arrival order was only ever true by luck; this makes it true.
        _key: obj.type === 'file_req'
          ? `chunk:${obj.file_id}:${obj.chunk_index}` : null,
        resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); },
        reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); },
      });
      this._send(obj);
    });
  }

  _send(obj) {
    if (!this._channel || this._channel.readyState !== 'open') {
      throw new Error(`DataChannel not open (state: ${this._channel?.readyState})`);
    }
    const encoded = msgpack_encode(obj);
    const header = new Uint8Array(4);
    new DataView(header.buffer).setUint32(0, encoded.byteLength, false);
    const frame = new Uint8Array(4 + encoded.byteLength);
    frame.set(header);
    frame.set(encoded, 4);
    this._channel.send(frame);
  }

  _onMessage(data) {
    const incoming = new Uint8Array(data);
    const combined = new Uint8Array(this._recvBuf.length + incoming.length);
    combined.set(this._recvBuf);
    combined.set(incoming, this._recvBuf.length);
    this._recvBuf = combined;

    while (this._recvBuf.length >= 4) {
      const len = new DataView(this._recvBuf.buffer, this._recvBuf.byteOffset).getUint32(0, false);
      if (this._recvBuf.length < 4 + len) break;
      const msgBytes = this._recvBuf.slice(4, 4 + len);
      this._recvBuf = this._recvBuf.slice(4 + len);

      const msg = msgpack_decode(msgBytes);
      this._dispatch(msg);
    }
  }

  _dispatch(msg) {
    // While an upload is in flight the acks are its own, and there are many of
    // them: they must not be handed to whatever request happens to be oldest in
    // the pending map.
    if (msg.type === 'file_upload_ack' && this._uploaders.has(msg.filename)) {
      this._uploaders.get(msg.filename)(msg);
      return;
    }
    // An error carries no filename. With one upload running it is that
    // upload's; with several there is no way to tell, so they all hear it and
    // stop — which is the safe reading of an error on a shared channel.
    if (msg.type === 'error' && this._uploaders.size) {
      for (const handler of [...this._uploaders.values()]) handler(msg);
      return;
    }
    if (msg.type === 'chat_msg' && this._onChat) {
      this._onChat(msg);
      return;
    }
    if (msg.type === 'stream_init') {
      if (this._onStreamInit) this._onStreamInit(msg);
      return;
    }
    if (msg.type === 'stream_data') {
      if (this._onStreamData) this._onStreamData(msg);
      return;
    }
    if (msg.type === 'stream_end') {
      if (this._onStreamEnd) this._onStreamEnd(msg);
      return;
    }

    if (msg.type === 'index_sync' && msg.entries) {
      if (this._onIndexSync) this._onIndexSync(msg);
      const oldest = this._pending.entries().next();
      if (!oldest.done && oldest.value[1]._reqType === 'index_sync') {
        oldest.value[1].resolve(msg);
      }
      return;
    }

    if (msg.type === 'file_chunk') {
      const key = `chunk:${msg.file_id}:${msg.chunk_index}`;
      for (const [, handler] of this._pending) {
        // A node from before the reply carried a file_id: fall back to the
        // index, which is still better than the oldest pending request.
        const match = msg.file_id
          ? handler._key === key
          : handler._key && handler._key.endsWith(`:${msg.chunk_index}`);
        if (match) {
          handler.resolve(msg);
          return;
        }
      }
      // Nobody asked for it any more — a cancelled download, most likely. It
      // must not be handed to whatever request happens to be waiting.
      return;
    }

    const oldest = this._pending.entries().next();
    if (!oldest.done) {
      const [, handler] = oldest.value;
      handler.resolve(msg);
    }
  }
}

// ── Minimal msgpack encode/decode ────────────────────────────────────────────
// Covers the subset used by MNP: maps, strings, integers, binary, arrays, null.

function msgpack_encode(obj) {
  const parts = [];
  _encodeValue(obj, parts);
  const total = parts.reduce((s, p) => s + p.length, 0);
  const result = new Uint8Array(total);
  let off = 0;
  for (const p of parts) { result.set(p, off); off += p.length; }
  return result;
}

function _encodeValue(val, parts) {
  if (val === null || val === undefined) {
    parts.push(new Uint8Array([0xc0]));
  } else if (typeof val === 'boolean') {
    parts.push(new Uint8Array([val ? 0xc3 : 0xc2]));
  } else if (typeof val === 'number') {
    if (Number.isInteger(val)) {
      if (val >= 0 && val <= 127) {
        parts.push(new Uint8Array([val]));
      } else if (val >= 0 && val <= 0xff) {
        parts.push(new Uint8Array([0xcc, val]));
      } else if (val >= 0 && val <= 0xffff) {
        const b = new Uint8Array(3); b[0] = 0xcd;
        new DataView(b.buffer).setUint16(1, val, false);
        parts.push(b);
      } else if (val >= 0 && val <= 0xffffffff) {
        const b = new Uint8Array(5); b[0] = 0xce;
        new DataView(b.buffer).setUint32(1, val, false);
        parts.push(b);
      } else if (val >= -32 && val < 0) {
        parts.push(new Uint8Array([val & 0xff]));
      } else if (val >= -128 && val < 0) {
        const b = new Uint8Array(2); b[0] = 0xd0; b[1] = val & 0xff;
        parts.push(b);
      } else {
        const b = new Uint8Array(5); b[0] = 0xd2;
        new DataView(b.buffer).setInt32(1, val, false);
        parts.push(b);
      }
    } else {
      const b = new Uint8Array(9); b[0] = 0xcb;
      new DataView(b.buffer).setFloat64(1, val, false);
      parts.push(b);
    }
  } else if (typeof val === 'string') {
    const encoded = new TextEncoder().encode(val);
    if (encoded.length <= 31) {
      parts.push(new Uint8Array([0xa0 | encoded.length]));
    } else if (encoded.length <= 0xff) {
      parts.push(new Uint8Array([0xd9, encoded.length]));
    } else if (encoded.length <= 0xffff) {
      const b = new Uint8Array(3); b[0] = 0xda;
      new DataView(b.buffer).setUint16(1, encoded.length, false);
      parts.push(b);
    } else {
      const b = new Uint8Array(5); b[0] = 0xdb;
      new DataView(b.buffer).setUint32(1, encoded.length, false);
      parts.push(b);
    }
    parts.push(encoded);
  } else if (val instanceof Uint8Array) {
    if (val.length <= 0xff) {
      parts.push(new Uint8Array([0xc4, val.length]));
    } else if (val.length <= 0xffff) {
      const b = new Uint8Array(3); b[0] = 0xc5;
      new DataView(b.buffer).setUint16(1, val.length, false);
      parts.push(b);
    } else {
      const b = new Uint8Array(5); b[0] = 0xc6;
      new DataView(b.buffer).setUint32(1, val.length, false);
      parts.push(b);
    }
    parts.push(val);
  } else if (Array.isArray(val)) {
    if (val.length <= 15) {
      parts.push(new Uint8Array([0x90 | val.length]));
    } else if (val.length <= 0xffff) {
      const b = new Uint8Array(3); b[0] = 0xdc;
      new DataView(b.buffer).setUint16(1, val.length, false);
      parts.push(b);
    } else {
      const b = new Uint8Array(5); b[0] = 0xdd;
      new DataView(b.buffer).setUint32(1, val.length, false);
      parts.push(b);
    }
    for (const item of val) _encodeValue(item, parts);
  } else if (typeof val === 'object') {
    const keys = Object.keys(val);
    if (keys.length <= 15) {
      parts.push(new Uint8Array([0x80 | keys.length]));
    } else if (keys.length <= 0xffff) {
      const b = new Uint8Array(3); b[0] = 0xde;
      new DataView(b.buffer).setUint16(1, keys.length, false);
      parts.push(b);
    } else {
      const b = new Uint8Array(5); b[0] = 0xdf;
      new DataView(b.buffer).setUint32(1, keys.length, false);
      parts.push(b);
    }
    for (const k of keys) {
      _encodeValue(k, parts);
      _encodeValue(val[k], parts);
    }
  }
}

function msgpack_decode(buf) {
  const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength);
  const [val] = _decodeValue(buf, view, 0);
  return val;
}

function _decodeValue(buf, view, offset) {
  const byte = buf[offset];

  if (byte <= 0x7f) return [byte, offset + 1];
  if ((byte & 0xe0) === 0xe0) return [byte - 256, offset + 1];
  if ((byte & 0xa0) === 0xa0) {
    const len = byte & 0x1f;
    return [new TextDecoder().decode(buf.slice(offset + 1, offset + 1 + len)), offset + 1 + len];
  }
  if ((byte & 0xf0) === 0x90) {
    const len = byte & 0x0f;
    return _decodeArray(buf, view, offset + 1, len);
  }
  if ((byte & 0xf0) === 0x80) {
    const len = byte & 0x0f;
    return _decodeMap(buf, view, offset + 1, len);
  }

  switch (byte) {
    case 0xc0: return [null, offset + 1];
    case 0xc2: return [false, offset + 1];
    case 0xc3: return [true, offset + 1];
    case 0xc4: { const len = buf[offset + 1]; return [buf.slice(offset + 2, offset + 2 + len), offset + 2 + len]; }
    case 0xc5: { const len = view.getUint16(offset + 1, false); return [buf.slice(offset + 3, offset + 3 + len), offset + 3 + len]; }
    case 0xc6: { const len = view.getUint32(offset + 1, false); return [buf.slice(offset + 5, offset + 5 + len), offset + 5 + len]; }
    case 0xcc: return [buf[offset + 1], offset + 2];
    case 0xcd: return [view.getUint16(offset + 1, false), offset + 3];
    case 0xce: return [view.getUint32(offset + 1, false), offset + 5];
    case 0xcb: return [view.getFloat64(offset + 1, false), offset + 9];
    case 0xd0: return [view.getInt8(offset + 1), offset + 2];
    case 0xd1: return [view.getInt16(offset + 1, false), offset + 3];
    case 0xd2: return [view.getInt32(offset + 1, false), offset + 5];
    case 0xd9: {
      const len = buf[offset + 1];
      return [new TextDecoder().decode(buf.slice(offset + 2, offset + 2 + len)), offset + 2 + len];
    }
    case 0xda: {
      const len = view.getUint16(offset + 1, false);
      return [new TextDecoder().decode(buf.slice(offset + 3, offset + 3 + len)), offset + 3 + len];
    }
    case 0xdb: {
      const len = view.getUint32(offset + 1, false);
      return [new TextDecoder().decode(buf.slice(offset + 5, offset + 5 + len)), offset + 5 + len];
    }
    case 0xdc: { const len = view.getUint16(offset + 1, false); return _decodeArray(buf, view, offset + 3, len); }
    case 0xdd: { const len = view.getUint32(offset + 1, false); return _decodeArray(buf, view, offset + 5, len); }
    case 0xde: { const len = view.getUint16(offset + 1, false); return _decodeMap(buf, view, offset + 3, len); }
    case 0xdf: { const len = view.getUint32(offset + 1, false); return _decodeMap(buf, view, offset + 5, len); }
    default: throw new Error(`Unknown msgpack type: 0x${byte.toString(16)}`);
  }
}

function _decodeArray(buf, view, offset, count) {
  const arr = [];
  for (let i = 0; i < count; i++) {
    const [val, newOff] = _decodeValue(buf, view, offset);
    arr.push(val);
    offset = newOff;
  }
  return [arr, offset];
}

function _decodeMap(buf, view, offset, count) {
  const obj = {};
  for (let i = 0; i < count; i++) {
    const [key, off1] = _decodeValue(buf, view, offset);
    const [val, off2] = _decodeValue(buf, view, off1);
    obj[key] = val;
    offset = off2;
  }
  return [obj, offset];
}

function _b64decode(b64) {
  const binary = atob(b64);
  const bytes = new Uint8Array(binary.length);
  for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i);
  return bytes;
}

function _extractDtlsFingerprint(sdp) {
  const match = sdp.match(/a=fingerprint:sha-256 ([0-9A-Fa-f:]+)/);
  if (!match) return new Uint8Array(0);
  const hex = match[1].replace(/:/g, '');
  const bytes = new Uint8Array(hex.length / 2);
  for (let i = 0; i < hex.length; i += 2)
    bytes[i / 2] = parseInt(hex.substring(i, i + 2), 16);
  return bytes;
}

// ── Node identity pinning (11.5.8) ───────────────────────────────────────────

const NODE_PIN_PREFIX = 'mb_nodepin_';

function _checkNodePin(nodeId, nodePk) {
  if (!nodeId || !nodePk) return;
  const key = NODE_PIN_PREFIX + nodeId;

  let pinned = null;
  try { pinned = localStorage.getItem(key); } catch { return; }

  if (pinned === null) {
    try { localStorage.setItem(key, nodePk); } catch {}
    return;
  }
  if (pinned !== nodePk) {
    throw new Error(
      'This node\'s identity key has changed. That is expected only if its ' +
      'operator reinstalled the node — otherwise someone may be impersonating ' +
      'it. Verify with the operator out of band, then clear the pin in ' +
      'Settings to accept the new key.');
  }
}

/** Forget a pinned node identity — the deliberate escape hatch for a legitimate rotation. */
function clearNodePin(nodeId) {
  try {
    if (nodeId) localStorage.removeItem(NODE_PIN_PREFIX + nodeId);
    else {
      for (const k of Object.keys(localStorage))
        if (k.startsWith(NODE_PIN_PREFIX)) localStorage.removeItem(k);
    }
  } catch {}
}

function pinnedNodeCount() {
  try {
    return Object.keys(localStorage).filter(k => k.startsWith(NODE_PIN_PREFIX)).length;
  } catch { return 0; }
}

// Export
MeshBayTransport.clearNodePin = clearNodePin;
MeshBayTransport.pinnedNodeCount = pinnedNodeCount;
window.MeshBayTransport = MeshBayTransport;