kernel-swarm/src/index.ts

// @ts-ignore
import Hyperswarm from "@lumeweb/hyperswarm-web";
import type { ActiveQuery } from "libkmodule";
import { addHandler, getSeed, handleMessage } from "libkmodule";
import { handlePresentSeed as handlePresentSeedModule } from "libkmodule/dist/seed.js";
import type { Buffer } from "buffer";
import * as ed from "@noble/ed25519";
import b4a from "b4a";

interface DhtConnection {
  swarm: number;
  conn: any;
}

const connections = new Map<number, DhtConnection>();
const swarmInstances = new Map<number, Hyperswarm>();

let defaultSwarm: Hyperswarm;

let moduleReadyResolve: Function;
let moduleReady: Promise<void> = new Promise((resolve) => {
  moduleReadyResolve = resolve;
});

onmessage = handleMessage;
function idFactory(start = 1) {
  let id = start;

  return function nextId() {
    const nextId = id;
    id += 1;
    return nextId;
  };
}

const getSwarmId = idFactory();
const getSocketId = idFactory();

addHandler("presentSeed", handlePresentSeed);
addHandler("joinPeer", handleJoinPeer);
addHandler("getPeerByPubkey", handleGetPeerByPubkey);
addHandler("listenSocketEvent", handleListenSocketEvent, {
  receiveUpdates: true,
});
addHandler("socketExists", handleSocketExists);
addHandler("close", handleCloseSocketEvent);
addHandler("socketWrite", handleWriteSocketEvent);
addHandler("addRelay", handleAddRelay);
addHandler("removeRelay", handleRemoveRelay);
addHandler("clearRelays", handleClearRelays);
addHandler("getRelays", handleGetRelays);
addHandler("ready", handleReady);

async function handlePresentSeed(aq: ActiveQuery) {
  handlePresentSeedModule({
    callerInput: {
      seed: {
        publicKey: await ed.getPublicKey(aq.callerInput.rootKey),
        secretKey: aq.callerInput.rootKey,
      },
    },
  } as ActiveQuery);

  if (!defaultSwarm) {
    defaultSwarm = swarmInstances.get(await createSwarm()) as Hyperswarm;
  }
  moduleReadyResolve();
}

async function createSwarm(): Promise<number> {
  const swarmInstance = new Hyperswarm({ keyPair: await getSeed() });
  const id = getSwarmId();
  swarmInstances.set(id, swarmInstance);

  swarmInstance.on("connection", (peer) => {
    const socketId = getSocketId();
    connections.set(socketId, { swarm: id, conn: peer });

    peer.on("close", () => {
      connections.delete(socketId);
    });
  });

  return id;
}

function handleListenSocketEvent(aq: ActiveQuery) {
  const { event = null } = aq.callerInput;

  const socket = validateConnection(aq);

  if (!socket) {
    return;
  }

  if (!event) {
    aq.reject("Invalid event");
    return;
  }

  let responded = false;
  const respond = () => {
    if (responded) {
      return;
    }

    responded = true;
    aq.respond();
  };

  const cb = (data: Buffer) => {
    aq.sendUpdate(data);
  };

  socket.on(event, cb);
  socket.on("close", () => {
    socket.off(event, cb);
    respond();
  });

  aq.setReceiveUpdate?.((data: any) => {
    switch (data?.action) {
      case "off":
        socket.off(event, cb);
        respond();
        break;
    }
  });
}

async function handleSocketExists(aq: ActiveQuery) {
  const { id = null } = aq.callerInput;

  aq.respond(connections.has(Number(id)));
}

function handleCloseSocketEvent(aq: ActiveQuery) {
  const socket = validateConnection(aq);

  if (!socket) {
    return;
  }

  socket.end();

  aq.respond();
}

function handleWriteSocketEvent(aq: ActiveQuery) {
  const socket = validateConnection(aq);

  if (!socket) {
    return;
  }
  const { message = null } = aq.callerInput;

  if (!message) {
    aq.reject("empty message");
    return false;
  }

  socket.write(message);

  aq.respond();
}

function validateConnection(aq: ActiveQuery): any | boolean {
  const { id = null } = aq.callerInput;

  if (!id || !connections.has(id)) {
    aq.reject("Invalid connection id");
    return false;
  }

  return connections.get(id)?.conn;
}

async function getSwarm(aq: ActiveQuery): Promise<Hyperswarm> {
  await moduleReady;
  let swarm;
  if ("callerInput" in aq && aq.callerInput) {
    swarm = aq.callerInput.swarm ?? null;

    if (swarm && !swarmInstances.has(swarm)) {
      const error = "Invalid swarm id";
      aq.reject(error);
      throw new Error(error);
    }
  }

  if (!swarm) {
    return defaultSwarm;
  }

  return swarmInstances.get(swarm) as Hyperswarm;
}

async function handleAddRelay(aq: ActiveQuery) {
  const { pubkey = null } = aq.callerInput;

  if (!pubkey) {
    aq.reject("invalid pubkey");
    return;
  }

  const dht = await getSwarm(aq);

  aq.respond(await dht.addRelay(pubkey));
}

async function handleRemoveRelay(aq: ActiveQuery) {
  const { pubkey = null } = aq.callerInput;

  if (!pubkey) {
    aq.reject("invalid pubkey");
    return;
  }

  const dht = await getSwarm(aq);

  aq.respond(dht.removeRelay(pubkey));
}

async function handleClearRelays(aq: ActiveQuery) {
  const dht = await getSwarm(aq);

  dht.clearRelays();

  aq.respond();
}

async function handleGetRelays(aq: ActiveQuery) {
  aq.respond(await (await getSwarm(aq)).relays);
}

async function handleJoinPeer(aq: ActiveQuery) {
  const { topic = null } = aq.callerInput;

  const swarm = await getSwarm(aq);

  if (!topic) {
    aq.reject("invalid topic");
    return;
  }
  if (!b4a.isBuffer(topic)) {
    aq.reject("topic must be a buffer");
    return;
  }

  // @ts-ignore
  swarm.join(topic);
  aq.respond();
}
async function handleGetPeerByPubkey(aq: ActiveQuery) {
  const { pubkey = null } = aq.callerInput;

  const swarm = await getSwarm(aq);

  if (!pubkey) {
    aq.reject("invalid topic");
    return;
  }

  if (!b4a.isBuffer(pubkey)) {
    aq.reject("pubkey must be a buffer");
    return;
  }

  // @ts-ignore
  if (!swarm._allConnections.has(pubkey)) {
    aq.reject("peer does not exist");
    return;
  }

  // @ts-ignore
  const peer = swarm._allConnections.get(pubkey);

  aq.respond(
    [...connections.entries()].filter((conn) => {
      return conn[1].conn === peer;
    })[0][0]
  );
}

async function handleReady(aq: ActiveQuery) {
  const swarm = await getSwarm(aq);
  // @ts-ignore
  await swarm.ready();
  aq.respond();
}