//
// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
// Copyright 2019 Devolutions <info@devolutions.net>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt).  A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#include <ctype.h>
#include <stdbool.h>
#include <string.h>

#include "core/nng_impl.h"
#include "supplemental/tls/tls_api.h"

#include "http_api.h"

#include <nng/supplemental/tls/tls.h>

// We insist that individual headers fit in 8K.
// If you need more than that, you need something we can't do.
#define HTTP_BUFSIZE 8192

// types of reads
enum read_flavor {
	HTTP_RD_RAW,
	HTTP_RD_FULL,
	HTTP_RD_REQ,
	HTTP_RD_RES,
	HTTP_RD_CHUNK,
};

enum write_flavor {
	HTTP_WR_RAW,
	HTTP_WR_FULL,
	HTTP_WR_REQ,
	HTTP_WR_RES,
};

#define SET_RD_FLAVOR(aio, f) \
	nni_aio_set_prov_extra(aio, 0, ((void *) (intptr_t)(f)))
#define GET_RD_FLAVOR(aio) (int) ((intptr_t) nni_aio_get_prov_extra(aio, 0))
#define SET_WR_FLAVOR(aio, f) \
	nni_aio_set_prov_extra(aio, 0, ((void *) (intptr_t)(f)))
#define GET_WR_FLAVOR(aio) (int) ((intptr_t) nni_aio_get_prov_extra(aio, 0))

struct nng_http_conn {
	nng_stream *sock;
	void *      ctx;
	bool        closed;
	nni_list    rdq; // high level http read requests
	nni_list    wrq; // high level http write requests

	nni_aio *rd_uaio; // user aio for read
	nni_aio *wr_uaio; // user aio for write
	nni_aio *rd_aio;  // bottom half read operations
	nni_aio *wr_aio;  // bottom half write operations

	nni_mtx mtx;

	uint8_t *rd_buf;
	size_t   rd_get;
	size_t   rd_put;
	size_t   rd_bufsz;
	bool     rd_buffered;
};

void
nni_http_conn_set_ctx(nni_http_conn *conn, void *ctx)
{
	conn->ctx = ctx;
}

void *
nni_http_conn_get_ctx(nni_http_conn *conn)
{
	return (conn->ctx);
}

static void
http_close(nni_http_conn *conn)
{
	// Call with lock held.
	nni_aio *aio;

	if (conn->closed) {
		return;
	}

	conn->closed = true;
	nni_aio_close(conn->wr_aio);
	nni_aio_close(conn->rd_aio);

	if ((aio = conn->rd_uaio) != NULL) {
		conn->rd_uaio = NULL;
		nni_aio_finish_error(aio, NNG_ECLOSED);
	}
	if ((aio = conn->wr_uaio) != NULL) {
		conn->wr_uaio = NULL;
		nni_aio_finish_error(aio, NNG_ECLOSED);
	}

	// Abort all operations except the one in flight.
	while ((aio = nni_list_first(&conn->wrq)) != NULL) {
		nni_aio_list_remove(aio);
		nni_aio_finish_error(aio, NNG_ECLOSED);
	}
	while ((aio = nni_list_first(&conn->rdq)) != NULL) {
		nni_aio_list_remove(aio);
		nni_aio_finish_error(aio, NNG_ECLOSED);
	}

	if (conn->sock != NULL) {
		nng_stream_close(conn->sock);
	}
}

void
nni_http_conn_close(nni_http_conn *conn)
{
	nni_mtx_lock(&conn->mtx);
	http_close(conn);
	nni_mtx_unlock(&conn->mtx);
}

// http_rd_buf attempts to satisfy the read from data in the buffer.
static int
http_rd_buf(nni_http_conn *conn, nni_aio *aio)
{
	size_t   cnt = conn->rd_put - conn->rd_get;
	size_t   n;
	uint8_t *rbuf = conn->rd_buf;
	int      rv;
	bool     raw = false;
	nni_iov *iov;
	unsigned nio;

	rbuf += conn->rd_get;

	switch (GET_RD_FLAVOR(aio)) {
	case HTTP_RD_RAW:
		raw = true; // FALLTHROUGH
	case HTTP_RD_FULL:
		nni_aio_get_iov(aio, &nio, &iov);
		while ((nio != 0) && (cnt != 0)) {
			// Pull up data from the buffer if possible.
			n = iov[0].iov_len;
			if (n > cnt) {
				n = cnt;
			}
			memcpy(iov[0].iov_buf, rbuf, n);
			iov[0].iov_len -= n;
			NNI_INCPTR(iov[0].iov_buf, n);
			conn->rd_get += n;
			rbuf += n;
			nni_aio_bump_count(aio, n);
			cnt -= n;

			if (iov[0].iov_len == 0) {
				nio--;
				iov = &iov[1];
			}
		}

		nni_aio_set_iov(aio, nio, iov);

		if ((nio == 0) || (raw && (nni_aio_count(aio) != 0))) {
			// Finished the read.  (We are finished if we either
			// got *all* the data, or we got *some* data for
			// a raw read.)
			return (0);
		}

		// No more data left in the buffer, so use a physio.
		// (Note that we get here if we either have not completed
		// a full transaction on a FULL read, or were not even able
		// to get *any* data for a partial RAW read.)
		conn->rd_buffered = false;
		nni_aio_set_iov(conn->rd_aio, nio, iov);
		nng_stream_recv(conn->sock, conn->rd_aio);
		return (NNG_EAGAIN);

	case HTTP_RD_REQ:
		rv = nni_http_req_parse(
		    nni_aio_get_prov_extra(aio, 1), rbuf, cnt, &n);
		conn->rd_get += n;
		if (conn->rd_get == conn->rd_put) {
			conn->rd_get = conn->rd_put = 0;
		}
		if (rv == NNG_EAGAIN) {
			nni_iov iov1;
			iov1.iov_buf      = conn->rd_buf + conn->rd_put;
			iov1.iov_len      = conn->rd_bufsz - conn->rd_put;
			conn->rd_buffered = true;
			nni_aio_set_iov(conn->rd_aio, 1, &iov1);
			nng_stream_recv(conn->sock, conn->rd_aio);
		}
		return (rv);

	case HTTP_RD_RES:
		rv = nni_http_res_parse(
		    nni_aio_get_prov_extra(aio, 1), rbuf, cnt, &n);
		conn->rd_get += n;
		if (conn->rd_get == conn->rd_put) {
			conn->rd_get = conn->rd_put = 0;
		}
		if (rv == NNG_EAGAIN) {
			nni_iov iov1;
			iov1.iov_buf      = conn->rd_buf + conn->rd_put;
			iov1.iov_len      = conn->rd_bufsz - conn->rd_put;
			conn->rd_buffered = true;
			nni_aio_set_iov(conn->rd_aio, 1, &iov1);
			nng_stream_recv(conn->sock, conn->rd_aio);
		}
		return (rv);

	case HTTP_RD_CHUNK:
		rv = nni_http_chunks_parse(
		    nni_aio_get_prov_extra(aio, 1), rbuf, cnt, &n);
		conn->rd_get += n;
		if (conn->rd_get == conn->rd_put) {
			conn->rd_get = conn->rd_put = 0;
		}
		if (rv == NNG_EAGAIN) {
			nni_iov iov1;
			iov1.iov_buf      = conn->rd_buf + conn->rd_put;
			iov1.iov_len      = conn->rd_bufsz - conn->rd_put;
			conn->rd_buffered = true;
			nni_aio_set_iov(conn->rd_aio, 1, &iov1);
			nng_stream_recv(conn->sock, conn->rd_aio);
		}
		return (rv);
	}
	return (NNG_EINVAL);
}

static void
http_rd_start(nni_http_conn *conn)
{
	for (;;) {
		nni_aio *aio;
		int      rv;

		if ((aio = conn->rd_uaio) == NULL) {
			if ((aio = nni_list_first(&conn->rdq)) == NULL) {
				// No more stuff waiting for read.
				return;
			}
			nni_list_remove(&conn->rdq, aio);
			conn->rd_uaio = aio;
		}

		if (conn->closed) {
			rv = NNG_ECLOSED;
		} else {
			rv = http_rd_buf(conn, aio);
		}
		switch (rv) {
		case NNG_EAGAIN:
			return;
		case 0:
			conn->rd_uaio = NULL;
			nni_aio_finish(aio, 0, nni_aio_count(aio));
			break;
		default:
			conn->rd_uaio = NULL;
			nni_aio_finish_error(aio, rv);
			http_close(conn);
			break;
		}
	}
}

static void
http_rd_cb(void *arg)
{
	nni_http_conn *conn = arg;
	nni_aio *      aio  = conn->rd_aio;
	nni_aio *      uaio;
	size_t         cnt;
	int            rv;
	unsigned       niov;
	nni_iov *      iov;

	nni_mtx_lock(&conn->mtx);

	if ((rv = nni_aio_result(aio)) != 0) {
		if ((uaio = conn->rd_uaio) != NULL) {
			conn->rd_uaio = NULL;
			nni_aio_finish_error(uaio, rv);
		}
		http_close(conn);
		nni_mtx_unlock(&conn->mtx);
		return;
	}

	cnt = nni_aio_count(aio);

	// If we were reading into the buffer, then advance location(s).
	if (conn->rd_buffered) {
		conn->rd_put += cnt;
		NNI_ASSERT(conn->rd_put <= conn->rd_bufsz);
		http_rd_start(conn);
		nni_mtx_unlock(&conn->mtx);
		return;
	}

	// Otherwise we are completing a USER request, and there should
	// be no data left in the user buffer.
	NNI_ASSERT(conn->rd_get == conn->rd_put);

	if ((uaio = conn->rd_uaio) == NULL) {
		// This indicates that a read request was canceled.  This
		// can occur only when shutting down, really.
		nni_mtx_unlock(&conn->mtx);
		return;
	}

	nni_aio_get_iov(uaio, &niov, &iov);

	while ((niov != 0) && (cnt != 0)) {
		// Pull up data from the buffer if possible.
		size_t n = iov[0].iov_len;
		if (n > cnt) {
			n = cnt;
		}
		iov[0].iov_len -= n;
		NNI_INCPTR(iov[0].iov_buf, n);
		nni_aio_bump_count(uaio, n);
		cnt -= n;

		if (iov[0].iov_len == 0) {
			niov--;
			iov = &iov[1];
		}
	}
	nni_aio_set_iov(uaio, niov, iov);

	// Resubmit the start.  This will attempt to consume data
	// from the read buffer (there won't be any), and then either
	// complete the I/O (for HTTP_RD_RAW, or if there is nothing left),
	// or submit another physio.
	http_rd_start(conn);
	nni_mtx_unlock(&conn->mtx);
}

static void
http_rd_cancel(nni_aio *aio, void *arg, int rv)
{
	nni_http_conn *conn = arg;

	nni_mtx_lock(&conn->mtx);
	if (aio == conn->rd_uaio) {
		conn->rd_uaio = NULL;
		nni_aio_abort(conn->rd_aio, rv);
		nni_aio_finish_error(aio, rv);
	} else if (nni_aio_list_active(aio)) {
		nni_aio_list_remove(aio);
		nni_aio_finish_error(aio, rv);
	}
	nni_mtx_unlock(&conn->mtx);
}

static void
http_rd_submit(nni_http_conn *conn, nni_aio *aio)
{
	int rv;

	if (nni_aio_begin(aio) != 0) {
		return;
	}
	if (conn->closed) {
		nni_aio_finish_error(aio, NNG_ECLOSED);
		return;
	}
	if ((rv = nni_aio_schedule(aio, http_rd_cancel, conn)) != 0) {
		nni_aio_finish_error(aio, rv);
		return;
	}
	nni_list_append(&conn->rdq, aio);
	if (conn->rd_uaio == NULL) {
		http_rd_start(conn);
	}
}

static void
http_wr_start(nni_http_conn *conn)
{
	nni_aio *aio;
	nni_iov *iov;
	unsigned niov;

	if ((aio = conn->wr_uaio) == NULL) {
		if ((aio = nni_list_first(&conn->wrq)) == NULL) {
			// No more stuff waiting for read.
			return;
		}
		nni_list_remove(&conn->wrq, aio);
		conn->wr_uaio = aio;
	}

	nni_aio_get_iov(aio, &niov, &iov);
	nni_aio_set_iov(conn->wr_aio, niov, iov);
	nng_stream_send(conn->sock, conn->wr_aio);
}

static void
http_wr_cb(void *arg)
{
	nni_http_conn *conn = arg;
	nni_aio *      aio  = conn->wr_aio;
	nni_aio *      uaio;
	int            rv;
	size_t         n;

	nni_mtx_lock(&conn->mtx);

	uaio = conn->wr_uaio;

	if ((rv = nni_aio_result(aio)) != 0) {
		// We failed to complete the aio.
		if (uaio != NULL) {
			conn->wr_uaio = NULL;
			nni_aio_finish_error(uaio, rv);
		}
		http_close(conn);
		nni_mtx_unlock(&conn->mtx);
		return;
	}

	if (uaio == NULL) {
		// Write canceled?  This happens pretty much only during
		// shutdown/close, so we don't want to resume writing.
		// The stream is probably corrupted at this point anyway.
		nni_mtx_unlock(&conn->mtx);
		return;
	}

	n = nni_aio_count(aio);
	nni_aio_bump_count(uaio, n);

	if (GET_WR_FLAVOR(uaio) == HTTP_WR_RAW) {
		// For raw data, we just send partial completion
		// notices to the consumer.
		goto done;
	}
	nni_aio_iov_advance(aio, n);
	if (nni_aio_iov_count(aio) > 0) {
		// We have more to transmit - start another and leave
		// (we will get called again when it is done).
		nng_stream_send(conn->sock, aio);
		nni_mtx_unlock(&conn->mtx);
		return;
	}

done:
	conn->wr_uaio = NULL;
	nni_aio_finish(uaio, 0, nni_aio_count(uaio));

	// Start next write if another is ready.
	http_wr_start(conn);

	nni_mtx_unlock(&conn->mtx);
}

static void
http_wr_cancel(nni_aio *aio, void *arg, int rv)
{
	nni_http_conn *conn = arg;

	nni_mtx_lock(&conn->mtx);
	if (aio == conn->wr_uaio) {
		conn->wr_uaio = NULL;
		nni_aio_abort(conn->wr_aio, rv);
		nni_aio_finish_error(aio, rv);
	} else if (nni_aio_list_active(aio)) {
		nni_aio_list_remove(aio);
		nni_aio_finish_error(aio, rv);
	}
	nni_mtx_unlock(&conn->mtx);
}

static void
http_wr_submit(nni_http_conn *conn, nni_aio *aio)
{
	int rv;

	if (nni_aio_begin(aio) != 0) {
		return;
	}
	if (conn->closed) {
		nni_aio_finish_error(aio, NNG_ECLOSED);
		return;
	}
	if ((rv = nni_aio_schedule(aio, http_wr_cancel, conn)) != 0) {
		nni_aio_finish_error(aio, rv);
		return;
	}
	nni_list_append(&conn->wrq, aio);

	if (conn->wr_uaio == NULL) {
		http_wr_start(conn);
	}
}

void
nni_http_read_req(nni_http_conn *conn, nni_http_req *req, nni_aio *aio)
{
	SET_RD_FLAVOR(aio, HTTP_RD_REQ);
	nni_aio_set_prov_extra(aio, 1, req);

	nni_mtx_lock(&conn->mtx);
	http_rd_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_read_res(nni_http_conn *conn, nni_http_res *res, nni_aio *aio)
{
	SET_RD_FLAVOR(aio, HTTP_RD_RES);
	nni_aio_set_prov_extra(aio, 1, res);

	nni_mtx_lock(&conn->mtx);
	http_rd_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_read_chunks(nni_http_conn *conn, nni_http_chunks *cl, nni_aio *aio)
{
	SET_RD_FLAVOR(aio, HTTP_RD_CHUNK);
	nni_aio_set_prov_extra(aio, 1, cl);

	nni_mtx_lock(&conn->mtx);
	http_rd_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_read_full(nni_http_conn *conn, nni_aio *aio)
{
	SET_RD_FLAVOR(aio, HTTP_RD_FULL);
	nni_aio_set_prov_extra(aio, 1, NULL);

	nni_mtx_lock(&conn->mtx);
	http_rd_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_read(nni_http_conn *conn, nni_aio *aio)
{
	SET_RD_FLAVOR(aio, HTTP_RD_RAW);
	nni_aio_set_prov_extra(aio, 1, NULL);

	nni_mtx_lock(&conn->mtx);
	http_rd_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_write_req(nni_http_conn *conn, nni_http_req *req, nni_aio *aio)
{
	int     rv;
	void *  buf;
	size_t  bufsz;
	void *  data;
	size_t  size;
	nni_iov iov[2];
	int     niov;

	if ((rv = nni_http_req_get_buf(req, &buf, &bufsz)) != 0) {
		nni_aio_finish_error(aio, rv);
		return;
	}
	nni_http_req_get_data(req, &data, &size);
	niov           = 1;
	iov[0].iov_len = bufsz;
	iov[0].iov_buf = buf;
	if ((size > 0) && (data != NULL)) {
		niov++;
		iov[1].iov_len = size;
		iov[1].iov_buf = data;
	}
	nni_aio_set_iov(aio, niov, iov);

	SET_WR_FLAVOR(aio, HTTP_WR_REQ);

	nni_mtx_lock(&conn->mtx);
	http_wr_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_write_res(nni_http_conn *conn, nni_http_res *res, nni_aio *aio)
{
	int     rv;
	void *  buf;
	size_t  bufsz;
	void *  data;
	size_t  size;
	nni_iov iov[2];
	int     nio;

	if ((rv = nni_http_res_get_buf(res, &buf, &bufsz)) != 0) {
		nni_aio_finish_error(aio, rv);
		return;
	}
	nni_http_res_get_data(res, &data, &size);
	nio            = 1;
	iov[0].iov_len = bufsz;
	iov[0].iov_buf = buf;
	if ((size > 0) && (data != NULL)) {
		nio++;
		iov[1].iov_len = size;
		iov[1].iov_buf = data;
	}
	nni_aio_set_iov(aio, nio, iov);

	SET_WR_FLAVOR(aio, HTTP_WR_RES);

	nni_mtx_lock(&conn->mtx);
	http_wr_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_write(nni_http_conn *conn, nni_aio *aio)
{
	SET_WR_FLAVOR(aio, HTTP_WR_RAW);

	nni_mtx_lock(&conn->mtx);
	http_wr_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

void
nni_http_write_full(nni_http_conn *conn, nni_aio *aio)
{
	SET_WR_FLAVOR(aio, HTTP_WR_FULL);

	nni_mtx_lock(&conn->mtx);
	http_wr_submit(conn, aio);
	nni_mtx_unlock(&conn->mtx);
}

int
nni_http_conn_getopt(
    nni_http_conn *conn, const char *name, void *buf, size_t *szp, nni_type t)
{
	int rv;
	nni_mtx_lock(&conn->mtx);
	if (conn->closed) {
		rv = NNG_ECLOSED;
	} else {
		rv = nni_stream_get(conn->sock, name, buf, szp, t);
	}
	nni_mtx_unlock(&conn->mtx);
	return (rv);
}

int
nni_http_conn_setopt(nni_http_conn *conn, const char *name, const void *buf,
    size_t sz, nni_type t)
{
	int rv;
	nni_mtx_lock(&conn->mtx);
	if (conn->closed) {
		rv = NNG_ECLOSED;
	} else {
		rv = nni_stream_set(conn->sock, name, buf, sz, t);
	}
	nni_mtx_unlock(&conn->mtx);
	return (rv);
}

void
nni_http_conn_fini(nni_http_conn *conn)
{
	nni_aio_stop(conn->wr_aio);
	nni_aio_stop(conn->rd_aio);

	nni_mtx_lock(&conn->mtx);
	http_close(conn);
	if (conn->sock != NULL) {
		nng_stream_free(conn->sock);
		conn->sock = NULL;
	}
	nni_mtx_unlock(&conn->mtx);

	nni_aio_free(conn->wr_aio);
	nni_aio_free(conn->rd_aio);
	nni_free(conn->rd_buf, conn->rd_bufsz);
	nni_mtx_fini(&conn->mtx);
	NNI_FREE_STRUCT(conn);
}

static int
http_init(nni_http_conn **connp, nng_stream *data)
{
	nni_http_conn *conn;
	int            rv;

	if ((conn = NNI_ALLOC_STRUCT(conn)) == NULL) {
		return (NNG_ENOMEM);
	}
	nni_mtx_init(&conn->mtx);
	nni_aio_list_init(&conn->rdq);
	nni_aio_list_init(&conn->wrq);

	if ((conn->rd_buf = nni_alloc(HTTP_BUFSIZE)) == NULL) {
		nni_http_conn_fini(conn);
		return (NNG_ENOMEM);
	}
	conn->rd_bufsz = HTTP_BUFSIZE;

	if (((rv = nni_aio_alloc(&conn->wr_aio, http_wr_cb, conn)) != 0) ||
	    ((rv = nni_aio_alloc(&conn->rd_aio, http_rd_cb, conn)) != 0)) {
		nni_http_conn_fini(conn);
		return (rv);
	}

	conn->sock = data;

	*connp = conn;

	return (0);
}

int
nni_http_conn_init(nni_http_conn **connp, nng_stream *stream)
{
	int rv;
	if ((rv = http_init(connp, stream)) != 0) {
		nng_stream_free(stream);
	}
	return (rv);
}