diff --git a/src/ssb.c b/src/ssb.c index 943f4f7d..c2d48889 100644 --- a/src/ssb.c +++ b/src/ssb.c @@ -237,6 +237,8 @@ typedef struct _tf_ssb_t void (*hitch_callback)(const char* name, uint64_t duration, void* user_data); void* hitch_user_data; + + tf_ssb_store_queue_t store_queue; } tf_ssb_t; typedef struct _tf_ssb_connection_message_request_t @@ -1666,7 +1668,7 @@ static bool _tf_ssb_connection_box_stream_recv(tf_ssb_connection_t* connection) return true; } -bool tf_ssb_append_message_with_keys(tf_ssb_t* ssb, const char* author, const uint8_t* private_key, JSValue message, char* out_id, size_t out_id_size) +JSValue tf_ssb_sign_message(tf_ssb_t* ssb, const char* author, const uint8_t* private_key, JSValue message) { char previous_id[crypto_hash_sha256_BYTES * 2]; int64_t previous_sequence = 0; @@ -1675,14 +1677,7 @@ bool tf_ssb_append_message_with_keys(tf_ssb_t* ssb, const char* author, const ui JSContext* context = ssb->context; JSValue root = JS_NewObject(context); - if (have_previous) - { - JS_SetPropertyStr(context, root, "previous", JS_NewString(context, previous_id)); - } - else - { - JS_SetPropertyStr(context, root, "previous", JS_NULL); - } + JS_SetPropertyStr(context, root, "previous", have_previous ? JS_NewString(context, previous_id) : JS_NULL); JS_SetPropertyStr(context, root, "author", JS_NewString(context, author)); JS_SetPropertyStr(context, root, "sequence", JS_NewInt64(context, previous_sequence + 1)); @@ -1696,53 +1691,29 @@ bool tf_ssb_append_message_with_keys(tf_ssb_t* ssb, const char* author, const ui JSValue jsonval = JS_JSONStringify(context, root, JS_NULL, JS_NewInt32(context, 2)); size_t len = 0; const char* json = JS_ToCStringLen(context, &len, jsonval); + JS_FreeValue(context, jsonval); uint8_t signature[crypto_sign_BYTES]; unsigned long long siglen; bool valid = crypto_sign_detached(signature, &siglen, (const uint8_t*)json, len, private_key) == 0; + JS_FreeCString(context, json); + if (!valid) { tf_printf("crypto_sign_detached failed\n"); - } - - char signature_base64[crypto_sign_BYTES * 2]; - tf_base64_encode(signature, sizeof(signature), signature_base64, sizeof(signature_base64)); - strcat(signature_base64, ".sig.ed25519"); - JSValue sigstr = JS_NewString(context, signature_base64); - JS_SetPropertyStr(context, root, "signature", sigstr); - - bool stored = false; - char id[sodium_base64_ENCODED_LEN(crypto_hash_sha256_BYTES, sodium_base64_VARIANT_ORIGINAL) + 7 + 1]; - if (valid && tf_ssb_verify_and_strip_signature(ssb->context, root, id, sizeof(id), NULL, 0, NULL)) - { - if (tf_ssb_db_store_message(ssb, ssb->context, id, root, signature_base64, false)) - { - tf_ssb_notify_message_added(ssb, id); - snprintf(out_id, out_id_size, "%s", id); - stored = true; - } - else - { - tf_printf("message not stored.\n"); - } + JS_FreeValue(context, root); + root = JS_UNDEFINED; } else { - tf_printf("Failed to verify message signature.\n"); - tf_printf("json = %s\n", json); - tf_printf("sig = %s\n", signature_base64); + char signature_base64[crypto_sign_BYTES * 2]; + tf_base64_encode(signature, sizeof(signature), signature_base64, sizeof(signature_base64)); + strcat(signature_base64, ".sig.ed25519"); + JSValue sigstr = JS_NewString(context, signature_base64); + JS_SetPropertyStr(context, root, "signature", sigstr); } - if (!stored && out_id && out_id_size) - { - *out_id = '\0'; - } - - JS_FreeCString(context, json); - JS_FreeValue(context, jsonval); - - JS_FreeValue(context, root); - return stored; + return root; } static void _tf_ssb_connection_dispatch_scheduled(tf_ssb_connection_t* connection) @@ -2123,7 +2094,7 @@ tf_ssb_t* tf_ssb_create(uv_loop_t* loop, JSContext* context, const char* db_path JS_NewClass(JS_GetRuntime(ssb->context), _connection_class_id, &def); ssb->db_path = tf_strdup(db_path); - sqlite3_open_v2(db_path, &ssb->db_writer, SQLITE_OPEN_READWRITE | SQLITE_OPEN_URI, NULL); + sqlite3_open_v2(db_path, &ssb->db_writer, SQLITE_OPEN_READWRITE | SQLITE_OPEN_URI | SQLITE_OPEN_CREATE, NULL); tf_ssb_db_init(ssb); if (loop) @@ -3419,32 +3390,54 @@ tf_ssb_blob_wants_t* tf_ssb_connection_get_blob_wants_state(tf_ssb_connection_t* return connection ? &connection->blob_wants : NULL; } -bool tf_ssb_verify_strip_and_store_message(tf_ssb_t* ssb, JSValue value, bool* out_is_new) +typedef struct _store_t +{ + tf_ssb_t* ssb; + bool verified; + bool stored; + char id[crypto_hash_sha256_BYTES * 2 + 1]; + + tf_ssb_verify_strip_store_callback_t* callback; + void* user_data; +} store_t; + +static void _tf_ssb_verify_strip_and_store_finish(store_t* store) +{ + if (store->callback) + { + store->callback(store->id, store->verified, store->stored, store->user_data); + } + tf_free(store); +} + +static void _tf_ssb_verify_strip_and_store_callback(const char* id, bool stored, void* user_data) +{ + store_t* store = user_data; + store->stored = stored; + _tf_ssb_verify_strip_and_store_finish(store); +} + +void tf_ssb_verify_strip_and_store_message(tf_ssb_t* ssb, JSValue value, tf_ssb_verify_strip_store_callback_t* callback, void* user_data) { JSContext* context = tf_ssb_get_context(ssb); + store_t* async = tf_malloc(sizeof(store_t)); + *async = (store_t) + { + .ssb = ssb, + .callback = callback, + .user_data = user_data, + }; char signature[crypto_sign_BYTES + 128] = { 0 }; - char id[crypto_hash_sha256_BYTES * 2 + 1] = { 0 }; bool sequence_before_author = false; - if (out_is_new) + if (tf_ssb_verify_and_strip_signature(context, value, async->id, sizeof(async->id), signature, sizeof(signature), &sequence_before_author)) { - *out_is_new = false; - } - if (tf_ssb_verify_and_strip_signature(context, value, id, sizeof(id), signature, sizeof(signature), &sequence_before_author)) - { - if (tf_ssb_db_store_message(ssb, context, id, value, signature, sequence_before_author)) - { - tf_ssb_notify_message_added(ssb, id); - if (out_is_new) - { - *out_is_new = true; - } - } - return true; + async->verified = true; + tf_ssb_db_store_message(ssb, context, async->id, value, signature, sequence_before_author, _tf_ssb_verify_strip_and_store_callback, async); } else { - tf_printf("failed to verify message\n"); - return false; + printf("nope\n"); + _tf_ssb_verify_strip_and_store_finish(async); } } @@ -3555,3 +3548,8 @@ void tf_ssb_set_hitch_callback(tf_ssb_t* ssb, void (*callback)(const char* name, ssb->hitch_callback = callback; ssb->hitch_user_data = user_data; } + +tf_ssb_store_queue_t* tf_ssb_get_store_queue(tf_ssb_t* ssb) +{ + return &ssb->store_queue; +} diff --git a/src/ssb.db.c b/src/ssb.db.c index 95295413..c3b156ab 100644 --- a/src/ssb.db.c +++ b/src/ssb.db.c @@ -17,13 +17,15 @@ #include #include +#include + static void _tf_ssb_db_exec(sqlite3* db, const char* statement) { char* error = NULL; int result = sqlite3_exec(db, statement, NULL, NULL, &error); if (result != SQLITE_OK) { - tf_printf("Error running '%s': %s.\n", statement, error); + tf_printf("Error running '%s': %s.\n", statement, error ? error : sqlite3_errmsg(db)); abort(); } } @@ -253,41 +255,25 @@ static bool _tf_ssb_db_previous_message_exists(sqlite3* db, const char* author, return exists; } -bool tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, JSValue val, const char* signature, bool sequence_before_author) +static int64_t _tf_ssb_db_store_message_raw( + tf_ssb_t* ssb, + const char* id, + const char* previous, + const char* author, + int64_t sequence, + double timestamp, + const char* content, + size_t content_len, + const char* signature, + bool sequence_before_author) { - bool stored = false; - - JSValue previousval = JS_GetPropertyStr(context, val, "previous"); - const char* previous = JS_IsNull(previousval) ? NULL : JS_ToCString(context, previousval); - JS_FreeValue(context, previousval); - - JSValue authorval = JS_GetPropertyStr(context, val, "author"); - const char* author = JS_ToCString(context, authorval); - JS_FreeValue(context, authorval); - - int64_t sequence = -1; - JSValue sequenceval = JS_GetPropertyStr(context, val, "sequence"); - JS_ToInt64(context, &sequence, sequenceval); - JS_FreeValue(context, sequenceval); - sqlite3* db = tf_ssb_acquire_db_writer(ssb); - sqlite3_stmt* statement; int64_t last_row_id = -1; if (_tf_ssb_db_previous_message_exists(db, author, sequence, previous)) { - double timestamp = -1.0; - JSValue timestampval = JS_GetPropertyStr(context, val, "timestamp"); - JS_ToFloat64(context, ×tamp, timestampval); - JS_FreeValue(context, timestampval); - - JSValue contentval = JS_GetPropertyStr(context, val, "content"); - JSValue content = JS_JSONStringify(context, contentval, JS_NULL, JS_NULL); - size_t content_len; - const char* contentstr = JS_ToCStringLen(context, &content_len, content); - JS_FreeValue(context, contentval); - const char* query = "INSERT INTO messages (id, previous, author, sequence, timestamp, content, hash, signature, sequence_before_author) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING"; + sqlite3_stmt* statement; if (sqlite3_prepare(db, query, -1, &statement, NULL) == SQLITE_OK) { if (sqlite3_bind_text(statement, 1, id, -1, NULL) == SQLITE_OK && @@ -295,7 +281,7 @@ bool tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, sqlite3_bind_text(statement, 3, author, -1, NULL) == SQLITE_OK && sqlite3_bind_int64(statement, 4, sequence) == SQLITE_OK && sqlite3_bind_double(statement, 5, timestamp) == SQLITE_OK && - sqlite3_bind_text(statement, 6, contentstr, content_len, NULL) == SQLITE_OK && + sqlite3_bind_text(statement, 6, content, content_len, NULL) == SQLITE_OK && sqlite3_bind_text(statement, 7, "sha256", 6, NULL) == SQLITE_OK && sqlite3_bind_text(statement, 8, signature, -1, NULL) == SQLITE_OK && sqlite3_bind_int(statement, 9, sequence_before_author) == SQLITE_OK) @@ -303,10 +289,10 @@ bool tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, int r = sqlite3_step(statement); if (r != SQLITE_DONE) { - tf_printf("%s\n", sqlite3_errmsg(db)); + tf_printf("_tf_ssb_db_store_message_raw: %s\n", sqlite3_errmsg(db)); + abort(); } - stored = r == SQLITE_DONE && sqlite3_changes(db) != 0; - if (stored) + if (r == SQLITE_DONE && sqlite3_changes(db) != 0) { last_row_id = sqlite3_last_insert_rowid(db); } @@ -321,45 +307,211 @@ bool tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, { tf_printf("%s: prepare failed: %s\n", __FUNCTION__, sqlite3_errmsg(db)); } - - JS_FreeCString(context, contentstr); - JS_FreeValue(context, content); } else { - tf_printf("Previous message doesn't exist.\n"); - } - - if (last_row_id != -1) - { - const char* query = "SELECT DISTINCT json.value FROM messages, json_tree(messages.content) AS json LEFT OUTER JOIN blobs ON json.value = blobs.id WHERE messages.rowid = ?1 AND json.value LIKE '&%%.sha256' AND length(json.value) = ?2 AND blobs.content IS NULL"; - if (sqlite3_prepare(db, query, -1, &statement, NULL) == SQLITE_OK) - { - if (sqlite3_bind_int64(statement, 1, last_row_id) == SQLITE_OK && - sqlite3_bind_int(statement, 2, k_blob_id_len - 1) == SQLITE_OK) - { - int r = SQLITE_OK; - while ((r = sqlite3_step(statement)) == SQLITE_ROW) - { - tf_ssb_notify_blob_want_added(ssb, (const char*)sqlite3_column_text(statement, 0)); - } - if (r != SQLITE_DONE) - { - tf_printf("%s\n", sqlite3_errmsg(db)); - } - } - sqlite3_finalize(statement); - } - else - { - tf_printf("%s: prepare failed: %s\n", __FUNCTION__, sqlite3_errmsg(db)); - } + tf_printf("%p: Previous message doesn't exist for author=%s sequence=%" PRId64 ".\n", db, author, sequence); } tf_ssb_release_db_writer(ssb, db); + return last_row_id; +} +static char* _tf_ssb_db_get_message_blob_wants(tf_ssb_t* ssb, int64_t rowid) +{ + sqlite3* db = tf_ssb_acquire_db_reader(ssb); + sqlite3_stmt* statement; + char* result = NULL; + size_t size = 0; + + if (sqlite3_prepare(db, "SELECT DISTINCT json.value FROM messages, json_tree(messages.content) AS json LEFT OUTER JOIN blobs ON json.value = blobs.id WHERE messages.rowid = ?1 AND json.value LIKE '&%%.sha256' AND length(json.value) = ?2 AND blobs.content IS NULL", -1, &statement, NULL) == SQLITE_OK) + { + if (sqlite3_bind_int64(statement, 1, rowid) == SQLITE_OK && + sqlite3_bind_int(statement, 2, k_blob_id_len - 1) == SQLITE_OK) + { + int r = SQLITE_OK; + while ((r = sqlite3_step(statement)) == SQLITE_ROW) + { + int id_size = sqlite3_column_bytes(statement, 0); + const uint8_t* id = sqlite3_column_text(statement, 0); + result = tf_realloc(result, size + id_size + 1); + memcpy(result + size, id, id_size + 1); + size += id_size + 1; + } + if (r != SQLITE_DONE) + { + tf_printf("%s\n", sqlite3_errmsg(db)); + } + } + sqlite3_finalize(statement); + } + else + { + tf_printf("%s: prepare failed: %s\n", __FUNCTION__, sqlite3_errmsg(db)); + } + result = tf_realloc(result, size + 1); + result[size] = '\0'; + + tf_ssb_release_db_reader(ssb, db); + return result; +} + +typedef struct _message_store_t message_store_t; +typedef struct _message_store_t +{ + uv_work_t work; + tf_ssb_t* ssb; + char id[k_id_base64_len]; + char signature[512]; + bool sequence_before_author; + char previous[k_id_base64_len]; + char author[k_id_base64_len]; + int64_t sequence; + double timestamp; + const char* content; + size_t length; + + bool out_stored; + char* out_blob_wants; + + tf_ssb_db_store_message_callback_t* callback; + void* user_data; + + message_store_t* next; +} message_store_t; + +static void _tf_ssb_db_store_message_work(uv_work_t* work) +{ + message_store_t* store = work->data; + int64_t last_row_id = _tf_ssb_db_store_message_raw(store->ssb, store->id, *store->previous ? store->previous : NULL, store->author, store->sequence, store->timestamp, store->content, store->length, store->signature, store->sequence_before_author); + if (last_row_id != -1) + { + store->out_stored = true; + store->out_blob_wants = _tf_ssb_db_get_message_blob_wants(store->ssb, last_row_id); + } +} + +static void _tf_ssb_db_store_message_work_finish(message_store_t* store); +static void _tf_ssb_db_store_message_after_work(uv_work_t* work, int status); + +static void _wake_up_queue(tf_ssb_t* ssb, tf_ssb_store_queue_t* queue) +{ + if (!queue->running) + { + message_store_t* next = queue->head; + if (next) + { + queue->head = next->next; + if (queue->tail == next) + { + queue->tail = NULL; + } + next->next = NULL; + queue->running = true; + int r = uv_queue_work(tf_ssb_get_loop(ssb), &next->work, _tf_ssb_db_store_message_work, _tf_ssb_db_store_message_after_work); + if (r) + { + _tf_ssb_db_store_message_work_finish(next); + } + } + } +} + +static void _tf_ssb_db_store_message_work_finish(message_store_t* store) +{ + JSContext* context = tf_ssb_get_context(store->ssb); + if (store->callback) + { + store->callback(store->id, store->out_stored, store->user_data); + } + JS_FreeCString(context, store->content); + tf_ssb_store_queue_t* queue = tf_ssb_get_store_queue(store->ssb); + queue->running = false; + _wake_up_queue(store->ssb, queue); + tf_free(store); +} + +static void _tf_ssb_db_store_message_after_work(uv_work_t* work, int status) +{ + message_store_t* store = work->data; + if (store->out_stored) + { + tf_ssb_notify_message_added(store->ssb, store->id); + } + if (store->out_blob_wants) + { + for (char* p = store->out_blob_wants; *p; p = p + strlen(p)) + { + tf_ssb_notify_blob_want_added(store->ssb, p); + } + tf_free(store->out_blob_wants); + } + _tf_ssb_db_store_message_work_finish(store); +} + +void tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, JSValue val, const char* signature, bool sequence_before_author, tf_ssb_db_store_message_callback_t* callback, void* user_data) +{ + JSValue previousval = JS_GetPropertyStr(context, val, "previous"); + const char* previous = JS_IsNull(previousval) ? NULL : JS_ToCString(context, previousval); + JS_FreeValue(context, previousval); + + JSValue authorval = JS_GetPropertyStr(context, val, "author"); + const char* author = JS_ToCString(context, authorval); + JS_FreeValue(context, authorval); + + int64_t sequence = -1; + JSValue sequenceval = JS_GetPropertyStr(context, val, "sequence"); + JS_ToInt64(context, &sequence, sequenceval); + JS_FreeValue(context, sequenceval); + + double timestamp = -1.0; + JSValue timestampval = JS_GetPropertyStr(context, val, "timestamp"); + JS_ToFloat64(context, ×tamp, timestampval); + JS_FreeValue(context, timestampval); + + JSValue contentval = JS_GetPropertyStr(context, val, "content"); + JSValue content = JS_JSONStringify(context, contentval, JS_NULL, JS_NULL); + size_t content_len; + const char* contentstr = JS_ToCStringLen(context, &content_len, content); + JS_FreeValue(context, content); + JS_FreeValue(context, contentval); + + message_store_t* store = tf_malloc(sizeof(message_store_t)); + *store = (message_store_t) + { + .work = + { + .data = store, + }, + .ssb = ssb, + .sequence = sequence, + .timestamp = timestamp, + .content = contentstr, + .length = content_len, + .sequence_before_author = sequence_before_author, + + .callback = callback, + .user_data = user_data, + }; + snprintf(store->id, sizeof(store->id), "%s", id); + snprintf(store->previous, sizeof(store->previous), "%s", previous ? previous : ""); + snprintf(store->author, sizeof(store->author), "%s", author); + snprintf(store->signature, sizeof(store->signature), "%s", signature); JS_FreeCString(context, author); JS_FreeCString(context, previous); - return stored; + + tf_ssb_store_queue_t* queue = tf_ssb_get_store_queue(ssb); + if (queue->tail) + { + message_store_t* tail = queue->tail; + tail->next = store; + queue->tail = store; + } + else + { + queue->head = store; + queue->tail = store; + } + _wake_up_queue(ssb, queue); } bool tf_ssb_db_message_content_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size) @@ -812,7 +964,7 @@ JSValue tf_ssb_db_visit_query(tf_ssb_t* ssb, const char* query, const JSValue bi JSValue tf_ssb_format_message(JSContext* context, const char* previous, const char* author, int64_t sequence, double timestamp, const char* hash, const char* content, const char* signature, bool sequence_before_author) { JSValue value = JS_NewObject(context); - JS_SetPropertyStr(context, value, "previous", previous ? JS_NewString(context, previous) : JS_NULL); + JS_SetPropertyStr(context, value, "previous", (previous && *previous) ? JS_NewString(context, previous) : JS_NULL); if (sequence_before_author) { JS_SetPropertyStr(context, value, "sequence", JS_NewInt64(context, sequence)); diff --git a/src/ssb.db.h b/src/ssb.db.h index a7d3f99f..e7b26c72 100644 --- a/src/ssb.db.h +++ b/src/ssb.db.h @@ -10,11 +10,13 @@ typedef struct _tf_ssb_t tf_ssb_t; void tf_ssb_db_init(tf_ssb_t* ssb); void tf_ssb_db_init_reader(sqlite3* db); -bool tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, JSValue val, const char* signature, bool sequence_before_author); bool tf_ssb_db_message_content_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size); bool tf_ssb_db_blob_has(tf_ssb_t* ssb, const char* id); bool tf_ssb_db_blob_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size); +typedef void (tf_ssb_db_store_message_callback_t)(const char* id, bool stored, void* user_data); +void tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, JSValue val, const char* signature, bool sequence_before_author, tf_ssb_db_store_message_callback_t* callback, void* user_data); + typedef void (tf_ssb_db_blob_store_callback_t)(const char* id, bool is_new, void* user_data); void tf_ssb_db_blob_store_async(tf_ssb_t* ssb, const uint8_t* blob, size_t size, tf_ssb_db_blob_store_callback_t* callback, void* user_data); bool tf_ssb_db_blob_store(tf_ssb_t* ssb, const uint8_t* blob, size_t size, char* out_id, size_t out_id_size, bool* out_new); diff --git a/src/ssb.h b/src/ssb.h index 034b6db0..b0ab3013 100644 --- a/src/ssb.h +++ b/src/ssb.h @@ -66,6 +66,13 @@ typedef struct _tf_ssb_blob_wants_t int wants_sent; } tf_ssb_blob_wants_t; +typedef struct _tf_ssb_store_queue_t +{ + bool running; + void* head; + void* tail; +} tf_ssb_store_queue_t; + tf_ssb_t* tf_ssb_create(uv_loop_t* loop, JSContext* context, const char* db_path); void tf_ssb_destroy(tf_ssb_t* ssb); @@ -87,7 +94,7 @@ JSContext* tf_ssb_get_context(tf_ssb_t* ssb); void tf_ssb_broadcast_listener_start(tf_ssb_t* ssb, bool linger); void tf_ssb_broadcast_sender_start(tf_ssb_t* ssb); void tf_ssb_run(tf_ssb_t* ssb); -bool tf_ssb_append_message_with_keys(tf_ssb_t* ssb, const char* author, const uint8_t* private_key, JSValue message, char* out_id, size_t out_id_size); +JSValue tf_ssb_sign_message(tf_ssb_t* ssb, const char* author, const uint8_t* private_key, JSValue message); bool tf_ssb_whoami(tf_ssb_t* ssb, char* out_id, size_t out_id_size); void tf_ssb_visit_broadcasts(tf_ssb_t* ssb, void (*callback)(const char* host, const struct sockaddr_in* addr, tf_ssb_connection_t* tunnel, const uint8_t* pub, void* user_data), void* user_data); @@ -107,7 +114,8 @@ bool tf_ssb_id_bin_to_str(char* str, size_t str_size, const uint8_t* bin); bool tf_ssb_verify_and_strip_signature(JSContext* context, JSValue val, char* out_id, size_t out_id_size, char* out_signature, size_t out_signature_size, bool* out_sequence_before_author); void tf_ssb_calculate_message_id(JSContext* context, JSValue message, char* out_id, size_t out_id_size); -bool tf_ssb_verify_strip_and_store_message(tf_ssb_t* ssb, JSValue value, bool* out_is_new); +typedef void (tf_ssb_verify_strip_store_callback_t)(const char* id, bool verified, bool is_new, void* user_data); +void tf_ssb_verify_strip_and_store_message(tf_ssb_t* ssb, JSValue value, tf_ssb_verify_strip_store_callback_t* callback, void* user_data); bool tf_ssb_connection_is_client(tf_ssb_connection_t* connection); const char* tf_ssb_connection_get_host(tf_ssb_connection_t* connection); @@ -190,3 +198,5 @@ void tf_ssb_record_thread_time(tf_ssb_t* ssb, int64_t thread_id, uint64_t hrtime uint64_t tf_ssb_get_average_thread_time(tf_ssb_t* ssb); void tf_ssb_set_hitch_callback(tf_ssb_t* ssb, void (*callback)(const char* name, uint64_t duration_ns, void* user_data), void* user_data); + +tf_ssb_store_queue_t* tf_ssb_get_store_queue(tf_ssb_t* ssb); diff --git a/src/ssb.js.c b/src/ssb.js.c index 880fc4c2..b13a9edd 100644 --- a/src/ssb.js.c +++ b/src/ssb.js.c @@ -107,9 +107,39 @@ static JSValue _tf_ssb_getAllIdentities(JSContext* context, JSValueConst this_va return result; } +typedef struct _append_message_t +{ + JSContext* context; + JSValue promise[2]; +} append_message_t; + +static void _tf_ssb_appendMessage_finish(append_message_t* async, bool success, JSValue result) +{ + JSValue error = JS_Call(async->context, success ? async->promise[0] : async->promise[1], JS_UNDEFINED, 1, &result); + tf_util_report_error(async->context, error); + JS_FreeValue(async->context, error); + JS_FreeValue(async->context, async->promise[0]); + JS_FreeValue(async->context, async->promise[1]); + tf_free(async); +} + +static void _tf_ssb_appendMessageWithIdentity_callback(const char* id, bool verified, bool is_new, void* user_data) +{ + append_message_t* async = user_data; + JSValue result = JS_UNDEFINED; + if (verified) + { + result = is_new ? JS_TRUE : JS_FALSE; + } + _tf_ssb_appendMessage_finish(async, verified, result); +} + static JSValue _tf_ssb_appendMessageWithIdentity(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv) { - JSValue result = JS_UNDEFINED; + append_message_t* async = tf_malloc(sizeof(append_message_t)); + *async = (append_message_t) { 0 }; + JSValue result = JS_NewPromiseCapability(context, async->promise); + tf_ssb_t* ssb = JS_GetOpaque(this_val, _tf_ssb_classId); if (ssb) { @@ -118,17 +148,19 @@ static JSValue _tf_ssb_appendMessageWithIdentity(JSContext* context, JSValueCons uint8_t private_key[crypto_sign_SECRETKEYBYTES]; if (tf_ssb_db_identity_get_private_key(ssb, user, id, private_key, sizeof(private_key))) { - char message_id[k_id_base64_len] = { 0 }; - tf_ssb_append_message_with_keys(ssb, id, private_key, argv[2], message_id, sizeof(message_id)); - result = JS_NewString(context, message_id); + tf_ssb_verify_strip_and_store_message(ssb, tf_ssb_sign_message(ssb, id, private_key, argv[2]), _tf_ssb_appendMessageWithIdentity_callback, async); } else { - result = JS_ThrowInternalError(context, "Unable to get private key for user %s with identity %s.", user, id); + _tf_ssb_appendMessage_finish(async, false, JS_ThrowInternalError(context, "Unable to get private key for user %s with identity %s.", user, id)); } JS_FreeCString(context, id); JS_FreeCString(context, user); } + else + { + _tf_ssb_appendMessage_finish(async, false, JS_ThrowInternalError(context, "No SSB instance.")); + } return result; } @@ -637,18 +669,31 @@ static JSValue _tf_ssb_sqlAsync(JSContext* context, JSValueConst this_val, int a return result; } +typedef struct _message_store_t +{ + JSContext* context; + JSValue promise[2]; +} message_store_t; + +void _tf_ssb_message_store_callback(const char* id, bool verified, bool is_new, void* user_data) +{ + message_store_t* store = user_data; + JSValue result = JS_Call(store->context, id ? store->promise[0] : store->promise[1], JS_UNDEFINED, 0, NULL); + tf_util_report_error(store->context, result); + JS_FreeValue(store->context, result); + JS_FreeValue(store->context, store->promise[0]); + JS_FreeValue(store->context, store->promise[1]); + tf_free(store); +} + static JSValue _tf_ssb_storeMessage(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv) { tf_ssb_t* ssb = JS_GetOpaque(this_val, _tf_ssb_classId); - bool is_new = false; - if (tf_ssb_verify_strip_and_store_message(ssb, argv[0], &is_new)) - { - return is_new ? JS_TRUE : JS_FALSE; - } - else - { - return JS_UNDEFINED; - } + message_store_t* store = tf_malloc(sizeof(message_store_t)); + *store = (message_store_t) { .context = context }; + JSValue result = JS_NewPromiseCapability(context, store->promise); + tf_ssb_verify_strip_and_store_message(ssb, argv[0], _tf_ssb_message_store_callback, store); + return result; } typedef struct _broadcasts_t diff --git a/src/ssb.rpc.c b/src/ssb.rpc.c index db27b824..ed5b2401 100644 --- a/src/ssb.rpc.c +++ b/src/ssb.rpc.c @@ -962,7 +962,7 @@ static void _tf_ssb_rpc_ebt_replicate(tf_ssb_connection_t* connection, uint8_t f if (!JS_IsUndefined(author)) { /* Looks like a message. */ - tf_ssb_verify_strip_and_store_message(ssb, args, NULL); + tf_ssb_verify_strip_and_store_message(ssb, args, NULL, NULL); } else { diff --git a/src/ssb.tests.c b/src/ssb.tests.c index df127c53..21304e12 100644 --- a/src/ssb.tests.c +++ b/src/ssb.tests.c @@ -14,6 +14,7 @@ #include #include #include +#include void tf_ssb_test_id_conversion(const tf_test_options_t* options) { @@ -120,6 +121,19 @@ static void _ssb_test_idle(uv_idle_t* idle) } } +static void _message_stored(const char* id, bool verified, bool is_new, void* user_data) +{ + *(bool*)user_data = true; +} + +static void _wait_stored(tf_ssb_t* ssb, bool* stored) +{ + while (!*stored) + { + uv_run(tf_ssb_get_loop(ssb), UV_RUN_ONCE); + } +} + void tf_ssb_test_ssb(const tf_test_options_t* options) { tf_printf("Testing SSB.\n"); @@ -127,9 +141,11 @@ void tf_ssb_test_ssb(const tf_test_options_t* options) uv_loop_t loop = { 0 }; uv_loop_init(&loop); - tf_ssb_t* ssb0 = tf_ssb_create(&loop, NULL, "file:db0?mode=memory&cache=shared"); + unlink("out/test_db0.sqlite"); + tf_ssb_t* ssb0 = tf_ssb_create(&loop, NULL, "file:out/test_db0.sqlite"); tf_ssb_register(tf_ssb_get_context(ssb0), ssb0); - tf_ssb_t* ssb1 = tf_ssb_create(&loop, NULL, "file:db1?mode=memory&cache=shared"); + unlink("out/test_db1.sqlite"); + tf_ssb_t* ssb1 = tf_ssb_create(&loop, NULL, "file:out/test_db1.sqlite"); tf_ssb_register(tf_ssb_get_context(ssb1), ssb1); uv_idle_t idle0 = { .data = ssb0 }; @@ -170,19 +186,25 @@ void tf_ssb_test_ssb(const tf_test_options_t* options) b = tf_ssb_db_blob_store(ssb0, (const uint8_t*)k_blob, strlen(k_blob), blob_id, sizeof(blob_id), NULL); assert(b); - char message_id[k_id_base64_len] = { 0 }; JSContext* context0 = tf_ssb_get_context(ssb0); JSValue obj = JS_NewObject(context0); JS_SetPropertyStr(context0, obj, "type", JS_NewString(context0, "post")); JS_SetPropertyStr(context0, obj, "text", JS_NewString(context0, "Hello, world!")); - tf_ssb_append_message_with_keys(ssb0, id0, priv0, obj, message_id, sizeof(message_id)); + bool stored = false; + JSValue signed_message = tf_ssb_sign_message(ssb0, id0, priv0, obj); + tf_ssb_verify_strip_and_store_message(ssb0, signed_message, _message_stored, &stored); + JS_FreeValue(context0, signed_message); + _wait_stored(ssb0, &stored); JS_FreeValue(context0, obj); - printf("appended %s\n", message_id); obj = JS_NewObject(context0); JS_SetPropertyStr(context0, obj, "type", JS_NewString(context0, "post")); JS_SetPropertyStr(context0, obj, "text", JS_NewString(context0, "First post.")); - tf_ssb_append_message_with_keys(ssb0, id0, priv0, obj, NULL, 0); + stored = false; + signed_message = tf_ssb_sign_message(ssb0, id0, priv0, obj); + tf_ssb_verify_strip_and_store_message(ssb0, signed_message, _message_stored, &stored); + JS_FreeValue(context0, signed_message); + _wait_stored(ssb0, &stored); JS_FreeValue(context0, obj); obj = JS_NewObject(context0); @@ -193,7 +215,11 @@ void tf_ssb_test_ssb(const tf_test_options_t* options) JS_SetPropertyStr(context0, mention, "link", JS_NewString(context0, blob_id)); JS_SetPropertyUint32(context0, mentions, 0, mention); JS_SetPropertyStr(context0, obj, "mentions", mentions); - tf_ssb_append_message_with_keys(ssb0, id0, priv0, obj, NULL, 0); + stored = false; + signed_message = tf_ssb_sign_message(ssb0, id0, priv0, obj); + tf_ssb_verify_strip_and_store_message(ssb0, signed_message, _message_stored, &stored); + JS_FreeValue(context0, signed_message); + _wait_stored(ssb0, &stored); JS_FreeValue(context0, obj); uint8_t* b0; @@ -249,7 +275,11 @@ void tf_ssb_test_ssb(const tf_test_options_t* options) obj = JS_NewObject(context0); JS_SetPropertyStr(context0, obj, "type", JS_NewString(context0, "post")); JS_SetPropertyStr(context0, obj, "text", JS_NewString(context0, "Message to self.")); - tf_ssb_append_message_with_keys(ssb0, id0, priv0, obj, NULL, 0); + stored = false; + signed_message = tf_ssb_sign_message(ssb0, id0, priv0, obj); + tf_ssb_verify_strip_and_store_message(ssb0, signed_message, _message_stored, &stored); + JS_FreeValue(context0, signed_message); + _wait_stored(ssb0, &stored); JS_FreeValue(context0, obj); while (count0 == 0) @@ -313,11 +343,14 @@ void tf_ssb_test_rooms(const tf_test_options_t* options) uv_loop_t loop = { 0 }; uv_loop_init(&loop); - tf_ssb_t* ssb0 = tf_ssb_create(&loop, NULL, "file:db0?mode=memory&cache=shared"); + unlink("out/test_db0.sqlite"); + tf_ssb_t* ssb0 = tf_ssb_create(&loop, NULL, "file:out/test_db0.sqlite"); tf_ssb_register(tf_ssb_get_context(ssb0), ssb0); - tf_ssb_t* ssb1 = tf_ssb_create(&loop, NULL, "file:db1?mode=memory&cache=shared"); + unlink("out/test_db1.sqlite"); + tf_ssb_t* ssb1 = tf_ssb_create(&loop, NULL, "file:out/test_db1.sqlite"); tf_ssb_register(tf_ssb_get_context(ssb1), ssb1); - tf_ssb_t* ssb2 = tf_ssb_create(&loop, NULL, "file:db2?mode=memory&cache=shared"); + unlink("out/test_db2.sqlite"); + tf_ssb_t* ssb2 = tf_ssb_create(&loop, NULL, "file:out/test_db2.sqlite"); tf_ssb_register(tf_ssb_get_context(ssb2), ssb2); uv_idle_t idle0 = { .data = ssb0 }; @@ -492,6 +525,8 @@ void tf_ssb_test_following(const tf_test_options_t* options) JSContext* context = NULL; JSValue message; + JSValue signed_message; + bool stored; #define FOLLOW(ssb, id, priv, follow) \ context = tf_ssb_get_context(ssb); \ @@ -499,7 +534,11 @@ void tf_ssb_test_following(const tf_test_options_t* options) JS_SetPropertyStr(context, message, "type", JS_NewString(context, "contact")); \ JS_SetPropertyStr(context, message, "contact", JS_NewString(context, id)); \ JS_SetPropertyStr(context, message, "following", follow ? JS_TRUE : JS_FALSE); \ - tf_ssb_append_message_with_keys(ssb, id, priv, message, NULL, 0); \ + signed_message = tf_ssb_sign_message(ssb, id, priv, message); \ + stored = false; \ + tf_ssb_verify_strip_and_store_message(ssb, signed_message, _message_stored, &stored); \ + _wait_stored(ssb, &stored); \ + JS_FreeValue(context, signed_message); \ JS_FreeValue(context, message); \ context = NULL @@ -557,7 +596,8 @@ void tf_ssb_test_bench(const tf_test_options_t* options) uv_loop_t loop = { 0 }; uv_loop_init(&loop); - tf_ssb_t* ssb0 = tf_ssb_create(&loop, NULL, "file:db0?mode=memory&cache=shared"); + unlink("out/test_db0.sqlite"); + tf_ssb_t* ssb0 = tf_ssb_create(&loop, NULL, "file:out/test_db0.sqlite"); tf_ssb_generate_keys(ssb0); char id0[k_id_base64_len] = { 0 }; @@ -575,13 +615,18 @@ void tf_ssb_test_bench(const tf_test_options_t* options) JS_SetPropertyStr(tf_ssb_get_context(ssb0), obj, "text", JS_NewString(tf_ssb_get_context(ssb0), "Hello, world!")); for (int i = 0; i < k_messages; i++) { - tf_ssb_append_message_with_keys(ssb0, id0, priv0, obj, NULL, 0); + bool stored = false; + JSValue signed_message = tf_ssb_sign_message(ssb0, id0, priv0, obj); + tf_ssb_verify_strip_and_store_message(ssb0, signed_message, _message_stored, &stored); + JS_FreeValue(tf_ssb_get_context(ssb0), signed_message); + _wait_stored(ssb0, &stored); } JS_FreeValue(tf_ssb_get_context(ssb0), obj); clock_gettime(CLOCK_REALTIME, &end_time); tf_printf("insert = %f seconds\n", (end_time.tv_sec - start_time.tv_sec) + (end_time.tv_nsec - start_time.tv_nsec) / 1e9); - tf_ssb_t* ssb1 = tf_ssb_create(&loop, NULL, "file:db1?mode=memory&cache=shared"); + unlink("out/test_db1.sqlite"); + tf_ssb_t* ssb1 = tf_ssb_create(&loop, NULL, "file:out/test_db1.sqlite"); tf_ssb_generate_keys(ssb1); uint8_t id0bin[k_id_bin_len]; tf_ssb_id_str_to_bin(id0bin, id0); @@ -602,9 +647,12 @@ void tf_ssb_test_bench(const tf_test_options_t* options) tf_printf("Waiting for messages.\n"); clock_gettime(CLOCK_REALTIME, &start_time); - while (_ssb_test_count_messages(ssb1) < k_messages) + int count = 0; + while (count < k_messages) { + //tf_printf("%d / %d\n", count, k_messages); uv_run(&loop, UV_RUN_ONCE); + count = _ssb_test_count_messages(ssb1); } clock_gettime(CLOCK_REALTIME, &end_time); tf_printf("Done.\n");