/* SPDX-License-Identifier: GPL-2.0 */

/** \file http_recv.c paraslash's http receiver */

#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <sys/un.h>
#include <netdb.h>
#include <lopsub.h>

#include "recv_cmd.lsg.h"
#include "para.h"
#include "error.h"
#include "http.h"
#include "list.h"
#include "sched.h"
#include "buffer_tree.h"
#include "recv.h"
#include "net.h"
#include "string.h"
#include "fd.h"

/**
 * the possible states of a http receiver node
 *
 * \sa \ref receiver_node.
 */
enum http_recv_status {HTTP_CONNECTED, HTTP_SENT_GET_REQUEST, HTTP_STREAMING};

/**
 * Data specific to the http receiver.
 *
 * Each running instance of the http receiver reserves space for one such struct.
 */
struct private_http_recv_data {
	/**
	 * The current status of the http receiver node.
	 *
	 * It gets initialized to \p HTTP_CONNECTED by the open function of the
	 * http receiver.
	 *
	 * \sa \ref receiver::open, \ref receiver_node.
	 */
	enum http_recv_status status;
};

static char *make_request_msg(void)
{
	return make_message("%s1.0\nHost: %s\nUser-Agent: para_recv/%s\n\n\n",
		HTTP_GET_MSG, para_hostname(), paraslash_version());
}

static void http_recv_pre_monitor(struct sched *s, void *context)
{
	struct receiver_node *rn = context;
	struct private_http_recv_data *phd = rn->private_data;

	if (generic_recv_pre_monitor(s, rn) <= 0)
		return;
	if  (phd->status == HTTP_CONNECTED)
		sched_monitor_writefd(rn->fd, s);
	else
		sched_monitor_readfd(rn->fd, s);
}

/*
 * Establish the http connection. If already established, fill the buffer pool
 * area with data read from the socket. In any case, update the state of the
 * connection if necessary.
 */
static int http_recv_post_monitor(struct sched *s, void *context)
{
	struct receiver_node *rn = context;
	struct private_http_recv_data *phd = rn->private_data;
	struct btr_node *btrn = rn->btrn;
	int ret, iovcnt;
	struct iovec iov[2];
	size_t num_bytes;

	ret = task_get_notification(rn->task);
	if (ret < 0)
		goto out;
	ret = btr_node_status(btrn, 0, BTR_NT_ROOT);
	if (ret < 0)
		goto out;
	if (ret == 0)
		return 0;
	if (phd->status == HTTP_CONNECTED) {
		char *rq;
		if (!sched_write_ok(rn->fd, s))
			return 0;
		rq = make_request_msg();
		PARA_INFO_LOG("sending http request\n");
		ret = write_va_buffer(rn->fd, "%s", rq);
		free(rq);
		if (ret < 0)
			goto out;
		phd->status = HTTP_SENT_GET_REQUEST;
		return 0;
	}
	if (phd->status == HTTP_SENT_GET_REQUEST) {
		ret = read_and_compare(rn->fd, HTTP_OK_MSG);
		if (ret < 0) {
			PARA_ERROR_LOG("did not receive HTTP OK message\n");
			goto out;
		}
		if (ret == 0)
			return 0;
		PARA_INFO_LOG("received ok msg, streaming\n");
		phd->status = HTTP_STREAMING;
		return 0;
	}
	ret = -E_HTTP_RECV_OVERRUN;
	iovcnt = btr_pool_get_buffers(rn->btrp, iov);
	if (iovcnt == 0)
		goto out;
	ret = readv_nonblock(rn->fd, iov, iovcnt, &num_bytes);
	if (num_bytes == 0)
		goto out;
	if (num_bytes <= iov[0].iov_len) /* only the first buffer was filled */
		btr_add_output_pool(rn->btrp, num_bytes, btrn);
	else { /* both buffers contain data */
		btr_add_output_pool(rn->btrp, iov[0].iov_len, btrn);
		btr_add_output_pool(rn->btrp, num_bytes - iov[0].iov_len, btrn);
	}
out:
	if (ret < 0) {
		PARA_NOTICE_LOG("%s\n", para_strerror(-ret));
		btr_remove_node(&rn->btrn);
	}
	return ret;
}

static void http_recv_close(struct receiver_node *rn)
{
	close(rn->fd);
	btr_pool_free(rn->btrp);
	free(rn->private_data);
}

static int http_recv_open(struct receiver_node *rn)
{
	struct private_http_recv_data *phd;
	struct lls_parse_result *lpr = rn->lpr;
	const char *r_i = RECV_CMD_OPT_STRING_VAL(HTTP, HOST, lpr);
	uint32_t r_p = RECV_CMD_OPT_UINT32_VAL(HTTP, PORT, lpr);
	int fd, ret = para_connect(IPPROTO_TCP, r_i, r_p);

	if (ret < 0)
		return ret;
	fd = ret;
	ret = mark_fd_nonblocking(fd);
	if (ret < 0) {
		close(fd);
		return ret;
	}
	rn->private_data = phd = zalloc(sizeof(struct private_http_recv_data));
	rn->fd = fd;
	phd->status = HTTP_CONNECTED;
	rn->btrp = btr_pool_new("http_recv", 320 * 1024);
	return 1;
}

/** \cond doxygen_ignore */
const struct receiver lsg_recv_cmd_com_http_user_data = {
	.open = http_recv_open,
	.close = http_recv_close,
	.pre_monitor = http_recv_pre_monitor,
	.post_monitor = http_recv_post_monitor,
};
/** \endcond */