bobbydigitales/dandelion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add deno-only bootstrap peer implementation. Uses libdatachannel for the RTCDatachannel implementation.

Browse Source 
Fix Typescript sourcemap serving
PeerManager: more robust when RTCPeerConnection fails or is not present
Separate source maps so the main files arent bloated
This commit is contained in:
bobbydigitales
2025-06-10 20:23:19 -07:00
parent a96d7e28c8
commit d0fd041f7e
27 changed files with 132 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/App.ts
View File

@@ -60,35 +60,35 @@ class StatusBar {
peerStatus = new Map();
headless = false;
setMessageHTML(html:string) {
setMessageHTML(html: string) {
let statusBarElement = document.getElementById('status_bar');
if (!statusBarElement) {
return;
}
statusBarElement.innerHTML = html;
statusBarElement.innerHTML = html;
}
setHeadless(headless:boolean) {
setHeadless(headless: boolean) {
this.headless = headless;
}
updatePeerMessage(peerID: string, message: string) {
this.peerStatus.set(peerID, {message, data:this.peerStatus.get(peerID)?.data});
this.peerStatus.set(peerID, { message, data: this.peerStatus.get(peerID)?.data });
this.render();
}
updatePeerData(peerID:PeerID, data:any) {
this.peerStatus.set(peerID, {message:this.peerStatus.get(peerID)?.message, data:data});
updatePeerData(peerID: PeerID, data: any) {
this.peerStatus.set(peerID, { message: this.peerStatus.get(peerID)?.message, data: data });
}
updatePeerStatus(peerID:PeerID, message:string="", data={}) {
this.peerStatus.set(peerID, {message, data});
updatePeerStatus(peerID: PeerID, message: string = "", data = {}) {
this.peerStatus.set(peerID, { message, data });
this.render();
}
getPeerData(peerID:PeerID) {
getPeerData(peerID: PeerID) {
let status = this.peerStatus.get(peerID);
if (status) {
return status.data;
@@ -105,8 +105,8 @@ class StatusBar {
let newStatus = "";
for (let [peerID, status] of this.peerStatus.entries()) {
let statusBarItem = `<span>(${logID(peerID)} | ${status.message}) </span>`;
newStatus += statusBarItem;
let statusBarItem = `<span>(${logID(peerID)} | ${status.message}) </span>`;
newStatus += statusBarItem;
}
this.setMessageHTML(newStatus);
@@ -187,7 +187,7 @@ export class App {
if (neededPostIDs.length > 0) {
console.log.apply(null, log(`[app] Need (${neededPostIDs.length}) posts for user ${logID(userID)} from peer ${logID(peerID)}`));
let neededPostCount = neededPostIDs.length;
this.statusBar.updatePeerStatus(peerID, `need(${logID(userID)} | ${neededPostCount})`, {havePostCount:0, neededPostCount:neededPostCount});
this.statusBar.updatePeerStatus(peerID, `need(${logID(userID)} | ${neededPostCount})`, { havePostCount: 0, neededPostCount: neededPostCount });
let neededPosts = await this.peerManager?.rpc.getPostsForUser(peerID, this.peerID, userID, neededPostIDs);
}
@@ -245,7 +245,7 @@ export class App {
this.statusBar.updatePeerStatus(sendingPeerID, `getPostIDs(${logID(userID)})⬆️`);
let postIDs = await this.peerManager?.rpc.getPostIDsForUser(sendingPeerID, userID);
this.statusBar.updatePeerStatus(sendingPeerID, `syncing(${logID(userID)} ${postIDs.length})`);
console.log.apply(null, log(`[app] Got (${postIDs.length}) post IDs for user [${logID(userID)}] from peer [${logID(sendingPeerID)}]`));
@@ -272,8 +272,9 @@ export class App {
}
let knownUsers = await this.sync.getKnownUsers();
this.peerManager.rpc.announceUsers(event.peerID, this.peerID, knownUsers);
// rpc saying what peers we have
this.peerManager.rpc.announceUsers(event.peerID, this.peerID, knownUsers);
});
this.peerManager.addEventListener(PeerEventTypes.PEER_DISCONNECTED, async (event: any) => {
@@ -310,7 +311,7 @@ export class App {
this.peerManager.registerRPC('getPostsForUser', async (requestingPeerID: string, userID: string, postIDs: string[]) => {
let posts = await this.sync.getPostsForUser(userID, postIDs);
let i=0;
let i = 0;
for (let post of posts) {
console.log.apply(null, log(`[app] sendPostForUser sending post [${logID(post.post_id)}] to [${logID(requestingPeerID)}]`, userID, post.author, post.text));
@@ -330,7 +331,7 @@ export class App {
// if (post.text === "image...") {
// debugger;
// }
let peerData = this.statusBar.getPeerData(sendingPeerID);
if (peerData) {
this.statusBar.updatePeerMessage(sendingPeerID, `⬇️${logID(userID)} ${peerData.havePostCount}/${peerData.neededPostCount}}`);
@@ -342,7 +343,7 @@ export class App {
peerData.havePostCount++
this.statusBar.updatePeerMessage(sendingPeerID, `⬇️${logID(userID)} ${peerData.havePostCount}/${peerData.neededPostCount}}`);
}
if (this.renderTimer) {
clearTimeout(this.renderTimer);
}
@@ -1198,9 +1199,9 @@ export class App {
console.log(`[headless]${this.isHeadless} [archive] ${this.isArchivePeer} [bootstrap] ${this.isBootstrapPeer}`);
this.statusBar.setHeadless(this.isHeadless);
let limitPostsParam = urlParams.get('limitPosts');
if (limitPostsParam) {

67
src/PeerManager.ts
View File

@@ -6,7 +6,7 @@
import { generateID } from "IDUtils";
import { log, logID } from "log";
import { App } from "./App";
// import { App } from "./App";
// Use a broadcast channel to only have one peer manager for multiple tabs,
// then we won't need to have a session ID as all queries for a peerID will be coming from the same peer manager
@@ -26,7 +26,7 @@ export class PeerManager {
searchQueryFunctions: Map<string, Function> = new Map();
RPC_remote: Map<string, Function> = new Map();
rpc: { [key: string]: Function } = {};
isBootstrapPeer: boolean = false;
_isBootstrapPeer: boolean = false;
bootstrapPeerConnections: Map<string, PeerConnection> | null = null;
sessionID = generateID();
userID: string;
@@ -84,6 +84,11 @@ export class PeerManager {
return peername;
}
isBootstrapPeer(peerID: string) {
return this.bootstrapPeerIDs?.has(peerID)
}
websocketSend(message: any) {
if (!this.websocket) {
throw new Error();
@@ -118,7 +123,7 @@ export class PeerManager {
if (message.type === "hello2") {
if (!this.isBootstrapPeer && Array.isArray(message?.bootstrapPeers)) {
if (!this._isBootstrapPeer && Array.isArray(message?.bootstrapPeers)) {
this.bootstrapPeerIDs = new Set(message.bootstrapPeers);
}
@@ -144,7 +149,7 @@ export class PeerManager {
if (!peerConnection) {
let remotePeerID = message.from;
let newPeer = new PeerConnection(this, remotePeerID, this.websocketSendPeerMessage.bind(this));
if (this.isBootstrapPeer) {
if (this._isBootstrapPeer) {
newPeer.setPolite(false);
}
peerConnection = newPeer;
@@ -177,7 +182,7 @@ export class PeerManager {
async onHello2Received(bootstrapPeerIDs: Set<string> | null) {
if (this.isBootstrapPeer) {
if (this._isBootstrapPeer) {
this.connectPromiseCallbacks?.resolve();
return;
}
@@ -195,7 +200,7 @@ export class PeerManager {
// this.bootstrapPeerConnection = await this.connectToPeer(peerID);
let bootstrapPeerConnectionPromise = new Promise( async (resolve, reject)=>{
let bootstrapPeerConnectionPromise = new Promise(async (resolve, reject) => {
let peerConnection = await this.connectToPeer(peerID);
if (!peerConnection) {
reject(peerConnection);
@@ -221,7 +226,7 @@ export class PeerManager {
peer_id: this.peerID,
session_id: this.sessionID,
// peer_name: app.peername,
is_bootstrap_peer: this.isBootstrapPeer,
is_bootstrap_peer: this._isBootstrapPeer,
// peer_description: this.rtcPeerDescription
});
}
@@ -246,7 +251,7 @@ export class PeerManager {
}
constructor(userID: string, peerID: string, isBootstrapPeer: boolean) {
this.isBootstrapPeer = isBootstrapPeer;
this._isBootstrapPeer = isBootstrapPeer;
this.peers = new Map();
this.routingTable = new Map();
this.userID = userID;
@@ -324,7 +329,7 @@ export class PeerManager {
}
if (!this.isBootstrapPeer && numActive === 0) {
if (!this._isBootstrapPeer && numActive === 0) {
console.log.apply(null, log(`No peers connected, will attempt to reconnect in ${this.reconnectPeriod} seconds...`));
// Websocket reconnect
@@ -342,6 +347,13 @@ export class PeerManager {
let output = `Current status:` + "\n" + `[${logID(this.peerID)}]${this.getPeername(this.peerID)}[local]` + "\n";
for (let [peerID, peer] of this.peers) {
output += `[${logID(peerID)}]${peer.rtcPeer?.connectionState}:${this.getPeername(peerID)}${this.bootstrapPeerIDs?.has(peerID) ? "[Bootstrap]" : ""}` + "\n";
if (peer.rpcSuperlog) {
for (let transactionID of peer.pendingRPCs.keys()) {
output += `[${logID(transactionID)}]`;
}
}
}
output += `numActivePeers: ${numActive}` + "\n";
@@ -588,7 +600,7 @@ class PeerConnection {
// console.log.apply(null, log([...this.peerManager.peers.keys()]));
if (this.peerManager.isBootstrapPeer) {
if (this.peerManager._isBootstrapPeer) {
this.send({ type: 'initial_peers', from: this.peerManager.peerID, peers: [...this.peerManager.peers.keys()].filter(entry => entry !== this.remotePeerID) });
// this.dataChannel.send(JSON.stringify());
}
@@ -616,6 +628,11 @@ class PeerConnection {
async connect() {
let connectionPromise = new Promise((resolve, reject) => { this.connectionPromise = { resolve, reject } });
if (!(typeof RTCPeerConnection === "function")) {
throw new Error("RTCPeerConnection is not a function, exiting.");
}
this.rtcPeer = new RTCPeerConnection(PeerConnection.config);
this.rtcPeer.onconnectionstatechange = async (e: any) => {
@@ -720,7 +737,11 @@ class PeerConnection {
async onWebsocketMessage(message: any) {
if (message.type == "rtc_connect") {
this.rtcPeer?.setRemoteDescription(message.description);
try {
this.rtcPeer?.setRemoteDescription(message.description);
} catch (e) {
console.log(e);
}
}
@@ -734,7 +755,7 @@ class PeerConnection {
if (!this.rtcPeer) {
throw new Error();
throw new Error("Unable to instantiate RTCPeerConnection, exiting.");
}
let description = null;
@@ -760,7 +781,12 @@ class PeerConnection {
return;
}
this.isSettingRemoteAnswerPending = description.type == "answer";
await this.rtcPeer.setRemoteDescription(description);
try {
await this.rtcPeer.setRemoteDescription(description);
} catch (e) {
console.log("PeerConnection:setRemoteDescription:failed:", e, description);
}
this.isSettingRemoteAnswerPending = false;
if (description.type === "offer") {
await this.rtcPeer.setLocalDescription();
@@ -769,9 +795,11 @@ class PeerConnection {
} else if (candidate) {
try {
await this.rtcPeer.addIceCandidate(candidate);
} catch (err) {
} catch (e) {
if (!this.ignoreOffer) {
throw err;
console.log("PeerConnection:addIceCandidate:failed:", e, candidate);
throw e;
}
}
}
@@ -869,7 +897,7 @@ class PeerConnection {
// Think about a timeout here to auto reject it after a while.
let promise = new Promise((resolve, reject) => {
this.pendingRPCs.set(transactionID, { resolve, reject, functionName });
// setTimeout(() => reject("bad"), 1000);
setTimeout(() => reject(`function:${functionName}[${transactionID}] failed to resolve after 10 seconds.`), 10_000);
});
let message = {
@@ -909,6 +937,7 @@ class PeerConnection {
}
pendingRPC.resolve(message.response);
this.pendingRPCs.delete(message.transaction_id);
}
if (type === "rpc_call") {
@@ -920,12 +949,6 @@ class PeerConnection {
this.rpcSuperlog && console.log.apply(null, log(`[rpc] call: response:`, response));
if (response === undefined) {
return;
}
let responseMessage = { type: 'rpc_response', function_name: message.function_name, transaction_id: message.transaction_id, response: response };
this.send(responseMessage);

3
src/Sync.ts
View File

@@ -137,7 +137,7 @@ export class Sync {
'622ecc28-2eff-44b9-b89d-fdea7c8dd2d5', // Hazel
]);
}
// Rob
if (userID === 'b38b623c-c3fa-4351-9cab-50233c99fa4e') {
following.push(...[
@@ -161,6 +161,7 @@ export class Sync {
following.push(...[
'b38b623c-c3fa-4351-9cab-50233c99fa4e', // Rob
'05a495a0-0dd8-4186-94c3-b8309ba6fc4c', // Martin
'622ecc28-2eff-44b9-b89d-fdea7c8dd2d5', // Hazel
]);
}