tildefriends/core/ssb.js

"use strict";
var g_wants_requests = {};
var g_database = new Database('core');
let g_attendants = {};
const k_use_create_history_stream = false;
const k_blobs_concurrent_target = 8;

function following(db, id) {
	var o = db.get(id + ":following");
	const k_version = 5;
	var f = o ? JSON.parse(o) : o;
	if (!f || f.version != k_version) {
		f = {users: [], sequence: 0, version: k_version};
	}
	f.users = new Set(f.users);
	ssb.sqlStream(
		"SELECT "+
		"  sequence, "+
		"  json_extract(content, '$.contact') AS contact, "+
		"  json_extract(content, '$.following') AS following "+
		"FROM messages "+
		"WHERE "+
		"  author = ?1 AND "+
		"  sequence > ?2 AND "+
		"  json_extract(content, '$.type') = 'contact' "+
		"UNION SELECT MAX(sequence) AS sequence, NULL, NULL FROM messages WHERE author = ?1 "+
		"ORDER BY sequence",
		[id, f.sequence],
		function(row) {
			if (row.following) {
				f.users.add(row.contact);
			} else {
				f.users.delete(row.contact);
			}
			f.sequence = row.sequence;
		});
	f.users = Array.from(f.users).sort();
	var j = JSON.stringify(f);
	if (o != j) {
		db.set(id + ":following", j);
	}
	return f.users;
}

function followingDeep(db, seed_ids, depth) {
	if (depth <= 0) {
		return seed_ids;
	}
	var f = seed_ids.map(x => following(db, x));
	var ids = [].concat(...f);
	var x = followingDeep(db, [...new Set(ids)].sort(), depth - 1);
	x = [...new Set([].concat(...x, ...seed_ids))].sort();
	return x;
}

function get_latest_sequence_for_author(author) {
	var sequence = 0;
	ssb.sqlStream(
		'SELECT MAX(sequence) AS sequence FROM messages WHERE author = ?1',
		[author],
		function(row) {
			if (row.sequence) {
				sequence = row.sequence;
			}
		});
	return sequence;
}

function storeMessage(message) {
	var payload = message.message.value ? message.message.value : message.message;
	if (typeof(payload) == 'object') {
		ssb.storeMessage(payload);
	}
}

function tunnel_attendants(request) {
	if (request.message.type !== 'state') {
		throw Error('Unexpected type: ' + request.message.type);
	}
	let state = new Set(request.message.ids);
	for (let id of state) {
		request.add_room_attendant(id);
	}
	request.more(function attendants(message) {
		if (message.message.type === 'joined') {
			request.add_room_attendant(message.message.id);
			state.add(message.message.id);
		} else if (message.message.type === 'left') {
			request.remove_room_attendant(message.message.id);
			state.delete(message.message.id);
		} else {
			throw Error('Unexpected type: ' + message.type);
		}
	});
}

function send_blobs_create_wants(connection) {
	connection.send_json({'name': ['blobs', 'createWants'], 'type': 'source', 'args': []}, function on_blob_create_wants(message) {
		if (message.message?.name === 'Error') {
			return;
		}
		Object.keys(message.message).forEach(function(id) {
			if (message.message[id] < 0) {
				if (g_wants_requests[connection.id]) {
					delete connection.active_blob_wants[id];
					var blob = ssb.blobGet(id);
					if (blob) {
						var out_message = {};
						out_message[id] = blob.byteLength;
						g_wants_requests[connection.id].send_json(out_message);
					}
				}
			} else {
				var received_bytes = 0;
				var expected_bytes = message.message[id];
				var buffer = new Uint8Array(expected_bytes);
				connection.send_json({'name': ['blobs', 'get'], 'type': 'source', 'args': [id]}, function(message) {
					if (message.flags & 0x4 /* end */) {
						delete connection.active_blob_wants[id];
					} else {
						buffer.set(new Uint8Array(message.message, 0, message.message.byteLength), received_bytes);
						received_bytes += message.message.byteLength;
						if (received_bytes == expected_bytes) {
							ssb.blobStore(buffer);
						}
					}
				});
				if (g_wants_requests[connection.id] && Object.keys(connection.active_blob_wants).length < k_blobs_concurrent_target) {
					requestMoreBlobs(g_wants_requests[connection.id]);
				}
			}
		});
	});
}

ssb.addEventListener('connections', function on_connections_changed(change, connection) {
	if (change == 'add') {
		connection.active_blob_wants = {};
		var sequence = get_latest_sequence_for_author(connection.id);
		if (k_use_create_history_stream) {
			connection.send_json({'name': ['createHistoryStream'], 'type': 'source', 'args': [{'id': connection.id, 'seq': sequence, 'live': true, 'keys': false}]}, storeMessage);
			var identities = ssb.getAllIdentities();
			followingDeep(g_database, identities, 2).then(function(ids) {
				for (let id of ids) {
					if (identities.indexOf(id) != -1) {
						continue;
					}
					var sequence = get_latest_sequence_for_author(id);
					connection.send_json({'name': ['createHistoryStream'], 'type': 'source', 'args': [{'id': id, 'seq': sequence, 'live': true, 'keys': false}]}, storeMessage);
				}
			});
		} else {
			if (connection.is_client) {
				connection.send_json({"name": ["ebt", "replicate"], "args": [{"version": 3, "format": "classic"}], "type": "duplex"}, ebtReplicateClient);

				connection.send_json_async({'name': ['tunnel', 'isRoom'], 'args': []}, function tunnel_is_room(request) {
					if (request.message) {
						request.connection.send_json({'name': ['room', 'attendants'], 'args': [], 'type': 'source'}, tunnel_attendants);
					}
				});
			}
			send_blobs_create_wants(connection);
		}
	} else if (change == 'remove') {
		print('REMOVE', connection.id);
		notify_attendant_changed(connection.id, 'left');
		delete g_attendants[connection.id];
		delete g_wants_requests[connection.id];
	} else {
		print('CHANGE', change);
	}
});

function blob_want_discovered(request, id) {
	if (!request || !request.connection || Object.keys(request.connection.active_blob_wants).length > k_blobs_concurrent_target) {
		return;
	}
	var message = {};
	message[id] = -1;
	request.send_json(message);
	request.connection.active_blob_wants[id] = true;
}

function requestMoreBlobs(request) {
	ssb.sqlStream(
		'SELECT id FROM blob_wants LIMIT ' + k_blobs_concurrent_target,
		[],
		row => blob_want_discovered(request, row.id));
}

ssb.addRpc(['blobs', 'createWants'], function(request) {
	g_wants_requests[request.connection.id] = request;
	ssb.addEventListener('blob_want_added', id => blob_want_discovered(request, id));
	requestMoreBlobs(request);
});

function notify_attendant_changed(id, type) {
	if (!id) {
		print(`notify_attendant_changed called with id=${id}`);
		return;
	}
	for (let r of Object.values(g_attendants)) {
		try {
			r.send_json({
				type: type,
				id: id,
			});
		} catch (e) {
			print(`Removing ${id} from g_attendants in ${type}.`, e);
			delete g_attendants[id];
		}
	}
}

ssb.addRpc(['room', 'attendants'], function(request) {
	let ids = Object.keys(g_attendants).sort();
	request.send_json({
		type: 'state',
		ids: ids,
	});
	notify_attendant_changed(request.connection.id, 'joined');
	g_attendants[request.connection.id] = request;
});

function ebtReplicateSendClock(request, have) {
	var identities = ssb.getAllIdentities();
	var message = {};
	var last_sent = request.connection.sent_clock || {};
	var ids = followingDeep(g_database, identities, 2).concat([request.connection.id]);
	if (!Object.keys(last_sent).length) {
		for (let id of ids) {
			message[id] = get_latest_sequence_for_author(id);
		}
	}
	for (let id of Object.keys(have)) {
		if (message[id] === undefined) {
			var sequence = get_latest_sequence_for_author(id);
			message[id] = sequence ? sequence : -1;
		}
	}

	var to_send = {}
	var offset = Math.floor(Math.random() * ids.length);
	for (var i = 0; i < ids.length; i++) {
		var id = ids[(i + offset) % ids.length];
		if (last_sent[id] === undefined || message[id] > last_sent[id]) {
			last_sent[id] = to_send[id] = message[id] === -1 ? -1 : message[id] << 1;
		}
		if (Object.keys(to_send).length >= 32) {
			request.send_json(to_send);
			to_send = {};
		}
	}
	request.connection.sent_clock = last_sent;

	if (Object.keys(to_send).length) {
		request.send_json(to_send);
	}
}

function formatMessage(row) {
	if (row.sequence_before_author) {
		return {
			previous: row.previous,
			sequence: row.sequence,
			author: row.author,
			timestamp: row.timestamp,
			hash: row.hash,
			content: JSON.parse(row.content),
			signature: row.signature,
		};
	} else {
		return {
			previous: row.previous,
			author: row.author,
			sequence: row.sequence,
			timestamp: row.timestamp,
			hash: row.hash,
			content: JSON.parse(row.content),
			signature: row.signature,
		};
	}
}

function ebtReplicateRegisterMessageCallback(request) {
	ssb.addEventListener('message', function(message_id) {
		ssb.sqlStream(
			'SELECT previous, author, id, sequence, timestamp, hash, content, signature FROM messages WHERE id = ?1',
			[message_id],
			function (row) {
				request.send_json(formatMessage(row));
			});
	});
}

function ebtReplicateCommon(request) {
	if (request.message.author) {
		storeMessage(request);
	} else {
		ebtReplicateSendClock(request, request.message);

		for (let id of Object.keys(request.message)) {
			if (request.message[id] >= 0 && (request.message[id] & 1) == 0) {
				ssb.sqlStream(
					'SELECT previous, author, id, sequence, timestamp, hash, content, signature FROM messages WHERE author = ?1 AND sequence >= ?2 ORDER BY sequence',
					[id, request.message[id] >> 1],
					function (row) {
						request.send_json(formatMessage(row));
					});
			}
		}
	}
}

function ebtReplicateClient(request) {
	if (request.message?.name !== 'Error') {
		if (!request.connection.message_registered) {
			ebtReplicateRegisterMessageCallback(request);
			request.connection.message_registered = true;
		}
		ebtReplicateCommon(request);
	}
}

function ebtReplicateServer(request) {
	ebtReplicateRegisterMessageCallback(request);
	ebtReplicateSendClock(request, {});
	request.more(ebtReplicateCommon);
}

ssb.addRpc(['ebt', 'replicate'], ebtReplicateServer);

ssb.addRpc(['createHistoryStream'], function(request) {
	var id = request.args[0].id;
	var seq = request.args[0].seq;
	var keys = request.args[0].keys || request.args[0].keys === undefined;
	function sendMessage(row) {
		if (keys) {
			var message = {
				key: row.id,
				value: formatMessage(row),
				timestamp: row.timestamp,
			};
		} else {
			var message = formatMessage(row);
		}
		request.send_json(message);
	}
	ssb.sqlStream(
		'SELECT previous, author, id, sequence, timestamp, hash, content, signature, sequence_before_author FROM messages WHERE author = ?1 AND sequence >= ?2 ORDER BY sequence',
		[id, seq ?? 0],
		sendMessage);
	ssb.addEventListener('message', function(message_id) {
		ssb.sqlStream(
			'SELECT previous, author, id, sequence, timestamp, hash, content, signature, sequence_before_author FROM messages WHERE id = ?1 AND author = ?2',
			[message_id, id],
			function (row) {
				sendMessage(row);
			});
	});
});