cory/tildefriends
5
6
Fork 2
Code Issues 39 Pull Requests Actions Releases 34 Wiki Activity

Work-in-progress, untested, naive peer exchange. Intended to be disabled by default by a setting.
Some checks failed
Build Tilde Friends / Build-All (push) Failing after 5m56s
Details

Browse Source
This commit is contained in:
Cory McWilliams
2024-08-14 21:07:16 -04:00
parent 107666cc60
commit 657bcadc7e
4 changed files with 180 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
src/ssb.rpc.c
View File

@ -19,6 +19,7 @@
#endif
static void _tf_ssb_connection_send_history_stream(tf_ssb_connection_t* connection, int32_t request_number, const char* author, int64_t sequence, bool keys, bool live);
static void _tf_ssb_rpc_send_peers_exchange(tf_ssb_connection_t* connection);
static void _tf_ssb_rpc_start_delete_blobs(tf_ssb_t* ssb, int delay_ms);
static int64_t _get_global_setting_int64(tf_ssb_t* ssb, const char* name, int64_t default_value)
@ -1216,6 +1217,11 @@ static void _tf_ssb_rpc_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_chang
_tf_ssb_rpc_connection_tunnel_isRoom_callback, NULL, NULL);
JS_FreeValue(context, message);
if (tf_ssb_is_peer_exchange(ssb))
{
_tf_ssb_rpc_send_peers_exchange(connection);
}
if (tf_ssb_is_replicator(ssb))
{
_tf_ssb_rpc_send_ebt_replicate(connection);
@ -1333,6 +1339,121 @@ void tf_ssb_rpc_start_periodic(tf_ssb_t* ssb)
_tf_ssb_rpc_start_delete_blobs(ssb, 30 * 1000);
}
typedef struct _peers_exchange_t
{
tf_ssb_t* ssb;
JSValue peers;
} peers_exchange_t;
static void _tf_ssb_get_peers_exhange_callback(
const char* host, const struct sockaddr_in* addr, tf_ssb_broadcast_origin_t origin, tf_ssb_connection_t* tunnel, const uint8_t* pub, void* user_data)
{
peers_exchange_t* data = user_data;
if (origin == k_tf_ssb_broadcast_origin_peer_exchange)
{
char fullid[k_id_base64_len] = { 0 };
tf_base64_encode(pub, sizeof(pub), fullid, sizeof(fullid));
char connection[1024] = { 0 };
snprintf(connection, sizeof(connection), "net:%s:%d~shs:%s", host, ntohs(addr->sin_port), fullid);
JSContext* context = tf_ssb_get_context(data->ssb);
JS_SetPropertyStr(context, data->peers, connection, JS_NewInt32(context, 0));
}
}
static JSValue _tf_ssb_get_peers_exchange(tf_ssb_t* ssb)
{
JSContext* context = tf_ssb_get_context(ssb);
JSValue peers = JS_NewObject(context);
tf_ssb_visit_broadcasts(ssb, _tf_ssb_get_peers_exhange_callback, &(peers_exchange_t) { .ssb = ssb, .peers = peers });
return peers;
}
static void _tf_ssb_rpc_peers_exchange_internal(
tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
{
JSContext* context = tf_ssb_connection_get_context(connection);
if (_is_error(context, args))
{
return;
}
/* The peer that participated in the exchange is now a peer exchange entry, too. */
tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
if (*tf_ssb_connection_get_host(connection))
{
char fullid[k_id_base64_len] = { 0 };
tf_ssb_connection_get_id(connection, fullid, sizeof(fullid));
char connection_string[1024] = { 0 };
snprintf(connection_string, sizeof(connection_string), "net:%s:%d~shs:%s", tf_ssb_connection_get_host(connection), tf_ssb_connection_get_port(connection), fullid);
tf_ssb_add_broadcast(ssb, connection_string, k_tf_ssb_broadcast_origin_peer_exchange, k_ssb_peer_exchange_expires_seconds);
}
JSValue in_peers = JS_GetPropertyStr(context, args, "peers");
JSPropertyEnum* ptab = NULL;
uint32_t plen = 0;
if (JS_GetOwnPropertyNames(context, &ptab, &plen, in_peers, JS_GPN_STRING_MASK) == 0)
{
for (uint32_t i = 0; i < plen; ++i)
{
JSValue key = JS_AtomToString(context, ptab[i].atom);
JSPropertyDescriptor desc;
JSValue key_value = JS_NULL;
if (JS_GetOwnProperty(context, &desc, args, ptab[i].atom) == 1)
{
key_value = desc.value;
JS_FreeValue(context, desc.setter);
JS_FreeValue(context, desc.getter);
}
const char* connection = JS_ToCString(context, key);
int64_t timestamp = 0;
JS_ToInt64(context, &timestamp, key_value);
/* ADD BROADCAST connection: timestamp */
JS_FreeCString(context, connection);
JS_FreeValue(context, key);
JS_FreeValue(context, key_value);
}
for (uint32_t i = 0; i < plen; ++i)
{
JS_FreeAtom(context, ptab[i].atom);
}
js_free(context, ptab);
}
JS_FreeValue(context, in_peers);
}
static void _tf_ssb_rpc_send_peers_exchange(tf_ssb_connection_t* connection)
{
int32_t request_number = tf_ssb_connection_next_request_number(connection);
JSContext* context = tf_ssb_connection_get_context(connection);
JSValue message = JS_NewObject(context);
tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
JS_SetPropertyStr(context, message, "peers", _tf_ssb_get_peers_exchange(ssb));
tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_new_request, request_number, NULL, message, _tf_ssb_rpc_peers_exchange_internal, NULL, NULL);
JS_FreeValue(context, message);
}
static void _tf_ssb_rpc_peers_exchange(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
{
tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
if (!tf_ssb_is_peer_exchange(ssb))
{
tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number, "peers.exchange");
return;
}
_tf_ssb_rpc_peers_exchange_internal(connection, flags, request_number, args, message, size, user_data);
JSContext* context = tf_ssb_connection_get_context(connection);
JSValue out_message = JS_NewObject(context);
JS_SetPropertyStr(context, out_message, "peers", _tf_ssb_get_peers_exchange(ssb));
tf_ssb_connection_rpc_send_json(connection, flags, -request_number, NULL, out_message, NULL, NULL, NULL);
JS_FreeValue(context, out_message);
}
void tf_ssb_rpc_register(tf_ssb_t* ssb)
{
tf_ssb_add_connections_changed_callback(ssb, _tf_ssb_rpc_connections_changed_callback, NULL, NULL);
@ -1346,4 +1467,5 @@ void tf_ssb_rpc_register(tf_ssb_t* ssb)
tf_ssb_add_rpc_callback(ssb, (const char*[]) { "room", "attendants", NULL }, _tf_ssb_rpc_room_attendants, NULL, NULL); /* SOURCE */
tf_ssb_add_rpc_callback(ssb, (const char*[]) { "createHistoryStream", NULL }, _tf_ssb_rpc_createHistoryStream, NULL, NULL); /* SOURCE */
tf_ssb_add_rpc_callback(ssb, (const char*[]) { "ebt", "replicate", NULL }, _tf_ssb_rpc_ebt_replicate_server, NULL, NULL); /* DUPLEX */
tf_ssb_add_rpc_callback(ssb, (const char*[]) { "peers", "exchange", NULL }, _tf_ssb_rpc_peers_exchange, NULL, NULL); /* ASYNC */
}