Make storing messages async. Phew.

git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@4355 ed5197a5-7fde-0310-b194-c3ffbd925b24

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", have_previous ? JS_NewString(context, previous_id) : JS_NULL);
-	{
-		JS_SetPropertyStr(context, root, "previous", JS_NewString(context, previous_id));
-	}
-	else
-	{
-		JS_SetPropertyStr(context, root, "previous", 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");
-	}
+		JS_FreeValue(context, root);
-
+		root = JS_UNDEFINED;
-	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");
-		}
 	}
 	else
 	{
-		tf_printf("Failed to verify message signature.\n");
+		char signature_base64[crypto_sign_BYTES * 2];
-		tf_printf("json = %s\n", json);
+		tf_base64_encode(signature, sizeof(signature), signature_base64, sizeof(signature_base64));
-		tf_printf("sig = %s\n", 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)
+	return root;
-	{
-		*out_id = '\0';
-	}
-
-	JS_FreeCString(context, json);
-	JS_FreeValue(context, jsonval);
-
-	JS_FreeValue(context, root);
-	return stored;
 }
 
 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;
+		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);
-	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;
 	}
 	else
 	{
-		tf_printf("failed to verify message\n");
+		printf("nope\n");
-		return false;
+		_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;
+}

src/ssb.db.c

@@ -17,13 +17,15 @@
 #include <stdlib.h>
 #include <string.h>
 
+#include <unistd.h>
+
 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, &timestamp, 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 (r == SQLITE_DONE && sqlite3_changes(db) != 0)
-				if (stored)
 				{
 					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");
+		tf_printf("%p: Previous message doesn't exist for author=%s sequence=%" PRId64 ".\n", db, author, sequence);
-	}
-
-	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_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, &timestamp, 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));

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);

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);

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_verify_strip_and_store_message(ssb, tf_ssb_sign_message(ssb, id, private_key, argv[2]), _tf_ssb_appendMessageWithIdentity_callback, async);
-			tf_ssb_append_message_with_keys(ssb, id, private_key, argv[2], message_id, sizeof(message_id));
-			result = JS_NewString(context, message_id);
 		}
 		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;
+	message_store_t* store = tf_malloc(sizeof(message_store_t));
-	if (tf_ssb_verify_strip_and_store_message(ssb, argv[0], &is_new))
+	*store = (message_store_t) { .context = context };
-	{
+	JSValue result = JS_NewPromiseCapability(context, store->promise);
-		return is_new ? JS_TRUE : JS_FALSE;
+	tf_ssb_verify_strip_and_store_message(ssb, argv[0], _tf_ssb_message_store_callback, store);
-	}
+	return result;
-	else
-	{
-		return JS_UNDEFINED;
-	}
 }
 
 typedef struct _broadcasts_t

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
 	{

src/ssb.tests.c

@@ -14,6 +14,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <time.h>
+#include <unistd.h>
 
 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");