Trying to get organized. Move things db, import, and export out of ssb.c.

git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@3655 ed5197a5-7fde-0310-b194-c3ffbd925b24
This commit is contained in:
Cory McWilliams 2021-08-22 19:34:28 +00:00
parent 45dfe34375
commit e922af4c55
9 changed files with 607 additions and 542 deletions

View File

@ -1,4 +1,6 @@
#include "ssb.h"
#include "ssb.import.h"
#include "ssb.export.h"
#include "task.h"
#include "taskstub.h"
#include "tests.h"

541
src/ssb.c
View File

@ -1,6 +1,7 @@
#include "ssb.h"
#include "ssb.connections.h"
#include "ssb.db.h"
#include "ssb.rpc.h"
#include "trace.h"
@ -385,162 +386,6 @@ void tf_ssb_connection_rpc_send(tf_ssb_connection_t* connection, uint8_t flags,
printf("RPC SEND flags=%x RN=%d: %.*s\n", flags, request_number, (int)size, message);
}
bool tf_ssb_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(ssb->db, 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(ssb->db));
}
return found;
}
bool tf_ssb_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(ssb->db, 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(ssb->db));
}
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);
}
}
} 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 ? SQLITE_OK : SQLITE_DENY;
break;
}
return SQLITE_DENY;
}
void tf_ssb_visit_query(tf_ssb_t* ssb, const char* query, const JSValue binds, void (*callback)(JSValue row, void* user_data), void* user_data)
{
sqlite3_stmt* statement;
sqlite3_set_authorizer(ssb->db, _tf_ssb_sqlite_authorizer, ssb);
if (sqlite3_prepare(ssb->db, query, -1, &statement, NULL) == SQLITE_OK) {
if (_tf_ssb_sqlite_bind_json(ssb->context, ssb->db, statement, binds)) {
while (sqlite3_step(statement) == SQLITE_ROW) {
JSValue row = _tf_ssb_sqlite_row_to_json(ssb->context, statement);
tf_trace_begin(ssb->trace, "callback");
callback(row, user_data);
tf_trace_end(ssb->trace);
JS_FreeValue(ssb->context, row);
}
}
sqlite3_finalize(statement);
} else {
printf("prepare failed: %s\n", sqlite3_errmsg(ssb->db));
}
sqlite3_set_authorizer(ssb->db, NULL, NULL);
}
void tf_ssb_calculate_message_id(JSContext* context, JSValue message, char* out_id, size_t out_id_size)
{
JSValue idval = JS_JSONStringify(context, message, JS_NULL, JS_NewInt32(context, 2));
@ -608,55 +453,6 @@ bool tf_ssb_verify_and_strip_signature(JSContext* context, JSValue val, char* ou
return verified;
}
bool tf_ssb_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_stmt* statement;
const char* query = "INSERT INTO messages (id, previous, author, sequence, timestamp, content, hash, signature) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING";
if (sqlite3_prepare(ssb->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(ssb->db));
}
stored = r == SQLITE_DONE && sqlite3_changes(ssb->db) != 0;
}
sqlite3_finalize(statement);
} else {
printf("prepare failed: %s\n", sqlite3_errmsg(ssb->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;
}
void tf_ssb_send_createHistoryStream(tf_ssb_t* ssb, const char* id)
{
for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next) {
@ -1496,47 +1292,7 @@ tf_ssb_t* tf_ssb_create(uv_loop_t* loop, JSContext* context, sqlite3* db, const
sqlite3_open("db.sqlite", &ssb->db);
ssb->owns_db = true;
}
sqlite3_exec(ssb->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(ssb->db, "CREATE INDEX IF NOT EXISTS messages_author_id_index ON messages (author, id)", NULL, NULL, NULL);
sqlite3_exec(ssb->db, "CREATE INDEX IF NOT EXISTS messages_author_sequence_index ON messages (author, sequence)", NULL, NULL, NULL);
sqlite3_exec(ssb->db, "CREATE INDEX IF NOT EXISTS messages_author_timestamp_index ON messages (author, timestamp)", NULL, NULL, NULL);
sqlite3_exec(ssb->db,
"CREATE TABLE IF NOT EXISTS blobs ("
" id TEXT PRIMARY KEY,"
" content BLOB,"
" created INTEGER"
")",
NULL, NULL, NULL);
sqlite3_exec(ssb->db,
"CREATE TABLE IF NOT EXISTS properties ("
" id TEXT,"
" key TEXT,"
" value TEXT,"
" UNIQUE(id, key)"
")",
NULL, NULL, NULL);
sqlite3_exec(ssb->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);
tf_ssb_db_init(ssb);
if (loop) {
ssb->loop = loop;
@ -1874,90 +1630,6 @@ bool tf_ssb_whoami(tf_ssb_t* ssb, char* out_id, size_t out_id_size)
return tf_ssb_id_bin_to_str(out_id, out_id_size, ssb->pub);
}
bool tf_ssb_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(ssb->db, 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_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_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(ssb->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(ssb->db));
}
sqlite3_finalize(statement);
} else {
printf("prepare failed: %s\n", sqlite3_errmsg(ssb->db));
}
if (result && out_id) {
snprintf(out_id, out_id_size, "%s", id);
}
return result;
}
bool tf_ssb_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(ssb->db, 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;
}
static bool _tf_ssb_parse_broadcast(const char* in_broadcast, tf_ssb_broadcast_t* out_broadcast)
{
char public_key_str[45] = { 0 };
@ -2147,215 +1819,6 @@ void tf_ssb_add_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_connections_c
ssb->connections_changed_count++;
}
static void _write_file(const char* path, void* blob, size_t size)
{
FILE* file = fopen(path, "wb");
if (file) {
fwrite(blob, 1, size, file);
fclose(file);
} else {
printf("Failed to open %s for write: %s.\n", path, strerror(errno));
}
}
void tf_ssb_export(tf_ssb_t* ssb, const char* key)
{
char user[256] = { 0 };
char path[256] = { 0 };
if (sscanf(key, "/~%255[^/]/%255s", user, path) != 2) {
printf("Unable to export %s.\n", key);
return;
}
char app_blob_id[64] = { 0 };
sqlite3_stmt* statement;
if (sqlite3_prepare(ssb->db, "SELECT value FROM properties WHERE id = $1 AND key = 'path:' || $2", -1, &statement, NULL) == SQLITE_OK) {
if (sqlite3_bind_text(statement, 1, user, -1, NULL) == SQLITE_OK &&
sqlite3_bind_text(statement, 2, path, -1, NULL) == SQLITE_OK &&
sqlite3_step(statement) == SQLITE_ROW) {
int len = sqlite3_column_bytes(statement, 0);
if (len >= (int)sizeof(app_blob_id)) {
len = sizeof(app_blob_id) - 1;
}
memcpy(app_blob_id, sqlite3_column_text(statement, 0), len);
app_blob_id[len] = '\0';
}
sqlite3_finalize(statement);
}
if (!*app_blob_id) {
printf("Did not find app blob ID for %s.\n", key);
return;
}
uint8_t* blob = NULL;
size_t size = 0;
if (!tf_ssb_blob_get(ssb, app_blob_id, &blob, &size)) {
printf("Did not find blob for %s: %s.\n", key, app_blob_id);
return;
}
char file_path[1024];
snprintf(file_path, sizeof(file_path), "apps/%s/%s.json", user, path);
_write_file(file_path, blob, size);
JSContext* context = ssb->context;
JSValue app = JS_ParseJSON(context, (const char*)blob, size, NULL);
free(blob);
JSValue files = JS_GetPropertyStr(context, app, "files");
JSPropertyEnum* ptab = NULL;
uint32_t plen = 0;
if (JS_GetOwnPropertyNames(context, &ptab, &plen, files, JS_GPN_STRING_MASK) == 0) {
for (uint32_t i = 0; i < plen; ++i) {
JSPropertyDescriptor desc;
if (JS_GetOwnProperty(context, &desc, files, ptab[i].atom) == 1) {
JSValue key = JS_AtomToString(context, ptab[i].atom);
const char* file_name = JS_ToCString(context, key);
const char* blob_id = JS_ToCString(context, desc.value);
uint8_t* file_blob = NULL;
size_t file_size = 0;
if (tf_ssb_blob_get(ssb, blob_id, &file_blob, &file_size)) {
snprintf(file_path, sizeof(file_path), "apps/%s/%s/%s", user, path, file_name);
_write_file(file_path, file_blob, file_size);
free(file_blob);
}
JS_FreeCString(context, file_name);
JS_FreeValue(context, key);
JS_FreeCString(context, blob_id);
JS_FreeValue(context, desc.value);
}
}
}
for (uint32_t i = 0; i < plen; ++i) {
JS_FreeAtom(context, ptab[i].atom);
}
js_free(context, ptab);
JS_FreeValue(context, files);
JS_FreeValue(context, app);
}
typedef struct _tf_import_file_t {
uv_fs_t req;
uv_file file;
tf_ssb_t* ssb;
const char* user;
const char* parent;
const char* name;
//uv_buf_t buf;
char data[k_ssb_blob_bytes_max];
int* work_left;
} tf_import_file_t;
static void _tf_ssb_import_file_close(uv_fs_t* req)
{
tf_import_file_t* file = req->data;
(*file->work_left)--;
uv_fs_req_cleanup(req);
free(req->data);
}
static void _tf_ssb_import_file_read(uv_fs_t* req)
{
tf_import_file_t* file = req->data;
char id[k_id_base64_len];
if (req->result >= 0) {
if (tf_ssb_blob_store(file->ssb, (const uint8_t*)file->data, req->result, id, sizeof(id))) {
printf("Stored %s/%s as %s.\n", file->parent, file->name, id);
if (strcasecmp(file->name + strlen(file->name) - strlen(".json"), ".json") == 0) {
sqlite3_stmt* statement;
if (sqlite3_prepare(file->ssb->db, "INSERT OR REPLACE INTO properties (id, key, value) VALUES ($1, 'path:' || $2, $3)", -1, &statement, NULL) == SQLITE_OK) {
((char*)file->name)[strlen(file->name) - strlen(".json")] = '\0';
if (sqlite3_bind_text(statement, 1, file->user, -1, NULL) == SQLITE_OK &&
sqlite3_bind_text(statement, 2, file->name, -1, NULL) == SQLITE_OK &&
sqlite3_bind_text(statement, 3, id, -1, NULL) == SQLITE_OK &&
sqlite3_step(statement) == SQLITE_DONE) {
printf("Registered %s path:%s as %s.\n", file->user, file->name, id);
}
sqlite3_finalize(statement);
}
}
}
}
uv_fs_req_cleanup(req);
uv_fs_close(file->ssb->loop, req, file->file, _tf_ssb_import_file_close);
}
static void _tf_ssb_import_file_open(uv_fs_t* req)
{
tf_import_file_t* file = req->data;
file->file = req->result;
uv_fs_req_cleanup(req);
uv_fs_read(file->ssb->loop, req, file->file, &(uv_buf_t) { .base = file->data, .len = k_ssb_blob_bytes_max }, 1, 0, _tf_ssb_import_file_read);
}
typedef struct _tf_import_t {
tf_ssb_t* ssb;
const char* user;
const char* parent;
uv_fs_t req;
int work_left;
} tf_import_t;
static void _tf_ssb_import_scandir(uv_fs_t* req)
{
tf_import_t* import = req->data;
uv_dirent_t ent;
while (uv_fs_scandir_next(req, &ent) == 0) {
size_t len = strlen(import->parent) + strlen(ent.name) + 2;
char* path = malloc(len);
snprintf(path, len, "%s/%s", import->parent, ent.name);
if (ent.type == UV_DIRENT_DIR) {
tf_ssb_import(import->ssb, import->user, path);
} else {
size_t size = sizeof(tf_import_file_t) + strlen(import->parent) +1 + strlen(ent.name) + 1;
tf_import_file_t* file = malloc(size);
memset(file, 0, size);
file->ssb = import->ssb;
file->user = import->user;
file->parent = (void*)(file + 1);
file->name = file->parent + strlen(import->parent) + 1;
file->req.data = file;
file->work_left = &import->work_left;
memcpy((char*)file->parent, import->parent, strlen(import->parent) + 1);
memcpy((char*)file->name, ent.name, strlen(ent.name) + 1);
import->work_left++;
int r = uv_fs_open(import->ssb->loop, &file->req, path, 0, 0, _tf_ssb_import_file_open);
if (r < 0) {
printf("Failed to open %s: %s.\n", path, uv_strerror(r));
free(file);
import->work_left--;
}
}
free(path);
}
import->work_left--;
}
void tf_ssb_import(tf_ssb_t* ssb, const char* user, const char* path)
{
tf_import_t import = {
.ssb = ssb,
.user = user,
.parent = path,
.work_left = 1,
};
import.req.data = &import;
int r = uv_fs_scandir(ssb->loop, &import.req, path, 0, _tf_ssb_import_scandir);
if (r) {
printf("Failed to scan directory %s: %s.", path, uv_strerror(r));
}
while (import.work_left > 0) {
uv_run(ssb->loop, UV_RUN_ONCE);
}
uv_fs_req_cleanup(&import.req);
}
void tf_ssb_register_rpc(tf_ssb_t* ssb, const char** name, tf_ssb_rpc_callback_t* callback, void* user_data)
{
size_t name_len = 0;

350
src/ssb.db.c Normal file
View File

@ -0,0 +1,350 @@
#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,
"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 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_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;
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;
}
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_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_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_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_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_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);
}
}
} 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 ? SQLITE_OK : SQLITE_DENY;
break;
}
return SQLITE_DENY;
}
void tf_ssb_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);
}

16
src/ssb.db.h Normal file
View File

@ -0,0 +1,16 @@
#pragma once
#include <quickjs.h>
#include <stdbool.h>
typedef struct _tf_ssb_t tf_ssb_t;
void tf_ssb_db_init(tf_ssb_t* ssb);
bool tf_ssb_store_message(tf_ssb_t* ssb, JSContext* context, const char* id, JSValue val, const char* signature);
bool tf_ssb_message_content_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size);
bool tf_ssb_blob_get(tf_ssb_t* ssb, const char* id, uint8_t** out_blob, size_t* out_size);
bool tf_ssb_blob_store(tf_ssb_t* ssb, const uint8_t* blob, size_t size, char* out_id, size_t out_id_size);
bool tf_ssb_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 tf_ssb_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);
void tf_ssb_visit_query(tf_ssb_t* ssb, const char* query, const JSValue binds, void (*callback)(JSValue row, void* user_data), void* user_data);

100
src/ssb.export.c Normal file
View File

@ -0,0 +1,100 @@
#include "ssb.export.h"
#include "ssb.h"
#include <stdlib.h>
#include <string.h>
#include <sqlite3.h>
#include <uv.h>
static void _write_file(const char* path, void* blob, size_t size)
{
FILE* file = fopen(path, "wb");
if (file) {
fwrite(blob, 1, size, file);
fclose(file);
} else {
printf("Failed to open %s for write: %s.\n", path, strerror(errno));
}
}
void tf_ssb_export(tf_ssb_t* ssb, const char* key)
{
char user[256] = { 0 };
char path[256] = { 0 };
if (sscanf(key, "/~%255[^/]/%255s", user, path) != 2) {
printf("Unable to export %s.\n", key);
return;
}
char app_blob_id[64] = { 0 };
sqlite3_stmt* statement;
if (sqlite3_prepare(tf_ssb_get_db(ssb), "SELECT value FROM properties WHERE id = $1 AND key = 'path:' || $2", -1, &statement, NULL) == SQLITE_OK) {
if (sqlite3_bind_text(statement, 1, user, -1, NULL) == SQLITE_OK &&
sqlite3_bind_text(statement, 2, path, -1, NULL) == SQLITE_OK &&
sqlite3_step(statement) == SQLITE_ROW) {
int len = sqlite3_column_bytes(statement, 0);
if (len >= (int)sizeof(app_blob_id)) {
len = sizeof(app_blob_id) - 1;
}
memcpy(app_blob_id, sqlite3_column_text(statement, 0), len);
app_blob_id[len] = '\0';
}
sqlite3_finalize(statement);
}
if (!*app_blob_id) {
printf("Did not find app blob ID for %s.\n", key);
return;
}
uint8_t* blob = NULL;
size_t size = 0;
if (!tf_ssb_blob_get(ssb, app_blob_id, &blob, &size)) {
printf("Did not find blob for %s: %s.\n", key, app_blob_id);
return;
}
char file_path[1024];
snprintf(file_path, sizeof(file_path), "apps/%s/%s.json", user, path);
_write_file(file_path, blob, size);
JSContext* context = tf_ssb_get_context(ssb);
JSValue app = JS_ParseJSON(context, (const char*)blob, size, NULL);
free(blob);
JSValue files = JS_GetPropertyStr(context, app, "files");
JSPropertyEnum* ptab = NULL;
uint32_t plen = 0;
if (JS_GetOwnPropertyNames(context, &ptab, &plen, files, JS_GPN_STRING_MASK) == 0) {
for (uint32_t i = 0; i < plen; ++i) {
JSPropertyDescriptor desc;
if (JS_GetOwnProperty(context, &desc, files, ptab[i].atom) == 1) {
JSValue key = JS_AtomToString(context, ptab[i].atom);
const char* file_name = JS_ToCString(context, key);
const char* blob_id = JS_ToCString(context, desc.value);
uint8_t* file_blob = NULL;
size_t file_size = 0;
if (tf_ssb_blob_get(ssb, blob_id, &file_blob, &file_size)) {
snprintf(file_path, sizeof(file_path), "apps/%s/%s/%s", user, path, file_name);
_write_file(file_path, file_blob, file_size);
free(file_blob);
}
JS_FreeCString(context, file_name);
JS_FreeValue(context, key);
JS_FreeCString(context, blob_id);
JS_FreeValue(context, desc.value);
}
}
}
for (uint32_t i = 0; i < plen; ++i) {
JS_FreeAtom(context, ptab[i].atom);
}
js_free(context, ptab);
JS_FreeValue(context, files);
JS_FreeValue(context, app);
}

5
src/ssb.export.h Normal file
View File

@ -0,0 +1,5 @@
#pragma once
typedef struct _tf_ssb_t tf_ssb_t;
void tf_ssb_export(tf_ssb_t* ssb, const char* key);

View File

@ -92,9 +92,6 @@ bool tf_ssb_id_bin_to_str(char* str, size_t str_size, const uint8_t* bin);
void tf_ssb_test();
void tf_ssb_export(tf_ssb_t* ssb, const char* key);
void tf_ssb_import(tf_ssb_t* ssb, const char* user, const char* path);
typedef void (tf_ssb_rpc_callback_t)(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data);
void tf_ssb_register_rpc(tf_ssb_t* ssb, const char** name, tf_ssb_rpc_callback_t* callback, void* user_data);

127
src/ssb.import.c Normal file
View File

@ -0,0 +1,127 @@
#include "ssb.import.h"
#include "ssb.h"
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <sqlite3.h>
#include <uv.h>
typedef struct _tf_import_file_t {
uv_fs_t req;
uv_file file;
tf_ssb_t* ssb;
const char* user;
const char* parent;
const char* name;
//uv_buf_t buf;
char data[k_ssb_blob_bytes_max];
int* work_left;
} tf_import_file_t;
static void _tf_ssb_import_file_close(uv_fs_t* req)
{
tf_import_file_t* file = req->data;
(*file->work_left)--;
uv_fs_req_cleanup(req);
free(req->data);
}
static void _tf_ssb_import_file_read(uv_fs_t* req)
{
tf_import_file_t* file = req->data;
char id[k_id_base64_len];
if (req->result >= 0) {
if (tf_ssb_blob_store(file->ssb, (const uint8_t*)file->data, req->result, id, sizeof(id))) {
printf("Stored %s/%s as %s.\n", file->parent, file->name, id);
if (strcasecmp(file->name + strlen(file->name) - strlen(".json"), ".json") == 0) {
sqlite3_stmt* statement;
if (sqlite3_prepare(tf_ssb_get_db(file->ssb), "INSERT OR REPLACE INTO properties (id, key, value) VALUES ($1, 'path:' || $2, $3)", -1, &statement, NULL) == SQLITE_OK) {
((char*)file->name)[strlen(file->name) - strlen(".json")] = '\0';
if (sqlite3_bind_text(statement, 1, file->user, -1, NULL) == SQLITE_OK &&
sqlite3_bind_text(statement, 2, file->name, -1, NULL) == SQLITE_OK &&
sqlite3_bind_text(statement, 3, id, -1, NULL) == SQLITE_OK &&
sqlite3_step(statement) == SQLITE_DONE) {
printf("Registered %s path:%s as %s.\n", file->user, file->name, id);
}
sqlite3_finalize(statement);
}
}
}
}
uv_fs_req_cleanup(req);
uv_fs_close(tf_ssb_get_loop(file->ssb), req, file->file, _tf_ssb_import_file_close);
}
static void _tf_ssb_import_file_open(uv_fs_t* req)
{
tf_import_file_t* file = req->data;
file->file = req->result;
uv_fs_req_cleanup(req);
uv_fs_read(tf_ssb_get_loop(file->ssb), req, file->file, &(uv_buf_t) { .base = file->data, .len = k_ssb_blob_bytes_max }, 1, 0, _tf_ssb_import_file_read);
}
typedef struct _tf_import_t {
tf_ssb_t* ssb;
const char* user;
const char* parent;
uv_fs_t req;
int work_left;
} tf_import_t;
static void _tf_ssb_import_scandir(uv_fs_t* req)
{
tf_import_t* import = req->data;
uv_dirent_t ent;
while (uv_fs_scandir_next(req, &ent) == 0) {
size_t len = strlen(import->parent) + strlen(ent.name) + 2;
char* path = malloc(len);
snprintf(path, len, "%s/%s", import->parent, ent.name);
if (ent.type == UV_DIRENT_DIR) {
tf_ssb_import(import->ssb, import->user, path);
} else {
size_t size = sizeof(tf_import_file_t) + strlen(import->parent) +1 + strlen(ent.name) + 1;
tf_import_file_t* file = malloc(size);
memset(file, 0, size);
file->ssb = import->ssb;
file->user = import->user;
file->parent = (void*)(file + 1);
file->name = file->parent + strlen(import->parent) + 1;
file->req.data = file;
file->work_left = &import->work_left;
memcpy((char*)file->parent, import->parent, strlen(import->parent) + 1);
memcpy((char*)file->name, ent.name, strlen(ent.name) + 1);
import->work_left++;
int r = uv_fs_open(tf_ssb_get_loop(import->ssb), &file->req, path, 0, 0, _tf_ssb_import_file_open);
if (r < 0) {
printf("Failed to open %s: %s.\n", path, uv_strerror(r));
free(file);
import->work_left--;
}
}
free(path);
}
import->work_left--;
}
void tf_ssb_import(tf_ssb_t* ssb, const char* user, const char* path)
{
tf_import_t import = {
.ssb = ssb,
.user = user,
.parent = path,
.work_left = 1,
};
import.req.data = &import;
int r = uv_fs_scandir(tf_ssb_get_loop(ssb), &import.req, path, 0, _tf_ssb_import_scandir);
if (r) {
printf("Failed to scan directory %s: %s.", path, uv_strerror(r));
}
while (import.work_left > 0) {
uv_run(tf_ssb_get_loop(ssb), UV_RUN_ONCE);
}
uv_fs_req_cleanup(&import.req);
}

5
src/ssb.import.h Normal file
View File

@ -0,0 +1,5 @@
#pragma once
typedef struct _tf_ssb_t tf_ssb_t;
void tf_ssb_import(tf_ssb_t* ssb, const char* user, const char* path);