#include "ssb.db.h" #include "ssb.h" #include "trace.h" #include #include #include #include #include void tf_ssb_db_init(tf_ssb_t* ssb) { sqlite3* db = tf_ssb_get_db(ssb); 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, ×tamp, 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); printf("blob store %s\n", id); const char* query = "INSERT INTO blobs (id, content, created) VALUES ($1, $2, CAST(strftime('%s') AS INTEGER)) 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 && sqlite3_bind_blob(statement, 2, blob, size, NULL) == SQLITE_OK) { result = sqlite3_step(statement) == SQLITE_DONE; } else { printf("bind failed: %s\n", sqlite3_errmsg(db)); } sqlite3_finalize(statement); } else { printf("prepare failed: %s\n", sqlite3_errmsg(db)); } 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\n", sqlite3_errmsg(db)); } sqlite3_set_authorizer(db, NULL, NULL); }