tildefriends/src/ssb.c

#include "ssb.h"

#include "ssb.connections.h"
#include "ssb.db.h"
#include "trace.h"

#include <assert.h>
#include <malloc.h>
#include <base64c.h>
#include <openssl/evp.h>
#include <openssl/rand.h>
#include <quickjs.h>
#include <sodium/crypto_auth.h>
#include <sodium/crypto_box.h>
#include <sodium/crypto_hash_sha256.h>
#include <sodium/crypto_scalarmult.h>
#include <sodium/crypto_scalarmult_curve25519.h>
#include <sodium/crypto_secretbox.h>
#include <sodium/crypto_sign.h>
#include <sodium/utils.h>
#include <sqlite3.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <uv.h>

#if !defined(_countof)
#define _countof(a) ((int)(sizeof((a)) / sizeof(*(a))))
#endif

static_assert(k_id_base64_len == sodium_base64_ENCODED_LEN(9 + crypto_box_PUBLICKEYBYTES, sodium_base64_VARIANT_ORIGINAL), "k_id_base64_len");
static_assert(k_id_bin_len == crypto_box_PUBLICKEYBYTES, "k_id_bin_len");
static_assert(k_blob_id_len == (sodium_base64_ENCODED_LEN(crypto_hash_sha256_BYTES, sodium_base64_VARIANT_ORIGINAL) + 8), "k_blob_id_len");

const uint8_t k_ssb_network[] = {
	0xd4, 0xa1, 0xcb, 0x88, 0xa6, 0x6f, 0x02, 0xf8,
	0xdb, 0x63, 0x5c, 0xe2, 0x64, 0x41, 0xcc, 0x5d,
	0xac, 0x1b, 0x08, 0x42, 0x0c, 0xea, 0xac, 0x23,
	0x08, 0x39, 0xb7, 0x55, 0x84, 0x5a, 0x9f, 0xfb
};

const char* k_secrets_path = "/.config/tildefriends/secret";

typedef enum {
	k_tf_ssb_state_invalid,
	k_tf_ssb_state_connected,
	k_tf_ssb_state_sent_hello,
	k_tf_ssb_state_sent_identity,
	k_tf_ssb_state_verified,

	k_tf_ssb_state_server_wait_hello,
	k_tf_ssb_state_server_wait_client_identity,
	k_tf_ssb_state_server_verified,

	k_tf_ssb_state_closing,
} tf_ssb_state_t;

enum {
	k_connections_changed_callbacks_max = 8,
	k_tf_ssb_rpc_message_body_length_max = 65536,
};

typedef struct _tf_ssb_broadcast_t tf_ssb_broadcast_t;
typedef struct _tf_ssb_connection_t tf_ssb_connection_t;
typedef struct _tf_ssb_request_t tf_ssb_request_t;

typedef struct _tf_ssb_request_t {
	tf_ssb_request_t* next;
	int32_t request_number;
	tf_ssb_rpc_callback_t* callback;
	void* user_data;
} tf_ssb_request_t;

typedef struct _tf_ssb_broadcast_t {
	tf_ssb_broadcast_t* next;
	time_t ctime;
	time_t mtime;
	char host[256];
	struct sockaddr_in addr;
	uint8_t pub[crypto_sign_PUBLICKEYBYTES];
} tf_ssb_broadcast_t;

typedef struct _tf_ssb_rpc_callback_node_t tf_ssb_rpc_callback_node_t;
typedef struct _tf_ssb_rpc_callback_node_t {
	const char** name;
	tf_ssb_rpc_callback_t* callback;
	tf_ssb_callback_cleanup_t* cleanup;
	void* user_data;
	tf_ssb_rpc_callback_node_t* next;
} tf_ssb_rpc_callback_node_t;

typedef struct _tf_ssb_connections_changed_callback_node_t tf_ssb_connections_changed_callback_node_t;
typedef struct _tf_ssb_connections_changed_callback_node_t {
	tf_ssb_connections_changed_callback_t* callback;
	tf_ssb_callback_cleanup_t* cleanup;
	void* user_data;
	tf_ssb_connections_changed_callback_node_t* next;
} tf_ssb_connections_changed_callback_node_t;

typedef struct _tf_ssb_message_added_callback_node_t tf_ssb_message_added_callback_node_t;
typedef struct _tf_ssb_message_added_callback_node_t {
	tf_ssb_message_added_callback_t* callback;
	tf_ssb_callback_cleanup_t* cleanup;
	void* user_data;
	tf_ssb_message_added_callback_node_t* next;
} tf_ssb_message_added_callback_node_t;

typedef struct _tf_ssb_blob_want_added_callback_node_t tf_ssb_blob_want_added_callback_node_t;
typedef struct _tf_ssb_blob_want_added_callback_node_t {
	tf_ssb_blob_want_added_callback_t* callback;
	tf_ssb_callback_cleanup_t* cleanup;
	void* user_data;
	tf_ssb_blob_want_added_callback_node_t* next;
} tf_ssb_blob_want_added_callback_node_t;

typedef struct _tf_ssb_broadcasts_changed_callback_node_t tf_ssb_broadcasts_changed_callback_node_t;
typedef struct _tf_ssb_broadcasts_changed_callback_node_t {
	tf_ssb_broadcasts_changed_callback_t* callback;
	tf_ssb_callback_cleanup_t* cleanup;
	void* user_data;
	tf_ssb_broadcasts_changed_callback_node_t* next;
} tf_ssb_broadcasts_changed_callback_node_t;

typedef struct _tf_ssb_t {
	bool own_context;
	JSRuntime* runtime;
	JSContext* context;

	tf_trace_t* trace;

	sqlite3* db;
	bool owns_db;

	const char* secrets_path;

	uv_loop_t own_loop;
	uv_loop_t* loop;
	uv_udp_t broadcast_listener;
	uv_udp_t broadcast_sender;
	uv_timer_t broadcast_cleanup_timer;
	uv_timer_t broadcast_timer;
	uv_timer_t trace_timer;
	uv_tcp_t server;

	uint8_t pub[crypto_sign_PUBLICKEYBYTES];
	uint8_t priv[crypto_sign_SECRETKEYBYTES];

	int messages_stored;
	int rpc_in;
	int rpc_out;

	tf_ssb_connection_t* connections;
	int connections_count;

	tf_ssb_connections_t* connections_tracker;

	tf_ssb_broadcast_t* broadcasts;
	int broadcasts_count;

	tf_ssb_rpc_callback_node_t* rpc;
	int rpc_count;

	tf_ssb_connections_changed_callback_node_t* connections_changed;
	int connections_changed_count;

	tf_ssb_message_added_callback_node_t* message_added;
	int message_added_count;

	tf_ssb_blob_want_added_callback_node_t* blob_want_added;
	int blob_want_added_count;

	tf_ssb_broadcasts_changed_callback_node_t* broadcasts_changed;
	int broadcasts_changed_count;
} tf_ssb_t;

typedef struct _tf_ssb_connection_t {
	tf_ssb_t* ssb;
	uv_tcp_t tcp;
	uv_connect_t connect;
	uv_async_t async;

	JSValue object;

	char host[256];
	int port;

	tf_ssb_state_t state;

	uint8_t epub[crypto_box_PUBLICKEYBYTES];
	uint8_t epriv[crypto_box_SECRETKEYBYTES];

	uint8_t serverpub[crypto_box_PUBLICKEYBYTES];
	uint8_t serverepub[crypto_box_PUBLICKEYBYTES];

	uint8_t detached_signature_A[crypto_sign_BYTES];

	uint8_t s_to_c_box_key[crypto_hash_sha256_BYTES];
	uint8_t c_to_s_box_key[crypto_hash_sha256_BYTES];

	uint8_t recv_buffer[8192];
	size_t recv_size;

	uint8_t nonce[crypto_secretbox_NONCEBYTES];
	uint8_t send_nonce[crypto_secretbox_NONCEBYTES];

	uint16_t body_len;
	uint8_t body_auth_tag[16];

	uint8_t rpc_recv_buffer[8 * 1024 * 1024];
	uint8_t pad;
	size_t rpc_recv_size;

	uint32_t send_request_number;

	tf_ssb_connection_t* next;
	tf_ssb_request_t* requests;
	const char* destroy_reason;
} tf_ssb_connection_t;

static JSClassID _connection_class_id;

static void _tf_ssb_connection_destroy(tf_ssb_connection_t* connection, const char* reason);
static void _tf_ssb_connection_client_send_hello(uv_stream_t* stream);
static void _tf_ssb_connection_on_close(uv_handle_t* handle);
static void _tf_ssb_connection_close(tf_ssb_connection_t* connection, const char* reason);
static void _tf_ssb_nonce_inc(uint8_t* nonce);
static void _tf_ssb_write(tf_ssb_connection_t* connection, void* data, size_t size);
static void _tf_ssb_connection_finalizer(JSRuntime* runtime, JSValue value);

static void _tf_ssb_connection_send_close(tf_ssb_connection_t* connection)
{
	uint8_t message_enc[34];

	uint8_t nonce1[crypto_secretbox_NONCEBYTES];
	memcpy(nonce1, connection->send_nonce, sizeof(nonce1));
	_tf_ssb_nonce_inc(connection->send_nonce);

	uint8_t header[18] = { 0 };
	if (crypto_secretbox_easy(message_enc, header, sizeof(header), nonce1, connection->c_to_s_box_key) == 0)
	{
		_tf_ssb_write(connection, message_enc, sizeof(message_enc));
	}
	_tf_ssb_connection_close(connection, "crypto_secretbox_easy close message");
}

static void _tf_ssb_connection_close(tf_ssb_connection_t* connection, const char* reason)
{
	if (connection->state == k_tf_ssb_state_closing)
	{
		return;
	}
	else if (connection->state == k_tf_ssb_state_verified ||
		connection->state == k_tf_ssb_state_server_verified)
	{
		printf("Connection %p is closing: %s.\n", connection, reason);
		connection->state = k_tf_ssb_state_closing;
		_tf_ssb_connection_send_close(connection);
	}
	_tf_ssb_connection_destroy(connection, reason);
}

static void _tf_ssb_connection_on_tcp_alloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf)
{
	tf_ssb_connection_t* connection = handle->data;
	size_t malloc_size = sizeof(connection->recv_buffer) - connection->recv_size;
	buf->base = malloc_size ? malloc(malloc_size) : NULL;
	buf->len = malloc_size;
}

static void _tf_ssb_connection_on_write(uv_write_t* req, int status)
{
	free(req);
}

static void _tf_ssb_write(tf_ssb_connection_t* connection, void* data, size_t size)
{
	uv_write_t* write = malloc(sizeof(uv_write_t) + size);
	*write = (uv_write_t) { .data = connection };
	memcpy(write + 1, data, size);
	int result = uv_write(write, (uv_stream_t*)&connection->tcp, &(uv_buf_t) { .base = (char*)(write + 1), .len = size }, 1, _tf_ssb_connection_on_write);
	if (result)
	{
		_tf_ssb_connection_close(connection, "write failed");
	}
}

static void _tf_ssb_connection_send_identity(tf_ssb_connection_t* connection, uint8_t* hmac, uint8_t* pubkey)
{
	memcpy(connection->serverepub, pubkey, sizeof(connection->serverepub));
	if (crypto_auth_hmacsha512256_verify(hmac, connection->serverepub, 32, k_ssb_network) != 0)
	{
		_tf_ssb_connection_close(connection, "invalid server hello");
		return;
	}

	uint8_t shared_secret_ab[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_ab, connection->epriv, connection->serverepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_ab as client");
		return;
	}

	uint8_t servercurvepub[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_sign_ed25519_pk_to_curve25519(servercurvepub, connection->serverpub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute key to curve25519 as client");
		return;
	}

	uint8_t shared_secret_aB[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_aB, connection->epriv, servercurvepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_aB as client");
		return;
	}

	uint8_t hash[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash, shared_secret_ab, sizeof(shared_secret_ab));

	uint8_t msg[sizeof(k_ssb_network) + sizeof(connection->serverpub) + crypto_hash_sha256_BYTES];
	memcpy(msg, k_ssb_network, sizeof(k_ssb_network));
	memcpy(msg + sizeof(k_ssb_network), connection->serverpub, sizeof(connection->serverpub));
	memcpy(msg + sizeof(k_ssb_network) + sizeof(connection->serverpub), hash, sizeof(hash));

	unsigned long long siglen;
	if (crypto_sign_detached(connection->detached_signature_A, &siglen, msg, sizeof(msg), connection->ssb->priv) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute detached_signature_A as client");
		return;
	}

	uint8_t tosend[crypto_sign_BYTES + sizeof(connection->ssb->pub)];
	memcpy(tosend, connection->detached_signature_A, sizeof(connection->detached_signature_A));
	memcpy(tosend + sizeof(connection->detached_signature_A), connection->ssb->pub, sizeof(connection->ssb->pub));
	uint8_t nonce[crypto_secretbox_NONCEBYTES] = { 0 };

	uint8_t tohash[sizeof(k_ssb_network) + sizeof(shared_secret_ab) + sizeof(shared_secret_aB)];
	memcpy(tohash, k_ssb_network, sizeof(k_ssb_network));
	memcpy(tohash + sizeof(k_ssb_network), shared_secret_ab, sizeof(shared_secret_ab));
	memcpy(tohash + sizeof(k_ssb_network) + sizeof(shared_secret_ab), shared_secret_aB, sizeof(shared_secret_aB));
	uint8_t hash2[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash2, tohash, sizeof(tohash));

	uint8_t c[crypto_secretbox_MACBYTES + sizeof(tosend)];
	if (crypto_secretbox_easy(c, tosend, sizeof(tosend), nonce, hash2) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to create initial secretbox as client");
		return;
	}

	static_assert(sizeof(c) == 112, "client send size");
	_tf_ssb_write(connection, c, sizeof(c));
	connection->state = k_tf_ssb_state_sent_identity;
}


static void _tf_ssb_nonce_inc(uint8_t* nonce)
{
	int i = 23;
	while (++nonce[i] == 0 && i > 0)
	{
		i--;
	}
}

static void _tf_ssb_connection_box_stream_send(tf_ssb_connection_t* connection, const uint8_t* message, size_t size)
{
	uint8_t* message_enc = malloc(size + 34);

	uint8_t nonce1[crypto_secretbox_NONCEBYTES];
	memcpy(nonce1, connection->send_nonce, sizeof(nonce1));
	_tf_ssb_nonce_inc(connection->send_nonce);
	uint8_t nonce2[crypto_secretbox_NONCEBYTES];
	memcpy(nonce2, connection->send_nonce, sizeof(nonce2));
	_tf_ssb_nonce_inc(connection->send_nonce);

	if (crypto_secretbox_easy(message_enc + 34 - 16, message, size, nonce2, connection->c_to_s_box_key) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to secretbox message");
		free(message_enc);
		return;
	}

	uint8_t header[18];
	*(uint16_t*)header = htons((uint16_t)size);
	memcpy(header + sizeof(uint16_t), message_enc + 34 - 16, 16);
	if (crypto_secretbox_easy(message_enc, header, sizeof(header), nonce1, connection->c_to_s_box_key) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to secretbox header");
		free(message_enc);
		return;
	}

	_tf_ssb_write(connection, message_enc, size + 34);
	free(message_enc);
}

static bool _tf_ssb_connection_get_request_callback(tf_ssb_connection_t* connection, int32_t request_number, tf_ssb_rpc_callback_t** out_callback, void** out_user_data)
{
	bool found = false;
	for (tf_ssb_request_t* it = connection->requests; it; it = it->next)
	{
		if (it->request_number == request_number)
		{
			if (out_callback)
			{
				*out_callback = it->callback;
			}
			if (out_user_data)
			{
				*out_user_data = it->user_data;
			}
			found = true;
			break;
		}
	}
	return found;
}

void tf_ssb_connection_add_request(tf_ssb_connection_t* connection, int32_t request_number, tf_ssb_rpc_callback_t* callback, void* user_data)
{
	if (_tf_ssb_connection_get_request_callback(connection, request_number, NULL, NULL))
	{
		/* TODO: This leaks the callback. */
		return;
	}
	tf_ssb_request_t* request = malloc(sizeof(tf_ssb_request_t));
	*request = (tf_ssb_request_t)
	{
		.next = connection->requests,
		.request_number = request_number,
		.callback = callback,
		.user_data = user_data,
	};
	connection->requests = request;
}

static void _tf_ssb_connection_remove_request(tf_ssb_connection_t* connection, int32_t request_number)
{
	for (tf_ssb_request_t** it = &connection->requests; *it; it = &(*it)->next)
	{
		if ((*it)->request_number == request_number)
		{
			tf_ssb_request_t* found = *it;
			if (found->user_data)
			{
				JS_FreeValue(tf_ssb_connection_get_context(connection), JS_MKPTR(JS_TAG_OBJECT, found->user_data));
			}
			*it = found->next;
			free(found);
			break;
		}
	}
}

void tf_ssb_connection_rpc_send(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, const uint8_t* message, size_t size, tf_ssb_rpc_callback_t* callback, void* user_data)
{
	if (!connection)
	{
		return;
	}
	if (request_number > 0)
	{
		tf_ssb_connection_add_request(connection, request_number, callback, user_data);
	}
	uint8_t* combined = malloc(9 + size);
	*combined = flags;
	uint32_t u32size = htonl((uint32_t)size);
	memcpy(combined + 1, &u32size, sizeof(u32size));
	uint32_t rn = htonl((uint32_t)request_number);
	memcpy(combined + 1 + sizeof(uint32_t), &rn, sizeof(rn));
	memcpy(combined + 1 + 2 * sizeof(uint32_t), message, size);
	_tf_ssb_connection_box_stream_send(connection, combined, 1 + 2 * sizeof(uint32_t) + size);
	free(combined);
	printf("RPC SEND flags=%x RN=%d: %.*s\n", flags, request_number, (int)size, message);
	connection->ssb->rpc_out++;
}

static int _utf8_len(uint8_t ch)
{
	static const uint8_t k_length[] =
	{
		1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 2, 2, 3, 4
	};
	return k_length[(ch & 0xf0) >> 4];
}

static uint32_t _utf8_decode(uint32_t c)
{
	if (c > 0x7f)
	{
		uint32_t mask = (c <= 0x00efbfbf) ? 0x000f0000 : 0x003f0000;
		c = ((c & 0x07000000) >> 6) |
			((c & mask) >> 4) |
			((c & 0x00003f00) >> 2) |
			(c & 0x0000003f);
	}
	return c;
}

static const uint8_t* _utf8_to_cp(const uint8_t* ch, uint32_t* out_cp)
{
	int len = _utf8_len(*ch);
	int actual_len = 0;
	uint32_t encoding = 0;
	for (int i = 0; i < len && ch[i]; i++, actual_len++)
	{
		encoding = (encoding << 8) | ch[i];
	}
	*out_cp = _utf8_decode(encoding);
	return ch + actual_len;
}

static uint32_t _cp_to_utf16(uint32_t cp, uint16_t* out_h, uint16_t* out_l)
{
	if (cp < 0x10000)
	{
		*out_h = 0;
		*out_l = cp & 0xffff;
		return cp;
	}
	else
	{
		uint32_t t = cp - 0x10000;
		uint32_t h = ((t << 12) >> 22) + 0xd800;
		uint32_t l = ((t << 22) >> 22) + 0xdc00;
		*out_h = h & 0xffff;
		*out_l = l & 0xffff;
		return (h << 16) | (l & 0xffff);
	}
}

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));
	size_t len = 0;
	const char* messagestr = JS_ToCStringLen(context, &len, idval);

	char* latin1 = strdup(messagestr);
	uint8_t* write_pos = (uint8_t*)latin1;
	const uint8_t* p = (const uint8_t*)messagestr;
	while (p && *p)
	{
		uint32_t cp = 0;
		p = _utf8_to_cp(p, &cp);
		uint16_t h = 0;
		uint16_t l = 0;
		_cp_to_utf16(cp, &h, &l);
		if (h)
		{
			*write_pos++ = h & 0xff;
		}
		if (l)
		{
			*write_pos++ = l & 0xff;
		}
	}
	size_t latin1_len = write_pos - (uint8_t*)latin1;
	*write_pos++ = '\0';

	uint8_t id[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(id, (uint8_t*)latin1, latin1_len);

	char id_base64[k_id_base64_len];
	base64c_encode(id, sizeof(id), (uint8_t*)id_base64, sizeof(id_base64));

	snprintf(out_id, out_id_size, "%%%s.sha256", id_base64);

	free(latin1);
	JS_FreeCString(context, messagestr);
	JS_FreeValue(context, idval);
}

static bool _tf_ssb_verify_and_strip_signature_internal(JSContext* context, JSValue val, char* out_id, size_t out_id_size, char* out_signature, size_t out_signature_size)
{
	bool verified = false;
	tf_ssb_calculate_message_id(context, val, out_id, out_id_size);
	JSValue signature = JS_GetPropertyStr(context, val, "signature");
	const char* str = JS_ToCString(context, signature);
	JSAtom sigatom = JS_NewAtom(context, "signature");
	JS_DeleteProperty(context, val, sigatom, 0);
	JS_FreeAtom(context, sigatom);

	if (out_signature)
	{
		memset(out_signature, 0, out_signature_size);
		strncpy(out_signature, str, out_signature_size - 1);
	}

	JSValue sigval = JS_JSONStringify(context, val, JS_NULL, JS_NewInt32(context, 2));
	const char* sigstr = JS_ToCString(context, sigval);
	const char* sigkind = strstr(str, ".sig.ed25519");

	JSValue authorval = JS_GetPropertyStr(context, val, "author");
	const char* author = JS_ToCString(context, authorval);
	const char* author_id = author && *author == '@' ? author + 1 : author;
	const char* type = strstr(author_id, ".ed25519");

	uint8_t publickey[crypto_box_PUBLICKEYBYTES];
	int r = base64c_decode((const uint8_t*)author_id, type - author_id, publickey, sizeof(publickey));
	if (r != -1)
	{
		uint8_t binsig[crypto_sign_BYTES];
		r = base64c_decode((const uint8_t*)str, sigkind - str, binsig, sizeof(binsig));
		if (r != -1)
		{
			r = crypto_sign_verify_detached(binsig, (const uint8_t*)sigstr, strlen(sigstr), publickey);
			verified = r == 0;
			if (!verified)
			{
				//printf("crypto_sign_verify_detached fail (r=%d)\n", r);
				if (false)
				{
					printf("val=[%.*s]\n", (int)strlen(sigstr), sigstr);
					printf("sig=%.*s\n", (int)(sigkind - str), str);
					printf("public key=%.*s\n", (int)(type - author_id), author_id);
				}
			}
		}
		else
		{
			printf("base64 decode sig fail [%.*s]\n", (int)(sigkind - str), str);
		}
	}
	else
	{
		printf("base64 decode author[%.*s] fail (%d)\n", (int)(type - author_id), author_id, r);
	}

	JS_FreeCString(context, author);
	JS_FreeCString(context, sigstr);
	JS_FreeCString(context, str);
	JS_FreeValue(context, sigval);
	JS_FreeValue(context, authorval);
	if (verified)
	{
		JS_FreeValue(context, signature);
	}
	else
	{
		JS_SetPropertyStr(context, val, "signature", signature);
	}
	return verified;
}

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)
{
	if (_tf_ssb_verify_and_strip_signature_internal(context, val, out_id, out_id_size, out_signature, out_signature_size))
	{
		if (out_sequence_before_author)
		{
			*out_sequence_before_author = false;
		}
		return true;
	}
	else if (out_sequence_before_author)
	{
		JSValue reordered = JS_NewObject(context);
		JS_SetPropertyStr(context, reordered, "previous", JS_GetPropertyStr(context, val, "previous"));
		JS_SetPropertyStr(context, reordered, "sequence", JS_GetPropertyStr(context, val, "sequence"));
		JS_SetPropertyStr(context, reordered, "author", JS_GetPropertyStr(context, val, "author"));
		JS_SetPropertyStr(context, reordered, "timestamp", JS_GetPropertyStr(context, val, "timestamp"));
		JS_SetPropertyStr(context, reordered, "hash", JS_GetPropertyStr(context, val, "hash"));
		JS_SetPropertyStr(context, reordered, "content", JS_GetPropertyStr(context, val, "content"));
		JS_SetPropertyStr(context, reordered, "signature", JS_GetPropertyStr(context, val, "signature"));
		bool result = _tf_ssb_verify_and_strip_signature_internal(context, reordered, out_id, out_id_size, out_signature, out_signature_size);
		JS_FreeValue(context, reordered);
		if (result)
		{
			*out_sequence_before_author = true;
			return true;
		}
	}
	return false;
}

void tf_ssb_send_close(tf_ssb_t* ssb)
{
	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
	{
		_tf_ssb_connection_send_close(connection);
	}
}

bool tf_ssb_id_bin_to_str(char* str, size_t str_size, const uint8_t* bin)
{
	char buffer[k_id_base64_len - 9];
	base64c_encode(bin, crypto_sign_PUBLICKEYBYTES, (uint8_t*)buffer, sizeof(buffer));
	return snprintf(str, str_size, "@%s.ed25519", buffer) < (int)str_size;
}

bool tf_ssb_id_str_to_bin(uint8_t* bin, const char* str)
{
	const char* author_id = str && *str == '@' ? str + 1 : str;
	const char* type = strstr(str, ".ed25519");
	return base64c_decode((const uint8_t*)author_id, type - author_id, bin, crypto_box_PUBLICKEYBYTES) != 0;
}

static void _tf_ssb_notify_connections_changed(tf_ssb_t* ssb, tf_ssb_change_t change, tf_ssb_connection_t* connection)
{
	tf_ssb_connections_changed_callback_node_t* next = NULL;
	for (tf_ssb_connections_changed_callback_node_t* node = ssb->connections_changed; node; node = next)
	{
		next = node->next;
		node->callback(ssb, change, connection, node->user_data);
	}
}

static void _tf_ssb_connection_verify_identity(tf_ssb_connection_t* connection, const uint8_t* message, size_t len)
{
	uint8_t nonce[crypto_secretbox_NONCEBYTES] = { 0 };

	uint8_t shared_secret_ab[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_ab, connection->epriv, connection->serverepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_ab");
		return;
	}

	uint8_t servercurvepub[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_sign_ed25519_pk_to_curve25519(servercurvepub, connection->serverpub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to convert key to curve25519");
		return;
	}

	uint8_t shared_secret_aB[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_aB, connection->epriv, servercurvepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_aB");
		return;
	}

	uint8_t clientcurvepriv[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_sign_ed25519_sk_to_curve25519(clientcurvepriv, connection->ssb->priv) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to convert key to curve25519");
		return;
	}

	uint8_t shared_secret_Ab[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_Ab, clientcurvepriv, connection->serverepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_Ab");
		return;
	}

	uint8_t tohash[sizeof(k_ssb_network) + sizeof(shared_secret_ab) + sizeof(shared_secret_aB) + sizeof(shared_secret_Ab)];
	memcpy(tohash, k_ssb_network, sizeof(k_ssb_network));
	memcpy(tohash + sizeof(k_ssb_network), shared_secret_ab, sizeof(shared_secret_ab));
	memcpy(tohash + sizeof(k_ssb_network) + sizeof(shared_secret_ab), shared_secret_aB, sizeof(shared_secret_aB));
	memcpy(tohash + sizeof(k_ssb_network) + sizeof(shared_secret_ab) + sizeof(shared_secret_aB), shared_secret_Ab, sizeof(shared_secret_Ab));
	uint8_t hash2[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash2, tohash, sizeof(tohash));

	uint8_t hash3a[crypto_hash_sha256_BYTES + crypto_sign_PUBLICKEYBYTES];
	crypto_hash_sha256(hash3a, hash2, sizeof(hash2));
	memcpy(hash3a + crypto_hash_sha256_BYTES, connection->ssb->pub, sizeof(connection->ssb->pub));
	crypto_hash_sha256(connection->s_to_c_box_key, hash3a, sizeof(hash3a));

	uint8_t hash3b[crypto_hash_sha256_BYTES + crypto_sign_PUBLICKEYBYTES];
	crypto_hash_sha256(hash3b, hash2, sizeof(hash2));
	memcpy(hash3b + crypto_hash_sha256_BYTES, connection->serverpub, sizeof(connection->serverpub));
	crypto_hash_sha256(connection->c_to_s_box_key, hash3b, sizeof(hash3b));

	uint8_t m[80];
	if (crypto_secretbox_open_easy(m, message, len, nonce, hash2) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to open initial secret box as client");
		return;
	}

	uint8_t hash3[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash3, shared_secret_ab, sizeof(shared_secret_ab));

	uint8_t msg[sizeof(k_ssb_network) + sizeof(connection->detached_signature_A) + sizeof(connection->ssb->pub) + sizeof(hash3)];
	memcpy(msg, k_ssb_network, sizeof(k_ssb_network));
	memcpy(msg + sizeof(k_ssb_network), connection->detached_signature_A, sizeof(connection->detached_signature_A));
	memcpy(msg + sizeof(k_ssb_network) + sizeof(connection->detached_signature_A), connection->ssb->pub, sizeof(connection->ssb->pub));
	memcpy(msg + sizeof(k_ssb_network) + sizeof(connection->detached_signature_A) + sizeof(connection->ssb->pub), hash3, sizeof(hash3));
	if (crypto_sign_verify_detached(m, msg, sizeof(msg), connection->serverpub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to verify server identity");
		return;
	}

	uint8_t nonce2[crypto_auth_hmacsha512256_BYTES];
	if (crypto_auth_hmacsha512256(nonce2, connection->epub, sizeof(connection->epub), k_ssb_network) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute client recv nonce");
		return;
	}
	memcpy(connection->nonce, nonce2, sizeof(connection->nonce));

	uint8_t nonce3[crypto_auth_hmacsha512256_BYTES];
	if (crypto_auth_hmacsha512256(nonce3, connection->serverepub, sizeof(connection->serverepub), k_ssb_network) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute client send nonce");
		return;
	}
	memcpy(connection->send_nonce, nonce3, sizeof(connection->send_nonce));

	char fullid[k_id_base64_len];
	tf_ssb_id_bin_to_str(fullid, sizeof(fullid), connection->serverpub);

	JSContext* context = connection->ssb->context;
	JS_SetPropertyStr(context, connection->object, "id", JS_NewString(context, fullid));
	JS_SetPropertyStr(context, connection->object, "is_client", JS_TRUE);

	connection->state = k_tf_ssb_state_verified;
	_tf_ssb_notify_connections_changed(connection->ssb, k_tf_ssb_change_connect, connection);
}

const char* tf_ssb_connection_get_host(tf_ssb_connection_t* connection)
{
	return connection->host;
}

int tf_ssb_connection_get_port(tf_ssb_connection_t* connection)
{
	return connection->port;
}

bool tf_ssb_connection_get_id(tf_ssb_connection_t* connection, char* out_id, size_t out_id_size)
{
	return tf_ssb_id_bin_to_str(out_id, out_id_size, connection->serverpub);
}

static bool _tf_ssb_is_already_connected(tf_ssb_t* ssb, uint8_t* id)
{
	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
	{
		if (memcmp(connection->serverpub, id, k_id_bin_len) == 0)
		{
			return true;
		}
		else if (memcmp(ssb->pub, id, k_id_bin_len) == 0)
		{
			return true;
		}
	}
	return false;
}

static void _tf_ssb_connection_verify_client_identity(tf_ssb_connection_t* connection, const uint8_t* message, size_t len)
{
	uint8_t nonce[crypto_secretbox_NONCEBYTES] = { 0 };

	/*
	** shared_secret_ab = nacl_scalarmult(
	**   server_ephemeral_sk,
	**   client_ephemeral_pk
	** )
	*/
	uint8_t shared_secret_ab[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_ab, connection->epriv, connection->serverepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_ab");
		return;
	}

	/*
	** shared_secret_aB = nacl_scalarmult(
	**   sk_to_curve25519(server_longterm_sk),
	**   client_ephemeral_pk
	** )
	*/
	uint8_t curvepriv[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_sign_ed25519_sk_to_curve25519(curvepriv, connection->ssb->priv) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to convert key to curve25519");
		return;
	}

	static_assert(sizeof(connection->ssb->priv) == crypto_sign_ed25519_SECRETKEYBYTES, "size");
	uint8_t shared_secret_aB[crypto_scalarmult_curve25519_SCALARBYTES] = { 0 };

	if (crypto_scalarmult(shared_secret_aB, curvepriv, connection->serverepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_aB");
		return;
	}

	static_assert(sizeof(k_ssb_network) == crypto_auth_KEYBYTES, "network key size");
	uint8_t tohash[sizeof(k_ssb_network) + sizeof(shared_secret_ab) + sizeof(shared_secret_aB)];
	memcpy(tohash, k_ssb_network, sizeof(k_ssb_network));
	memcpy(tohash + sizeof(k_ssb_network), shared_secret_ab, sizeof(shared_secret_ab));
	memcpy(tohash + sizeof(k_ssb_network) + sizeof(shared_secret_ab), shared_secret_aB, sizeof(shared_secret_aB));
	uint8_t hash2[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash2, tohash, sizeof(tohash));

	/*
	** msg3_plaintext = assert_nacl_secretbox_open(
	**   ciphertext: msg3,
	**   nonce: 24_bytes_of_zeros,
	**   key: sha256(
	**     concat(
	**       network_identifier,
	**       shared_secret_ab,
	**       shared_secret_aB
	**     )
	**   )
	** )
	*/

	uint8_t m[96];
	if (crypto_secretbox_open_easy(m, message, len, nonce, hash2) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to open initial secret box as server");
		return;
	}
	uint8_t* detached_signature_A = m;

	if (_tf_ssb_is_already_connected(connection->ssb, m + 64))
	{
		_tf_ssb_connection_close(connection, "already connected");
		return;
	}

	memcpy(connection->serverpub, m + 64, sizeof(connection->serverpub));

	uint8_t hash3[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(hash3, shared_secret_ab, sizeof(shared_secret_ab));

	uint8_t msg[sizeof(k_ssb_network) + sizeof(connection->ssb->pub) + sizeof(hash3)];
	memcpy(msg, k_ssb_network, sizeof(k_ssb_network));
	memcpy(msg + sizeof(k_ssb_network), connection->ssb->pub, sizeof(connection->ssb->pub));
	memcpy(msg + sizeof(k_ssb_network) + sizeof(connection->ssb->pub), hash3, sizeof(hash3));
	if (crypto_sign_verify_detached(detached_signature_A, msg, sizeof(msg), connection->serverpub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to verify client identity");
		return;
	}

	uint8_t nonce2[crypto_auth_hmacsha512256_BYTES];
	if (crypto_auth_hmacsha512256(nonce2, connection->epub, sizeof(connection->epub), k_ssb_network) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute initial recv nonce as server");
		return;
	}
	memcpy(connection->nonce, nonce2, sizeof(connection->nonce));

	uint8_t nonce3[crypto_auth_hmacsha512256_BYTES];
	if (crypto_auth_hmacsha512256(nonce3, connection->serverepub, sizeof(connection->serverepub), k_ssb_network) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute initial send nonce as server");
		return;
	}
	memcpy(connection->send_nonce, nonce3, sizeof(connection->send_nonce));

	int detached_signature_A_size = 64;
	uint8_t sign_b[sizeof(k_ssb_network) + detached_signature_A_size + sizeof(connection->serverpub) + sizeof(hash3)];
	memcpy(sign_b, k_ssb_network, sizeof(k_ssb_network));
	memcpy(sign_b + sizeof(k_ssb_network), detached_signature_A, detached_signature_A_size);
	memcpy(sign_b + sizeof(k_ssb_network) + detached_signature_A_size, connection->serverpub, sizeof(connection->serverpub));
	memcpy(sign_b + sizeof(k_ssb_network) + detached_signature_A_size + sizeof(connection->serverpub), hash3, sizeof(hash3));

	uint8_t detached_signature_B[crypto_sign_BYTES];
	unsigned long long siglen;
	if (crypto_sign_detached(detached_signature_B, &siglen, sign_b, sizeof(sign_b), connection->ssb->priv) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute detached_signature_B as server");
		return;
	}

	uint8_t clientcurvepub[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_sign_ed25519_pk_to_curve25519(clientcurvepub, connection->serverpub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to convert key to curve25519");
		return;
	}

	uint8_t shared_secret_Ab[crypto_scalarmult_curve25519_SCALARBYTES];
	if (crypto_scalarmult_curve25519(shared_secret_Ab, connection->epriv, clientcurvepub) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to compute shared_secret_Ab as server");
		return;
	}

	uint8_t key_buf[sizeof(k_ssb_network) + sizeof(shared_secret_ab) + sizeof(shared_secret_aB) + sizeof(shared_secret_Ab)];
	memcpy(key_buf, k_ssb_network, sizeof(k_ssb_network));
	memcpy(key_buf + sizeof(k_ssb_network), shared_secret_ab, sizeof(shared_secret_ab));
	memcpy(key_buf + sizeof(k_ssb_network) + sizeof(shared_secret_ab), shared_secret_aB, sizeof(shared_secret_aB));
	memcpy(key_buf + sizeof(k_ssb_network) + sizeof(shared_secret_ab) + sizeof(shared_secret_aB), shared_secret_Ab, sizeof(shared_secret_Ab));

	uint8_t key_hash[crypto_hash_sha256_BYTES];
	crypto_hash_sha256(key_hash, key_buf, sizeof(key_buf));

	uint8_t hash3a[crypto_hash_sha256_BYTES + crypto_sign_PUBLICKEYBYTES];
	crypto_hash_sha256(hash3a, key_hash, sizeof(key_hash));
	memcpy(hash3a + crypto_hash_sha256_BYTES, connection->ssb->pub, sizeof(connection->ssb->pub));
	crypto_hash_sha256(connection->s_to_c_box_key, hash3a, sizeof(hash3a));

	uint8_t hash3b[crypto_hash_sha256_BYTES + crypto_sign_PUBLICKEYBYTES];
	crypto_hash_sha256(hash3b, key_hash, sizeof(key_hash));
	memcpy(hash3b + crypto_hash_sha256_BYTES, connection->serverpub, sizeof(connection->serverpub));
	crypto_hash_sha256(connection->c_to_s_box_key, hash3b, sizeof(hash3b));

	uint8_t c[crypto_secretbox_MACBYTES + sizeof(detached_signature_B)];
	if (crypto_secretbox_easy(c, detached_signature_B, sizeof(detached_signature_B), nonce, key_hash) != 0)
	{
		_tf_ssb_connection_close(connection, "unable to create initial secret box as server");
		return;
	}

	static_assert(sizeof(c) == 80, "server send size");
	_tf_ssb_write(connection, c, sizeof(c));

	char fullid[k_id_base64_len];
	tf_ssb_id_bin_to_str(fullid, sizeof(fullid), connection->serverpub);

	JSContext* context = connection->ssb->context;
	JS_SetPropertyStr(context, connection->object, "id", JS_NewString(context, fullid));
	JS_SetPropertyStr(context, connection->object, "is_client", JS_FALSE);

	connection->state = k_tf_ssb_state_server_verified;
	_tf_ssb_notify_connections_changed(connection->ssb, k_tf_ssb_change_connect, connection);
}

static bool _tf_ssb_connection_recv_pop(tf_ssb_connection_t* connection, uint8_t* buffer, size_t size)
{
	if (connection->recv_size < size)
	{
		return false;
	}

	memcpy(buffer, connection->recv_buffer, size);
	if (connection->recv_size - size)
	{
		memmove(connection->recv_buffer, connection->recv_buffer + size, connection->recv_size - size);
	}
	connection->recv_size -= size;
	return true;
}

static bool _tf_ssb_name_equals(JSContext* context, JSValue object, const char** match)
{
	bool result = true;
	JSValue name = JS_GetPropertyStr(context, object, "name");

	if (JS_IsArray(context, name))
	{
		int length;
		JSValue lengthval = JS_GetPropertyStr(context, name, "length");
		if (JS_ToInt32(context, &length, lengthval) == 0)
		{
			for (int i = 0; i < length; i++)
			{
				if (!match[i])
				{
					result = false;
					break;
				}

				JSValue element = JS_GetPropertyUint32(context, name, i);
				const char* str = JS_ToCString(context, element);
				if (!str || strcmp(str, match[i]) != 0)
				{
					result = false;
				}
				JS_FreeCString(context, str);
				JS_FreeValue(context, element);
			}
			if (result && match[length])
			{
				result = false;
			}
		}
		JS_FreeValue(context, lengthval);
	}
	else
	{
		result = false;
	}

	JS_FreeValue(context, name);
	return result;
}

static void _tf_ssb_connection_rpc_recv(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, const uint8_t* message, size_t size)
{
	connection->ssb->rpc_in++;
	if (flags & k_ssb_rpc_flag_json)
	{
		char id[k_id_base64_len] = "";
		tf_ssb_id_bin_to_str(id, sizeof(id), connection->serverpub);
		printf("RPC RECV from %s flags=%x RN=%d: %.*s\n", id, flags, request_number, (int)size, message);
		JSContext* context = connection->ssb->context;
		JSValue val = JS_ParseJSON(context, (const char*)message, size, NULL);

		if (!JS_IsUndefined(val))
		{
			bool found = false;
			if (JS_IsObject(val))
			{
				for (tf_ssb_rpc_callback_node_t* it = connection->ssb->rpc; it; it = it->next)
				{
					if (_tf_ssb_name_equals(context, val, it->name))
					{
						it->callback(connection, flags, request_number, JS_DupValue(context, val), NULL, 0, it->user_data);
						found = true;
						break;
					}
				}
			}
			if (!found)
			{
				tf_ssb_rpc_callback_t* callback = NULL;
				void* user_data = NULL;
				if (_tf_ssb_connection_get_request_callback(connection, -request_number, &callback, &user_data))
				{
					if (callback)
					{
						callback(connection, flags, request_number, JS_DupValue(context, val), NULL, 0, user_data);
					}
				}
				else
				{
					const char* k_unsupported = "{\"message\": \"method: is not in list of allowed methods\", \"name\": \"Error\", \"stack\": \"none\"}";
					tf_ssb_connection_rpc_send(connection, k_ssb_rpc_flag_json | k_ssb_rpc_flag_end_error | (flags & k_ssb_rpc_flag_stream), -request_number,
						(const uint8_t*)k_unsupported, strlen(k_unsupported), NULL, NULL);
				}
			}
		}
		else
		{
			printf("Failed to parse %.*s\n", (int)size, message);
		}

		JS_FreeValue(context, val);
	}
	else if ((flags & k_ssb_rpc_mask_type) == k_ssb_rpc_flag_binary)
	{
		printf("RPC RECV flags=%x RN=%d: %zd bytes\n", flags, request_number, size);
		tf_ssb_rpc_callback_t* callback = NULL;
		void* user_data = NULL;
		if (_tf_ssb_connection_get_request_callback(connection, -request_number, &callback, &user_data))
		{
			if (callback)
			{
				callback(connection, flags, request_number, JS_UNDEFINED, message, size, user_data);
			}
		}
	}

	if (request_number < 0 &&
		(flags & k_ssb_rpc_flag_end_error))
	{
		_tf_ssb_connection_remove_request(connection, -request_number);
	}
}

static void _tf_ssb_connection_rpc_recv_push(tf_ssb_connection_t* connection, const uint8_t* data, size_t size)
{
	if (connection->rpc_recv_size + size > sizeof(connection->rpc_recv_buffer))
	{
		_tf_ssb_connection_close(connection, "recv buffer overflow");
		return;
	}

	memcpy(connection->rpc_recv_buffer + connection->rpc_recv_size, data, size);
	connection->rpc_recv_size += size;

	while (connection->rpc_recv_size >= 9)
	{
		uint8_t flags = *connection->rpc_recv_buffer;
		uint32_t body_len;
		int32_t request_number;
		memcpy(&body_len, connection->rpc_recv_buffer + 1, sizeof(body_len));
		body_len = htonl(body_len);
		memcpy(&request_number, connection->rpc_recv_buffer + 1 + sizeof(body_len), sizeof(request_number));
		request_number = htonl(request_number);

		size_t rpc_size = 9 + body_len;
		if (connection->rpc_recv_size >= rpc_size)
		{
			uint8_t* end = &connection->rpc_recv_buffer[9 + body_len];
			uint8_t tmp = *end;
			*end = '\0';
			_tf_ssb_connection_rpc_recv(connection, flags, request_number, connection->rpc_recv_buffer + 9, body_len);
			*end = tmp;
			memmove(connection->rpc_recv_buffer, connection->rpc_recv_buffer + rpc_size, connection->rpc_recv_size - rpc_size);
			connection->rpc_recv_size -= rpc_size;
		}
		else
		{
			/* Wait for more body. */
			break;
		}
	}
}

static bool _tf_ssb_connection_box_stream_recv(tf_ssb_connection_t* connection)
{
	if (!connection->body_len)
	{
		uint8_t header_enc[34];
		if (_tf_ssb_connection_recv_pop(connection, header_enc, sizeof(header_enc)))
		{
			uint8_t header[18];
			if (crypto_secretbox_open_easy(header, header_enc, sizeof(header_enc), connection->nonce, connection->s_to_c_box_key) != 0)
			{
				_tf_ssb_connection_close(connection, "failed to open header secret box");
				return false;
			}
			_tf_ssb_nonce_inc(connection->nonce);
			connection->body_len = htons(*(uint16_t*)header);
			memcpy(connection->body_auth_tag, header + sizeof(uint16_t), sizeof(connection->body_auth_tag));
			if (!connection->body_len)
			{
				_tf_ssb_connection_close(connection, "empty body, graceful close");
			}
		}
		else
		{
			return false;
		}
	}

	if (connection->body_len)
	{
		uint8_t buf[16 + k_tf_ssb_rpc_message_body_length_max];
		memcpy(buf, connection->body_auth_tag, sizeof(connection->body_auth_tag));
		if (_tf_ssb_connection_recv_pop(connection, buf + 16, connection->body_len))
		{
			uint8_t body[k_tf_ssb_rpc_message_body_length_max];
			if (crypto_secretbox_open_easy(body, buf, 16 + connection->body_len, connection->nonce, connection->s_to_c_box_key) != 0)
			{
				_tf_ssb_connection_close(connection, "failed to open secret box");
				return false;
			}
			_tf_ssb_nonce_inc(connection->nonce);
			_tf_ssb_connection_rpc_recv_push(connection, body, connection->body_len);
			connection->body_len = 0;
		}
		else
		{
			return false;
		}
	}

	return true;
}

void tf_ssb_append_message(tf_ssb_t* ssb, JSValue message)
{
	char author[k_id_base64_len];
	tf_ssb_id_bin_to_str(author, sizeof(author), ssb->pub);

	char previous_id[crypto_hash_sha256_BYTES * 2];
	int64_t previous_sequence = 0;
	bool have_previous = tf_ssb_db_get_latest_message_by_author(ssb, author, &previous_sequence, previous_id, sizeof(previous_id));

	JSContext* context = ssb->context;
	JSValue root = JS_NewObject(context);

	if (have_previous)
	{
		JS_SetPropertyStr(context, root, "previous", JS_NewString(context, previous_id));
	}
	else
	{
		JS_SetPropertyStr(context, root, "previous", JS_NULL);
	}
	JSValue authorstr = JS_NewString(context, author);
	JS_SetPropertyStr(context, root, "author", authorstr);
	JS_SetPropertyStr(context, root, "sequence", JS_NewInt64(context, previous_sequence + 1));

	int64_t now = (int64_t)time(NULL);
	JS_SetPropertyStr(context, root, "timestamp", JS_NewInt64(context, now * 1000LL));
	JSValue hashstr = JS_NewString(context, "sha256");
	JS_SetPropertyStr(context, root, "hash", hashstr);
	JSValue content = JS_DupValue(context, message);
	JS_SetPropertyStr(context, root, "content", content);

	JSValue jsonval = JS_JSONStringify(context, root, JS_NULL, JS_NewInt32(context, 2));
	size_t len = 0;
	const char* json = JS_ToCStringLen(context, &len, jsonval);

	uint8_t signature[crypto_sign_BYTES];
	unsigned long long siglen;
	bool valid = crypto_sign_detached(signature, &siglen, (const uint8_t*)json, len, ssb->priv) == 0;

	JS_FreeCString(context, json);
	JS_FreeValue(context, jsonval);

	char signature_base64[crypto_sign_BYTES * 2];
	base64c_encode(signature, sizeof(signature), (uint8_t*)signature_base64, sizeof(signature_base64));
	strcat(signature_base64, ".sig.ed25519");
	JSValue sigstr = JS_NewString(context, signature_base64);
	JS_SetPropertyStr(context, root, "signature", sigstr);

	jsonval = JS_JSONStringify(context, root, JS_NULL, JS_NewInt32(context, 2));
	len = 0;
	json = JS_ToCStringLen(context, &len, jsonval);

	JS_FreeCString(context, json);
	JS_FreeValue(context, jsonval);

	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);
		}
		else
		{
			printf("message not stored.\n");
		}
	}
	else
	{
		printf("Failed to verify message signature.\n");
	}

	JS_FreeValue(context, root);
}

void _tf_ssb_connection_destroy(tf_ssb_connection_t* connection, const char* reason)
{
	tf_ssb_t* ssb = connection->ssb;
	if (!connection->destroy_reason)
	{
		connection->destroy_reason = reason;
		printf("destroying connection: %s\n", reason);
	}
	for (tf_ssb_connection_t** it = &connection->ssb->connections; *it; it = &(*it)->next)
	{
		if (*it == connection)
		{
			*it = connection->next;
			connection->next = NULL;
			ssb->connections_count--;
			_tf_ssb_notify_connections_changed(ssb, k_tf_ssb_change_remove, connection);
			break;
		}
	}
	while (connection->requests)
	{
		_tf_ssb_connection_remove_request(connection, connection->requests->request_number);
	}
	if (!JS_IsUndefined(connection->object))
	{
		JSValue object = connection->object;
		connection->object = JS_UNDEFINED;
		JS_SetOpaque(object, NULL);
		JS_FreeValue(ssb->context, object);
	}
	if (connection->async.data && !uv_is_closing((uv_handle_t*)&connection->async))
	{
		uv_close((uv_handle_t*)&connection->async, _tf_ssb_connection_on_close);
	}
	if (connection->tcp.data && !uv_is_closing((uv_handle_t*)&connection->tcp))
	{
		uv_close((uv_handle_t*)&connection->tcp, _tf_ssb_connection_on_close);
	}
	if (connection->connect.data && !uv_is_closing((uv_handle_t*)&connection->connect))
	{
		uv_close((uv_handle_t*)&connection->connect, _tf_ssb_connection_on_close);
	}

	if (JS_IsUndefined(connection->object) &&
		!connection->async.data &&
		!connection->tcp.data &&
		!connection->connect.data)
	{
		free(connection);
	}
}

static void _tf_ssb_connection_on_close(uv_handle_t* handle)
{
	tf_ssb_connection_t* connection = handle->data;
	handle->data = NULL;
	if (connection)
	{
		_tf_ssb_connection_destroy(connection, "handle closed");
	}
}

static void _tf_ssb_connection_on_tcp_recv(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf)
{
	tf_ssb_connection_t* connection = stream->data;
	if (nread >= 0)
	{
		if (connection->recv_size + nread > sizeof(connection->recv_buffer))
		{
			_tf_ssb_connection_close(connection, "recv buffer overflow");
			free(buf->base);
			return;
		}
		memcpy(connection->recv_buffer + connection->recv_size, buf->base, nread);
		connection->recv_size += nread;

		switch (connection->state)
		{
		case k_tf_ssb_state_invalid:
			_tf_ssb_connection_close(connection, "received a message in invalid state");
			break;
		case k_tf_ssb_state_connected:
			_tf_ssb_connection_close(connection, "received a message in connected state");
			break;
		case k_tf_ssb_state_sent_hello:
			{
				uint8_t hello[64];
				if (_tf_ssb_connection_recv_pop(connection, hello, sizeof(hello)))
				{
					_tf_ssb_connection_send_identity(connection, hello, hello + 32);
				}
			}
			break;
		case k_tf_ssb_state_sent_identity:
			{
				uint8_t identity[80];
				if (_tf_ssb_connection_recv_pop(connection, identity, sizeof(identity)))
				{
					_tf_ssb_connection_verify_identity(connection, identity, sizeof(identity));
				}
			}
			break;
		case k_tf_ssb_state_verified:
			uv_async_send(&connection->async);
			break;
		case k_tf_ssb_state_server_wait_hello:
			{
				uint8_t hello[64];
				if (_tf_ssb_connection_recv_pop(connection, hello, sizeof(hello)))
				{
					uint8_t* hmac = hello;
					memcpy(connection->serverepub, hello + crypto_box_PUBLICKEYBYTES, crypto_box_PUBLICKEYBYTES);
					static_assert(sizeof(connection->serverepub) == crypto_box_PUBLICKEYBYTES, "serverepub size");
					if (crypto_auth_hmacsha512256_verify(hmac, connection->serverepub, 32, k_ssb_network) != 0)
					{
						_tf_ssb_connection_close(connection, "crypto_auth_hmacsha512256_verify failed");
					}
					else
					{
						_tf_ssb_connection_client_send_hello((uv_stream_t*)&connection->tcp);
						connection->state = k_tf_ssb_state_server_wait_client_identity;
					}
				}
			}
			break;
		case k_tf_ssb_state_server_wait_client_identity:
			{
				uint8_t identity[112];
				if (_tf_ssb_connection_recv_pop(connection, identity, sizeof(identity)))
				{
					_tf_ssb_connection_verify_client_identity(connection, identity, sizeof(identity));
				}
			}
			break;
		case k_tf_ssb_state_server_verified:
			uv_async_send(&connection->async);
			break;
		case k_tf_ssb_state_closing:
			break;
		}
	}
	else if (nread == UV_ENOBUFS)
	{
		/* Our read buffer is full.  Try harder to process messages. */
		uv_async_send(&connection->async);
	}
	else
	{
		_tf_ssb_connection_close(connection, "read zero");
	}
	free(buf->base);
}

static void _tf_ssb_connection_client_send_hello(uv_stream_t* stream)
{
	tf_ssb_connection_t* connection = stream->data;
	char write[crypto_auth_BYTES + crypto_box_PUBLICKEYBYTES];

	if (crypto_box_keypair(connection->epub, connection->epriv) != 0)
	{
		_tf_ssb_connection_close(connection, "failed to generate ephemeral keypair");
		return;
	}

	uint8_t a[crypto_auth_hmacsha512256_BYTES];
	if (crypto_auth_hmacsha512256(a, connection->epub, sizeof(connection->epub), k_ssb_network) != 0)
	{
		_tf_ssb_connection_close(connection, "failed to create hello message");
		return;
	}

	char* b = write;
	memcpy(b, a, crypto_auth_BYTES);
	memcpy(b + crypto_auth_BYTES, connection->epub, sizeof(connection->epub));

	_tf_ssb_write(connection, b, crypto_auth_BYTES + sizeof(connection->epub));
	connection->state = k_tf_ssb_state_sent_hello;
}

static void _tf_ssb_connection_on_connect(uv_connect_t* connect, int status)
{
	tf_ssb_connection_t* connection = connect->data;
	connect->data = NULL;
	if (status == 0)
	{
		connection->state = k_tf_ssb_state_connected;
		int result = uv_read_start(connect->handle, _tf_ssb_connection_on_tcp_alloc, _tf_ssb_connection_on_tcp_recv);
		if (result)
		{
			printf("uv_read_start => %s\n", uv_strerror(status));
			_tf_ssb_connection_close(connection, "uv_read_start failed");
		}
		else
		{
			_tf_ssb_connection_client_send_hello(connect->handle);
		}
	}
	else
	{
		printf("connect => %s\n", uv_strerror(status));
		_tf_ssb_connection_close(connection, "uv_tcp_connect failed");
	}
}

static bool _tf_ssb_load_keys(tf_ssb_t* ssb)
{
	const char* home = getenv("HOME");
	if (!home)
	{
		return false;
	}

	bool result = false;
	char* json = NULL;
	size_t path_size = strlen(home) + strlen(ssb->secrets_path) + 1;
	char* path = malloc(path_size);
	snprintf(path, path_size, "%s%s", home, ssb->secrets_path);

	FILE* file = fopen(path, "rb");
	if (!file)
	{
		printf("Failed to open %s: %s.\n", path, strerror(errno));
		goto failed;
	}

	if (fseek(file, 0, SEEK_END) != 0)
	{
		printf("Failed to seek %s: %s\n.", path, strerror(errno));
		goto failed;
	}
	long len = ftell(file);
	if (len < 0 ||
		fseek(file, 0, SEEK_SET) != 0)
	{
		printf("Failed to seek %s: %s\n.", path, strerror(errno));
		goto failed;
	}

	json = malloc(len + 1);
	if (fread(json, 1, len, file) != (size_t)len)
	{
		printf("Failed to read %s: %s\n.", path, strerror(errno));
		goto failed;
	}

	json[len] = '\0';

	JSContext* context = ssb->context;
	JSValue root = JS_ParseJSON(context, (const char*)json, len, NULL);

	JSValue pubvalue = JS_GetPropertyStr(context, root, "public");
	size_t pubstrlen = 0;
	const char* pubstr = JS_ToCStringLen(context, &pubstrlen, pubvalue);
	size_t privstrlen = 0;
	JSValue privvalue = JS_GetPropertyStr(context, root, "private");
	const char* privstr = JS_ToCStringLen(context, &privstrlen, privvalue);

	if (pubstr && privstr)
	{
		result =
			base64c_decode((const uint8_t*)pubstr, pubstrlen - strlen(".ed25519"), ssb->pub, sizeof(ssb->pub)) != 0 &&
			base64c_decode((const uint8_t*)privstr, privstrlen - strlen(".ed25519"), ssb->priv, sizeof(ssb->priv)) != 0;
	}

	JS_FreeCString(context, pubstr);
	JS_FreeCString(context, privstr);

	JS_FreeValue(context, pubvalue);
	JS_FreeValue(context, privvalue);

	JS_FreeValue(context, root);

failed:
	if (json)
	{
		free(json);
	}
	if (file)
	{
		fclose(file);
	}
	if (path)
	{
		free(path);
	}
	return result;
}

static bool _tf_ssb_save_keys(tf_ssb_t* ssb)
{
	bool result = false;
	char private_base64[crypto_sign_SECRETKEYBYTES * 2];
	char public_base64[crypto_sign_PUBLICKEYBYTES * 2];
	char private[crypto_sign_SECRETKEYBYTES * 2 + 16];
	char public[crypto_sign_PUBLICKEYBYTES * 2 + 16];
	char id[crypto_sign_PUBLICKEYBYTES * 2 + 16];
	base64c_encode(ssb->pub, sizeof(ssb->pub), (uint8_t*)public_base64, sizeof(public_base64));
	base64c_encode(ssb->priv, sizeof(ssb->priv), (uint8_t*)private_base64, sizeof(private_base64));

	snprintf(private, sizeof(private), "%s.ed25519", private_base64);
	snprintf(public, sizeof(public), "%s.ed25519", public_base64);
	snprintf(id, sizeof(id), "@%s.ed25519", public_base64);

	const char* home = getenv("HOME");
	if (!home)
	{
		return false;
	}

	size_t path_size = strlen(home) + strlen(ssb->secrets_path) + 1;
	char* path = malloc(path_size);
	snprintf(path, path_size, "%s%s", home, ssb->secrets_path);

	JSContext* context = ssb->context;
	JSValue root = JS_NewObject(context);
	JS_SetPropertyStr(context, root, "curve", JS_NewString(context, "ed25519"));
	JS_SetPropertyStr(context, root, "public", JS_NewString(context, public));
	JS_SetPropertyStr(context, root, "private", JS_NewString(context, private));
	JS_SetPropertyStr(context, root, "id", JS_NewString(context, id));

	JSValue jsonval = JS_JSONStringify(context, root, JS_NULL, JS_NewInt32(context, 2));
	size_t len = 0;
	const char* json = JS_ToCStringLen(context, &len, jsonval);

	FILE* file = fopen(path, "wb");
	if (file)
	{
		result = fwrite(json, 1, len, file) == len;
		fclose(file);
	}

	JS_FreeCString(context, json);
	JS_FreeValue(context, jsonval);
	JS_FreeValue(context, root);

	free(path);
	return result;
}

static void _tf_ssb_trace_timer(uv_timer_t* timer)
{
	tf_ssb_t* ssb = timer->data;

	const char* names[] =
	{
		"connections",
		"broadcasts",
		"rpc",
		"connections_changed",
		"message_added",
		"blob_want_added",
		"broadcasts_changed",
	};
	int64_t values[] =
	{
		ssb->connections_count,
		ssb->broadcasts_count,
		ssb->rpc_count,
		ssb->connections_changed_count,
		ssb->message_added_count,
		ssb->blob_want_added_count,
		ssb->broadcasts_changed_count,
	};

	tf_trace_counter(ssb->trace, "ssb", _countof(values), names, values);
}

void tf_ssb_get_stats(tf_ssb_t* ssb, tf_ssb_stats_t* out_stats)
{
	*out_stats = (tf_ssb_stats_t)
	{
		.connections = ssb->connections_count,
		.broadcasts = ssb->broadcasts_count,
		.messages_stored = ssb->messages_stored,
		.rpc_in = ssb->rpc_in,
		.rpc_out = ssb->rpc_out,
		.callbacks =
		{
			.rpc = ssb->rpc_count,
			.connections_changed = ssb->connections_changed_count,
			.message_added = ssb->message_added_count,
			.blob_want_added = ssb->blob_want_added_count,
			.broadcasts_changed = ssb->broadcasts_changed_count,
		},
	};
	ssb->messages_stored = 0;
	ssb->rpc_in = 0;
	ssb->rpc_out = 0;
}

tf_ssb_t* tf_ssb_create(uv_loop_t* loop, JSContext* context, sqlite3* db, const char* secrets_path)
{
	tf_ssb_t* ssb = malloc(sizeof(tf_ssb_t));
	memset(ssb, 0, sizeof(*ssb));
	ssb->secrets_path = secrets_path ? secrets_path : k_secrets_path;
	if (context)
	{
		ssb->context = context;
	}
	else
	{
		ssb->own_context = true;
		ssb->runtime = JS_NewRuntime();
		ssb->context = JS_NewContext(ssb->runtime);
	}

	JS_NewClassID(&_connection_class_id);
	JSClassDef def =
	{
		.class_name = "connection",
		.finalizer = _tf_ssb_connection_finalizer,
	};
	JS_NewClass(JS_GetRuntime(ssb->context), _connection_class_id, &def);

	if (db)
	{
		ssb->db = db;
	}
	else
	{
		sqlite3_open("db.sqlite", &ssb->db);
		ssb->owns_db = true;
	}
	tf_ssb_db_init(ssb);

	if (loop)
	{
		ssb->loop = loop;
	}
	else
	{
		uv_loop_init(&ssb->own_loop);
		ssb->loop = &ssb->own_loop;
	}

	ssb->broadcast_timer.data = ssb;
	uv_timer_init(ssb->loop, &ssb->broadcast_timer);

	ssb->broadcast_sender.data = ssb;
	uv_udp_init(ssb->loop, &ssb->broadcast_sender);
	struct sockaddr_in broadcast_from = {
		.sin_family = AF_INET,
		.sin_addr.s_addr = INADDR_ANY,
	};
	uv_udp_bind(&ssb->broadcast_sender, (struct sockaddr*)&broadcast_from, 0);
	uv_udp_set_broadcast(&ssb->broadcast_sender, 1);

	ssb->trace_timer.data = ssb;
	uv_timer_init(ssb->loop, &ssb->trace_timer);
	uv_timer_start(&ssb->trace_timer, _tf_ssb_trace_timer, 100, 100);
	uv_unref((uv_handle_t*)&ssb->trace_timer);

	if (!_tf_ssb_load_keys(ssb))
	{
		printf("Generating a new keypair.\n");
		tf_ssb_generate_keys(ssb);
		_tf_ssb_save_keys(ssb);
	}

	ssb->connections_tracker = tf_ssb_connections_create(ssb);

	return ssb;
}

sqlite3* tf_ssb_get_db(tf_ssb_t* ssb)
{
	return ssb->db;
}

uv_loop_t* tf_ssb_get_loop(tf_ssb_t* ssb)
{
	return ssb->loop;
}

void tf_ssb_generate_keys(tf_ssb_t* ssb)
{
	crypto_sign_ed25519_keypair(ssb->pub, ssb->priv);
}

void tf_ssb_set_trace(tf_ssb_t* ssb, tf_trace_t* trace)
{
	ssb->trace = trace;
	if (trace && ssb->db)
	{
		tf_trace_sqlite(trace, ssb->db);
	}
}

tf_trace_t* tf_ssb_get_trace(tf_ssb_t* ssb)
{
	return ssb->trace;
}

JSContext* tf_ssb_get_context(tf_ssb_t* ssb)
{
	return ssb->context;
}

static void _tf_ssb_on_handle_close(uv_handle_t* handle)
{
	handle->data = NULL;
}

void tf_ssb_destroy(tf_ssb_t* ssb)
{
	tf_ssb_connections_destroy(ssb->connections_tracker);
	ssb->connections_tracker = NULL;

	if (ssb->broadcast_listener.data && !uv_is_closing((uv_handle_t*)&ssb->broadcast_listener))
	{
		uv_close((uv_handle_t*)&ssb->broadcast_listener, _tf_ssb_on_handle_close);
	}

	if (ssb->broadcast_sender.data && !uv_is_closing((uv_handle_t*)&ssb->broadcast_sender))
	{
		uv_close((uv_handle_t*)&ssb->broadcast_sender, _tf_ssb_on_handle_close);
	}

	if (ssb->broadcast_timer.data && !uv_is_closing((uv_handle_t*)&ssb->broadcast_timer))
	{
		uv_close((uv_handle_t*)&ssb->broadcast_timer, _tf_ssb_on_handle_close);
	}

	if (ssb->broadcast_cleanup_timer.data && !uv_is_closing((uv_handle_t*)&ssb->broadcast_cleanup_timer))
	{
		uv_close((uv_handle_t*)&ssb->broadcast_cleanup_timer, _tf_ssb_on_handle_close);
	}

	if (ssb->trace_timer.data && !uv_is_closing((uv_handle_t*)&ssb->trace_timer))
	{
		uv_close((uv_handle_t*)&ssb->trace_timer, _tf_ssb_on_handle_close);
	}

	if (ssb->server.data && !uv_is_closing((uv_handle_t*)&ssb->server))
	{
		uv_close((uv_handle_t*)&ssb->server, _tf_ssb_on_handle_close);
	}

	while (ssb->broadcast_listener.data ||
		ssb->broadcast_sender.data ||
		ssb->broadcast_timer.data ||
		ssb->broadcast_cleanup_timer.data ||
		ssb->trace_timer.data ||
		ssb->server.data)
	{
		uv_run(ssb->loop, UV_RUN_ONCE);
	}

	if (ssb->loop == &ssb->own_loop)
	{
		uv_loop_close(ssb->loop);
	}
	while (ssb->rpc)
	{
		tf_ssb_rpc_callback_node_t* node = ssb->rpc;
		ssb->rpc = node->next;
		ssb->rpc_count--;
		if (node->cleanup)
		{
			node->cleanup(ssb, node->user_data);
			node->cleanup = NULL;
		}
		free(node);
	}
	while (ssb->connections_changed)
	{
		tf_ssb_connections_changed_callback_node_t* node = ssb->connections_changed;
		ssb->connections_changed = node->next;
		ssb->connections_changed_count--;
		if (node->cleanup)
		{
			node->cleanup(ssb, node->user_data);
		}
		free(node);
	}
	while (ssb->message_added)
	{
		tf_ssb_message_added_callback_node_t* node = ssb->message_added;
		ssb->message_added = node->next;
		ssb->message_added_count--;
		if (node->cleanup)
		{
			node->cleanup(ssb, node->user_data);
		}
		free(node);
	}
	while (ssb->blob_want_added)
	{
		tf_ssb_blob_want_added_callback_node_t* node = ssb->blob_want_added;
		ssb->blob_want_added = node->next;
		ssb->blob_want_added_count--;
		if (node->cleanup)
		{
			node->cleanup(ssb, node->user_data);
		}
		free(node);
	}
	while (ssb->broadcasts_changed)
	{
		tf_ssb_broadcasts_changed_callback_node_t* node = ssb->broadcasts_changed;
		ssb->broadcasts_changed = node->next;
		ssb->broadcasts_changed_count--;
		if (node->cleanup)
		{
			node->cleanup(ssb, node->user_data);
		}
		free(node);
	}
	if (ssb->own_context)
	{
		JS_FreeContext(ssb->context);
		JS_FreeRuntime(ssb->runtime);
	}
	if (ssb->owns_db)
	{
		sqlite3_close(ssb->db);
	}
	while (ssb->broadcasts)
	{
		tf_ssb_broadcast_t* broadcast = ssb->broadcasts;
		ssb->broadcasts = broadcast->next;
		ssb->broadcasts_count--;
		free(broadcast);
	}
	free(ssb);
}

void tf_ssb_run(tf_ssb_t* ssb)
{
	uv_run(ssb->loop, UV_RUN_DEFAULT);
}

static void _tf_ssb_connection_finalizer(JSRuntime* runtime, JSValue value)
{
	tf_ssb_connection_t* connection = JS_GetOpaque(value, _connection_class_id);
	if (connection)
	{
		connection->object = JS_UNDEFINED;
		_tf_ssb_connection_destroy(connection, "object finalized");
	}
}

static void _tf_ssb_connection_send_json_response(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
{
	if (!user_data)
	{
		return;
	}

	void _tf_ssb_on_rpc(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data);
	_tf_ssb_on_rpc(connection, flags, request_number, args, message, size, user_data);
}

static JSValue _tf_ssb_connection_send_json(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv)
{
	tf_ssb_connection_t* connection = JS_GetOpaque(this_val, _connection_class_id);
	if (!connection)
	{
		return JS_UNDEFINED;
	}

	uint32_t request_number = tf_ssb_connection_next_request_number(connection);

	JSValue message_val = JS_JSONStringify(context, argv[0], JS_NULL, JS_NULL);
	size_t size;
	const char* message = JS_ToCStringLen(context, &size, message_val);
	JS_FreeValue(context, message_val);

	tf_ssb_connection_rpc_send(
		connection,
		k_ssb_rpc_flag_json | k_ssb_rpc_flag_stream,
		request_number,
		(const uint8_t*)message,
		size,
		_tf_ssb_connection_send_json_response,
		JS_IsFunction(context, argv[1]) ? JS_VALUE_GET_PTR(JS_DupValue(context, argv[1])) : NULL);
	JS_FreeCString(context, message);
	return JS_UNDEFINED;
}

static void _tf_ssb_connection_process_message_async(uv_async_t* async)
{
	tf_ssb_connection_t* connection = async->data;
	if (_tf_ssb_connection_box_stream_recv(connection))
	{
		uv_async_send(&connection->async);
	}
}

tf_ssb_connection_t* tf_ssb_connection_create(tf_ssb_t* ssb, const char* host, const struct sockaddr_in* addr, const uint8_t* public_key)
{
	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
	{
		if (memcmp(connection->serverpub, public_key, k_id_bin_len) == 0)
		{
			char id[k_id_base64_len];
			tf_ssb_id_bin_to_str(id, sizeof(id), public_key);
			printf("Not connecting to %s:%d, because we are already connected to %s.\n", host, ntohs(addr->sin_port), id);
			return NULL;
		}
		else if (memcmp(ssb->pub, public_key, k_id_bin_len) == 0)
		{
			char id[k_id_base64_len];
			tf_ssb_id_bin_to_str(id, sizeof(id), public_key);
			printf("Not connecting to %s:%d, because they appear to be ourselves %s.\n", host, ntohs(addr->sin_port), id);
			return NULL;
		}
	}

	JSContext* context = ssb->context;
	tf_ssb_connection_t* connection = malloc(sizeof(tf_ssb_connection_t));
	memset(connection, 0, sizeof(*connection));
	connection->ssb = ssb;
	connection->tcp.data = connection;
	connection->connect.data = connection;
	connection->send_request_number = 1;
	snprintf(connection->host, sizeof(connection->host), "%s", host);
	connection->port = ntohs(addr->sin_port);
	connection->async.data = connection;
	uv_async_init(ssb->loop, &connection->async, _tf_ssb_connection_process_message_async);

	connection->object = JS_NewObjectClass(ssb->context, _connection_class_id);
	JS_SetOpaque(connection->object, connection);
	JS_SetPropertyStr(context, connection->object, "send_json", JS_NewCFunction(context, _tf_ssb_connection_send_json, "send_json", 2));
	char public_key_str[k_id_base64_len] = { 0 };
	if (tf_ssb_id_bin_to_str(public_key_str, sizeof(public_key_str), public_key))
	{
		JS_SetPropertyStr(context, connection->object, "id", JS_NewString(context, public_key_str));
		JS_SetPropertyStr(context, connection->object, "is_client", JS_TRUE);
	}

	memcpy(connection->serverpub, public_key, sizeof(connection->serverpub));

	uv_tcp_init(ssb->loop, &connection->tcp);
	printf("uv_tcp_connect\n");
	int result = uv_tcp_connect(&connection->connect, &connection->tcp, (const struct sockaddr*)addr, _tf_ssb_connection_on_connect);
	if (result)
	{
		printf("uv_tcp_connect(%s): %s\n", host, uv_strerror(result));
		_tf_ssb_connection_destroy(connection, "connect failed");
	}
	else
	{
		connection->next = ssb->connections;
		ssb->connections = connection;
		ssb->connections_count++;
		_tf_ssb_notify_connections_changed(ssb, k_tf_ssb_change_create, connection);
	}
	return connection;
}

typedef struct _connect_t {
	tf_ssb_t* ssb;
	uv_getaddrinfo_t req;
	char host[256];
	int port;
	uint8_t key[k_id_bin_len];
} connect_t;

static void _tf_on_connect_getaddrinfo(uv_getaddrinfo_t* addrinfo, int result, struct addrinfo* info)
{
	connect_t* connect = addrinfo->data;
	if (result == 0 && info)
	{
		struct sockaddr_in addr = *(struct sockaddr_in*)info->ai_addr;
		addr.sin_port = htons(connect->port);
		tf_ssb_connection_create(connect->ssb, connect->host, &addr, connect->key);
	}
	else
	{
		printf("getaddrinfo => %s\n", uv_strerror(result));
	}
	free(connect);
	uv_freeaddrinfo(info);
}

void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key)
{
	connect_t* connect = malloc(sizeof(connect_t));
	*connect = (connect_t)
	{
		.ssb = ssb,
		.port = port,
		.req.data = connect,
	};
	snprintf(connect->host, sizeof(connect->host), "%s", host);
	memcpy(connect->key, key, k_id_bin_len);
	int r = uv_getaddrinfo(ssb->loop, &connect->req, _tf_on_connect_getaddrinfo, host, NULL, &(struct addrinfo) { .ai_family = AF_INET });
	if (r < 0)
	{
		printf("uv_getaddrinfo: %s\n", uv_strerror(r));
		free(connect);
	}
}

static void _tf_ssb_on_connection(uv_stream_t* stream, int status)
{
	tf_ssb_t* ssb = stream->data;
	if (status < 0)
	{
		printf("uv_listen failed: %s\n", uv_strerror(status));
		return;
	}

	tf_ssb_connection_t* connection = malloc(sizeof(tf_ssb_connection_t));
	memset(connection, 0, sizeof(*connection));
	connection->ssb = ssb;
	connection->tcp.data = connection;
	connection->send_request_number = 1;
	connection->async.data = connection;
	uv_async_init(ssb->loop, &connection->async, _tf_ssb_connection_process_message_async);

	connection->object = JS_NewObjectClass(ssb->context, _connection_class_id);
	JS_SetPropertyStr(ssb->context, connection->object, "send_json", JS_NewCFunction(ssb->context, _tf_ssb_connection_send_json, "send_json", 2));
	JS_SetOpaque(connection->object, connection);

	if (uv_tcp_init(ssb->loop, &connection->tcp) != 0)
	{
		_tf_ssb_connection_destroy(connection, "init failed");
		return;
	}

	if (uv_accept(stream, (uv_stream_t*)&connection->tcp) != 0)
	{
		_tf_ssb_connection_destroy(connection, "accept failed");
		return;
	}

	connection->next = ssb->connections;
	ssb->connections = connection;
	ssb->connections_count++;
	_tf_ssb_notify_connections_changed(ssb, k_tf_ssb_change_create, connection);

	connection->state = k_tf_ssb_state_server_wait_hello;
	uv_read_start((uv_stream_t*)&connection->tcp, _tf_ssb_connection_on_tcp_alloc, _tf_ssb_connection_on_tcp_recv);
}

static void _tf_ssb_send_broadcast(tf_ssb_t* ssb, struct sockaddr_in* address)
{
	struct sockaddr server_addr;
	int len = (int)sizeof(server_addr);
	if (uv_tcp_getsockname(&ssb->server, &server_addr, &len) != 0 ||
		server_addr.sa_family != AF_INET)
	{
		printf("Unable to get server's address.\n");
	}

	char address_str[256];
	if (uv_ip4_name(address, address_str, sizeof(address_str)) != 0)
	{
		printf("Unable to convert address to string.\n");
	}

	char fullid[k_id_base64_len];
	base64c_encode(ssb->pub, sizeof(ssb->pub), (uint8_t*)fullid, sizeof(fullid));

	char message[512];
	snprintf(message, sizeof(message), "net:%s:%d~shs:%s", address_str, ntohs(((struct sockaddr_in*)&server_addr)->sin_port), fullid);

	uv_buf_t buf = { .base = message, .len = strlen(message) };

	struct sockaddr_in broadcast_addr = { 0 };
	broadcast_addr.sin_family = AF_INET;
	broadcast_addr.sin_port = htons(8008);
	broadcast_addr.sin_addr.s_addr = INADDR_BROADCAST;
	int r = uv_udp_try_send(&ssb->broadcast_sender, &buf, 1, (struct sockaddr*)&broadcast_addr);
	if (r < 0)
	{
		printf("failed to send broadcast %d: %s\n", r, uv_strerror(r));
	}
}

static void _tf_ssb_broadcast_timer(uv_timer_t* timer)
{
	tf_ssb_t* ssb = timer->data;
	uv_interface_address_t* info = NULL;
	int count = 0;
	if (uv_interface_addresses(&info, &count) == 0)
	{
		for (int i = 0; i < count; i++)
		{
			if (!info[i].is_internal &&
				info[i].address.address4.sin_family == AF_INET)
			{
				_tf_ssb_send_broadcast(ssb, &info[i].address.address4);
			}
		}
		uv_free_interface_addresses(info, count);
	}
}

void tf_ssb_server_open(tf_ssb_t* ssb, int port)
{
	if (ssb->server.data)
	{
		printf("Already listening.\n");
		return;
	}

	ssb->server.data = ssb;
	if (uv_tcp_init(ssb->loop, &ssb->server) != 0)
	{
		printf("uv_tcp_init failed\n");
		return;
	}

	struct sockaddr_in addr = { 0 };
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr.s_addr = INADDR_ANY;
	if (uv_tcp_bind(&ssb->server, (struct sockaddr*)&addr, 0) != 0)
	{
		printf("uv_tcp_bind failed\n");
		return;
	}

	int status = uv_listen((uv_stream_t*)&ssb->server, SOMAXCONN, _tf_ssb_on_connection);
	if (status != 0)
	{
		printf("uv_listen failed: %s\n", uv_strerror(status));
		/* TODO: cleanup */
		return;
	}

	printf("Starting broadcasts.\n");
	uv_timer_start(&ssb->broadcast_timer, _tf_ssb_broadcast_timer, 2000, 2000);
}

void tf_ssb_server_close(tf_ssb_t* ssb)
{
	if (ssb->server.data && !uv_is_closing((uv_handle_t*)&ssb->server))
	{
		uv_close((uv_handle_t*)&ssb->server, _tf_ssb_on_handle_close);
	}

	uv_timer_stop(&ssb->broadcast_timer);
	printf("Stopped broadcasts.\n");
}

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

static bool _tf_ssb_parse_broadcast(const char* in_broadcast, tf_ssb_broadcast_t* out_broadcast)
{
	char public_key_str[45] = { 0 };
	int port = 0;
	static_assert(sizeof(out_broadcast->host) == 256, "host field size");
	if (sscanf(in_broadcast, "net:%255[0-9.]:%d~shs:%44s", out_broadcast->host, &port, public_key_str) == 3)
	{
		if (uv_inet_pton(AF_INET, out_broadcast->host, &out_broadcast->addr.sin_addr) == 0)
		{
			out_broadcast->addr.sin_family = AF_INET;
			out_broadcast->addr.sin_port = htons((uint16_t)port);
			int r = base64c_decode((const uint8_t*)public_key_str, strlen(public_key_str), out_broadcast->pub, crypto_sign_PUBLICKEYBYTES);
			return r != -1;
		}
		else
		{
			printf("pton failed\n");
		}
	}
	else if (strncmp(in_broadcast, "ws:", 3))
	{
		printf("Unsupported broadcast: %s\n", in_broadcast);
	}
	return false;
}

void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address)
{
	tf_ssb_broadcast_t broadcast = { 0 };
	if (_tf_ssb_parse_broadcast(address, &broadcast))
	{
		tf_ssb_connection_create(ssb, broadcast.host, &broadcast.addr, broadcast.pub);
	}
	else
	{
		printf("Unable to parse: %s\n", address);
	}
}

static void _tf_ssb_on_broadcast_listener_alloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf)
{
	buf->base = malloc(suggested_size);
	buf->len = suggested_size;
}

static void _tf_ssb_notify_broadcasts_changed(tf_ssb_t* ssb)
{
	tf_ssb_broadcasts_changed_callback_node_t* next = NULL;
	for (tf_ssb_broadcasts_changed_callback_node_t* node = ssb->broadcasts_changed; node; node = next)
	{
		next = node->next;
		if (node->callback)
		{
			node->callback(ssb, node->user_data);
		}
	}
}

static void _tf_ssb_add_broadcast(tf_ssb_t* ssb, const tf_ssb_broadcast_t* broadcast)
{
	if (memcmp(broadcast->pub, ssb->pub, sizeof(ssb->pub)) == 0)
	{
		return;
	}

	for (tf_ssb_broadcast_t* node = ssb->broadcasts; node; node = node->next)
	{
		if (node->addr.sin_family == broadcast->addr.sin_family &&
			node->addr.sin_port == broadcast->addr.sin_port &&
			node->addr.sin_addr.s_addr == broadcast->addr.sin_addr.s_addr &&
			memcmp(node->pub, broadcast->pub, sizeof(node->pub)) == 0)
		{
			node->mtime = time(NULL);
			return;
		}
	}

	char key[k_id_base64_len];
	if (tf_ssb_id_bin_to_str(key, sizeof(key), broadcast->pub))
	{
		tf_ssb_connections_store(ssb->connections_tracker, broadcast->host, ntohs(broadcast->addr.sin_port), key);
	}

	printf("Received new broadcast: host=%s, pub=%s.\n", broadcast->host, key);
	tf_ssb_broadcast_t* node = malloc(sizeof(tf_ssb_broadcast_t));
	*node = *broadcast;
	node->next = ssb->broadcasts;
	node->ctime = time(NULL);
	node->mtime = node->ctime;
	ssb->broadcasts = node;
	ssb->broadcasts_count++;

	_tf_ssb_notify_broadcasts_changed(ssb);
}

static void _tf_ssb_on_broadcast_listener_recv(uv_udp_t* handle, ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned flags)
{
	if (nread <= 0)
	{
		free(buf->base);
		return;
	}

	tf_ssb_t* ssb = handle->data;
	((char*)buf->base)[nread] = '\0';

	const char* k_delim = ";";
	char* state = NULL;
	char* entry = strtok_r(buf->base, k_delim, &state);
	while (entry)
	{
		tf_ssb_broadcast_t broadcast = { 0 };
		if (_tf_ssb_parse_broadcast(entry, &broadcast))
		{
			_tf_ssb_add_broadcast(ssb, &broadcast);
		}
		entry = strtok_r(NULL, k_delim, &state);
	}
	free(buf->base);
}

void tf_ssb_visit_broadcasts(tf_ssb_t* ssb, void (*callback)(const struct sockaddr_in* addr, const uint8_t* pub, void* user_data), void* user_data)
{
	time_t now = time(NULL);
	tf_ssb_broadcast_t* next = NULL;
	for (tf_ssb_broadcast_t* node = ssb->broadcasts; node; node = next)
	{
		next = node->next;
		if (node->mtime - now < 60)
		{
			callback(&node->addr, node->pub, user_data);
		}
	}
}

static void _tf_ssb_broadcast_cleanup_timer(uv_timer_t* timer)
{
	tf_ssb_t* ssb = timer->data;
	int modified = 0;
	time_t now = time(NULL);
	for (tf_ssb_broadcast_t** it = &ssb->broadcasts; *it;)
	{
		if ((*it)->mtime < now - 10)
		{
			tf_ssb_broadcast_t* node = *it;
			*it = node->next;
			free(node);
			ssb->broadcasts_count--;
			modified++;
		}
		else
		{
			it = &(*it)->next;
		}
	}

	if (modified)
	{
		_tf_ssb_notify_broadcasts_changed(ssb);
	}
}

void tf_ssb_broadcast_listener_start(tf_ssb_t* ssb, bool linger)
{
	if (ssb->broadcast_listener.data)
	{
		return;
	}

	ssb->broadcast_listener.data = ssb;
	uv_udp_init(ssb->loop, &ssb->broadcast_listener);

	struct sockaddr_in addr = { 0 };
	addr.sin_family = AF_INET;
	addr.sin_port = htons(8008);
	addr.sin_addr.s_addr = INADDR_ANY;
	int result = uv_udp_bind(&ssb->broadcast_listener, (const struct sockaddr*)&addr, UV_UDP_REUSEADDR);
	if (result != 0)
	{
		printf("bind: %s\n", uv_strerror(result));
	}

	result = uv_udp_recv_start(&ssb->broadcast_listener, _tf_ssb_on_broadcast_listener_alloc, _tf_ssb_on_broadcast_listener_recv);
	if (result != 0)
	{
		printf("uv_udp_recv_start: %s\n", uv_strerror(result));
	}

	if (!linger)
	{
		uv_unref((uv_handle_t*)&ssb->broadcast_listener);
	}

	ssb->broadcast_cleanup_timer.data = ssb;
	uv_timer_init(ssb->loop, &ssb->broadcast_cleanup_timer);
	uv_timer_start(&ssb->broadcast_cleanup_timer, _tf_ssb_broadcast_cleanup_timer, 2000, 2000);
}

void tf_ssb_append_post(tf_ssb_t* ssb, const char* text)
{
	JSValue obj = JS_NewObject(ssb->context);
	JS_SetPropertyStr(ssb->context, obj, "type", JS_NewString(ssb->context, "post"));
	JS_SetPropertyStr(ssb->context, obj, "text", JS_NewString(ssb->context, text));
	tf_ssb_append_message(ssb, obj);
	JS_FreeValue(ssb->context, obj);
}

const char** tf_ssb_get_connection_ids(tf_ssb_t* ssb)
{
	int count = 0;
	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
	{
		if (connection->state == k_tf_ssb_state_verified || connection->state == k_tf_ssb_state_server_verified)
		{
			count++;
		}
	}

	char* buffer = malloc(sizeof(char*) * (count + 1) + k_id_base64_len * count);
	char** array = (char**)buffer;
	char* strings = buffer + sizeof(char*) * (count + 1);
	int i = 0;
	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
	{
		if (connection->state == k_tf_ssb_state_verified || connection->state == k_tf_ssb_state_server_verified)
		{
			char* write_pos = strings + k_id_base64_len * i;
			tf_ssb_id_bin_to_str(write_pos, k_id_base64_len, connection->serverpub);
			array[i++] = write_pos;
		}
	}
	array[i] = NULL;
	return (const char**)array;
}

int tf_ssb_get_connections(tf_ssb_t* ssb, tf_ssb_connection_t** out_connections, int out_connections_count)
{
	int i = 0;
	for (tf_ssb_connection_t* connection = ssb->connections;
		connection && i < out_connections_count;
		connection = connection->next)
	{
		out_connections[i++] = connection;
	}
	return i;
}

void tf_ssb_add_broadcasts_changed_callback(tf_ssb_t* ssb, tf_ssb_broadcasts_changed_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data)
{
	tf_ssb_broadcasts_changed_callback_node_t* node = malloc(sizeof(tf_ssb_broadcasts_changed_callback_node_t));
	*node = (tf_ssb_broadcasts_changed_callback_node_t)
	{
		.callback = callback,
		.cleanup = cleanup,
		.user_data = user_data,
		.next = ssb->broadcasts_changed,
	};
	ssb->broadcasts_changed = node;
	ssb->broadcasts_changed_count++;
}

void tf_ssb_remove_broadcasts_changed_callback(tf_ssb_t* ssb, tf_ssb_broadcasts_changed_callback_t* callback, void* user_data)
{
	tf_ssb_broadcasts_changed_callback_node_t** it = &ssb->broadcasts_changed;
	while (*it)
	{
		if ((*it)->callback == callback &&
			(*it)->user_data == user_data)
		{
			tf_ssb_broadcasts_changed_callback_node_t* node = *it;
			*it = node->next;
			ssb->broadcasts_changed_count--;
			if (node->cleanup)
			{
				node->cleanup(ssb, node->user_data);
			}
			free(node);
		}
		else
		{
			it = &(*it)->next;
		}
	}
}

void tf_ssb_add_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_connections_changed_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data)
{
	tf_ssb_connections_changed_callback_node_t* node = malloc(sizeof(tf_ssb_connections_changed_callback_node_t));
	*node = (tf_ssb_connections_changed_callback_node_t)
	{
		.callback = callback,
		.cleanup = cleanup,
		.user_data = user_data,
		.next = ssb->connections_changed,
	};
	ssb->connections_changed = node;
	ssb->connections_changed_count++;
}

void tf_ssb_remove_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_connections_changed_callback_t* callback, void* user_data)
{
	tf_ssb_connections_changed_callback_node_t** it = &ssb->connections_changed;
	while (*it)
	{
		if ((*it)->callback == callback &&
			(*it)->user_data == user_data)
		{
			tf_ssb_connections_changed_callback_node_t* node = *it;
			*it = node->next;
			ssb->connections_changed_count--;
			if (node->cleanup)
			{
				node->cleanup(ssb, node->user_data);
			}
			free(node);
		}
		else
		{
			it = &(*it)->next;
		}
	}
}

void tf_ssb_add_rpc_callback(tf_ssb_t* ssb, const char** name, tf_ssb_rpc_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data)
{
	size_t name_len = 0;
	int name_count = 0;
	for (int i = 0; name[i]; i++)
	{
		name_count++;
		name_len += strlen(name[i]) + 1;
	}
	tf_ssb_rpc_callback_node_t* node = malloc(sizeof(tf_ssb_rpc_callback_node_t) + (name_count + 1) * sizeof(const char*) + name_len);
	*node = (tf_ssb_rpc_callback_node_t)
	{
		.name = (const char**)(node + 1),
		.callback = callback,
		.cleanup = cleanup,
		.user_data = user_data,
		.next = ssb->rpc,
	};
	char* p = (char*)(node + 1) + (name_count + 1) * sizeof(const char*);
	for (int i = 0; i < name_count; i++)
	{
		size_t len = strlen(name[i]);
		memcpy(p, name[i], len + 1);
		node->name[i] = p;
		p += len + 1;
	}
	node->name[name_count] = NULL;
	ssb->rpc = node;
	ssb->rpc_count++;
}

JSContext* tf_ssb_connection_get_context(tf_ssb_connection_t* connection)
{
	return connection->ssb->context;
}

tf_ssb_t* tf_ssb_connection_get_ssb(tf_ssb_connection_t* connection)
{
	return connection->ssb;
}

int32_t tf_ssb_connection_next_request_number(tf_ssb_connection_t* connection)
{
	return connection->send_request_number++;
}

JSClassID tf_ssb_get_connection_class_id()
{
	return _connection_class_id;
}

JSValue tf_ssb_connection_get_object(tf_ssb_connection_t* connection)
{
	return connection ? connection->object : JS_UNDEFINED;
}

void tf_ssb_add_message_added_callback(tf_ssb_t* ssb, void (*callback)(tf_ssb_t* ssb, const char* id, void* user_data), void (*cleanup)(tf_ssb_t* ssb, void* user_data), void* user_data)
{
	tf_ssb_message_added_callback_node_t* node = malloc(sizeof(tf_ssb_message_added_callback_node_t));
	*node = (tf_ssb_message_added_callback_node_t)
	{
		.callback = callback,
		.cleanup = cleanup,
		.user_data = user_data,
		.next = ssb->message_added,
	};
	ssb->message_added = node;
	ssb->message_added_count++;
}

void tf_ssb_remove_message_added_callback(tf_ssb_t* ssb, tf_ssb_message_added_callback_t* callback, void* user_data)
{
	tf_ssb_message_added_callback_node_t** it = &ssb->message_added;
	while (*it)
	{
		if ((*it)->callback == callback &&
			(*it)->user_data == user_data)
		{
			tf_ssb_message_added_callback_node_t* node = *it;
			*it = node->next;
			ssb->message_added_count--;
			if (node->cleanup)
			{
				node->cleanup(ssb, node->user_data);
			}
			free(node);
		}
		else
		{
			it = &(*it)->next;
		}
	}
}

void tf_ssb_notify_message_added(tf_ssb_t* ssb, const char* id)
{
	tf_ssb_message_added_callback_node_t* next = NULL;
	ssb->messages_stored++;
	for (tf_ssb_message_added_callback_node_t* node = ssb->message_added; node; node = next)
	{
		next = node->next;
		node->callback(ssb, id, node->user_data);
	}
}

void tf_ssb_add_blob_want_added_callback(tf_ssb_t* ssb, void (*callback)(tf_ssb_t* ssb, const char* id, void* user_data), void (*cleanup)(tf_ssb_t* ssb, void* user_data), void* user_data)
{
	tf_ssb_blob_want_added_callback_node_t* node = malloc(sizeof(tf_ssb_blob_want_added_callback_node_t));
	*node = (tf_ssb_blob_want_added_callback_node_t)
	{
		.callback = callback,
		.cleanup = cleanup,
		.user_data = user_data,
		.next = ssb->blob_want_added,
	};
	ssb->blob_want_added = node;
	ssb->blob_want_added_count++;
}

void tf_ssb_remove_blob_want_added_callback(tf_ssb_t* ssb, tf_ssb_blob_want_added_callback_t* callback, void* user_data)
{
	tf_ssb_blob_want_added_callback_node_t** it = &ssb->blob_want_added;
	while (*it)
	{
		if ((*it)->callback == callback &&
			(*it)->user_data == user_data)
		{
			tf_ssb_blob_want_added_callback_node_t* node = *it;
			*it = node->next;
			ssb->blob_want_added_count--;
			if (node->cleanup)
			{
				node->cleanup(ssb, node->user_data);
			}
			free(node);
		}
		else
		{
			it = &(*it)->next;
		}
	}
}

void tf_ssb_notify_blob_want_added(tf_ssb_t* ssb, const char* id)
{
	tf_ssb_blob_want_added_callback_node_t* next = NULL;
	for (tf_ssb_blob_want_added_callback_node_t* node = ssb->blob_want_added; node; node = next)
	{
		next = node->next;
		node->callback(ssb, id, node->user_data);
	}
}