tildefriends/core/ssb.js

419 lines
12 KiB
JavaScript

"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) {
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);
});
ssb.addRpc(['blobs', 'has'], function(request) {
var found = false;
ssb.sqlStream(
'SELECT 1 FROM blobs where id = ?1',
[request.args[0]],
function(row) {
found = true;
});
request.send_json(found);
});
ssb.addRpc(['blobs', 'get'], function(request) {
for (let arg of request.args) {
var blob;
if (arg.key) {
blob = ssb.blobGet(arg.key);
} else {
blob = ssb.blobGet(arg);
}
const k_send_max = 8192;
if (blob.byteLength > k_send_max) {
for (var i = 0; i < blob.byteLength; i += k_send_max) {
var buffer = new Uint8Array(blob, i, Math.min(blob.byteLength - i, k_send_max));
request.send_binary(buffer);
}
} else {
request.send_binary(blob);
}
request.send_json_end(true);
}
});
ssb.addRpc(['gossip', 'ping'], function(request) {
request.more(function ping(message) {
message.send_json(Date.now());
});
});
ssb.addRpc(['tunnel', 'isRoom'], function(request) {
request.send_json({"name":"tilde friends tunnel","membership":false,"features":["tunnel","room1"]});
});
function notify_attendant_changed(id, type) {
for (let r of Object.values(g_attendants)) {
try {
r.send_json({
type: type,
id: id,
});
} catch (e) {
print(`Removing ${r.connection.id} from g_attendants in ${type}.`, e);
delete g_attendants[r.connection.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;
});
ssb.addRpc(['tunnel', 'connect'], function(request) {
if (!request.args[0].origin &&
request.args[0].portal &&
request.args[0].target) {
let target_connection = ssb.getConnection(request.args[0].target);
let target_request_number = target_connection.send_json({
'name': ['tunnel', 'connect'],
'args': [{
'origin': request.connection.id,
'portal': request.args[0].portal,
'target': request.args[0].target,
}],
'type': 'duplex',
});
ssb.tunnel(request.connection, -request.request_number, target_connection, target_request_number);
} else if (request.args[0].origin &&
request.args[0].portal &&
request.args[0].target) {
ssb.createTunnel(request.connection, -request.request_number, request.args[0].origin);
}
});
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);
});
});
});