network.js

/*jslint node: true */
"use strict";
var bCordova = (typeof window === 'object' && window && window.cordova);
var WebSocket = bCordova ? global.WebSocket : require('ws');
const { SocksProxyAgent } = bCordova ? {} : require('socks-proxy-agent');
const { HttpsProxyAgent } = bCordova ? {} : require('https-proxy-agent');
var WebSocketServer = WebSocket.Server;
var crypto = require('crypto');
var _ = require('lodash');
var async = require('async');
var db = require('./db.js');
const kvstore = require('./kvstore.js');
var constants = require('./constants.js');
var storage = require('./storage.js');
const archiving = require('./archiving.js');
var myWitnesses = require('./my_witnesses.js');
var joint_storage = require('./joint_storage.js');
var validation = require('./validation.js');
var ValidationUtils = require("./validation_utils.js");
var writer = require('./writer.js');
var conf = require('./conf.js');
var mutex = require('./mutex.js');
var catchup = require('./catchup.js');
var privatePayment = require('./private_payment.js');
var objectHash = require('./object_hash.js');
var objectLength = require('./object_length.js');
const { isTooDeeplyNestedOrHasTooManyNodes, sortObject } = require("./string_utils.js");
var ecdsaSig = require('./signature.js');
var eventBus = require('./event_bus.js');
var light = require('./light.js');
var inputs = require('./inputs.js');
var breadcrumbs = require('./breadcrumbs.js');
var mail = require('./mail.js');
var aa_composer = require('./aa_composer.js');
const main_chain = require('./main_chain.js');
var formulaEvaluation = require('./formula/evaluation.js');
const errorToString = require('./formula/error/errorToString.js');
var dataFeeds = require('./data_feeds.js');
var libraryPackageJson = require('./package.json');

var FORWARDING_TIMEOUT = 10*1000; // don't forward if the joint was received more than FORWARDING_TIMEOUT ms ago
var STALLED_TIMEOUT = 5000; // a request is treated as stalled if no response received within STALLED_TIMEOUT ms
var RESPONSE_TIMEOUT = 300*1000; // after this timeout, the request is abandoned
var HEARTBEAT_TIMEOUT = conf.HEARTBEAT_TIMEOUT || 10*1000;
var HEARTBEAT_RESPONSE_TIMEOUT = 60*1000;
var HEARTBEAT_PAUSE_TIMEOUT = 2*HEARTBEAT_TIMEOUT;
var MAX_STATE_VARS = 2000;
const maxPayload = 6 * 1024 * 1024; // 6 MB

var wss;
var arrOutboundPeers = [];
var assocConnectingOutboundWebsockets = {};
var assocUnitsInWork = {};
var assocRequestedUnits = {};

// joints that couldn't be validated because they spend a locally pruned final-bad output; keyed by the unit they are waiting on.
// Can't use the usual unhandled_joints/dependencies tables for this: those consider a dependency satisfied as soon as its row
// exists in the units table, which is already true for a pruned unit even before we recover its content.
// Only small pointers {unit, expiry_ts} are kept here; the (possibly large) joint bodies live in rocksdb under 'wj\n'+unit,
// so that a peer can't exhaust our memory by triggering many/large withheld joints that never get resolved.
let assocUnitsWaitingForPrunedContent = {};

var bStarted = false;
var bCatchingUp = false;
var bWaitingForCatchupChain = false;
var coming_online_time = Date.now();
var assocReroutedConnectionsByTag = {};
var arrWatchedAddresses = []; // does not include my addresses, therefore always empty
var arrTempWatchedAddresses = [];
var last_hearbeat_wake_ts = Date.now();
var peer_events_buffer = [];
var assocKnownPeers = {};
var assocBlockedPeers = {};
var exchangeRates = {};
var knownWitnesses = {};
var bWatchingForLight = false;
var prev_bugreport_hash = '';
let definitions = {}; // cache
let largeHistoryTags = {};

let messagesInWork = {};
let requestsInWork = {};
let currentJointHost = null;

if (bCordova){ // browser
	console.log("defining .on() on ws");
	WebSocket.prototype.on = function(event, callback) {
		var self = this;
		if (event === 'message'){
			this['on'+event] = function(event){
				callback.call(self, event.data);
			};
			return;
		}
		if (event !== 'open'){
			this['on'+event] = callback;
			return;
		}
		// allow several handlers for 'open' event
		if (!this['open_handlers'])
			this['open_handlers'] = [];
		this['open_handlers'].push(callback);
		this['on'+event] = function(){
			self['open_handlers'].forEach(function(cb){
				cb();
			});
		};
	};
	WebSocket.prototype.once = WebSocket.prototype.on;
	WebSocket.prototype.setMaxListeners = function(){};
}

// if not using a hub and accepting messages directly (be your own hub)
var my_device_address;
var objMyTempPubkeyPackage;

function setMyDeviceProps(device_address, objTempPubkey){
	my_device_address = device_address;
	objMyTempPubkeyPackage = objTempPubkey;
}

exports.light_vendor_url = null;

// general network functions

function sendMessage(ws, type, content) {
	var message = JSON.stringify([type, content]);
	if (!ws)
		return console.log("ws is not defined, will not send " + message);
	if (ws.readyState !== ws.OPEN)
		return console.log("readyState="+ws.readyState+' on peer '+ws.peer+', will not send '+message);
	console.log("SENDING "+message.length+" chars "+message+" to "+ws.peer);
	if (bCordova) {
		ws.send(message);
	} else {
		ws.send(message, function(err){
			if (err)
				ws.emit('error', 'From send: '+err);
		});
	}
}


function sendJustsayingToLightVendor(subject, body, handle){
	if (!handle)
		handle = function(){};
	if (!conf.bLight)
		return handle("sendJustsayingToLightVendor cannot be called as full node")
	if (!exports.light_vendor_url){
		console.log("light_vendor_url not set yet");
		return setTimeout(function(){
			sendJustsayingToLightVendor(subject, body, handle);
		}, 1000);
	}
	findOutboundPeerOrConnect(exports.light_vendor_url, function(err, ws){
		if (err)
			return handle("connect to light vendor failed: "+err);
		sendMessage(ws, 'justsaying', {subject: subject, body: body});
		return handle(null);
	});
}


function sendJustsaying(ws, subject, body){
	sendMessage(ws, 'justsaying', {subject: subject, body: body});
}

function sendAllInboundJustsaying(subject, body){
	wss.clients.forEach(function(ws){
		sendMessage(ws, 'justsaying', {subject: subject, body: body});
	});
}

function sendSysVars(ws) {
	if (storage.systemVars.threshold_size.length === 0) {
		console.log(`sys vars not initialized yet, will send later`);
		return setTimeout(sendSysVars, 100, ws);
	}
	sendJustsaying(ws, 'system_vars', storage.systemVars);
}

function sendUpdatedSysVarsToAllLight() {
	wss.clients.forEach(function (ws) {
		if (ws.bSentSysVars || ws.bWatchingSystemVars)
			sendSysVars(ws);
	});
}

function sendSysVarVoteToAllWatchers(subject, value, author_addresses, unit, is_stable) {
	wss.clients.forEach(function (ws) {
		if (ws.bWatchingSystemVars)
			sendJustsaying(ws, 'system_var_vote', { subject, value, author_addresses, unit, is_stable });
	});
}

function sendError(ws, error) {
	sendJustsaying(ws, 'error', error);
}

function sendInfo(ws, content) {
	sendJustsaying(ws, 'info', content);
}

function sendResult(ws, content) {
	sendJustsaying(ws, 'result', content);
}

function sendErrorResult(ws, unit, error) {
	sendResult(ws, {unit: unit, result: 'error', error: error});
}

function sendVersion(ws){
	sendJustsaying(ws, 'version', {
		protocol_version: constants.version, 
		alt: constants.alt, 
		library: libraryPackageJson.name, 
		library_version: libraryPackageJson.version, 
		program: conf.program, 
		program_version: conf.program_version
	});
}

function sendResponse(ws, tag, response){
	var command = ws.assocCommandsInPreparingResponse[tag];
	delete ws.assocCommandsInPreparingResponse[tag];
	if (command && command !== 'custom') // for custom, we decrement in handleRequest
		decrementPending(requestsInWork, ws.host);
	sendMessage(ws, 'response', {tag: tag, command: command, response: response});
}

function sendErrorResponse(ws, tag, error) {
	sendResponse(ws, tag, {error: error});
}


// if a 2nd identical request is issued before we receive a response to the 1st request, then:
// 1. its responseHandler will be called too but no second request will be sent to the wire
// 2. bReroutable flag must be the same
function sendRequest(ws, command, params, bReroutable, responseHandler){
	var request = {command: command};
	if (params)
		request.params = params;
	var content = _.clone(request);
	var tag = objectHash.getBase64Hash(request, true);
	//if (ws.assocPendingRequests[tag]) // ignore duplicate requests while still waiting for response from the same peer
	//    return console.log("will not send identical "+command+" request");
	if (ws.assocPendingRequests[tag]){
		console.log('already sent a '+command+' request to '+ws.peer+', will add one more response handler rather than sending a duplicate request to the wire');
		ws.assocPendingRequests[tag].responseHandlers.push(responseHandler);
	}
	else{
		content.tag = tag;
		// after STALLED_TIMEOUT, reroute the request to another peer
		// it'll work correctly even if the current peer is already disconnected when the timeout fires
		var reroute = !bReroutable ? null : function(){
			console.log('will try to reroute a '+command+' request stalled at '+ws.peer);
			if (!ws.assocPendingRequests[tag])
				return console.log('will not reroute - the request was already handled by another peer');
			ws.assocPendingRequests[tag].bRerouted = true;
			findNextPeer(ws, function(next_ws){ // the callback may be called much later if findNextPeer has to wait for connection
				if (!ws.assocPendingRequests[tag])
					return console.log('will not reroute after findNextPeer - the request was already handled by another peer');
				if (next_ws === ws || assocReroutedConnectionsByTag[tag] && assocReroutedConnectionsByTag[tag].indexOf(next_ws) >= 0){
					console.log('will not reroute '+command+' to the same peer, will rather wait for a new connection');
					eventBus.once('connected_to_source', function(){ // try again
						console.log('got new connection, retrying reroute '+command);
						reroute();
					});
					return;
				}
				console.log('rerouting '+command+' from '+ws.peer+' to '+next_ws.peer);
				ws.assocPendingRequests[tag].responseHandlers.forEach(function(rh){
					sendRequest(next_ws, command, params, bReroutable, rh);
				});
				if (!assocReroutedConnectionsByTag[tag])
					assocReroutedConnectionsByTag[tag] = [ws];
				assocReroutedConnectionsByTag[tag].push(next_ws);
			});
		};
		var reroute_timer = !bReroutable ? null : setTimeout(reroute, STALLED_TIMEOUT);
		var cancel_timer = bReroutable ? null : setTimeout(function(){
			ws.assocPendingRequests[tag].responseHandlers.forEach(function(rh){
				rh(ws, request, {error: "[internal] response timeout"});
			});
			delete ws.assocPendingRequests[tag];
		}, RESPONSE_TIMEOUT);
		ws.assocPendingRequests[tag] = {
			request: request,
			responseHandlers: [responseHandler], 
			reroute: reroute,
			reroute_timer: reroute_timer,
			cancel_timer: cancel_timer
		};
		sendMessage(ws, 'request', content);
	}
	return tag;
}


function deletePendingRequest(ws, tag){
	if (ws && ws.assocPendingRequests && ws.assocPendingRequests[tag]){
		var pendingRequest = ws.assocPendingRequests[tag];
		clearTimeout(pendingRequest.reroute_timer);
		clearTimeout(pendingRequest.cancel_timer);
		delete ws.assocPendingRequests[tag];

		// if the request was rerouted, cancel all other pending requests
		if (assocReroutedConnectionsByTag[tag]){
			assocReroutedConnectionsByTag[tag].forEach(function(client){
				if (client.assocPendingRequests[tag]){
					clearTimeout(client.assocPendingRequests[tag].reroute_timer);
					clearTimeout(client.assocPendingRequests[tag].cancel_timer);
					delete client.assocPendingRequests[tag];
				}
			});
			delete assocReroutedConnectionsByTag[tag];
		}
		return true;
	}else{
		return false;
	}
}


function handleResponse(ws, tag, response){
	if (!ValidationUtils.isNonemptyString(tag))
		return console.log("no tag in response");
	var pendingRequest = ValidationUtils.hasOwnProperty(ws.assocPendingRequests, tag) && ws.assocPendingRequests[tag];
	if (!pendingRequest) // was canceled due to timeout or rerouted and answered by another peer
		//throw "no req by tag "+tag;
		return console.log("no req by tag "+tag);
	pendingRequest.responseHandlers.forEach(function(responseHandler){
		process.nextTick(function(){
			responseHandler(ws, pendingRequest.request, response);
		});
	});

	deletePendingRequest(ws, tag);
}

function cancelRequestsOnClosedConnection(ws){
	console.log("websocket closed, will complete all outstanding requests");
	for (var tag in ws.assocPendingRequests){
		var pendingRequest = ws.assocPendingRequests[tag];
		clearTimeout(pendingRequest.reroute_timer);
		clearTimeout(pendingRequest.cancel_timer);
		if (pendingRequest.reroute){ // reroute immediately, not waiting for STALLED_TIMEOUT
			if (!pendingRequest.bRerouted)
				pendingRequest.reroute();
			// we still keep ws.assocPendingRequests[tag] because we'll need it when we find a peer to reroute to
		}
		else{
			pendingRequest.responseHandlers.forEach(function(rh){
				rh(ws, pendingRequest.request, {error: "[internal] connection closed"});
			});
			delete ws.assocPendingRequests[tag];
		}
	}
	printConnectionStatus();
}



// peers

function findNextPeer(ws, handleNextPeer){
	tryFindNextPeer(ws, function(next_ws){
		if (next_ws)
			return handleNextPeer(next_ws);
		var peer = ws ? ws.peer : '[none]';
		console.log('findNextPeer after '+peer+' found no appropriate peer, will wait for a new connection');
		eventBus.once('connected_to_source', function(new_ws){
			console.log('got new connection, retrying findNextPeer after '+peer);
			findNextPeer(ws, handleNextPeer);
		});
	});
}

function tryFindNextPeer(ws, handleNextPeer){
	var arrOutboundSources = arrOutboundPeers.filter(function(outbound_ws){ return outbound_ws.bSource; });
	var len = arrOutboundSources.length;
	if (len > 0){
		var peer_index = arrOutboundSources.indexOf(ws); // -1 if it is already disconnected by now, or if it is inbound peer, or if it is null
		var next_peer_index = (peer_index === -1) ? getRandomInt(0, len-1) : ((peer_index+1)%len);
		handleNextPeer(arrOutboundSources[next_peer_index]);
	}
	else
		findRandomInboundPeer(handleNextPeer);
}

function getRandomInt(min, max) {
	return Math.floor(Math.random() * (max+1 - min)) + min;
}

function findRandomInboundPeer(handleInboundPeer){
	var arrInboundSources = [...wss.clients].filter(function(inbound_ws){ return inbound_ws.bSource; });
	if (arrInboundSources.length === 0)
		return handleInboundPeer(null);
	var arrInboundHosts = arrInboundSources.map(function(ws){ return ws.host; });
	// filter only those inbound peers that are reversible
	db.query(
		"SELECT peer_host FROM peer_host_urls JOIN peer_hosts USING(peer_host) \n\
		WHERE is_active=1 AND peer_host IN(?) \n\
			AND (count_invalid_joints/count_new_good_joints<? \n\
			OR count_new_good_joints=0 AND count_nonserial_joints=0 AND count_invalid_joints=0) \n\
		ORDER BY (count_new_good_joints=0), "+db.getRandom()+" LIMIT 1", 
		[arrInboundHosts, conf.MAX_TOLERATED_INVALID_RATIO], 
		function(rows){
			console.log(rows.length+" inbound peers");
			if (rows.length === 0)
				return handleInboundPeer(constants.bDevnet ? arrInboundSources[Math.floor(Math.random()*arrInboundSources.length)] : null);
			var host = rows[0].peer_host;
			console.log("selected inbound peer "+host);
			var ws = arrInboundSources.filter(function(ws){ return (ws.host === host); })[0];
			if (!ws)
				throw Error("inbound ws not found");
			handleInboundPeer(ws);
		}
	);
}

function checkIfHaveEnoughOutboundPeersAndAdd(){
	var arrOutboundPeerUrls = arrOutboundPeers.map(function(ws){ return ws.peer; });
	db.query(
		"SELECT peer FROM peers JOIN peer_hosts USING(peer_host) \n\
		WHERE (count_invalid_joints/count_new_good_joints<? \n\
			OR count_new_good_joints=0 AND count_nonserial_joints=0 AND count_invalid_joints=0) \n\
			AND peer IN(?)", 
		[conf.MAX_TOLERATED_INVALID_RATIO, (arrOutboundPeerUrls.length > 0) ? arrOutboundPeerUrls : null],
		function(rows){
			var count_good_peers = rows.length;
			if (count_good_peers >= conf.MIN_COUNT_GOOD_PEERS)
				return;
			if (count_good_peers === 0) // nobody trusted enough to ask for new peers, can't do anything
				return;
			var arrGoodPeerUrls = rows.map(function(row){ return row.peer; });
			for (var i=0; i<arrOutboundPeers.length; i++){
				var ws = arrOutboundPeers[i];
				if (arrGoodPeerUrls.indexOf(ws.peer) !== -1)
					requestPeers(ws);
			}
		}
	);
}

function connectToPeer(url, onOpen, dontAddPeer) {
	if (!dontAddPeer)
		addPeer(url);
	var options = { maxPayload };
	if (SocksProxyAgent && conf.socksHost && conf.socksPort) {
		let socksUrl = 'socks5h://';
		if (conf.socksUsername && conf.socksPassword)
			socksUrl += conf.socksUsername + ':' + conf.socksPassword + '@';
		socksUrl += conf.socksHost + ':' + conf.socksPort;
		console.log('Using socks proxy: ' + socksUrl);
		options.agent = new SocksProxyAgent(socksUrl);
	} else if (HttpsProxyAgent && conf.httpsProxy) {
		options.agent = new HttpsProxyAgent(conf.httpsProxy);
		console.log('Using httpsProxy: ' + conf.httpsProxy);
	}

	var ws = !bCordova ? new WebSocket(url,options) : new WebSocket(url);
	assocConnectingOutboundWebsockets[url] = ws;
	setTimeout(function(){
		if (assocConnectingOutboundWebsockets[url]){
			console.log('abandoning connection to '+url+' due to timeout');
			delete assocConnectingOutboundWebsockets[url];
			// after this, new connection attempts will be allowed to the wire, but this one can still succeed.  See the check for duplicates below.
		}
	}, 5000);
	ws.setMaxListeners(40); // avoid warning
	ws.once('open', function onWsOpen() {
		breadcrumbs.add('connected to '+url);
		delete assocConnectingOutboundWebsockets[url];
		ws.assocPendingRequests = {};
		ws.assocCommandsInPreparingResponse = {};
		if (!ws.url)
			throw Error("no url on ws");
		if (new URL(ws.url).href !== new URL(url).href) // browser implementation of Websocket might add /
			throw Error("url is different: "+ws.url);
		var another_ws_to_same_peer = getOutboundPeerWsByUrl(url);
		if (another_ws_to_same_peer){ // duplicate connection.  May happen if we abondoned a connection attempt after timeout but it still succeeded while we opened another connection
			console.log('already have a connection to '+url+', will keep the old one and close the duplicate');
			ws.close(1000, 'duplicate connection');
			if (onOpen)
				onOpen(null, another_ws_to_same_peer);
			return;
		}
		ws.peer = url;
		ws.host = getHostByPeer(ws.peer);
		ws.bOutbound = true;
		ws.last_ts = Date.now();
		console.log('connected to '+url+", host "+ws.host);
		arrOutboundPeers.push(ws);
		sendVersion(ws);
		if (conf.myUrl) // I can listen too, this is my url to connect to
			sendJustsaying(ws, 'my_url', conf.myUrl);
		if (!conf.bLight)
			subscribe(ws);
		if (onOpen)
			onOpen(null, ws);
		eventBus.emit('connected', ws);
		eventBus.emit('open-'+url);
	});
	ws.on('close', function onWsClose() {
		var i = arrOutboundPeers.indexOf(ws);
		console.log('close event, removing '+i+': '+url);
		if (i !== -1)
			arrOutboundPeers.splice(i, 1);
		cancelRequestsOnClosedConnection(ws);
		if (options.agent && options.agent.destroy)
			options.agent.destroy();
	});
	ws.on('error', function onWsError(e){
		delete assocConnectingOutboundWebsockets[url];
		console.log("error from server "+url+": "+e);
		var err = e.toString();
		// !ws.bOutbound means not connected yet. This is to distinguish connection errors from later errors that occur on open connection
		if (!ws.bOutbound && onOpen)
			onOpen(err);
		if (!ws.bOutbound)
			eventBus.emit('open-'+url, err);
	});
	ws.on('message', onWebsocketMessage);
	console.log('connectToPeer done');
}

function addOutboundPeers(multiplier){
	if (!multiplier)
		multiplier = 1;
	if (multiplier >= 32) // limit recursion
		return;
	var order_by = (multiplier <= 4) ? "count_new_good_joints DESC" : db.getRandom(); // don't stick to old peers with most accumulated good joints
	var arrOutboundPeerUrls = arrOutboundPeers.map(function(ws){ return ws.peer; });
	var arrInboundHosts = [...wss.clients].map(function(ws){ return ws.host; });
	var max_new_outbound_peers = Math.min(conf.MAX_OUTBOUND_CONNECTIONS-arrOutboundPeerUrls.length, 5); // having too many connections being opened creates odd delays in db functions
	if (max_new_outbound_peers <= 0)
		return;
	db.query(
		"SELECT peer, peers.peer_host \n\
		FROM peers \n\
		JOIN peer_hosts USING(peer_host) \n\
		LEFT JOIN peer_host_urls ON peer=url AND is_active=1 \n\
		WHERE (count_invalid_joints/count_new_good_joints<? \n\
			OR count_new_good_joints=0 AND count_nonserial_joints=0 AND count_invalid_joints=0) \n\
			"+((arrOutboundPeerUrls.length > 0) ? "AND peer NOT IN("+arrOutboundPeerUrls.map(db.escape).join(', ')+") \n" : "")+"\n\
			"+((arrInboundHosts.length > 0) ? "AND (peer_host_urls.peer_host IS NULL OR peer_host_urls.peer_host NOT IN("+arrInboundHosts.map(db.escape).join(', ')+")) \n" : "")+"\n\
			AND peer_hosts.peer_host != 'byteball.org' \n\
			AND is_self=0 \n\
		ORDER BY "+order_by+" LIMIT ?", 
		[conf.MAX_TOLERATED_INVALID_RATIO*multiplier, max_new_outbound_peers], 
		function(rows){
			for (var i=0; i<rows.length; i++){
				if (peerIsBlocked(rows[i].peer_host))
					continue;
				assocKnownPeers[rows[i].peer] = true;
				findOutboundPeerOrConnect(rows[i].peer);
			}
			if (arrOutboundPeerUrls.length === 0 && rows.length === 0) // if no outbound connections at all, get less strict
				addOutboundPeers(multiplier*2);
		}
	);
}

function getHostByPeer(peer){
	try {
		return new URL(peer).hostname;
	}
	catch (e) {
		return peer;
	}
}

function addPeerHost(host, onDone){
	db.query("INSERT "+db.getIgnore()+" INTO peer_hosts (peer_host) VALUES (?)", [host], function(){
		if (onDone)
			onDone();
	});
}

function addPeer(peer, learnt_from_peer_host){
	if (assocKnownPeers[peer])
		return;
	assocKnownPeers[peer] = true;
	var host = getHostByPeer(peer);
	addPeerHost(host, function(){
		console.log("will insert peer "+peer);
		db.query("INSERT "+db.getIgnore()+" INTO peers (peer_host, peer, learnt_from_peer_host) VALUES (?,?,?)", [host, peer, learnt_from_peer_host || null]);
	});
}

function getOutboundPeerWsByUrl(url){
	console.log("outbound peers: "+arrOutboundPeers.map(function(o){ return o.peer; }).join(", "));
	for (var i=0; i<arrOutboundPeers.length; i++)
		if (arrOutboundPeers[i].peer === url)
			return arrOutboundPeers[i];
	return null;
}

function getPeerWebSocket(peer){
	for (var i=0; i<arrOutboundPeers.length; i++)
		if (arrOutboundPeers[i].peer === peer)
			return arrOutboundPeers[i];
	for (let client of wss.clients)
		if (client.peer === peer)
			return client;
	return null;
}

function getInboundDeviceWebSocket(device_address){
	for (let client of wss.clients){
		if (client.device_address === device_address)
			return client;
	}
	return null;
}



function findOutboundPeerOrConnect(url, onOpen, dontAddPeer){
	if (!url)
		throw Error('no url');
	if (!onOpen)
		onOpen = function(){};
	if (!bStarted)
		return onOpen("[internal] network not started yet");
	url = url.toLowerCase();
	if (url.startsWith('wss://byteball.org'))
		url = url.replace(/^wss:\/\/byteball\.org/, 'wss://obyte.org');
	var ws = getOutboundPeerWsByUrl(url);
	if (ws)
		return onOpen(null, ws);
	// check if we are already connecting to the peer
	ws = assocConnectingOutboundWebsockets[url];
	if (ws){ // add second event handler
		breadcrumbs.add('already connecting to '+url);
		return eventBus.once('open-'+url, function secondOnOpen(err){
			console.log('second open '+url+", err="+err);
			if (err)
				return onOpen(err);
			if (ws.readyState === ws.OPEN)
				onOpen(null, ws);
			else{
				// can happen e.g. if the ws was abandoned but later succeeded, we opened another connection in the meantime, 
				// and had another_ws_to_same_peer on the first connection
				console.log('in second onOpen, websocket already closed');
				onOpen('[internal] websocket already closed');
			}
		});
	}
	console.log("will connect to "+url);
	connectToPeer(url, onOpen, dontAddPeer);
}

function purgePeerEvents(){
    if (conf.storage !== 'sqlite')
        return;
    console.log('will purge peer events');
    db.query("DELETE FROM peer_events WHERE event_date <= datetime('now', '-0.5 day')", function() {
        console.log("deleted some old peer_events");
    });
}

function purgeDeadPeers(){
	if (conf.storage !== 'sqlite')
		return;
	console.log('will purge dead peers');
	var arrOutboundPeerUrls = arrOutboundPeers.map(function(ws){ return ws.peer; });
	db.query("SELECT rowid, "+db.getUnixTimestamp('event_date')+" AS ts FROM peer_events ORDER BY rowid DESC LIMIT 1", function(lrows){
		if (lrows.length === 0)
			return;
		var last_rowid = lrows[0].rowid;
		var last_event_ts = lrows[0].ts;
		db.query("SELECT peer, peer_host FROM peers", function(rows){
			async.eachSeries(rows, function(row, cb){
				if (arrOutboundPeerUrls.indexOf(row.peer) >= 0)
					return cb();
				db.query(
					"SELECT MAX(rowid) AS max_rowid, MAX("+db.getUnixTimestamp('event_date')+") AS max_event_ts FROM peer_events WHERE peer_host=?", 
					[row.peer_host], 
					function(mrows){
						var max_rowid = mrows[0].max_rowid || 0;
						var max_event_ts = mrows[0].max_event_ts || 0;
						var count_other_events = last_rowid - max_rowid;
						var days_since_last_event = (last_event_ts - max_event_ts)/24/3600;
						if (count_other_events < 20000 || days_since_last_event < 7)
							return cb();
						console.log('peer '+row.peer+' is dead, will delete');
						db.query("DELETE FROM peers WHERE peer=?", [row.peer], function(){
							delete assocKnownPeers[row.peer];
							cb();
						});
					}
				);
			});
		});
	});
}

function requestPeers(ws){
	sendRequest(ws, 'get_peers', null, false, handleNewPeers);
}

async function handleNewPeers(ws, request, arrPeerUrls){
	if (!arrPeerUrls || arrPeerUrls.error)
		return console.log('get_peers failed: ', arrPeerUrls ? arrPeerUrls.error : 'no response');
	if (!Array.isArray(arrPeerUrls))
		return sendError(ws, "peer urls is not an array");
	if (arrPeerUrls.length > 100)
		return sendError(ws, "too many peer urls");
	if (!arrPeerUrls.every(ValidationUtils.isNonemptyString))
		return sendError(ws, "peer urls must be non-empty strings");
	for (let i=0; i<arrPeerUrls.length; i++){
		const url = arrPeerUrls[i];
		if (conf.myUrl && conf.myUrl.toLowerCase() === url.toLowerCase())
			continue;
		if (!isValidWsUrl(url)) {
			console.log('ignoring new peer '+url+' because of incompatible ws protocol');
			continue;
		}
		const host = getHostByPeer(url);
		if (host === 'byteball.org')
			continue;
		const peer_ws = getPeerWebSocket(url);
		if (peer_ws) {
			console.log(`already connected to peer ${url} shared by ${ws.host}`);
			continue;
		}
		const [row] = await db.query("SELECT 1 FROM peers WHERE peer=?", [url]);
		if (row) {
			console.log(`peer ${url} shared by ${ws.host} is already known`);
			continue;
		}
		if (arrOutboundPeers.length >= conf.MAX_OUTBOUND_CONNECTIONS)
			return console.log(`will not connect to peer #${i} ${url} shared by ${ws.host} because max outbound connections already exhausted, will skip the remaining peers too`);
		console.log(`will try to connect to peer ${url} shared by ${ws.host}`);
		addPeerHost(host);
		connectToPeer(url, err => {
			if (!err) // add only if successfully connected
				addPeer(url, ws.host);
		}, true);
	}
}

function isValidWsUrl(url) {
	try {
		const { protocol, hash, username, password } = new URL(url);
		if (hash)
			return false;
		if (url.includes('#')) // empty trailing hash
			return false;
		if (username || password) // no legitimate peer url needs credentials
			return false;
		if (!['wss:', 'ws:'].includes(protocol))
			return false;
		if (conf.WS_PROTOCOL === 'wss://' && protocol !== 'wss:')
			return false;
		return true;
	}
	catch {
		return false;
	}
}

function heartbeat(){
	// just resumed after sleeping
	var bJustResumed = (bCordova && Date.now() - last_hearbeat_wake_ts > HEARTBEAT_PAUSE_TIMEOUT);
	last_hearbeat_wake_ts = Date.now();
	[...wss.clients].concat(arrOutboundPeers).forEach(function(ws){
		if (ws.bSleeping || ws.readyState !== ws.OPEN)
			return;
		var elapsed_since_last_received = Date.now() - ws.last_ts;
		if (elapsed_since_last_received < HEARTBEAT_TIMEOUT)
			return;
		if (!ws.last_sent_heartbeat_ts || bJustResumed){
			ws.last_sent_heartbeat_ts = Date.now();
			return sendRequest(ws, 'heartbeat', null, false, handleHeartbeatResponse);
		}
		var elapsed_since_last_sent_heartbeat = Date.now() - ws.last_sent_heartbeat_ts;
		if (elapsed_since_last_sent_heartbeat < HEARTBEAT_RESPONSE_TIMEOUT)
			return;
		console.log('will disconnect peer '+ws.peer+' who was silent for '+elapsed_since_last_received+'ms');
		ws.close(1000, "lost connection");
	});
}

function handleHeartbeatResponse(ws, request, response){
	delete ws.last_sent_heartbeat_ts;
	if (response === 'sleep') // the peer doesn't want to be bothered with heartbeats any more, but still wants to keep the connection open
		ws.bSleeping = true;
	// as soon as the peer sends a heartbeat himself, we'll think he's woken up and resume our heartbeats too
}

function requestFromLightVendor(command, params, responseHandler){
	if (!responseHandler)
		return new Promise((resolve, reject) => requestFromLightVendor(command, params, (ws, request, response) => response && response.error ? reject(response.error) : resolve(response)));
	if (!exports.light_vendor_url){
		console.log("light_vendor_url not set yet");
		return setTimeout(function(){
			requestFromLightVendor(command, params, responseHandler);
		}, 1000);
	}
	findOutboundPeerOrConnect(exports.light_vendor_url, function(err, ws){
		if (err)
			return responseHandler(null, null, {error: "[connect to light vendor failed]: "+err});
		sendRequest(ws, command, params, false, responseHandler);
	});
}


function getConnectionStatus(){
	return {
		incoming: wss ? wss.clients.size : 0,
		outgoing: arrOutboundPeers.length,
		outgoing_being_opened: Object.keys(assocConnectingOutboundWebsockets).length
	}
}

function printConnectionStatus(){
	var objConnectionStatus = getConnectionStatus();
	console.log(objConnectionStatus.incoming+" incoming connections, "+objConnectionStatus.outgoing+" outgoing connections, "+
	objConnectionStatus.outgoing_being_opened+" outgoing connections being opened");
}

function subscribe(ws){
	ws.subscription_id = crypto.randomBytes(30).toString("base64"); // this is to detect self-connect
	storage.readLastMainChainIndex(function(last_mci){
		sendRequest(ws, 'subscribe', {subscription_id: ws.subscription_id, last_mci: last_mci, library_version: libraryPackageJson.version}, false, function(ws, request, response){
			delete ws.subscription_id;
			if (response && response.error)
				return;
			ws.bSource = true;
			eventBus.emit('connected_to_source', ws);
		});
	});
}

// joints

// sent as justsaying or as response to a request
function sendJoint(ws, objJoint, tag) {
	console.log('sending joint identified by unit ' + objJoint.unit.unit + ' to', ws.peer);
	tag ? sendResponse(ws, tag, {joint: objJoint}) : sendJustsaying(ws, 'joint', objJoint);
}

// sent by light clients to their vendors
function postJointToLightVendor(objJoint, handleResponse) {
	console.log('posting joint identified by unit ' + objJoint.unit.unit + ' to light vendor');
	requestFromLightVendor('post_joint', objJoint, function(ws, request, response){
		handleResponse(response);
	});
}

function sendFreeJoints(ws) {
	storage.readFreeJoints(function(objJoint){
		sendJoint(ws, objJoint);
	}, function(){
		sendJustsaying(ws, 'free_joints_end', null);
	});
}

function sendJointsSinceMci(ws, mci) {
	joint_storage.readJointsSinceMci(
		mci, 
		function(objJoint){
			sendJoint(ws, objJoint);
		},
		function(){
			sendJustsaying(ws, 'free_joints_end', null);
		}
	);
}

function requestFreeJointsFromAllOutboundPeers(){
	for (var i=0; i<arrOutboundPeers.length; i++)
		sendJustsaying(arrOutboundPeers[i], 'refresh', null);
}

function requestNewJoints(ws){
	storage.readLastMainChainIndex(function(last_mci){
		sendJustsaying(ws, 'refresh', last_mci);
	});
}

function rerequestLostJoints(bForce){
	//console.log("rerequestLostJoints");
	if (bCatchingUp && !bForce)
		return;
	joint_storage.findLostJoints(function(arrUnits){
		console.log("lost units", arrUnits.length > 0 ? arrUnits : 'none');
		tryFindNextPeer(null, function(ws){
			if (!ws)
				return;
			console.log("found next peer "+ws.peer);
			requestJoints(ws, arrUnits.filter(function(unit){ return (!assocUnitsInWork[unit] && !havePendingJointRequest(unit)); }));
		});
	});
}

function requestNewMissingJoints(ws, arrUnits){
	var arrNewUnits = [];
	async.eachSeries(
		arrUnits,
		function(unit, cb){
			if (assocUnitsInWork[unit])
				return cb();
			if (havePendingJointRequest(unit)){
				console.log("unit "+unit+" was already requested");
				return cb();
			}
			joint_storage.checkIfNewUnit(unit, {
				ifNew: function(){
					arrNewUnits.push(unit);
					cb();
				},
				ifKnown: function(){console.log("known"); cb();}, // it has just been handled
				ifKnownUnverified: function(){console.log("known unverified"); cb();}, // I was already waiting for it
				ifKnownBad: function(error){
					console.log("known bad "+unit+": "+error); // might happen if it's immediately used as parent but not written to known_bad_joints yet
					cb();
				}
			});
		},
		function(){
			//console.log(arrNewUnits.length+" of "+arrUnits.length+" left", assocUnitsInWork);
			// filter again as something could have changed each time we were paused in checkIfNewUnit
			arrNewUnits = arrNewUnits.filter(function(unit){ return (!assocUnitsInWork[unit] && !havePendingJointRequest(unit)); });
			if (arrNewUnits.length > 0)
				requestJoints(ws, arrNewUnits);
		}
	);
}

function requestJoints(ws, arrUnits) {
	if (arrUnits.length === 0)
		return;
	arrUnits.forEach(function(unit){
		if (assocRequestedUnits[unit]){
			var diff = Date.now() - assocRequestedUnits[unit];
			// since response handlers are called in nextTick(), there is a period when the pending request is already cleared but the response
			// handler is not yet called, hence assocRequestedUnits[unit] not yet cleared
			if (diff <= STALLED_TIMEOUT)
				return console.log("unit "+unit+" already requested "+diff+" ms ago, assocUnitsInWork="+assocUnitsInWork[unit]);
			//	throw new Error("unit "+unit+" already requested "+diff+" ms ago, assocUnitsInWork="+assocUnitsInWork[unit]);
		}
		if (ws.readyState === ws.OPEN)
			assocRequestedUnits[unit] = Date.now();
		// even if readyState is not ws.OPEN, we still send the request, it'll be rerouted after timeout
		sendRequest(ws, 'get_joint', unit, true, handleResponseToJointRequest);
	});
}

function handleResponseToJointRequest(ws, request, response){
	delete assocRequestedUnits[request.params];
	if (!response)
		return console.log("empty response to get_joint " + request.params);
	if (!response.joint){
		var unit = request.params;
		if (response.joint_not_found === unit){
			if (conf.bLight) // we trust the light vendor that if it doesn't know about the unit after 1 day, it doesn't exist
				db.query("DELETE FROM unhandled_private_payments WHERE unit=? AND creation_date<"+db.addTime('-1 DAY'), [unit]);
			if (!bCatchingUp)
				return console.log("unit "+unit+" does not exist"); // if it is in unhandled_joints, it'll be deleted in 1 hour
			//	return purgeDependenciesAndNotifyPeers(unit, "unit "+unit+" does not exist");
			db.query("SELECT 1 FROM hash_tree_balls WHERE unit=?", [unit], function(rows){
				if (rows.length === 0)
					return console.log("unit "+unit+" does not exist (catching up)");
				//	return purgeDependenciesAndNotifyPeers(unit, "unit "+unit+" does not exist (catching up)");
				findNextPeer(ws, function(next_ws){
					breadcrumbs.add("found next peer to reroute joint_not_found "+unit+": "+next_ws.peer);
					requestJoints(next_ws, [unit]);
				});
			});
		}
		// if it still exists, we'll request it again
		// we requst joints in two cases:
		// - when referenced from parents, in this case we request it from the same peer who sent us the referencing joint, 
		//   he should know, or he is attempting to DoS us
		// - when catching up and requesting old joints from random peers, in this case we are pretty sure it should exist
		return;
	}
	var objJoint = response.joint;
	if (!objJoint.unit || !objJoint.unit.unit || typeof objJoint.unit.unit !== 'string')
		return sendError(ws, 'no unit');
	if ("unsigned" in objJoint)
		return sendError(ws, 'unsigned unit');
	var unit = objJoint.unit.unit;
	if (request.params !== unit)
		return sendError(ws, "I didn't request this unit from you: " + unit);
	if (conf.bLight && objJoint.ball && !objJoint.unit.content_hash){
		// accept it as unfinished (otherwise we would have to require a proof)
		delete objJoint.ball;
		delete objJoint.skiplist_units;
	}
	if (isTooDeeplyNestedOrHasTooManyNodes(objJoint))
		return sendError(ws, "joint is too deeply nested or has too many nodes");
	if (conf.bLight)
		return handleLightOnlineJoint(ws, objJoint);
	if (objJoint.unit.content_hash || !objJoint.unit.messages)
		return handleOnlineJoint(ws, objJoint);
	// full content of a unit that might be a locally pruned stable final-bad unit we are trying to recover; if so, handle it
	// without going through the normal new-unit pipeline (it is already fully judged and saved, just its content was stripped)
	if (!tryRecoverPrunedContentOfFinalBadUnit(ws, objJoint))
		handleOnlineJoint(ws, objJoint);
}

// nobody here is waiting for this unit's content unless it's in assocUnitsWaitingForPrunedContent, so that map alone
// (populated only from the validated pruned-final-bad branch in validation.js) tells us whether recovery is relevant
function tryRecoverPrunedContentOfFinalBadUnit(ws, objJoint){
	const unit = objJoint.unit.unit;
	const arrWaitingJoints = assocUnitsWaitingForPrunedContent[unit];
	if (!arrWaitingJoints)
		return false;
	if (!validation.allHashesAreValid(objJoint)){
		console.log('recovered content for pruned unit ' + unit + ' from ' + ws.peer + ' failed hash check');
		return false;
	}
	archiving.cachePrunedJoint(objJoint);
	breadcrumbs.add('recovered pruned content of final-bad unit ' + unit + ' from ' + ws.peer);
	delete assocUnitsWaitingForPrunedContent[unit];
	// replay the waiting joints in the background, don't make the peer who sent us the recovered content wait for this
	async.eachSeries(arrWaitingJoints, function(o, cb){
		const key = 'wj\n' + o.unit;
		kvstore.get(key, strSavedJoint => {
			kvstore.del(key);
			if (!strSavedJoint) // already replayed via another missing unit it was also waiting on, or expired
				return cb();
			const { objJoint: objWaitingJoint, peer } = JSON.parse(strSavedJoint);
			handleOnlineJoint(peer ? getPeerWebSocket(peer) : null, objWaitingJoint, cb);
		});
	});
	return true;
}

function purgeExpiredUnitsWaitingForPrunedContent(){
	const now = Date.now();
	for (let missing_unit in assocUnitsWaitingForPrunedContent){
		const arrExpired = assocUnitsWaitingForPrunedContent[missing_unit].filter(o => o.expiry_ts < now);
		arrExpired.forEach(o => kvstore.del('wj\n' + o.unit));
		assocUnitsWaitingForPrunedContent[missing_unit] = assocUnitsWaitingForPrunedContent[missing_unit].filter(o => o.expiry_ts >= now);
		if (assocUnitsWaitingForPrunedContent[missing_unit].length === 0)
			delete assocUnitsWaitingForPrunedContent[missing_unit];
	}
}
setInterval(purgeExpiredUnitsWaitingForPrunedContent, 60 * 60 * 1000);

// assocUnitsWaitingForPrunedContent always starts empty on restart, so any 'wj\n' pointers left in rocksdb
// from a previous run would otherwise never be replayed or purged again
function clearStalePrunedContentPointers(){
	return new Promise(function(resolve, reject){
		const batch = kvstore.batch();
		let count = 0;
		kvstore.createKeyStream({ gte: 'wj\n', lte: 'wj\n\uFFFF' })
			.on('data', key => {
				batch.del(key);
				count++;
			})
			.on('end', () => {
				if (count === 0)
					return resolve();
				batch.write(err => {
					if (err)
						return reject(Error("failed to clear stale pruned content pointers: " + err));
					console.log(`cleared ${count} stale pruned content pointer(s) left over from the previous run`);
					resolve();
				});
			})
			.on('error', reject);
	});
}

function havePendingRequest(command){
	var arrPeers = [...wss.clients].concat(arrOutboundPeers);
	for (var i=0; i<arrPeers.length; i++){
		var assocPendingRequests = arrPeers[i].assocPendingRequests;
		for (var tag in assocPendingRequests)
			if (assocPendingRequests[tag].request.command === command)
				return true;
	}
	return false;
}

function havePendingJointRequest(unit){
	var arrPeers = [...wss.clients].concat(arrOutboundPeers);
	for (var i=0; i<arrPeers.length; i++){
		var assocPendingRequests = arrPeers[i].assocPendingRequests;
		for (var tag in assocPendingRequests){
			var request = assocPendingRequests[tag].request;
			if (request.command === 'get_joint' && request.params === unit)
				return true;
		}
	}
	return false;
}

// We may receive a reference to a nonexisting unit in parents. We are not going to keep the referencing joint forever.
function purgeJunkUnhandledJoints(){
	if (bCatchingUp || Date.now() - coming_online_time < 3600*1000 || wss.clients.size === 0 && arrOutboundPeers.length === 0)
		return;
	joint_storage.purgeOldUnhandledJoints();
}

function purgeJointAndDependenciesAndNotifyPeers(objJoint, error, onDone){
	if (error.indexOf('is not stable in view of your parents') >= 0){ // give it a chance to be retried after adding other units
		eventBus.emit('nonfatal_error', "error on unit "+objJoint.unit.unit+": "+error+"; "+JSON.stringify(objJoint), new Error());
		// schedule a retry
		console.log("will schedule a retry of " + objJoint.unit.unit);
		setTimeout(function () {
			console.log("retrying " + objJoint.unit.unit);
			rerequestLostJoints(true);
			joint_storage.readDependentJointsThatAreReady(null, handleSavedJoint);
		}, 60 * 1000);
		return onDone();
	}
	joint_storage.purgeJointAndDependencies(
		objJoint, 
		error, 
		// this callback is called for each dependent unit
		function(purged_unit, peer){
			var ws = getPeerWebSocket(peer);
			if (ws)
				sendErrorResult(ws, purged_unit, "error on (indirect) parent unit "+objJoint.unit.unit+": "+error);
		}, 
		onDone
	);
}

function purgeDependenciesAndNotifyPeers(unit, error, onDone){
	joint_storage.purgeDependencies(
		unit, 
		error, 
		// this callback is called for each dependent unit
		function(purged_unit, peer){
			var ws = getPeerWebSocket(peer);
			if (ws)
				sendErrorResult(ws, purged_unit, "error on (indirect) parent unit "+unit+": "+error);
		}, 
		onDone
	);
}

function forwardJoint(ws, objJoint){
	[...wss.clients].concat(arrOutboundPeers).forEach(function(client) {
		if (client != ws && client.bSubscribed)
			sendJoint(client, objJoint);
	});
}

function handleJoint(ws, objJoint, bSaved, bPosted, callbacks){
	if ('aa' in objJoint)
		return callbacks.ifJointError("AA unit cannot be broadcast");
	var unit = objJoint.unit.unit;
	if (typeof unit !== 'string')
		return callbacks.ifJointError("invalid unit");
	const version = objJoint.unit.version;
	if (typeof version !== 'string')
		return callbacks.ifJointError("invalid version");
	const fVersion = parseFloat(version);
	if (!(fVersion >= constants.fVersion4 || objJoint.ball)) // covers NaN too
		return callbacks.ifTransientError("version is too old");
	if (assocUnitsInWork[unit])
		return callbacks.ifUnitInWork();
	assocUnitsInWork[unit] = true;
	
	var validate = function(){
		mutex.lock(['handleJoint'], function(unlock){
			if (ws && !conf.bLight)
				currentJointHost = ws.host;
			// clear host only if validation completed with any result, otherwise it crashed and we keep it for a while to avoid DoS from the same peer
			const clearHost = () => {
				if (ws && !conf.bLight)
					currentJointHost = null;
			};
			validation.validate(objJoint, {
				ifUnitError: function(error){
					console.log(objJoint.unit.unit+" validation failed: "+error);
					clearHost();
					callbacks.ifUnitError(error);
					if (constants.bDevnet)
						throw Error(error);
					purgeJointAndDependenciesAndNotifyPeers(objJoint, error, function(){
						delete assocUnitsInWork[unit];
					});
					unlock();
					if (ws && error !== 'authentifier verification failed' && !error.match(/bad merkle proof at path/) && !bPosted)
						writeEvent('invalid', ws.host);
					if (objJoint.unsigned)
						eventBus.emit("validated-"+unit, false);
				},
				ifJointError: function(error){
					clearHost();
					callbacks.ifJointError(error);
				//	throw Error(error);
					joint_storage.saveKnownBadJoint(objJoint, error, function(){
						delete assocUnitsInWork[unit];
					});
					unlock();
					if (ws)
						writeEvent('invalid', ws.host);
					if (objJoint.unsigned)
						eventBus.emit("validated-"+unit, false);
				},
				ifTransientError: function(error){
				//	throw Error(error);
					console.log("############################## transient error "+error);
					clearHost();
					callbacks.ifTransientError ? callbacks.ifTransientError(error) : callbacks.ifUnitError(error);
					process.nextTick(unlock);
					joint_storage.removeUnhandledJointAndDependencies(unit, function(){
					//	if (objJoint.ball)
					//		db.query("DELETE FROM hash_tree_balls WHERE ball=? AND unit=?", [objJoint.ball, objJoint.unit.unit]);
						delete assocUnitsInWork[unit];
					});
					if (error.includes("last ball just advanced"))
						setTimeout(rerequestLostJoints, 10 * 1000, true);
					if (error === "possible AA" && bCatchingUp)
						tryToAdvanceStabilityPointForCatchupAATrigger(objJoint);
				},
				ifNeedHashTree: function(){
					console.log('need hash tree for unit '+unit);
					clearHost();
					if (objJoint.unsigned)
						throw Error("ifNeedHashTree() unsigned");
					callbacks.ifNeedHashTree();
					// we are not saving unhandled joint because we don't know dependencies
					delete assocUnitsInWork[unit];
					unlock();
				},
				ifNeedParentUnits: function(arrMissingUnits, bRequestPrunedContent){
					clearHost();
					if (bRequestPrunedContent && !bPosted) {
						// e.g. spending a locally pruned final-bad output: hold on to this joint ourselves and retry it if/when we
						// recover the missing content, since the missing unit already has a row in the units table and the usual
						// dependencies table would (wrongly) consider it satisfied right away.
						// the joint body goes to rocksdb (see comment on assocUnitsWaitingForPrunedContent), keyed by its own unit
						kvstore.put('wj\n' + unit, JSON.stringify({ objJoint, peer: ws ? ws.peer : null }), () => {});
						const expiry_ts = Date.now() + 60 * 60 * 1000;
						for (let missing_unit of arrMissingUnits) {
							if (!assocUnitsWaitingForPrunedContent[missing_unit])
								assocUnitsWaitingForPrunedContent[missing_unit] = [];
							assocUnitsWaitingForPrunedContent[missing_unit].push({ unit, expiry_ts });
						}
						delete assocUnitsInWork[unit];
						// requestNewMissingJoints relies on checkIfNewUnit, which can say ifKnown() for a unit that only has
						// stale props cached in assocCachedUnits, even though its full content is exactly what we are missing
						// Also, don't check havePendingJointRequest as the peer of an earlier request could intentionally withhold the unit (there is rerouting, but it would give up at the first peer that doesn't have the unit)
						// ws can be null here (e.g. replaying a saved joint whose original sender already disconnected)
						if (ws)
							requestJoints(ws, arrMissingUnits);
						else
							findNextPeer(null, function(next_ws){
								requestJoints(next_ws, arrMissingUnits);
							});
					}
					callbacks.ifNeedParentUnits(arrMissingUnits, bRequestPrunedContent);
					unlock();
				},
				ifOk: async function(objValidationState, validation_unlock){
					clearHost();
					if (objJoint.unsigned)
						throw Error("ifOk() unsigned");
					if (bPosted && objValidationState.sequence !== 'good') {
						validation_unlock();
						callbacks.ifUnitError("The transaction would be non-serial (a double spend)");
						delete assocUnitsInWork[unit];
						unlock();
						if (ws)
							writeEvent('nonserial', ws.host);
						return;
					}
					if (conf.bDryRunNewTriggers && !conf.bLight && !objJoint.ball && objValidationState.count_primary_aa_triggers) {
						const outputAddresses = objJoint.unit.messages
							.filter(msg => msg.app === 'payment')
							.reduce((acc, msg) => acc.concat(msg.payload.outputs.map(output => output.address)), []);
						const rows = await db.query("SELECT address, definition FROM aa_addresses WHERE address IN (?)", [outputAddresses]);
						for (let { address, definition } of rows) {
							console.log(`dry run trigger for AA address ${address} in submitted unit ${unit}`);
							const trigger = aa_composer.getTrigger(objJoint.unit, address);
							// if it would crash, let it crash now, not when we execute the trigger for real
							await aa_composer.dryRunPrimaryAATrigger(trigger, address, JSON.parse(definition));
						}
					}
					writer.saveJoint(objJoint, objValidationState, null, function(){
						validation_unlock();
						callbacks.ifOk();
						unlock();
						if (ws)
							writeEvent((objValidationState.sequence !== 'good') ? 'nonserial' : 'new_good', ws.host);
						notifyWatchers(objJoint, objValidationState.sequence === 'good', ws);
						if (objValidationState.arrUnitsGettingBadSequence)
							notifyWatchersAboutUnitsGettingBadSequence(objValidationState.arrUnitsGettingBadSequence);
						if (!bCatchingUp)
							eventBus.emit('new_joint', objJoint);
					});
				},
				ifOkUnsigned: function(bSerial){
					clearHost();
					if (!objJoint.unsigned)
						throw Error("ifOkUnsigned() signed");
					callbacks.ifOkUnsigned();
					unlock();
					eventBus.emit("validated-"+unit, bSerial);
				}
			});
		});
	};

	joint_storage.checkIfNewJoint(objJoint, {
		ifNew: function(){
			bSaved ? callbacks.ifNew() : validate();
		},
		ifKnown: function(){
			callbacks.ifKnown();
			delete assocUnitsInWork[unit];
		},
		ifKnownBad: function(){
			callbacks.ifKnownBad();
			delete assocUnitsInWork[unit];
		},
		ifKnownUnverified: function(){
			bSaved ? validate() : callbacks.ifKnownUnverified();
		}
	});
}

// we rejected a unit as a "possible AA" while catching up because we haven't executed its predecessor's AA triggers yet.
// try to advance the stability point to an unstable on-MC parent, which might stabilize the actual AA trigger and let the retransmitted unit through
async function tryToAdvanceStabilityPointForCatchupAATrigger(objJoint){
	const objUnit = objJoint.unit;
	const unit = objUnit.unit;
	const ball = objJoint.ball;
	if (!ball || storage.assocHashTreeUnitsByBall[ball] !== unit)
		return;
	// the parent might not be known to us yet, hence not in assocUnstableUnits
	const parent_unit = objUnit.parent_units.find(parent_unit => {
		const props = storage.assocUnstableUnits[parent_unit];
		return props && props.is_on_main_chain && !props.is_stable;
	});
	if (!parent_unit)
		return;
	const arrFreeUnits = main_chain.getFreeUnits();
	console.log(`possible AA trigger for unit ${unit}: trying to advance the stability point to its MC parent ${parent_unit} using free units ${arrFreeUnits.join(', ')}`);
	// use a dedicated connection so that interleaving writes from other tasks don't interfere with this check
	const conn = await db.takeConnectionFromPool();
	await conn.query("BEGIN");
	const bStable = await main_chain.determineIfStableInLaterUnitsAndUpdateStableMcFlag(conn, parent_unit, arrFreeUnits, false);
	console.log(`AA parent ${parent_unit} stable in free units ${arrFreeUnits.join(', ')}? ${bStable}`);
	await conn.query("COMMIT");
	conn.release();
}

// handle joint posted to me by a light client
function handlePostedJoint(ws, objJoint, onDone){
	
	if (!objJoint || !objJoint.unit || !objJoint.unit.unit || typeof objJoint.unit.unit !== 'string')
		return onDone('no unit');
	if ("unsigned" in objJoint)
		return onDone('unsigned unit');
	
	var unit = objJoint.unit.unit;
	delete objJoint.unit.main_chain_index;
	delete objJoint.unit.actual_tps_fee;
	objJoint = sortObject(objJoint);
	
	handleJoint(ws, objJoint, false, true, {
		ifUnitInWork: function(){
			onDone("already handling this unit");
		},
		ifUnitError: function(error){
			onDone(error);
		},
		ifTransientError: function(error){
			onDone(error);
		},
		ifJointError: function(error){
			onDone(error);
		},
		ifNeedHashTree: function(){
			onDone("need hash tree");
		},
		ifNeedParentUnits: function(arrMissingUnits){
			delete assocUnitsInWork[unit];
			onDone("unknown parents");
		},
		ifOk: function(){
			onDone();
			
			// forward to other peers
			if (!bCatchingUp && !conf.bLight)
				forwardJoint(ws, objJoint);

			delete assocUnitsInWork[unit];
		},
		ifOkUnsigned: function(){
			delete assocUnitsInWork[unit];
			onDone("you can't send unsigned units");
		},
		ifKnown: function(){
			if (objJoint.unsigned)
				return onDone("you can't send unsigned units");
			onDone("known");
			writeEvent('known_good', ws.host);
		},
		ifKnownBad: function(){
			onDone("known bad");
			writeEvent('known_bad', ws.host);
		},
		ifKnownUnverified: function(){ // impossible unless the peer also sends this joint by 'joint' justsaying
			onDone("known unverified");
			delete assocUnitsInWork[unit];
		}
	});
}

function handleOnlineJoint(ws, objJoint, onDone){
	if (!onDone)
		onDone = function(){};
	var unit = objJoint.unit.unit;
	delete objJoint.unit.main_chain_index;
	delete objJoint.unit.actual_tps_fee;
	if (bCatchingUp && !objJoint.ball) {
		console.log("catching up, will not handle online joint " + unit + " without a ball");
		return onDone();
	}
	objJoint = sortObject(objJoint);
	
	handleJoint(ws, objJoint, false, false, {
		ifUnitInWork: onDone,
		ifUnitError: function(error){
			sendErrorResult(ws, unit, error);
			onDone();
		},
		ifTransientError: function(error) {
			sendErrorResult(ws, unit, error);
			onDone();
			if (error.includes("tps fee"))
				setTimeout(handleOnlineJoint, 10 * 1000, ws, objJoint);
		},
		ifJointError: function(error){
			sendErrorResult(ws, unit, error);
			onDone();
		},
		ifNeedHashTree: function(){
			if (!bCatchingUp && !bWaitingForCatchupChain)
				requestCatchup(ws);
			// we are not saving the joint so that in case requestCatchup() fails, the joint will be requested again via findLostJoints, 
			// which will trigger another attempt to request catchup
			onDone();
		},
		ifNeedParentUnits: function(arrMissingUnits, bRequestPrunedContent){
			sendInfo(ws, {unit: unit, info: "unresolved dependencies: "+arrMissingUnits.join(", ")});
			if (!bRequestPrunedContent){
				joint_storage.saveUnhandledJointAndDependencies(objJoint, arrMissingUnits, ws.peer, function(){
					delete assocUnitsInWork[unit];
				});
				requestNewMissingJoints(ws, arrMissingUnits);
			}
			onDone();
		},
		ifOk: function(){
			sendResult(ws, {unit: unit, result: 'accepted'});
			
			// forward to other peers
			if (!bCatchingUp && !conf.bLight)
				forwardJoint(ws, objJoint);

			delete assocUnitsInWork[unit];

			// wake up other joints that depend on me
			findAndHandleJointsThatAreReady(unit);
			onDone();
		},
		ifOkUnsigned: function(){
			delete assocUnitsInWork[unit];
			onDone();
		},
		ifKnown: function(){
			if (objJoint.unsigned)
				return onDone();
			sendResult(ws, {unit: unit, result: 'known'});
			if (ws)
				writeEvent('known_good', ws.host);
			onDone();
		},
		ifKnownBad: function(){
			sendResult(ws, {unit: unit, result: 'known_bad'});
			if (ws)
				writeEvent('known_bad', ws.host);
			if (objJoint.unsigned)
				eventBus.emit("validated-"+unit, false);
			onDone();
		},
		ifKnownUnverified: function(){
			sendResult(ws, {unit: unit, result: 'known_unverified'});
			delete assocUnitsInWork[unit];
			onDone();
		}
	});
}


function handleSavedJoint(objJoint, creation_ts, peer){
	
	var unit = objJoint.unit.unit;
	var ws = getPeerWebSocket(peer);
	if (ws && ws.readyState !== ws.OPEN)
		ws = null;

	handleJoint(ws, objJoint, true, false, {
		ifUnitInWork: function(){
			setTimeout(function(){
				handleSavedJoint(objJoint, creation_ts, peer);
			}, 1000);
		},
		ifUnitError: function(error){
			if (ws)
				sendErrorResult(ws, unit, error);
		},
		ifTransientError: function(error){
			if (ws)
				sendErrorResult(ws, unit, error);
		},
		ifJointError: function(error){
			if (ws)
				sendErrorResult(ws, unit, error);
		},
		ifNeedHashTree: function(){
			console.log("handleSavedJoint "+objJoint.unit.unit+": need hash tree, will retry later");
			setTimeout(function(){
				handleSavedJoint(objJoint, creation_ts, peer);
			}, 1000);
		//	throw Error("handleSavedJoint "+objJoint.unit.unit+": need hash tree");
		},
		ifNeedParentUnits: function(arrMissingUnits, bRequestPrunedContent){
			if (bRequestPrunedContent) return;
			db.query("SELECT 1 FROM archived_joints WHERE unit IN(?) LIMIT 1", [arrMissingUnits], function(rows){
				if (rows.length === 0)
					throw Error("unit "+unit+" still has unresolved dependencies: "+arrMissingUnits.join(", "));
				breadcrumbs.add("unit "+unit+" has unresolved dependencies that were archived: "+arrMissingUnits.join(", "))
				if (ws)
					requestNewMissingJoints(ws, arrMissingUnits);
				else
					findNextPeer(null, function(next_ws){
						requestNewMissingJoints(next_ws, arrMissingUnits);
					});
				delete assocUnitsInWork[unit];
			});
		},
		ifOk: function(){
			if (ws)
				sendResult(ws, {unit: unit, result: 'accepted'});
			
			// forward to other peers
			if (!bCatchingUp && !conf.bLight && creation_ts > Date.now() - FORWARDING_TIMEOUT)
				forwardJoint(ws, objJoint);

			joint_storage.removeUnhandledJointAndDependencies(unit, function(){
				delete assocUnitsInWork[unit];
				// wake up other saved joints that depend on me
				findAndHandleJointsThatAreReady(unit);
			});
		},
		ifOkUnsigned: function(){
			joint_storage.removeUnhandledJointAndDependencies(unit, function(){
				delete assocUnitsInWork[unit];
			});
		},
		// readDependentJointsThatAreReady can read the same joint twice before it's handled. If not new, just ignore (we've already responded to peer).
		ifKnown: function(){},
		ifKnownBad: function(){},
		ifNew: function(){
			// that's ok: may be simultaneously selected by readDependentJointsThatAreReady and deleted by purgeJunkUnhandledJoints when we wake up after sleep
			delete assocUnitsInWork[unit];
			console.log("new in handleSavedJoint: "+unit);
		//	throw Error("new in handleSavedJoint: "+unit);
		}
	});
}

function handleLightOnlineJoint(ws, objJoint){
	// the lock ensures that we do not overlap with history processing which might also write new joints
	mutex.lock(["light_joints"], function(unlock){
		breadcrumbs.add('got light_joints for handleLightOnlineJoint '+objJoint.unit.unit);
		handleOnlineJoint(ws, objJoint, function(){
			breadcrumbs.add('handleLightOnlineJoint done');
			unlock();
		});
	});
}

function setWatchedAddresses(_arrWatchedAddresses){
	arrWatchedAddresses = _arrWatchedAddresses;
}

function addWatchedAddress(address){
	arrWatchedAddresses.push(address);
}

function notifyWatchersAboutUnitsGettingBadSequence(arrUnits){
	if (conf.bLight)
		throw Error("light node cannot notify about bad sequence");

	// - one unit can concern several addresses
	// - same address can be concerned by several units
	// - several addresses can be watched locally or by same peer
	// we have to sort that in order to provide duplicated units to sequence_became_bad event
	arrUnits = _.uniq(arrUnits);
	var assocAddressesByUnit = {};
	var assocUnitsByAddress = {};
	async.each(arrUnits, function(unit, cb){
		storage.readJoint(db, unit, {
			ifFound: function(objJoint) {
				var objAddresses = getAllAuthorsAndOutputAddresses(objJoint.unit);
				if (!objAddresses) // voided unit
					return cb();
				var arrAddresses = objAddresses.addresses;
				assocAddressesByUnit[unit] = arrAddresses;
				arrAddresses.forEach(function(address){
					if (!assocUnitsByAddress[address])
						assocUnitsByAddress[address] = [];
					assocUnitsByAddress[address].push(unit);
				});
				cb();
			},
			ifNotFound: function(){
				cb();
			}
		});
	},
	function () {	
		// notify local watchers
		var assocUniqueUnits = {};
		for (var unit in assocAddressesByUnit){
			if (_.intersection(arrWatchedAddresses, assocAddressesByUnit[unit]).length > 0){
				assocUniqueUnits[unit] = true;
			}
		}
		var arrConcernedAddresses = Object.keys(assocUnitsByAddress);
		db.query(
			"SELECT address FROM my_addresses WHERE address IN(?) UNION SELECT shared_address AS address FROM shared_addresses WHERE address IN(?) UNION SELECT address FROM my_watched_addresses WHERE address IN(?)",  
			[arrConcernedAddresses, arrConcernedAddresses, arrConcernedAddresses],
			function(rows){
				if (rows.length > 0){
					var arrMyConcernedAddresses = rows.map(function(row){return row.address});
					for (var unit in assocAddressesByUnit){
						if (_.intersection(arrMyConcernedAddresses, assocAddressesByUnit[unit]).length > 0)
							assocUniqueUnits[unit] = true;
					}
				}
				var arrUniqueUnits = Object.keys(assocUniqueUnits);
				if (arrUniqueUnits.length > 0)
					eventBus.emit("sequence_became_bad", arrUniqueUnits);
			}
		);
		// notify light watchers
		if (!bWatchingForLight)
			return;
		db.query("SELECT peer, address, NULL AS unit FROM watched_light_addresses WHERE address IN(?) UNION SELECT peer, NULL AS address, unit FROM watched_light_units WHERE unit IN(?)", [arrConcernedAddresses, arrUnits], function(rows){
			var assocUniqueUnitsByPeer = {};
			rows.forEach(function(row){
				if (row.address){
					if (assocUnitsByAddress[row.address]){
						assocUnitsByAddress[row.address].forEach(function(unit){
							if (!assocUniqueUnitsByPeer[row.peer])
								assocUniqueUnitsByPeer[row.peer] = {};
							assocUniqueUnitsByPeer[row.peer][unit] = true;
						});
					}
				}
				if (row.unit){
					if (!assocUniqueUnitsByPeer[row.peer])
						assocUniqueUnitsByPeer[row.peer] = {};
					assocUniqueUnitsByPeer[row.peer][row.unit] = true;
				}
			});
			Object.keys(assocUniqueUnitsByPeer).forEach(function(peer){
				var ws = getPeerWebSocket(peer);
				if (ws && ws.readyState === ws.OPEN)
					sendJustsaying(ws, 'light/sequence_became_bad', Object.keys(assocUniqueUnitsByPeer[peer]));
			});
		});
	});
}

function getAllAuthorsAndOutputAddresses(objUnit){
	var arrAuthorAddresses = objUnit.authors.map(function(author){ return author.address; });
	if (!objUnit.messages) // voided unit
		return null;
	var arrOutputAddresses = [];
	var arrBaseAAAddresses = [];
	for (var i=0; i<objUnit.messages.length; i++){
		var message = objUnit.messages[i];
		var payload = message.payload;
		if (message.app === "payment" && payload) {
			for (var j = 0; j < payload.outputs.length; j++) {
				var address = payload.outputs[j].address;
				if (arrOutputAddresses.indexOf(address) === -1)
					arrOutputAddresses.push(address);
			}
		}
		else if (message.app === 'definition' && payload.definition[1].base_aa)
			arrBaseAAAddresses.push(payload.definition[1].base_aa);
	}
	var arrAddresses = _.union(arrAuthorAddresses, arrOutputAddresses, arrBaseAAAddresses);
	return {
		author_addresses: arrAuthorAddresses,
		output_addresses: arrOutputAddresses,
		base_aa_addresses: arrBaseAAAddresses,
		addresses: arrAddresses,
	};
}

// if any of the watched addresses are affected, notifies:  1. own UI  2. light clients
function notifyWatchers(objJoint, bGoodSequence, source_ws){
	var bAA = objJoint.new_aa;
	delete objJoint.new_aa;
	var objUnit = objJoint.unit;
	var objAddresses = getAllAuthorsAndOutputAddresses(objUnit);
	if (!objAddresses) // voided unit
		return;
	var arrAddresses = objAddresses.addresses;

	if (_.intersection(arrWatchedAddresses, arrAddresses).length > 0){
		eventBus.emit("new_my_transactions", [objJoint.unit.unit]);
		eventBus.emit("new_my_unit-"+objJoint.unit.unit, objJoint);
	}
	else
		db.query(
			"SELECT 1 FROM my_addresses WHERE address IN(?) UNION SELECT 1 FROM shared_addresses WHERE shared_address IN(?) UNION SELECT 1 FROM my_watched_addresses WHERE address IN(?)", 
			[arrAddresses, arrAddresses, arrAddresses], 
			function(rows){
				if (rows.length > 0){
					eventBus.emit("new_my_transactions", [objJoint.unit.unit]);
					eventBus.emit("new_my_unit-"+objJoint.unit.unit, objJoint);
				}
			}
		);
	
	if (conf.bLight)
		return;
	if (objJoint.ball || !bGoodSequence) // If already stable, light clients will require a proof. We notify them only good sequence unit.
		return;
	if (!bWatchingForLight)
		return;
	// this is a new unstable joint, light clients will accept it without proof
	db.query("SELECT peer FROM watched_light_addresses WHERE address IN(?)", [arrAddresses], function(rows){
		if (rows.length === 0)
			return;
		if (!objUnit.timestamp)
			objUnit.timestamp = Math.round(Date.now()/1000); // light clients need timestamp
		rows.forEach(function(row){
			var ws = getPeerWebSocket(row.peer);
			if (ws && ws.readyState === ws.OPEN && ws !== source_ws) {
				if (bAA) // trigger a get_history request to receive the aa_response
					sendJustsaying(ws, 'light/have_updates');
				else
					sendJoint(ws, objJoint);
			}
		});
	});

	// look for transactions sent to watched AAs
	if (bAA) // skip secondary triggers, we'll see their responses in a moment
		return;
	var arrOutputAddresses = objAddresses.output_addresses;
	var arrAuthorAddresses = objAddresses.author_addresses;
	var arrBaseAAAddresses = objAddresses.base_aa_addresses;
	var arrAllAAAddresses = arrOutputAddresses.concat(arrBaseAAAddresses);
	db.query("SELECT peer, address, aa FROM watched_light_aas WHERE aa IN(?)", [arrAllAAAddresses], function (rows) {
		rows.forEach(function (row) {
			if ((!row.address || arrAuthorAddresses.includes(row.address)) && arrOutputAddresses.includes(row.aa)) {
				var ws = getPeerWebSocket(row.peer);
				if (ws && ws.readyState === ws.OPEN && ws !== source_ws)
					sendJustsaying(ws, 'light/aa_request', { aa_address: row.aa, unit: objUnit });
			}
			if (arrBaseAAAddresses.includes(row.aa)) {
				var ws = getPeerWebSocket(row.peer);
				if (ws && ws.readyState === ws.OPEN && ws !== source_ws)
					sendJustsaying(ws, 'light/aa_definition', objUnit);
			}
		});
	});
}

eventBus.on('mci_became_stable', notifyWatchersAboutStableJoints);

function notifyWatchersAboutStableJoints(mci){
	// the event was emitted from inside mysql transaction, make sure it completes so that the changes are visible
	// If the mci became stable in determineIfStableInLaterUnitsAndUpdateStableMcFlag (rare), write lock is released before the validation commits, 
	// so we might not see this mci as stable yet. Hopefully, it'll complete before light/have_updates roundtrip
	mutex.lock(["write"], function(unlock){
		unlock(); // we don't need to block writes, we requested the lock just to wait that the current write completes
		notifyLocalWatchedAddressesAboutStableJoints(mci);
		console.log("notifyWatchersAboutStableJoints "+mci);
		if (mci <= 1 || !bWatchingForLight)
			return;
		storage.findLastBallMciOfMci(db, mci, function(last_ball_mci){
			storage.findLastBallMciOfMci(db, mci-1, function(prev_last_ball_mci){
				if (prev_last_ball_mci === last_ball_mci)
					return;
				notifyLightClientsAboutStableJoints(prev_last_ball_mci, last_ball_mci);
			});
		});
	});
}

// from_mci is non-inclusive, to_mci is inclusive
function notifyLightClientsAboutStableJoints(from_mci, to_mci){
	db.query(
		"SELECT peer FROM units CROSS JOIN unit_authors USING(unit) CROSS JOIN watched_light_addresses USING(address) \n\
		WHERE main_chain_index>? AND main_chain_index<=? \n\
		UNION \n\
		SELECT peer FROM units CROSS JOIN outputs USING(unit) CROSS JOIN watched_light_addresses USING(address) \n\
		WHERE main_chain_index>? AND main_chain_index<=? \n\
		UNION \n\
		SELECT peer FROM units CROSS JOIN watched_light_units USING(unit) \n\
		WHERE main_chain_index>? AND main_chain_index<=?",
		[from_mci, to_mci, from_mci, to_mci, from_mci, to_mci],
		function(rows){
			rows.forEach(function(row){
				var ws = getPeerWebSocket(row.peer);
				if (ws && ws.readyState === ws.OPEN)
					sendJustsaying(ws, 'light/have_updates');
			});
			db.query("DELETE FROM watched_light_units \n\
				WHERE unit IN (SELECT unit FROM units WHERE main_chain_index>? AND main_chain_index<=?)", [from_mci, to_mci], function() {
				
			});
		}
	);
}

function notifyLocalWatchedAddressesAboutStableJoints(mci){
	function handleRows(rows){
		if (rows.length > 0){
			eventBus.emit('my_transactions_became_stable', rows.map(function(row){ return row.unit; }));
			rows.forEach(function(row){
				eventBus.emit('my_stable-'+row.unit);
			});
		}
	}
	if (arrWatchedAddresses.length > 0)
		db.query(
			"SELECT unit FROM units CROSS JOIN unit_authors USING(unit) \n\
			WHERE main_chain_index=? AND address IN("+arrWatchedAddresses.map(db.escape).join(', ')+") AND sequence='good' \n\
			UNION \n\
			SELECT unit FROM units CROSS JOIN outputs USING(unit) \n\
			WHERE main_chain_index=? AND address IN("+arrWatchedAddresses.map(db.escape).join(', ')+") AND sequence='good'",
			[mci, mci],
			handleRows
		);
	db.query(
		"SELECT unit FROM units CROSS JOIN unit_authors USING(unit) CROSS JOIN my_addresses USING(address) WHERE main_chain_index=? AND sequence='good' \n\
		UNION \n\
		SELECT unit FROM units CROSS JOIN outputs USING(unit) CROSS JOIN my_addresses USING(address) WHERE main_chain_index=? AND sequence='good' \n\
		UNION \n\
		SELECT unit FROM units CROSS JOIN unit_authors USING(unit) CROSS JOIN shared_addresses ON address=shared_address WHERE main_chain_index=? AND sequence='good' \n\
		UNION \n\
		SELECT unit FROM units CROSS JOIN outputs USING(unit) CROSS JOIN shared_addresses ON address=shared_address WHERE main_chain_index=? AND sequence='good'\n\
		UNION \n\
		SELECT unit FROM units CROSS JOIN unit_authors USING(unit) CROSS JOIN my_watched_addresses USING(address) WHERE main_chain_index=? AND sequence='good' \n\
		UNION \n\
		SELECT unit FROM units CROSS JOIN outputs USING(unit) CROSS JOIN my_watched_addresses USING(address) WHERE main_chain_index=? AND sequence='good'",
		[mci, mci, mci, mci, mci, mci],
		handleRows
	);
}

function aaResponseAffectsAddress(objAAResponse, address) {
	if (objAAResponse.trigger_address === address)
		return true;
	if (objAAResponse.objResponseUnit && objAAResponse.objResponseUnit.messages.find(function (message) {
		return message.app === 'payment' && message.payload.outputs.find(function (output) {
			return output.address === address;
		});
	}))
		return true;
	// check if any updated state variable name contains our address
	var aa_address = objAAResponse.aa_address;
	if (objAAResponse.updatedStateVars && objAAResponse.updatedStateVars[aa_address]) {
		for (var var_name in objAAResponse.updatedStateVars[aa_address]) {
			if (var_name.indexOf(address) >= 0)
				return true;
		}
	}
	// check if the error message contains our address
	if (objAAResponse.response.error && errorToString(objAAResponse.response.error).indexOf(address) >= 0)
		return true;
	// check if any response variable name contains our address
	if (objAAResponse.response.responseVars) {
		for (var var_name in objAAResponse.response.responseVars) {
			if (var_name.indexOf(address) >= 0)
				return true;
		}
	}
	return false;
}

eventBus.on('aa_response', function (objAAResponse) {
	if (!bWatchingForLight)
		return;
	db.query("SELECT peer, address FROM watched_light_aas WHERE aa=?", [objAAResponse.aa_address], function (rows) {
		if (rows.length === 0)
			return;
		if (objAAResponse.logs) { // do not send the logs over the wire
			objAAResponse = _.clone(objAAResponse);
			delete objAAResponse.logs;
		}
		rows.forEach(function (row) {
			if (!row.address || aaResponseAffectsAddress(objAAResponse, row.address)) {
				var ws = getPeerWebSocket(row.peer);
				if (ws && ws.readyState === ws.OPEN)
					sendJustsaying(ws, 'light/aa_response', objAAResponse);
			}
		});
	});
});

// full wallets only
eventBus.on('aa_definition_saved', function (payload, unit) {
	if (!bWatchingForLight)
		return;
	var base_aa = payload.definition[1].base_aa;
	if (!base_aa)
		return;
	db.query("SELECT peer FROM watched_light_aas WHERE aa=?", [base_aa], function (rows) {
		var arrWses = [];
		rows.forEach(function (row) {
			var ws = getPeerWebSocket(row.peer);
			if (ws && ws.readyState === ws.OPEN)
				arrWses.push(ws);
		});
		if (arrWses.length === 0)
			return;
		storage.readJoint(db, unit, {
			ifNotFound: function () {
				console.log('recently saved unit ' + unit + ' not found');
			},
			ifFound: function (objJoint) {
				var objUnit = objJoint.unit;
				arrWses.forEach(function (ws) {
					sendJustsaying(ws, 'light/aa_definition_saved', objUnit);
					// posted by an AA, it was skipped when handling the request
					if (objUnit.authors.length === 1 && !objUnit.authors[0].authentifiers)
						sendJustsaying(ws, 'light/aa_definition', objUnit);
				});
			}
		})
	});
});

function addLightWatchedAddress(address, handle){
	sendJustsayingToLightVendor('light/new_address_to_watch', address, handle);
}

function addTempLightWatchedAddress(address, handle) {
	if (!arrTempWatchedAddresses.includes(address))
		arrTempWatchedAddresses.push(address);
	addLightWatchedAddress(address, handle);
	eventBus.on('connected', () => addLightWatchedAddress(address));
}

function addLightWatchedAa(aa, address, handle){
	var params = { aa: aa };
	if (address)
		params.address = address;
	sendJustsayingToLightVendor('light/new_aa_to_watch', params, handle);
	eventBus.on('connected', () => sendJustsayingToLightVendor('light/new_aa_to_watch', params));
}

async function flushEvents(forceFlushing) {
	if (peer_events_buffer.length == 0 || (!forceFlushing && peer_events_buffer.length != 100)) {
		return;
	}

	var arrQueryParams = [];
	var objUpdatedHosts = {};
	peer_events_buffer.forEach(function(event_row){
		var host = event_row.host;
		var event = event_row.event;
		var event_date = event_row.event_date;
		if (event === 'new_good'){
			var column = "count_"+event+"_joints";
			_.set(objUpdatedHosts, [host, column], _.get(objUpdatedHosts, [host, column], 0)+1);
		}
		arrQueryParams.push("(" + db.escape(host) +"," + db.escape(event) + "," + db.getFromUnixTime(event_date) + ")");
	});

	for (var host in objUpdatedHosts) {
		var columns_obj = objUpdatedHosts[host];
		var sql_columns_updates = [];
		for (var column in columns_obj) {
			sql_columns_updates.push(column + "=" + column + "+" + columns_obj[column]);
		}
		db.query("UPDATE peer_hosts SET "+sql_columns_updates.join()+" WHERE peer_host=?", [host]);
	}

	db.query("INSERT INTO peer_events (peer_host, event, event_date) VALUES " + arrQueryParams.join());
	peer_events_buffer = [];
}

function writeEvent(event, host){
	if (conf.bLight)
		return;
	if (host === 'byteball.org') host = 'obyte.org';
	if (event === 'invalid' || event === 'nonserial'){
		var column = "count_"+event+"_joints";
		db.query("UPDATE peer_hosts SET "+column+"="+column+"+1 WHERE peer_host=?", [host]);
		db.query("INSERT INTO peer_events (peer_host, event) VALUES (?,?)", [host, event]);
		if (event === 'invalid')
			assocBlockedPeers[host] = Date.now();
		return;
	}
	var event_date = Math.floor(Date.now() / 1000);
	peer_events_buffer.push({host: host, event: event, event_date: event_date});
	flushEvents();
}

function peerIsBlocked(host) {
	if (constants.bTestnet || constants.bDevnet)
		return false;
	return !!assocBlockedPeers[host];
}

function unblockPeers(){
	for (var host in assocBlockedPeers)
		if (assocBlockedPeers[host] < Date.now() - 3600*1000)
			delete assocBlockedPeers[host];
}

function initBlockedPeers(){
	db.query(
		"SELECT peer_host, MAX("+db.getUnixTimestamp('event_date')+") AS ts FROM peer_events \n\
		WHERE event_date>"+db.addTime("-1 HOUR")+" AND event='invalid' \n\
		GROUP BY peer_host",
		function(rows){
			rows.forEach(function(row){
				assocBlockedPeers[row.peer_host] = row.ts*1000;
			});
			try {
				const fs = require('fs');
				const app_data_dir = require('./desktop_app.js').getAppDataDir();
				const filepath = `${app_data_dir}/uncaught_exception_clients.txt`;
				const hosts = fs.readFileSync(filepath, 'utf8').split('\n');
				for (let host of hosts) {
					assocBlockedPeers[host] = Date.now();
					console.log(`added blocked peer ${host} from uncaught_exception_clients.txt`);
				}
				setTimeout(() => {
					fs.unlinkSync(filepath);
					console.log('deleted uncaught_exception_clients.txt file');
				}, 3600*1000);
			}
			catch (e) {
				console.log('no uncaught_exception_clients.txt file found', e);
			}
		}
	);
}


function findAndHandleJointsThatAreReady(unit){
	joint_storage.readDependentJointsThatAreReady(unit, handleSavedJoint);
	handleSavedPrivatePayments(unit);
}

function comeOnline(){
	bCatchingUp = false;
	coming_online_time = Date.now();
	waitTillIdle(function(){
		requestFreeJointsFromAllOutboundPeers();
		setTimeout(cleanBadSavedPrivatePayments, 300*1000);
	});
	waitTillRecentTimestamp();
}

function waitTillRecentTimestamp() {
	if (bCatchingUp) return;
	db.query("SELECT timestamp FROM units ORDER BY main_chain_index DESC LIMIT 1", function (rows) {
		if (rows.length === 0 || rows[0].timestamp > Date.now() / 1000 - 300)
			return eventBus.emit('catching_up_done');
		console.log('last unit is old, will wait');
		setTimeout(waitTillRecentTimestamp, 10 * 1000);
	});
}

function isIdle(){
	//console.log(db._freeConnections.length +"/"+ db._allConnections.length+" connections are free, "+mutex.getCountOfQueuedJobs()+" jobs queued, "+mutex.getCountOfLocks()+" locks held, "+Object.keys(assocUnitsInWork).length+" units in work");
	const db_conns = db.getCountUsedConnections();
	const queued = mutex.getCountOfQueuedJobs();
	const locks = mutex.getCountOfLocks();
	const units_in_work = Object.keys(assocUnitsInWork).length;
	const is_idle = (db_conns === 0 && queued === 0 && locks === 0 && units_in_work === 0);
	if (!is_idle)
		console.log({ db_conns, queued, locks, units_in_work }, units_in_work <= 3 ? assocUnitsInWork : undefined);
	return is_idle;
}

function _waitTillIdle(onIdle){
	if (isIdle()){
		eventBus.emit('idle'); // first call the callbacks that were queued earlier
		if (onIdle)
			onIdle();
	}
	else{
		console.log('not idle, will wait');
		if (onIdle)
			eventBus.once('idle', onIdle);
		setTimeout(_waitTillIdle, 100);
	}
}

function waitTillIdle(onIdle) {
	if (!onIdle)
		return new Promise(resolve => _waitTillIdle(resolve));
	_waitTillIdle(onIdle);
}

function isSyncIdle() {
	return (db.getCountUsedConnections() === 0 && Object.keys(assocUnitsInWork).length === 0);
}

function _waitTillSyncIdle(onIdle){
	if (isSyncIdle()){
		eventBus.emit('sync_idle'); // first call the callbacks that were queued earlier
		if (onIdle)
			onIdle();
	}
	else {
		console.log('sync is active, will wait');
		if (onIdle)
			eventBus.once('sync_idle', onIdle);
		setTimeout(_waitTillSyncIdle, 100);
	}
}

function waitTillSyncIdle(onIdle) {
	if (!onIdle)
		return new Promise(resolve => _waitTillSyncIdle(resolve));
	_waitTillSyncIdle(onIdle);
}

function broadcastJoint(objJoint){
	if (!conf.bLight) // the joint was already posted to light vendor before saving
		[...wss.clients].concat(arrOutboundPeers).forEach(function(client) {
			if (client.bSubscribed)
				sendJoint(client, objJoint);
		});
	notifyWatchers(objJoint, true);
}

function onNewAA(objUnit) {
	findAndHandleJointsThatAreReady(objUnit.unit);
	notifyWatchers({ unit: objUnit, new_aa: true }, true);
}

function onSystemVarUpdated(subject, value) {
	console.log('onSystemVarUpdated', subject, value);
	sendUpdatedSysVarsToAllLight();
	// update my witnesses with the new OP list unless catching up
	if (subject === 'op_list' && !bCatchingUp) {
		const arrOPs = JSON.parse(value);
		myWitnesses.readMyWitnesses(arrWitnesses => {
			if (arrWitnesses.length === 0)
				return console.log('no witnesses yet');
			const diff1 = _.difference(arrWitnesses, arrOPs);
			if (diff1.length === 0)
				return console.log("witnesses didn't change");
			const diff2 = _.difference(arrOPs, arrWitnesses);
			if (diff2.length !== diff1.length)
				throw Error(`different lengths of diffs: ${JSON.stringify(diff1)} vs ${JSON.stringify(diff2)}`);
			for (let i = 0; i < diff1.length; i++) {
				const old_witness = diff1[i];
				const new_witness = diff2[i];
				console.log(`replacing witness ${old_witness} with ${new_witness}`);
				myWitnesses.replaceWitness(old_witness, new_witness, err => {
					if (err)
						throw Error(`failed to replace witness ${old_witness} with ${new_witness}: ${err}`);
				});
			}
		}, 'ignore');
	}
}


// catchup

function checkCatchupLeftovers(){
	db.query(
		"SELECT 1 FROM hash_tree_balls \n\
		UNION \n\
		SELECT 1 FROM catchup_chain_balls \n\
		LIMIT 1",
		function(rows){
			if (rows.length === 0)
				return console.log('no leftovers');
			console.log('have catchup leftovers from the previous run');
			findNextPeer(null, function(ws){
				console.log('will request leftovers from '+ws.peer);
				if (!bCatchingUp && !bWaitingForCatchupChain)
					requestCatchup(ws);
			});
		}
	);
}

function requestCatchup(ws){
	console.log("will request catchup from "+ws.peer);
	eventBus.emit('catching_up_started');
//	if (conf.storage === 'sqlite')
//		db.query("PRAGMA cache_size=-200000", function(){});
	catchup.purgeHandledBallsFromHashTree(db, function(){
		db.query(
			"SELECT hash_tree_balls.unit FROM hash_tree_balls LEFT JOIN units USING(unit) WHERE units.unit IS NULL ORDER BY ball_index", 
			function(tree_rows){ // leftovers from previous run
				if (tree_rows.length > 0){
					bCatchingUp = true;
					console.log("will request balls found in hash tree");
					requestNewMissingJoints(ws, tree_rows.map(function(tree_row){ return tree_row.unit; }));
					waitTillHashTreeFullyProcessedAndRequestNext(ws);
					return;
				}
				db.query("SELECT 1 FROM catchup_chain_balls LIMIT 1", function(chain_rows){ // leftovers from previous run
					if (chain_rows.length > 0){
						bCatchingUp = true;
						requestNextHashTree(ws);
						return;
					}
					// we are not switching to catching up mode until we receive a catchup chain - don't allow peers to throw us into 
					// catching up mode by just sending a ball
					
					// to avoid duplicate requests, we are raising this flag before actually sending the request 
					// (will also reset the flag only after the response is fully processed)
					bWaitingForCatchupChain = true;
					
					console.log('will read last stable mci for catchup');
					storage.readLastStableMcIndex(db, function(last_stable_mci){
						storage.readLastMainChainIndex(function(last_known_mci){
							myWitnesses.readMyWitnesses(function(arrWitnesses){
								var params = {witnesses: arrWitnesses, last_stable_mci: last_stable_mci, last_known_mci: last_known_mci};
								sendRequest(ws, 'catchup', params, true, handleCatchupChain);
							}, 'wait');
						});
					});
				});
			}
		);
	});
}

function handleCatchupChain(ws, request, response){
	if (!response || response.error){
		bWaitingForCatchupChain = false;
		console.log('catchup request got error response: ', response ? response.error : 'no response');
		// findLostJoints will wake up and trigger another attempt to request catchup
		return;
	}
	var catchupChain = response;
	console.log('received catchup chain from '+ws.peer);
	catchup.processCatchupChain(catchupChain, ws.peer, request.params.witnesses, {
		ifError: function(error){
			bWaitingForCatchupChain = false;
			sendError(ws, error);
		},
		ifOk: function(){
			bWaitingForCatchupChain = false;
			bCatchingUp = true;
			requestNextHashTree(ws);
		},
		ifCurrent: function(){
			bWaitingForCatchupChain = false;
		}
	});
}



// hash tree

function requestNextHashTree(ws){
	eventBus.emit('catchup_next_hash_tree');
	db.query("SELECT ball FROM catchup_chain_balls ORDER BY member_index LIMIT 2", function(rows){
		if (rows.length === 0)
			return comeOnline();
		if (rows.length === 1){
			db.query("DELETE FROM catchup_chain_balls WHERE ball=?", [rows[0].ball], function(){
				comeOnline();
			});
			return;
		}
		var from_ball = rows[0].ball;
		var to_ball = rows[1].ball;
		
		// don't send duplicate requests
		for (var tag in ws.assocPendingRequests)
			if (ws.assocPendingRequests[tag].request.command === 'get_hash_tree'){
				console.log("already requested hash tree from this peer");
				return;
			}
		sendRequest(ws, 'get_hash_tree', {from_ball: from_ball, to_ball: to_ball}, true, handleHashTree);
	});
}

function handleHashTree(ws, request, response){
	if (!response || response.error){
		console.log('get_hash_tree got error response: ', response ? response.error : 'no response');
		waitTillHashTreeFullyProcessedAndRequestNext(ws); // after 1 sec, it'll request the same hash tree, likely from another peer
		return;
	}
	console.log('received hash tree from '+ws.peer);
	var hashTree = response;
	catchup.processHashTree(hashTree.balls, {
		ifError: function(error){
			sendError(ws, error);
			waitTillHashTreeFullyProcessedAndRequestNext(ws); // after 1 sec, it'll request the same hash tree, likely from another peer
		},
		ifOk: function(){
			requestNewMissingJoints(ws, hashTree.balls.map(function(objBall){ return objBall.unit; }));
			waitTillHashTreeFullyProcessedAndRequestNext(ws);
		}
	});
}

function haveManyUnhandledHashTreeBalls(){
	var count = 0;
	for (var ball in storage.assocHashTreeUnitsByBall){
		var unit = storage.assocHashTreeUnitsByBall[ball];
		if (!storage.assocUnstableUnits[unit]){
			count++;
			if (count > 30)
				return true;
		}
	}
	return false;
}

function waitTillHashTreeFullyProcessedAndRequestNext(ws){
	setTimeout(function(){
	//	db.query("SELECT COUNT(*) AS count FROM hash_tree_balls LEFT JOIN units USING(unit) WHERE units.unit IS NULL", function(rows){
		//	var count = Object.keys(storage.assocHashTreeUnitsByBall).length;
			if (!haveManyUnhandledHashTreeBalls()){
				findNextPeer(ws, function(next_ws){
					requestNextHashTree(next_ws);
				});
			}
			else
				waitTillHashTreeFullyProcessedAndRequestNext(ws);
	//	});
	}, 100);
}




// private payments

function sendPrivatePaymentToWs(ws, arrChains){
	// each chain is sent as separate ws message
	arrChains.forEach(function(arrPrivateElements){
		sendJustsaying(ws, 'private_payment', arrPrivateElements);
	});
}

// sends multiple private payloads and their corresponding chains
function sendPrivatePayment(peer, arrChains){
	var ws = getPeerWebSocket(peer);
	if (ws)
		return sendPrivatePaymentToWs(ws, arrChains);
	findOutboundPeerOrConnect(peer, function(err, ws){
		if (!err)
			sendPrivatePaymentToWs(ws, arrChains);
	}, true);
}

// handles one private payload and its chain
function handleOnlinePrivatePayment(ws, arrPrivateElements, bViaHub, callbacks){
	if (!ValidationUtils.isNonemptyArray(arrPrivateElements))
		return callbacks.ifError("private_payment content must be non-empty array");
	
	var unit = arrPrivateElements[0].unit;
	var message_index = arrPrivateElements[0].message_index;
	var output_index = arrPrivateElements[0].payload.denomination ? arrPrivateElements[0].output_index : -1;
	if (!ValidationUtils.isValidBase64(unit, constants.HASH_LENGTH))
		return callbacks.ifError("invalid unit");
	if (!ValidationUtils.isNonnegativeInteger(message_index))
		return callbacks.ifError("invalid message_index");
	if (!(ValidationUtils.isNonnegativeInteger(output_index) || output_index === -1))
		return callbacks.ifError("invalid output_index");

	var savePrivatePayment = function(cb){
		// we may receive the same unit and message index but different output indexes if recipient and cosigner are on the same device.
		// in this case, we also receive the same (unit, message_index, output_index) twice - as cosigner and as recipient.  That's why IGNORE.
		db.query(
			"INSERT "+db.getIgnore()+" INTO unhandled_private_payments (unit, message_index, output_index, json, peer) VALUES (?,?,?,?,?)", 
			[unit, message_index, output_index, JSON.stringify(arrPrivateElements), bViaHub ? '' : ws.peer], // forget peer if received via hub
			function(){
				callbacks.ifQueued();
				if (cb)
					cb();
			}
		);
	};
	
	if (conf.bLight && arrPrivateElements.length > 1){
		savePrivatePayment(function(){
			updateLinkProofsOfPrivateChain(arrPrivateElements, unit, message_index, output_index);
			rerequestLostJointsOfPrivatePayments(); // will request the head element
		});
		return;
	}

	joint_storage.checkIfNewUnit(unit, {
		ifKnown: function(){
			//assocUnitsInWork[unit] = true;
			privatePayment.validateAndSavePrivatePaymentChain(arrPrivateElements, {
				ifOk: function(){
					//delete assocUnitsInWork[unit];
					callbacks.ifAccepted(unit);
					eventBus.emit("new_my_transactions", [unit]);
				},
				ifError: function(error){
					//delete assocUnitsInWork[unit];
					callbacks.ifValidationError(unit, error);
				},
				ifWaitingForChain: function(){
					savePrivatePayment();
				}
			});
		},
		ifNew: function(){
			savePrivatePayment();
			// if received via hub, I'm requesting from the same hub, thus telling the hub that this unit contains a private payment for me.
			// It would be better to request missing joints from somebody else
			requestNewMissingJoints(ws, [unit]);
		},
		ifKnownUnverified: savePrivatePayment,
		ifKnownBad: function(){
			callbacks.ifValidationError(unit, "known bad");
		}
	});
}
	
// if unit is undefined, find units that are ready
function handleSavedPrivatePayments(unit){
	//if (unit && assocUnitsInWork[unit])
	//    return;
	if (!my_device_address) return; // skip if we don't have a wallet
	if (!unit && mutex.isAnyOfKeysLocked(["private_chains"])) // we are still downloading the history (light)
		return console.log("skipping handleSavedPrivatePayments because history download is still under way");
	var lock = unit ? mutex.lock : mutex.lockOrSkip;
	lock(["saved_private"], function(unlock){
		var sql = unit
			? "SELECT json, peer, unit, message_index, output_index, linked FROM unhandled_private_payments WHERE unit="+db.escape(unit)
			: "SELECT json, peer, unit, message_index, output_index, linked FROM unhandled_private_payments CROSS JOIN units USING(unit)";
		db.query(sql, function(rows){
			if (rows.length === 0)
				return unlock();
			var assocNewUnits = {};
			async.each( // handle different chains in parallel
				rows,
				function(row, cb){
					var arrPrivateElements = JSON.parse(row.json);
					var ws = getPeerWebSocket(row.peer);
					if (ws && ws.readyState !== ws.OPEN)
						ws = null;
					
					var validateAndSave = function(){
						var objHeadPrivateElement = arrPrivateElements[0];
						try {
							var json_payload_hash = objectHash.getBase64Hash(objHeadPrivateElement.payload, true);
						}
						catch (e) {
							console.log("getBase64Hash failed for private element", objHeadPrivateElement.payload, e);
							if (ws)
								sendResult(ws, {private_payment_in_unit: row.unit, result: 'error', error: e.toString()});
							deleteHandledPrivateChain(row.unit, row.message_index, row.output_index, cb);
						}
						var key = 'private_payment_validated-'+objHeadPrivateElement.unit+'-'+json_payload_hash+'-'+row.output_index;
						privatePayment.validateAndSavePrivatePaymentChain(arrPrivateElements, {
							ifOk: function(){
								if (ws)
									sendResult(ws, {private_payment_in_unit: row.unit, result: 'accepted'});
								if (row.peer) // received directly from a peer, not through the hub
									eventBus.emit("new_direct_private_chains", [arrPrivateElements]);
								assocNewUnits[row.unit] = true;
								deleteHandledPrivateChain(row.unit, row.message_index, row.output_index, cb);
								console.log('emit '+key);
								eventBus.emit(key, true);
							},
							ifError: function(error){
								console.log("validation of priv: "+error);
							//	throw Error(error);
								if (ws)
									sendResult(ws, {private_payment_in_unit: row.unit, result: 'error', error: error});
								deleteHandledPrivateChain(row.unit, row.message_index, row.output_index, cb);
								eventBus.emit(key, false);
							},
							// light only. Means that chain joints (excluding the head) not downloaded yet or not stable yet
							ifWaitingForChain: function(){
								console.log('waiting for chain: unit '+row.unit+', message '+row.message_index+' output '+row.output_index);
								cb();
							}
						});
					};
					
					if (conf.bLight && arrPrivateElements.length > 1 && !row.linked)
						updateLinkProofsOfPrivateChain(arrPrivateElements, row.unit, row.message_index, row.output_index, cb, validateAndSave);
					else
						validateAndSave();
					
				},
				function(){
					unlock();
					var arrNewUnits = Object.keys(assocNewUnits);
					if (arrNewUnits.length > 0)
						eventBus.emit("new_my_transactions", arrNewUnits);
				}
			);
		});
	});
}

function deleteHandledPrivateChain(unit, message_index, output_index, cb){
	db.query("DELETE FROM unhandled_private_payments WHERE unit=? AND message_index=? AND output_index=?", [unit, message_index, output_index], function(){
		cb();
	});
}

// full only
function cleanBadSavedPrivatePayments(){
	if (conf.bLight || bCatchingUp)
		return;
	db.query(
		"SELECT DISTINCT unhandled_private_payments.unit FROM unhandled_private_payments LEFT JOIN units USING(unit) \n\
		WHERE units.unit IS NULL AND unhandled_private_payments.creation_date<"+db.addTime('-1 DAY'),
		function(rows){
			rows.forEach(function(row){
				breadcrumbs.add('deleting bad saved private payment '+row.unit);
				db.query("DELETE FROM unhandled_private_payments WHERE unit=?", [row.unit]);
			});
		}
	);
	
}

// light only
function rerequestLostJointsOfPrivatePayments(){
	if (!conf.bLight || !exports.light_vendor_url)
		return;
	db.query(
		"SELECT DISTINCT unhandled_private_payments.unit FROM unhandled_private_payments LEFT JOIN units USING(unit) WHERE units.unit IS NULL",
		function(rows){
			if (rows.length === 0)
				return;
			var arrUnits = rows.map(function(row){ return row.unit; });
			findOutboundPeerOrConnect(exports.light_vendor_url, function(err, ws){
				if (err)
					return;
				requestNewMissingJoints(ws, arrUnits);
			});
		}
	);
}

// light only
function requestUnfinishedPastUnitsOfPrivateChains(arrChains, onDone){
	mutex.lock(["private_chains"], function(unlock){
		function finish(){
			unlock();
			if (onDone)
				onDone();
		}
		privatePayment.findUnfinishedPastUnitsOfPrivateChains(arrChains, true, function(arrUnits){
			if (arrUnits.length === 0)
				return finish();
			breadcrumbs.add(arrUnits.length+" unfinished past units of private chains");
			requestHistoryFor(arrUnits, [], err => {
				if (err) {
					console.log(`error getting history for unfinished units of private payments`, err);
					return finish();
				}
				// get units that are still new or unstable after refreshing the history
				storage.filterNewOrUnstableUnits(arrUnits, async arrMissingUnits => {
					if (arrMissingUnits.length === 0) return finish();
					console.log(`will delete unhandled private payments whose units are not known after 1 day`, arrMissingUnits);
					await db.query(`DELETE FROM unhandled_private_payments WHERE unit IN(${arrMissingUnits.map(db.escape).join(', ')}) AND creation_date < ${db.addTime('-1 DAY')}`);
					finish();
				});
			});
		});
	});
}

function requestHistoryAfterMCI(arrUnits, addresses, minMCI, onDone){
	if (!onDone)
		onDone = function(){};
	var arrAddresses = Array.isArray(addresses) ? addresses : [];
	if (!arrUnits.every(unit => ValidationUtils.isValidBase64(unit, constants.HASH_LENGTH)))
		throw Error("some units are invalid: " + arrUnits.join(', '));
	myWitnesses.readMyWitnesses(function(arrWitnesses){
		var objHistoryRequest = {witnesses: arrWitnesses};
		if (arrUnits.length)
			objHistoryRequest.requested_joints = arrUnits;
		if (arrAddresses.length)
			objHistoryRequest.addresses = arrAddresses;
		if (minMCI !== -1)
			objHistoryRequest.min_mci = minMCI;
		requestFromLightVendor('light/get_history', objHistoryRequest, function(ws, request, response){
			if (!response || response.error){
				console.log(response ? response.error : 'no response');
				return onDone(response ? response.error : 'no response');
			}
			light.processHistory(response, arrWitnesses, {
				ifError: function(err){
					sendError(ws, err);
					onDone(err);
				},
				ifOk: function(){
					onDone();
				}
			});
		});
	}, 'wait');
}

function requestHistoryFor(arrUnits, addresses, onDone) {
	return requestHistoryAfterMCI(arrUnits, addresses, -1, onDone);
}

function requestProofsOfJointsIfNewOrUnstable(arrUnits, onDone){
	if (!onDone)
		onDone = function(){};
	storage.filterNewOrUnstableUnits(arrUnits, function(arrNewOrUnstableUnits){
		if (arrNewOrUnstableUnits.length === 0)
			return onDone();
		requestHistoryFor(arrUnits, [], onDone);
	});
}

// light only
function requestUnfinishedPastUnitsOfSavedPrivateElements(){
	mutex.lockOrSkip(['saved_private_chains'], function(unlock){
		db.query("SELECT json FROM unhandled_private_payments", function(rows){
			eventBus.emit('unhandled_private_payments_left', rows.length);
			if (rows.length === 0)
				return unlock();
			breadcrumbs.add(rows.length+" unhandled private payments");
			var arrChains = [];
			rows.forEach(function(row){
				var arrPrivateElements = JSON.parse(row.json);
				arrChains.push(arrPrivateElements);
			});
			requestUnfinishedPastUnitsOfPrivateChains(arrChains, function onPrivateChainsReceived(err){
				if (err){
					console.log("error from requestUnfinishedPastUnitsOfPrivateChains: "+err);
					return unlock();
				}
				console.log("requestUnfinishedPastUnitsOfPrivateChains done");
				handleSavedPrivatePayments();
				setTimeout(unlock, 2000);
			});
		});
	});
}

// light only
// Note that we are leaking to light vendor information about the full chain. 
// If the light vendor was a party to any previous transaction in this chain, he'll know how much we received.
function checkThatEachChainElementIncludesThePrevious(arrPrivateElements, handleResult){
	if (arrPrivateElements.length === 1) // an issue
		return handleResult(true);
	var arrUnits = arrPrivateElements.map(function(objPrivateElement){ return objPrivateElement.unit; });
	requestFromLightVendor('light/get_link_proofs', arrUnits, function(ws, request, response){
		if (!response || response.error)
			return handleResult(null); // undefined result
		var arrChain = response;
		if (!ValidationUtils.isNonemptyArray(arrChain))
			return handleResult(null); // undefined result
		light.processLinkProofs(arrUnits, arrChain, {
			ifError: function(err){
				console.log("linkproof validation failed: "+err);
			//	throw Error(err);
				handleResult(false);
			},
			ifOk: function(){
				console.log("linkproof validated ok");
				handleResult(true);
			}
		});
	});
}

// light only
function updateLinkProofsOfPrivateChain(arrPrivateElements, unit, message_index, output_index, onFailure, onSuccess){
	if (!conf.bLight)
		throw Error("not light but updateLinkProofsOfPrivateChain");
	if (!onFailure)
		onFailure = function(){};
	if (!onSuccess)
		onSuccess = function(){};
	checkThatEachChainElementIncludesThePrevious(arrPrivateElements, function(bLinked){
		if (bLinked === null)
			return onFailure();
		if (!bLinked)
			return deleteHandledPrivateChain(unit, message_index, output_index, onFailure);
		// the result cannot depend on output_index
		db.query("UPDATE unhandled_private_payments SET linked=1 WHERE unit=? AND message_index=?", [unit, message_index], function(){
			onSuccess();
		});
	});
}

function initWitnessesIfNecessary(ws, onDone){
	onDone = onDone || function(){};
	myWitnesses.readMyWitnesses(function(arrWitnesses){
		if (arrWitnesses.length > 0) // already have witnesses
			return onDone();
		sendRequest(ws, 'get_witnesses', null, false, function(ws, request, arrWitnesses){
			if (!arrWitnesses || arrWitnesses.error){
				console.log('get_witnesses returned error: ', arrWitnesses ? arrWitnesses.error : 'no response');
				return onDone();
			}
			if (!ValidationUtils.isArrayOfLength(arrWitnesses, constants.COUNT_WITNESSES) || !arrWitnesses.every(ValidationUtils.isValidAddress)) {
				console.log('get_witnesses returned invalid witnesses:', arrWitnesses);
				return onDone();
			}
			myWitnesses.insertWitnesses(arrWitnesses, onDone);
		});
	}, 'ignore');
}


// hub

function deleteOverlengthMessagesIfLimitIsSet(ws, device_address, handle){
	if (ws.max_message_length)
		db.query("DELETE FROM device_messages WHERE device_address=? AND LENGTH(message)>?", [device_address, ws.max_message_length], function(){
			return handle();
		});
	else
		return handle();
}


function sendStoredDeviceMessages(ws, device_address){
	deleteOverlengthMessagesIfLimitIsSet(ws, device_address, function(){
		var max_message_count = ws.max_message_count ? ws.max_message_count : 100;
		db.query("SELECT message_hash, message FROM device_messages WHERE device_address=? ORDER BY creation_date LIMIT ?", [device_address, max_message_count], function(rows){
			rows.forEach(function(row){
				sendJustsaying(ws, 'hub/message', {message_hash: row.message_hash, message: JSON.parse(row.message)});
			});
			sendInfo(ws, rows.length+" messages sent");
			sendJustsaying(ws, 'hub/message_box_status', (rows.length === max_message_count) ? 'has_more' : 'empty');
		});
	});
}

function version2int(version){
	if (!version) return 0;
	var arr = version.split('.');
	return arr[0]*1000000 + arr[1]*1000 + arr[2]*1;
}


// switch/case different message types

function handleJustsaying(ws, subject, body){
	switch (subject){
		case 'refresh':
			if (bCatchingUp)
				return;
			var mci = body;
			if (ValidationUtils.isNonnegativeInteger(mci))
				return sendJointsSinceMci(ws, mci);
			else
				return sendFreeJoints(ws);
			
		case 'version':
			if (!body)
				return;
			ws.library_version = body.library_version;
			if (typeof ws.library_version !== 'string') {
				sendError(ws, "invalid library_version");
				return ws.close(1000, "invalid library_version");
			}
			if (typeof body.protocol_version !== 'string') {
				sendError(ws, "invalid protocol_version");
				return ws.close(1000, "invalid protocol_version");
			}
			if (typeof body.alt !== 'string') {
				sendError(ws, "invalid alt");
				return ws.close(1000, "invalid alt");
			}
			if (version2int(ws.library_version) < version2int(constants.minCoreVersion)){
				ws.old_core = true;
				ws.bSubscribed = false;
				return disconnectOldVersion(ws, "old core");
			}
			if (version2int(ws.library_version) < version2int(constants.minCoreVersionForFullNodes)){
				ws.old_core = true;
				if (ws.bSubscribed){
					ws.bSubscribed = false;
					return disconnectOldVersion(ws, "old core (full)");
				}
			}
			if (constants.supported_versions.indexOf(body.protocol_version) === -1){
				sendError(ws, 'Incompatible versions, I support '+constants.supported_versions.join(', ')+', yours '+body.protocol_version);
				ws.close(1000, 'incompatible versions');
				return;
			}
			if (body.alt !== constants.alt){
				sendError(ws, 'Incompatible alts, mine '+constants.alt+', yours '+body.alt);
				ws.close(1000, 'incompatible alts');
				return;
			}
			if (version2int(ws.library_version) < version2int(constants.minCoreVersionToSharePeers)){
				ws.dontSharePeers = true;
				sendJustsaying(ws, "please upgrade the core to at least " + constants.minCoreVersionToSharePeers);
			}
			eventBus.emit('peer_version', ws, body); // handled elsewhere
			break;

		case 'new_version': // a new version is available
			if (!body)
				return;
			if (ws.bLoggingIn || ws.bLoggedIn) // accept from hub only
				eventBus.emit('new_version', ws, body);
			break;

		case 'hub/push_project_number':
			if (!body)
				return;
			if (ws.bLoggingIn || ws.bLoggedIn)
				eventBus.emit('receivedPushProjectNumber', ws, body);
			break;
		
		case 'bugreport':
			if (!conf.bug_sink_email)
				return console.log("no bug_sink_email, not accepting bugreport");
			if (!body || !body.exception || !ValidationUtils.isNonemptyString(body.message))
				return console.log("invalid bugreport");
			try {
				var arrParts = body.exception.toString().split("Breadcrumbs", 2);
				var text = body.message + ' ' + arrParts[0];
				var matches = body.message.match(/message encrypted to unknown key, device (0\w{32})/);
				var hash = matches ? matches[1] : crypto.createHash("sha256").update(text, "utf8").digest("base64");
				if (hash === prev_bugreport_hash)
					return console.log("ignoring known bug report");
				prev_bugreport_hash = hash;
				if (conf.ignoreBugreportRegexp && new RegExp(conf.ignoreBugreportRegexp).test(text))
					return console.log('ignoring bugreport');
				mail.sendBugEmail(body.message, body.exception);
			}
			catch (e) {
				console.log("failed to process bugreport", e);
			}
			break;
			
		case 'joint':
			var objJoint = body;
			if (!objJoint || !objJoint.unit || !objJoint.unit.unit)
				return sendError(ws, 'no unit');
			if ("unsigned" in objJoint)
				return sendError(ws, 'unsigned unit');
			if (typeof objJoint.unit.unit !== 'string' || !ValidationUtils.isValidBase64(objJoint.unit.unit, constants.HASH_LENGTH))
				return sendError(ws, 'invalid unit');
			if (objJoint.ball && !storage.isGenesisUnit(objJoint.unit.unit))
				return sendError(ws, 'only requested joint can contain a ball');
			if (conf.bLight && !ws.bLightVendor)
				return sendError(ws, "I'm a light client and you are not my vendor");
			db.query("SELECT 1 FROM archived_joints WHERE unit=? AND reason='uncovered'", [objJoint.unit.unit], function (rows) {
				if (rows.length > 0) // ignore it as long is it was unsolicited
					return sendError(ws, "this unit is already known and archived");
				if (isTooDeeplyNestedOrHasTooManyNodes(objJoint))
					return sendError(ws, "joint is too deeply nested or has too many nodes");
				if (conf.bLight && objJoint.unit.authors && arrTempWatchedAddresses.length > 0) {
					if (!ValidationUtils.isNonemptyArray(objJoint.unit.authors) || !objJoint.unit.authors.every(ValidationUtils.isNonemptyObject))
						return sendError(ws, "invalid authors");
					var author_addresses = objJoint.unit.authors.map(a => a.address);
					if (_.intersection(author_addresses, arrTempWatchedAddresses).length > 0) {
						console.log('new joint from a temporarily watched address', author_addresses);
						return eventBus.emit('new_joint', objJoint);
					}
				}
				// light clients accept the joint without proof, it'll be saved as unconfirmed (non-stable)
				return conf.bLight ? handleLightOnlineJoint(ws, objJoint) : handleOnlineJoint(ws, objJoint);
			});
			break;
			
		case 'free_joints_end':
		case 'result':
		case 'info':
		case 'error':
			break;
			
		case 'private_payment':
			return sendError(ws, `direct sending of private payments disabled, use chat instead`);
			if (!body)
				return;
			var arrPrivateElements = body;
			handleOnlinePrivatePayment(ws, arrPrivateElements, false, {
				ifError: function(error){
					sendError(ws, error);
				},
				ifAccepted: function(unit){
					sendResult(ws, {private_payment_in_unit: unit, result: 'accepted'});
					eventBus.emit("new_direct_private_chains", [arrPrivateElements]);
				},
				ifValidationError: function(unit, error){
					sendResult(ws, {private_payment_in_unit: unit, result: 'error', error: error});
				},
				ifQueued: function(){
				}
			});
			break;
			
		case 'my_url':
			if (!ValidationUtils.isNonemptyString(body))
				return;
			var url = body;
			if (ws.bOutbound) // ignore: if you are outbound, I already know your url
				break;
			// inbound only
			if (ws.bAdvertisedOwnUrl) // allow it only once per connection
				break;
			ws.bAdvertisedOwnUrl = true;
			if (!isValidWsUrl(url)) {
				console.log("ignoring peer's my_url " + url + " because of incompatible ws protocol");
				break;
			}
			ws.claimed_url = url;
			db.query("SELECT creation_date AS latest_url_change_date, url FROM peer_host_urls WHERE peer_host=? ORDER BY creation_date DESC LIMIT 1", [ws.host], function(rows){
				var latest_change = rows[0];
				if (latest_change && latest_change.url === url) // advertises the same url
					return;
				//var elapsed_time = Date.now() - Date.parse(latest_change.latest_url_change_date);
				//if (elapsed_time < 24*3600*1000) // change allowed no more often than once per day
				//    return;
				
				// verify it is really your url by connecting to this url, sending a random string through this new connection, 
				// and expecting this same string over existing inbound connection
				if (assocConnectingOutboundWebsockets[url])
					return console.log("ignoring my_url of a known peer we are already connecting to:", url);
				const arrExistingOutboundPeers = arrOutboundPeers.slice();
				findOutboundPeerOrConnect(url, function(err, reverse_ws){
					if (err)
						return console.log("failed to connect to peer's my_url", url, err);
					if (arrExistingOutboundPeers.includes(reverse_ws))
						return console.log("ignoring my_url of a known peer", url);
					db.query("INSERT " + db.getIgnore() + " INTO peer_hosts (peer_host) VALUES (?)", [reverse_ws.host]); // writeEvent() needs it
					ws.sent_echo_string = crypto.randomBytes(30).toString("base64");
					sendJustsaying(reverse_ws, 'want_echo', ws.sent_echo_string); // we expect it to be echoed back over the original connection ws
					setTimeout(() => {
						if (ws.sent_echo_string) {
							console.log(`echo not received in time for my_url ${url} received from ${ws.host}, closing connection`);
							reverse_ws.close(1000, "echo not received in time");
						}
					}, 10000); // 10 seconds timeout for echo response
				}, true);
			});
			break;
			
		case 'want_echo':
			var echo_string = body;
			if (ws.bOutbound || !echo_string) // ignore
				break;
			// inbound only
			if (!ws.claimed_url)
				break;
			var reverse_ws = getOutboundPeerWsByUrl(ws.claimed_url);
			if (!reverse_ws) // no reverse outbound connection
				break;
			sendJustsaying(reverse_ws, 'your_echo', echo_string);
			break;
			
		case 'your_echo': // comes on the same ws as my_url, claimed_url is already set
			var echo_string = body;
			if (ws.bOutbound || !echo_string) // ignore
				break;
			// inbound only
			if (!ws.claimed_url)
				break;
			if (ws.sent_echo_string !== echo_string)
				break;
			var outbound_host = getHostByPeer(ws.claimed_url);
			var arrQueries = [];
			db.addQuery(arrQueries, "INSERT "+db.getIgnore()+" INTO peer_hosts (peer_host) VALUES (?)", [outbound_host]);
			db.addQuery(arrQueries, "INSERT "+db.getIgnore()+" INTO peers (peer_host, peer, learnt_from_peer_host) VALUES (?,?,?)", 
				[outbound_host, ws.claimed_url, ws.host]);
			db.addQuery(arrQueries, "UPDATE peer_host_urls SET is_active=NULL, revocation_date="+db.getNow()+" WHERE peer_host=?", [ws.host]);
			db.addQuery(arrQueries, "INSERT INTO peer_host_urls (peer_host, url) VALUES (?,?)", [ws.host, ws.claimed_url]);
			async.series(arrQueries);
			ws.sent_echo_string = null;
			break;
			
			
		// I'm a hub, the peer wants to authenticate
		case 'hub/login':
			if (!body)
				return;
			if (!conf.bServeAsHub)
				return sendError(ws, "I'm not a hub");
			var objLogin = body;
			if (!ws.challenge || objLogin.challenge !== ws.challenge)
				return sendError(ws, "wrong challenge");
			if (!objLogin.pubkey || !objLogin.signature)
				return sendError(ws, "no login params");
			if (!ValidationUtils.isStringOfLength(objLogin.pubkey, constants.PUBKEY_LENGTH))
				return sendError(ws, "wrong pubkey length");
			if (!ValidationUtils.isStringOfLength(objLogin.signature, constants.SIG_LENGTH))
				return sendError(ws, "wrong signature length");
			if (objLogin.max_message_length && !ValidationUtils.isPositiveInteger(objLogin.max_message_length))
				return sendError(ws, "max_message_length must be an integer");
			if (objLogin.max_message_count && (!ValidationUtils.isPositiveInteger(objLogin.max_message_count) || objLogin.max_message_count > 100))
				return sendError(ws, "max_message_count must be an integer > 0 and <= 100");
			if (ValidationUtils.hasFieldsExcept(objLogin, ["challenge", "pubkey", "signature", "max_message_length", "max_message_count"]))
				return sendError(ws, "unexpected fields in login object");
			try {
				if (!ecdsaSig.verify(objectHash.getDeviceMessageHashToSign(objLogin), objLogin.signature, objLogin.pubkey))
					return sendError(ws, "wrong signature");
			}
			catch (e) {
				return sendError(ws, "message hash failed: " + e.toString());
			}
			ws.device_address = objectHash.getDeviceAddress(objLogin.pubkey);
			ws.max_message_length = objLogin.max_message_length;
			ws.max_message_count = objLogin.max_message_count;
			// after this point the device is authenticated and can send further commands

			// temp fix for password-protected light clients: send sys vars (again) as they were ignored in the history response
			if (!ws.bSubscribed)
				sendSysVars(ws);

			var finishLogin = function(){
				ws.bLoginComplete = true;
				if (ws.onLoginComplete){
					ws.onLoginComplete();
					delete ws.onLoginComplete;
				}
			};
			db.query("SELECT 1 FROM devices WHERE device_address=?", [ws.device_address], function(rows){
				if (rows.length === 0)
					db.query("INSERT "+db.getIgnore()+" INTO devices (device_address, pubkey) VALUES (?,?)", [ws.device_address, objLogin.pubkey], function(){
						sendInfo(ws, "address created");
						finishLogin();
					});
				else {
					if (!ws.blockChat)
						sendStoredDeviceMessages(ws, ws.device_address);
					finishLogin();
				}
			});
			sendJustsaying(ws, 'hub/push_project_number', {projectNumber: (conf.pushApiProjectNumber && conf.pushApiKey ? conf.pushApiProjectNumber : 0), hasKeyId: !!conf.keyId});
			eventBus.emit('client_logged_in', ws);
			break;
			
		// I'm a hub, the peer wants to download new messages
		case 'hub/refresh':
			if (!conf.bServeAsHub)
				return sendError(ws, "I'm not a hub");
			if (!ws.device_address)
				return sendError(ws, "please log in first");
			if (ws.blockChat)
				return sendError(ws, "chat is blocked, please upgrade");
			sendStoredDeviceMessages(ws, ws.device_address);
			break;
			
		// I'm a hub, the peer wants to remove a message that he's just handled
		case 'hub/delete':
			if (!conf.bServeAsHub)
				return sendError(ws, "I'm not a hub");
			var message_hash = body;
			if (!message_hash)
				return sendError(ws, "no message hash");
			if (typeof message_hash !== 'string' || !ValidationUtils.isValidBase64(message_hash, constants.HASH_LENGTH))
				return sendError(ws, "invalid message hash");
			if (!ws.device_address)
				return sendError(ws, "please log in first");
			db.query("DELETE FROM device_messages WHERE device_address=? AND message_hash=?", [ws.device_address, message_hash], function(){
				sendInfo(ws, "deleted message "+message_hash);
			});
			break;
			
		// I'm a hub, the peer wants update settings for a correspondent device
		case 'hub/update_correspondent_settings':
			if (!body)
				return;
			if (!conf.bServeAsHub)
				return sendError(ws, "I'm not a hub");
			if (!ws.device_address)
				return sendError(ws, "please log in first");
			if (body.push_enabled !== 0 && body.push_enabled !== 1)
				return sendError(ws, "invalid push_enabled");
			if (!ValidationUtils.isValidDeviceAddress(body.correspondent_address))
				return sendError(ws, "invalid correspondent_address");
			db.query(
				"INSERT "+db.getIgnore()+" INTO correspondent_settings (device_address, correspondent_address, push_enabled) VALUES(?,?,?)",
				[ws.device_address, body.correspondent_address, body.push_enabled],
				function(res){
					if (res.affectedRows === 0)
						db.query("UPDATE correspondent_settings SET push_enabled=? WHERE device_address=? AND correspondent_address=?", [body.push_enabled, ws.device_address, body.correspondent_address]);
					sendInfo(ws, "updated push "+body.push_enabled);
				}
			);
			break;
			
		// I'm a hub, the peer wants to get the challenge again
		case 'hub/repeat_challenge':
			if (!conf.bServeAsHub)
				return sendError(ws, "I'm not a hub");
			if (!ws.challenge)
				return sendError(ws, "No challenge yet");
			sendJustsaying(ws, 'hub/challenge', ws.challenge);
			break;
		
		// I'm connected to a hub
		case 'hub/challenge':
		case 'hub/message':
		case 'hub/message_box_status':
			if (!body)
				return;
			if (['hub/message', 'hub/message_box_status'].includes(subject) && !ws.bLoggingIn && !ws.bLoggedIn) // accept from hub only
				return;
			eventBus.emit("message_from_hub", ws, subject, body);
			break;
			
		// I'm light client
		case 'light/have_updates':
		case 'light/sequence_became_bad':
		case 'light/aa_request':
		case 'light/aa_definition':
		case 'light/aa_response':
		case 'light/aa_definition_saved':
		case 'system_var_vote':
			if (!conf.bLight)
				return sendError(ws, "I'm not light");
			if (!ws.bLightVendor)
				return sendError(ws, "You are not my light vendor");
			eventBus.emit("message_for_light", ws, subject, body);
			break;
			
		// I'm light vendor
		case 'light/new_address_to_watch':
			if (conf.bLight)
				return sendError(ws, "I'm light myself, can't serve you");
			if (ws.bOutbound)
				return sendError(ws, "light clients have to be inbound");
			var address = body;
			if (!ValidationUtils.isValidAddress(address))
				return sendError(ws, "address not valid");
			bWatchingForLight = true;
			db.query("INSERT "+db.getIgnore()+" INTO watched_light_addresses (peer, address) VALUES (?,?)", [ws.peer, address], function(){
				sendInfo(ws, "now watching "+address);
				// check if we already have something on this address
				/*
				db.query(
					"SELECT unit, is_stable FROM unit_authors JOIN units USING(unit) WHERE address=? \n\
					UNION \n\
					SELECT unit, is_stable FROM outputs JOIN units USING(unit) WHERE address=? \n\
					ORDER BY is_stable LIMIT 10", 
					[address, address], 
					function(rows){
						if (rows.length === 0)
							return;
						if (rows.length === 10 || rows.some(function(row){ return row.is_stable; }))
							sendJustsaying(ws, 'light/have_updates');
						rows.forEach(function(row){
							if (row.is_stable)
								return;
							storage.readJoint(db, row.unit, {
								ifFound: function(objJoint){
									sendJoint(ws, objJoint);
								},
								ifNotFound: function(){
									throw Error("watched unit "+row.unit+" not found");
								}
							});
						});
					}
				);
				*/
			});            
			break;
			
		case 'light/new_aa_to_watch':
			if (!body)
				return;
			if (conf.bLight)
				return sendError(ws, "I'm light myself, can't serve you");
			if (ws.bOutbound)
				return sendError(ws, "light clients have to be inbound");
			if (!ValidationUtils.isValidAddress(body.aa))
				return sendError(ws, "invalid AA");
			if (body.address && !ValidationUtils.isValidAddress(body.address))
				return sendError(ws, "invalid address");
			storage.readAADefinition(db, body.aa, null, function (arrDefinition) {
				if (!arrDefinition) {
					arrDefinition = storage.getUnconfirmedAADefinition(body.aa);
					if (!arrDefinition)
						return sendError(ws, "not an AA: " + body.aa);
				}
				bWatchingForLight = true;
				db.query("INSERT " + db.getIgnore() + " INTO watched_light_aas (peer, aa, address) VALUES (?,?,?)", [ws.peer, body.aa, body.address || ''], function () {
					sendInfo(ws, "now watching AA " + body.aa + " address " + (body.address || 'all'));
				});
			});
			break;
			
		case 'watch_system_vars':
			ws.bWatchingSystemVars = true;
			sendSysVars(ws);
			break;
			
		case 'exchange_rates':
			if (!ws.bLoggingIn && !ws.bLoggedIn) // accept from hub only
				return;
			_.assign(exchangeRates, body);
			eventBus.emit('rates_updated');
			break;
			
		case 'known_witnesses':
			if (!ws.bLoggingIn && !ws.bLoggedIn) // accept from hub only
				return console.log('ignoring known_witnesses from non-hub');
			_.assign(knownWitnesses, body);
			eventBus.emit('known_witnesses_updated');
			break;
			
		case 'system_vars':
			if (!ws.bLoggingIn && !ws.bLoggedIn && !ws.bLightVendor || !conf.bLight) // accept from hub or light vendor only and only if light
				return;
			_.assign(storage.systemVars, body);
			eventBus.emit("message_for_light", ws, subject, body);
			break;
			
		case 'upgrade_required':
			if (!ws.bLoggingIn && !ws.bLoggedIn) // accept from hub only
				return;
			ws.close(1000, "my core is old");
			throw Error("Mandatory upgrade required, please check the release notes at https://github.com/byteball/obyte-gui-wallet/releases and upgrade.");
			break;
			
		case 'custom':
			eventBus.emit('custom_justsaying', ws, body);
			break;
	}
}

function handleRequest(ws, tag, command, params){
	if (!ValidationUtils.isNonemptyString(command))
		return console.log("no command");
	if (!ValidationUtils.isNonemptyString(tag))
		return console.log("invalid tag"); // can be even an object
	if (tag.length > constants.HASH_LENGTH)
		return sendErrorResponse(ws, tag, "tag too long");
	if (ws.assocCommandsInPreparingResponse[tag]) // ignore repeated request while still preparing response to a previous identical request
		return console.log("ignoring identical "+command+" request");
	if (peerIsBlocked(ws.host)) // requests from the hub's webserver
		return sendErrorResponse(ws, tag, "you are blocked");
	ws.assocCommandsInPreparingResponse[tag] = command;
	incrementPending(requestsInWork, ws.host);
	if (command.startsWith('light/')) {
		if (conf.bLight)
			return sendErrorResponse(ws, tag, "I'm light myself, can't serve you");
		if (ws.bOutbound)
			return sendErrorResponse(ws, tag, "light clients have to be inbound");
	}
	else if (command.startsWith('hub/')) {
		if (!conf.bServeAsHub)
			return sendErrorResponse(ws, tag, "I'm not a hub, can't serve hub/ requests");
	}
	switch (command){
		case 'heartbeat':
			ws.bSleeping = false; // the peer is sending heartbeats, therefore he is awake
			
			// true if our timers were paused
			// Happens only on android, which suspends timers when the app becomes paused but still keeps network connections
			// Handling 'pause' event would've been more straightforward but with preference KeepRunning=false, the event is delayed till resume
			var bPaused = (bCordova && Date.now() - last_hearbeat_wake_ts > HEARTBEAT_PAUSE_TIMEOUT);
			if (bPaused)
				return sendResponse(ws, tag, 'sleep'); // opt out of receiving heartbeats and move the connection into a sleeping state
			sendResponse(ws, tag);
			break;
			
		case 'subscribe':
			if (!ValidationUtils.isNonemptyObject(params))
				return sendErrorResponse(ws, tag, 'no params');
			var subscription_id = params.subscription_id;
			if (typeof subscription_id !== 'string')
				return sendErrorResponse(ws, tag, 'no subscription_id');
			if ([...wss.clients].concat(arrOutboundPeers).some(function(other_ws) { return (other_ws.subscription_id === subscription_id); })){
				if (ws.bOutbound)
					db.query("UPDATE peers SET is_self=1 WHERE peer=?", [ws.peer]);
				sendErrorResponse(ws, tag, "self-connect");
				return ws.close(1000, "self-connect");
			}
			if (conf.bLight){
				//if (ws.peer === exports.light_vendor_url)
				//    sendFreeJoints(ws);
				return sendErrorResponse(ws, tag, "I'm light, cannot subscribe you to updates");
			}
			if (typeof params.library_version !== 'string') {
				sendErrorResponse(ws, tag, "invalid library_version");
				return ws.close(1000, "invalid library_version");
			}
			if (version2int(params.library_version) < version2int(constants.minCoreVersionForFullNodes))
				ws.old_core = true;
			if (ws.old_core){ // can be also set in 'version'
				sendErrorResponse(ws, tag, "old core (full)");
				return disconnectOldVersion(ws, "old core (full)");
			}
			ws.bSubscribed = true;
			sendResponse(ws, tag, "subscribed");
			if (bCatchingUp)
				return;
			if (ValidationUtils.isNonnegativeInteger(params.last_mci))
				sendJointsSinceMci(ws, params.last_mci);
			else
				sendFreeJoints(ws);
			break;
			
		case 'get_joint': // peer needs a specific joint
			//if (bCatchingUp)
			//    return;
			if (ws.old_core)
				return sendErrorResponse(ws, tag, "old core, will not serve get_joint");
			var unit = params;
			if (!ValidationUtils.isNonemptyString(unit))
				return sendErrorResponse(ws, tag, "unit must be a string");
			const objCachedPrunedJoint = archiving.getCachedPrunedJoint(unit);
			if (objCachedPrunedJoint)
				return sendJoint(ws, objCachedPrunedJoint, tag);
			storage.readJoint(db, unit, {
				ifFound: function(objJoint){
					// make the peer go a bit deeper into stable units and request catchup only when and if it reaches min retrievable and we can deliver a catchup
					if (objJoint.ball && objJoint.unit.main_chain_index > storage.getMinRetrievableMci()) {
						delete objJoint.ball;
						delete objJoint.skiplist_units;
					}
					sendJoint(ws, objJoint, tag);
				},
				ifNotFound: function(){
					sendResponse(ws, tag, {joint_not_found: unit});
				}
			});
			break;
			
		case 'post_joint': // only light clients use this command to post joints they created
			var objJoint = params;
			if (!objJoint || !objJoint.unit || !objJoint.unit.unit)
				return sendErrorResponse(ws, tag, 'no unit');
			if (objJoint.ball)
				return sendErrorResponse(ws, tag, 'no ball via post_joint');
			if (bCatchingUp)
				return sendErrorResponse(ws, tag, "catching up, will not handle posted joint");
			if (isTooDeeplyNestedOrHasTooManyNodes(objJoint))
				return sendErrorResponse(ws, tag, "joint is too deeply nested or has too many nodes");
			handlePostedJoint(ws, objJoint, function(error){
				error ? sendErrorResponse(ws, tag, error) : sendResponse(ws, tag, 'accepted');
			});
			break;
			
		case 'catchup':
			if (!ws.bSubscribed)
				return sendErrorResponse(ws, tag, "not subscribed, will not serve catchup");
			var catchupRequest = params;
			mutex.lock(['catchup_request'], function(unlock){
				if (!ws || ws.readyState !== ws.OPEN) // may be already gone when we receive the lock
					return process.nextTick(unlock);
				catchup.prepareCatchupChain(catchupRequest, {
					ifError: function(error){
						sendErrorResponse(ws, tag, error);
						unlock();
					},
					ifOk: function(objCatchupChain){
						sendResponse(ws, tag, objCatchupChain);
						unlock();
					}
				});
			});
			break;
			
		case 'get_hash_tree':
			if (!ws.bSubscribed)
				return sendErrorResponse(ws, tag, "not subscribed, will not serve get_hash_tree");
			var hashTreeRequest = params;
			mutex.lock(['get_hash_tree_request'], function(unlock){
				if (!ws || ws.readyState !== ws.OPEN) // may be already gone when we receive the lock
					return process.nextTick(unlock);
				catchup.readHashTree(hashTreeRequest, {
					ifError: function(error){
						sendErrorResponse(ws, tag, error);
						unlock();
					},
					ifOk: function(arrBalls){
						// we have to wrap arrBalls into an object because the peer will check .error property first
						sendResponse(ws, tag, {balls: arrBalls});
						unlock();
					}
				});
			});
			break;
			
		case 'get_peers':
			var arrPeerUrls = arrOutboundPeers.filter(function(ws){ return (ws.host !== 'byteball.org' && ws.readyState === ws.OPEN && ws.bSubscribed && ws.bSource); }).map(function(ws){ return ws.peer; });
			if (ws.dontSharePeers)
				arrPeerUrls = [];
			// empty array is ok
			sendResponse(ws, tag, arrPeerUrls);
			break;
			
		case 'get_witnesses':
			myWitnesses.readMyWitnesses(function(arrWitnesses){
				sendResponse(ws, tag, arrWitnesses);
			}, 'wait');
			break;
			
		case 'get_last_mci':
			storage.readLastMainChainIndex(function(last_mci){
				sendResponse(ws, tag, last_mci);
			});
			break;
			
		case 'get_last_stable_unit_props':
			storage.readLastStableMcUnitProps(db, function(props){
				sendResponse(ws, tag, {
					unit: props.unit,
					main_chain_index: props.main_chain_index,
					timestamp: props.timestamp,
				});
			});
			break;
			
		case 'get_system_vars':
			sendResponse(ws, tag, storage.systemVars);
			break;
			
		case 'get_system_var_votes':
			let votes = {
				op_list: [],
				threshold_size: [],
				base_tps_fee: [],
				tps_interval: [],
				tps_fee_multiplier: [],			
			};
			let assocAddresses = {}
			let is_stable = 1;
			db.query("SELECT * FROM op_votes ORDER BY address, op_address", op_rows => {
				let prev_address, vote;
				for (let { address, op_address, unit, timestamp } of op_rows) {
					if (address !== prev_address) {
						vote = { address, unit, timestamp, value: [], is_stable };
						votes.op_list.push(vote);
						assocAddresses[address] = true;
						prev_address = address;
					}
					vote.value.push(op_address);
				}
				db.query("SELECT * FROM numerical_votes", n_rows => {
					for (let { subject, address, unit, value, timestamp } of n_rows) {
						votes[subject].push({ address, unit, timestamp, value, is_stable });
						assocAddresses[address] = true;
					}

					// unconfirmed votes
					is_stable = 0;
					for (let unit in storage.assocUnstableMessages) { // undefined order of iteration, we might handle unstable votes from the same address in the wrong order
						const arrUnstableMessages = storage.assocUnstableMessages[unit];
						for (let message of arrUnstableMessages) {
							if (message.app !== 'system_vote')
								continue;
							const { subject, value } = message.payload;
							const { author_addresses, timestamp, sequence } = storage.assocUnstableUnits[unit];
							if (sequence !== 'good')
								continue;
							for (let address of author_addresses) {
								const prev_vote = votes[subject].find(vote => vote.address === address);
								if (prev_vote) {
									prev_vote.value = value;
									prev_vote.timestamp = timestamp;
									prev_vote.unit = unit;
									prev_vote.is_stable = 0;
								}
								else {
									votes[subject].push({ address, unit, timestamp, value, is_stable });
									assocAddresses[address] = true;
								}
							}
						}
					}
					// read the balances of all voting addresses
					let balances = {};
					const arrAddresses = Object.keys(assocAddresses);
					const strAddresses = arrAddresses.map(db.escape).join(', ');
					db.query(`SELECT address, SUM(amount) AS balance 
						FROM outputs
						LEFT JOIN units USING(unit)
						WHERE address IN(${strAddresses}) AND is_spent=0 AND asset IS NULL AND sequence='good' 
						GROUP BY address`,
						bal_rows => {
							for (let { address, balance } of bal_rows) {
								balances[address] = balance;
							}
							sendResponse(ws, tag, { votes, balances });
						}
					);
				});
			});
			break;
			
		// I'm a hub, the peer wants to deliver a message to one of my clients
		case 'hub/deliver':
			var objDeviceMessage = params;
			if (!objDeviceMessage || !objDeviceMessage.signature || !objDeviceMessage.pubkey || !objDeviceMessage.to
					|| !objDeviceMessage.encrypted_package || !objDeviceMessage.encrypted_package.dh
					|| !objDeviceMessage.encrypted_package.dh.sender_ephemeral_pubkey 
					|| !objDeviceMessage.encrypted_package.dh.recipient_ephemeral_pubkey 
					|| !objDeviceMessage.encrypted_package.encrypted_message
					|| !objDeviceMessage.encrypted_package.iv || !objDeviceMessage.encrypted_package.authtag)
				return sendErrorResponse(ws, tag, "missing fields");
			if (!ValidationUtils.isValidDeviceAddress(objDeviceMessage.to))
				return sendErrorResponse(ws, tag, "invalid to address");
			if (isTooDeeplyNestedOrHasTooManyNodes(objDeviceMessage, 5, 100))
				return sendErrorResponse(ws, tag, "device message is too deeply nested or has too many nodes");
			if (ValidationUtils.hasFieldsExcept(objDeviceMessage, ["signature", "pubkey", "to", "encrypted_package"]))
				return sendErrorResponse(ws, tag, "unexpected fields in device message");
			var bToMe = (my_device_address && my_device_address === objDeviceMessage.to);
			if (!conf.bServeAsHub && !bToMe)
				return sendErrorResponse(ws, tag, "I'm not a hub");
			try {
				if (!ecdsaSig.verify(objectHash.getDeviceMessageHashToSign(objDeviceMessage), objDeviceMessage.signature, objDeviceMessage.pubkey))
					return sendErrorResponse(ws, tag, "wrong message signature");
				var message_hash = objectHash.getBase64Hash(objDeviceMessage);
			}
			catch (e) {
				return sendErrorResponse(ws, tag, "message hash failed: " + e.toString());
			}
			
			// if i'm always online and i'm my own hub
			if (bToMe){
				sendResponse(ws, tag, "accepted");
				eventBus.emit("message_from_hub", ws, 'hub/message', {
					message_hash,
					message: objDeviceMessage
				});
				return;
			}
			
			db.query("SELECT pubkey, temp_pubkey_package FROM devices WHERE device_address=?", [objDeviceMessage.to], function(rows){
				if (rows.length === 0)
					return sendErrorResponse(ws, tag, "address "+objDeviceMessage.to+" not registered here");
				let keys = [rows[0].pubkey];
				if (rows[0].temp_pubkey_package) {
					var objTempPubkey = JSON.parse(rows[0].temp_pubkey_package);
					keys.push(objTempPubkey.temp_pubkey);
				}
				if (!keys.includes(objDeviceMessage.encrypted_package.dh.recipient_ephemeral_pubkey))
					return sendErrorResponse(ws, tag, "wrong recipient ephemeral pubkey");
				var message_string = JSON.stringify(objDeviceMessage);
				db.query(
					"INSERT "+db.getIgnore()+" INTO device_messages (message_hash, message, device_address) VALUES (?,?,?)", 
					[message_hash, message_string, objDeviceMessage.to],
					function(){
						// if the addressee is connected, deliver immediately
						[...wss.clients].concat(arrOutboundPeers).forEach(function(client){
							if (client.device_address === objDeviceMessage.to && (!client.max_message_length || message_string.length <= client.max_message_length) && !client.blockChat) {
								sendJustsaying(client, 'hub/message', {
									message_hash: message_hash,
									message: objDeviceMessage
								});
							}
						});
						sendResponse(ws, tag, "accepted");
						var sender_device_address = objectHash.getDeviceAddress(objDeviceMessage.pubkey);
						db.query(
							"SELECT push_enabled FROM correspondent_settings WHERE device_address=? AND correspondent_address=?",
							[objDeviceMessage.to, sender_device_address],
							function(rows){
								if (rows.length === 0 || rows[0].push_enabled === 1)
									eventBus.emit('peer_sent_new_message', ws, objDeviceMessage);
							}
						);
					}
				);
			});
			break;
			
		// I'm a hub, the peer wants to get a correspondent's temporary pubkey
		case 'hub/get_temp_pubkey':
			var permanent_pubkey = params;
			if (!ValidationUtils.isStringOfLength(permanent_pubkey, constants.PUBKEY_LENGTH))
				return sendErrorResponse(ws, tag, "wrong permanent_pubkey length");
			var device_address = objectHash.getDeviceAddress(permanent_pubkey);
			if (device_address === my_device_address) // to me
				return sendResponse(ws, tag, objMyTempPubkeyPackage); // this package signs my permanent key
			if (!conf.bServeAsHub)
				return sendErrorResponse(ws, tag, "I'm not a hub");
			db.query("SELECT temp_pubkey_package FROM devices WHERE device_address=?", [device_address], function(rows){
				if (rows.length === 0)
					return sendErrorResponse(ws, tag, "device with this pubkey is not registered here");
				if (!rows[0].temp_pubkey_package)
					return sendErrorResponse(ws, tag, "temp pub key not set yet");
				var objTempPubkey = JSON.parse(rows[0].temp_pubkey_package);
				sendResponse(ws, tag, objTempPubkey);
			});
			break;
			
		// I'm a hub, the peer wants to update its temporary pubkey
		case 'hub/temp_pubkey':
			if (!conf.bServeAsHub)
				return sendErrorResponse(ws, tag, "I'm not a hub");
			if (!ws.device_address)
				return sendErrorResponse(ws, tag, "please log in first");
			var objTempPubkey = params;
			if (!objTempPubkey || !objTempPubkey.temp_pubkey || !objTempPubkey.pubkey || !objTempPubkey.signature)
				return sendErrorResponse(ws, tag, "no temp_pubkey params");
			if (ValidationUtils.hasFieldsExcept(objTempPubkey, ["temp_pubkey", "pubkey", "signature"]))
				return sendErrorResponse(ws, tag, "unexpected fields in temp_pubkey package");
			if (!ValidationUtils.isStringOfLength(objTempPubkey.temp_pubkey, constants.PUBKEY_LENGTH))
				return sendErrorResponse(ws, tag, "wrong temp_pubkey length");
			if (!ValidationUtils.isStringOfLength(objTempPubkey.pubkey, constants.PUBKEY_LENGTH))
				return sendErrorResponse(ws, tag, "wrong pubkey length");
			if (objectHash.getDeviceAddress(objTempPubkey.pubkey) !== ws.device_address)
				return sendErrorResponse(ws, tag, "signed by another pubkey");
			try {
				if (!ecdsaSig.verify(objectHash.getDeviceMessageHashToSign(objTempPubkey), objTempPubkey.signature, objTempPubkey.pubkey))
					return sendErrorResponse(ws, tag, "wrong signature");
			}
			catch (e) {
				return sendErrorResponse(ws, tag, "message hash failed: " + e.toString());
			}
			var fnUpdate = function(onDone){
				db.query("UPDATE devices SET temp_pubkey_package=? WHERE device_address=?", [JSON.stringify(objTempPubkey), ws.device_address], function(){
					if (onDone)
						onDone();
				});
			};
			fnUpdate(function(){
				sendResponse(ws, tag, "updated");
			});
			if (!ws.bLoginComplete)
				ws.onLoginComplete = fnUpdate;
			break;
			
		case 'light/get_history':
			if (largeHistoryTags[tag])
				return sendErrorResponse(ws, tag, constants.lightHistoryTooLargeErrorMessage);
			if (!ws.bSentSysVars) {
				ws.bSentSysVars = true;
				sendSysVars(ws);
			}
			mutex.lock(['get_history_request'], function(unlock){
				if (!ws || ws.readyState !== ws.OPEN) // may be already gone when we receive the lock
					return process.nextTick(unlock);
				light.prepareHistory(params, {
					ifError: function(err){
						if (err === constants.lightHistoryTooLargeErrorMessage)
							largeHistoryTags[tag] = true;
						sendErrorResponse(ws, tag, err);
						unlock();
					},
					ifOk: function(objResponse){
						sendResponse(ws, tag, objResponse);
						bWatchingForLight = true;
						if (params.addresses)
							db.query(
								"INSERT "+db.getIgnore()+" INTO watched_light_addresses (peer, address) VALUES "+
								params.addresses.map(function(address){ return "("+db.escape(ws.peer)+", "+db.escape(address)+")"; }).join(", ")
							);
						if (params.requested_joints && ValidationUtils.isNonemptyArray(params.requested_joints)) {
							storage.sliceAndExecuteQuery("SELECT unit FROM units WHERE main_chain_index >= ? AND unit IN(?)",
								[storage.getMinRetrievableMci(), params.requested_joints], params.requested_joints, function(rows) {
								if(rows.length) {
									db.query(
										"INSERT " + db.getIgnore() + " INTO watched_light_units (peer, unit) VALUES " +
										rows.map(function(row) {
											return "(" + db.escape(ws.peer) + ", " + db.escape(row.unit) + ")";
										}).join(", ")
									);
								}
							});
						}
						//db.query("INSERT "+db.getIgnore()+" INTO light_peer_witnesses (peer, witness_address) VALUES "+
						//    params.witnesses.map(function(address){ return "("+db.escape(ws.peer)+", "+db.escape(address)+")"; }).join(", "));
						unlock();
					}
				});
			});
			break;
			
		case 'light/get_link_proofs':
			mutex.lock(['get_link_proofs_request'], function(unlock){
				if (!ws || ws.readyState !== ws.OPEN) // may be already gone when we receive the lock
					return process.nextTick(unlock);
				light.prepareLinkProofs(params, {
					ifError: function(err){
						sendErrorResponse(ws, tag, err);
						unlock();
					},
					ifOk: function(objResponse){
						sendResponse(ws, tag, objResponse);
						unlock();
					}
				});
			});
			break;
			
		case 'light/get_parents_and_last_ball_and_witness_list_unit':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in get_parents_and_last_ball_and_witness_list_unit");
			var callbacks = {
				ifError: function(err){
					sendErrorResponse(ws, tag, err);
				},
				ifOk: function(objResponse){
					sendResponse(ws, tag, objResponse);
				}
			}
			if (params.witnesses)
				light.prepareParentsAndLastBallAndWitnessListUnit(params.witnesses, params.from_addresses, params.output_addresses, params.max_aa_responses||0, callbacks);
			else
				myWitnesses.readMyWitnesses(function(arrWitnesses){
					light.prepareParentsAndLastBallAndWitnessListUnit(arrWitnesses, params.from_addresses, params.output_addresses, params.max_aa_responses||0, callbacks);
				});
			break;

	   case 'light/get_attestation':
			// find an attestation posted by the given attestor and attesting field=value
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_attestation");
			if (!params.attestor_address || !params.field || !params.value)
				return sendErrorResponse(ws, tag, "missing params in light/get_attestation");
			if (!ValidationUtils.isValidAddress(params.attestor_address))
				return sendErrorResponse(ws, tag, "invalid attestor_address in light/get_attestation");
			if (typeof params.field !== 'string' || params.field.length > 50)
				return sendErrorResponse(ws, tag, "invalid field in light/get_attestation");
			if (typeof params.value !== 'string' || params.value.length > 100)
				return sendErrorResponse(ws, tag, "invalid value in light/get_attestation");
			var order = (conf.storage === 'sqlite') ? 'rowid' : 'creation_date';
			var join = (conf.storage === 'sqlite') ? '' : 'JOIN units USING(unit)';
			db.query(
				"SELECT unit FROM attested_fields "+join+" WHERE attestor_address=? AND field=? AND value=? ORDER BY "+order+" DESC LIMIT 1", 
				[params.attestor_address, params.field, params.value],
				function(rows){
					var attestation_unit = (rows.length > 0) ? rows[0].unit : "";
					sendResponse(ws, tag, attestation_unit);
				}
			);
			break;

	   case 'light/get_attestations':
			// get list of all attestations of an address
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_attestations");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "invalid address in light/get_attestations");
			var order = (conf.storage === 'sqlite') ? 'attestations.rowid' : 'creation_date';
			var join = (conf.storage === 'sqlite') ? '' : 'JOIN units USING(unit)';
			db.query(
				"SELECT unit, attestor_address, payload \n\
				FROM attestations CROSS JOIN messages USING(unit, message_index) "+join+" \n\
				WHERE address=? ORDER BY "+order, 
				[params.address],
				function(rows){
					var arrAttestations = rows.map(function(row){
						var payload = JSON.parse(row.payload);
						if (payload.address !== params.address)
							throw Error("not matching addresses, expected "+params.address+", got "+payload.address);
						return {unit: row.unit, attestor_address: row.attestor_address, profile: payload.profile};
					});
					sendResponse(ws, tag, arrAttestations);
				}
			);
			break;

		case 'light/pick_divisible_coins_for_amount':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/pick_divisible_coins_for_amount");
			if (!params.addresses || !params.last_ball_mci || !params.amount)
				return sendErrorResponse(ws, tag, "missing params in light/pick_divisible_coins_for_amount");
			if (params.asset && !ValidationUtils.isValidBase64(params.asset, constants.HASH_LENGTH))
				return sendErrorResponse(ws, tag, "asset is not valid");
			if (!ValidationUtils.isNonemptyArray(params.addresses))
				return sendErrorResponse(ws, tag, "addresses must be non-empty array");
			if (params.addresses.length > constants.MAX_AUTHORS_PER_UNIT)
				return sendErrorResponse(ws, tag, "too many addresses");
			if (!params.addresses.every(ValidationUtils.isValidAddress))
				return sendErrorResponse(ws, tag, "some addresses are not valid");
			if (!ValidationUtils.isPositiveInteger(params.last_ball_mci))
				return sendErrorResponse(ws, tag, "last_ball_mci is not valid");
			if (!ValidationUtils.isPositiveInteger(params.amount))
				return sendErrorResponse(ws, tag, "amount is not valid");
			if (params.amount > constants.MAX_CAP)
				return sendErrorResponse(ws, tag, "amount is too large");
			if (params.spend_unconfirmed && (params.spend_unconfirmed !== "own" && params.spend_unconfirmed !== "none" && params.spend_unconfirmed !== "all"))
				return sendErrorResponse(ws, tag, "spend_unconfirmed is not valid");
			var getAssetInfoOrNull = function(asset, cb){
				if (!asset)
					return cb(null, null);
				storage.readAssetInfo(db, asset, function(objAsset){
					if (!objAsset)
						return cb("asset " + asset + " not found", null);
					return cb(null, objAsset);
				});
			};
			getAssetInfoOrNull(params.asset, function(err, objAsset){
				if (err)
					return sendErrorResponse(ws, tag, err);
				var bMultiAuthored = (params.addresses.length > 1);
				let size = 2000;
				inputs.pickDivisibleCoinsForAmount(db, objAsset, params.addresses, params.last_ball_mci, params.amount + size, size, 0, bMultiAuthored, params.spend_unconfirmed || 'own', function(arrInputsWithProofs, total_amount) {
					var objResponse = {inputs_with_proofs: arrInputsWithProofs || [], total_amount: total_amount || 0};
					sendResponse(ws, tag, objResponse);
				});
			});
			break;

		case 'light/get_definition_chash':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_definition_chash");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "address not valid");
			if (params.max_mci && !ValidationUtils.isPositiveInteger(params.max_mci))
				return sendErrorResponse(ws, tag, "max_mci not a positive integer");
			storage.readDefinitionChashByAddress(db, params.address, params.max_mci, function(definition_chash){
				sendResponse(ws, tag, definition_chash);
			});
			break;
		
		case 'light/get_definition':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_definition");
			if (!ValidationUtils.isValidAddress(params))
				return sendErrorResponse(ws, tag, "address not valid");
			if (definitions[params])
				return sendResponse(ws, tag, definitions[params]);
			db.query("SELECT definition FROM definitions WHERE definition_chash=? UNION SELECT definition FROM aa_addresses WHERE address=? LIMIT 1", [params, params], function(rows){
				var arrDefinition = rows[0]
					? JSON.parse(rows[0].definition)
					: storage.getUnconfirmedAADefinition(params);
				if (arrDefinition) // save in cache
					definitions[params] = arrDefinition;
				sendResponse(ws, tag, arrDefinition);
			});
			break;

		case 'light/get_definition_for_address':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_definition_for_address");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "address not valid");

			db.query("SELECT definition_chash,is_stable FROM address_definition_changes CROSS JOIN units USING(unit) WHERE address=? AND sequence='good'\n\
			ORDER BY main_chain_index DESC, level DESC LIMIT 1",[params.address],function(address_definition_changes){
				if (address_definition_changes[0] && address_definition_changes[0].is_stable === 0){
					return sendResponse(ws, tag, {
						definition_chash: address_definition_changes[0].definition_chash,
						is_stable: false
					});
				}
				var definition_chash = address_definition_changes[0] ? address_definition_changes[0].definition_chash : params.address;
				db.query("SELECT definition,is_stable FROM definitions CROSS JOIN unit_authors USING(definition_chash) CROSS JOIN units USING(unit) \n\
					WHERE definition_chash=? AND sequence='good'",[definition_chash], function(definitions){
						if (definitions[0]){
							return sendResponse(ws, tag, {
								definition_chash: definition_chash,
								definition: JSON.parse(definitions[0].definition),
								is_stable: definitions[0].is_stable === 1
							});
						} else {
							return sendResponse(ws, tag, {
								definition_chash: definition_chash,
								is_stable: true
							});
						}
					})
			});
			break;

		case 'light/get_balances':
			var addresses = params;
			if (!addresses)
				return sendErrorResponse(ws, tag, "no params in light/get_balances");
			if (!ValidationUtils.isNonemptyArray(addresses))
				return sendErrorResponse(ws, tag, "addresses must be non-empty array");
			if (!addresses.every(ValidationUtils.isValidAddress))
				return sendErrorResponse(ws, tag, "some addresses are not valid");
			if (addresses.length > 100)
				return sendErrorResponse(ws, tag, "too many addresses");
			db.query(
				"SELECT address, asset, is_stable, SUM(CAST(amount AS DOUBLE)) AS balance, COUNT(*) AS outputs_count \n\
				FROM outputs JOIN units USING(unit) \n\
				WHERE is_spent=0 AND address IN(?) AND sequence='good' \n\
				GROUP BY address, asset, is_stable", [addresses], function(rows) {
					var balances = {};
					rows.forEach(function(row) {
						if (!balances[row.address])
							balances[row.address] = { base: { stable: 0, pending: 0, stable_outputs_count: 0, pending_outputs_count: 0}};
						if (row.asset && !balances[row.address][row.asset])
							balances[row.address][row.asset] = { stable: 0, pending: 0, stable_outputs_count: 0, pending_outputs_count: 0};
						balances[row.address][row.asset || 'base'][row.is_stable ? 'stable' : 'pending'] = row.balance;
						balances[row.address][row.asset || 'base'][row.is_stable ? 'stable_outputs_count' : 'pending_outputs_count'] = row.outputs_count;
					});
					for (var address in balances)
						for (var asset in balances[address])
							balances[address][asset].total = (balances[address][asset].stable || 0) + (balances[address][asset].pending || 0);
					sendResponse(ws, tag, balances);
				}
			);
			break;

		case 'light/get_profile_units':
			var addresses = params;
			if (!addresses)
				return sendErrorResponse(ws, tag, "no params in light/get_profiles_units");
			if (!ValidationUtils.isNonemptyArray(addresses))
				return sendErrorResponse(ws, tag, "addresses must be non-empty array");
			if (!addresses.every(ValidationUtils.isValidAddress))
				return sendErrorResponse(ws, tag, "some addresses are not valid");
			if (addresses.length > 100)
				return sendErrorResponse(ws, tag, "too many addresses");
			db.query(
				"SELECT unit FROM messages JOIN unit_authors USING(unit) \n\
				JOIN units USING(unit) WHERE app='profile' AND address IN(?) \n\
				ORDER BY main_chain_index ASC", [addresses], function(rows) {
					var units = rows.map(function(row) { return row.unit; });
					sendResponse(ws, tag, units);
				}
			);
			break;

		case 'light/get_data_feed':
			if (!ValidationUtils.isNonemptyObject(params))
				return sendErrorResponse(ws, tag, "no params in light/get_data_feed");
			if ("max_mci" in params && !ValidationUtils.isPositiveInteger(params.max_mci))
				return sendErrorResponse(ws, tag, "max_mci must be positive integer");
			dataFeeds.readDataFeedValueByParams(params, params.max_mci || 1e15, 'all_unstable', function (err, value) {
				if (err)
					return sendErrorResponse(ws, tag, err);
				sendResponse(ws, tag, value);
			});
			break;

		case 'light/dry_run_aa':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/dry_run_aa");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "address not valid");
		
			storage.readAADefinition(db, params.address, null, function (arrDefinition) {
				if (!arrDefinition)
					return sendErrorResponse(ws, tag, "not an AA");
				aa_composer.validateAATriggerObject(params.trigger, function(error){
					if (error)
						return sendErrorResponse(ws, tag, error);
					aa_composer.dryRunPrimaryAATrigger(params.trigger, params.address, arrDefinition, function (arrResponses) {
						if (constants.COUNT_WITNESSES === 1) { // the temp unit might have rebuilt the MC
							db.executeInTransaction(function (conn, onDone) {
								storage.resetMemory(conn, onDone);
							});
						}
						if (ws.library_version === '0.4.2' && arrResponses.length === 1 && arrResponses[0].bounced && typeof arrResponses[0].response.error === 'object')
							arrResponses[0].response.error = arrResponses[0].response.error.message;
						sendResponse(ws, tag, arrResponses);
					});
				})
			});
			break;

		case 'light/get_aa_state_vars':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_aa_state_vars");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "address not valid");
			if ('var_prefix_from' in params && typeof params.var_prefix_from !== 'string')
				return sendErrorResponse(ws, tag, "var_prefix_from must be string");
			if ('var_prefix_to' in params && typeof params.var_prefix_to !== 'string')
				return sendErrorResponse(ws, tag, "var_prefix_to must be string");
			if ('var_prefix' in params && typeof params.var_prefix !== 'string')
				return sendErrorResponse(ws, tag, "var_prefix must be string");
			if ('var_prefix' in params && ('var_prefix_from' in params || 'var_prefix_to' in params))
				return sendErrorResponse(ws, tag, "var_prefix cannot be used with var_prefix_from or var_prefix_to");
			if ('var_prefix' in params){
				params.var_prefix_from = params.var_prefix;
				params.var_prefix_to = params.var_prefix;
			}
			if ('limit' in params && !ValidationUtils.isPositiveInteger(params.limit))
				return sendErrorResponse(ws, tag, "limit must be a positive integer");
			if ('limit' in params && params.limit > MAX_STATE_VARS)
				return sendErrorResponse(ws, tag, "limit cannot be greater than " + MAX_STATE_VARS);
			storage.readAADefinition(db, params.address, null, function (arrDefinition) {
				if (!arrDefinition) {
					arrDefinition = storage.getUnconfirmedAADefinition(params.address);
					if (!arrDefinition)
						return sendErrorResponse(ws, tag, "not an AA");
				}
				storage.readAAStateVars(params.address, params.var_prefix_from || '', params.var_prefix_to || '', params.limit || MAX_STATE_VARS, function (objStateVars) {
					sendResponse(ws, tag, objStateVars);
				});
			});
			break;
			
		case 'light/get_aa_balances':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_aa_balances");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "address not valid");
			storage.readAABalances(db, params.address, function(assocBalances) {
				sendResponse(ws, tag, { balances: assocBalances });
			});
			break;
			
		case 'light/execute_getter':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/execute_getter");
			if (!ValidationUtils.isValidAddress(params.address))
				return sendErrorResponse(ws, tag, "address not valid");
			if (!ValidationUtils.isNonemptyString(params.getter))
				return sendErrorResponse(ws, tag, "no getter");
			if ('args' in params) {
				if (!Array.isArray(params.args))
					return sendErrorResponse(ws, tag, "args must be array");
				// remove the trailing null as this type doesn't exist in oscript
				while (params.args.length && params.args[params.args.length - 1] === null)
					params.args.length--;
				if (params.args.some(a => a === null))
					return sendErrorResponse(ws, tag, "args cannot include null");
				if (isTooDeeplyNestedOrHasTooManyNodes(params.args))
					return sendErrorResponse(ws, tag, "args are too deeply nested or have too many nodes");
			}
			formulaEvaluation.executeGetter(db, params.address, params.getter, params.args || [], function (err, res) {
				if (err)
					return sendErrorResponse(ws, tag, err);
				sendResponse(ws, tag, { result: res });
			});
			break;
			
		case 'light/get_aas_by_base_aas':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_aas_by_base_aas");
			var base_aas = params.base_aas || [params.base_aa];
			if (!ValidationUtils.isNonemptyArray(base_aas))
				return sendErrorResponse(ws, tag, "no base_aas in light/get_aas_by_base_aas");
			if (base_aas.length > 20)
				return sendErrorResponse(ws, tag, "too many base_aas in light/get_aas_by_base_aas, max 20");
			if (!base_aas.every(ValidationUtils.isValidAddress))
				return sendErrorResponse(ws, tag, "base_aa address not valid");
			var aa_params = params.params || {};
			for (var name in aa_params) {
				var value = aa_params[name];
				if (typeof value === 'object') {
					if (!ValidationUtils.isArrayOfLength(value, 2))
						return sendErrorResponse(ws, tag, "value of param " + name + " must be a 2-element array");
					var comp = value[0];
					value = value[1];
					if (typeof comp !== 'string')
						return sendErrorResponse(ws, tag, "bad comparison");
					if (!['=', '!=', '>', '>=', '<', '<='].includes(comp))
						return sendErrorResponse(ws, tag, "invalid comparison of param " + name + ": " + comp);
					if (!['string', 'number', 'boolean'].includes(typeof value))
						return sendErrorResponse(ws, tag, "invalid type of param " + name + ": " + (typeof value));
				}
			}
			db.query("SELECT address, definition, unit, creation_date FROM aa_addresses WHERE base_aa IN(?)", [base_aas], function (rows) {
				var arrAAs = [];
				rows.forEach(function (row) {
					var arrDefinition = JSON.parse(row.definition);
					var this_aa_params = arrDefinition[1].params;
					for (var name in aa_params) {
						if (!satisfiesSearchCriteria(this_aa_params[name], aa_params[name]))
							return;
					}
					arrAAs.push({ address: row.address, definition: arrDefinition, unit: row.unit, creation_date: row.creation_date });
				});
				sendResponse(ws, tag, arrAAs);
			});
			break;
			
		case 'light/get_aa_responses':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_aa_responses");
			var aas = params.aas || [params.aa];
			if (!ValidationUtils.isNonemptyArray(aas))
				return sendErrorResponse(ws, tag, "no aas in light/get_aa_responses");
			if (aas.length > 20)
				return sendErrorResponse(ws, tag, "too many aas in light/get_aa_responses, max 20");
			if (!aas.every(ValidationUtils.isValidAddress))
				return sendErrorResponse(ws, tag, "aa address not valid");
			if ("max_mci" in params && !ValidationUtils.isPositiveInteger(params.max_mci))
				return sendErrorResponse(ws, tag, "max_mci must be positive integer");
			const max_mci = params.max_mci || 1e15;
			if ("min_mci" in params && !ValidationUtils.isPositiveInteger(params.min_mci))
				return sendErrorResponse(ws, tag, "min_mci must be positive integer");
			const min_mci = params.min_mci || 0;
			var order = params.order || 'DESC';
			if (!['ASC', 'DESC'].includes(order))
				return sendErrorResponse(ws, tag, "bad order");
			// for pagination, note the mci of the last response and use it as max_mci or min_mci (depending on order) in the next request. You'll receive duplicates, filter them out.
			db.query(
				`SELECT mci, trigger_address, aa_address, trigger_unit, bounced, response_unit, response, timestamp
				FROM aa_responses
				CROSS JOIN units ON trigger_unit=unit
				WHERE aa_address IN(?) AND mci>=? AND mci<=?
				ORDER BY mci ${order}, aa_response_id ${order}
				LIMIT 100`,
				[aas, min_mci, max_mci],
				function (rows) {
					light.enrichAAResponses(rows, () => {
						sendResponse(ws, tag, rows);
					});
				}
			);
			break;

		// get a chain of AA responses starting from a specific trigger unit. The chain can branch.
		case 'light/get_aa_response_chain':
			if (!params)
				return sendErrorResponse(ws, tag, "no params in light/get_aa_response_chain");
			var trigger_unit = params.trigger_unit;
			if (!ValidationUtils.isValidBase64(trigger_unit, constants.HASH_LENGTH))
				return sendErrorResponse(ws, tag, "invalid trigger unit in light/get_aa_response_chain");
			var all_rows = [];
			var addAAResponse = function (trigger_unit, onDone) {
				db.query(
					"SELECT mci, trigger_address, aa_address, trigger_unit, bounced, response_unit, response, timestamp \n\
					FROM aa_responses CROSS JOIN units ON trigger_unit=unit \n\
					WHERE trigger_unit=? ORDER BY aa_address",
					[trigger_unit],
					function (rows) {
						async.eachSeries(
							rows,
							function (row, cb) {
								all_rows.push(row);
								row.response_unit ? addAAResponse(row.response_unit, cb) : cb();
							},
							onDone
						);
					}
				);
			};
			addAAResponse(trigger_unit, () => {
				if (all_rows.length === 0)
					return sendErrorResponse(ws, tag, "no such trigger unit in light/get_aa_response_chain");
				light.enrichAAResponses(all_rows, () => {
					sendResponse(ws, tag, all_rows);
				});
			});
			break;

		// I'm a hub, the peer wants to enable push notifications
		case 'hub/enable_notification':
			if(ws.device_address)
				eventBus.emit("enableNotification", ws.device_address, params);
			sendResponse(ws, tag, 'ok');
			break;

		// I'm a hub, the peer wants to disable push notifications
		case 'hub/disable_notification':
			if(ws.device_address)
				eventBus.emit("disableNotification", ws.device_address, params);
			sendResponse(ws, tag, 'ok');
			break;
			
		case 'hub/get_bots':
			db.query("SELECT id, name, pairing_code, description FROM bots ORDER BY rank DESC, id", [], function(rows){
				sendResponse(ws, tag, rows);
			});
			break;
			
		case 'hub/get_asset_metadata':
			var asset = params;
			if (!ValidationUtils.isStringOfLength(asset, constants.HASH_LENGTH))
				return sendErrorResponse(ws, tag, "bad asset");
			db.query("SELECT metadata_unit, registry_address, suffix FROM asset_metadata WHERE asset=?", [asset], function(rows){
				if (rows.length === 0)
					return sendErrorResponse(ws, tag, "no metadata");
				sendResponse(ws, tag, rows[0]);
			});
			break;

		case 'hub/get_arbstore_url':
			if (version2int(ws.library_version) < version2int('0.4.5')) {
				sendErrorResponse(ws, tag, "old core (arbstores)");
				return disconnectOldVersion(ws, "old core (arbstores)");
			}
			if (!conf.arbstores)
				return sendErrorResponse(ws, tag, "arbstores not defined");
			var arbiter_address = params;
			if (!ValidationUtils.isValidAddress(arbiter_address))
				return sendErrorResponse(ws, tag, "invalid arbiter address");
			db.query("SELECT arbstore_address FROM arbiter_locations WHERE arbiter_address=?", [arbiter_address], function(rows){
				if (!rows.length)
					return sendErrorResponse(ws, tag, "arbiter is not known");
				sendResponse(ws, tag, conf.arbstores[rows[0].arbstore_address]);
			});
			break;
		
		case 'hub/get_arbstore_address':
			if (version2int(ws.library_version) < version2int('0.4.5')) {
				sendErrorResponse(ws, tag, "old core (arbstores)");
				return disconnectOldVersion(ws, "old core (arbstores)");
			}
			if (!conf.arbstores)
				return sendErrorResponse(ws, tag, "arbstores not defined");
			var arbiter_address = params;
			if (!ValidationUtils.isValidAddress(arbiter_address))
				return sendErrorResponse(ws, tag, "invalid arbiter address");
			db.query("SELECT arbstore_address FROM arbiter_locations WHERE arbiter_address=?", [arbiter_address], function(rows){
				if (!rows.length)
					return sendErrorResponse(ws, tag, "arbiter is not known");
				sendResponse(ws, tag, rows[0].arbstore_address);
			});
			break;
		
		case 'hub/get_arbstore_url_by_address':
			if (version2int(ws.library_version) < version2int('0.4.5')) {
				sendErrorResponse(ws, tag, "old core (arbstores)");
				return disconnectOldVersion(ws, "old core (arbstores)");
			}
			var arbstore_address = params;
			if (!conf.arbstores || !conf.arbstores[arbstore_address])
				return sendErrorResponse(ws, tag, "arbstore is not known");
			sendResponse(ws, tag, conf.arbstores[arbstore_address]);
			break;
		
		case 'hub/get_exchange_rates':
			sendResponse(ws, tag, exchangeRates);
			break;

		case 'custom':
			eventBus.emit('custom_request', ws, params,tag);
			delete ws.assocCommandsInPreparingResponse[tag];
			decrementPending(requestsInWork, ws.host);
			break;
		
		default:
			delete ws.assocCommandsInPreparingResponse[tag];
			decrementPending(requestsInWork, ws.host);
	}
}

function disconnectOldVersion(ws, errMsg) {
	sendJustsaying(ws, 'upgrade_required');
	sendJustsaying(ws, errMsg);
	ws.close(1000, errMsg);
}

function satisfiesSearchCriteria(this_param_value, searched_param_value) {
	var comp = '=';
	if (typeof searched_param_value === 'object') {
		comp = searched_param_value[0];
		searched_param_value = searched_param_value[1];
	}
	if (typeof this_param_value !== typeof searched_param_value)
		return (comp === '!=');
	switch (comp) {
		case '=':  return this_param_value === searched_param_value;
		case '!=': return this_param_value !== searched_param_value;
		case '>':  return this_param_value > searched_param_value;
		case '>=': return this_param_value >= searched_param_value;
		case '<':  return this_param_value < searched_param_value;
		case '<=': return this_param_value <= searched_param_value;
		default: throw Error("unknown comp: " + comp);
	}
}

function onWebsocketMessage(message) {
		
	var ws = this;
	
	if (ws.readyState !== ws.OPEN)
		return console.log("received a message on socket with ready state "+ws.readyState);
	
	if (typeof message !== 'string') // ws 8+
		message = message.toString();
	console.log('RECEIVED ' + message.length + ' chars ' + (message.length > 1000 ? message.substr(0,1000)+'...' : message) + ' from '+ws.peer);
	ws.last_ts = Date.now();
	
	try{
		var arrMessage = JSON.parse(message);
		var message_type = arrMessage[0];
		var content = arrMessage[1];
		if (!content || typeof content !== 'object')
			return console.log("content is not object: "+content);
	}
	catch(e){
		return console.log('failed to json.parse message '+message);
	}
	
	incrementPending(messagesInWork, ws.host)
	switch (message_type){
		case 'justsaying':
			handleJustsaying(ws, content.subject, content.body);
			break;
			
		case 'request':
			handleRequest(ws, content.tag, content.command, content.params);
			break;
		
		case 'response':
			handleResponse(ws, content.tag, content.response);
			break;
			
		default: 
			console.log("unknown type: ", message_type);
		//	throw Error("unknown type: "+message_type);
	}
	decrementPending(messagesInWork, ws.host);
}

function watchMemory() {
	const v8 = require('v8');
	const THRESHOLD_PERCENT = 80;

	setInterval(() => {
		const stats = v8.getHeapStatistics();
		const usagePercent = (stats.used_heap_size / stats.heap_size_limit) * 100;
		if (usagePercent > THRESHOLD_PERCENT) // crash with uncaughtException before OOM
			throw new Error(`Memory usage is at ${usagePercent.toFixed(2)}%`);
	}, 5000); // Check every 5 seconds
}


// @see https://www.npmjs.com/package/ws#multiple-servers-sharing-a-single-https-server
function handleUpgradeConnection(incomingRequest, socket, head) {
	if (!(wss instanceof WebSocketServer)) throw new Error('reuse port and upgrade connection in light node is not supported')

	if (incomingRequest instanceof require('net').Server && !socket && !head) {
		incomingRequest.on('upgrade', function(_request, _socket, _head) {
			upgrade(_request, _socket, _head)
		})
	} else upgrade(incomingRequest, socket, head)

	function upgrade($request, $socket, $head) {
		wss.handleUpgrade($request, $socket, $head, function(ws) {
			wss.emit('connection', ws, $request);
		})
	}
}

function startAcceptingConnections(){
	db.query("DELETE FROM watched_light_addresses");
	db.query("DELETE FROM watched_light_aas");
	db.query("DELETE FROM watched_light_units");
	//db.query("DELETE FROM light_peer_witnesses");
	// listen for new connections
	wss = new WebSocketServer(conf.portReuse ? { noServer: true, maxPayload } : { port: conf.port, maxPayload });
	wss.on('connection', function(ws, req) {
		if (!req.socket.remoteAddress){
			console.log("no ip in accepted connection");
			ws.terminate();
			return;
		}
		let ip = req.socket.remoteAddress.replace(/^::ffff:/, '');
		if (req.headers['x-real-ip'] && (ip === '127.0.0.1' || ip.match(/^192\.168\./) || ip.match(/^10\./))) // we are behind a proxy
			ip = req.headers['x-real-ip'];
		ws.peer = ip + ":" + req.socket.remotePort;
		ws.host = ip;
		ws.assocPendingRequests = {};
		ws.assocCommandsInPreparingResponse = {};
		ws.bInbound = true;
		ws.last_ts = Date.now();
		console.log('got connection from '+ws.peer+", host "+ws.host);
		ws.on('error', function (e) {
			console.log("error on client " + ws.peer + ": " + e);
			ws.close(1000, "received error");
		});
		if (wss.clients.size >= conf.MAX_INBOUND_CONNECTIONS){
			console.log("inbound connections maxed out, rejecting new client "+ip);
			ws.close(1000, "inbound connections maxed out"); // 1001 doesn't work in cordova
			return;
		}
		if (peerIsBlocked(ws.host)) {
			console.log("rejecting new client "+ws.host+" because of bad stats");
			return ws.terminate();
		}

		// welcome the new peer with the list of free joints
		//if (!bCatchingUp)
		//    sendFreeJoints(ws);

		sendVersion(ws);

		// I'm a hub, send challenge
		if (conf.bServeAsHub){
			ws.challenge = crypto.randomBytes(30).toString("base64");
			sendJustsaying(ws, 'hub/challenge', ws.challenge);
		}
		if (!conf.bLight)
			subscribe(ws);
		eventBus.emit('connected', ws);
		ws.on('message', function(message){
			onWebsocketMessage.call(ws, message);
		});
		ws.on('close', function(){
			if (bWatchingForLight){
				db.query("DELETE FROM watched_light_addresses WHERE peer=?", [ws.peer]);
				db.query("DELETE FROM watched_light_aas WHERE peer=?", [ws.peer]);
				db.query("DELETE FROM watched_light_units WHERE peer=?", [ws.peer]);
				//db.query("DELETE FROM light_peer_witnesses WHERE peer=?", [ws.peer]);
			}
			console.log("client "+ws.peer+" disconnected");
			cancelRequestsOnClosedConnection(ws);
		});
		addPeerHost(ws.host);
	});
	console.log('WSS running at port ' + conf.port);
}


function startPeerExchange() {
	if (conf.bWantNewPeers){
		// outbound connections
		addOutboundPeers();
		// retry lost and failed connections every 1 minute
		setInterval(addOutboundPeers, 60*1000);
		setTimeout(checkIfHaveEnoughOutboundPeersAndAdd, 30*1000);
		setInterval(checkIfHaveEnoughOutboundPeersAndAdd, 3600 * 1000);
		setInterval(purgeDeadPeers, 30*60*1000);
	}
}

async function startRelay(){
	setInterval(unblockPeers, 10*60*1000);
	initBlockedPeers();
	await clearStalePrunedContentPointers();
	
	if (bCordova || !conf.port) // no listener on mobile
		wss = {clients: new Set()};
	else
		startAcceptingConnections();
	
	const unlock = await mutex.lock(['handleJoint', 'initial']); // make newly received units wait until we finish initializing
	await storage.initCaches();
	joint_storage.initUnhandledAndKnownBad();
	checkCatchupLeftovers();

	startPeerExchange();

	// purge peer_events every 6 hours, removing those older than 0.5 days ago.
	setInterval(purgePeerEvents, 6*60*60*1000);
	setInterval(function(){flushEvents(true)}, 1000 * 60);
	
	// request needed joints that were not received during the previous session
	rerequestLostJoints();
	setInterval(rerequestLostJoints, 8*1000);
	
	setInterval(purgeJunkUnhandledJoints, 30*60*1000);
	setInterval(joint_storage.purgeUncoveredNonserialJointsUnderLock, 60*1000);
	setInterval(handleSavedPrivatePayments, 5*1000);
	joint_storage.readDependentJointsThatAreReady(null, handleSavedJoint);
	watchMemory();

	eventBus.on('new_aa_unit', onNewAA);
	eventBus.on('system_vars_updated', onSystemVarUpdated);
	eventBus.on('system_var_vote', sendSysVarVoteToAllWatchers);
	await aa_composer.handleAATriggers(); // in case anything's left from the previous run
	await storage.updateMissingTpsFees();
	unlock();
}

async function startLightClient(){
	wss = {clients: new Set()};
	await storage.initUnstableUnits(); // necessary for archiveJointAndDescendants()
	rerequestLostJointsOfPrivatePayments();
	setInterval(rerequestLostJointsOfPrivatePayments, 5*1000);
	setInterval(handleSavedPrivatePayments, 5*1000);
	setInterval(requestUnfinishedPastUnitsOfSavedPrivateElements, 12*1000);
}

async function start(){
	if (bStarted)
		return console.log("network already started");
	bStarted = true;
	console.log("starting network");
	conf.bLight ? await startLightClient() : await startRelay();
	setInterval(printConnectionStatus, 6*1000);
	// if we have exactly same intervals on two clints, they might send heartbeats to each other at the same time
	setInterval(heartbeat, 3*1000 + getRandomInt(0, 1000));
	eventBus.emit('network_started');
}

function closeAllWsConnections() {
	arrOutboundPeers.forEach(function(ws) {
		ws.close(1000,'Re-connect');
	});
}

function isStarted(){
	return bStarted;
}

function isConnected(){
	return (arrOutboundPeers.length + wss.clients.size);
}

function isCatchingUp(){
	return bCatchingUp;
}

function waitUntilCatchedUp(cb) {
	if (conf.bLight)
		throw Error("call waitUntilCatchedUp in full wallets only");
	if (!cb)
		return new Promise(resolve => waitUntilCatchedUp(resolve));
	waitTillSyncIdle(() => {
		if (!bCatchingUp && !bWaitingForCatchupChain)
			return cb();
		console.log('bCatchingUp =', bCatchingUp, 'bWaitingForCatchupChain =', bWaitingForCatchupChain, ' will wait for catchup to finish');
		eventBus.once('catching_up_done', () => waitUntilCatchedUp(cb));
	});
}

if (!conf.explicitStart) {
	start();
}

function setExchangeRates(rates) {
	exchangeRates = rates;
}

function incrementPending(assoc, host) {
	if (!assoc[host])
		assoc[host] = 0;
	assoc[host]++;
}

function decrementPending(assoc, host) {
	if (!assoc[host]) throw Error(`no pending for ${host}: ${assoc[host]}`);
	assoc[host]--;
	if (assoc[host] === 0)
		delete assoc[host];
}

process.on('uncaughtException', (err) => {
	console.log('Uncaught exception:', err);
	console.error('Uncaught exception:', err);
	if (!conf.bLight) {
		let hosts = [...Object.keys(messagesInWork), ...Object.keys(requestsInWork)];
		if (currentJointHost)
			hosts.push(currentJointHost);
		console.log('Clients with pending requests/messages at the time of uncaught exception:', hosts);
		const fs = require('fs');
		const app_data_dir = require('./desktop_app.js').getAppDataDir();
		fs.writeFileSync(`${app_data_dir}/uncaught_exception_clients.txt`, hosts.concat(Object.keys(assocBlockedPeers)).join('\n'), 'utf8');
	}
	throw err; // crash the process to avoid ending up in an inconsistent state
});

exports.start = start;
exports.startAcceptingConnections = startAcceptingConnections;
exports.startPeerExchange = startPeerExchange;

exports.postJointToLightVendor = postJointToLightVendor;
exports.broadcastJoint = broadcastJoint;
exports.sendPrivatePayment = sendPrivatePayment;

exports.sendJustsaying = sendJustsaying;
exports.sendAllInboundJustsaying = sendAllInboundJustsaying;
exports.sendError = sendError;
exports.sendRequest = sendRequest;
exports.sendResponse = sendResponse;
exports.findOutboundPeerOrConnect = findOutboundPeerOrConnect;
exports.handleOnlineJoint = handleOnlineJoint;
exports.handleUpgradeConnection = handleUpgradeConnection;
exports.handleRequest = handleRequest;

exports.handleOnlinePrivatePayment = handleOnlinePrivatePayment;
exports.requestUnfinishedPastUnitsOfPrivateChains = requestUnfinishedPastUnitsOfPrivateChains;
exports.requestProofsOfJointsIfNewOrUnstable = requestProofsOfJointsIfNewOrUnstable;

exports.requestFromLightVendor = requestFromLightVendor;

exports.addPeer = addPeer;
exports.isValidWsUrl = isValidWsUrl;

exports.initWitnessesIfNecessary = initWitnessesIfNecessary;

exports.setMyDeviceProps = setMyDeviceProps;

exports.setWatchedAddresses = setWatchedAddresses;
exports.addWatchedAddress = addWatchedAddress;
exports.addLightWatchedAddress = addLightWatchedAddress;
exports.addTempLightWatchedAddress = addTempLightWatchedAddress;
exports.addLightWatchedAa = addLightWatchedAa;

exports.getConnectionStatus = getConnectionStatus;
exports.closeAllWsConnections = closeAllWsConnections;
exports.isStarted = isStarted;
exports.isConnected = isConnected;
exports.isCatchingUp = isCatchingUp;
exports.waitUntilCatchedUp = waitUntilCatchedUp;
exports.waitTillIdle = waitTillIdle;
exports.waitTillSyncIdle = waitTillSyncIdle;
exports.requestHistoryFor = requestHistoryFor;
exports.requestHistoryAfterMCI = requestHistoryAfterMCI;
exports.exchangeRates = exchangeRates;
exports.setExchangeRates = setExchangeRates;
exports.knownWitnesses = knownWitnesses;
exports.getInboundDeviceWebSocket = getInboundDeviceWebSocket;
exports.deletePendingRequest = deletePendingRequest;

exports.MAX_STATE_VARS = MAX_STATE_VARS;