/* SPDX-License-Identifier: GPL-2.0 */
/** \file udp_recv.c Paraslash's udp receiver */

#include <netinet/in.h>
#include <net/if.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <lopsub.h>

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

static void udp_recv_pre_monitor(struct sched *s, void *context)
{
	struct receiver_node *rn = context;

	if (generic_recv_pre_monitor(s, rn) <= 0)
		return;
	sched_monitor_readfd(rn->fd, s);
}

static int udp_check_eof(size_t sz, struct iovec iov[2])
{
	if (sz < FEC_EOF_PACKET_LEN)
		return 0;
	if (iov[0].iov_len >= FEC_EOF_PACKET_LEN) {
		if (memcmp(iov[0].iov_base, FEC_EOF_PACKET,
				FEC_EOF_PACKET_LEN) != 0)
			return 0;
		return -E_EOF;
	}
	if (memcmp(iov[0].iov_base, FEC_EOF_PACKET, iov[0].iov_len) != 0)
		return 0;
	if (memcmp(iov[1].iov_base, &FEC_EOF_PACKET[iov[0].iov_len],
			FEC_EOF_PACKET_LEN - iov[0].iov_len) != 0)
		return 0;
	return -E_EOF;
}

static int udp_recv_post_monitor(__a_unused struct sched *s, void *context)
{
	struct receiver_node *rn = context;
	struct btr_node *btrn = rn->btrn;
	size_t num_bytes;
	struct iovec iov[2];
	int ret, iovcnt;

	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;
	iovcnt = btr_pool_get_buffers(rn->btrp, iov);
	ret = -E_UDP_OVERRUN;
	if (iovcnt == 0)
		goto out;
	ret = readv_nonblock(rn->fd, iov, iovcnt, &num_bytes);
	if (num_bytes == 0)
		goto out;
	ret = udp_check_eof(num_bytes, iov);
	if (ret < 0)
		goto out;
	if (iov[0].iov_len >= num_bytes)
		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);
	}
	return 1;
out:
	if (ret < 0) {
		btr_remove_node(&rn->btrn);
		close(rn->fd);
		rn->fd = -1;
	}
	return ret;
}

static void udp_recv_close(struct receiver_node *rn)
{
	if (rn->fd >= 0)
		close(rn->fd);
	btr_pool_free(rn->btrp);
}

/*
 * Perform AF-independent joining of multicast receive addresses.
 *
 * fd: Bound socket descriptor.
 * iface: The receiving multicast interface, or NULL for the default.
 */
static int setup_multicast(int fd, const char *iface)
{
	struct sockaddr_storage ss;
	socklen_t sslen = sizeof(ss);
	int id = iface? if_nametoindex(iface) : 0;
#ifdef HAVE_IP_MREQN
	struct ip_mreqn m4 = {.imr_ifindex = id};
#else
	struct ip_mreq m4 = {.imr_interface.s_addr = INADDR_ANY};
#endif
	struct in_addr *in4;

	if (getsockname(fd, (struct sockaddr *)&ss, &sslen) < 0)
		return -ERRNO_TO_PARA_ERROR(errno);
	if (iface && id == 0)
		PARA_WARNING_LOG("cannot resolve %s, using default\n", iface);
	if (ss.ss_family == AF_INET6) {
		struct in6_addr *in6 = &((struct sockaddr_in6 *)&ss)->sin6_addr;
		struct ipv6_mreq m6;
		if (!IN6_IS_ADDR_MULTICAST(in6))
			return 0;
		memset(&m6, 0, sizeof(m6));
		memcpy(&m6.ipv6mr_multiaddr, in6, 16);
		m6.ipv6mr_interface = id;
		if (setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &m6,
				sizeof(m6)) < 0)
			return -ERRNO_TO_PARA_ERROR(errno);
		return 1;
	}
	/* AF_INET */
	in4 = &((struct sockaddr_in *)&ss)->sin_addr;
	if (!IN_MULTICAST(htonl(in4->s_addr)))
		return 0;
	m4.imr_multiaddr = *in4;
	if (setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &m4, sizeof(m4)) < 0)
		return -ERRNO_TO_PARA_ERROR(errno);
	return 1;
}

static int udp_recv_open(struct receiver_node *rn)
{
	struct lls_parse_result *lpr = rn->lpr;
	const char *iface = RECV_CMD_OPT_STRING_VAL(UDP, IFACE, lpr);
	const char *host = RECV_CMD_OPT_STRING_VAL(UDP, HOST, lpr);
	uint32_t port = RECV_CMD_OPT_UINT32_VAL(UDP, PORT, lpr);
	int ret;

	ret = makesock(IPPROTO_UDP, true /* passive */, host, port);
	if (ret < 0)
		return ret;
	rn->fd = ret;
	ret = setup_multicast(rn->fd, iface);
	if (ret < 0)
		goto err;
	ret = mark_fd_nonblocking(rn->fd);
	if (ret < 0)
		goto err;
	PARA_INFO_LOG("receiving from %s:%u, fd=%d\n", host, port, rn->fd);
	rn->btrp = btr_pool_new("udp_recv", 320 * 1024);
	return rn->fd;
err:
	close(rn->fd);
	return ret;
}

/** \cond doxygen_ignore */
const struct receiver lsg_recv_cmd_com_udp_user_data = {
	.open = udp_recv_open,
	.close = udp_recv_close,
	.pre_monitor = udp_recv_pre_monitor,
	.post_monitor = udp_recv_post_monitor,
};
/** \endcond */