index.js

/*
	Copyright (C) 2023 by Eric Chen

	This file is part of p2p_share_compute.

	p2p_share_compute is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

	p2p_share_compute is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

	You should have received a copy of the GNU General Public License along with p2p_share_compute. If not, see <https://www.gnu.org/licenses/>.
*/

const express = require('express');
const { performance } = require('perf_hooks');
const os = require('os');
const msgpack = require('msgpack');
const { WebSocket, WebSocketServer } = require('ws');
const { v4: uuidv4 } = require('uuid');
const { verify } = require('./verifyClient.js');
const { VM } = require('vm2');
const EventEmitter = require('events');
const fetch = require('cross-fetch');
const http = require('http');
const qs = require('querystring');
// idea: use websockets to connect

/* Error Handling */
process.on('uncaughtException', (err, origin) => {
	console.error('Error:');
	console.error(err);
	console.error(`Origin: ${origin}`);
});
process.on('unhandledRejection', (reason, promise) => {
	console.log('Unhandled Promise Rejection:');
	console.error(reason);
	console.error(`Origin: ${promise}`)
});


function getFlops() {
	const data = [];
	const operations = 1e6;
	for (let i = 0; i < 50; ++i) {
		const start = performance.now();
		let j = 0.5;
		const endval = 0.5+1.5*operations;
		while (j < endval) {
			j += 1.5;
		}
		const end = performance.now();
		const time = (end - start)/1000;
		const flops = operations/time;
		data.push(flops);
	}
	return Math.floor(data.reduce((a, b) => a+b, 0)/data.length);
}
function getFreeMemory() {
	return Math.floor(os.freemem());
}
class Peer extends EventEmitter {
	constructor({bootstrap = false /* is this the bootstrap server? */ , peerlist = [], origin, bootstrapServer, initialConnectNumber, verifyOrigin = true, ttl = 7, rejectOnFailureMessage, searchTries, logMessages, timeout, onReady = () => {}}) {
		super();
		// Setup basic variables
		if (!origin) throw Error('no origin provided');
		this.peerlist = peerlist;
		this.ttl = ttl;
		this.rejectOnFailureMessage = rejectOnFailureMessage;
		this.searchTries = searchTries;
		this.logMessages = logMessages;
		this.bootstrapServer = bootstrapServer;
		this.origin = origin;
		this.verifyOrigin = verifyOrigin;
		this.timeout = timeout;
		this.peerCache = {};
		this.connected = {};
		this.tokens = {};
		this.sessions = {}; // { <peer ip>: {<id>: { code: <code>, id: <session id> (persists for 5 minutes before it is deleted), vm: <>, stdin: [], ... }, ...}, ... }
		this.runSessions = {};
		for (const peer of peerlist) {
			this.peerCache[peer] = {covered:{}};
		}
		this.updateStats();
		setTimeout((async () => {
			await this.updateStats();
		}).bind(this), 5 * 60 * 1000); // every 5 minutes

		// Create HTTP server
		//console.log('creating peer');
		this.server = express();
		this.server.use(express.raw({ type: '*/*' }));
		this.server.post('/:type', (req, res) => {
			let data;
			try {
				data = JSON.parse(msgpack.unpack(req.body));
			} catch (err) {
				console.error(`Error while decoding HTTP data from IP ${req.ip} (not necessarily peer, just the origin IP)`);
				console.error(err);
				// don't log actual data as that might flood the console
				this.replyHttp(res, {type:'error', error:'INVALID'});
				return;
			}
			if (!data.origin || (this.verifyOrigin && req.ip !== data.origin) || !req.params.type || typeof req.params.type !== 'string' || req.params.type.slice(0,3) === 'ws_') {
				return this.replyHttp(res, {type:'error', error:'INVALID'});
			}
			if (this.logMessages) console.log(`HTTP Request ${data.origin} -> ${this.origin}: ${JSON.stringify(data)}`);
			this.emit(req.params.type, data, req, res);
		});
		this.httpServer = http.createServer(this.server);
		this.httpServer.listen(19292, () => {
			console.log('Listening on port 19292');
		});

		// setup websocket server
		this.wss = new WebSocketServer({ noServer: true });

		this.httpServer.on('upgrade', (req, socket, head) => {
			if (this.logMessages) console.log(`Upgrade requested at URL ${req.url}`);
			const parts = req.url.split('?');
			//console.log(parts);
			let parsed;
			if (parts.length >= 2) {
				parsed = qs.parse(parts.slice(-1)[0]);
			}
			//console.log(parsed, this.tokens);
			if (parts.length <= 1 || !parsed.token || !this.tokens[parsed.token]) {
				socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n');
				socket.destroy();
				return;
			}
			this.wss.handleUpgrade(req, socket, head, socket => {
				this.wss.emit('connection', socket, req);
			});
		});

		this.wss.on('connection', socket => {
			let origin;
			socket.on('message', msg => {
				let data;
				try {
					data = JSON.parse(msgpack.unpack(msg));
				} catch (err) {
					if (!origin) {
						console.error(`Error while decoding Websocket data from IP ${req.ip} (not necessarily peer, just the origin IP):`);
					} else {
						console.error(`Error while decoding Websocket data from peer ${origin}:`);
					}
					console.error(err);
					// don't log actual data as that might flood the console with binary data, clogging it up
					this.sendWs(socket, {type:'error', type:'INVALID'});
					return;
				}
				if (!data.origin || (this.verifyOrigin && req.ip !== data.origin) || !data.type || typeof data.type !== 'string' || data.type.slice(0,3) === 'ws_') {
					return this.sendWs(socket, {type:'error', error:'INVALID'});
				}
				if (!origin) origin = data.origin;
				if (this.logMessages) console.log(`Websocket Client ${data.origin} -> ${this.origin}: ${JSON.stringify(data)}`);
				this.emit('ws_' + data.type, data, socket);
			});
			socket.on('close', () => {
				if (!origin) return;
				for (const id in this.sessions[origin]) {
					setTimeout((() => {
						delete this.sessions[origin][id];
					}).bind(this), 5 * 60 * 1000 /* clear after 5 minutes */);
				}
			});
			// the session storing stuff is done 
			// there's not really any point in logging intended disconnects/abrupt disconnects since it doesn't really change anything and is just extra clunk/storage used up
		}); 

		// Setup bootstrapping
		if (bootstrap) {
			this.on('getpeers', async ({index, number, origin}, req, res) => {
				if (!(typeof index === 'number') || !(typeof number === 'number')) return (await this.replyHttp(res, {type:'error',error:'INVALID'}));
				if (!(origin in this.peerCache)) {
					this.peerCache[origin] = {};
					peerlist.push(origin);
				}
				if (index >= peerlist.length) {
					return (await this.replyHttp(res, {type:'error', error:'PASTINDEX'}));
				}
				return (await this.replyHttp(res, {type: "peers", peers: peerlist.slice(index, index+number)}));
			});
			onReady();
		} else {
			// find peers
			// ask bootstrap server
			this.findAndConnect(initialConnectNumber, onReady);
		}

		// Handle different HTTP message types
		this.on('compinfo', async (data, req, res) => {
			if (!data.ram || !data.flops) console.warn(`Warning: missing RAM or FLOPS data from peer ${data.origin}. Available data: ${data}`);
			await this.replyHttp(res, {type:'compinfo', ram: this.memory, flops: this.flops });
		});
		this.on('error', (data, req, res) => {
			console.error(`Error from ${data.origin}: ${data.error}`);
		});
		this.on('search', async (data, req, res) => {
			if (!data.ram || !data.flops || typeof data.ttl !== 'number' || !data.start || typeof data.origTtl !== 'number' || !data.id) return (await this.replyHttp(res, {type:'error', error:'INVALID'}));
			// just report ok first to sending peer
			await this.replyHttp(res, {type:'ok'});
			data.ignore = data.ignore || {};
			// FIRST, check our own specs to see if they work
			if (this.memory >= data.ram && this.flops >= data.flops && !data.ignore[this.origin]) {
				return (await this.sendHttp(data.start, {type:'searchRes', peer: this.origin, id: data.id, status: true}));
			} else {
				// NOW, check peer cache specs
				let minPower = -1, minPowerPeer;
				const ratio = data.ram / data.flops;
				for (const peer of Object.keys(this.peerCache)) {
					const { ram, flops } = this.peerCache[peer];
					const power = Math.floor(ram + (ratio * flops));
					if ((minPower === -1 || power < minPower) && !this.peerCache[peer].covered[data.id]) {
						minPower = power;
						minPowerPeer = peer;
					}
					if (ram >= data.ram && flops >= data.flops && !data.ignore[peer]) {
						return (await this.sendHttp(data.start, { type: 'searchRes', peer, id: data.id, status: true }));
					}
				}
				// this means that a working peer hasn't been found in this machine OR a peer in the peerCache
				// discard if ttl <= 1 and report failure because the hop from peer X to our peer is the last hop
				// forward otherwise
				if (data.ttl <= 1) {
					return 
				}
				
				// now subtract 1 from ttl and forward to lowest spec peer since that one is the most likely to connect with higher spec peers
				// we must also check if it has been covered already
				// if all have been covered, then there's nothing else we can do at this point other than exit
				if (minPower === -1) {
					return (await this.sendHttp(data.start, {type:'searchRes', id: data.id, status: false}));
				}
				this.peerCache[minPowerPeer].covered[data.id] = true;
				await this.sendHttp(minPowerPeer, {
					type:'search', 
					ram: data.ram, 
					flops: data.flops, 
					ignore: data.ignore, 
					start: data.start, 
					origTtl: data.origTtl, 
					ttl: data.ttl - 1,
					id: data.id
				});
			}
		});
			// This is handled seperately in the Peer.search() function
		/*this.on('searchRes', (data, req, res) => {

		});*/
		this.on('reqconn', async (data, req, res) => {
			if (await verify(data, req, res)) {
				// Generate a UUID as a token
				const token = uuidv4();
				this.tokens[token] = true;
				await this.replyHttp(res, {type:'conninfo', token });
				setTimeout((() => {
					this.tokens[token] = false;
				}).bind(this), 5 * 60 * 1000 /* clear after 5 minutes */ );
			} else {
				await this.replyHttp(res, {type:'error',error:'UNAUTH'});
			}
		});

		// Handle different Websocket message types
		// repeating this here for reference:
		// { <peer ip>: {<id>: { code: <code>, vm: <>, stdin, ... }, ...}, ... }
		this.on('ws_run', (data, socket) => {
			if (!data.code || !data.msgId) return this.sendWs(socket, {type:'error', error:'INVALID'});
			if (!this.sessions[data.origin]) this.sessions[data.origin] = [];
			const id = uuidv4();
			this.sendWs(socket, {type:'sessId', id, reply: data.msgId /* data.msgId should be a UUID */ });
			if ('stdin' in data && !(data.stdin instanceof Array)) data.stdin = [data.stdin];
			this.sessions[data.origin][id] = {
				code: data.code,
				id,
				stdin: data.stdin || []
			};

			// printing functions
			const printLine = ((...args) => this.sendWs(socket, {type:'stdout', data: args.join(' ') + '\n', id})).bind(this);
			const printError = ((...args) => this.sendWs(socket, {type:'stderr', data: args.join(' ') + '\n', id})).bind(this);
			const readLine = ((...args) => {
				const input = this.sessions[data.origin][id].stdin;
				if (input.length === 0) {
					return '';
				} else {
					return input.shift();
				}
			}).bind(this);

			const vm = new VM({
				timeout: this.timeout,
				allowAsync: false,
				sandbox: { printLine, printError, readLine }
			});
			try {
				vm.run(data.code);
			} catch (err) {
				if (socket.readyState === socket.OPEN) {
					printError(err.toString());
				} else {
					console.error('Error from program from ' + data.origin + ' : ');
					console.error(err);
				}
			}
		});
		this.on('ws_stdin', (data, socket) => {
			if (typeof data.data !== 'string' || !data.id) return this.sendWs(socket, {type:'error', error:'INVALID'});
			this.sessions[data.origin][data.id].stdin.push(data.data);
		});
		this.on('ws_clear', (data, socket) => {
			if (!data.id) this.sendWs(socket, {type:'error', error:'INVALID'});
			delete this.sessions[data.origin][data.id];
		});
	}
	updateStats() {
		//console.log('updating stats');
		this.flops = getFlops();
		//console.log('im running');
		this.memory = getFreeMemory();
		if (this.logMessages) {
			console.log('Flops: ' + this.flops);
			console.log('Free memory: ' + this.memory);
		}
	}
	async findAndConnect(initialConnectNumber, cb = () => {}) {
		let index = 0;
		let errors = 0;
		await this.connectTo(this.bootstrapServer);
		do {
			let res;
			try {
				res = await this.sendHttp(this.bootstrapServer, {type:'getpeers', index, number: initialConnectNumber});
				if (res.type === 'error') {
					console.error('Error code from bootstrap server: ' + res.error);
					const connectedTo = Object.keys(this.connected).length;
					if (connectedTo < initialConnectNumber) {
						console.error(`Unable to connect to config.initialConnectNumber (${initialConnectNumber}), not enough peers found. Currently connected to ${Object.keys(this.connected).length} peers`);
					}
					break;
				}
			} catch (err) {
				console.error('Error while connecting to bootstrapServer: ' + err);
				++errors;
				if (errors < 10) {
					continue;
				} else {
					console.error('10 failed attempts to connect to bootstrap server, quitting.');
					console.error('You should probably restart the program and try again.');
					return;
				}
			}
			index += initialConnectNumber;
			res = res.peers;
			for (const peer of res) {
				await this.connectTo(peer);
			}
		} while (Object.keys(this.connected).length < initialConnectNumber);
		console.log(`Connected to ${Object.keys(this.connected).length} peers`);
		cb();
	}
	async connectTo(peer) {
		if (this.connected[peer] || peer === this.origin) return;
		try {
			const peerData = await this.sendHttp(peer, {type:'compinfo', ram: this.memory, flops: this.flops });
			if (!peerData.ram || !peerData.flops) {
				throw Error('Missing RAM or FLOPS data from peer, available data: ' + JSON.stringify(peerData));
			}
			this.peerCache[peer] = { ram: peerData.ram, flops: peerData.flops, covered: {} };
			this.connected[peer] = true;
		} catch (err) {
			console.error(`error while connecting to peer ${peer}:`);
			console.error(err);
		}
	}
	async sendHttp(peer, data) {
		data['origin'] = this.origin;
		if (this.logMessages) console.log(`HTTP Request ${this.origin} -> ${peer}: ${JSON.stringify(data)}`);
		const res = await fetch(`http://${peer}:19292/${data.type}`, {
			method: 'POST',
			headers: {
				'Content-Type': 'application/json'
			},
			body: msgpack.pack(JSON.stringify(data))
		});
		const body = Buffer.from(await res.arrayBuffer());
		const decoded = JSON.parse(msgpack.unpack(body)); // not much we can do since the other server has probably already "left"
		if (this.logMessages) console.log(`HTTP Response ${peer} -> ${this.origin}: ${JSON.stringify(decoded)}`);
		//this.emit(decoded.type, decoded); // should be handled seperately at where it was called
		// .on should only be for events/queries coming from outside
		return decoded;
	}
	async replyHttp(res, data) {
		data['origin'] = this.origin;
		if (this.logMessages) console.log(`HTTP Response ${this.origin} -> ?: ${JSON.stringify(data)}`);
		res.send(msgpack.pack(JSON.stringify(data)));
	}
	async sendWs(socket, data) {
		data['origin'] = this.origin;
		if (this.logMessages) console.log(`Websocket Message ${this.origin} -> ${socket._socket.remoteAddress} (IP, not peer): ${JSON.stringify(data)}`);
		socket.send(msgpack.pack(JSON.stringify(data)));
	}
	// no replyWs, since sendWs is just the equivalent for sendWs
	listenWs(socket, listener) {
		socket.on('message', msg => {
			let data;
			try {
				data = JSON.parse(msgpack.unpack(msg));
			} catch (err) {
				console.error(`Error while decoding Websocket data from IP ${socket._socket.remoteAddress} (not necessarily peer, just the origin IP)`);
				console.error(err);
				// don't log actual data as that might flood the console with binary data, clogging it up
				this.sendWs(socket, {type:'error', type:'INVALID'});
				return;
			}
			if (!data.origin || (this.verifyOrigin && req.ip !== data.origin) || !data.type || typeof data.type !== 'string' || data.type.slice(0,3) === 'ws_') {
				return this.sendWs(socket, {type:'error', error:'INVALID'});
			}
			listener(data);
		});
	}

	search(ram, flops, ignore = []) {
		// basically send a request to ourselves to initiate the search
		// if we ourselves satisfy the requirements ig we can run on our own machine??? 

		// start the chain
		return new Promise((resolve, reject) => {
			const ignoreObj = {};
			for (const p of ignore) {
				ignoreObj[p] = true;
			}
			const id = uuidv4();
			this.sendHttp(this.origin, {type:'search', ram, flops, ignore: ignoreObj, start: this.origin, origTtl: this.ttl, ttl: this.ttl, id });
			this.on('searchRes', async (data, req, res) => {
				if (data.id === id) {
					if (data.status) {
						if (!data.peer || ignoreObj[data.peer]) {
							return this.replyHttp(res, {type:'error', error:'INVALID'});
						}
						resolve(data.peer);
					} else {
						if (this.rejectOnFailureMessage) {
							const err = new Error('Reported failure from ' + data.origin);
							err.name = 'SearchFailureError';
							reject(err);
						} else {
							console.error('Reported failure from ' + data.origin);
						}
					}
				}
			});
		});
	}
	
	async findPeer(ram, flops, ignore = []) {
		let attempts = 0;
		let found = false;
		while (attempts < this.searchTries && !found) {
			attempts += 1;
			let peer;
			try {
				peer = await this.search(ram, flops, ignore);
			} catch (err) {
				if (err.type === 'SearchFailureError') {
					console.error('Search failed, consider increasing TTL');
				}
				throw err;
			}
			// then we attempt to connect
			const connData = await this.sendHttp(peer, {type:'reqconn'});
			// if that doesn't work we initiate another search with that peer excluded
			// and so on, until we find a suitable peer or reach the limit
			if (connData.error === 'UNAUTH') {
				ignore.push(peer);
				continue;
			}
			if (!connData.token) {
				ignore.push(peer);
				await this.sendHttp(peer, {type:'error', error:'INVALID'});
				continue;
			}
			found = true;
			// now connect with ws
			const ws = new WebSocket(`ws://${peer}:19292/execute?token=${connData.token}`);
			ws.on('error', error => {
				console.error('Failed to connect to peer ' + peer + ': ');
				console.error(error)
				throw error;
			});
			ws.on('close', () => {
				console.log(`WebSocket connection with peer ${peer} closed`);
			});
			return (await new Promise((resolve, reject) => {
				try {
					ws.on('open', (() => {
						this.runSessions[peer] = {ws, sessions: {}};
						resolve(peer);
						this.listenWs(ws, data => {
							if (this.logMessages) console.log(`Websocket Server ${data.origin} -> ${this.origin}: ${JSON.stringify(data)}`);
						});
					}).bind(this));
				} catch (err) {
					reject(err);
				}
			}));
		}
		throw Error(`Unable to find suitable peer in ${this.searchTries} tries, maybe consider increasing searchTries?`)
	}
	run(peer, code, startingStdin = [], onStdout = () => {}, onStderr = () => {}) {
		return new Promise((resolve, reject) => {
			try {
				const id = uuidv4();
				const ws = this.runSessions[peer].ws;
				this.listenWs(ws, async data => {
					if (data.type === 'sessId' && (!data.id || !data.reply)) return (await this.sendWs(ws, {type:'error', error:'INVALID'}));
					if (data.reply === id) {
						this.runSessions[peer].sessions[data.id] = {};
						this.listenWs(ws, async data1 => {
							if (data1.id === data.id) {
								if (data1.type === 'stdout') onStdout(data1.data);
								else if (data1.type === 'stderr') onStderr(data1.data);
							}
						});
						resolve(data.id);
					}
				});
				this.sendWs(ws, {type:'run', code, msgId:id, stdin: startingStdin});
			} catch (err) {
				reject(err);
			}
		});
	}
	async sendStdin(peer, session, data) {
		await this.sendWs(this.runSessions[peer].ws, {type:'stdin', data, id: session});
	}
	async clearProgram(peer, session) {
		await this.sendWs(this.runSessions[peer].ws, {type:'clear', id: session})
	}
	disconnect(peer, persist = true) {
		this.runSessions[peer].ws.close();
		if (!persist) delete this.runSessions[peer];
		else delete this.runSessions[peer].ws;
	}
	async reconnect(peer) {
		if (!this.runSessions[peer]) throw Error('No data persisted for peer ' + peer);
		const connData = await this.sendHttp(peer, {type:'reqconn'});
		
		if (connData.error === 'UNAUTH') {
			throw Error(`Failed to reconnect to peer ${peer}: authorization failed`)
		}
		if (!connData.token) {
			await this.sendHttp(peer, {type:'error', error:'INVALID'});
			throw Error(`Failed to reconnect to peer ${peer}: No token given by server`);
		}
		
		const ws = new WebSocket(`ws://${peer}:19292/execute?token=${connData.token}`);
		ws.on('error', error => {
			console.error('Failed to connect to peer ' + peer + ': ');
			console.error(error);
		});
		ws.on('close', () => {
			console.log(`WebSocket connection with peer ${peer} closed`);
		});
		ws.on('open', (() => {
			this.runSessions[peer].ws = ws;
			this.listenWs(ws, data => {
				if (this.logMessages) console.log(`Websocket Server ${data.origin} -> ${this.origin}: ${JSON.stringify(data)}`);
			});
		}).bind(this));
		return ws;
	}
}

module.exports = Peer;