path: root/demo-json-rpc-server.c

/*
 * demo-json-rpc-server.c: JSON-RPC 2.0 demo server
 *
 * Copyright (c) 2015, Přemysl Janouch <p.janouch@gmail.com>
 * All rights reserved.
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
 * SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION
 * OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 *
 */

#define print_fatal_data    ((void *) LOG_ERR)
#define print_error_data    ((void *) LOG_ERR)
#define print_warning_data  ((void *) LOG_WARNING)
#define print_status_data   ((void *) LOG_INFO)
#define print_debug_data    ((void *) LOG_DEBUG)

#define LIBERTY_WANT_SSL

#include "config.h"
#include "liberty/liberty.c"

#include <langinfo.h>
#include <locale.h>
#include <signal.h>
#include <strings.h>

#include <ev.h>
#include <jansson.h>
#include <magic.h>

#include "http-parser/http_parser.h"

// --- Extensions to liberty ---------------------------------------------------

// These should be incorporated into the library ASAP

#define UNPACKER_INT_BEGIN                                                     \
	if (self->len - self->offset < sizeof *value)                              \
		return false;                                                          \
	uint8_t *x = (uint8_t *) self->data + self->offset;                        \
	self->offset += sizeof *value;

static bool
msg_unpacker_u16 (struct msg_unpacker *self, uint16_t *value)
{
	UNPACKER_INT_BEGIN
	*value
		= (uint16_t) x[0] << 24 | (uint16_t) x[1] << 16;
	return true;
}

static bool
msg_unpacker_u32 (struct msg_unpacker *self, uint32_t *value)
{
	UNPACKER_INT_BEGIN
	*value
		= (uint32_t) x[0] << 24 | (uint32_t) x[1] << 16
		| (uint32_t) x[2] << 8  | (uint32_t) x[3];
	return true;
}

#undef UNPACKER_INT_BEGIN

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

// "msg_writer" should be rewritten on top of this

static void
str_pack_u8 (struct str *self, uint8_t x)
{
	str_append_data (self, &x, 1);
}

static void
str_pack_u16 (struct str *self, uint64_t x)
{
	uint8_t tmp[2] = { x >> 8, x };
	str_append_data (self, tmp, sizeof tmp);
}

static void
str_pack_u32 (struct str *self, uint32_t x)
{
	uint32_t u = x;
	uint8_t tmp[4] = { u >> 24, u >> 16, u >> 8, u };
	str_append_data (self, tmp, sizeof tmp);
}

static void
str_pack_i32 (struct str *self, int32_t x)
{
	str_pack_u32 (self, (uint32_t) x);
}

static void
str_pack_u64 (struct str *self, uint64_t x)
{
	uint8_t tmp[8] =
		{ x >> 56, x >> 48, x >> 40, x >> 32, x >> 24, x >> 16, x >> 8, x };
	str_append_data (self, tmp, sizeof tmp);
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static int
tolower_ascii (int c)
{
	return c >= 'A' && c <= 'Z' ? c + ('a' - 'A') : c;
}

static size_t
tolower_ascii_strxfrm (char *dest, const char *src, size_t n)
{
	size_t len = strlen (src);
	while (n-- && (*dest++ = tolower_ascii (*src++)))
		;
	return len;
}

static int
strcasecmp_ascii (const char *a, const char *b)
{
	while (*a && tolower_ascii (*a) == tolower_ascii (*b))
	{
		a++;
		b++;
	}
	return *(const unsigned char *) a - *(const unsigned char *) b;
}

static bool
isspace_ascii (int c)
{
	return c == '\f' || c == '\n' || c == '\r' || c == '\t' || c == '\v';
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

/// Return a pointer to the next UTF-8 character, or NULL on error
// TODO: decode the sequence while we're at it
static const char *
utf8_next (const char *s, size_t len)
{
	// End of string, we go no further
	if (!len)
		return NULL;

	// In the middle of a character -> error
	const uint8_t *p = (const unsigned char *) s;
	if ((*p & 0xC0) == 0x80)
		return NULL;

	// Find out how long the sequence is
	unsigned mask = 0xC0;
	unsigned tail_len = 0;
	while ((*p & mask) == mask)
	{
		// Invalid start of sequence
		if (mask == 0xFE)
			return NULL;

		mask |= mask >> 1;
		tail_len++;
	}

	p++;

	// Check the rest of the sequence
	if (tail_len > --len)
		return NULL;

	while (tail_len--)
		if ((*p++ & 0xC0) != 0x80)
			return NULL;

	return (const char *) p;
}

/// Very rough UTF-8 validation, just makes sure codepoints can be iterated
// TODO: also validate the codepoints
static bool
utf8_validate (const char *s, size_t len)
{
	const char *next;
	while (len)
	{
		if (!(next = utf8_next (s, len)))
			return false;

		len -= next - s;
		s = next;
	}
	return true;
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static uint8_t g_base64_table[256] =
{
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 62, 64, 64, 64, 63,
	52, 53, 54, 55, 56, 57, 58, 59,  60, 61, 64, 64, 64,  0, 64, 64,
	64,  0,  1,  2,  3,  4,  5,  6,   7,  8,  9, 10, 11, 12, 13, 14,
	15, 16, 17, 18, 19, 20, 21, 22,  23, 24, 25, 64, 64, 64, 64, 64,
	64, 26, 27, 28, 29, 30, 31, 32,  33, 34, 35, 36, 37, 38, 39, 40,
	41, 42, 43, 44, 45, 46, 47, 48,  49, 50, 51, 64, 64, 64, 64, 64,

	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
	64, 64, 64, 64, 64, 64, 64, 64,  64, 64, 64, 64, 64, 64, 64, 64,
};

static inline bool
base64_decode_group (const char **s, bool ignore_ws, struct str *output)
{
	uint8_t input[4];
	size_t loaded = 0;
	for (; loaded < 4; (*s)++)
	{
		if (!**s)
			return loaded == 0;
		if (!ignore_ws || !isspace_ascii (**s))
			input[loaded++] = **s;
	}

	size_t len = 3;
	if (input[0] == '=' || input[1] == '=')
		return false;
	if (input[2] == '=' && input[3] != '=')
		return false;
	if (input[2] == '=')
		len--;
	if (input[3] == '=')
		len--;

	uint8_t a = g_base64_table[input[0]];
	uint8_t b = g_base64_table[input[1]];
	uint8_t c = g_base64_table[input[2]];
	uint8_t d = g_base64_table[input[3]];

	if (((a | b) | (c | d)) & 0x40)
		return false;

	uint32_t block = a << 18 | b << 12 | c << 6 | d;
	switch (len)
	{
	case 1:
		str_append_c (output, block >> 16);
		break;
	case 2:
		str_append_c (output, block >> 16);
		str_append_c (output, block >> 8);
		break;
	case 3:
		str_append_c (output, block >> 16);
		str_append_c (output, block >> 8);
		str_append_c (output, block);
	}
	return true;
}

static bool
base64_decode (const char *s, bool ignore_ws, struct str *output)
{
	while (*s)
		if (!base64_decode_group (&s, ignore_ws, output))
			return false;
	return true;
}

static void
base64_encode (const void *data, size_t len, struct str *output)
{
	const char *alphabet =
		"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

	const uint8_t *p = data;
	size_t n_groups = len / 3;
	size_t tail = len - n_groups * 3;
	uint32_t group;

	for (; n_groups--; p += 3)
	{
		group = p[0] << 16 | p[1] << 8 | p[2];
		str_append_c (output, alphabet[(group >> 18) & 63]);
		str_append_c (output, alphabet[(group >> 12) & 63]);
		str_append_c (output, alphabet[(group >>  6) & 63]);
		str_append_c (output, alphabet[ group        & 63]);
	}

	switch (tail)
	{
	case 2:
		group = p[0] << 16 | p[1] << 8;
		str_append_c (output, alphabet[(group >> 18) & 63]);
		str_append_c (output, alphabet[(group >> 12) & 63]);
		str_append_c (output, alphabet[(group >>  6) & 63]);
		str_append_c (output, '=');
		break;
	case 1:
		group = p[0] << 16;
		str_append_c (output, alphabet[(group >> 18) & 63]);
		str_append_c (output, alphabet[(group >> 12) & 63]);
		str_append_c (output, '=');
		str_append_c (output, '=');
	default:
		break;
	}
}

// --- HTTP parsing ------------------------------------------------------------

// Basic tokenizer for HTTP header field values, to be used in various parsers.
// The input should already be unwrapped.

// Recommended literature:
//   http://tools.ietf.org/html/rfc7230#section-3.2.6
//   http://tools.ietf.org/html/rfc7230#appendix-B
//   http://tools.ietf.org/html/rfc5234#appendix-B.1

#define HTTP_TOKENIZER_CLASS(name, definition)                                 \
	static inline bool                                                         \
	http_tokenizer_is_ ## name (int c)                                         \
	{                                                                          \
		return (definition);                                                   \
	}

HTTP_TOKENIZER_CLASS (vchar, c >= 0x21 && c <= 0x7E)
HTTP_TOKENIZER_CLASS (delimiter, !!strchr ("\"(),/:;<=>?@[\\]{}", c))
HTTP_TOKENIZER_CLASS (whitespace, c == '\t' || c == ' ')
HTTP_TOKENIZER_CLASS (obs_text, c >= 0x80 && c <= 0xFF)

HTTP_TOKENIZER_CLASS (tchar,
	http_tokenizer_is_vchar (c) && !http_tokenizer_is_delimiter (c))

HTTP_TOKENIZER_CLASS (qdtext,
	c == '\t' || c == ' ' || c == '!'
	|| (c >= 0x23 && c <= 0x5B)
	|| (c >= 0x5D && c <= 0x7E)
	|| http_tokenizer_is_obs_text (c))

HTTP_TOKENIZER_CLASS (quoted_pair,
	c == '\t' || c == ' '
	|| http_tokenizer_is_vchar (c)
	|| http_tokenizer_is_obs_text (c))

#undef HTTP_TOKENIZER_CLASS

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

enum http_tokenizer_token
{
	HTTP_T_EOF,                         ///< Input error
	HTTP_T_ERROR,                       ///< End of input

	HTTP_T_TOKEN,                       ///< "token"
	HTTP_T_QUOTED_STRING,               ///< "quoted-string"
	HTTP_T_DELIMITER,                   ///< "delimiters"
	HTTP_T_WHITESPACE                   ///< RWS/OWS/BWS
};

struct http_tokenizer
{
	const unsigned char *input;         ///< The input string
	size_t input_len;                   ///< Length of the input
	size_t offset;                      ///< Position in the input

	char delimiter;                     ///< The delimiter character
	struct str string;                  ///< "token" / "quoted-string" content
};

static void
http_tokenizer_init (struct http_tokenizer *self, const char *input, size_t len)
{
	memset (self, 0, sizeof *self);
	self->input = (const unsigned char *) input;
	self->input_len = len;

	str_init (&self->string);
}

static void
http_tokenizer_free (struct http_tokenizer *self)
{
	str_free (&self->string);
}

static enum http_tokenizer_token
http_tokenizer_quoted_string (struct http_tokenizer *self)
{
	bool quoted_pair = false;
	while (self->offset < self->input_len)
	{
		int c = self->input[self->offset++];
		if (quoted_pair)
		{
			if (!http_tokenizer_is_quoted_pair (c))
				return HTTP_T_ERROR;

			str_append_c (&self->string, c);
			quoted_pair = false;
		}
		else if (c == '\\')
			quoted_pair = true;
		else if (c == '"')
			return HTTP_T_QUOTED_STRING;
		else if (http_tokenizer_is_qdtext (c))
			str_append_c (&self->string, c);
		else
			return HTTP_T_ERROR;
	}

	// Premature end of input
	return HTTP_T_ERROR;
}

static enum http_tokenizer_token
http_tokenizer_next (struct http_tokenizer *self, bool skip_ows)
{
	str_reset (&self->string);
	if (self->offset >= self->input_len)
		return HTTP_T_EOF;

	int c = self->input[self->offset++];

	if (skip_ows)
		while (http_tokenizer_is_whitespace (c))
		{
			if (self->offset >= self->input_len)
				return HTTP_T_EOF;
			c = self->input[self->offset++];
		}

	if (c == '"')
		return http_tokenizer_quoted_string (self);

	if (http_tokenizer_is_delimiter (c))
	{
		self->delimiter = c;
		return HTTP_T_DELIMITER;
	}

	// Simple variable-length tokens
	enum http_tokenizer_token result;
	bool (*eater) (int c) = NULL;
	if (http_tokenizer_is_whitespace (c))
	{
		eater = http_tokenizer_is_whitespace;
		result = HTTP_T_WHITESPACE;
	}
	else if (http_tokenizer_is_tchar (c))
	{
		eater = http_tokenizer_is_tchar;
		result = HTTP_T_TOKEN;
	}
	else
		return HTTP_T_ERROR;

	str_append_c (&self->string, c);
	while (self->offset < self->input_len)
	{
		if (!eater (c = self->input[self->offset]))
			break;

		str_append_c (&self->string, c);
		self->offset++;
	}
	return result;
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static bool
http_parse_media_type_parameter
	(struct http_tokenizer *t, struct str_map *parameters)
{
	bool result = false;
	char *attribute = NULL;

	if (http_tokenizer_next (t, true) != HTTP_T_TOKEN)
		goto end;
	attribute = xstrdup (t->string.str);

	if (http_tokenizer_next (t, false) != HTTP_T_DELIMITER
	 || t->delimiter != '=')
		goto end;

	switch (http_tokenizer_next (t, false))
	{
	case HTTP_T_TOKEN:
	case HTTP_T_QUOTED_STRING:
		str_map_set (parameters, attribute, xstrdup (t->string.str));
		result = true;
	default:
		break;
	}

end:
	free (attribute);
	return result;
}

/// Parser for "Content-Type".  @a type and @a subtype may be non-NULL
/// even if the function fails.  @a parameters should be case-insensitive.
static bool
http_parse_media_type (const char *media_type,
	char **type, char **subtype, struct str_map *parameters)
{
	bool result = false;
	struct http_tokenizer t;
	http_tokenizer_init (&t, media_type, strlen (media_type));

	if (http_tokenizer_next (&t, true) != HTTP_T_TOKEN)
		goto end;
	*type = xstrdup (t.string.str);

	if (http_tokenizer_next (&t, false) != HTTP_T_DELIMITER
	 || t.delimiter != '/')
		goto end;

	if (http_tokenizer_next (&t, false) != HTTP_T_TOKEN)
		goto end;
	*subtype = xstrdup (t.string.str);

	while (true)
	switch (http_tokenizer_next (&t, true))
	{
	case HTTP_T_DELIMITER:
		if (t.delimiter != ';')
			goto end;
		if (!http_parse_media_type_parameter (&t, parameters))
			goto end;
		break;
	case HTTP_T_EOF:
		result = true;
	default:
		goto end;
	}

end:
	http_tokenizer_free (&t);
	return result;
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

struct http_protocol
{
	LIST_HEADER (struct http_protocol)

	char *name;                         ///< The protocol to upgrade to
	char *version;                      ///< Version of the protocol, if any
};

static void
http_protocol_destroy (struct http_protocol *self)
{
	free (self->name);
	free (self->version);
	free (self);
}

static bool
http_parse_upgrade (const char *upgrade, struct http_protocol **out)
{
	// HTTP grammar makes this more complicated than it should be

	bool result = false;
	struct http_protocol *list = NULL;
	struct http_protocol *tail = NULL;

	struct http_tokenizer t;
	http_tokenizer_init (&t, upgrade, strlen (upgrade));

	enum {
		STATE_PROTOCOL_NAME,
		STATE_SLASH,
		STATE_PROTOCOL_VERSION,
		STATE_EXPECT_COMMA
	} state = STATE_PROTOCOL_NAME;
	struct http_protocol *proto = NULL;

	while (true)
	switch (state)
	{
	case STATE_PROTOCOL_NAME:
		switch (http_tokenizer_next (&t, false))
		{
		case HTTP_T_DELIMITER:
			if (t.delimiter != ',')
				goto end;
		case HTTP_T_WHITESPACE:
			break;
		case HTTP_T_TOKEN:
			proto = xcalloc (1, sizeof *proto);
			proto->name = xstrdup (t.string.str);
			LIST_APPEND_WITH_TAIL (list, tail, proto);
			state = STATE_SLASH;
			break;
		case HTTP_T_EOF:
			result = true;
		default:
			goto end;
		}
		break;
	case STATE_SLASH:
		switch (http_tokenizer_next (&t, false))
		{
		case HTTP_T_DELIMITER:
			if (t.delimiter == '/')
				state = STATE_PROTOCOL_VERSION;
			else if (t.delimiter == ',')
				state = STATE_PROTOCOL_NAME;
			else
				goto end;
			break;
		case HTTP_T_WHITESPACE:
			state = STATE_EXPECT_COMMA;
			break;
		case HTTP_T_EOF:
			result = true;
		default:
			goto end;
		}
		break;
	case STATE_PROTOCOL_VERSION:
		switch (http_tokenizer_next (&t, false))
		{
		case HTTP_T_TOKEN:
			proto->version = xstrdup (t.string.str);
			state = STATE_EXPECT_COMMA;
			break;
		default:
			goto end;
		}
		break;
	case STATE_EXPECT_COMMA:
		switch (http_tokenizer_next (&t, false))
		{
		case HTTP_T_DELIMITER:
			if (t.delimiter != ',')
				goto end;
			state = STATE_PROTOCOL_NAME;
		case HTTP_T_WHITESPACE:
			break;
		case HTTP_T_EOF:
			result = true;
		default:
			goto end;
		}
	}

end:
	if (result)
		*out = list;
	else
		LIST_FOR_EACH (struct http_protocol, iter, list)
			http_protocol_destroy (iter);

	http_tokenizer_free (&t);
	return result;
}

// --- libev helpers -----------------------------------------------------------

static bool
flush_queue (write_queue_t *queue, ev_io *watcher)
{
	struct iovec vec[queue->len], *vec_iter = vec;
	for (write_req_t *iter = queue->head; iter; iter = iter->next)
		*vec_iter++ = iter->data;

	ssize_t written;
again:
	written = writev (watcher->fd, vec, N_ELEMENTS (vec));
	if (written < 0)
	{
		if (errno == EAGAIN)
			goto skip;
		if (errno == EINTR)
			goto again;
		return false;
	}

	write_queue_processed (queue, written);

skip:
	if (write_queue_is_empty (queue))
		ev_io_stop (EV_DEFAULT_ watcher);
	else
		ev_io_start (EV_DEFAULT_ watcher);
	return true;
}

// --- Logging -----------------------------------------------------------------

static void
log_message_syslog (void *user_data, const char *quote, const char *fmt,
	va_list ap)
{
	int prio = (int) (intptr_t) user_data;

	va_list va;
	va_copy (va, ap);
	int size = vsnprintf (NULL, 0, fmt, va);
	va_end (va);
	if (size < 0)
		return;

	char buf[size + 1];
	if (vsnprintf (buf, sizeof buf, fmt, ap) >= 0)
		syslog (prio, "%s%s", quote, buf);
}

// --- FastCGI -----------------------------------------------------------------

// Constants from the FastCGI specification document

#define FCGI_HEADER_LEN       8

#define FCGI_VERSION_1        1
#define FCGI_NULL_REQUEST_ID  0
#define FCGI_KEEP_CONN        1

enum fcgi_type
{
	FCGI_BEGIN_REQUEST     =  1,
	FCGI_ABORT_REQUEST     =  2,
	FCGI_END_REQUEST       =  3,
	FCGI_PARAMS            =  4,
	FCGI_STDIN             =  5,
	FCGI_STDOUT            =  6,
	FCGI_STDERR            =  7,
	FCGI_DATA              =  8,
	FCGI_GET_VALUES        =  9,
	FCGI_GET_VALUES_RESULT = 10,
	FCGI_UNKNOWN_TYPE      = 11,
	FCGI_MAXTYPE           = FCGI_UNKNOWN_TYPE
};

enum fcgi_role
{
	FCGI_RESPONDER         =  1,
	FCGI_AUTHORIZER        =  2,
	FCGI_FILTER            =  3
};

enum fcgi_protocol_status
{
	FCGI_REQUEST_COMPLETE  =  0,
	FCGI_CANT_MPX_CONN     =  1,
	FCGI_OVERLOADED        =  2,
	FCGI_UNKNOWN_ROLE      =  3
};

#define FCGI_MAX_CONNS   "FCGI_MAX_CONNS"
#define FCGI_MAX_REQS    "FCGI_MAX_REQS"
#define FCGI_MPXS_CONNS  "FCGI_MPXS_CONNS"

// - - Message stream parser - - - - - - - - - - - - - - - - - - - - - - - - - -

struct fcgi_parser;

typedef void (*fcgi_message_fn)
	(const struct fcgi_parser *parser, void *user_data);

enum fcgi_parser_state
{
	FCGI_READING_HEADER,                ///< Reading the fixed header portion
	FCGI_READING_CONTENT,               ///< Reading the message content
	FCGI_READING_PADDING                ///< Reading the padding
};

struct fcgi_parser
{
	enum fcgi_parser_state state;       ///< Parsing state
	struct str input;                   ///< Input buffer

	// The next block of fields is considered public:

	uint8_t version;                    ///< FastCGI protocol version
	uint8_t type;                       ///< FastCGI record type
	uint16_t request_id;                ///< FastCGI request ID
	struct str content;                 ///< Message data

	uint16_t content_length;            ///< Message content length
	uint8_t padding_length;             ///< Message padding length

	fcgi_message_fn on_message;         ///< Callback on message
	void *user_data;                    ///< User data
};

static void
fcgi_parser_init (struct fcgi_parser *self)
{
	memset (self, 0, sizeof *self);
	str_init (&self->input);
	str_init (&self->content);
}

static void
fcgi_parser_free (struct fcgi_parser *self)
{
	str_free (&self->input);
	str_free (&self->content);
}

static void
fcgi_parser_unpack_header (struct fcgi_parser *self)
{
	struct msg_unpacker unpacker;
	msg_unpacker_init (&unpacker, self->input.str, self->input.len);

	bool success = true;
	uint8_t reserved;
	success &= msg_unpacker_u8  (&unpacker, &self->version);
	success &= msg_unpacker_u8  (&unpacker, &self->type);
	success &= msg_unpacker_u16 (&unpacker, &self->request_id);
	success &= msg_unpacker_u16 (&unpacker, &self->content_length);
	success &= msg_unpacker_u8  (&unpacker, &self->padding_length);
	success &= msg_unpacker_u8  (&unpacker, &reserved);
	hard_assert (success);

	str_remove_slice (&self->input, 0, unpacker.offset);
}

static void
fcgi_parser_push (struct fcgi_parser *self, const void *data, size_t len)
{
	// This could be made considerably faster for high-throughput applications
	// if we use a circular buffer instead of constantly calling memmove()
	str_append_data (&self->input, data, len);

	while (true)
	switch (self->state)
	{
	case FCGI_READING_HEADER:
		if (self->input.len < FCGI_HEADER_LEN)
			return;

		fcgi_parser_unpack_header (self);
		self->state = FCGI_READING_CONTENT;
		break;
	case FCGI_READING_CONTENT:
		if (self->input.len < self->content_length)
			return;

		// Move an appropriate part of the input buffer to the content buffer
		str_reset (&self->content);
		str_append_data (&self->content, self->input.str, self->content_length);
		str_remove_slice (&self->input, 0, self->content_length);
		self->state = FCGI_READING_PADDING;
		break;
	case FCGI_READING_PADDING:
		if (self->input.len < self->padding_length)
			return;

		// Call the callback to further process the message
		self->on_message (self, self->user_data);

		// Remove the padding from the input buffer
		str_remove_slice (&self->input, 0, self->padding_length);
		self->state = FCGI_READING_HEADER;
		break;
	}
}

// - - Name-value pair parser  - - - - - - - - - - - - - - - - - - - - - - - - -

enum fcgi_nv_parser_state
{
	FCGI_NV_PARSER_NAME_LEN,            ///< The first name length octet
	FCGI_NV_PARSER_NAME_LEN_FULL,       ///< Remaining name length octets
	FCGI_NV_PARSER_VALUE_LEN,           ///< The first value length octet
	FCGI_NV_PARSER_VALUE_LEN_FULL,      ///< Remaining value length octets
	FCGI_NV_PARSER_NAME,                ///< Reading the name
	FCGI_NV_PARSER_VALUE                ///< Reading the value
};

struct fcgi_nv_parser
{
	struct str_map *output;             ///< Where the pairs will be stored

	enum fcgi_nv_parser_state state;    ///< Parsing state
	struct str input;                   ///< Input buffer

	uint32_t name_len;                  ///< Length of the name
	uint32_t value_len;                 ///< Length of the value

	char *name;                         ///< The current name, 0-terminated
	char *value;                        ///< The current value, 0-terminated
};

static void
fcgi_nv_parser_init (struct fcgi_nv_parser *self)
{
	memset (self, 0, sizeof *self);
	str_init (&self->input);
}

static void
fcgi_nv_parser_free (struct fcgi_nv_parser *self)
{
	str_free (&self->input);
	free (self->name);
	free (self->value);
}

static void
fcgi_nv_parser_push (struct fcgi_nv_parser *self, const void *data, size_t len)
{
	// This could be optimized significantly; I'm not even trying
	str_append_data (&self->input, data, len);

	while (true)
	{
		struct msg_unpacker unpacker;
		msg_unpacker_init (&unpacker, self->input.str, self->input.len);

	switch (self->state)
	{
		uint8_t len;
		uint32_t len_full;

	case FCGI_NV_PARSER_NAME_LEN:
		if (!msg_unpacker_u8 (&unpacker, &len))
			return;

		if (len >> 7)
			self->state = FCGI_NV_PARSER_NAME_LEN_FULL;
		else
		{
			self->name_len = len;
			str_remove_slice (&self->input, 0, unpacker.offset);
			self->state = FCGI_NV_PARSER_VALUE_LEN;
		}
		break;
	case FCGI_NV_PARSER_NAME_LEN_FULL:
		if (!msg_unpacker_u32 (&unpacker, &len_full))
			return;

		self->name_len = len_full & ~(1U << 31);
		str_remove_slice (&self->input, 0, unpacker.offset);
		self->state = FCGI_NV_PARSER_VALUE_LEN;
		break;
	case FCGI_NV_PARSER_VALUE_LEN:
		if (!msg_unpacker_u8 (&unpacker, &len))
			return;

		if (len >> 7)
			self->state = FCGI_NV_PARSER_VALUE_LEN_FULL;
		else
		{
			self->value_len = len;
			str_remove_slice (&self->input, 0, unpacker.offset);
			self->state = FCGI_NV_PARSER_NAME;
		}
		break;
	case FCGI_NV_PARSER_VALUE_LEN_FULL:
		if (!msg_unpacker_u32 (&unpacker, &len_full))
			return;

		self->value_len = len_full & ~(1U << 31);
		str_remove_slice (&self->input, 0, unpacker.offset);
		self->state = FCGI_NV_PARSER_NAME;
		break;
	case FCGI_NV_PARSER_NAME:
		if (self->input.len < self->name_len)
			return;

		self->name = xmalloc (self->name_len + 1);
		self->name[self->name_len] = '\0';
		memcpy (self->name, self->input.str, self->name_len);
		str_remove_slice (&self->input, 0, self->name_len);
		self->state = FCGI_NV_PARSER_VALUE;
		break;
	case FCGI_NV_PARSER_VALUE:
		if (self->input.len < self->value_len)
			return;

		self->value = xmalloc (self->value_len + 1);
		self->value[self->value_len] = '\0';
		memcpy (self->value, self->input.str, self->value_len);
		str_remove_slice (&self->input, 0, self->value_len);
		self->state = FCGI_NV_PARSER_NAME_LEN;

		// The map takes ownership of the value
		str_map_set (self->output, self->name, self->value);
		free (self->name);

		self->name  = NULL;
		self->value = NULL;
		break;
	}
	}
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static void
fcgi_nv_convert_len (size_t len, struct str *output)
{
	if (len < 0x80)
		str_pack_u8 (output, len);
	else
	{
		len |= (uint32_t) 1 << 31;
		str_pack_u32 (output, len);
	}
}

static void
fcgi_nv_convert (struct str_map *map, struct str *output)
{
	struct str_map_iter iter;
	str_map_iter_init (&iter, map);
	while (str_map_iter_next (&iter))
	{
		const char *name  = iter.link->key;
		const char *value = iter.link->data;
		size_t name_len   = iter.link->key_length;
		size_t value_len  = strlen (value);

		fcgi_nv_convert_len (name_len,  output);
		fcgi_nv_convert_len (value_len, output);
		str_append_data (output, name,  name_len);
		str_append_data (output, value, value_len);
	}
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

enum fcgi_request_state
{
	FCGI_REQUEST_PARAMS,                ///< Reading headers
	FCGI_REQUEST_STDIN                  ///< Reading input
};

struct fcgi_request
{
	struct fcgi_muxer *muxer;           ///< The parent muxer
	uint16_t request_id;                ///< The ID of this request
	uint8_t flags;                      ///< Request flags

	enum fcgi_request_state state;      ///< Parsing state
	struct str_map headers;             ///< Headers
	struct fcgi_nv_parser hdr_parser;   ///< Header parser

	struct str output_buffer;           ///< Output buffer

	void *handler_data;                 ///< Handler data
};

struct fcgi_muxer
{
	struct fcgi_parser parser;          ///< FastCGI message parser

	// TODO: bool quitting; that causes us to reject all requests?

	/// Requests assigned to request IDs
	// TODO: allocate this dynamically
	struct fcgi_request *requests[1 << 16];

	void (*write_cb) (void *user_data, const void *data, size_t len);
	void (*close_cb) (void *user_data);

	void *(*request_start_cb) (void *user_data, struct fcgi_request *request);
	void (*request_push_cb) (void *handler_data, const void *data, size_t len);
	void (*request_destroy_cb) (void *handler_data);

	void *user_data;                    ///< User data for callbacks
};

static void
fcgi_muxer_send (struct fcgi_muxer *self,
	enum fcgi_type type, uint16_t request_id, const void *data, size_t len)
{
	hard_assert (len <= UINT16_MAX);

	struct str message;
	str_init (&message);

	str_pack_u8  (&message, FCGI_VERSION_1);
	str_pack_u8  (&message, type);
	str_pack_u16 (&message, request_id);
	str_pack_u16 (&message, len);  // content length
	str_pack_u8  (&message, 0);    // padding length

	str_append_data (&message, data, len);

	// XXX: we should probably have another write_cb that assumes ownership
	self->write_cb (self->user_data, message.str, message.len);
	str_free (&message);
}

static void
fcgi_muxer_send_end_request (struct fcgi_muxer *self, uint16_t request_id,
	uint32_t app_status, enum fcgi_protocol_status protocol_status)
{
	uint8_t content[8] = { app_status >> 24, app_status >> 16,
		app_status << 8, app_status, protocol_status };
	fcgi_muxer_send (self, FCGI_END_REQUEST, request_id,
		content, sizeof content);
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static void
fcgi_request_init (struct fcgi_request *self)
{
	memset (self, 0, sizeof *self);

	str_map_init (&self->headers);
	self->headers.free = free;

	fcgi_nv_parser_init (&self->hdr_parser);
	self->hdr_parser.output = &self->headers;
}

static void
fcgi_request_free (struct fcgi_request *self)
{
	str_map_free (&self->headers);
	fcgi_nv_parser_free (&self->hdr_parser);
}

static void
fcgi_request_push_params
	(struct fcgi_request *self, const void *data, size_t len)
{
	if (self->state != FCGI_REQUEST_PARAMS)
	{
		// TODO: probably reject the request
		return;
	}

	if (len)
		fcgi_nv_parser_push (&self->hdr_parser, data, len);
	else
	{
		// TODO: probably check the state of the header parser
		// TODO: request_start() can return false, end the request here?
		self->handler_data = self->muxer->request_start_cb
			(self->muxer->user_data, self);
		self->state = FCGI_REQUEST_STDIN;
	}
}

static void
fcgi_request_push_stdin
	(struct fcgi_request *self, const void *data, size_t len)
{
	if (self->state != FCGI_REQUEST_STDIN)
	{
		// TODO: probably reject the request
		return;
	}

	self->muxer->request_push_cb (self->handler_data, data, len);
}

static void
fcgi_request_flush (struct fcgi_request *self)
{
	if (!self->output_buffer.len)
		return;

	fcgi_muxer_send (self->muxer, FCGI_STDOUT, self->request_id,
		self->output_buffer.str, self->output_buffer.len);
	str_reset (&self->output_buffer);
}

static void
fcgi_request_write (struct fcgi_request *self, const void *data, size_t len)
{
	// We're buffering the output and splitting it into messages
	bool need_flush = true;
	while (len)
	{
		size_t to_write = UINT16_MAX - self->output_buffer.len;
		if (to_write > len)
		{
			to_write = len;
			need_flush = false;
		}

		str_append_data (&self->output_buffer, data, to_write);
		data = (uint8_t *) data + to_write;
		len -= to_write;

		if (need_flush)
			fcgi_request_flush (self);
	}
}

static void
fcgi_request_finish (struct fcgi_request *self)
{
	// TODO: flush(), end_request(), delete self, muxer->request_destroy_cb()?
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

typedef void (*fcgi_muxer_handler_fn)
	(struct fcgi_muxer *, const struct fcgi_parser *);

static void
fcgi_muxer_on_get_values
	(struct fcgi_muxer *self, const struct fcgi_parser *parser)
{
	struct str_map values;    str_map_init (&values);    values.free   = free;
	struct str_map response;  str_map_init (&response);  response.free = free;

	struct fcgi_nv_parser nv_parser;
	fcgi_nv_parser_init (&nv_parser);
	nv_parser.output = &values;

	fcgi_nv_parser_push (&nv_parser, parser->content.str, parser->content.len);

	struct str_map_iter iter;
	str_map_iter_init (&iter, &values);
	while (str_map_iter_next (&iter))
	{
		const char *key = iter.link->key;

		// TODO: if (!strcmp (key, FCGI_MAX_CONNS))
		// TODO: if (!strcmp (key, FCGI_MAX_REQS))

		if (!strcmp (key, FCGI_MPXS_CONNS))
			str_map_set (&response, key, xstrdup ("1"));
	}

	struct str content;
	str_init (&content);
	fcgi_nv_convert (&response, &content);
	fcgi_muxer_send (self, FCGI_GET_VALUES_RESULT, parser->request_id,
		content.str, content.len);
	str_free (&content);

	str_map_free (&values);
	str_map_free (&response);
}

static void
fcgi_muxer_on_begin_request
	(struct fcgi_muxer *self, const struct fcgi_parser *parser)
{
	struct msg_unpacker unpacker;
	msg_unpacker_init (&unpacker, parser->content.str, parser->content.len);

	uint16_t role;
	uint8_t flags;
	bool success = true;
	success &= msg_unpacker_u16 (&unpacker, &role);
	success &= msg_unpacker_u8 (&unpacker, &flags);
	// Ignoring 5 reserved bytes

	if (!success)
	{
		print_debug ("FastCGI: ignoring invalid %s message",
			STRINGIFY (FCGI_BEGIN_REQUEST));
		return;
	}

	// We can only act as a responder, reject everything else up front
	if (role != FCGI_RESPONDER)
	{
		fcgi_muxer_send_end_request (self,
			parser->request_id, 0, FCGI_UNKNOWN_ROLE);
		return;
	}

	struct fcgi_request *request = self->requests[parser->request_id];
	if (request)
	{
		// TODO: fail
		return;
	}

	request = xcalloc (1, sizeof *request);
	fcgi_request_init (request);
	request->muxer = self;
	request->request_id = parser->request_id;
	request->flags = flags;

	self->requests[parser->request_id] = request;
}

static void
fcgi_muxer_on_abort_request
	(struct fcgi_muxer *self, const struct fcgi_parser *parser)
{
	struct fcgi_request *request = self->requests[parser->request_id];
	if (!request)
	{
		print_debug ("FastCGI: received %s for an unknown request",
			STRINGIFY (FCGI_ABORT_REQUEST));
		return;
	}

	// TODO: abort the request: let it somehow produce FCGI_END_REQUEST
}

static void
fcgi_muxer_on_params (struct fcgi_muxer *self, const struct fcgi_parser *parser)
{
	struct fcgi_request *request = self->requests[parser->request_id];
	if (!request)
	{
		print_debug ("FastCGI: received %s for an unknown request",
			STRINGIFY (FCGI_PARAMS));
		return;
	}

	fcgi_request_push_params (request,
		parser->content.str, parser->content.len);
}

static void
fcgi_muxer_on_stdin (struct fcgi_muxer *self, const struct fcgi_parser *parser)
{
	struct fcgi_request *request = self->requests[parser->request_id];
	if (!request)
	{
		print_debug ("FastCGI: received %s for an unknown request",
			STRINGIFY (FCGI_STDIN));
		return;
	}

	fcgi_request_push_stdin (request,
		parser->content.str, parser->content.len);
}

static void
fcgi_muxer_on_message (const struct fcgi_parser *parser, void *user_data)
{
	struct fcgi_muxer *self = user_data;

	if (parser->version != FCGI_VERSION_1)
	{
		print_debug ("FastCGI: unsupported version %d", parser->version);
		// TODO: also return false to stop processing on protocol error?
		return;
	}

	static const fcgi_muxer_handler_fn handlers[] =
	{
		[FCGI_GET_VALUES]    = fcgi_muxer_on_get_values,
		[FCGI_BEGIN_REQUEST] = fcgi_muxer_on_begin_request,
		[FCGI_ABORT_REQUEST] = fcgi_muxer_on_abort_request,
		[FCGI_PARAMS]        = fcgi_muxer_on_params,
		[FCGI_STDIN]         = fcgi_muxer_on_stdin,
	};

	fcgi_muxer_handler_fn handler;
	if (parser->type >= N_ELEMENTS (handlers)
	 || !(handler = handlers[parser->type]))
	{
		uint8_t content[8] = { parser->type };
		fcgi_muxer_send (self, FCGI_UNKNOWN_TYPE, parser->request_id,
			content, sizeof content);
		return;
	}

	handler (self, parser);
}

static void
fcgi_muxer_init (struct fcgi_muxer *self)
{
	fcgi_parser_init (&self->parser);
	self->parser.on_message = fcgi_muxer_on_message;
	self->parser.user_data = self;
}

static void
fcgi_muxer_free (struct fcgi_muxer *self)
{
	fcgi_parser_free (&self->parser);
}

static void
fcgi_muxer_push (struct fcgi_muxer *self, const void *data, size_t len)
{
	fcgi_parser_push (&self->parser, data, len);
}

// --- SCGI --------------------------------------------------------------------

enum scgi_parser_state
{
	SCGI_READING_NETSTRING_LENGTH,      ///< The length of the header netstring
	SCGI_READING_NAME,                  ///< Header name
	SCGI_READING_VALUE,                 ///< Header value
	SCGI_READING_CONTENT                ///< Incoming data
};

struct scgi_parser
{
	enum scgi_parser_state state;       ///< Parsing state
	struct str input;                   ///< Input buffer

	struct str_map headers;             ///< Headers parsed

	size_t headers_len;                 ///< Length of the netstring contents
	struct str name;                    ///< Header name so far
	struct str value;                   ///< Header value so far

	/// Finished parsing request headers.
	/// Return false to abort further processing of input.
	bool (*on_headers_read) (void *user_data);

	/// Content available; len == 0 means end of file.
	/// Return false to abort further processing of input.
	bool (*on_content) (void *user_data, const void *data, size_t len);

	void *user_data;                    ///< User data passed to callbacks
};

static void
scgi_parser_init (struct scgi_parser *self)
{
	memset (self, 0, sizeof *self);

	str_init (&self->input);
	str_map_init (&self->headers);
	self->headers.free = free;
	str_init (&self->name);
	str_init (&self->value);
}

static void
scgi_parser_free (struct scgi_parser *self)
{
	str_free (&self->input);
	str_map_free (&self->headers);
	str_free (&self->name);
	str_free (&self->value);
}

static bool
scgi_parser_push (struct scgi_parser *self,
	const void *data, size_t len, struct error **e)
{
	if (!len)
	{
		if (self->state != SCGI_READING_CONTENT)
		{
			error_set (e, "premature EOF");
			return false;
		}

		// Indicate end of file
		return self->on_content (self->user_data, NULL, 0);
	}

	// Notice that this madness is significantly harder to parse than FastCGI;
	// this procedure could also be optimized significantly
	str_append_data (&self->input, data, len);

	bool keep_running = true;
	while (keep_running)
	switch (self->state)
	{
	case SCGI_READING_NETSTRING_LENGTH:
	{
		if (self->input.len < 1)
			return true;

		char digit = *self->input.str;
		// XXX: this allows for omitting the netstring length altogether
		if (digit == ':')
		{
			self->state = SCGI_READING_NAME;
			break;
		}

		if (digit < '0' || digit >= '9')
		{
			error_set (e, "invalid header netstring");
			return false;
		}

		size_t new_len = self->headers_len * 10 + (digit - '0');
		if (new_len < self->headers_len)
		{
			error_set (e, "header netstring is too long");
			return false;
		}
		self->headers_len = new_len;
		str_remove_slice (&self->input, 0, 1);
		break;
	}
	case SCGI_READING_NAME:
	{
		if (self->input.len < 1)
			return true;

		char c = *self->input.str;
		if (!self->headers_len)
		{
			// The netstring is ending but we haven't finished parsing it,
			// or the netstring doesn't end with a comma
			if (self->name.len || c != ',')
			{
				error_set (e, "invalid header netstring");
				return false;
			}
			self->state = SCGI_READING_CONTENT;
			keep_running = self->on_headers_read (self->user_data);
		}
		else if (c != '\0')
			str_append_c (&self->name, c);
		else
			self->state = SCGI_READING_VALUE;

		str_remove_slice (&self->input, 0, 1);
		break;
	}
	case SCGI_READING_VALUE:
	{
		if (self->input.len < 1)
			return true;

		char c = *self->input.str;
		if (!self->headers_len)
		{
			// The netstring is ending but we haven't finished parsing it
			error_set (e, "invalid header netstring");
			return false;
		}
		else if (c != '\0')
			str_append_c (&self->value, c);
		else
		{
			// We've got a name-value pair, let's put it in the map
			str_map_set (&self->headers,
				self->name.str, str_steal (&self->value));

			str_reset (&self->name);
			str_init (&self->value);

			self->state = SCGI_READING_NAME;
		}

		str_remove_slice (&self->input, 0, 1);
		break;
	}
	case SCGI_READING_CONTENT:
		keep_running = self->on_content
			(self->user_data, self->input.str, self->input.len);
		str_remove_slice (&self->input, 0, self->input.len);
		return keep_running;
	}
	return false;
}

// --- WebSockets --------------------------------------------------------------

#define WS_GUID "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

#define SEC_WS_KEY         "Sec-WebSocket-Key"
#define SEC_WS_ACCEPT      "Sec-WebSocket-Accept"
#define SEC_WS_PROTOCOL    "Sec-WebSocket-Protocol"
#define SEC_WS_EXTENSIONS  "Sec-WebSocket-Extensions"
#define SEC_WS_VERSION     "Sec-WebSocket-Version"

#define WS_MAX_CONTROL_PAYLOAD_LEN  125

static char *
ws_encode_response_key (const char *key)
{
	char *response_key = xstrdup_printf ("%s" WS_GUID, key);
	unsigned char hash[SHA_DIGEST_LENGTH];
	SHA1 ((unsigned char *) response_key, strlen (response_key), hash);
	free (response_key);

	struct str base64;
	str_init (&base64);
	base64_encode (hash, sizeof hash, &base64);
	return str_steal (&base64);
}

enum ws_status
{
	// Named according to the meaning specified in RFC 6455, section 11.2

	WS_STATUS_NORMAL_CLOSURE         = 1000,
	WS_STATUS_GOING_AWAY             = 1001,
	WS_STATUS_PROTOCOL_ERROR         = 1002,
	WS_STATUS_UNSUPPORTED_DATA       = 1003,
	WS_STATUS_INVALID_PAYLOAD_DATA   = 1007,
	WS_STATUS_POLICY_VIOLATION       = 1008,
	WS_STATUS_MESSAGE_TOO_BIG        = 1009,
	WS_STATUS_MANDATORY_EXTENSION    = 1010,
	WS_STATUS_INTERNAL_SERVER_ERROR  = 1011,

	// Reserved for internal usage
	WS_STATUS_NO_STATUS_RECEIVED     = 1005,
	WS_STATUS_ABNORMAL_CLOSURE       = 1006,
	WS_STATUS_TLS_HANDSHAKE          = 1015
};

// - - Frame parser  - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

enum ws_parser_state
{
	WS_PARSER_FIXED,                    ///< Parsing fixed length part
	WS_PARSER_PAYLOAD_LEN_16,           ///< Parsing extended payload length
	WS_PARSER_PAYLOAD_LEN_64,           ///< Parsing extended payload length
	WS_PARSER_MASK,                     ///< Parsing masking-key
	WS_PARSER_PAYLOAD                   ///< Parsing payload
};

enum ws_opcode
{
	// Non-control
	WS_OPCODE_CONT   =  0,
	WS_OPCODE_TEXT   =  1,
	WS_OPCODE_BINARY =  2,

	// Control
	WS_OPCODE_CLOSE  =  8,
	WS_OPCODE_PING   =  9,
	WS_OPCODE_PONG   = 10
};

static bool
ws_is_control_frame (int opcode)
{
	return opcode >= WS_OPCODE_CLOSE;
}

struct ws_parser
{
	struct str input;                   ///< External input buffer
	enum ws_parser_state state;         ///< Parsing state

	unsigned is_fin     : 1;            ///< Final frame of a message?
	unsigned is_masked  : 1;            ///< Is the frame masked?
	unsigned reserved_1 : 1;            ///< Reserved
	unsigned reserved_2 : 1;            ///< Reserved
	unsigned reserved_3 : 1;            ///< Reserved
	enum ws_opcode opcode;              ///< Opcode
	uint32_t mask;                      ///< Frame mask
	uint64_t payload_len;               ///< Payload length

	bool (*on_frame_header) (void *user_data, const struct ws_parser *self);

	/// Callback for when a message is successfully parsed.
	/// The actual payload is stored in "input", of length "payload_len".
	bool (*on_frame) (void *user_data, const struct ws_parser *self);

	void *user_data;                    ///< User data for callbacks
};

static void
ws_parser_init (struct ws_parser *self)
{
	memset (self, 0, sizeof *self);
	str_init (&self->input);
}

static void
ws_parser_free (struct ws_parser *self)
{
	str_free (&self->input);
}

static void
ws_parser_unmask (struct ws_parser *self)
{
	// This could be made faster.  For example by reading the mask in
	// native byte ordering and applying it directly here.

	uint64_t end = self->payload_len & ~(uint64_t) 3;
	for (uint64_t i = 0; i < end; i += 4)
	{
		self->input.str[i + 3] ^=  self->mask        & 0xFF;
		self->input.str[i + 2] ^= (self->mask >>  8) & 0xFF;
		self->input.str[i + 1] ^= (self->mask >> 16) & 0xFF;
		self->input.str[i    ] ^= (self->mask >> 24) & 0xFF;
	}

	switch (self->payload_len - end)
	{
	case 3:
		self->input.str[end + 2] ^= (self->mask >>  8) & 0xFF;
	case 2:
		self->input.str[end + 1] ^= (self->mask >> 16) & 0xFF;
	case 1:
		self->input.str[end    ] ^= (self->mask >> 24) & 0xFF;
		break;
	}
}

static bool
ws_parser_push (struct ws_parser *self, const void *data, size_t len)
{
	bool success = false;
	str_append_data (&self->input, data, len);

	struct msg_unpacker unpacker;
	msg_unpacker_init (&unpacker, self->input.str, self->input.len);

	while (true)
	switch (self->state)
	{
		uint8_t u8;
		uint16_t u16;

	case WS_PARSER_FIXED:
		if (unpacker.len - unpacker.offset < 2)
			goto need_data;

		(void) msg_unpacker_u8 (&unpacker, &u8);
		self->is_fin      = (u8 >> 7) &   1;
		self->reserved_1  = (u8 >> 6) &   1;
		self->reserved_2  = (u8 >> 5) &   1;
		self->reserved_3  = (u8 >> 4) &   1;
		self->opcode      =  u8       &  15;

		(void) msg_unpacker_u8 (&unpacker, &u8);
		self->is_masked   = (u8 >> 7) &   1;
		self->payload_len =  u8       & 127;

		if (self->payload_len == 127)
			self->state = WS_PARSER_PAYLOAD_LEN_64;
		else if (self->payload_len == 126)
			self->state = WS_PARSER_PAYLOAD_LEN_16;
		else
			self->state = WS_PARSER_MASK;
		break;

	case WS_PARSER_PAYLOAD_LEN_16:
		if (!msg_unpacker_u16 (&unpacker, &u16))
			goto need_data;
		self->payload_len = u16;

		self->state = WS_PARSER_MASK;
		break;

	case WS_PARSER_PAYLOAD_LEN_64:
		if (!msg_unpacker_u64 (&unpacker, &self->payload_len))
			goto need_data;

		self->state = WS_PARSER_MASK;
		break;

	case WS_PARSER_MASK:
		if (!self->is_masked)
			goto end_of_header;
		if (!msg_unpacker_u32 (&unpacker, &self->mask))
			goto need_data;

	end_of_header:
		self->state = WS_PARSER_PAYLOAD;
		if (!self->on_frame_header (self->user_data, self))
			goto fail;
		break;

	case WS_PARSER_PAYLOAD:
		// Move the buffer so that payload data is at the front
		str_remove_slice (&self->input, 0, unpacker.offset);

		// And continue unpacking frames past the payload
		msg_unpacker_init (&unpacker, self->input.str, self->input.len);
		unpacker.offset = self->payload_len;

		if (self->input.len < self->payload_len)
			goto need_data;
		if (self->is_masked)
			ws_parser_unmask (self);
		if (!self->on_frame (self->user_data, self))
			goto fail;

		self->state = WS_PARSER_FIXED;
		break;
	}

need_data:
	success = true;
fail:
	str_remove_slice (&self->input, 0, unpacker.offset);
	return success;
}

// - - Server handler  - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

// WebSockets aren't CGI-compatible, therefore we must handle the initial HTTP
// handshake ourselves.  Luckily it's not too much of a bother with http-parser.
// Typically there will be a normal HTTP server in front of us, proxying the
// requests based on the URI.

enum ws_handler_state
{
	WS_HANDLER_CONNECTING,              ///< Parsing HTTP
	WS_HANDLER_OPEN,                    ///< Parsing WebSockets frames
	WS_HANDLER_CLOSING,                 ///< Closing the connection
	WS_HANDLER_CLOSED                   ///< Dead
};

struct ws_handler
{
	enum ws_handler_state state;        ///< State

	http_parser hp;                     ///< HTTP parser
	bool parsing_header_value;          ///< Parsing header value or field?
	struct str field;                   ///< Field part buffer
	struct str value;                   ///< Value part buffer
	struct str_map headers;             ///< HTTP Headers
	struct str url;                     ///< Request URL

	struct ws_parser parser;            ///< Protocol frame parser
	bool expecting_continuation;        ///< For non-control traffic

	enum ws_opcode message_opcode;      ///< Opcode for the current message
	struct str message_data;            ///< Concatenated message data

	unsigned ping_interval;             ///< Ping interval in seconds
	uint64_t max_payload_len;           ///< Maximum length of any message

	// TODO: handshake_timeout
	// TODO: a close timer

	// TODO: a ping timer (when no pong is received by the second time the
	//   timer triggers, it is a ping timeout)
	ev_timer ping_timer;                ///< Ping timer
	bool received_pong;                 ///< Received PONG since the last PING

	// TODO: void (*on_handshake) (protocols) that will allow the user
	//   to choose any sub-protocol, if the client has provided any.

	// TODO: "on_connected" after the handshake has finished?

	/// Called upon reception of a single full message
	bool (*on_message) (void *user_data,
		enum ws_opcode type, const void *data, size_t len);

	/// The connection has been closed.  @a close_code may, or may not, be one
	/// of enum ws_status.  The @a reason is never NULL.
	// TODO; also note that ideally, the handler should (be able to) first
	//   receive a notification about the connection being closed because of
	//   an error (recv()) returns -1, and call on_close() in reaction.
	//   Actually, calling push() could work pretty fine for this.
	void (*on_close) (void *user_data, int close_code, const char *reason);

	/// Write a chunk of data to the stream
	void (*write_cb) (void *user_data, const void *data, size_t len);

	// TODO: "close_cb"; to be used from a ping timer e.g.

	void *user_data;                    ///< User data for callbacks
};

static void
ws_handler_send_control (struct ws_handler *self,
	enum ws_opcode opcode, const void *data, size_t len)
{
	if (len > WS_MAX_CONTROL_PAYLOAD_LEN)
	{
		print_debug ("truncating output control frame payload"
			" from %zu to %zu bytes", len, (size_t) WS_MAX_CONTROL_PAYLOAD_LEN);
		len = WS_MAX_CONTROL_PAYLOAD_LEN;
	}

	uint8_t header[2] = { 0x80 | (opcode & 0x0F), len };
	self->write_cb (self->user_data, header, sizeof header);
	self->write_cb (self->user_data, data, len);
}

static void
ws_handler_fail (struct ws_handler *self, enum ws_status reason)
{
	uint8_t payload[2] = { reason << 8, reason };
	ws_handler_send_control (self, WS_OPCODE_CLOSE, payload, sizeof payload);

	// TODO: set the close timer, ignore all further incoming input (either set
	//   some flag for the case that we're in the middle of ws_handler_push(),
	//   and/or add a mechanism to stop the caller from polling the socket for
	//   reads).
	// TODO: set the state to FAILED (not CLOSED as that means the TCP
	//   connection is closed) and wait until all is sent?
	// TODO: make sure we don't send pings after the close
}

// TODO: ws_handler_close() that behaves like ws_handler_fail() but doesn't
//   ignore frames up to a corresponding close from the client.
//   Read the RFC once again to see if we can really process the frames.

// TODO: add support for fragmented responses
static void
ws_handler_send (struct ws_handler *self,
	enum ws_opcode opcode, const void *data, size_t len)
{
	// TODO: make sure (just assert?) we're in the OPEN state

	struct str header;
	str_init (&header);
	str_pack_u8 (&header, 0x80 | (opcode & 0x0F));

	if (len > UINT16_MAX)
	{
		str_pack_u8 (&header, 127);
		str_pack_u64 (&header, len);
	}
	else if (len > 125)
	{
		str_pack_u8 (&header, 126);
		str_pack_u16 (&header, len);
	}
	else
		str_pack_u8 (&header, len);

	self->write_cb (self->user_data, header.str, header.len);
	self->write_cb (self->user_data, data, len);
	str_free (&header);
}

static bool
ws_handler_on_frame_header (void *user_data, const struct ws_parser *parser)
{
	struct ws_handler *self = user_data;

	// Note that we aren't expected to send any close frame before closing the
	// connection when the frame is unmasked

	if (parser->reserved_1 || parser->reserved_2 || parser->reserved_3
	 || !parser->is_masked  // client -> server payload must be masked
	 || (ws_is_control_frame (parser->opcode) &&
		(!parser->is_fin || parser->payload_len > WS_MAX_CONTROL_PAYLOAD_LEN))
	 || (!ws_is_control_frame (parser->opcode) &&
		(self->expecting_continuation && parser->opcode != WS_OPCODE_CONT))
	 || parser->payload_len >= 0x8000000000000000ULL)
		ws_handler_fail (self, WS_STATUS_PROTOCOL_ERROR);
	else if (parser->payload_len > self->max_payload_len)
		ws_handler_fail (self, WS_STATUS_MESSAGE_TOO_BIG);
	else
		return true;
	return false;
}

static bool
ws_handler_on_control_frame
	(struct ws_handler *self, const struct ws_parser *parser)
{
	switch (parser->opcode)
	{
	case WS_OPCODE_CLOSE:
		// TODO: confirm the close
		// TODO: change the state to CLOSING
		// TODO: call "on_close"
		// NOTE: the reason is an empty string if omitted
		break;
	case WS_OPCODE_PING:
		ws_handler_send_control (self, WS_OPCODE_PONG,
			parser->input.str, parser->payload_len);
		break;
	case WS_OPCODE_PONG:
		// XXX: maybe we should check the payload
		self->received_pong = true;
		break;
	default:
		// Unknown control frame
		ws_handler_fail (self, WS_STATUS_PROTOCOL_ERROR);
		return false;
	}
	return true;
}

static bool
ws_handler_on_frame (void *user_data, const struct ws_parser *parser)
{
	struct ws_handler *self = user_data;
	if (ws_is_control_frame (parser->opcode))
		return ws_handler_on_control_frame (self, parser);

	// TODO: do this rather in "on_frame_header"
	if (self->message_data.len + parser->payload_len > self->max_payload_len)
	{
		ws_handler_fail (self, WS_STATUS_MESSAGE_TOO_BIG);
		return false;
	}

	if (!self->expecting_continuation)
		self->message_opcode = parser->opcode;

	str_append_data (&self->message_data,
		parser->input.str, parser->payload_len);
	self->expecting_continuation = !parser->is_fin;

	if (!parser->is_fin)
		return true;

	if (self->message_opcode == WS_OPCODE_TEXT
	 && !utf8_validate (self->parser.input.str, self->parser.input.len))
	{
		ws_handler_fail (self, WS_STATUS_INVALID_PAYLOAD_DATA);
		return false;
	}

	bool result = self->on_message (self->user_data, self->message_opcode,
		self->parser.input.str, self->parser.payload_len);
	str_reset (&self->message_data);
	return result;
}

static void
ws_handler_on_ping_timer (EV_P_ ev_timer *watcher, int revents)
{
	(void) loop;
	(void) revents;

	struct ws_handler *self = watcher->data;
	if (!self->received_pong)
	{
		// TODO: close/fail the connection?
		return;
	}

	ws_handler_send_control (self, WS_OPCODE_PING, NULL, 0);
}

static void
ws_handler_init (struct ws_handler *self)
{
	memset (self, 0, sizeof *self);

	http_parser_init (&self->hp, HTTP_REQUEST);
	self->hp.data = self;

	str_init (&self->field);
	str_init (&self->value);
	str_map_init (&self->headers);
	self->headers.free = free;
	self->headers.key_xfrm = tolower_ascii_strxfrm;
	str_init (&self->url);

	ws_parser_init (&self->parser);
	self->parser.on_frame_header = ws_handler_on_frame_header;
	self->parser.on_frame = ws_handler_on_frame;

	str_init (&self->message_data);

	self->ping_interval = 60;
	// This is still ridiculously high.  Note that the most significant bit
	// must always be zero, i.e. the protocol maximum is 0x7FFF FFFF FFFF FFFF.
	self->max_payload_len = UINT32_MAX;

	// Just so we can safely stop it
	ev_timer_init (&self->ping_timer, ws_handler_on_ping_timer, 0., 0.);
	self->ping_timer.data = self;
	// So that the first ping timer doesn't timeout the connection
	self->received_pong = true;
}

static void
ws_handler_free (struct ws_handler *self)
{
	str_free (&self->field);
	str_free (&self->value);
	str_map_free (&self->headers);
	str_free (&self->url);
	ws_parser_free (&self->parser);
	str_free (&self->message_data);
	ev_timer_stop (EV_DEFAULT_ &self->ping_timer);
}

static bool
ws_handler_header_field_is_a_list (const char *name)
{
	// This must contain all header fields we use for anything
	static const char *concatenable[] =
		{ SEC_WS_PROTOCOL, SEC_WS_EXTENSIONS, "Connection", "Upgrade" };

	for (size_t i = 0; i < N_ELEMENTS (concatenable); i++)
		if (!strcasecmp_ascii (name, concatenable[i]))
			return true;
	return false;
}

static void
ws_handler_on_header_read (struct ws_handler *self)
{
	// The HTTP parser unfolds values and removes preceding whitespace, but
	// otherwise doesn't touch the values or the following whitespace.

	// RFC 7230 states that trailing whitespace is not part of a field value
	char *value = self->field.str;
	size_t len = self->field.len;
	while (len--)
		if (value[len] == '\t' || value[len] == ' ')
			value[len] = '\0';
		else
			break;
	self->field.len = len;

	const char *field = self->field.str;
	const char *current = str_map_find (&self->headers, field);
	if (ws_handler_header_field_is_a_list (field) && current)
		str_map_set (&self->headers, field,
			xstrdup_printf ("%s, %s", current, self->value.str));
	else
		// If the field cannot be concatenated, just overwrite the last value.
		// Maybe we should issue a warning or something.
		str_map_set (&self->headers, field, xstrdup (self->value.str));
}

static int
ws_handler_on_header_field (http_parser *parser, const char *at, size_t len)
{
	struct ws_handler *self = parser->data;
	if (self->parsing_header_value)
	{
		ws_handler_on_header_read (self);
		str_reset (&self->field);
		str_reset (&self->value);
	}
	str_append_data (&self->field, at, len);
	self->parsing_header_value = false;
	return 0;
}

static int
ws_handler_on_header_value (http_parser *parser, const char *at, size_t len)
{
	struct ws_handler *self = parser->data;
	str_append_data (&self->value, at, len);
	self->parsing_header_value = true;
	return 0;
}

static int
ws_handler_on_headers_complete (http_parser *parser)
{
	// We strictly require a protocol upgrade
	if (!parser->upgrade)
		return 2;

	return 0;
}

static int
ws_handler_on_url (http_parser *parser, const char *at, size_t len)
{
	struct ws_handler *self = parser->data;
	str_append_data (&self->value, at, len);
	return 0;
}

#define HTTP_101_SWITCHING_PROTOCOLS    "101 Switching Protocols"
#define HTTP_400_BAD_REQUEST            "400 Bad Request"
#define HTTP_405_METHOD_NOT_ALLOWED     "405 Method Not Allowed"
#define HTTP_417_EXPECTATION_FAILED     "407 Expectation Failed"
#define HTTP_505_VERSION_NOT_SUPPORTED  "505 HTTP Version Not Supported"

static void
ws_handler_http_responsev (struct ws_handler *self,
	const char *status, char *const *fields)
{
	hard_assert (status != NULL);

	struct str response;
	str_init (&response);
	str_append_printf (&response, "HTTP/1.1 %s\r\n", status);

	while (*fields)
		str_append_printf (&response, "%s\r\n", *fields++);

	str_append (&response, "Server: "
		PROGRAM_NAME "/" PROGRAM_VERSION "\r\n\r\n");
	self->write_cb (self->user_data, response.str, response.len);
	str_free (&response);
}

static void
ws_handler_http_response (struct ws_handler *self, const char *status, ...)
{
	struct str_vector v;
	str_vector_init (&v);

	va_list ap;
	va_start (ap, status);

	const char *s;
	while ((s = va_arg (ap, const char *)))
		str_vector_add (&v, s);

	va_end (ap);

	ws_handler_http_responsev (self, status, v.vector);
	str_vector_free (&v);
}

// TODO: also set the connection to some FAILED state or anything that's neither
//   CONNECTING nor OPEN
#define FAIL_HANDSHAKE(status, ...)                                            \
	BLOCK_START                                                                \
		ws_handler_http_response (self, (status), __VA_ARGS__);                \
		return false;                                                          \
	BLOCK_END

static bool
ws_handler_finish_handshake (struct ws_handler *self)
{
	// XXX: we probably shouldn't use 505 to reject the minor version but w/e
	if (self->hp.http_major != 1 || self->hp.http_minor < 1)
		FAIL_HANDSHAKE (HTTP_505_VERSION_NOT_SUPPORTED, NULL);
	if (self->hp.method != HTTP_GET)
		FAIL_HANDSHAKE (HTTP_405_METHOD_NOT_ALLOWED, "Allow: GET", NULL);

	// Your expectations are way too high
	if (str_map_find (&self->headers, "Expect"))
		FAIL_HANDSHAKE (HTTP_417_EXPECTATION_FAILED, NULL);

	// Reject URLs specifying the schema and host; we're not parsing that
	// TODO: actually do parse this and let our user decide if it matches
	struct http_parser_url url;
	if (http_parser_parse_url (self->url.str, self->url.len, false, &url)
	 || (url.field_set & (1 << UF_SCHEMA | 1 << UF_HOST | 1 << UF_PORT))
	 || !str_map_find (&self->headers, "Host"))
		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);

	const char *connection = str_map_find (&self->headers, "Connection");
	if (!connection || strcasecmp_ascii (connection, "Upgrade"))
		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);

	// Check if we can actually upgrade the protocol to WebSockets
	const char *upgrade = str_map_find (&self->headers, "Upgrade");
	struct http_protocol *offered_upgrades = NULL;
	bool can_upgrade = false;
	if (upgrade && http_parse_upgrade (upgrade, &offered_upgrades))
		// Case-insensitive according to RFC 6455; neither RFC 2616 nor 7230
		// say anything at all about case-sensitivity for this field
		LIST_FOR_EACH (struct http_protocol, iter, offered_upgrades)
		{
			if (!iter->version && !strcasecmp_ascii (iter->name, "websocket"))
				can_upgrade = true;
			http_protocol_destroy (iter);
		}
	if (!can_upgrade)
		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);

	// Okay, we've finally got past basic HTTP/1.1 stuff
	const char *key      = str_map_find (&self->headers, SEC_WS_KEY);
	const char *version  = str_map_find (&self->headers, SEC_WS_VERSION);
	const char *protocol = str_map_find (&self->headers, SEC_WS_PROTOCOL);

	struct str tmp;
	str_init (&tmp);
	bool key_is_valid = base64_decode (key, false, &tmp) && tmp.len == 16;
	str_free (&tmp);
	if (!key_is_valid)
		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);

	if (!version)
		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);
	if (strcmp (version, "13"))
		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, SEC_WS_VERSION ": 13", NULL);

	struct str_vector fields;
	str_vector_init (&fields);

	str_vector_add_args (&fields,
		"Upgrade: websocket",
		"Connection: Upgrade",
		NULL);

	char *response_key = ws_encode_response_key (key);
	str_vector_add_owned (&fields,
		xstrdup_printf (SEC_WS_ACCEPT ": %s", response_key));
	free (response_key);

	// TODO: check and set Sec-Websocket-{Extensions,Protocol}

	ws_handler_http_responsev (self,
		HTTP_101_SWITCHING_PROTOCOLS, fields.vector);

	str_vector_free (&fields);

	// XXX: maybe we should start it earlier so that the handshake can
	//   timeout as well.  ws_handler_connected()?
	//
	//   But it should rather be named "connect_timer"
	ev_timer_start (EV_DEFAULT_ &self->ping_timer);
	return true;
}

static bool
ws_handler_push (struct ws_handler *self, const void *data, size_t len)
{
	if (!len)
	{
		if (self->state == WS_HANDLER_OPEN)
			self->on_close (self->user_data, WS_STATUS_ABNORMAL_CLOSURE, "");
		else
		{
			// TODO: anything to do besides just closing the connection?
		}

		self->state = WS_HANDLER_CLOSED;
		return false;
	}

	if (self->state != WS_HANDLER_CONNECTING)
		return ws_parser_push (&self->parser, data, len);

	// The handshake hasn't been done yet, process HTTP headers
	static const http_parser_settings http_settings =
	{
		.on_header_field     = ws_handler_on_header_field,
		.on_header_value     = ws_handler_on_header_value,
		.on_headers_complete = ws_handler_on_headers_complete,
		.on_url              = ws_handler_on_url,
	};

	size_t n_parsed = http_parser_execute (&self->hp,
		&http_settings, data, len);

	if (self->hp.upgrade)
	{
		// The handshake hasn't been finished, yet there is more data
		//   to be processed after the headers already
		if (len - n_parsed)
			FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);

		if (!ws_handler_finish_handshake (self))
			return false;

		self->state = WS_HANDLER_OPEN;
		return true;
	}

	enum http_errno err = HTTP_PARSER_ERRNO (&self->hp);
	if (n_parsed != len || err != HPE_OK)
	{
		if (err == HPE_CB_headers_complete)
			print_debug ("WS handshake failed: %s", "missing `Upgrade' field");
		else
			print_debug ("WS handshake failed: %s",
				http_errno_description (err));

		FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL);
	}
	return true;
}

// --- Server ------------------------------------------------------------------

static struct config_item g_config_table[] =
{
	{ "bind_host",       NULL,              "Address of the server"          },
	{ "port_fastcgi",    "9000",            "Port to bind for FastCGI"       },
	{ "port_scgi",       NULL,              "Port to bind for SCGI"          },
	{ "port_ws",         NULL,              "Port to bind for WebSockets"    },
	{ "pid_file",        NULL,              "Full path for the PID file"     },
	// XXX: here belongs something like a web SPA that interfaces with us
	{ "static_root",     NULL,              "The root for static content"    },
	{ NULL,              NULL,              NULL                             }
};

struct server_context
{
	ev_signal sigterm_watcher;          ///< Got SIGTERM
	ev_signal sigint_watcher;           ///< Got SIGINT
	bool quitting;                      ///< User requested quitting

	struct listener *listeners;         ///< Listeners
	size_t n_listeners;                 ///< Number of listening sockets

	struct client *clients;             ///< Clients
	unsigned n_clients;                 ///< Current number of connections

	struct request_handler *handlers;   ///< Request handlers
	struct str_map config;              ///< Server configuration
};

static void
server_context_init (struct server_context *self)
{
	memset (self, 0, sizeof *self);

	str_map_init (&self->config);
	load_config_defaults (&self->config, g_config_table);
}

static void close_listeners (struct server_context *self);

static void
server_context_free (struct server_context *self)
{
	// We really shouldn't attempt a quit without closing the clients first
	soft_assert (!self->clients);

	close_listeners (self);
	free (self->listeners);

	str_map_free (&self->config);
}

// --- JSON-RPC ----------------------------------------------------------------

#define JSON_RPC_ERROR_TABLE(XX)                                               \
	XX (-32700, PARSE_ERROR,      "Parse error")                               \
	XX (-32600, INVALID_REQUEST,  "Invalid Request")                           \
	XX (-32601, METHOD_NOT_FOUND, "Method not found")                          \
	XX (-32602, INVALID_PARAMS,   "Invalid params")                            \
	XX (-32603, INTERNAL_ERROR,   "Internal error")

enum json_rpc_error
{
#define XX(code, name, message) JSON_RPC_ERROR_ ## name,
	JSON_RPC_ERROR_TABLE (XX)
#undef XX
	JSON_RPC_ERROR_COUNT
};

static json_t *
json_rpc_error (enum json_rpc_error id, json_t *data)
{
#define XX(code, name, message) { code, message },
	static const struct json_rpc_error
	{
		int code;
		const char *message;
	}
	errors[JSON_RPC_ERROR_COUNT] =
	{
		JSON_RPC_ERROR_TABLE (XX)
	};
#undef XX

	json_t *error = json_object ();
	json_object_set_new (error, "code",    json_integer (errors[id].code));
	json_object_set_new (error, "message", json_string  (errors[id].message));

	if (data)
		json_object_set_new (error, "data", data);

	return error;
}

static json_t *
json_rpc_response (json_t *id, json_t *result, json_t *error)
{
	json_t *x = json_object ();
	json_object_set_new (x, "jsonrpc", json_string ("2.0"));
	json_object_set_new (x, "id", id ? id : json_null ());
	if (result)  json_object_set_new (x, "result", result);
	if (error)   json_object_set_new (x, "error",  error);
	return x;
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static bool
validate_json_rpc_content_type (const char *content_type)
{
	char *type = NULL;
	char *subtype = NULL;

	struct str_map parameters;
	str_map_init (&parameters);
	parameters.free = free;
	parameters.key_xfrm = tolower_ascii_strxfrm;

	bool result = http_parse_media_type
		(content_type, &type, &subtype, &parameters);
	if (!result)
		goto end;

	if (strcasecmp_ascii (type, "application")
	 || (strcasecmp_ascii (subtype, "json") &&
		 strcasecmp_ascii (subtype, "json-rpc" /* obsolete */)))
		result = false;

	const char *charset = str_map_find (&parameters, "charset");
	if (charset && strcasecmp_ascii (charset, "UTF-8"))
		result = false;

	// Currently ignoring all unknown parametrs

end:
	free (type);
	free (subtype);
	str_map_free (&parameters);
	return result;
}

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

typedef json_t *(*json_rpc_handler_fn) (struct server_context *, json_t *);

struct json_rpc_handler_info
{
	const char *method_name;            ///< JSON-RPC method name
	json_rpc_handler_fn handler;        ///< Method handler
};

static int
json_rpc_handler_info_cmp (const void *first, const void *second)
{
	return strcmp (((struct json_rpc_handler_info *) first)->method_name,
		((struct json_rpc_handler_info *) second)->method_name);
}

// TODO: a method that queues up a ping over IRC: this has to be owned by the
//   server context as a background job that removes itself upon completion.

static json_t *
json_rpc_ping (struct server_context *ctx, json_t *params)
{
	(void) ctx;
	(void) params;

	return json_rpc_response (NULL, json_string ("pong"), NULL);
}

static json_t *
process_json_rpc_request (struct server_context *ctx, json_t *request)
{
	// A list of all available methods; this list has to be ordered.
	// Eventually it might be better to move this into a map in the context.
	static struct json_rpc_handler_info handlers[] =
	{
		{ "ping", json_rpc_ping },
	};

	if (!json_is_object (request))
		return json_rpc_response (NULL, NULL,
			json_rpc_error (JSON_RPC_ERROR_INVALID_REQUEST, NULL));

	json_t *v      = json_object_get (request, "jsonrpc");
	json_t *m      = json_object_get (request, "method");
	json_t *params = json_object_get (request, "params");
	json_t *id     = json_object_get (request, "id");

	const char *version;
	const char *method;

	bool ok = true;
	ok &= v && (version = json_string_value (v)) && !strcmp (version, "2.0");
	ok &= m && (method  = json_string_value (m));
	ok &= !params || json_is_array (params) || json_is_object (params);
	ok &= !id || json_is_null (id) ||
		json_is_string (id) || json_is_number (id);
	if (!ok)
		return json_rpc_response (id, NULL,
			json_rpc_error (JSON_RPC_ERROR_INVALID_REQUEST, NULL));

	struct json_rpc_handler_info key = { .method_name = method };
	struct json_rpc_handler_info *handler = bsearch (&key, handlers,
		N_ELEMENTS (handlers), sizeof key, json_rpc_handler_info_cmp);
	if (!handler)
		return json_rpc_response (id, NULL,
			json_rpc_error (JSON_RPC_ERROR_METHOD_NOT_FOUND, NULL));

	json_t *response = handler->handler (ctx, params);
	if (id)
		return response;

	// Notifications don't get responses
	json_decref (response);
	return NULL;
}

static void
flush_json (json_t *json, struct str *output)
{
	char *utf8 = json_dumps (json, JSON_ENCODE_ANY);
	str_append (output, utf8);
	free (utf8);
	json_decref (json);
}

static void
process_json_rpc (struct server_context *ctx,
	const void *data, size_t len, struct str *output)
{

	json_error_t e;
	json_t *request;
	if (!(request = json_loadb (data, len, JSON_DECODE_ANY, &e)))
	{
		flush_json (json_rpc_response (NULL, NULL,
			json_rpc_error (JSON_RPC_ERROR_PARSE_ERROR, NULL)),
			output);
		return;
	}

	if (json_is_array (request))
	{
		if (!json_array_size (request))
		{
			flush_json (json_rpc_response (NULL, NULL,
				json_rpc_error (JSON_RPC_ERROR_INVALID_REQUEST, NULL)),
				output);
			return;
		}

		json_t *response = json_array ();
		json_t *iter;
		size_t i;

		json_array_foreach (request, i, iter)
		{
			json_t *result = process_json_rpc_request (ctx, iter);
			if (result)
				json_array_append_new (response, result);
		}

		if (json_array_size (response))
			flush_json (response, output);
		else
			json_decref (response);
	}
	else
	{
		json_t *result = process_json_rpc_request (ctx, request);
		if (result)
			flush_json (result, output);
	}
}

// --- Requests ----------------------------------------------------------------

struct request
{
	struct server_context *ctx;         ///< Server context

	struct request_handler *handler;    ///< Current request handler
	void *handler_data;                 ///< User data for the handler

	/// Callback to write some CGI response data to the output
	void (*write_cb) (void *user_data, const void *data, size_t len);

	/// Callback to close the connection.
	/// CALLING THIS MAY CAUSE THE REQUEST TO BE DESTROYED.
	void (*close_cb) (void *user_data);

	void *user_data;                    ///< User data argument for callbacks
};

struct request_handler
{
	LIST_HEADER (struct request_handler)

	/// Install ourselves as the handler for the request if applicable.
	/// Set @a continue_ to false if further processing should be stopped.
	bool (*try_handle) (struct request *request,
		struct str_map *headers, bool *continue_);

	/// Handle incoming data.
	/// Return false if further processing should be stopped.
	bool (*push_cb) (struct request *request, const void *data, size_t len);

	/// Destroy the handler
	void (*destroy_cb) (struct request *request);
};

static void
request_init (struct request *self)
{
	memset (self, 0, sizeof *self);
}

static void
request_free (struct request *self)
{
	if (self->handler)
		self->handler->destroy_cb (self);
}

/// This function is only intended to be run from asynchronous event handlers
/// such as timers, not as a direct result of starting the request or receiving
/// request data.  CALLING THIS MAY CAUSE THE REQUEST TO BE DESTROYED.
static void
request_finish (struct request *self)
{
	self->close_cb (self->user_data);
}

static bool
request_start (struct request *self, struct str_map *headers)
{
	// XXX: it feels like this should rather be two steps:
	//   bool (*can_handle) (request *, headers)
	//   ... install the handler ...
	//   bool (*handle) (request *)
	//
	//   However that might cause some stuff to be done twice.
	//
	//   Another way we could get rid off the continue_ argument is via adding
	//   some way of marking the request as finished from within the handler.

	bool continue_ = true;
	LIST_FOR_EACH (struct request_handler, handler, self->ctx->handlers)
		if (handler->try_handle (self, headers, &continue_))
		{
			self->handler = handler;
			return continue_;
		}

	// Unable to serve the request
	struct str response;
	str_init (&response);
	str_append (&response, "Status: 404 Not Found\n");
	str_append (&response, "Content-Type: text/plain\n\n");
	self->write_cb (self->user_data, response.str, response.len);
	str_free (&response);
	return false;
}

static bool
request_push (struct request *self, const void *data, size_t len)
{
	if (!soft_assert (self->handler))
		// No handler, nothing to do with any data
		return false;

	return self->handler->push_cb (self, data, len);
}

// --- Requests handlers -------------------------------------------------------

static bool
request_handler_json_rpc_try_handle
	(struct request *request, struct str_map *headers, bool *continue_)
{
	const char *content_type = str_map_find (headers, "CONTENT_TYPE");
	const char *method = str_map_find (headers, "REQUEST_METHOD");

	if (!method || strcmp (method, "POST")
	 || !content_type || !validate_json_rpc_content_type (content_type))
		return false;

	struct str *buf = xcalloc (1, sizeof *buf);
	str_init (buf);

	request->handler_data = buf;
	*continue_ = true;
	return true;
}

static bool
request_handler_json_rpc_push
	(struct request *request, const void *data, size_t len)
{
	struct str *buf = request->handler_data;
	if (len)
	{
		str_append_data (buf, data, len);
		return true;
	}

	struct str response;
	str_init (&response);
	str_append (&response, "Status: 200 OK\n");
	str_append_printf (&response, "Content-Type: %s\n\n", "application/json");
	process_json_rpc (request->ctx, buf->str, buf->len, &response);
	request->write_cb (request->user_data, response.str, response.len);
	str_free (&response);
	return false;
}

static void
request_handler_json_rpc_destroy (struct request *request)
{
	struct str *buf = request->handler_data;
	str_free (buf);
	free (buf);

	request->handler_data = NULL;
}

struct request_handler g_request_handler_json_rpc =
{
	.try_handle = request_handler_json_rpc_try_handle,
	.push_cb    = request_handler_json_rpc_push,
	.destroy_cb = request_handler_json_rpc_destroy,
};

// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

static char *
canonicalize_url_path (const char *path)
{
	// XXX: this strips any slashes at the end
	struct str_vector v;
	str_vector_init (&v);
	split_str_ignore_empty (path, '/', &v);

	struct str_vector canonical;
	str_vector_init (&canonical);

	// So that the joined path always begins with a slash
	str_vector_add (&canonical, "");

	for (size_t i = 0; i < v.len; i++)
	{
		const char *dir = v.vector[i];
		if (!strcmp (dir, "."))
			continue;

		if (strcmp (dir, ".."))
			str_vector_add (&canonical, dir);
		else if (canonical.len > 1)
			// ".." never goes above the root
			str_vector_remove (&canonical, canonical.len - 1);
	}
	str_vector_free (&v);

	char *joined = join_str_vector (&canonical, '/');
	str_vector_free (&canonical);
	return joined;
}

static char *
detect_magic (const void *data, size_t len)
{
	magic_t cookie;
	char *mime_type = NULL;

	if (!(cookie = magic_open (MAGIC_MIME)))
		return NULL;

	const char *magic = NULL;
	if (!magic_load (cookie, NULL)
	 && (magic = magic_buffer (cookie, data, len)))
		mime_type = xstrdup (magic);
	else
		print_debug ("MIME type detection failed: %s", magic_error (cookie));

	magic_close (cookie);
	return mime_type;
}

static bool
request_handler_static_try_handle
	(struct request *request, struct str_map *headers, bool *continue_)
{
	// Serving static files is actually quite complicated as it turns out;
	// but this is only meant to serve a few tiny text files

	struct server_context *ctx = request->ctx;
	const char *root = str_map_find (&ctx->config, "static_root");
	if (!root)
	{
		print_debug ("static document root not configured");
		return false;
	}

	const char *method = str_map_find (headers, "REQUEST_METHOD");
	if (!method || strcmp (method, "GET"))
		return false;

	// TODO: look at <SCRIPT_NAME, PATH_INFO>, REQUEST_URI in the headers
	const char *path_info = str_map_find (headers, "PATH_INFO");
	if (!path_info)
	{
		print_debug ("PATH_INFO not defined");
		return false;
	}

	// We need to filter the path to stay in our root
	// Being able to read /etc/passwd would be rather embarrasing
	char *suffix = canonicalize_url_path (path_info);
	char *path = xstrdup_printf ("%s%s", root, suffix);

	// TODO: check that this is a regular file
	FILE *fp = fopen (path, "rb");
	if (!fp)
	{
		struct str response;
		str_init (&response);
		str_append (&response, "Status: 404 Not Found\n");
		str_append (&response, "Content-Type: text/plain\n\n");
		str_append_printf (&response,
			"File %s was not found on this server\n", suffix);
		request->write_cb (request->user_data, response.str, response.len);
		str_free (&response);

		free (suffix);
		free (path);
		return false;
	}

	free (suffix);
	free (path);

	uint8_t buf[8192];
	size_t len;

	// Try to detect the Content-Type from the actual contents
	char *mime_type = NULL;
	if ((len = fread (buf, 1, sizeof buf, fp)))
		mime_type = detect_magic (buf, len);
	if (!mime_type)
		mime_type = xstrdup ("application/octet_stream");

	struct str response;
	str_init (&response);
	str_append (&response, "Status: 200 OK\n");
	str_append_printf (&response, "Content-Type: %s\n\n", mime_type);
	request->write_cb (request->user_data, response.str, response.len);
	str_free (&response);
	free (mime_type);

	// Write the chunk we've used to help us with magic detection;
	// obviously we have to do it after we've written the headers
	if (len)
		request->write_cb (request->user_data, buf, len);

	while ((len = fread (buf, 1, sizeof buf, fp)))
		request->write_cb (request->user_data, buf, len);
	fclose (fp);

	// TODO: this should rather not be returned all at once but in chunks;
	//   file read requests never return EAGAIN
	// TODO: actual file data should really be returned by a callback when
	//   the socket is writable with nothing to be sent (pumping the entire
	//   file all at once won't really work if it's huge).
	*continue_ = false;
	return true;
}

static bool
request_handler_static_push
	(struct request *request, const void *data, size_t len)
{
	(void) request;
	(void) data;
	(void) len;

	// Ignoring all content; we shouldn't receive any (GET)
	return false;
}

static void
request_handler_static_destroy (struct request *request)
{
	(void) request;
}

struct request_handler g_request_handler_static =
{
	.try_handle = request_handler_static_try_handle,
	.push_cb    = request_handler_static_push,
	.destroy_cb = request_handler_static_destroy,
};

// --- Client communication handlers -------------------------------------------

struct client
{
	LIST_HEADER (struct client)

	struct server_context *ctx;         ///< Server context

	int socket_fd;                      ///< The TCP socket
	write_queue_t write_queue;          ///< Write queue

	ev_io read_watcher;                 ///< The socket can be read from
	ev_io write_watcher;                ///< The socket can be written to

	struct client_impl *impl;           ///< Client behaviour
	void *impl_data;                    ///< Client behaviour data
};

struct client_impl
{
	/// Initialize the client as needed
	void (*init) (struct client *client);

	// TODO: a method for graceful shutdown which will, in the case of
	//   WebSockets, actually send a "shutdown" close packet, and in the case
	//   of FastCGI will FCGI_END_REQUEST everything with FCGI_REQUEST_COMPLETE
	//   and FCGI_OVERLOADED all incoming requests in the meantime (the FastCGI
	//   specification isn't very clear about how we should respond to this).
	//
	//   We then should set up a timer for about a second until we kill all
	//   clients for good.

	/// Do any additional cleanup
	void (*destroy) (struct client *client);

	/// Process incoming data; "len == 0" means EOF
	bool (*push) (struct client *client, const void *data, size_t len);
};

static void
client_init (struct client *self)
{
	memset (self, 0, sizeof *self);
	write_queue_init (&self->write_queue);
}

static void
client_free (struct client *self)
{
	write_queue_free (&self->write_queue);
}

static void
client_write (struct client *client, const void *data, size_t len)
{
	write_req_t *req = xcalloc (1, sizeof *req);
	req->data.iov_base = memcpy (xmalloc (len), data, len);
	req->data.iov_len = len;

	write_queue_add (&client->write_queue, req);
	ev_io_start (EV_DEFAULT_ &client->write_watcher);
}

static void
client_remove (struct client *client)
{
	struct server_context *ctx = client->ctx;

	LIST_UNLINK (ctx->clients, client);
	ctx->n_clients--;

	// First uninitialize the higher-level implementation
	client->impl->destroy (client);

	ev_io_stop (EV_DEFAULT_ &client->read_watcher);
	ev_io_stop (EV_DEFAULT_ &client->write_watcher);
	xclose (client->socket_fd);
	client_free (client);
	free (client);
}

// - - FastCGI - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

struct client_fcgi
{
	struct fcgi_muxer muxer;            ///< FastCGI de/multiplexer
};

struct client_fcgi_request
{
	struct fcgi_request *fcgi_request;  ///< FastCGI request
	struct request request;             ///< Request
};

static void
client_fcgi_request_write (void *user_data, const void *data, size_t len)
{
	struct client_fcgi_request *request = user_data;
	fcgi_request_write (request->fcgi_request, data, len);
}

static void
client_fcgi_request_close (void *user_data)
{
	struct client_fcgi_request *request = user_data;
	// TODO: fcgi_request_finish()?  That will most probably end up with us
	//   receiving client_fcgi_request_destroy()
}

static void *
client_fcgi_request_start (void *user_data, struct fcgi_request *fcgi_request)
{
	struct client *client = user_data;

	// TODO: what if the request is aborted by ;
	struct client_fcgi_request *request = xmalloc (sizeof *request);
	request->fcgi_request = fcgi_request;
	request_init (&request->request);
	request->request.ctx = client->ctx;
	request->request.write_cb = client_fcgi_request_write;
	request->request.close_cb = client_fcgi_request_close;
	request->request.user_data = request;
	return request;
}

static void
client_fcgi_request_push (void *handler_data, const void *data, size_t len)
{
	struct client_fcgi_request *request = handler_data;
	request_push (&request->request, data, len);
}

static void
client_fcgi_request_destroy (void *handler_data)
{
	struct client_fcgi_request *request = handler_data;
	request_free (&request->request);
	free (handler_data);
}

static void
client_fcgi_write (void *user_data, const void *data, size_t len)
{
	struct client *client = user_data;
	client_write (client, data, len);
}

static void
client_fcgi_close (void *user_data)
{
	struct client *client = user_data;
	client_remove (client);
}

static void
client_fcgi_init (struct client *client)
{
	struct client_fcgi *self = xcalloc (1, sizeof *self);
	client->impl_data = self;

	fcgi_muxer_init (&self->muxer);
	self->muxer.write_cb           = client_fcgi_write;
	self->muxer.close_cb           = client_fcgi_close;
	self->muxer.request_start_cb   = client_fcgi_request_start;
	self->muxer.request_push_cb    = client_fcgi_request_push;
	self->muxer.request_destroy_cb = client_fcgi_request_destroy;
	self->muxer.user_data          = client;
}

static void
client_fcgi_destroy (struct client *client)
{
	struct client_fcgi *self = client->impl_data;
	client->impl_data = NULL;

	fcgi_muxer_free (&self->muxer);
	free (self);
}

static bool
client_fcgi_push (struct client *client, const void *data, size_t len)
{
	struct client_fcgi *self = client->impl_data;
	fcgi_muxer_push (&self->muxer, data, len);
	return true;
}

static struct client_impl g_client_fcgi =
{
	.init    = client_fcgi_init,
	.destroy = client_fcgi_destroy,
	.push    = client_fcgi_push,
};

// - - SCGI  - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

struct client_scgi
{
	struct scgi_parser parser;          ///< SCGI stream parser
	struct request request;             ///< Request (only one per connection)
};

static void
client_scgi_write (void *user_data, const void *data, size_t len)
{
	struct client *client = user_data;
	client_write (client, data, len);
}

static void
client_scgi_close (void *user_data)
{
	// NOTE: this rather really means "close me [the request]"
	struct client *client = user_data;
	client_remove (client);
}

static bool
client_scgi_on_headers_read (void *user_data)
{
	struct client *client = user_data;
	struct client_scgi *self = client->impl_data;
	return request_start (&self->request, &self->parser.headers);
}

static bool
client_scgi_on_content (void *user_data, const void *data, size_t len)
{
	struct client *client = user_data;
	struct client_scgi *self = client->impl_data;

	// XXX: do we have to count CONTENT_LENGTH and supply our own EOF?
	//   If we do produce our own EOF, we should probably make sure we don't
	//   send it twice in a row.
	return request_push (&self->request, data, len);
}

static void
client_scgi_init (struct client *client)
{
	struct client_scgi *self = xcalloc (1, sizeof *self);
	client->impl_data = self;

	request_init (&self->request);
	self->request.ctx            = client->ctx;
	self->request.write_cb       = client_scgi_write;
	self->request.close_cb       = client_scgi_close;
	self->request.user_data      = client;

	scgi_parser_init (&self->parser);
	self->parser.on_headers_read = client_scgi_on_headers_read;
	self->parser.on_content      = client_scgi_on_content;
	self->parser.user_data       = client;
}

static void
client_scgi_destroy (struct client *client)
{
	struct client_scgi *self = client->impl_data;
	client->impl_data = NULL;

	request_free (&self->request);
	scgi_parser_free (&self->parser);
	free (self);
}

static bool
client_scgi_push (struct client *client, const void *data, size_t len)
{
	struct client_scgi *self = client->impl_data;
	struct error *e = NULL;
	if (scgi_parser_push (&self->parser, data, len, &e))
		return true;

	if (e != NULL)
	{
		print_debug ("SCGI parser failed: %s", e->message);
		error_free (e);
	}
	return false;
}

static struct client_impl g_client_scgi =
{
	.init    = client_scgi_init,
	.destroy = client_scgi_destroy,
	.push    = client_scgi_push,
};

// - - WebSockets  - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

struct client_ws
{
	struct ws_handler handler;          ///< WebSockets connection handler
};

static void
client_ws_write (void *user_data, const void *data, size_t len)
{
	struct client *client = user_data;
	client_write (client, data, len);
}

static bool
client_ws_on_message (void *user_data,
	enum ws_opcode type, const void *data, size_t len)
{
	struct client *client = user_data;
	struct client_ws *self = client->impl_data;

	if (type != WS_OPCODE_TEXT)
	{
		ws_handler_fail (&self->handler, WS_STATUS_UNSUPPORTED_DATA);
		return false;
	}

	struct str response;
	str_init (&response);
	process_json_rpc (client->ctx, data, len, &response);
	ws_handler_send (&self->handler,
		WS_OPCODE_TEXT, response.str, response.len);
	str_free (&response);
	return true;
}

static void
client_ws_init (struct client *client)
{
	struct client_ws *self = xmalloc (sizeof *self);
	client->impl_data = self;

	ws_handler_init (&self->handler);
	self->handler.write_cb   = client_ws_write;
	self->handler.on_message = client_ws_on_message;
	self->handler.user_data  = client;
	// TODO: configure the handler some more, e.g. regarding the protocol

	// One mebibyte seems to be a reasonable value
	self->handler.max_payload_len = 1 << 10;
}

static void
client_ws_destroy (struct client *client)
{
	struct client_ws *self = client->impl_data;
	client->impl_data = NULL;

	ws_handler_free (&self->handler);
	free (self);
}

static bool
client_ws_push (struct client *client, const void *data, size_t len)
{
	struct client_ws *self = client->impl_data;
	return ws_handler_push (&self->handler, data, len);
}

static struct client_impl g_client_ws =
{
	.init    = client_ws_init,
	.destroy = client_ws_destroy,
	.push    = client_ws_push,
};


// --- Basic server stuff ------------------------------------------------------

struct listener
{
	int fd;                             ///< Listening socket FD
	ev_io watcher;                      ///< New connection available
	struct client_impl *impl;           ///< Client behaviour
};

static void
close_listeners (struct server_context *self)
{
	// TODO: factor out the closing act, to be used in initiate_quit()
	for (size_t i = 0; i < self->n_listeners; i++)
	{
		struct listener *listener = &self->listeners[i];
		if (listener->fd == -1)
			continue;

		ev_io_stop (EV_DEFAULT_ &listener->watcher);
		xclose (listener->fd);
		listener->fd = -1;
	}
}

static bool
client_read_loop (EV_P_ struct client *client, ev_io *watcher)
{
	char buf[8192];
	while (true)
	{
		ssize_t n_read = recv (watcher->fd, buf, sizeof buf, 0);
		if (n_read >= 0)
		{
			if (!client->impl->push (client, buf, n_read))
				return false;
			if (!n_read)
				break;
		}
		else if (errno == EAGAIN)
			return true;
		else if (errno != EINTR)
			return false;
	}

	// Don't receive the EOF condition repeatedly
	ev_io_stop (EV_A_ watcher);

	// We can probably still write, so let's just return
	return true;
}

static void
on_client_ready (EV_P_ ev_io *watcher, int revents)
{
	struct client *client = watcher->data;

	if (revents & EV_READ)
		if (!client_read_loop (EV_A_ client, watcher))
			goto close;
	if (revents & EV_WRITE)
		// TODO: shouldn't we at least provide an option (to be used by a client
		//   implementation if it so desires) to close the connection once we've
		//   finished flushing the write queue?  This should probably even be
		//   the default behaviour, as it's fairly uncommon for clients to
		//   shutdown the socket for writes while leaving it open for reading.
		// TODO: some sort of "on_buffers_flushed" callback for streaming huge
		//   chunks of external (or generated) data.
		if (!flush_queue (&client->write_queue, watcher))
			goto close;
	return;

close:
	client_remove (client);
}

static void
make_client (EV_P_ struct client_impl *impl, int sock_fd)
{
	struct server_context *ctx = ev_userdata (loop);
	set_blocking (sock_fd, false);

	struct client *client = xmalloc (sizeof *client);
	client_init (client);
	client->socket_fd = sock_fd;
	client->impl = impl;

	ev_io_init (&client->read_watcher,  on_client_ready, sock_fd, EV_READ);
	ev_io_init (&client->write_watcher, on_client_ready, sock_fd, EV_WRITE);
	client->read_watcher.data = client;
	client->write_watcher.data = client;

	// We're only interested in reading as the write queue is empty now
	ev_io_start (EV_A_ &client->read_watcher);

	// Initialize the higher-level implementation
	client->impl->init (client);

	LIST_PREPEND (ctx->clients, client);
	ctx->n_clients++;
}

static void
on_client_available (EV_P_ ev_io *watcher, int revents)
{
	struct listener *listener = watcher->data;
	(void) revents;

	while (true)
	{
		int sock_fd = accept (watcher->fd, NULL, NULL);
		if (sock_fd != -1)
			make_client (EV_A_ listener->impl, sock_fd);
		else if (errno == EAGAIN)
			return;
		else if (errno != EINTR && errno != ECONNABORTED)
			break;
	}

	// Stop accepting connections to prevent busy looping
	ev_io_stop (EV_A_ watcher);

	print_fatal ("%s: %s", "accept", strerror (errno));
	// TODO: initiate_quit (ctx);
}

// --- Application setup -------------------------------------------------------

/// This function handles values that require validation before their first use,
/// or some kind of a transformation (such as conversion to an integer) needs
/// to be done before they can be used directly.
static bool
parse_config (struct server_context *ctx, struct error **e)
{
	(void) ctx;
	(void) e;

	return true;
}

static int
listener_bind (struct addrinfo *gai_iter)
{
	int fd = socket (gai_iter->ai_family,
		gai_iter->ai_socktype, gai_iter->ai_protocol);
	if (fd == -1)
		return -1;
	set_cloexec (fd);

	int yes = 1;
	soft_assert (setsockopt (fd, SOL_SOCKET, SO_KEEPALIVE,
		&yes, sizeof yes) != -1);
	soft_assert (setsockopt (fd, SOL_SOCKET, SO_REUSEADDR,
		&yes, sizeof yes) != -1);

	char host[NI_MAXHOST], port[NI_MAXSERV];
	host[0] = port[0] = '\0';
	int err = getnameinfo (gai_iter->ai_addr, gai_iter->ai_addrlen,
		host, sizeof host, port, sizeof port,
		NI_NUMERICHOST | NI_NUMERICSERV);
	if (err)
		print_debug ("%s: %s", "getnameinfo", gai_strerror (err));

	char *address = format_host_port_pair (host, port);
	if (bind (fd, gai_iter->ai_addr, gai_iter->ai_addrlen))
		print_error ("bind to %s failed: %s", address, strerror (errno));
	else if (listen (fd, 16 /* arbitrary number */))
		print_error ("listen on %s failed: %s", address, strerror (errno));
	else
	{
		print_status ("listening on %s", address);
		free (address);
		return fd;
	}

	free (address);
	xclose (fd);
	return -1;
}

static void
listener_add (struct server_context *ctx, const char *host, const char *port,
	const struct addrinfo *gai_hints, struct client_impl *impl)
{
	struct addrinfo *gai_result, *gai_iter;
	int err = getaddrinfo (host, port, gai_hints, &gai_result);
	if (err)
	{
		char *address = format_host_port_pair (host, port);
		print_error ("bind to %s failed: %s: %s",
			address, "getaddrinfo", gai_strerror (err));
		free (address);
		return;
	}

	int fd;
	for (gai_iter = gai_result; gai_iter; gai_iter = gai_iter->ai_next)
	{
		if ((fd = listener_bind (gai_iter)) == -1)
			continue;
		set_blocking (fd, false);

		struct listener *listener = &ctx->listeners[ctx->n_listeners++];
		ev_io_init (&listener->watcher, on_client_available, fd, EV_READ);
		ev_io_start (EV_DEFAULT_ &listener->watcher);
		listener->watcher.data = listener;
		listener->impl = impl;
		break;
	}
	freeaddrinfo (gai_result);
}

static void
get_ports_from_config (struct server_context *ctx,
	const char *key, struct str_vector *out)
{
	const char *ports;
	if ((ports = str_map_find (&ctx->config, key)))
		split_str_ignore_empty (ports, ',', out);
}

static bool
setup_listen_fds (struct server_context *ctx, struct error **e)
{
	static const struct addrinfo gai_hints =
	{
		.ai_socktype = SOCK_STREAM,
		.ai_flags = AI_PASSIVE,
	};

	struct str_vector ports_fcgi;  str_vector_init (&ports_fcgi);
	struct str_vector ports_scgi;  str_vector_init (&ports_scgi);
	struct str_vector ports_ws;    str_vector_init (&ports_ws);

	get_ports_from_config (ctx, "port_fastcgi", &ports_fcgi);
	get_ports_from_config (ctx, "port_scgi",    &ports_scgi);
	get_ports_from_config (ctx, "port_ws",      &ports_ws);

	const char *bind_host = str_map_find (&ctx->config, "bind_host");
	size_t n_ports = ports_fcgi.len + ports_scgi.len + ports_ws.len;
	ctx->listeners = xcalloc (n_ports, sizeof *ctx->listeners);

	for (size_t i = 0; i < ports_fcgi.len; i++)
		listener_add (ctx, bind_host, ports_fcgi.vector[i],
			&gai_hints, &g_client_fcgi);
	for (size_t i = 0; i < ports_scgi.len; i++)
		listener_add (ctx, bind_host, ports_scgi.vector[i],
			&gai_hints, &g_client_scgi);
	for (size_t i = 0; i < ports_ws.len; i++)
		listener_add (ctx, bind_host, ports_ws.vector[i],
			&gai_hints, &g_client_ws);

	str_vector_free (&ports_fcgi);
	str_vector_free (&ports_scgi);
	str_vector_free (&ports_ws);

	if (!ctx->n_listeners)
	{
		error_set (e, "%s: %s",
			"network setup failed", "no ports to listen on");
		return false;
	}
	return true;
}

static bool
lock_pid_file (struct server_context *ctx, struct error **e)
{
	const char *path = str_map_find (&ctx->config, "pid_file");
	if (!path)
		return true;

	int fd = open (path, O_RDWR | O_CREAT,
		S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH /* 644 */);
	if (fd < 0)
	{
		error_set (e, "can't open `%s': %s", path, strerror (errno));
		return false;
	}

	struct flock lock =
	{
		.l_type = F_WRLCK,
		.l_start = 0,
		.l_whence = SEEK_SET,
		.l_len = 0,
	};
	if (fcntl (fd, F_SETLK, &lock))
	{
		error_set (e, "can't lock `%s': %s", path, strerror (errno));
		return false;
	}

	struct str pid;
	str_init (&pid);
	str_append_printf (&pid, "%ld", (long) getpid ());

	if (ftruncate (fd, 0)
	 || write (fd, pid.str, pid.len) != (ssize_t) pid.len)
	{
		error_set (e, "can't write to `%s': %s", path, strerror (errno));
		return false;
	}
	str_free (&pid);

	// Intentionally not closing the file descriptor; it must stay alive
	// for the entire life of the application
	return true;
}

// --- Tests -------------------------------------------------------------------

static void
test_utf8 (void)
{
	const char valid  [] = "2H₂ + O₂ ⇌ 2H₂O, R = 4.7 kΩ, ⌀ 200 mm";
	const char invalid[] = "\xf0\x90\x28\xbc";
	soft_assert ( utf8_validate (valid,   sizeof valid));
	soft_assert (!utf8_validate (invalid, sizeof invalid));
}

static void
test_base64 (void)
{
	char data[65];
	for (size_t i = 0; i < N_ELEMENTS (data); i++)
		data[i] = i;

	struct str encoded;  str_init (&encoded);
	struct str decoded;  str_init (&decoded);

	base64_encode (data, sizeof data, &encoded);
	soft_assert (base64_decode (encoded.str, false, &decoded));
	soft_assert (decoded.len == sizeof data);
	soft_assert (!memcmp (decoded.str, data, sizeof data));

	str_free (&encoded);
	str_free (&decoded);
}

static void
test_http_parser (void)
{
	struct str_map parameters;
	str_map_init (&parameters);
	parameters.key_xfrm = tolower_ascii_strxfrm;

	char *type = NULL;
	char *subtype = NULL;
	soft_assert (http_parse_media_type ("TEXT/html; CHARset=\"utf\\-8\"",
		&type, &subtype, &parameters));
	soft_assert (!strcasecmp_ascii (type, "text"));
	soft_assert (!strcasecmp_ascii (subtype, "html"));
	soft_assert (parameters.len == 1);
	soft_assert (!strcmp (str_map_find (&parameters, "charset"), "utf-8"));
	str_map_free (&parameters);

	struct http_protocol *protocols;
	soft_assert (http_parse_upgrade ("websocket, HTTP/2.0, , ", &protocols));

	soft_assert (!strcmp (protocols->name, "websocket"));
	soft_assert (!protocols->version);

	soft_assert (!strcmp (protocols->next->name, "HTTP"));
	soft_assert (!strcmp (protocols->next->version, "2.0"));

	soft_assert (!protocols->next->next);

	LIST_FOR_EACH (struct http_protocol, iter, protocols)
		http_protocol_destroy (iter);
}

static bool
test_scgi_parser_on_headers_read (void *user_data)
{
	struct scgi_parser *parser = user_data;
	soft_assert (parser->headers.len == 4);
	soft_assert (!strcmp (str_map_find (&parser->headers,
		"CONTENT_LENGTH"), "27"));
	soft_assert (!strcmp (str_map_find (&parser->headers,
		"SCGI"), "1"));
	soft_assert (!strcmp (str_map_find (&parser->headers,
		"REQUEST_METHOD"), "POST"));
	soft_assert (!strcmp (str_map_find (&parser->headers,
		"REQUEST_URI"), "/deepthought"));
	return true;
}

static bool
test_scgi_parser_on_content (void *user_data, const void *data, size_t len)
{
	(void) user_data;
	soft_assert (!strncmp (data, "What is the answer to life?", len));
	return true;
}

static void
test_scgi_parser (void)
{
	struct scgi_parser parser;
	scgi_parser_init (&parser);
	parser.on_headers_read = test_scgi_parser_on_headers_read;
	parser.on_content      = test_scgi_parser_on_content;
	parser.user_data       = &parser;

	// This is an example straight from the specification
	const char example[] =
		"70:"
			"CONTENT_LENGTH" "\0" "27" "\0"
			"SCGI" "\0" "1" "\0"
			"REQUEST_METHOD" "\0" "POST" "\0"
			"REQUEST_URI" "\0" "/deepthought" "\0"
		","
		"What is the answer to life?";

	soft_assert (scgi_parser_push (&parser, example, sizeof example, NULL));
	scgi_parser_free (&parser);
}

static bool
test_websockets_on_frame_header (void *user_data, const struct ws_parser *self)
{
	(void) user_data;
	soft_assert (self->is_fin);
	soft_assert (self->is_masked);
	soft_assert (self->opcode == WS_OPCODE_TEXT);
	return true;
}

static bool
test_websockets_on_frame (void *user_data, const struct ws_parser *self)
{
	(void) user_data;
	soft_assert (self->input.len == self->payload_len);
	soft_assert (!strncmp (self->input.str, "Hello", self->input.len));
	return true;
}

static void
test_websockets (void)
{
	char *accept = ws_encode_response_key ("dGhlIHNhbXBsZSBub25jZQ==");
	soft_assert (!strcmp (accept, "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="));
	free (accept);

	struct ws_parser parser;
	ws_parser_init (&parser);
	parser.on_frame_header = test_websockets_on_frame_header;
	parser.on_frame        = test_websockets_on_frame;
	parser.user_data       = &parser;

	const char frame[] = "\x81\x85\x37\xfa\x21\x3d\x7f\x9f\x4d\x51\x58";
	soft_assert (ws_parser_push (&parser, frame, sizeof frame - 1));
	ws_parser_free (&parser);

	// TODO: test the server handler (happy path)
}

static void
test_misc (void)
{
	soft_assert ( validate_json_rpc_content_type
		("application/JSON; charset=\"utf-8\""));
	soft_assert (!validate_json_rpc_content_type
		("text/html; charset=\"utf-8\""));

	char *canon = canonicalize_url_path ("///../../../etc/./passwd");
	soft_assert (!strcmp (canon, "/etc/passwd"));
	free (canon);
}

int
test_main (int argc, char *argv[])
{
	struct test test;
	test_init (&test, argc, argv);

	test_add_simple (&test, "/utf-8",       NULL, test_utf8);
	test_add_simple (&test, "/base64",      NULL, test_base64);
	test_add_simple (&test, "/http-parser", NULL, test_http_parser);
	test_add_simple (&test, "/scgi-parser", NULL, test_scgi_parser);
	test_add_simple (&test, "/websockets",  NULL, test_websockets);

	test_add_simple (&test, "/misc",        NULL, test_misc);

	// TODO: write more tests

	return test_run (&test);
}

// --- Main program ------------------------------------------------------------

static void
on_termination_signal (EV_P_ ev_signal *handle, int revents)
{
	struct server_context *ctx = ev_userdata (loop);
	(void) handle;
	(void) revents;

	// TODO: initiate_quit (ctx);
}

static void
daemonize (void)
{
	print_status ("daemonizing...");

	if (chdir ("/"))
		exit_fatal ("%s: %s", "chdir", strerror (errno));

	pid_t pid;
	if ((pid = fork ()) < 0)
		exit_fatal ("%s: %s", "fork", strerror (errno));
	else if (pid)
		exit (EXIT_SUCCESS);

	setsid ();
	signal (SIGHUP, SIG_IGN);

	if ((pid = fork ()) < 0)
		exit_fatal ("%s: %s", "fork", strerror (errno));
	else if (pid)
		exit (EXIT_SUCCESS);

	openlog (PROGRAM_NAME, LOG_NDELAY | LOG_NOWAIT | LOG_PID, 0);
	g_log_message_real = log_message_syslog;

	// XXX: we may close our own descriptors this way, crippling ourselves;
	//   there is no real guarantee that we will start with all three
	//   descriptors open.  In theory we could try to enumerate the descriptors
	//   at the start of main().
	for (int i = 0; i < 3; i++)
		xclose (i);

	int tty = open ("/dev/null", O_RDWR);
	if (tty != 0 || dup (0) != 1 || dup (0) != 2)
		exit_fatal ("failed to reopen FD's: %s", strerror (errno));
}

static void
parse_program_arguments (int argc, char **argv)
{
	static const struct opt opts[] =
	{
		{ 't', "test", NULL, 0, "self-test" },
		{ 'd', "debug", NULL, 0, "run in debug mode" },
		{ 'h', "help", NULL, 0, "display this help and exit" },
		{ 'V', "version", NULL, 0, "output version information and exit" },
		{ 'w', "write-default-cfg", "FILENAME",
		  OPT_OPTIONAL_ARG | OPT_LONG_ONLY,
		  "write a default configuration file and exit" },
		{ 0, NULL, NULL, 0, NULL }
	};

	struct opt_handler oh;
	opt_handler_init (&oh, argc, argv, opts, NULL, "JSON-RPC 2.0 demo server.");

	int c;
	while ((c = opt_handler_get (&oh)) != -1)
	switch (c)
	{
	case 't':
		test_main (argc, argv);
		exit (EXIT_SUCCESS);
	case 'd':
		g_debug_mode = true;
		break;
	case 'h':
		opt_handler_usage (&oh, stdout);
		exit (EXIT_SUCCESS);
	case 'V':
		printf (PROGRAM_NAME " " PROGRAM_VERSION "\n");
		exit (EXIT_SUCCESS);
	case 'w':
		call_write_default_config (optarg, g_config_table);
		exit (EXIT_SUCCESS);
	default:
		print_error ("wrong options");
		opt_handler_usage (&oh, stderr);
		exit (EXIT_FAILURE);
	}

	argc -= optind;
	argv += optind;

	if (argc)
	{
		opt_handler_usage (&oh, stderr);
		exit (EXIT_FAILURE);
	}
	opt_handler_free (&oh);
}

int
main (int argc, char *argv[])
{
	parse_program_arguments (argc, argv);

	print_status (PROGRAM_NAME " " PROGRAM_VERSION " starting");

	struct server_context ctx;
	server_context_init (&ctx);

	struct error *e = NULL;
	if (!read_config_file (&ctx.config, &e))
	{
		print_error ("error loading configuration: %s", e->message);
		error_free (e);
		exit (EXIT_FAILURE);
	}

	struct ev_loop *loop;
	if (!(loop = EV_DEFAULT))
		exit_fatal ("libev initialization failed");

	ev_set_userdata (loop, &ctx);

	ev_signal_init (&ctx.sigterm_watcher, on_termination_signal, SIGTERM);
	ev_signal_start (EV_DEFAULT_ &ctx.sigterm_watcher);

	ev_signal_init (&ctx.sigint_watcher, on_termination_signal, SIGINT);
	ev_signal_start (EV_DEFAULT_ &ctx.sigint_watcher);

	(void) signal (SIGPIPE, SIG_IGN);

	LIST_PREPEND (ctx.handlers, &g_request_handler_static);
	LIST_PREPEND (ctx.handlers, &g_request_handler_json_rpc);

	if (!parse_config (&ctx, &e)
	 || !lock_pid_file (&ctx, &e)
	 || !setup_listen_fds (&ctx, &e))
	{
		print_error ("%s", e->message);
		error_free (e);
		exit (EXIT_FAILURE);
	}

	if (!g_debug_mode)
		daemonize ();

	ev_run (loop, 0);
	ev_loop_destroy (loop);

	server_context_free (&ctx);
	return EXIT_SUCCESS;
}