#include "ssb.rpc.h" #include "mem.h" #include "ssb.h" #include "ssb.db.h" #include "util.js.h" #include #include #include #include #if !defined(_countof) #define _countof(a) ((int)(sizeof((a)) / sizeof(*(a)))) #endif static void _tf_ssb_rpc_gossip_ping_callback(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data) { char buffer[256]; snprintf(buffer, sizeof(buffer), "%" PRId64, (int64_t)time(NULL) * 1000); tf_ssb_connection_rpc_send( connection, flags, -request_number, (const uint8_t*)buffer, strlen(buffer), NULL, NULL, NULL); } static void _tf_ssb_rpc_gossip_ping(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_connection_add_request(connection, -request_number, _tf_ssb_rpc_gossip_ping_callback, NULL, NULL, NULL); } static void _tf_ssb_rpc_blobs_get(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data) { if (flags & k_ssb_rpc_flag_end_error) { return; } tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_connection_get_context(connection); JSValue ids = JS_GetPropertyStr(context, args, "args"); int length = tf_util_get_length(context, ids); bool success = false; for (int i = 0; i < length; i++) { JSValue arg = JS_GetPropertyUint32(context, ids, i); const char* id = NULL; if (JS_IsString(arg)) { id = JS_ToCString(context, arg); } else { JSValue key = JS_GetPropertyStr(context, arg, "key"); id = JS_ToCString(context, key); JS_FreeValue(context, key); } uint8_t* blob = NULL; size_t size = 0; const int k_send_max = 8192; if (tf_ssb_db_blob_get(ssb, id, &blob, &size)) { for (size_t offset = 0; offset < size; offset += k_send_max) { tf_ssb_connection_rpc_send( connection, k_ssb_rpc_flag_binary | k_ssb_rpc_flag_stream, -request_number, blob + offset, offset + k_send_max <= size ? k_send_max : (size - offset), NULL, NULL, NULL); } success = true; tf_free(blob); } JS_FreeCString(context, id); JS_FreeValue(context, arg); } JS_FreeValue(context, ids); tf_ssb_connection_rpc_send( connection, k_ssb_rpc_flag_json | k_ssb_rpc_flag_end_error | k_ssb_rpc_flag_stream, -request_number, (const uint8_t*)(success ? "true" : "false"), strlen(success ? "true" : "false"), NULL, NULL, NULL); } static void _tf_ssb_rpc_blobs_has(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); JSContext* context = tf_ssb_connection_get_context(connection); JSValue ids = JS_GetPropertyStr(context, args, "args"); JSValue id = JS_GetPropertyUint32(context, ids, 0); const char* id_str = JS_ToCString(context, id); bool has = tf_ssb_db_blob_has(ssb, id_str); JS_FreeCString(context, id_str); JS_FreeValue(context, id); JS_FreeValue(context, ids); tf_ssb_connection_rpc_send( connection, k_ssb_rpc_flag_json | k_ssb_rpc_flag_end_error, -request_number, (const uint8_t*)(has ? "true" : "false"), strlen(has ? "true" : "false"), NULL, NULL, NULL); } static void _tf_ssb_rpc_blob_wants_added_callback(tf_ssb_t* ssb, const char* id, void* user_data) { tf_ssb_connection_t* connection = user_data; tf_ssb_blob_wants_t* blob_wants = tf_ssb_connection_get_blob_wants_state(connection); JSContext* context = tf_ssb_get_context(ssb); JSValue message = JS_NewObject(context); JS_SetPropertyStr(context, message, id, JS_NewInt64(context, -1)); tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream, -blob_wants->request_number, message, NULL, NULL, NULL); JS_FreeValue(context, message); } static void _tf_ssb_rpc_request_more_blobs(tf_ssb_connection_t* connection) { JSContext* context = tf_ssb_connection_get_context(connection); tf_ssb_blob_wants_t* blob_wants = tf_ssb_connection_get_blob_wants_state(connection); tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); sqlite3* db = tf_ssb_get_db(ssb); sqlite3_stmt* statement; if (sqlite3_prepare(db, "SELECT id FROM blob_wants WHERE id > ? ORDER BY id LIMIT 32", -1, &statement, NULL) == SQLITE_OK) { if (sqlite3_bind_text(statement, 1, blob_wants->last_id, -1, NULL) == SQLITE_OK) { while (sqlite3_step(statement) == SQLITE_ROW) { const char* blob = (const char*)sqlite3_column_text(statement, 0); JSValue message = JS_NewObject(context); JS_SetPropertyStr(context, message, blob, JS_NewInt32(context, -1)); tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream, -blob_wants->request_number, message, NULL, NULL, NULL); JS_FreeValue(context, message); snprintf(blob_wants->last_id, sizeof(blob_wants->last_id), "%s", blob); blob_wants->wants_sent++; } } sqlite3_finalize(statement); } } static void _tf_ssb_rpc_blobs_createWants(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_blob_wants_t* blob_wants = tf_ssb_connection_get_blob_wants_state(connection); tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); tf_ssb_add_blob_want_added_callback(ssb, _tf_ssb_rpc_blob_wants_added_callback, NULL, connection); blob_wants->request_number = request_number; _tf_ssb_rpc_request_more_blobs(connection); } typedef struct tunnel_t { tf_ssb_connection_t* connection; int32_t request_number; } tunnel_t; void _tf_ssb_rpc_tunnel_callback(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data) { tunnel_t* tun = user_data; tf_ssb_connection_rpc_send(tun->connection, flags, tun->request_number, message, size, NULL, NULL, NULL); } void _tf_ssb_rpc_tunnel_cleanup(tf_ssb_t* ssb, void* user_data) { tf_free(user_data); } static bool _get_global_setting_bool(tf_ssb_t* ssb, const char* name, bool default_value) { bool result = default_value; JSContext* context = tf_ssb_get_context(ssb); sqlite3* db = tf_ssb_get_db(ssb); sqlite3_stmt* statement; if (sqlite3_prepare(db, "SELECT value FROM properties WHERE id = 'core' AND key = 'settings'", -1, &statement, NULL) == SQLITE_OK) { if (sqlite3_step(statement) == SQLITE_ROW) { JSValue value = JS_ParseJSON(context, (const char*)sqlite3_column_text(statement, 0), sqlite3_column_bytes(statement, 0), NULL); JSValue property = JS_GetPropertyStr(context, value, name); if (JS_IsBool(property)) { result = JS_ToBool(context, property); } JS_FreeValue(context, property); JS_FreeValue(context, value); } sqlite3_finalize(statement); } return result; } static bool _get_global_setting_string(tf_ssb_t* ssb, const char* name, char* out_value, size_t size) { bool result = false; JSContext* context = tf_ssb_get_context(ssb); sqlite3* db = tf_ssb_get_db(ssb); sqlite3_stmt* statement; if (sqlite3_prepare(db, "SELECT value FROM properties WHERE id = 'core' AND key = 'settings'", -1, &statement, NULL) == SQLITE_OK) { if (sqlite3_step(statement) == SQLITE_ROW) { JSValue value = JS_ParseJSON(context, (const char*)sqlite3_column_text(statement, 0), sqlite3_column_bytes(statement, 0), NULL); JSValue property = JS_GetPropertyStr(context, value, name); const char* value_string = JS_ToCString(context, property); if (value_string) { snprintf(out_value, size, "%s", value_string); result = true; JS_FreeCString(context, value_string); } JS_FreeValue(context, property); JS_FreeValue(context, value); } sqlite3_finalize(statement); } return result; } static void _tf_ssb_rpc_tunnel_connect(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 (!_get_global_setting_bool(ssb, "room", true)) { tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number); return; } JSContext* context = tf_ssb_connection_get_context(connection); JSValue arg_array = JS_GetPropertyStr(context, args, "args"); JSValue arg = JS_GetPropertyUint32(context, arg_array, 0); JSValue origin = JS_GetPropertyStr(context, arg, "origin"); JSValue portal = JS_GetPropertyStr(context, arg, "portal"); JSValue target = JS_GetPropertyStr(context, arg, "target"); if (JS_IsUndefined(origin) && !JS_IsUndefined(portal) && !JS_IsUndefined(target)) { const char* target_str = JS_ToCString(context, target); tf_ssb_connection_t* target_connection = tf_ssb_connection_get(ssb, target_str); if (target_connection) { int32_t tunnel_request_number = tf_ssb_connection_next_request_number(target_connection); const char* portal_str = JS_ToCString(context, portal); JSValue message = JS_NewObject(context); JSValue name = JS_NewArray(context); JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "tunnel")); JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "connect")); JS_SetPropertyStr(context, message, "name", name); JSValue arg_obj = JS_NewObject(context); char origin_str[k_id_base64_len] = ""; tf_ssb_connection_get_id(connection, origin_str, sizeof(origin_str)); JS_SetPropertyStr(context, arg_obj, "origin", JS_NewString(context, origin_str)); JS_SetPropertyStr(context, arg_obj, "portal", JS_NewString(context, portal_str)); JS_SetPropertyStr(context, arg_obj, "target", JS_NewString(context, target_str)); JSValue arg_array = JS_NewArray(context); JS_SetPropertyUint32(context, arg_array, 0, arg_obj); JS_SetPropertyStr(context, message, "args", arg_array); JS_SetPropertyStr(context, message, "type", JS_NewString(context, "duplex")); tf_ssb_connection_rpc_send_json( target_connection, k_ssb_rpc_flag_stream, tunnel_request_number, message, NULL, NULL, NULL); tunnel_t* data0 = tf_malloc(sizeof(tunnel_t)); *data0 = (tunnel_t) { .connection = target_connection, .request_number = tunnel_request_number, }; tunnel_t* data1 = tf_malloc(sizeof(tunnel_t)); *data1 = (tunnel_t) { .connection = connection, .request_number = -request_number, }; tf_ssb_connection_add_request(connection, -request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data0, target_connection); tf_ssb_connection_add_request(target_connection, tunnel_request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data1, connection); JS_FreeValue(context, message); JS_FreeCString(context, portal_str); } JS_FreeCString(context, target_str); } else if (!JS_IsUndefined(origin) && !JS_IsUndefined(portal) && !JS_IsUndefined(target)) { const char* origin_str = JS_ToCString(context, origin); const char* portal_str = JS_ToCString(context, portal); const char* target_str = JS_ToCString(context, target); printf("TUNNEL CREATE\n"); tf_ssb_connection_tunnel_create(ssb, portal_str, -request_number, origin_str); JS_FreeCString(context, origin_str); JS_FreeCString(context, portal_str); JS_FreeCString(context, target_str); } JS_FreeValue(context, origin); JS_FreeValue(context, portal); JS_FreeValue(context, target); JS_FreeValue(context, arg); JS_FreeValue(context, arg_array); } static void _tf_ssb_rpc_room_meta(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); JSContext* context = tf_ssb_get_context(ssb); JSValue response = JS_FALSE; if (_get_global_setting_bool(ssb, "room", true)) { char room_name[1024] = "tilde friends tunnel"; _get_global_setting_string(ssb, "room_name", room_name, sizeof(room_name)); response = JS_NewObject(context); JS_SetPropertyStr(context, response, "name", JS_NewString(context, room_name)); JS_SetPropertyStr(context, response, "membership", JS_FALSE); JSValue features = JS_NewArray(context); JS_SetPropertyUint32(context, features, 0, JS_NewString(context, "tunnel")); JS_SetPropertyUint32(context, features, 1, JS_NewString(context, "room1")); JS_SetPropertyUint32(context, features, 2, JS_NewString(context, "room2")); JS_SetPropertyStr(context, response, "features", features); } tf_ssb_connection_rpc_send_json( connection, flags, -request_number, response, NULL, NULL, NULL); JS_FreeValue(context, response); } static void _tf_ssb_rpc_room_attendants_callback(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data) { printf("room attendants callback??\n"); } static void _tf_ssb_rpc_room_attendants(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 (!_get_global_setting_bool(ssb, "room", true)) { tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number); return; } JSContext* context = tf_ssb_get_context(ssb); JSValue joined = JS_NewObject(context); JS_SetPropertyStr(context, joined, "type", JS_NewString(context, "joined")); char my_id[k_id_base64_len] = ""; if (tf_ssb_connection_get_id(connection, my_id, sizeof(my_id))) { JS_SetPropertyStr(context, joined, "id", JS_NewString(context, my_id)); } JSValue state = JS_NewObject(context); JS_SetPropertyStr(context, state, "type", JS_NewString(context, "state")); JSValue ids = JS_NewArray(context); int id_count = 0; tf_ssb_connection_t* connections[1024]; int count = tf_ssb_get_connections(ssb, connections, _countof(connections)); for (int i = 0; i < count; i++) { char id[k_id_base64_len] = { 0 }; if (tf_ssb_connection_is_attendant(connections[i]) && tf_ssb_connection_get_id(connections[i], id, sizeof(id))) { JS_SetPropertyUint32(context, ids, id_count++, JS_NewString(context, id)); tf_ssb_connection_rpc_send_json( connections[i], flags, -tf_ssb_connection_get_attendant_request_number(connections[i]), joined, NULL, NULL, NULL); } } JS_SetPropertyStr(context, state, "ids", ids); tf_ssb_connection_rpc_send_json(connection, flags, -request_number, state, NULL, NULL, NULL); JS_FreeValue(context, joined); JS_FreeValue(context, state); tf_ssb_connection_set_attendant(connection, true, request_number); tf_ssb_connection_add_request(connection, -request_number, _tf_ssb_rpc_room_attendants_callback, NULL, NULL, NULL); } typedef struct _blobs_get_t { char id[k_blob_id_len]; size_t received; size_t expected_size; char buffer[]; } blobs_get_t; static void _tf_ssb_rpc_connection_blobs_get_callback(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); JSContext* context = tf_ssb_get_context(ssb); blobs_get_t* get = user_data; if ((flags & k_ssb_rpc_mask_type) == k_ssb_rpc_flag_binary && size > 0 && get->received + size <= get->expected_size) { memcpy(get->buffer + get->received, message, size); get->received += size; } else if ((flags & k_ssb_rpc_mask_type) == k_ssb_rpc_flag_json) { bool stored = false; if (JS_ToBool(context, args)) { char id[256]; stored = tf_ssb_db_blob_store(ssb, (uint8_t*)get->buffer, get->expected_size, id, sizeof(id), NULL); } tf_ssb_connection_rpc_send( connection, k_ssb_rpc_flag_json | k_ssb_rpc_flag_stream | k_ssb_rpc_flag_end_error, -request_number, (const uint8_t*)(stored ? "true" : "false"), strlen(stored ? "true" : "false"), NULL, NULL, NULL); } } static void _tf_ssb_rpc_connection_blobs_get_cleanup(tf_ssb_t* ssb, void* user_data) { tf_free(user_data); } static void _tf_ssb_rpc_connection_blobs_get(tf_ssb_connection_t* connection, const char* blob_id, size_t size) { blobs_get_t* get = tf_malloc(sizeof(blobs_get_t) + size); *get = (blobs_get_t) { .expected_size = size }; snprintf(get->id, sizeof(get->id), "%s", blob_id); memset(get->buffer, 0, size); tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_get_context(ssb); JSValue message = JS_NewObject(context); JSValue name = JS_NewArray(context); JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "blobs")); JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "get")); JS_SetPropertyStr(context, message, "name", name); JS_SetPropertyStr(context, message, "type", JS_NewString(context, "source")); JSValue args = JS_NewArray(context); JS_SetPropertyUint32(context, args, 0, JS_NewString(context, blob_id)); JS_SetPropertyStr(context, message, "args", args); tf_ssb_connection_rpc_send_json( connection, k_ssb_rpc_flag_stream, tf_ssb_connection_next_request_number(connection), message, _tf_ssb_rpc_connection_blobs_get_callback, _tf_ssb_rpc_connection_blobs_get_cleanup, get); JS_FreeValue(context, message); } static void _tf_ssb_rpc_connection_blobs_createWants_callback(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_blob_wants_t* blob_wants = tf_ssb_connection_get_blob_wants_state(connection); if (!blob_wants) { return; } tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_connection_get_context(connection); JSValue name = JS_GetPropertyStr(context, args, "name"); if (!JS_IsUndefined(name)) { /* { name: "Error" } */ JS_FreeValue(context, name); return; } JSPropertyEnum* ptab; uint32_t plen; JS_GetOwnPropertyNames(context, &ptab, &plen, args, JS_GPN_STRING_MASK); 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* blob_id = JS_ToCString(context, key); int64_t size = 0; JS_ToInt64(context, &size, key_value); if (--blob_wants->wants_sent == 0) { _tf_ssb_rpc_request_more_blobs(connection); } if (size < 0) { size_t blob_size = 0; if (tf_ssb_db_blob_get(ssb, blob_id, NULL, &blob_size)) { JSValue message = JS_NewObject(context); JS_SetPropertyStr(context, message, blob_id, JS_NewInt64(context, blob_size)); tf_ssb_connection_rpc_send_json( connection, k_ssb_rpc_flag_stream, -blob_wants->request_number, message, NULL, NULL, NULL); JS_FreeValue(context, message); } else if (size == -1LL) { JSValue message = JS_NewObject(context); JS_SetPropertyStr(context, message, blob_id, JS_NewInt64(context, -2)); tf_ssb_connection_rpc_send_json( connection, k_ssb_rpc_flag_stream, -blob_wants->request_number, message, NULL, NULL, NULL); JS_FreeValue(context, message); } } else { _tf_ssb_rpc_connection_blobs_get(connection, blob_id, size); } JS_FreeCString(context, blob_id); 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); } static void _tf_ssb_rpc_connection_room_attendants_callback(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); JSContext* context = tf_ssb_get_context(ssb); JSValue type = JS_GetPropertyStr(context, args, "type"); const char* type_string = JS_ToCString(context, type); if (!type_string) { tf_ssb_connection_rpc_send_error(connection, flags, -request_number, "Missing type."); } else if (strcmp(type_string, "state") == 0) { tf_ssb_connection_clear_room_attendants(connection); JSValue ids = JS_GetPropertyStr(context, args, "ids"); int length = tf_util_get_length(context, ids); for (int i = 0; i < length; i++) { JSValue id = JS_GetPropertyUint32(context, ids, i); const char* id_string = JS_ToCString(context, id); if (id_string) { tf_ssb_connection_add_room_attendant(connection, id_string); } JS_FreeCString(context, id_string); JS_FreeValue(context, id); } JS_FreeValue(context, ids); } else if (strcmp(type_string, "joined") == 0) { JSValue id = JS_GetPropertyStr(context, args, "id"); const char* id_string = JS_ToCString(context, id); if (id_string) { tf_ssb_connection_add_room_attendant(connection, id_string); } JS_FreeCString(context, id_string); JS_FreeValue(context, id); } else if (strcmp(type_string, "left") == 0) { JSValue id = JS_GetPropertyStr(context, args, "id"); const char* id_string = JS_ToCString(context, id); if (id_string) { tf_ssb_connection_remove_room_attendant(connection, id_string); } JS_FreeCString(context, id_string); JS_FreeValue(context, id); } else { tf_ssb_connection_rpc_send_error(connection, flags, -request_number, "Unexpected room.attendants response type."); } JS_FreeCString(context, type_string); JS_FreeValue(context, type); } static bool _is_error(JSContext* context, JSValue message) { JSValue name = JS_GetPropertyStr(context, message, "name"); const char* name_string = JS_ToCString(context, name); bool is_error = false; if (name_string && strcmp(name_string, "Error") == 0) { is_error = true; } JS_FreeCString(context, name_string); JS_FreeValue(context, name); return is_error; } static void _tf_ssb_rpc_connection_tunnel_isRoom_callback(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); JSContext* context = tf_ssb_get_context(ssb); if (_is_error(context, args)) { return; } if (JS_IsObject(args)) { JSValue message = JS_NewObject(context); JSValue name = JS_NewArray(context); JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "room")); JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "attendants")); JS_SetPropertyStr(context, message, "name", name); JS_SetPropertyStr(context, message, "type", JS_NewString(context, "source")); JS_SetPropertyStr(context, message, "args", JS_NewArray(context)); tf_ssb_connection_rpc_send_json( connection, k_ssb_rpc_flag_stream, tf_ssb_connection_next_request_number(connection), message, _tf_ssb_rpc_connection_room_attendants_callback, NULL, NULL); JS_FreeValue(context, message); } } 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) { tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_get_context(ssb); sqlite3* db = tf_ssb_get_db(ssb); sqlite3_stmt* statement; if (sqlite3_prepare(db, "SELECT previous, author, id, sequence, timestamp, hash, content, signature, sequence_before_author FROM messages WHERE author = ?1 AND sequence >= ?2 ORDER BY sequence", -1, &statement, NULL) == SQLITE_OK) { if (sqlite3_bind_text(statement, 1, author, -1, NULL) == SQLITE_OK && sqlite3_bind_int64(statement, 2, sequence) == SQLITE_OK) { while (sqlite3_step(statement) == SQLITE_ROW) { JSValue message = JS_UNDEFINED; JSValue formatted = tf_ssb_format_message( context, (const char*)sqlite3_column_text(statement, 0), (const char*)sqlite3_column_text(statement, 1), sqlite3_column_int64(statement, 3), sqlite3_column_double(statement, 4), (const char*)sqlite3_column_text(statement, 5), (const char*)sqlite3_column_text(statement, 6), (const char*)sqlite3_column_text(statement, 7), sqlite3_column_int(statement, 8)); if (keys) { message = JS_NewObject(context); JS_SetPropertyStr(context, message, "key", JS_NewString(context, (const char*)sqlite3_column_text(statement, 2))); JS_SetPropertyStr(context, message, "value", formatted); JS_SetPropertyStr(context, message, "timestamp", JS_NewString(context, (const char*)sqlite3_column_text(statement, 4))); } else { message = formatted; } tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream, request_number, message, NULL, NULL, NULL); JS_FreeValue(context, message); } } sqlite3_finalize(statement); } } static void _tf_ssb_rpc_createHistoryStream(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); JSContext* context = tf_ssb_get_context(ssb); JSValue arg_array = JS_GetPropertyStr(context, args, "args"); JSValue arg = JS_GetPropertyUint32(context, arg_array, 0); if (JS_IsUndefined(arg)) { tf_ssb_connection_rpc_send_error(connection, flags, -request_number, "Missing request.args in createHistoryStream."); } JSValue id = JS_GetPropertyStr(context, arg, "id"); JSValue seq = JS_GetPropertyStr(context, arg, "seq"); JSValue keys = JS_GetPropertyStr(context, arg, "keys"); JSValue live = JS_GetPropertyStr(context, arg, "live"); bool is_keys = JS_IsUndefined(keys) || JS_ToBool(context, keys) > 0; bool is_live = JS_ToBool(context, live) > 0; int64_t sequence = 0; JS_ToInt64(context, &sequence, seq); const char* author = JS_ToCString(context, id); _tf_ssb_connection_send_history_stream(connection, -request_number, author, sequence, is_keys); if (is_live) { tf_ssb_connection_add_new_message_request(connection, author, -request_number, is_keys); } JS_FreeCString(context, author); JS_FreeValue(context, id); JS_FreeValue(context, seq); JS_FreeValue(context, keys); JS_FreeValue(context, arg); JS_FreeValue(context, arg_array); } static void _tf_ssb_rpc_ebt_replicate_send_clock(tf_ssb_connection_t* connection, int32_t request_number, JSValue message) { tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_get_context(ssb); JSValue full_clock = JS_NewObject(context); /* Ask for every identity we know is being followed from local accounts. */ const char** visible = tf_ssb_db_get_all_visible_identities(ssb, 2); for (int i = 0; visible[i]; i++) { int64_t sequence = 0; tf_ssb_db_get_latest_message_by_author(ssb, visible[i], &sequence, NULL, 0); JS_SetPropertyStr(context, full_clock, visible[i], JS_NewInt64(context, sequence == -1 ? -1 : (sequence << 1))); } /* Ask about the incoming connection, too. */ char id[k_id_base64_len] = ""; if (tf_ssb_connection_get_id(connection, id, sizeof(id))) { JSValue in_clock = JS_GetPropertyStr(context, full_clock, id); if (JS_IsUndefined(in_clock)) { int64_t sequence = 0; tf_ssb_db_get_latest_message_by_author(ssb, id, &sequence, NULL, 0); JS_SetPropertyStr(context, full_clock, id, JS_NewInt64(context, sequence == -1 ? -1 : (sequence << 1))); } JS_FreeValue(context, in_clock); } /* Also respond with what we know about all requested identities. */ if (!JS_IsUndefined(message)) { JSPropertyEnum* ptab; uint32_t plen; JS_GetOwnPropertyNames(context, &ptab, &plen, message, JS_GPN_STRING_MASK); for (uint32_t i = 0; i < plen; ++i) { JSValue in_clock = JS_GetProperty(context, full_clock, ptab[i].atom); if (JS_IsUndefined(in_clock)) { JSValue key = JS_AtomToString(context, ptab[i].atom); const char* key_string = JS_ToCString(context, key); if (key_string) { int64_t sequence = -1; tf_ssb_db_get_latest_message_by_author(ssb, key_string, &sequence, NULL, 0); JS_SetPropertyStr(context, full_clock, key_string, JS_NewInt64(context, sequence == -1 ? -1 : (sequence << 1))); } JS_FreeCString(context, key_string); JS_FreeValue(context, key); } } for (uint32_t i = 0; i < plen; ++i) { JS_FreeAtom(context, ptab[i].atom); } js_free(context, ptab); } tf_free(visible); /* TODO: Send it in bite-size chunks. */ tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream, -request_number, full_clock, NULL, NULL, NULL); JS_FreeValue(context, full_clock); } static void _tf_ssb_rpc_ebt_replicate_send_messages(tf_ssb_connection_t* connection, JSValue message) { if (JS_IsUndefined(message)) { return; } tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_get_context(ssb); JSPropertyEnum* ptab = NULL; uint32_t plen = 0; JS_GetOwnPropertyNames(context, &ptab, &plen, message, JS_GPN_STRING_MASK); for (uint32_t i = 0; i < plen; ++i) { JSValue in_clock = JS_GetProperty(context, message, ptab[i].atom); if (!JS_IsUndefined(in_clock)) { JSValue key = JS_AtomToString(context, ptab[i].atom); int64_t sequence = -1; JS_ToInt64(context, &sequence, in_clock); const char* author = JS_ToCString(context, key); if (sequence >= 0 && (sequence & 1) == 0) { int32_t request_number = tf_ssb_connection_get_ebt_request_number(connection); _tf_ssb_connection_send_history_stream(connection, request_number, author, sequence >> 1, false); tf_ssb_connection_add_new_message_request(connection, author, request_number, false); } else { tf_ssb_connection_remove_new_message_request(connection, author); } JS_FreeCString(context, author); JS_FreeValue(context, key); } JS_FreeValue(context, in_clock); } for (uint32_t i = 0; i < plen; ++i) { JS_FreeAtom(context, ptab[i].atom); } js_free(context, ptab); } static void _tf_ssb_rpc_ebt_replicate(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); JSContext* context = tf_ssb_get_context(ssb); if (_is_error(context, args)) { /* TODO: Send createHistoryStream. */ return; } if (!tf_ssb_connection_get_ebt_request_number(connection)) { tf_ssb_connection_set_ebt_request_number(connection, -request_number); } JSValue author = JS_GetPropertyStr(context, args, "author"); JSValue name = JS_GetPropertyStr(context, args, "name"); JSValue in_clock = JS_IsUndefined(name) ? args : JS_UNDEFINED; if (!JS_IsUndefined(author)) { /* Looks like a message. */ tf_ssb_verify_strip_and_store_message(ssb, args); } else { /* EBT clock. */ tf_ssb_connection_set_ebt_send_clock(connection, args); if (!tf_ssb_connection_get_sent_clock(connection)) { _tf_ssb_rpc_ebt_replicate_send_clock(connection, request_number, in_clock); tf_ssb_connection_set_sent_clock(connection, true); } _tf_ssb_rpc_ebt_replicate_send_messages(connection, in_clock); } JS_FreeValue(context, name); JS_FreeValue(context, author); } static void _tf_ssb_rpc_ebt_replicate_client(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_rpc_ebt_replicate(connection, flags, request_number, args, message, size, user_data); } static void _tf_ssb_rpc_send_ebt_replicate(tf_ssb_connection_t* connection) { tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); JSContext* context = tf_ssb_get_context(ssb); JSValue message = JS_NewObject(context); JSValue name = JS_NewArray(context); JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "ebt")); JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "replicate")); JS_SetPropertyStr(context, message, "name", name); JSValue arg = JS_NewObject(context); JS_SetPropertyStr(context, arg, "version", JS_NewInt32(context, 3)); JS_SetPropertyStr(context, arg, "format", JS_NewString(context, "classic")); JSValue args = JS_NewArray(context); JS_SetPropertyUint32(context, args, 0, arg); JS_SetPropertyStr(context, message, "args", args); JS_SetPropertyStr(context, message, "type", JS_NewString(context, "duplex")); int32_t request_number = tf_ssb_connection_next_request_number(connection); tf_ssb_connection_rpc_send_json( connection, k_ssb_rpc_flag_stream, request_number, message, _tf_ssb_rpc_ebt_replicate_client, NULL, NULL); if (!tf_ssb_connection_get_ebt_request_number(connection)) { tf_ssb_connection_set_ebt_request_number(connection, request_number); } JS_FreeValue(context, message); } static void _tf_ssb_rpc_ebt_replicate_server(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data) { if (flags & k_ssb_rpc_flag_end_error) { return; } _tf_ssb_rpc_ebt_replicate(connection, flags, request_number, args, message, size, user_data); tf_ssb_connection_add_request(connection, -request_number, _tf_ssb_rpc_ebt_replicate, NULL, NULL, NULL); } static void _tf_ssb_rpc_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_change_t change, tf_ssb_connection_t* connection, void* user_data) { JSContext* context = tf_ssb_get_context(ssb); if (change == k_tf_ssb_change_connect) { JSValue message = JS_NewObject(context); JSValue name = JS_NewArray(context); JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "blobs")); JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "createWants")); JS_SetPropertyStr(context, message, "name", name); JS_SetPropertyStr(context, message, "type", JS_NewString(context, "source")); JS_SetPropertyStr(context, message, "args", JS_NewArray(context)); tf_ssb_connection_rpc_send_json( connection, k_ssb_rpc_flag_stream, tf_ssb_connection_next_request_number(connection), message, _tf_ssb_rpc_connection_blobs_createWants_callback, NULL, NULL); JS_FreeValue(context, message); if (tf_ssb_connection_is_client(connection)) { message = JS_NewObject(context); name = JS_NewArray(context); JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "tunnel")); JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "isRoom")); JS_SetPropertyStr(context, message, "name", name); JS_SetPropertyStr(context, message, "args", JS_NewArray(context)); tf_ssb_connection_rpc_send_json( connection, 0, tf_ssb_connection_next_request_number(connection), message, _tf_ssb_rpc_connection_tunnel_isRoom_callback, NULL, NULL); JS_FreeValue(context, message); _tf_ssb_rpc_send_ebt_replicate(connection); } } else if (change == k_tf_ssb_change_remove) { tf_ssb_remove_blob_want_added_callback(ssb, _tf_ssb_rpc_blob_wants_added_callback, connection); char id[k_id_base64_len] = ""; if (tf_ssb_connection_get_id(connection, id, sizeof(id))) { JSValue left = JS_NewObject(context); JS_SetPropertyStr(context, left, "type", JS_NewString(context, "left")); JS_SetPropertyStr(context, left, "id", JS_NewString(context, id)); tf_ssb_connection_t* connections[1024]; int count = tf_ssb_get_connections(ssb, connections, _countof(connections)); for (int i = 0; i < count; i++) { if (tf_ssb_connection_is_attendant(connections[i])) { tf_ssb_connection_rpc_send_json( connections[i], k_ssb_rpc_flag_stream, -tf_ssb_connection_get_attendant_request_number(connections[i]), left, NULL, NULL, NULL); } } JS_FreeValue(context, left); } } } void tf_ssb_rpc_register(tf_ssb_t* ssb) { tf_ssb_add_connections_changed_callback(ssb, _tf_ssb_rpc_connections_changed_callback, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "gossip", "ping", NULL }, _tf_ssb_rpc_gossip_ping, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "blobs", "get", NULL }, _tf_ssb_rpc_blobs_get, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "blobs", "has", NULL }, _tf_ssb_rpc_blobs_has, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "blobs", "createWants", NULL }, _tf_ssb_rpc_blobs_createWants, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "tunnel", "connect", NULL }, _tf_ssb_rpc_tunnel_connect, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "tunnel", "isRoom", NULL }, _tf_ssb_rpc_room_meta, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "room", "metadata", NULL }, _tf_ssb_rpc_room_meta, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "room", "attendants", NULL }, _tf_ssb_rpc_room_attendants, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "createHistoryStream", NULL }, _tf_ssb_rpc_createHistoryStream, NULL, NULL); tf_ssb_add_rpc_callback(ssb, (const char*[]) { "ebt", "replicate", NULL }, _tf_ssb_rpc_ebt_replicate_server, NULL, NULL); }