tildefriends/src/ssb.db.c

#include "ssb.db.h"

#include "ssb.h"
#include "trace.h"

#include <base64c.h>
#include <sodium/crypto_hash_sha256.h>
#include <sqlite3.h>
#include <stdlib.h>
#include <string.h>

void tf_ssb_db_init(tf_ssb_t* ssb)
{
	sqlite3* db = tf_ssb_get_db(ssb);
	sqlite3_exec(db, "PRAGMA journal_mode = WAL", NULL, NULL, NULL);
	sqlite3_exec(db, "PRAGMA synchronous = NORMAL", NULL, NULL, NULL);
	sqlite3_exec(db,
		"CREATE TABLE IF NOT EXISTS messages ("
		"  author TEXT,"
		"  id TEXT PRIMARY KEY,"
		"  sequence INTEGER,"
		"  timestamp INTEGER,"
		"  previous TEXT,"
		"  hash TEXT,"
		"  content TEXT,"
		"  signature TEXT,"
		"  UNIQUE(author, sequence)"
		")",
		NULL, NULL, NULL);
	sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS messages_author_id_index ON messages (author, id)", NULL, NULL, NULL);
	sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS messages_author_sequence_index ON messages (author, sequence)", NULL, NULL, NULL);
	sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS messages_author_timestamp_index ON messages (author, timestamp)", NULL, NULL, NULL);
	sqlite3_exec(db,
		"CREATE TABLE IF NOT EXISTS blobs ("
		"  id TEXT PRIMARY KEY,"
		"  content BLOB,"
		"  created INTEGER"
		")",
		NULL, NULL, NULL);
	sqlite3_exec(db,
		"CREATE TABLE IF NOT EXISTS blob_wants ("
		"  id TEXT PRIMARY KEY"
		")",
		NULL, NULL, NULL);
	sqlite3_exec(db,
		"CREATE TABLE IF NOT EXISTS properties ("
		"  id TEXT,"
		"  key TEXT,"
		"  value TEXT,"
		"  UNIQUE(id, key)"
		")",
		NULL, NULL, NULL);
	sqlite3_exec(db,
		"CREATE TABLE IF NOT EXISTS connections ("
		"  host TEXT,"
		"  port INTEGER,"
		"  key TEXT,"
		"  last_attempt INTEGER,"
		"  last_success INTEGER,"
		"  UNIQUE(host, port, key)"
		")",
		NULL, NULL, NULL);
}

bool tf_ssb_db_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, JSValue val, const char* signature)
{
	bool stored = false;
	JSValue previousval = JS_GetPropertyStr(context, val, "previous");
	const char* previous = JS_IsNull(previousval) ? NULL : JS_ToCString(context, previousval);
	JSValue authorval = JS_GetPropertyStr(context, val, "author");
	const char* author = JS_ToCString(context, authorval);
	int64_t sequence = -1;
	JS_ToInt64(context, &sequence, JS_GetPropertyStr(context, val, "sequence"));
	int64_t timestamp = -1;
	JS_ToInt64(context, &timestamp, JS_GetPropertyStr(context, val, "timestamp"));

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

	sqlite3* db = tf_ssb_get_db(ssb);
	sqlite3_stmt* statement;
	int64_t last_row_id = -1;
	const char* query = "INSERT INTO messages (id, previous, author, sequence, timestamp, content, hash, signature) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING";
	if (sqlite3_prepare(db, query, -1, &statement, NULL) == SQLITE_OK)
	{
		if (sqlite3_bind_text(statement, 1, id, -1, NULL) == SQLITE_OK &&
			(previous ? sqlite3_bind_text(statement, 2, previous, -1, NULL) : sqlite3_bind_null(statement, 2)) == SQLITE_OK &&
			sqlite3_bind_text(statement, 3, author, -1, NULL) == SQLITE_OK &&
			sqlite3_bind_int64(statement, 4, sequence) == SQLITE_OK &&
			sqlite3_bind_int64(statement, 5, timestamp) == SQLITE_OK &&
			sqlite3_bind_text(statement, 6, contentstr, 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)
		{
			int r = sqlite3_step(statement);
			if (r != SQLITE_DONE)
			{
				printf("%s\n", sqlite3_errmsg(db));
			}
			stored = r == SQLITE_DONE && sqlite3_changes(db) != 0;
			if (stored)
			{
				last_row_id = sqlite3_last_insert_rowid(db);
			}
		}
		sqlite3_finalize(statement);
	}
	else
	{
		printf("prepare failed: %s\n", sqlite3_errmsg(db));
	}

	if (last_row_id != -1)
	{
		query = "INSERT INTO blob_wants (id) 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 ON CONFLICT DO NOTHING RETURNING id";
		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)
				{
					printf("%s\n", sqlite3_errmsg(db));
				}
			}
			sqlite3_finalize(statement);
		}
		else
		{
			printf("prepare failed: %s\n", sqlite3_errmsg(db));
		}
	}

	JS_FreeValue(context, previousval);
	JS_FreeCString(context, author);
	JS_FreeValue(context, authorval);
	JS_FreeCString(context, previous);
	JS_FreeCString(context, contentstr);
	JS_FreeValue(context, content);
	return stored;
}

bool tf_ssb_db_message_content_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size)
{
	bool result = false;
	sqlite3_stmt* statement;
	const char* query = "SELECT content FROM messages WHERE id = ?";
	if (sqlite3_prepare(tf_ssb_get_db(ssb), query, -1, &statement, NULL) == SQLITE_OK)
	{
		if (sqlite3_bind_text(statement, 1, id, -1, NULL) == SQLITE_OK &&
			sqlite3_step(statement) == SQLITE_ROW)
		{
			const uint8_t* blob = sqlite3_column_blob(statement, 0);
			int size = sqlite3_column_bytes(statement, 0);
			if (out_blob)
			{
				*out_blob = malloc(size + 1);
				memcpy(*out_blob, blob, size);
				(*out_blob)[size] = '\0';
			}
			if (out_size)
			{
				*out_size = size;
			}
			result = true;
		}
		sqlite3_finalize(statement);
	}
	return result;
}

bool tf_ssb_db_blob_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size)
{
	bool result = false;
	sqlite3_stmt* statement;
	const char* query = "SELECT content FROM blobs WHERE id = $1";
	if (sqlite3_prepare(tf_ssb_get_db(ssb), query, -1, &statement, NULL) == SQLITE_OK)
	{
		if (sqlite3_bind_text(statement, 1, id, -1, NULL) == SQLITE_OK &&
			sqlite3_step(statement) == SQLITE_ROW)
		{
			const uint8_t* blob = sqlite3_column_blob(statement, 0);
			int size = sqlite3_column_bytes(statement, 0);
			if (out_blob)
			{
				*out_blob = malloc(size + 1);
				memcpy(*out_blob, blob, size);
				(*out_blob)[size] = '\0';
			}
			if (out_size)
			{
				*out_size = size;
			}
			result = true;
		}
		sqlite3_finalize(statement);
	}
	return result;
}

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 result = false;
	sqlite3* db = tf_ssb_get_db(ssb);
	sqlite3_stmt* statement;

	uint8_t hash[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash, blob, size);

	char hash64[256];
	base64c_encode(hash, sizeof(hash), (uint8_t*)hash64, sizeof(hash64));

	char id[512];
	snprintf(id, sizeof(id), "&%s.sha256", hash64);

	int rows = 0;

	if (sqlite3_prepare(db, "INSERT INTO blobs (id, content, created) VALUES ($1, $2, CAST(strftime('%s') AS INTEGER)) ON CONFLICT DO NOTHING", -1, &statement, NULL) == SQLITE_OK)
	{
		if (sqlite3_bind_text(statement, 1, id, -1, NULL) == SQLITE_OK &&
			sqlite3_bind_blob(statement, 2, blob, size, NULL) == SQLITE_OK)
		{
			result = sqlite3_step(statement) == SQLITE_DONE;
			rows = sqlite3_changes(db);
		}
		else
		{
			printf("bind failed: %s\n", sqlite3_errmsg(db));
		}
		sqlite3_finalize(statement);
	}
	else
	{
		printf("prepare failed: %s\n", sqlite3_errmsg(db));
	}

	if (rows)
	{
		printf("blob stored %s %zd => %d\n", id, size, result);
	}

	if (result)
	{
		if (sqlite3_prepare(db, "DELETE FROM blob_wants WHERE id = ?1", -1, &statement, NULL) == SQLITE_OK)
		{
			if (sqlite3_bind_text(statement, 1, id, -1, NULL) == SQLITE_OK)
			{
				sqlite3_step(statement);
			}
			sqlite3_finalize(statement);
		}
	}

	if (result && out_id)
	{
		snprintf(out_id, out_id_size, "%s", id);
	}
	return result;
}

bool tf_ssb_db_get_message_by_author_and_sequence(tf_ssb_t* ssb, const char* author, int64_t sequence, char* out_message_id, size_t out_message_id_size, int64_t* out_timestamp, char** out_content)
{
	bool found = false;
	sqlite3_stmt* statement;
	const char* query = "SELECT id, timestamp, content FROM messages WHERE author = $1 AND sequence = $2";
	if (sqlite3_prepare(tf_ssb_get_db(ssb), query, -1, &statement, NULL) == SQLITE_OK)
	{
		if (sqlite3_bind_text(statement, 1, author, -1, NULL) == SQLITE_OK &&
			sqlite3_bind_int64(statement, 2, sequence) == SQLITE_OK &&
			sqlite3_step(statement) == SQLITE_ROW)
		{
			if (out_message_id)
			{
				strncpy(out_message_id, (const char*)sqlite3_column_text(statement, 0), out_message_id_size - 1);
			}
			if (out_timestamp)
			{
				*out_timestamp = sqlite3_column_int64(statement, 1);
			}
			if (out_content)
			{
				*out_content = strdup((const char*)sqlite3_column_text(statement, 2));
			}
			found = true;
		}
		sqlite3_finalize(statement);
	}
	else
	{
		printf("prepare failed: %s\n", sqlite3_errmsg(tf_ssb_get_db(ssb)));
	}
	return found;
}

bool tf_ssb_db_get_latest_message_by_author(tf_ssb_t* ssb, const char* author, int64_t* out_sequence, char* out_message_id, size_t out_message_id_size)
{
	bool found = false;
	sqlite3_stmt* statement;
	const char* query = "SELECT id, sequence FROM messages WHERE author = $1 AND sequence = (SELECT MAX(sequence) FROM messages WHERE author = $1)";
	if (sqlite3_prepare(tf_ssb_get_db(ssb), query, -1, &statement, NULL) == SQLITE_OK)
	{
		if (sqlite3_bind_text(statement, 1, author, -1, NULL) == SQLITE_OK &&
			sqlite3_step(statement) == SQLITE_ROW)
		{
			if (out_sequence)
			{
				*out_sequence = sqlite3_column_int64(statement, 1);
			}
			if (out_message_id)
			{
				strncpy(out_message_id, (const char*)sqlite3_column_text(statement, 0), out_message_id_size - 1);
			}
			found = true;
		}
		sqlite3_finalize(statement);
	}
	else
	{
		printf("prepare failed: %s\n", sqlite3_errmsg(tf_ssb_get_db(ssb)));
	}
	return found;
}

static bool _tf_ssb_sqlite_bind_json(JSContext* context, sqlite3* db, sqlite3_stmt* statement, JSValue binds)
{
	bool all_bound = true;
	int32_t length = 0;
	if (JS_IsUndefined(binds))
	{
		return true;
	}
	JSValue lengthval = JS_GetPropertyStr(context, binds, "length");
	if (JS_ToInt32(context, &length, lengthval) == 0)
	{
		for (int i = 0; i < length; i++)
		{
			JSValue value = JS_GetPropertyUint32(context, binds, i);
			if (JS_IsString(value))
			{
				size_t str_len = 0;
				const char* str = JS_ToCStringLen(context, &str_len, value);
				if (str)
				{
					if (sqlite3_bind_text(statement, i + 1, str, str_len, SQLITE_TRANSIENT) != SQLITE_OK)
					{
						printf("failed to bind: %s\n", sqlite3_errmsg(db));
						all_bound = false;
					}
					JS_FreeCString(context, str);
				}
				else
				{
					printf("expected cstring\n");
				}
			}
			else if (JS_IsNumber(value))
			{
				int64_t number = 0;
				JS_ToInt64(context, &number, value);
				if (sqlite3_bind_int64(statement, i + 1, number) != SQLITE_OK)
				{
					printf("failed to bind: %s\n", sqlite3_errmsg(db));
					all_bound = false;
				}
			}
			else if (JS_IsNull(value))
			{
				if (sqlite3_bind_null(statement, i + 1) != SQLITE_OK)
				{
					printf("failed to bind: %s\n", sqlite3_errmsg(db));
					all_bound = false;
				}
			}
			else
			{
				const char* str = JS_ToCString(context, value);
				printf("expected string: %s\n", str);
				JS_FreeCString(context, str);
			}
			JS_FreeValue(context, value);
		}
	}
	else
	{
		printf("expected array\n");
	}
	JS_FreeValue(context, lengthval);
	return all_bound;
}

static JSValue _tf_ssb_sqlite_row_to_json(JSContext* context, sqlite3_stmt* row)
{
	JSValue result = JS_NewObject(context);
	for (int i = 0; i < sqlite3_column_count(row); i++)
	{
		const char* name = sqlite3_column_name(row, i);
		switch (sqlite3_column_type(row, i))
		{
		case SQLITE_INTEGER:
			JS_SetPropertyStr(context, result, name, JS_NewInt64(context, sqlite3_column_int64(row, i)));
			break;
		case SQLITE_FLOAT:
			JS_SetPropertyStr(context, result, name, JS_NewFloat64(context, sqlite3_column_double(row, i)));
			break;
		case SQLITE_TEXT:
			JS_SetPropertyStr(context, result, name, JS_NewStringLen(context, (const char*)sqlite3_column_text(row, i), sqlite3_column_bytes(row, i)));
			break;
		case SQLITE_BLOB:
			JS_SetPropertyStr(context, result, name, JS_NewArrayBufferCopy(context, sqlite3_column_blob(row, i), sqlite3_column_bytes(row, i)));
			break;
		case SQLITE_NULL:
			JS_SetPropertyStr(context, result, name, JS_NULL);
			break;
		}
	}
	return result;
}

static int _tf_ssb_sqlite_authorizer(void* user_data, int action_code, const char* arg0, const char* arg1, const char* arg2, const char* arg3)
{
	switch (action_code)
	{
	case SQLITE_SELECT:
	case SQLITE_FUNCTION:
		return SQLITE_OK;
	case SQLITE_READ:
		return
			(strcmp(arg0, "messages") == 0 ||
			strcmp(arg0, "blob_wants") == 0)
			? SQLITE_OK : SQLITE_DENY;
		break;
	}
	return SQLITE_DENY;
}

void tf_ssb_db_visit_query(tf_ssb_t* ssb, const char* query, const JSValue binds, void (*callback)(JSValue row, void* user_data), void* user_data)
{
	sqlite3* db = tf_ssb_get_db(ssb);
	sqlite3_stmt* statement;
	sqlite3_set_authorizer(db, _tf_ssb_sqlite_authorizer, ssb);
	if (sqlite3_prepare(db, query, -1, &statement, NULL) == SQLITE_OK)
	{
		JSContext* context = tf_ssb_get_context(ssb);
		if (_tf_ssb_sqlite_bind_json(context, db, statement, binds))
		{
			while (sqlite3_step(statement) == SQLITE_ROW)
			{
				JSValue row = _tf_ssb_sqlite_row_to_json(context, statement);
				tf_trace_t* trace = tf_ssb_get_trace(ssb);
				tf_trace_begin(trace, "callback");
				callback(row, user_data);
				tf_trace_end(trace);
				JS_FreeValue(context, row);
			}
		}
		sqlite3_finalize(statement);
	}
	else
	{
		printf("prepare failed: %s: %s\n", sqlite3_errmsg(db), query);
	}
	sqlite3_set_authorizer(db, NULL, NULL);
}