#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdarg.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>

#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netdb.h>
#ifndef NI_MAXHOST
#	define NI_MAXHOST 1025
#endif
#ifndef NI_MAXSERV
#	define NI_MAXSERV 32
#endif

#include <sys/resource.h> // needs _GNU_SOURCE for RUSAGE_THREAD

#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/event.h>
#include <event2/util.h>

#include <ossp/uuid.h>

#include "common.h"
#include "notify.h"
#include "connections.h"
#include "command.h"
#include "server.h"

/**
 *  A reference-counted data buffer.
 *  Used internally here for sending out one message to multiple connections, without
 *  creating a copy for each connection.
 *  Frees data pointer when the last reference is removed.
 *  A connection thread increases the reference count as it sends the message to each
 *  destination connection, while the libevent thread decrements as it completes each
 *  transmission.
 */
struct rc_data_ {
	pthread_mutex_t mutex;
	size_t reference_count;
	size_t data_len;
	char *data;
};

/**
 *  Allocates, initializes, and returns a new struct rc_data_ entity.
 */
static struct rc_data_ *
rc_data_new_(char *data, size_t len)
{
	struct rc_data_ *rc;
	int r;

	if (!data) {
		NOTIFY_DEBUG("data:%p len:%zu", data, len);
		return NULL;
	}

	rc = malloc(sizeof *rc);
	if (!rc) {
		NOTIFY_ERROR("malloc(%zu): %s", sizeof *rc, strerror(errno));
		return NULL;
	}

	if ( (r = pthread_mutex_init(&rc->mutex, NULL)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_init", strerror(r));
		free(rc);
		return NULL;
	}

	rc->reference_count = 1;
	rc->data_len = len;
	rc->data = data;

	return rc;
}

/**
 *  Increments the reference count of a struct rc_data_ entity.
 */
static void
rc_data_ref_inc_(struct rc_data_ *rc)
{
	int r;

	if ( (r = pthread_mutex_lock(&rc->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return;
	}

	rc->reference_count += 1;

	if ( (r = pthread_mutex_unlock(&rc->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
	}
}

/**
 *  Decrements the reference count of a struct rc_data_ entity
 *  and frees everything if there are no more references.
 */
static void
rc_data_free_(struct rc_data_ *rc)
{
	int r;

	if ( (r = pthread_mutex_lock(&rc->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return;
	}
	rc->reference_count -= 1;
	if (rc->reference_count > 0) {
		if ( (r = pthread_mutex_unlock(&rc->mutex)) ) {
			NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		}
		return;
	}

	free(rc->data);
	rc->data = NULL;
	if ( (r = pthread_mutex_unlock(&rc->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
	}
	if ( (r = pthread_mutex_destroy(&rc->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
	}

	memset(rc, 0, sizeof *rc);

	free(rc);
}

/**
 *  Wrapper for rc_data_free_() of the proper callback type to be used by
 *  evbuffer_add_reference() in connection_multi_send().
 */
static void
rc_cleanup_cb_(const void *data UNUSED, size_t len UNUSED, void *arg)
{
	struct rc_data_ *r = arg;

	rc_data_free_(r);
}

/**
 *  Initializes a struct connections entity.
 */
int
connections_init(struct connections *cs)
{
	int r;

	cs->count = 0;
	TAILQ_INIT(&cs->head);

	if ( (r = pthread_mutex_init(&cs->mutex, NULL)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_init", strerror(r));
		return -1;
	}

	return 0;
}

/**
 * Cleanup a struct connections.
 */
void
connections_fini(struct connections *cs)
{
	struct connection *c1, *c2;
	int r;

#if DEBUG_SILLY
	NOTIFY_DEBUG("cs:%p", cs);
#endif

	/* free any connection entities */
	c1 = TAILQ_FIRST(&cs->head);
	while (c1 != NULL) {
		c2 = TAILQ_NEXT(c1, tailq_entry);
		connection_free(c1);
		c1 = c2;
	}
	TAILQ_INIT(&cs->head);
	cs->count = 0;

	if ( (r = pthread_mutex_destroy(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
	}
}


/**
 *  Inserts a connection at the end of connections list.
 */
void
connections_append(struct connection *c)
{
	struct connections *cs = &c->server->connections;
	int r;

#if DEBUG_SILLY
	NOTIFY_DEBUG("c:%p", c);
#endif

	if ( (r = pthread_mutex_lock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return;
	}

	TAILQ_INSERT_TAIL(&cs->head, c, tailq_entry);
	cs->count += 1;

	if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		return;
	}
}

/**
 *  Removes a connection from its struct connections list.
 */
void
connections_remove(struct connection *c)
{
	struct connections *cs = &c->server->connections;
	int r;

#if DEBUG_SILLY
	NOTIFY_DEBUG("c:%p", c);
#endif

	if ( !c || !cs) {
		NOTIFY_DEBUG("c:%p connections:%p", c, cs);
		return;
	}

	if ( (r = pthread_mutex_lock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return;
	}

	TAILQ_REMOVE(&cs->head, c, tailq_entry);
	cs->count -= 1;

	if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		return;
	}
}

/**
 *  Allocates and returns a null-terminated array of pointers to all connection entities in
 *  connections list, minus the 'c_exclude' connection, if set.
 *  Used as a simple implementation of broadcasting a message.
 *  This increments the reference count of each connection added to the array
 *  so they need to individually be freed before freeing the array.
 */
struct connection **
connections_all_as_array(struct connections *cs, struct connection *c_exclude)
{
	struct connection **all;
	struct connection *c;
	size_t i = 0;
	int r;

	if ( (r = pthread_mutex_lock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return NULL;
	}

	all = calloc(cs->count + 1, sizeof *all);
	if (!all) {
		NOTIFY_ERROR("calloc(%zu, %zu): %s", cs->count + 1, sizeof *all, strerror(errno));
		goto err_unlock;
	}

	TAILQ_FOREACH(c, &cs->head, tailq_entry) {
		if (c != c_exclude
		&&  c->state < CONNECTION_STATE_WANT_CLOSE
		&&  c->state > CONNECTION_STATE_INIT) {
			connection_inc_ref(c);
			all[i++] = c;
		}
	}

	if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		while (i) {
			i--;
			connection_free(c);
		}
		free(all);
		return NULL;
	}

	return all;

err_unlock:
	if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
	}
	return NULL;
}


/**
 * Wrap ev timeout event to re-resolve hostname.
 */
static void
connection_resolve_retry_event_(evutil_socket_t fd UNUSED, short what UNUSED, void *arg)
{
	struct connection *c = arg;

	NOTIFY_DEBUG("retrying hostname resolution");

	c->dns_retry_ev = NULL;
	c->dns_retries += 1;
	connection_resolve_hostname(c);
}

/**
 *  A callback used by connection_init to populate the reverse-lookup of a connection's
 *  client_address.
**/
static void
connection_reverse_dns_cb_(int result, char type, int count, int ttl UNUSED, void *addresses, void *ctx)
{
	struct connection *c = ctx;
	char *old_client_address = c->client_address;

	if (result != DNS_ERR_NONE) {
		NOTIFY_ERROR("Error resolving: %s",
		             evdns_err_to_string(result) );
		c->evdns_request = NULL;

		// TODO: configurable
		// if (server_get_config_integer(c->server, "dns_retries_max")) {
		// if (server_get_config_integer(c->server, "dns_retry_seconds")) {

		if (c->dns_retries < 4) {
			struct timeval _seconds = {30, 0};
			c->dns_retry_ev = event_new(c->server->base, -1, EV_TIMEOUT, connection_resolve_retry_event_, c);
			event_add(c->dns_retry_ev, &_seconds);
		};
		return;
	}

	switch (type) {
		case DNS_PTR:
			if (count < 1) {
				return;
			}
			if (count > 1) {
				NOTIFY_DEBUG("multiple records returned, using first");
			}
			c->client_address = strdup(((char **)addresses)[0]);
			NOTIFY_DEBUG("resolved [%s] as %s for '%s' (%p)", old_client_address, c->client_address, c->name, c);
			break;

		default:
			NOTIFY_DEBUG("reverse lookup of [%s] returned dns type %d", c->client_address, type);
			break;
	}
	c->evdns_request = NULL;

	if (old_client_address) {
		free(old_client_address);
	}
}


/**
 * Perform reverse lookup of connection address.
 */
void
connection_resolve_hostname(struct connection *c)
{
	if (c->evdns_request) {
		NOTIFY_DEBUG("resolution already in progress");
		return;
	}

	/* does libevent's evdns have a getnameinfo style call? */

	switch (c->sa->sa_family) {
	case AF_INET:
		c->evdns_request = evdns_base_resolve_reverse(c->server->evdns_base, &((struct sockaddr_in *)c->sa)->sin_addr, 0, connection_reverse_dns_cb_, c);
		break;

	case AF_INET6:
		c->evdns_request = evdns_base_resolve_reverse_ipv6(c->server->evdns_base, &((struct sockaddr_in6 *)c->sa)->sin6_addr, 0, connection_reverse_dns_cb_, c);
		break;

	default:
		NOTIFY_DEBUG("unhandled address family %u", c->sa->sa_family);
	}
	if (!c->evdns_request) {
		NOTIFY_DEBUG("could not submit PTR lookup request");
	}
}


/**
 *  Populates a connection with initial information and sets reference count.
 */
int
connection_init(struct server *s, struct connection *c, struct bufferevent *bev, struct sockaddr *sock, int socklen, connection_flags_t flags)
{
	char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV];
	int r;
	uuid_rc_t rc;

	if ( (r = pthread_mutex_init(&c->rc_mutex, NULL)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_init", strerror(r));
		goto err;
	}

	if ( (r = pthread_mutex_init(&c->commands_mutex, NULL)) ) {
		NOTIFY_ERROR("%s:%s", "pthead_mutex_init", strerror(r));
		goto err_rc_mutex_destroy;
	}

	rc = uuid_create(&c->uuid);
	if (rc != UUID_RC_OK) {
		NOTIFY_ERROR("%s:%s", "uuid_create", uuid_error(rc));
		goto err_commands_mutex_destroy;
	}

	rc = uuid_make(c->uuid, UUID_MAKE_V1|UUID_MAKE_MC);
	if (rc != UUID_RC_OK) {
		NOTIFY_ERROR("%s:%s", "uuid_make", uuid_error(rc));
		goto err_uuid_destroy;
	}

	c->state = CONNECTION_STATE_INIT;
	c->bev = bev;
	c->flags = flags;
	c->server = s;

	/* render an IP from a sockaddr */
	r = getnameinfo(sock, socklen, hbuf, sizeof hbuf, sbuf, sizeof sbuf, NI_NUMERICHOST | NI_NUMERICSERV);
	if (r) {
		NOTIFY_ERROR("%s:%s", "getnameinfo", gai_strerror(r));
		strncpy(hbuf, "(unknown)", sizeof hbuf);
		strncpy(sbuf, "(?)", sizeof sbuf);
	}
	c->client_address = strdup(hbuf);
	if (!c->client_address) {
		NOTIFY_ERROR("%s:%s", "strdup", strerror(errno));
		goto err_uuid_destroy;
	}

	/* Past this point, errors are non-fatal. */
	c->reference_count = 1;

	/* Now try to resolve a name from client ip, in background. */
	c->sa = sock;
	c->sa_len = socklen;
	c->dns_retries = 0;
	c->dns_retry_ev = NULL;
	connection_resolve_hostname(c);

	/* FIXME: temporary name of connection for POC */
	static int current_connection_number;
	r = snprintf((char *)c->name, sizeof(c->name), "conn %d", current_connection_number++);
	if ((size_t)r >= sizeof c->name) {
		NOTIFY_ERROR("buffer truncated [%s:%d]", __FILE__, __LINE__);
	}

#if DARWIN
	/* Darwin systems <10.12 lack clock_gettime (!) */
	if (gettimeofday(&c->connect_time, NULL)) {
		NOTIFY_ERROR("%s:%s", "gettimeofday", strerror(errno));
	}
	NOTIFY_DEBUG("c:%p [%s] from %s at %.24s", c, c->name, c->client_address, ctime(&c->connect_time.tv_sec));
#else
	if (clock_gettime(CLOCK_REALTIME, &c->connect_timespec)) {
		NOTIFY_ERROR("%s:%s", "clock_gettime", strerror(errno));
	}
	NOTIFY_DEBUG("c:%p [%s] from %s at %.24s", c, c->name, c->client_address, ctime(&c->connect_timespec.tv_sec));
#endif /* DARWIN */

	return 0;

err_uuid_destroy:
	rc = uuid_destroy(c->uuid);
	if (rc != UUID_RC_OK) {
		NOTIFY_ERROR("%s:%s", "uuid_destroy", uuid_error(rc));
	}
err_commands_mutex_destroy:
	if ( (r = pthread_mutex_destroy(&c->commands_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_destroy", strerror(r));
	}
err_rc_mutex_destroy:
	if ( (r = pthread_mutex_destroy(&c->rc_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_destroy", strerror(r));
	}
err:
	return -1;
}


/**
 * Process a connection's commands as long as there are commands to process.
 * FIXME: track how much time has been spend consuming commands, re-queue
 *        processor job when over some threshold considering pending workqueue
 */
static void
connection_process_queue_(void *data, void *context, size_t id)
{
	struct connection *c = data;
	struct server_worker_context *ctx = context;
	struct command *command;
	struct rusage ru_start, ru_end;
	struct timespec ts_start, ts_current;
	int r;

	if (clock_gettime(CLOCK_MONOTONIC, &ts_start) < 0) {
		NOTIFY_DEBUG("%s:%s", "clock_gettime", strerror(errno));
	}
	while (1) {
		if ( (r = pthread_mutex_lock(&c->commands_mutex)) ) {
			NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
			return;
		}

		command = STAILQ_FIRST(&c->commands_head);
		if (command) {
			STAILQ_REMOVE_HEAD(&c->commands_head, stailq_entry);
		}

		if ( (r = pthread_mutex_unlock(&c->commands_mutex)) ) {
			NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
			return;
		}

		if (command == NULL) {
			break;
		}

		if ( (r = getrusage(RUSAGE_THREAD, &ru_start)) ) {
			NOTIFY_ERROR("%s:%s", "getrusage", strerror(r));
		}

		command_parse(c, context, id, command);

		if ( (r = getrusage(RUSAGE_THREAD, &ru_end)) ) {
			NOTIFY_ERROR("%s:%s", "getrusage", strerror(r));
		}

		/* track how much time was spent procesing */
		connection_accounting_increment(c, &ru_start, &ru_end);

		command_free(command);

		if (clock_gettime(CLOCK_MONOTONIC, &ts_current) < 0) {
			NOTIFY_ERROR("%s:%s", "clock_gettime", strerror(errno));
		}

	}
	connection_free(c);

	// Done processing for a connection, force GC cycle to release additional c references.
	lua_gc(ctx->L, LUA_GCCOLLECT);
}


/**
 *  Adds a command to a connection's list of things to process.
 */
int
connection_command_enqueue(struct connection *c, struct command *command)
{
	bool was_empty;
	int r;

	if ( (r = pthread_mutex_lock(&c->commands_mutex)) ) {
		command_free(command);
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return -1;
	}

	was_empty = STAILQ_EMPTY(&c->commands_head) ? true : false;
	if (was_empty == true) {
		STAILQ_INSERT_HEAD(&c->commands_head, command, stailq_entry);
	} else {
		STAILQ_INSERT_TAIL(&c->commands_head, command, stailq_entry);
	}

#if DARWIN
	/* Darwin systems <10.12 lack clock_gettime (!) */
	if (gettimeofday(&c->last_received_time, NULL)) {
		NOTIFY_ERROR("%s:%s", "gettimeofday", strerror(errno));
	}
#else
	if (clock_gettime(CLOCK_REALTIME, &c->last_received_timespec)) {
		NOTIFY_ERROR("%s:%s", "clock_gettime", strerror(errno));
	}
#endif /* DARWIN */

	if ( (r = pthread_mutex_unlock(&c->commands_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		return -1;
	}

	/* if there were already queued commands, a processor should already be scheduled */
	if (was_empty == true) {
		connection_inc_ref(c);
		if (workqueue_add(&c->server->workqueue, connection_process_queue_, c)) {
			NOTIFY_ERROR("%s:%s", "workqueue_add", "failed");
			return -1;
		}
	}

	return 0;
}


/**
 *  Add the utime and stime values to a connection's totals.
 */
void
connection_accounting_increment(struct connection *c, struct rusage *ru_start, struct rusage *ru_end)
{
	struct timeval utime_diff, stime_diff;

	timeval_diff(&utime_diff, &ru_end->ru_utime, &ru_start->ru_utime);
	timeval_diff(&stime_diff, &ru_end->ru_stime, &ru_start->ru_stime);

	timeval_increment(&c->utime, &utime_diff);
	timeval_increment(&c->stime, &stime_diff);
}

/**
 *  Locks the output buffer, for sending multiple
 *  contiguous messages without possible interleaving.
 */
void
connection_lock_output(struct connection *c)
{
	struct evbuffer *output = bufferevent_get_output(c->bev);

	evbuffer_lock(output);
}

/**
 *  Unlocks the output buffer.
 */
void
connection_unlock_output(struct connection *c)
{
	struct evbuffer *output = bufferevent_get_output(c->bev);

	evbuffer_unlock(output);
}

/**
 *  Sends a text string to a connection.
 */
int
connection_printf(struct connection *c, const char *fmt, ...)
{
	struct evbuffer *output = bufferevent_get_output(c->bev);
	va_list ap;
	int r;

	va_start(ap, fmt);
	r = evbuffer_add_vprintf(output, fmt, ap);
	va_end(ap);

	return r;
}


/**
 *  Sends a text string to a connection.
 */
int
connection_vprintf(struct connection *c, const char *fmt, va_list ap)
{
	struct evbuffer *output = bufferevent_get_output(c->bev);
	int r;

	r = evbuffer_add_vprintf(output, fmt, ap);

	return r;
}


/**
 *  Send one data buffer to multiple connections.
 *  Data buffer must be malloced, and will be freed when last connection
 *  has finished with it.
 *  Returns number of items sent, or negative on error.
 */
int
connection_multi_send(struct connection **clist, char *data, size_t len)
{
	struct connection **clist_iter;
	struct rc_data_ *rc;
	int r = 0;

	if (!clist) {
		NOTIFY_DEBUG("null clist [%s:%d]", __FILE__, __LINE__);
		return -1;
	}

	/* Track our data so it can be freed once all connections are done with it. */
	rc = rc_data_new_(data, len);
	if (rc == NULL) {
		NOTIFY_ERROR("rc_data_new_('%s', %zu) failed", data, len);
		return -1;
	}

	clist_iter = clist;
	while (*clist_iter) {
		struct evbuffer *output = bufferevent_get_output((*clist_iter)->bev);

		/* FIXME: if connection is still valid... */

		/* FIXME: clang claims this is a use-after-free if evbuffer_add_reference fails, but I don't yet believe it. */
		rc_data_ref_inc_(rc);
		if (evbuffer_add_reference(output, rc->data, rc->data_len, rc_cleanup_cb_, rc)) {
			NOTIFY_ERROR("%s:%s", "evbuffer_add_reference", "failed");
			rc_data_free_(rc);
		} else {
			r++;
		}
		clist_iter++;
	}

	/* FIXME: clang also claims this is use-after-free, same as above. */
	rc_data_free_(rc);

	return r;
}


int
connection_printf_broadcast(struct connection *sender, bool exclude_sender, const char *fmt, ...)
{
	struct connection **clist;
	char *message = NULL;
	ssize_t message_len = 0;
	va_list ap;
	int r;

	va_start(ap, fmt);
	message_len = vsnprintf(message, message_len, fmt, ap);
	va_end(ap);
	if (message_len < 0) {
		NOTIFY_ERROR("%s:%s (%zd)", "vsnprintf", strerror(errno), message_len);
		return -1;
	}
	message_len += 1;
	message = malloc(message_len);
	if (!message) {
		NOTIFY_ERROR("%s(%zd):%s", "malloc", message_len, strerror(errno));
		return -1;
	}
	va_start(ap, fmt);
	message_len = vsnprintf(message, message_len, fmt, ap);
	va_end(ap);
	if (message_len < 0) {
		NOTIFY_ERROR("%s:%s (%zd)", "vsnprintf", strerror(errno), message_len);
		free(message);
		return -1;
	}

#ifdef DEBUG_SILLY
		NOTIFY_DEBUG("message_len:%zu message:%s", message_len, message);
#endif

	clist = connections_all_as_array(&sender->server->connections, exclude_sender == true ? sender : NULL);
	r = connection_multi_send(clist, message, message_len);
	if (r < 0) {
		NOTIFY_ERROR("%s:%s", "connection_multi_send", "failed");
	}
	for (struct connection **clist_iter = clist; *clist_iter; clist_iter++) {
		connection_free(*clist_iter);
	}
	free(clist);

	return r;
}

/**
 *
 */
void
connection_inc_ref(struct connection *c)
{
	int r;

	if ( (r = pthread_mutex_lock(&c->rc_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return;
	}

	c->reference_count += 1;

#if DEBUG_SILLY
	NOTIFY_DEBUG("c:%p new reference_count:%zu", c, c->reference_count);
#endif

	if ( (r = pthread_mutex_unlock(&c->rc_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		return;
	}
}

/**
 *  Frees a connection (and all data owned by a connection) if there are no more references to it.
 */
void
connection_free(struct connection *c)
{
	struct evbuffer *tmp;
	evutil_socket_t fd;
	int r;

	if ( (r = pthread_mutex_lock(&c->rc_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
		return;
	}

	c->reference_count -= 1;

#if DEBUG_SILLY
	NOTIFY_DEBUG("c:%p new reference_count:%zu", c, c->reference_count);
#endif

	if (c->reference_count > 0) {
		if ( (r = pthread_mutex_unlock(&c->rc_mutex)) ) {
			NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
		}

		return;
	}

	NOTIFY_DEBUG("c:%p freeing", c);

	connections_remove(c);

	bufferevent_disable(c->bev, EV_READ|EV_WRITE);
	tmp = bufferevent_get_output(c->bev);
	evbuffer_drain(tmp, evbuffer_get_length(tmp));
	tmp = bufferevent_get_input(c->bev);
	evbuffer_drain(tmp, evbuffer_get_length(tmp));

	fd = bufferevent_getfd(c->bev);
	if (fd != -1) {
		EVUTIL_CLOSESOCKET(fd);
		bufferevent_setfd(c->bev, -1);
	}

	if (c->dns_retry_ev) {
		if (event_del(c->dns_retry_ev)) {
			NOTIFY_ERROR("%s:%s", "event_del", strerror(errno));
		}
		c->dns_retry_ev = NULL;
	}

	if (c->evdns_request) {
		evdns_cancel_request(c->server->evdns_base, c->evdns_request);
		c->evdns_request = NULL;
	}

	if (c->client_address) {
		free(c->client_address);
		c->client_address = NULL;
	}

	bufferevent_free(c->bev);

	if (c->uuid) {
		uuid_rc_t rc;

		rc = uuid_destroy(c->uuid);
		if (rc != UUID_RC_OK) {
			NOTIFY_ERROR("%s:%s", "uuid_destroy", uuid_error(rc));
		}
		c->uuid = NULL;
	}

	/* empty out the command queue */
	if ( (r = pthread_mutex_lock(&c->commands_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
	}

	struct command *c1, *c2;
	c1 = STAILQ_FIRST(&c->commands_head);
	while (c1 != NULL) {
		c2 = STAILQ_NEXT(c1, stailq_entry);
		free(c1);
		c1 = c2;
	}
	STAILQ_INIT(&c->commands_head);

	if ( (r = pthread_mutex_unlock(&c->commands_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
	}
	if ( (r = pthread_mutex_destroy(&c->commands_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
	}

	if ( (r = pthread_mutex_unlock(&c->rc_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
	}
	if ( (r = pthread_mutex_destroy(&c->rc_mutex)) ) {
		NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
	}

	memset(c, 0, sizeof *c);

	free(c);
}