uce/src/linux_fastcgi.cpp

#include "lib/uce_lib.cpp"
#include <csetjmp>
#include <deque>
#include <errno.h>
#include <fcntl.h>
#include <sys/socket.h>

ServerState server_state;

#include "fastcgi/src/fcgicc.cc"

FastCGIServer server;
pid_t http_worker_pid = 0;
pid_t proactive_compiler_pid = 0;
pid_t websocket_exec_pid = 0;
bool worker_accepts_http = false;
static sigjmp_buf request_fault_jmp;
static volatile sig_atomic_t request_fault_active = 0;
static volatile sig_atomic_t request_fault_signal = 0;
static Request* request_fault_request = 0;
static String request_fault_trace = "";
static int websocket_exec_fd = -1;
static String websocket_exec_read_buffer = "";
static std::deque<DTree> websocket_exec_pending_jobs;
static DTree websocket_exec_inflight_job;
static String websocket_exec_write_buffer = "";

void close_inherited_server_sockets();
u64 request_seed_from_time(f64 time_value);
void prepare_request_body_maps(Request& request);

Request* set_active_request(Request& request)
{
	Request* previous_context = context;
	context = &request;
	return(previous_context);
}

void restore_active_request(Request* previous_context)
{
	context = previous_context;
}

String request_status_line(Request& request, int status_code, String reason)
{
	String status = std::to_string(status_code) + " " + reason;
	if(request.params["GATEWAY_INTERFACE"] != "")
		return("Status: " + status);
	return("HTTP/1.1 " + status);
}

void clear_request_output(Request& request)
{
	for(auto* stream : request.ob_stack)
		delete stream;
	request.ob_stack.clear();
	request.ob_start();
}

void render_request_failure(Request& request, String title, String details, String trace, int status_code = 500)
{
	request.response_code = request_status_line(request, status_code, "Internal Server Error");
	request.header.clear();
	request.set_cookies.clear();
	request.header["Content-Type"] = "text/plain; charset=utf-8";
	request.err.clear();

	Request* previous_context = set_active_request(request);
	clear_request_output(request);

	String body;
	body += "UCE runtime error\n";
	body += "Request: " + first(request.params["REQUEST_URI"], request.params["SCRIPT_FILENAME"]) + "\n";
	body += "Script: " + request.params["SCRIPT_FILENAME"] + "\n";
	body += "Error: " + title + "\n";
	if(details != "")
		body += "Details: " + details + "\n";
	if(request.server && request.params["SCRIPT_FILENAME"] != "")
	{
		String generated = compiler_generated_cpp_path(&request, request.params["SCRIPT_FILENAME"]);
		body += "Generated C++: " + generated + "\n";
		body += "Hint: if this came from template code, inspect the generated C++ and the nearest .uce literal/code delimiter before changing runtime code.\n";
	}
	else
		body += "Hint: inspect the requested .uce file, its generated C++, and any component/unit called immediately before this failure.\n";
	if(request_fault_signal != 0)
	{
		String sig_label = signal_name((int)request_fault_signal);
		body += "Signal: " + std::to_string((int)request_fault_signal);
		if(sig_label != "")
			body += " (" + sig_label + ")";
		body += "\n";
	}
	if(trace != "")
		body += "\nTrace:\n" + trace;

	print(body);
	request.err = body;
	request.flags.status = status_code;
	restore_active_request(previous_context);
}

void on_request_fault_signal(int sig)
{
	request_fault_signal = sig;
	if(request_fault_active && request_fault_request)
	{
		request_fault_trace = capture_backtrace_string(32, 1);
		siglongjmp(request_fault_jmp, 1);
	}
	on_segfault(sig);
}

void install_request_fault_handlers()
{
	signal(SIGSEGV, on_request_fault_signal);
	signal(SIGABRT, on_request_fault_signal);
	signal(SIGBUS, on_request_fault_signal);
	signal(SIGILL, on_request_fault_signal);
	signal(SIGFPE, on_request_fault_signal);
	signal(SIGALRM, on_request_fault_signal);
}

void restore_request_fault_handlers()
{
	signal(SIGSEGV, on_segfault);
	signal(SIGABRT, on_segfault);
	signal(SIGBUS, on_segfault);
	signal(SIGILL, on_segfault);
	signal(SIGFPE, on_segfault);
	signal(SIGALRM, on_segfault);
}

namespace {

bool websocket_ipc_set_nonblocking(int fd)
{
	int flags = fcntl(fd, F_GETFL, 0);
	if(flags == -1)
		return(false);
	return(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
}

bool websocket_exec_enabled_for_process()
{
	return(worker_accepts_http);
}

u64 websocket_exec_queue_limit_bytes()
{
	u64 configured = int_val(server_state.config["WEBSOCKET_EXEC_QUEUE_BYTES"]);
	if(configured < 64 * 1024)
		configured = 1024 * 1024;
	return(configured);
}

bool websocket_exec_has_inflight_job()
{
	return(websocket_exec_inflight_job.to_bool());
}

FastCGIServer::Connection* websocket_find_connection(String connection_id)
{
	for(auto& item : server.client_sockets)
	{
		FastCGIServer::Connection* connection = item.second;
		if(connection->is_websocket && connection->websocket_connection_id == connection_id)
			return(connection);
	}
	return(0);
}

void websocket_exec_clear_ipc_state()
{
	if(websocket_exec_fd != -1)
		close(websocket_exec_fd);
	websocket_exec_fd = -1;
	websocket_exec_read_buffer = "";
	websocket_exec_write_buffer = "";
	websocket_exec_inflight_job.clear();
}

void websocket_exec_close_connection(String connection_id, u16 status_code = 1011, String reason = "websocket handler unavailable")
{
	if(connection_id == "")
		return;
	server.websocket_close(connection_id, status_code, reason);
}

void websocket_exec_fail_inflight_job(String reason = "websocket handler unavailable")
{
	if(!websocket_exec_has_inflight_job())
		return;
	websocket_exec_close_connection(websocket_exec_inflight_job["connection_id"].to_string(), 1011, reason);
	websocket_exec_inflight_job.clear();
	websocket_exec_write_buffer = "";
}

void websocket_exec_queue_job(DTree job)
{
	if(job["connection_id"].to_string() == "")
		return;

	u64 queued_bytes = websocket_exec_write_buffer.length();
	if(websocket_exec_has_inflight_job())
		queued_bytes += websocket_exec_inflight_job["serialized"].to_string().length();
	for(auto& pending : websocket_exec_pending_jobs)
		queued_bytes += pending["serialized"].to_string().length();
	queued_bytes += json_encode(job).length();

	if(queued_bytes > websocket_exec_queue_limit_bytes())
	{
		printf("(!) websocket dispatch queue overflow for %s\n", job["connection_id"].to_string().c_str());
		websocket_exec_close_connection(job["connection_id"].to_string(), 1013, "websocket server busy");
		return;
	}

	job["serialized"] = json_encode(job) + "\n";
	websocket_exec_pending_jobs.push_back(job);
}

StringList websocket_exec_snapshot_connections(String scope)
{
	return(server.websocket_connection_ids(scope));
}

void websocket_exec_append_command(DTree command)
{
	if(!context)
		return;
	context->resources.websocket_dispatch_commands.push(command);
}

DTree websocket_exec_make_message_command(String action, String message, bool binary)
{
	DTree command;
	command["action"] = action;
	command["binary"].set_bool(binary);
	command["message_b64"] = base64_encode(message);
	return(command);
}

DTree websocket_exec_make_close_command(String connection_id, u16 status_code = 1000, String reason = "")
{
	DTree command;
	command["action"] = "close";
	command["connection_id"] = connection_id;
	command["status_code"] = (f64)status_code;
	command["reason"] = reason;
	return(command);
}

void websocket_exec_apply_command(DTree command)
{
	String action = command["action"].to_string();
	if(action == "broadcast")
	{
		bool ok = false;
		String payload = base64_decode(command["message_b64"].to_string(), ok);
		if(!ok)
			return;
		server.websocket_broadcast(command["scope"].to_string(), payload, command["binary"].to_bool());
		return;
	}
	if(action == "send_to")
	{
		bool ok = false;
		String payload = base64_decode(command["message_b64"].to_string(), ok);
		if(!ok)
			return;
		server.websocket_send_to(command["connection_id"].to_string(), payload, command["binary"].to_bool());
		return;
	}
	if(action == "close")
	{
		server.websocket_close(
			command["connection_id"].to_string(),
			(u16)command["status_code"].to_u64(),
			command["reason"].to_string()
		);
	}
}

void websocket_exec_apply_result(DTree result)
{
	if(result["type"].to_string() != "result")
		return;

	String inflight_connection_id = websocket_exec_inflight_job["connection_id"].to_string();
	String result_connection_id = result["connection_id"].to_string();
	if(inflight_connection_id != "" && result_connection_id != "" && inflight_connection_id != result_connection_id)
	{
		printf("(!) websocket dispatch result mismatch: expected %s got %s\n",
			inflight_connection_id.c_str(),
			result_connection_id.c_str());
	}

	FastCGIServer::Connection* connection = websocket_find_connection(result_connection_id);
	if(connection)
		connection->websocket_state = result["connection_state"];

	result["commands"].each([] (DTree command, String) {
		websocket_exec_apply_command(command);
	});

	websocket_exec_inflight_job.clear();
}

void websocket_exec_handle_ipc_line(String line)
{
	line = trim(line);
	if(line == "")
		return;
	DTree result = json_decode(line);
	websocket_exec_apply_result(result);
}

void websocket_exec_read_results()
{
	if(websocket_exec_fd == -1)
		return;

	char buffer[4096];
	for(;;)
	{
		ssize_t read_result = read(websocket_exec_fd, buffer, sizeof(buffer));
		if(read_result == 0)
		{
			printf("(!) websocket executor disconnected\n");
			websocket_exec_fail_inflight_job();
			websocket_exec_clear_ipc_state();
			return;
		}
		if(read_result < 0)
		{
			if(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
				break;
			perror("websocket executor read");
			websocket_exec_fail_inflight_job();
			websocket_exec_clear_ipc_state();
			return;
		}

		websocket_exec_read_buffer.append(buffer, read_result);
		for(;;)
		{
			size_t line_end = websocket_exec_read_buffer.find('\n');
			if(line_end == String::npos)
				break;
			String line = websocket_exec_read_buffer.substr(0, line_end);
			websocket_exec_read_buffer.erase(0, line_end + 1);
			websocket_exec_handle_ipc_line(line);
		}
	}
}

void websocket_exec_flush_queue()
{
	if(websocket_exec_fd == -1)
		return;

	if(websocket_exec_write_buffer == "" && !websocket_exec_has_inflight_job() && websocket_exec_pending_jobs.size() > 0)
	{
		websocket_exec_inflight_job = websocket_exec_pending_jobs.front();
		websocket_exec_pending_jobs.pop_front();
		websocket_exec_write_buffer = websocket_exec_inflight_job["serialized"].to_string();
	}

	while(websocket_exec_write_buffer != "")
	{
		ssize_t write_result = write(websocket_exec_fd, websocket_exec_write_buffer.data(), websocket_exec_write_buffer.length());
		if(write_result < 0)
		{
			if(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
				return;
			perror("websocket executor write");
			websocket_exec_fail_inflight_job();
			websocket_exec_clear_ipc_state();
			return;
		}
		if(write_result == 0)
			return;
		websocket_exec_write_buffer.erase(0, write_result);
	}
}

Request websocket_exec_build_event_request(DTree job, String message)
{
	Request event_request;
	event_request.server = &server_state;
	event_request.params = job["params"].to_stringmap();
	event_request.params["REQUEST_METHOD"] = "WEBSOCKET";
	prepare_request_body_maps(event_request);
	event_request.resources.is_websocket = true;
	event_request.resources.websocket_connection_id = job["connection_id"].to_string();
	event_request.resources.websocket_scope = job["scope"].to_string();
	event_request.resources.websocket_opcode = (u8)job["opcode"].to_u64();
	event_request.resources.websocket_is_binary = job["is_binary"].to_bool();
	event_request.resources.websocket_is_text = job["is_text"].to_bool();
	event_request.resources.websocket_dispatch_capture = true;
	job["scope_connections"].each([&] (DTree item, String) {
		event_request.resources.websocket_scope_connection_ids.push_back(item.to_string());
	});
	event_request.connection = job["connection_state"];
	event_request.stats.time_init = time_precise();
	event_request.stats.time_start = event_request.stats.time_init;
	event_request.random_index = 0;
	event_request.random_seed = request_seed_from_time(event_request.stats.time_start);
	event_request.response_code = "WEBSOCKET";
	event_request.header["Content-Type"] = server_state.config["CONTENT_TYPE"];
	event_request.in = message;

	event_request.params["WS_MESSAGE"] = message;
	event_request.params["WS_CONNECTION_ID"] = event_request.resources.websocket_connection_id;
	event_request.params["WS_SCOPE"] = event_request.resources.websocket_scope;
	event_request.params["WS_CONNECTION_COUNT"] = (f64)event_request.resources.websocket_scope_connection_ids.size();
	event_request.params["WS_OPCODE"] = (f64)event_request.resources.websocket_opcode;
	event_request.params["WS_MESSAGE_TYPE"] = (event_request.resources.websocket_is_binary ? "BINARY" : "TEXT");
	event_request.params["WS_DOCUMENT_URI"] = first(
		event_request.params["DOCUMENT_URI"],
		event_request.params["REQUEST_URI"]
	);

	return(event_request);
}

bool websocket_exec_send_response(int fd, DTree response)
{
	String encoded = json_encode(response) + "\n";
	size_t offset = 0;
	while(offset < encoded.length())
	{
		ssize_t write_result = write(fd, encoded.data() + offset, encoded.length() - offset);
		if(write_result < 0)
		{
			if(errno == EINTR)
				continue;
			return(false);
		}
		offset += (size_t)write_result;
	}
	return(true);
}

void websocket_exec_process_job_line(int fd, String line)
{
	line = trim(line);
	if(line == "")
		return;

	DTree job = json_decode(line);
	if(job["type"].to_string() != "dispatch")
		return;

	bool decoded = false;
	String message = base64_decode(job["message_b64"].to_string(), decoded);
	if(!decoded)
	{
		printf("(!) invalid websocket IPC payload for %s\n", job["connection_id"].to_string().c_str());
		return;
	}

	Request event_request = websocket_exec_build_event_request(job, message);
	Request* previous_context = set_active_request(event_request);
	server_state.request_count += 1;

	compiler_invoke_websocket(&event_request, event_request.params["SCRIPT_FILENAME"]);

	if(event_request.session_id.length() > 0)
		save_session_data(event_request.session_id, event_request.session);
	cleanup_mysql_connections();
	cleanup_sqlite_connections();

	DTree response;
	response["type"] = "result";
	response["connection_id"] = event_request.resources.websocket_connection_id;
	response["connection_state"] = event_request.connection;
	response["commands"] = event_request.resources.websocket_dispatch_commands;

	restore_active_request(previous_context);
	if(!websocket_exec_send_response(fd, response))
		exit(1);
}

void websocket_exec_child_loop(int fd)
{
	Request background_context;
	my_pid = getpid();
	context = &background_context;
	close_inherited_server_sockets();
	signal(SIGSEGV, on_segfault);
	signal(SIGABRT, on_segfault);
	signal(SIGBUS, on_segfault);
	signal(SIGILL, on_segfault);
	signal(SIGFPE, on_segfault);
	signal(SIGPIPE, SIG_IGN);
	setpriority(PRIO_PROCESS, 0, 5);

	String read_buffer;
	char buffer[4096];
	for(;;)
	{
		ssize_t read_result = read(fd, buffer, sizeof(buffer));
		if(read_result == 0)
			exit(0);
		if(read_result < 0)
		{
			if(errno == EINTR)
				continue;
			exit(1);
		}

		read_buffer.append(buffer, read_result);
		for(;;)
		{
			size_t line_end = read_buffer.find('\n');
			if(line_end == String::npos)
				break;
			String line = read_buffer.substr(0, line_end);
			read_buffer.erase(0, line_end + 1);
			websocket_exec_process_job_line(fd, line);
		}
	}
}

bool websocket_exec_alive()
{
	return(websocket_exec_pid > 0 && task_kill(websocket_exec_pid, 0) == 0 && websocket_exec_fd != -1);
}

void ensure_websocket_executor()
{
	if(!websocket_exec_enabled_for_process())
		return;
	if(websocket_exec_alive())
		return;

	if(websocket_exec_pid > 0 || websocket_exec_fd != -1)
	{
		websocket_exec_fail_inflight_job();
		websocket_exec_clear_ipc_state();
		websocket_exec_pid = 0;
	}

	int sockets[2] = {-1, -1};
	if(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) != 0)
	{
		perror("socketpair");
		return;
	}

	pid_t p = fork();
	if(p < 0)
	{
		perror("fork");
		close(sockets[0]);
		close(sockets[1]);
		return;
	}
	if(p == 0)
	{
		parent_pid = getppid();
		file_release_process_locks("websocket executor fork");
		prctl(PR_SET_PDEATHSIG, SIGHUP);
		close(sockets[0]);
		websocket_exec_child_loop(sockets[1]);
		exit(0);
	}

	close(sockets[1]);
	if(!websocket_ipc_set_nonblocking(sockets[0]))
	{
		printf("(!) failed to set websocket executor socket nonblocking\n");
		close(sockets[0]);
		return;
	}

	websocket_exec_fd = sockets[0];
	websocket_exec_pid = p;
	printf("(P) websocket executor spawned: PID %i\n", p);
}

void websocket_exec_tick()
{
	if(!websocket_exec_enabled_for_process())
		return;
	if(!websocket_exec_alive())
	{
		websocket_exec_fail_inflight_job();
		websocket_exec_clear_ipc_state();
		websocket_exec_pid = 0;
		ensure_websocket_executor();
	}
	websocket_exec_read_results();
	websocket_exec_flush_queue();
}

}

String current_ws_scope()
{
	if(!context)
		return("");
	return(first(
		context->resources.websocket_scope,
		context->params["SCRIPT_FILENAME"]
	));
}

String normalize_ws_scope(String scope)
{
	if(scope == "")
		return(current_ws_scope());
	if(scope[0] == '/')
		return(scope);
	return(expand_path(scope, cwd_get()));
}

String ws_connection_id()
{
	if(!context)
		return("");
	return(context->resources.websocket_connection_id);
}

String ws_message()
{
	if(!context)
		return("");
	return(context->in);
}

String ws_scope()
{
	return(current_ws_scope());
}

u8 ws_opcode()
{
	if(!context)
		return(0);
	return(context->resources.websocket_opcode);
}

bool ws_is_binary()
{
	if(!context)
		return(false);
	return(context->resources.websocket_is_binary);
}

StringList ws_connections(String scope)
{
	if(context && context->resources.websocket_dispatch_capture)
	{
		String normalized_scope = normalize_ws_scope(scope);
		if(normalized_scope == context->resources.websocket_scope)
			return(context->resources.websocket_scope_connection_ids);
		return(StringList());
	}
	return(server.websocket_connection_ids(normalize_ws_scope(scope)));
}

u64 ws_connection_count(String scope)
{
	return(ws_connections(scope).size());
}

bool ws_send(String message, bool binary, String scope)
{
	String normalized_scope = normalize_ws_scope(scope);
	if(context && context->resources.websocket_dispatch_capture)
	{
		DTree command = websocket_exec_make_message_command("broadcast", message, binary);
		command["scope"] = normalized_scope;
		websocket_exec_append_command(command);
		return(true);
	}
	return(server.websocket_broadcast(normalized_scope, message, binary) > 0);
}

bool ws_send_to(String connection_id, String message, bool binary)
{
	if(context && context->resources.websocket_dispatch_capture)
	{
		DTree command = websocket_exec_make_message_command("send_to", message, binary);
		command["connection_id"] = connection_id;
		websocket_exec_append_command(command);
		return(true);
	}
	return(server.websocket_send_to(connection_id, message, binary));
}

bool ws_close(String connection_id)
{
	if(connection_id == "")
		connection_id = ws_connection_id();
	if(connection_id == "")
		return(false);
	if(context && context->resources.websocket_dispatch_capture)
	{
		websocket_exec_append_command(websocket_exec_make_close_command(connection_id));
		return(true);
	}
	return(server.websocket_close(connection_id));
}

bool cli_path_is_safe(String command)
{
	for(auto& segment : split(command, "/"))
	{
		if(segment == "..")
			return(false);
	}
	return(true);
}

String cli_resolve_unit_path(Request& request, String command, String& document_root)
{
	document_root = first(request.params["DOCUMENT_ROOT"], cwd_get());
	if(document_root.length() > 1 && document_root[document_root.length()-1] == '/')
		document_root.resize(document_root.length()-1);

	String script_filename = document_root + command;
	if(file_exists(script_filename))
		return(script_filename);

	String site_filename = document_root + "/site" + command;
	if(file_exists(site_filename))
	{
		document_root = document_root + "/site";
		return(site_filename);
	}

	return("");
}

u64 request_seed_from_time(f64 time_value)
{
	u64 bits = 0;
	static_assert(sizeof(bits) == sizeof(time_value));
	memcpy(&bits, &time_value, sizeof(bits));
	return(gen_noise64(bits));
}

void prepare_request_body_maps(Request& request)
{
	String route_path;
	request.get = parse_query(request.params["QUERY_STRING"], &route_path);
	request_populate_context_params_from_route(request, route_path);
	if(request.params["HTTP_COOKIE"].length() > 0)
		request.cookies = parse_cookies(request.params["HTTP_COOKIE"]);

	String ct_info = request.params["CONTENT_TYPE"];
	String ct_type = nibble(ct_info, ";");
	if(request.params["REQUEST_METHOD"] == "POST")
	{
		if(ct_type == "multipart/form-data")
		{
			nibble("boundary=", ct_info);
			request.post = parse_multipart(request.in, String("--")+ct_info, request.uploaded_files);
		}
		else
		{
			request.post = parse_query(request.in);
		}
	}
}

int handle_cli_complete(FastCGIRequest& request)
{
	Request* previous_context = set_active_request(request);
	server_state.request_count += 1;
	request.server = &server_state;
	request.resources.is_cli = true;
	request.params["UCE_CLI"] = "1";
	request.stats.time_start = time_precise();
	request.header["Content-Type"] = "text/plain; charset=utf-8";
	request.random_index = 0;
	request.random_seed = request_seed_from_time(request.stats.time_start);
	request.ob_start();

	String method = trim(request.params["REQUEST_METHOD"]);
	String command = trim(first(request.params["DOCUMENT_URI"], request.params["REQUEST_URI"]));

	try
	{
		if(method != "GET" && method != "POST")
		{
			request.set_status(405, "Method Not Allowed");
			request.header["Allow"] = "GET, POST";
			print("UCE CLI socket accepts GET and POST commands only\n");
		}
		else if(command == "/" || command == "/help")
		{
			print("UCE CLI command socket\n\nAvailable test hooks:\n  GET /ping\n  GET /status\n");
		}
		else if(command == "/ping" || command == "/test")
		{
			print("uce-cli: ok\n");
		}
		else if(command == "/status")
		{
			print("pid=", std::to_string(getpid()), "\nclients=", std::to_string(server.client_sockets.size()), "\n");
		}
		else if(command.length() >= 4 && command.substr(command.length() - 4) == ".uce")
		{
			if(!cli_path_is_safe(command))
			{
				request.set_status(400, "Bad Request");
				print("invalid UCE CLI unit path\n");
			}
			else
			{
				String document_root;
				String script_filename = cli_resolve_unit_path(request, command, document_root);
				if(script_filename == "")
				{
					request.set_status(404, "Not Found");
					print("UCE CLI unit not found: ", command, "\n");
				}
				else
				{
					request.params["DOCUMENT_ROOT"] = document_root;
					request.params["SCRIPT_FILENAME"] = script_filename;
					prepare_request_body_maps(request);
					request.props = DTree();
					compiler_invoke_cli(&request, script_filename);
				}
			}
		}
		else
		{
			request.set_status(404, "Not Found");
			print("unknown UCE CLI command: ", command, "\n");
		}
	}
	catch(const std::exception& e)
	{
		render_request_failure(request, "uncaught exception during CLI request", e.what(), capture_backtrace_string(32, 1), 500);
	}
	catch(...)
	{
		render_request_failure(request, "unknown uncaught exception during CLI request", "", capture_backtrace_string(32, 1), 500);
	}

	for(auto &f : request.uploaded_files)
		file_unlink(f.tmp_name);
	cleanup_mysql_connections();
	cleanup_sqlite_connections();
	restore_active_request(previous_context);
	return(request.flags.status);
}

int handle_request(FastCGIRequest& request) {
    // This is always the first event to occur.  It occurs when the
    // server receives all parameters.  There may be more data coming on the
    // standard input stream.
	if (request.params.count("REQUEST_URI"))
        return 0;  // OK, continue processing
    else
        return 1;  // stop processing and return error code
}

int handle_data(FastCGIRequest& request) {
    // Request bodies are accumulated by the FastCGI transport and parsed once
    // the input stream is closed in handle_complete().
    return 0;
}

int handle_complete(FastCGIRequest& request) {
    // The event handler can also be a class member function. This
    // event occurs when the parameters and standard input streams are
    // both closed, and thus the request is complete.
	// printf("(i) request handle\n");

	Request* previous_context = set_active_request(request);
	server_state.request_count += 1;
	request.server = &server_state;
	request.stats.time_start = time_precise();
	//request.stats.mem_alloc = 0;
	//request.stats.mem_high = 0;
    request.header["Content-Type"] = (request.resources.is_cli ? "text/plain; charset=utf-8" : context->server->config["CONTENT_TYPE"]);
	request.random_index = 0;
	request.random_seed = request_seed_from_time(request.stats.time_start);
	request.ob_start();
	request_fault_request = &request;
	request_fault_active = 1;
	request_fault_signal = 0;
	request_fault_trace = "";
	install_request_fault_handlers();

	String failure_title = "";
	String failure_details = "";
	String failure_trace = "";

	if(sigsetjmp(request_fault_jmp, 1) != 0)
	{
		failure_title = "fatal signal during request";
		failure_details = "worker recovered before closing the upstream connection";
		failure_trace = request_fault_trace;
	}
	else
	{
		try
		{
			prepare_request_body_maps(request);
			request.props = DTree();
			if(request.resources.is_cli)
				compiler_invoke_cli(&request, request.params["SCRIPT_FILENAME"]);
			else if(request.params["UCE_SERVE_HTTP"] == "1")
				compiler_invoke_serve_http(&request, request.params["SCRIPT_FILENAME"], request.params["UCE_SERVE_HTTP_FUNCTION"]);
			else
				compiler_invoke(&request, request.params["SCRIPT_FILENAME"]);
		}
		catch(const std::exception& e)
		{
			failure_title = "uncaught exception during request";
			failure_details = e.what();
			failure_trace = capture_backtrace_string(32, 1);
		}
		catch(...)
		{
			failure_title = "unknown uncaught exception during request";
			failure_trace = capture_backtrace_string(32, 1);
		}
	}

	request_fault_active = 0;
	request_fault_request = 0;
	restore_request_fault_handlers();

	if(failure_title != "")
		render_request_failure(request, failure_title, failure_details, failure_trace, 500);

	for( auto &f : request.uploaded_files)
	{
		file_unlink(f.tmp_name);
	}

	if(failure_title == "" && request.session_id.length() > 0)
		save_session_data(request.session_id, request.session);

	cleanup_mysql_connections();
	cleanup_sqlite_connections();
	restore_active_request(previous_context);

    return request.flags.status;
}

int handle_websocket_message(FastCGIRequest& request, const String& message, u8 opcode)
{
	ensure_websocket_executor();
	if(!websocket_exec_alive())
	{
		printf("(!) websocket executor unavailable for %s\n", request.resources.websocket_connection_id.c_str());
		server.websocket_close(request.resources.websocket_connection_id, 1011, "websocket handler unavailable");
		return(0);
	}

	DTree job;
	job["type"] = "dispatch";
	job["connection_id"] = request.resources.websocket_connection_id;
	job["scope"] = request.resources.websocket_scope;
	job["opcode"] = (f64)opcode;
	job["is_binary"].set_bool(request.resources.websocket_is_binary);
	job["is_text"].set_bool(request.resources.websocket_is_text);
	job["message_b64"] = base64_encode(message);
	if(request.resources.websocket_connection_state)
		job["connection_state"] = *request.resources.websocket_connection_state;

	for(auto& item : request.params)
		job["params"][item.first] = item.second;

	for(auto& connection_id : websocket_exec_snapshot_connections(request.resources.websocket_scope))
	{
		DTree snapshot_item;
		snapshot_item = connection_id;
		job["scope_connections"].push(snapshot_item);
	}

	websocket_exec_queue_job(job);
	return 0;
}

volatile bool termination_signal_received = false;

void on_terminate(int sig)
{
	if(termination_signal_received)
		return;
	termination_signal_received = true;
	if(getpid() != parent_pid)
		exit(1);
	printf("Terminating... PID %i:%i\n", getpid(), parent_pid);
	server.shutdown();
	exit(1);
}

void clear_shared_unit_cache(ServerState& state)
{
	for(auto& it : state.units)
		delete it.second;
	state.units.clear();
}

void close_inherited_server_sockets()
{
	for(auto socket_handle : server.server_sockets)
		close(socket_handle);
	server.server_sockets.clear();
	server.server_socket_types.clear();
}

void install_process_fault_handlers()
{
	signal(SIGSEGV, on_segfault);
	signal(SIGABRT, on_segfault);
	signal(SIGBUS, on_segfault);
	signal(SIGILL, on_segfault);
	signal(SIGFPE, on_segfault);
	signal(SIGPIPE, SIG_IGN);
}

String custom_server_safe_key(String key)
{
	return(runtime_safe_key(key, "server key"));
}

String custom_server_config_file(String key)
{
	return(path_join(server_state.config["BIN_DIRECTORY"], "server-" + custom_server_safe_key(key) + ".cfg"));
}

String custom_server_task_key(String key)
{
	return("server:" + custom_server_safe_key(key));
}

String custom_server_config_encode(String key, String type, String bind, String file_name, String function_name)
{
	DTree config;
	config["key"] = key;
	config["type"] = type;
	config["bind"] = bind;
	config["file"] = file_name;
	config["function"] = function_name;
	return(json_encode(config));
}

StringMap custom_server_config_decode(String content)
{
	StringMap result;
	content = trim(content);
	if(content == "")
		return(result);

	if(content[0] == '{')
		return(json_decode(content).to_stringmap());

	// Compatibility for early newline/key=value config files from development.
	for(auto line : split(content, "\n"))
	{
		line = trim(line);
		if(line == "")
			continue;
		String key = trim(nibble(line, "="));
		if(key != "")
			result[key] = json_decode(line).to_string();
	}
	return(result);
}

bool custom_server_is_numeric_port(String value)
{
	value = trim(value);
	if(value == "")
		return(false);
	for(char c : value)
	{
		if(c < '0' || c > '9')
			return(false);
	}
	return(true);
}

u64 custom_server_registry_count()
{
	u64 count = 0;
	for(auto file_name : ls(server_state.config["BIN_DIRECTORY"]))
	{
		if(str_starts_with(file_name, "server-") && str_ends_with(file_name, ".cfg"))
			count++;
	}
	return(count);
}

bool custom_server_wait_for_stop(String task_key, f64 timeout_seconds = 2.0)
{
	f64 deadline = time_precise() + timeout_seconds;
	while(time_precise() < deadline)
	{
		if(task_pid(task_key) == 0)
			return(true);
		usleep(50000);
	}
	return(task_pid(task_key) == 0);
}

int custom_server_bind_http(FastCGIServer& dispatcher, String bind)
{
	bind = trim(bind);
	if(custom_server_is_numeric_port(bind))
	{
		u64 port = int_val(bind);
		u64 min_port = config_map_u64(server_state.config, "CUSTOM_SERVER_MIN_PORT", 1024);
		u64 max_port = config_map_u64(server_state.config, "CUSTOM_SERVER_MAX_PORT", 65535);
		if(port < min_port || port > max_port)
			throw std::runtime_error("server_start_http(): TCP port is outside configured custom server range");
		String bind_address = config_map_bool(server_state.config, "CUSTOM_SERVER_ALLOW_PUBLIC_BIND", false) ? "0.0.0.0" : "127.0.0.1";
		return(dispatcher.listen_http((unsigned)port, bind_address));
	}
	String socket_prefix = first(server_state.config["CUSTOM_SERVER_UNIX_SOCKET_PREFIX"], "/tmp/uce/custom-servers/");
	if(!str_starts_with(bind, socket_prefix))
		throw std::runtime_error("server_start_http(): Unix socket path is outside configured custom server prefix");
	int socket_handle = dispatcher.listen(bind);
	dispatcher.server_socket_types[socket_handle] = 'H';
	return(socket_handle);
}

StringMap custom_server_read_config(String key)
{
	String config_file = custom_server_config_file(key);
	if(!file_exists(config_file))
		return(StringMap());
	return(custom_server_config_decode(file_get_contents(config_file)));
}

static String custom_server_dispatcher_key = "";

int custom_server_http_complete(FastCGIRequest& request)
{
	String key = first(request.params["UCE_SERVER_KEY"], custom_server_dispatcher_key);
	StringMap cfg = custom_server_read_config(key);
	if(cfg["type"] != "http" || cfg["file"] == "")
	{
		request.set_status(503, "Service Unavailable");
		request.header["Content-Type"] = "text/plain; charset=utf-8";
		request.ob_start();
		(*request.ob) << "custom HTTP server is not configured\n";
		return(request.flags.status);
	}

	request.params["UCE_SERVE_HTTP"] = "1";
	request.params["UCE_SERVE_HTTP_KEY"] = key;
	request.params["UCE_SERVE_HTTP_BIND"] = cfg["bind"];
	request.params["UCE_SERVE_HTTP_FUNCTION"] = cfg["function"];
	request.params["SCRIPT_FILENAME"] = cfg["file"];
	u64 timeout = config_map_u64(server_state.config, "CUSTOM_SERVER_HANDLER_TIMEOUT_SECONDS", 30);
	if(timeout > 0)
		alarm(timeout);
	int status = handle_complete(request);
	if(timeout > 0)
		alarm(0);
	return(status);
}

void custom_server_http_dispatcher_loop(String key)
{
	my_pid = getpid();
	custom_server_dispatcher_key = key;
	close_inherited_server_sockets();
	install_process_fault_handlers();

	StringMap cfg = custom_server_read_config(key);
	if(cfg["type"] != "http" || cfg["bind"] == "")
	{
		fprintf(stderr, "server_start_http(): missing config for key '%s'\n", key.c_str());
		exit(1);
	}

	try
	{
		FastCGIServer dispatcher;
		custom_server_bind_http(dispatcher, cfg["bind"]);
		dispatcher.calls_until_termination = -1;
		dispatcher.resolve_http_script_filename = false;
		dispatcher.on_complete = &custom_server_http_complete;
		for(;;)
			dispatcher.process(-1);
	}
	catch(const std::exception& e)
	{
		fprintf(stderr, "server_start_http(): dispatcher for key '%s' stopped: %s\n", key.c_str(), e.what());
		exit(1);
	}
	catch(...)
	{
		fprintf(stderr, "server_start_http(): dispatcher for key '%s' stopped with unknown error\n", key.c_str());
		exit(1);
	}
}

pid_t server_start_http(String key, String socket_fn_or_port, String call_uce_filename, String call_function)
{
	key = trim(key);
	socket_fn_or_port = trim(socket_fn_or_port);
	call_uce_filename = trim(call_uce_filename);
	call_function = trim(call_function);
	if(key == "")
		throw std::runtime_error("server_start_http(): key cannot be empty");
	if(socket_fn_or_port == "")
		throw std::runtime_error("server_start_http(): socket_fn_or_port cannot be empty");
	if(call_uce_filename == "")
		throw std::runtime_error("server_start_http(): call_uce_filename cannot be empty");
	if(call_uce_filename[0] != '/')
		call_uce_filename = expand_path(call_uce_filename, cwd_get());
	String allowed_root = first(server_state.config["CUSTOM_SERVER_UCE_ROOT"], path_join(server_state.config["COMPILER_SYS_PATH"], server_state.config["SITE_DIRECTORY"]));
	String real_call_uce_filename = path_real(call_uce_filename);
	if(real_call_uce_filename == "" || !path_is_within(real_call_uce_filename, allowed_root))
		throw std::runtime_error("server_start_http(): call_uce_filename is outside configured custom server UCE root");
	call_uce_filename = real_call_uce_filename;

	String config_file = custom_server_config_file(key);
	StringMap previous_config;
	if(file_exists(config_file))
		previous_config = custom_server_config_decode(file_get_contents(config_file));
	String new_config = custom_server_config_encode(key, "http", socket_fn_or_port, call_uce_filename, call_function);
	String task_key = custom_server_task_key(key);
	u64 max_servers = config_map_u64(server_state.config, "CUSTOM_SERVER_MAX_SERVERS", 16);
	if(!file_exists(config_file) && max_servers > 0 && custom_server_registry_count() >= max_servers)
		throw std::runtime_error("server_start_http(): custom server quota exceeded");
	pid_t existing_pid = task_pid(task_key);
	if(existing_pid != 0 && previous_config["bind"] != "" && previous_config["bind"] != socket_fn_or_port)
	{
		task_kill(existing_pid, SIGTERM);
		custom_server_wait_for_stop(task_key);
		existing_pid = 0;
	}
	if(!file_put_contents(config_file, new_config))
		throw std::runtime_error("server_start_http(): could not write server config");
	if(existing_pid != 0)
		return(existing_pid);
	return(task(task_key, [key]() {
		custom_server_http_dispatcher_loop(key);
	}, 0));
}

bool server_stop(String key)
{
	key = trim(key);
	if(key == "")
		return(false);
	String task_key = custom_server_task_key(key);
	pid_t pid = task_pid(task_key);
	file_unlink(custom_server_config_file(key));
	if(pid == 0)
		return(true);
	if(task_kill(pid, SIGTERM) != 0)
		return(false);
	return(custom_server_wait_for_stop(task_key));
}

bool proactive_compile_queue_has(StringList& queue, String file_name)
{
	return(std::find(queue.begin(), queue.end(), file_name) != queue.end());
}

void proactive_compile_queue_push(StringList& queue, String file_name)
{
	if(file_name == "" || proactive_compile_queue_has(queue, file_name))
		return;
	queue.push_back(file_name);
}

void run_proactive_compiler()
{
	Request background_context;
	StringList compile_queue;
	background_context.server = &server_state;
	set_active_request(background_context);
	if(!config_map_bool(server_state.config, "PROACTIVE_COMPILE_ENABLED", true))
		return;
	f64 check_interval = float_val(server_state.config["PROACTIVE_COMPILE_CHECK_INTERVAL"]);
	f64 failure_retry_interval = 0;
	f64 next_scan_at = 0;
	std::map<String, f64> retry_after;
	if(check_interval < 1)
		check_interval = 1;
	failure_retry_interval = std::max(
		check_interval,
		(f64)std::max((s64)10, (s64)int_val(server_state.config["COMPILE_FAILURE_RETRY_SECONDS"]))
	);

	my_pid = getpid();

	close_inherited_server_sockets();
	signal(SIGSEGV, on_segfault);
	signal(SIGABRT, on_segfault);
	signal(SIGBUS, on_segfault);
	signal(SIGILL, on_segfault);
	signal(SIGFPE, on_segfault);
	signal(SIGPIPE, SIG_IGN);
	setpriority(PRIO_PROCESS, 0, 10);

	try
	{
		auto known_units = compiler_list_known_units(&background_context);
		auto site_units = compiler_scan_site_units(&background_context);
		known_units.insert(known_units.end(), site_units.begin(), site_units.end());
		compiler_set_known_units(&background_context, known_units);
	}
	catch(const std::exception& e)
	{
		printf("(!) proactive compiler initial scan failed: %s\n", e.what());
	}
	catch(...)
	{
		printf("(!) proactive compiler initial scan failed: unknown exception\n");
	}
	next_scan_at = time_precise();

	for(;;)
	{
		try
		{
		if(compile_queue.size() == 0 && time_precise() >= next_scan_at)
		{
			auto tracked_units = compiler_list_known_units(&background_context);
			StringList existing_units;

			for(auto& file_name : tracked_units)
			{
				bool source_missing = false;
				auto retry_it = retry_after.find(file_name);
				bool retry_allowed = (retry_it == retry_after.end() || time_precise() >= retry_it->second);
				if(compiler_unit_needs_recompile(&background_context, file_name, &source_missing) && retry_allowed)
					proactive_compile_queue_push(compile_queue, file_name);
				if(source_missing)
				{
					printf("(i) proactive compiler forget removed unit %s\n", file_name.c_str());
					retry_after.erase(file_name);
					continue;
				}
				existing_units.push_back(file_name);
			}

			if(existing_units.size() != tracked_units.size())
				compiler_set_known_units(&background_context, existing_units);

			next_scan_at = time_precise() + check_interval;
		}

		if(compile_queue.size() > 0)
		{
			auto file_name = compile_queue.front();
			compile_queue.erase(compile_queue.begin());
			bool source_missing = false;
			auto retry_it = retry_after.find(file_name);
			if(retry_it != retry_after.end() && time_precise() < retry_it->second)
				continue;
			if(compiler_unit_needs_recompile(&background_context, file_name, &source_missing))
			{
				printf("(i) proactive compile %s\n", file_name.c_str());
				auto su = get_shared_unit(&background_context, file_name, false);
				if(su && su->compiler_messages == "")
					retry_after.erase(file_name);
				else
					retry_after[file_name] = time_precise() + failure_retry_interval;
			}
			else if(source_missing)
			{
				printf("(i) proactive compiler forget removed unit %s\n", file_name.c_str());
				compiler_untrack_known_unit(&background_context, file_name);
				retry_after.erase(file_name);
			}
			else
			{
				retry_after.erase(file_name);
			}
			background_context.session.clear();
			background_context.session_loaded_hash = "";
			clear_shared_unit_cache(server_state);
			usleep(250000);
			continue;
		}

		usleep(250000);
		}
		catch(const std::exception& e)
		{
			printf("(!) proactive compiler scan failed: %s\n", e.what());
			next_scan_at = time_precise() + check_interval;
			usleep(250000);
		}
		catch(...)
		{
			printf("(!) proactive compiler scan failed: unknown exception\n");
			next_scan_at = time_precise() + check_interval;
			usleep(250000);
		}
	}
}

bool proactive_compiler_alive()
{
	return(proactive_compiler_pid > 0 && task_kill(proactive_compiler_pid, 0) == 0);
}

void ensure_proactive_compiler()
{
	if(!config_map_bool(server_state.config, "PROACTIVE_COMPILE_ENABLED", true))
		return;
	if(float_val(server_state.config["PROACTIVE_COMPILE_CHECK_INTERVAL"]) <= 0)
		return;

	if(proactive_compiler_alive())
		return;

	pid_t p = fork();
	if(p == 0)
	{
		file_release_process_locks("proactive compiler fork");
		prctl(PR_SET_PDEATHSIG, SIGHUP);
		run_proactive_compiler();
		exit(0);
	}

	proactive_compiler_pid = p;
	printf("(P) proactive compiler spawned: PID %i\n", p);
}

void listen_for_connections()
{
	install_process_fault_handlers();
	if(worker_accepts_http)
	{
		// Keep the dedicated HTTP/WebSocket worker alive. If it ages out like a
		// normal FastCGI worker, nginx can connect to the shared listening socket
		// while no child is actively accepting, which makes `.ws.uce` page loads
		// appear to hang until the parent respawns a replacement worker.
		server.calls_until_termination = -1;
	}
	if(!worker_accepts_http)
		server.close_http_listeners();
	server.on_request = &handle_request;
	server.on_data = &handle_data;
	server.on_complete = &handle_complete;
	server.on_cli_complete = &handle_cli_complete;
	server.on_websocket_message = &handle_websocket_message;
	if(worker_accepts_http)
		ensure_websocket_executor();
	for(;;)
	{
		file_release_process_locks("worker loop cleanup");
		server.process(worker_accepts_http ? 50 : -1);
		if(worker_accepts_http)
			websocket_exec_tick();
	}
}

void init_base_process()
{
	printf("(P) Starting parent server PID:%i\n", getpid());

	server_state.config = make_server_settings();
	server_state.config["COMPILER_SYS_PATH"] = cwd_get();
	printf("Compiler base path: %s\n", server_state.config["COMPILER_SYS_PATH"].c_str());

	server_state.config["COMPILE_SCRIPT"] =
		server_state.config["COMPILER_SYS_PATH"] + "/" + server_state.config["COMPILE_SCRIPT"];

	if(server_state.config["FCGI_PORT"] != "")
		server.listen(int_val(server_state.config["FCGI_PORT"]));

	printf("%s\n", var_dump(server_state.config).c_str());

	if(server_state.config["FCGI_SOCKET_PATH"] != "")
	{
		server.listen(server_state.config["FCGI_SOCKET_PATH"]);
		chmod(server_state.config["FCGI_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
	}

	if(server_state.config["CLI_SOCKET_PATH"] != "")
	{
		server.listen_cli(server_state.config["CLI_SOCKET_PATH"]);
		chmod(server_state.config["CLI_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP);
	}

	if(server_state.config["HTTP_PORT"] != "")
		server.listen_http(int_val(server_state.config["HTTP_PORT"]));

	mkdir(server_state.config["BIN_DIRECTORY"]);
	mkdir(server_state.config["TMP_UPLOAD_PATH"]);
	mkdir(server_state.config["SESSION_PATH"]);

	signal(SIGCHLD, on_child_exit);
	signal(SIGINT, on_terminate);
	signal(SIGPIPE, SIG_IGN);
	srand(time());
}

int main(int argc, char** argv)
{
	init_base_process();
	ensure_proactive_compiler();

	for(;;)
	{
		if(!proactive_compiler_alive())
			proactive_compiler_pid = 0;
		if(!termination_signal_received)
			ensure_proactive_compiler();

		while(workers.size() < int_val(server_state.config["WORKER_COUNT"]))
		{
			if(!termination_signal_received)
			{
				worker_accepts_http = (http_worker_pid == 0 || workers.count(http_worker_pid) == 0);
				pid_t child_pid = spawn_subprocess(listen_for_connections);
				if(child_pid > 0 && worker_accepts_http)
					http_worker_pid = child_pid;
			}
		}
		sleep(1);
	}

	return 0;
}