tildefriends/src/http.c

#include "http.h"

#include "log.h"
#include "mem.h"
#include "tls.h"
#include "trace.h"
#include "util.js.h"

#include "picohttpparser.h"
#include "uv.h"

#include <assert.h>
#include <stdalign.h>
#include <stdlib.h>
#include <string.h>

#if !defined(__APPLE__) && !defined(__OpenBSD__) && !defined(_WIN32)
#include <alloca.h>
#endif

static const int k_timeout_ms = 60000;

typedef struct _tf_http_connection_t
{
	tf_http_t* http;
	tf_tls_session_t* tls;
	uv_tcp_t tcp;
	uv_shutdown_t shutdown;
	uv_timer_t timeout;

	int ref_count;
	bool is_handshaking;
	bool is_receiving_headers;
	bool is_response_sent;
	bool is_shutting_down;

	const char* method;
	const char* path;
	const char* query;
	int minor_version;

	char incoming[8192];

	char headers_buffer[8192];
	size_t headers_buffer_length;
	int parsed_length;

	struct phr_header headers[32];
	int headers_length;

	tf_http_callback_t* callback;
	const char* trace_name;
	tf_http_request_t* request;
	void* user_data;

	bool is_websocket;
	int websocket_message_index;
	void* body;
	void* fragment;
	int fragment_op_code;
	size_t fragment_length;
	size_t body_length;
	size_t content_length;
	bool connection_close;
} tf_http_connection_t;

typedef struct _tf_http_handler_t
{
	const char* pattern;
	tf_http_callback_t* callback;
	tf_http_cleanup_t* cleanup;
	void* user_data;
} tf_http_handler_t;

typedef struct _tf_http_listener_t
{
	tf_http_t* http;
	tf_tls_context_t* tls;
	uv_tcp_t tcp;
	tf_http_cleanup_t* cleanup;
	void* user_data;
} tf_http_listener_t;

typedef struct _tf_http_t
{
	bool is_shutting_down;

	tf_http_listener_t** listeners;
	int listeners_count;

	tf_http_connection_t** connections;
	int connections_count;

	tf_http_handler_t* handlers;
	int handlers_count;

	int pending_closes;
	uv_loop_t* loop;
	tf_trace_t* trace;

	void* user_data;
	tf_http_cleanup_t* user_data_cleanup;
} tf_http_t;

static const char* _http_connection_get_header(const tf_http_connection_t* connection, const char* name);
static void _http_connection_destroy(tf_http_connection_t* connection, const char* reason);
static void _http_timer_reset(tf_http_connection_t* connection);
static void _http_tls_update(tf_http_connection_t* connection);

tf_http_t* tf_http_create(uv_loop_t* loop)
{
	tf_http_t* http = tf_malloc(sizeof(tf_http_t));
	*http = (tf_http_t) {
		.loop = loop,
	};
	return http;
}

void tf_http_set_trace(tf_http_t* http, tf_trace_t* trace)
{
	http->trace = trace;
}

static void _http_allocate_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf)
{
	tf_http_connection_t* connection = handle->data;
	*buf = uv_buf_init(connection->incoming, sizeof(connection->incoming));
}

static bool _http_find_handler(tf_http_t* http, const char* path, tf_http_callback_t** out_callback, const char** out_trace_name, void** out_user_data)
{
	for (int i = 0; i < http->handlers_count; i++)
	{
		if (!http->handlers[i].pattern || !*http->handlers[i].pattern || strcmp(path, http->handlers[i].pattern) == 0 ||
			(*http->handlers[i].pattern && strncmp(path, http->handlers[i].pattern, strlen(http->handlers[i].pattern)) == 0 && path[strlen(http->handlers[i].pattern) - 1] == '/'))
		{
			*out_callback = http->handlers[i].callback;
			*out_trace_name = http->handlers[i].pattern;
			*out_user_data = http->handlers[i].user_data;
			return true;
		}
	}
	return false;
}

static void _http_on_write(uv_write_t* write, int status)
{
	_http_timer_reset(write->data);
	tf_free(write);
}

static void _http_connection_on_close(uv_handle_t* handle)
{
	tf_http_connection_t* connection = handle->data;
	handle->data = NULL;
	_http_connection_destroy(connection, "handle closed");
}

static void _http_request_destroy(tf_http_request_t* request)
{
	tf_http_close_callback* on_close = request->on_close;
	if (on_close)
	{
		tf_trace_t* trace = request->http->trace;
		request->on_close = NULL;
		tf_trace_begin(trace, request->connection && request->connection->trace_name ? request->connection->trace_name : "websocket");
		on_close(request);
		tf_trace_end(trace);
	}
}

static void _http_connection_destroy(tf_http_connection_t* connection, const char* reason)
{
	connection->is_shutting_down = true;

	if (connection->request)
	{
		tf_http_request_t* request = connection->request;
		connection->request = NULL;
		_http_request_destroy(request);
	}

	if (connection->tcp.data && !uv_is_closing((uv_handle_t*)&connection->tcp))
	{
		uv_close((uv_handle_t*)&connection->tcp, _http_connection_on_close);
	}
	if (connection->timeout.data && !uv_is_closing((uv_handle_t*)&connection->timeout))
	{
		uv_close((uv_handle_t*)&connection->timeout, _http_connection_on_close);
	}
	if (connection->tls)
	{
		tf_tls_session_destroy(connection->tls);
		connection->tls = NULL;
	}

	if (connection->ref_count == 0 && !connection->tcp.data && !connection->shutdown.data && !connection->timeout.data)
	{
		tf_http_t* http = connection->http;
		for (int i = 0; i < http->connections_count; i++)
		{
			if (http->connections[i] == connection)
			{
				http->connections[i] = http->connections[--http->connections_count];
			}
		}
		if (connection->body)
		{
			tf_free(connection->body);
			connection->body = NULL;
		}
		if (connection->fragment)
		{
			tf_free(connection->fragment);
			connection->fragment = NULL;
		}
		tf_free(connection);

		if (http->is_shutting_down && http->connections_count == 0)
		{
			tf_http_destroy(http);
		}
	}
}

static void _http_builtin_404_handler(tf_http_request_t* request)
{
	const char* k_payload = tf_http_status_text(404);
	tf_http_respond(request, 404, NULL, 0, k_payload, strlen(k_payload));
}

static void _http_reset_connection(tf_http_connection_t* connection)
{
	connection->fragment_op_code = 0;
	connection->fragment_length = 0;
	connection->body_length = 0;
	connection->content_length = 0;
	connection->headers_buffer_length = 0;
	connection->headers_length = 0;
	connection->is_receiving_headers = true;
	connection->is_response_sent = false;
	connection->parsed_length = 0;
	connection->path = NULL;
}

static void _http_websocket_mask_in_place(uint8_t* p, uint32_t mask, size_t size)
{
	int i = 0;
	for (; ((intptr_t)(p + i)) % alignof(uint32_t); i++)
	{
		p[i] ^= (mask & 0xff);
		mask = ((mask >> 8) & 0xffffff) | ((mask & 0xff) << 24);
	}
	int aligned_start = i;
	for (; i + 4 < (int)size; i += 4)
	{
		*(uint32_t*)(p + i) ^= mask;
	}
	for (; i < (int)size; i++)
	{
		p[i] ^= ((mask >> (8 * ((i - aligned_start) % 4))) & 0xff);
	}
}

static void _http_add_body_bytes(tf_http_connection_t* connection, const void* data, size_t size)
{
	if (connection->is_websocket)
	{
		if (size)
		{
			connection->body = tf_resize_vec(connection->body, connection->body_length + size);
			memcpy((char*)connection->body + connection->body_length, data, size);
			connection->body_length += size;
		}

		while (connection->body_length >= 2)
		{
			uint8_t* p = connection->body;
			uint8_t bits0 = p[0];
			uint8_t bits1 = p[1];
			if ((bits1 & (1 << 7)) == 0)
			{
				/* Unmasked message. */
				_http_connection_destroy(connection, "websocket server received unmasked bytes");
				return;
			}
			uint8_t op_code = bits0 & 0xf;
			bool fin = (bits0 & (1 << 7)) != 0;
			size_t length = bits1 & 0x7f;
			int mask_start = 2;
			if (length == 126)
			{
				length = 0;
				for (int i = 0; i < 2; i++)
				{
					length <<= 8;
					length |= p[2 + i];
				}
				mask_start = 4;
			}
			else if (length == 127)
			{
				length = 0;
				for (int i = 0; i < 8; i++)
				{
					length <<= 8;
					length |= p[2 + i];
				}
				mask_start = 10;
			}
			size_t total_length = mask_start + 4 + length;
			if (connection->body_length >= total_length)
			{
				uint32_t mask = (uint32_t)p[mask_start + 0] | (uint32_t)p[mask_start + 1] << 8 | (uint32_t)p[mask_start + 2] << 16 | (uint32_t)p[mask_start + 3] << 24;
				uint8_t* message = p + mask_start + 4;
				_http_websocket_mask_in_place(message, mask, length);

				if (!fin && !connection->fragment_op_code)
				{
					connection->fragment_op_code = op_code;
				}

				if (!fin || connection->fragment_length)
				{
					connection->fragment = tf_resize_vec(connection->fragment, connection->fragment_length + length);
					memcpy((uint8_t*)connection->fragment + connection->fragment_length, message, length);
					connection->fragment_length += length;
				}

				if (fin)
				{
					if (connection->request->on_message)
					{
						tf_trace_begin(connection->http->trace, connection->trace_name ? connection->trace_name : "websocket");
						connection->request->on_message(connection->request, connection->fragment_length ? connection->fragment_op_code : op_code,
							connection->fragment_length ? connection->fragment : message, connection->fragment_length ? connection->fragment_length : length);
						tf_trace_end(connection->http->trace);
					}
					connection->fragment_length = 0;
				}
				connection->websocket_message_index++;
				if (connection->body_length >= total_length)
				{
					if (connection->body_length > total_length)
					{
						memmove(connection->body, (char*)connection->body + total_length, connection->body_length - total_length);
					}
					connection->body_length -= total_length;
				}
			}
			else
			{
				break;
			}
		}
	}
	else
	{
		size_t fit = tf_min(connection->content_length - connection->body_length, size);
		if (fit > 0)
		{
			memcpy((char*)connection->body + connection->body_length, data, fit);
			connection->body_length += fit;
		}

		if (connection->body_length == connection->content_length)
		{
			tf_http_request_t* request = tf_malloc(sizeof(tf_http_request_t));
			*request = (tf_http_request_t) {
				.http = connection->http,
				.connection = connection,
				.is_tls = connection->tls != NULL,
				.method = connection->method,
				.path = connection->path,
				.query = connection->query,
				.body = connection->body,
				.content_length = connection->content_length,
				.headers = connection->headers,
				.headers_count = connection->headers_length,
				.user_data = connection->user_data,
			};
			connection->request = request;

			if (!connection->http->is_shutting_down)
			{
				tf_http_request_ref(request);
				tf_trace_begin(connection->http->trace, connection->trace_name ? connection->trace_name : "http");
				connection->callback(request);
				tf_trace_end(connection->http->trace);
				tf_http_request_unref(request);
			}
			else
			{
				const char* k_payload = tf_http_status_text(503);
				tf_http_respond(request, 503, NULL, 0, k_payload, strlen(k_payload));
			}
		}
	}
}

static size_t _http_on_read_plain_internal(tf_http_connection_t* connection, const void* data, size_t read_size)
{
	if (connection->is_receiving_headers)
	{
		size_t used_read_size = tf_min(read_size, sizeof(connection->headers_buffer) - connection->headers_buffer_length);
		memcpy(connection->headers_buffer + connection->headers_buffer_length, data, used_read_size);
		connection->headers_buffer_length += used_read_size;

		const char* method = NULL;
		size_t method_length = 0;
		const char* path = NULL;
		size_t path_length = 0;
		size_t header_count = sizeof(connection->headers) / sizeof(*connection->headers);

		int parse_result = phr_parse_request(connection->headers_buffer, connection->headers_buffer_length, &method, &method_length, &path, &path_length,
			&connection->minor_version, connection->headers, &header_count, connection->parsed_length);
		connection->parsed_length = connection->headers_buffer_length;
		if (parse_result > 0)
		{
			connection->is_receiving_headers = false;
			connection->headers_length = header_count;
			connection->method = method;
			((char*)connection->method)[method_length] = '\0';
			connection->path = path;
			((char*)connection->path)[path_length] = '\0';
			char* q = strchr(connection->path, '?');
			if (q)
			{
				*q = '\0';
				connection->query = q + 1;
			}

			connection->connection_close = connection->minor_version == 0;

			for (int i = 0; i < (int)header_count; i++)
			{
				for (size_t j = 0; j < connection->headers[i].name_len; j++)
				{
					if (connection->headers[i].name[j] >= 'A' && connection->headers[i].name[j] <= 'Z')
					{
						((char*)connection->headers[i].name)[j] += 'a' - 'A';
					}
				}
				((char*)connection->headers[i].name)[connection->headers[i].name_len] = '\0';
				((char*)connection->headers[i].value)[connection->headers[i].value_len] = '\0';
				if (strcasecmp(connection->headers[i].name, "content-length") == 0)
				{
					connection->content_length = strtoull(connection->headers[i].value, NULL, 10);
				}
				else if (strcasecmp(connection->headers[i].name, "connection") == 0)
				{
					if (strcasecmp(connection->headers[i].value, "close") == 0)
					{
						connection->connection_close = true;
					}
				}
			}

			if (connection->content_length)
			{
				connection->body = tf_realloc(connection->body, connection->content_length);
			}

			if (!_http_find_handler(connection->http, connection->path, &connection->callback, &connection->trace_name, &connection->user_data) || !connection->callback)
			{
				connection->callback = _http_builtin_404_handler;
				connection->trace_name = "404";
			}
			size_t consumed = read_size - (connection->headers_buffer_length - parse_result) - (read_size - used_read_size);
			_http_add_body_bytes(connection, NULL, 0);
			return consumed;
		}
		else if (parse_result == -2)
		{
			/* Incomplete.  Will try again next time. */
			return used_read_size;
		}
		else
		{
			tf_printf("phr_parse_request: %d\n", parse_result);
			_http_connection_destroy(connection, "failed to parse request headers");
			return used_read_size;
		}
	}
	else
	{
		_http_add_body_bytes(connection, data, read_size);
		return read_size;
	}
}

static void _http_on_read_plain(tf_http_connection_t* connection, const void* data, size_t read_size)
{
	size_t total_consumed = 0;
	while (total_consumed < read_size)
	{
		size_t consumed = _http_on_read_plain_internal(connection, ((const uint8_t*)data) + total_consumed, read_size - total_consumed);
		if (!consumed)
		{
			_http_connection_destroy(connection, "_http_on_read_plain_internal didn't consume any data");
			break;
		}
		total_consumed += consumed;
	}
}

static void _http_on_read(uv_stream_t* stream, ssize_t read_size, const uv_buf_t* buffer)
{
	tf_http_connection_t* connection = stream->data;
	_http_timer_reset(connection);
	if (read_size > 0)
	{
		if (connection->tls)
		{
			if (tf_tls_session_write_encrypted(connection->tls, buffer->base, read_size) < 0)
			{
				_http_connection_destroy(connection, "tf_tls_session_write_encrypted");
			}
			else
			{
				_http_tls_update(connection);
			}
		}
		else
		{
			_http_on_read_plain(connection, buffer->base, read_size);
		}
	}
	else if (read_size < 0)
	{
		_http_connection_destroy(connection, uv_strerror(read_size));
	}
}

static void _http_timer_callback(uv_timer_t* timer)
{
	_http_connection_destroy(timer->data, "_http_timer_callback");
}

static void _http_timer_reset(tf_http_connection_t* connection)
{
	if (connection->timeout.data)
	{
		int r = uv_timer_stop(&connection->timeout);
		if (r)
		{
			tf_printf("uv_timer_stop: %s\n", uv_strerror(r));
		}
		r = uv_timer_start(&connection->timeout, _http_timer_callback, k_timeout_ms, 0);
		if (r)
		{
			tf_printf("uv_timer_start: %s\n", uv_strerror(r));
		}
	}
}

static void _http_on_connection(uv_stream_t* stream, int status)
{
	tf_http_listener_t* listener = stream->data;
	tf_http_t* http = listener->http;
	tf_http_connection_t* connection = tf_malloc(sizeof(tf_http_connection_t));
	*connection = (tf_http_connection_t) { .http = http, .tcp = { .data = connection }, .is_receiving_headers = true };
	if (listener->tls)
	{
		connection->tls = tf_tls_context_create_session(listener->tls);
		if (!connection->tls)
		{
			_http_connection_destroy(connection, "tf_tls_context_create_session");
			return;
		}
		tf_tls_session_start_accept(connection->tls);
		connection->is_handshaking = true;
	}
	int r = uv_tcp_init(connection->http->loop, &connection->tcp);
	if (r)
	{
		tf_printf("uv_tcp_init: %s\n", uv_strerror(r));
		_http_connection_destroy(connection, "uv_tcp_init");
		return;
	}

	r = uv_timer_init(connection->http->loop, &connection->timeout);
	connection->timeout.data = connection;
	if (r)
	{
		tf_printf("uv_timer_init: %s\n", uv_strerror(r));
		_http_connection_destroy(connection, "uv_timer_init");
		return;
	}
	r = uv_timer_start(&connection->timeout, _http_timer_callback, k_timeout_ms, 0);
	if (r)
	{
		tf_printf("uv_timer_start: %s\n", uv_strerror(r));
		_http_connection_destroy(connection, "uv_timer_start");
		return;
	}

	r = uv_accept(stream, (uv_stream_t*)&connection->tcp);
	if (r)
	{
		tf_printf("uv_accept: %s\n", uv_strerror(r));
		_http_connection_destroy(connection, "uv_accept");
		return;
	}

	r = uv_read_start((uv_stream_t*)&connection->tcp, _http_allocate_buffer, _http_on_read);
	if (r)
	{
		tf_printf("uv_read_start: %s\n", uv_strerror(r));
		_http_connection_destroy(connection, "uv_read_start");
		return;
	}

	if (connection->tls)
	{
		_http_tls_update(connection);
	}

	http->connections = tf_resize_vec(http->connections, sizeof(tf_http_connection_t*) * (http->connections_count + 1));
	http->connections[http->connections_count++] = connection;
}

int tf_http_listen(tf_http_t* http, int port, tf_tls_context_t* tls, tf_http_cleanup_t* cleanup, void* user_data)
{
	tf_http_listener_t* listener = tf_malloc(sizeof(tf_http_listener_t));
	*listener = (tf_http_listener_t) {
		.http = http,
		.tls = tls,
		.tcp = { .data = listener },
		.cleanup = cleanup,
		.user_data = user_data,
	};
	int r = uv_tcp_init(http->loop, &listener->tcp);
	if (r)
	{
		tf_printf("uv_tcp_init: %s\n", uv_strerror(r));
	}

	if (r == 0)
	{
#if defined(__HAIKU__)
		/*
		** Binding to IPv6 here fails with an odd error, and the socket
		** becomes unusable.  Since we probably want localhost only
		** on this single-user OS, let's just assume IPv4.
		*/
		struct sockaddr_in addr = {
			.sin_family = AF_INET,
			.sin_addr = { .s_addr = INADDR_ANY },
			.sin_port = ntohs(port),
		};
#else
		struct sockaddr_in6 addr = {
			.sin6_family = AF_INET6,
			.sin6_addr = IN6ADDR_ANY_INIT,
			.sin6_port = ntohs(port),
		};
#endif
		r = uv_tcp_bind(&listener->tcp, (struct sockaddr*)&addr, 0);
		if (r)
		{
			tf_printf("%s:%d: uv_tcp_bind: %s\n", __FILE__, __LINE__, uv_strerror(r));
		}
	}

	int assigned_port = 0;
	if (r == 0)
	{
		struct sockaddr_storage name = { 0 };
		int size = (int)sizeof(name);
		r = uv_tcp_getsockname(&listener->tcp, (struct sockaddr*)&name, &size);
		assigned_port = ntohs(((struct sockaddr_in*)&name)->sin_port);
	}

	if (r == 0)
	{
		r = uv_listen((uv_stream_t*)&listener->tcp, 16, _http_on_connection);
		if (r)
		{
			tf_printf("uv_listen: %s\n", uv_strerror(r));
		}
	}

	if (r == 0)
	{
		http->listeners = tf_resize_vec(http->listeners, sizeof(tf_http_listener_t*) * (http->listeners_count + 1));
		http->listeners[http->listeners_count++] = listener;
	}
	return assigned_port;
}

void tf_http_add_handler(tf_http_t* http, const char* pattern, tf_http_callback_t* callback, tf_http_cleanup_t* cleanup, void* user_data)
{
	http->handlers = tf_resize_vec(http->handlers, sizeof(tf_http_handler_t) * (http->handlers_count + 1));
	http->handlers[http->handlers_count++] = (tf_http_handler_t) {
		.pattern = tf_strdup(pattern),
		.callback = callback,
		.cleanup = cleanup,
		.user_data = user_data,
	};
}

static void _http_free_listener_on_close(uv_handle_t* handle)
{
	tf_http_listener_t* listener = handle->data;
	handle->data = NULL;
	tf_free(listener);
}

void tf_http_destroy(tf_http_t* http)
{
	http->is_shutting_down = true;

	for (int i = 0; i < http->connections_count; i++)
	{
		_http_connection_destroy(http->connections[i], "tf_http_destroy");
	}

	for (int i = 0; i < http->listeners_count; i++)
	{
		tf_http_listener_t* listener = http->listeners[i];
		if (listener->cleanup)
		{
			listener->cleanup(listener->user_data);
			listener->cleanup = NULL;
		}
	}

	for (int i = 0; i < http->handlers_count; i++)
	{
		if (http->handlers[i].cleanup)
		{
			http->handlers[i].cleanup(http->handlers[i].user_data);
			http->handlers[i].cleanup = NULL;
		}
	}

	if (http->user_data_cleanup)
	{
		http->user_data_cleanup(http->user_data);
		http->user_data = NULL;
	}

	if (http->connections_count == 0)
	{
		tf_free(http->connections);
		http->connections = NULL;

		for (int i = 0; i < http->listeners_count; i++)
		{
			tf_http_listener_t* listener = http->listeners[i];
			uv_close((uv_handle_t*)&listener->tcp, _http_free_listener_on_close);
		}
		tf_free(http->listeners);
		http->listeners = NULL;
		http->listeners_count = 0;

		for (int i = 0; i < http->handlers_count; i++)
		{
			if (http->handlers[i].pattern)
			{
				tf_free((void*)http->handlers[i].pattern);
				http->handlers[i].pattern = NULL;
			}
		}
		tf_free(http->handlers);
		http->handlers_count = 0;

		tf_free(http);
	}
}

const char* tf_http_status_text(int status)
{
	switch (status)
	{
	case 101:
		return "Switching Protocols";
	case 200:
		return "OK";
	case 303:
		return "See other";
	case 304:
		return "Not Modified";
	case 400:
		return "Bad Request";
	case 401:
		return "Unauthorized";
	case 403:
		return "Forbidden";
	case 404:
		return "File not found";
	case 500:
		return "Internal server error";
	case 503:
		return "Service Unavailable";
	default:
		return "Unknown";
	}
}

static void _http_on_shutdown(uv_shutdown_t* request, int status)
{
	request->data = NULL;
}

static void _http_write_internal(tf_http_connection_t* connection, const void* data, size_t size)
{
	if (size && !connection->is_shutting_down)
	{
		uv_write_t* write = tf_malloc(sizeof(uv_write_t) + size);
		*write = (uv_write_t) { .data = connection };
		memcpy(write + 1, data, size);
		int r = uv_write(write, (uv_stream_t*)&connection->tcp, &(uv_buf_t) { .base = (void*)(write + 1), .len = size }, 1, _http_on_write);
		if (r)
		{
			tf_printf("uv_write: %s\n", uv_strerror(r));
		}
	}
}

static void _http_tls_update(tf_http_connection_t* connection)
{
	bool again = true;
	while (again)
	{
		again = false;

		if (connection->is_handshaking && connection->tls)
		{
			switch (tf_tls_session_handshake(connection->tls))
			{
			case k_tls_handshake_done:
				connection->is_handshaking = false;
				break;
			case k_tls_handshake_more:
				break;
			case k_tls_handshake_failed:
				_http_connection_destroy(connection, "tf_tls_session_handshake");
				return;
			}
		}

		/* Maybe we became disconnected and cleaned up our TLS session. */
		if (connection->tls)
		{
			char buffer[8192];
			int r = tf_tls_session_read_encrypted(connection->tls, buffer, sizeof(buffer));
			if (r > 0)
			{
				_http_write_internal(connection, buffer, r);
				again = true;
			}
		}

		if (connection->tls)
		{
			char buffer[8192];
			int r = tf_tls_session_read_plain(connection->tls, buffer, sizeof(buffer));
			if (r > 0)
			{
				_http_on_read_plain(connection, buffer, r);
				again = true;
			}
		}
	}
}

static void _http_write(tf_http_connection_t* connection, const void* data, size_t size)
{
	_http_timer_reset(connection);
	if (connection->tls)
	{
		int r = tf_tls_session_write_plain(connection->tls, data, size);
		if (r < (ssize_t)size)
		{
			char buffer[8192];
			tf_tls_session_get_error(connection->tls, buffer, sizeof(buffer));
			tf_printf("tf_tls_session_write_plain: %s\n", buffer);
		}
		_http_tls_update(connection);
	}
	else
	{
		_http_write_internal(connection, data, size);
	}
}

void tf_http_request_send(tf_http_request_t* request, const void* data, size_t size)
{
	_http_write(request->connection, data, size);
}

void tf_http_respond(tf_http_request_t* request, int status, const char** headers, int headers_count, const void* body, size_t content_length)
{
	if (request->connection->is_response_sent)
	{
		return;
	}
	request->connection->is_response_sent = true;

	const char* status_text = tf_http_status_text(status);
	/* HTTP/1.x 200 OK\r\n */
	bool sent_content_length = false;
	int headers_length = 8 + 1 + 3 + 1 + strlen(status_text) + 2;
	if (headers)
	{
		for (int i = 0; i < headers_count * 2; i += 2)
		{
			/* Key: Value\r\n */
			headers_length += strlen(headers[i]) + 2 + strlen(headers[i + 1]) + 2;
			if (strcasecmp(headers[i], "content-length") == 0)
			{
				sent_content_length = true;
			}
		}
	}
	/* \r\n */
	headers_length += 2;

	char content_length_buffer[32] = { 0 };
	int content_length_buffer_length = 0;
	if (!sent_content_length && status != 101)
	{
		content_length_buffer_length = snprintf(content_length_buffer, sizeof(content_length_buffer), "Content-Length: %zd\r\n", content_length);
		headers_length += content_length_buffer_length;
	}

	char* buffer = alloca(headers_length + 1);
	int offset = snprintf(buffer, headers_length + 1, "HTTP/1.%d %03d %s\r\n", request->connection->minor_version, status, status_text);
	if (headers)
	{
		for (int i = 0; i < headers_count * 2; i += 2)
		{
			offset += snprintf(buffer + offset, headers_length + 1 - offset, "%s: %s\r\n", headers[i], headers[i + 1]);
		}
	}
	if (!sent_content_length)
	{
		memcpy(buffer + offset, content_length_buffer, content_length_buffer_length);
		offset += content_length_buffer_length;
	}
	offset += snprintf(buffer + offset, headers_length + 1 - offset, "\r\n");
	assert(offset == headers_length);
	_http_write(request->connection, buffer, headers_length);
	if (content_length)
	{
		_http_write(request->connection, body, content_length);
	}
	_http_timer_reset(request->connection);

	if (request->connection->connection_close && !request->connection->shutdown.data)
	{
		request->connection->shutdown.data = request->connection;
		uv_shutdown(&request->connection->shutdown, (uv_stream_t*)&request->connection->tcp, _http_on_shutdown);
	}
}

size_t tf_http_get_body(const tf_http_request_t* request, const void** out_data)
{
	*out_data = request->connection->body;
	return request->connection->content_length;
}

void tf_http_request_ref(tf_http_request_t* request)
{
	request->ref_count++;
	request->connection->ref_count++;
}

void tf_http_request_unref(tf_http_request_t* request)
{
	tf_http_connection_t* connection = request->connection;
	if (--request->ref_count == 0)
	{
		if (connection)
		{
			connection->request = NULL;
		}
		_http_request_destroy(request);
		tf_free(request);
	}

	if (--connection->ref_count == 0)
	{
		if (connection->http->is_shutting_down)
		{
			_http_connection_destroy(connection, "unref during shutdown");
		}
		else if (!connection->is_websocket)
		{
			_http_reset_connection(connection);
		}
	}
}

static const char* _http_connection_get_header(const tf_http_connection_t* connection, const char* name)
{
	for (int i = 0; i < connection->headers_length; i++)
	{
		if (strcasecmp(connection->headers[i].name, name) == 0)
		{
			return connection->headers[i].value;
		}
	}
	return NULL;
}

const char* tf_http_request_get_header(tf_http_request_t* request, const char* name)
{
	return _http_connection_get_header(request->connection, name);
}

void tf_http_request_websocket_upgrade(tf_http_request_t* request)
{
	request->connection->is_websocket = true;
}

void tf_http_set_user_data(tf_http_t* http, void* user_data, tf_http_cleanup_t* cleanup)
{
	if (http->user_data && http->user_data_cleanup)
	{
		http->user_data_cleanup(http->user_data);
	}
	http->user_data = user_data;
	http->user_data_cleanup = cleanup;
}

void* tf_http_get_user_data(tf_http_t* http)
{
	return http->user_data;
}