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

/**
 * \file afs.c The implementation of the audio file selector.
 *
 * The functions of this file execute either in afs or in command handler
 * process context where the afs and the command handler processes are siblings
 * of each other, i.e. direct descendants of the server process. Inter-process
 * communication between command handlers, server and afs is performed through
 * Unix domain sockets.
 */

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

#include "server.lsg.h"
#include "server_cmd.lsg.h"
#include "para.h"
#include "error.h"
#include "crypt.h"
#include "string.h"
#include "afh.h"
#include "afs.h"
#include "net.h"
#include "list.h"
#include "server.h"
#include "daemon.h"
#include "ipc.h"
#include "sched.h"
#include "fd.h"
#include "signal.h"
#include "sideband.h"
#include "command.h"

/**
 * The array of tables of the audio file selector.
 *
 * We organize them in an array to be able to loop over all tables.
 */
static const struct afs_table {
	/** The name is no table operation, so define it here. */
	const char * const name;
	/** The only way to invoke the ops is via this pointer. */
	const struct afs_table_operations *ops;
} afs_tables[] = {
	{.name = "audio_files", .ops = &aft_ops},
	{.name = "attributes", .ops = &attr_ops},
	{.name = "moods", .ops = &moods_ops},
	{.name = "lyrics", .ops = &lyrics_ops},
	{.name = "images", .ops = &images_ops},
	{.name = "playlists", .ops = &playlists_ops},
};
/** Used to loop over the afs tables. */
#define NUM_AFS_TABLES ARRAY_SIZE(afs_tables)

extern const struct selector_operations mood_selector_operations;
extern const struct selector_operations playlist_selector_operations;
const struct selector_operations *selector_ops[NUM_SELECTORS] = {
	[SEL_MOOD] = &mood_selector_operations,
	[SEL_PLAYLIST] = &playlist_selector_operations,
};
struct selector_instance *current_selector_instance;
enum selector_id current_selector_id, previous_selector_id;

struct command_task {
	/** The file descriptor for the local socket. */
	int fd;
	/** The associated task structure. */
	struct task *task;
};

extern int mmd_mutex;
extern struct misc_meta_data *mmd;

static int server_socket;
static struct command_task command_task_struct;

char *current_mop, *previous_mop; /* NULL means dummy mood */

/**
 * Passed from command handlers to afs.
 *
 * Command handlers cannot change the afs database directly because they run in
 * a separate process. The callback query structure circumvents this
 * restriction as follows. To instruct the afs process to execute a particular
 * function, the command hander writes an instance of this structure to a
 * shared memory area, along with the arguments to the callback function. The
 * identifier of the shared memory area is transferred to the afs process via
 * the command socket.
 *
 * The afs process reads the shared memory id from the command socket, attaches
 * the corresponding area, and calls the callback function whose address is
 * stored in the area.
 *
 * The command output, if any, is transferred back to the command handler in
 * the same way: The afs process writes the output to a second shared memory
 * area together with a fixed size metadata header whose format corresponds to
 * the \ref callback_result structure. The identifier of this area is sent back
 * to the command handler which attaches the area and forwards the output to
 * the remote client.
 *
 * \sa \ref struct callback_result.
 */
struct callback_query {
	/** The function to be called. */
	afs_callback *cb;
	/** The number of bytes of the query */
	size_t query_size;
};

static int dispatch_result(int result_shmid, callback_result_handler *handler,
		void *private_result_data)
{
	struct osl_object result;
	void *result_shm;
	/* must attach r/w as result.data might get encrypted in-place. */
	int ret2, ret = shm_attach(result_shmid, ATTACH_RW, &result_shm);
	struct callback_result *cr = result_shm;

	if (ret < 0) {
		PARA_ERROR_LOG("attach failed: %s\n", para_strerror(-ret));
		return ret;
	}
	result.size = cr->result_size;
	result.data = result_shm + sizeof(*cr);
	assert(handler);
	ret = handler(&result, cr->band, private_result_data);
	ret2 = shm_detach(result_shm);
	if (ret2 < 0) {
		PARA_ERROR_LOG("detach failed: %s\n", para_strerror(-ret2));
		if (ret >= 0)
			ret = ret2;
	}
	return ret;
}

static int connect_afs(int afs_fd)
{
	int spair[2];
	char control[255] __a_aligned(8);
	struct iovec iov = {.iov_base = "cmd\0", .iov_len = 4};
	struct msghdr msg = {
		.msg_iov = &iov,
		.msg_iovlen = 1,
		.msg_control = control,
		.msg_controllen = sizeof(control),
	};
	struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);

	if (socketpair(PF_UNIX, SOCK_STREAM, 0, spair) < 0)
		return -ERRNO_TO_PARA_ERROR(errno);
	cmsg->cmsg_level = SOL_SOCKET;
	cmsg->cmsg_type = SCM_RIGHTS;
	cmsg->cmsg_len = CMSG_LEN(sizeof(int));
	*(int *)CMSG_DATA(cmsg) = spair[0];
	/* Sum of the length of all control messages in the buffer */
	msg.msg_controllen = cmsg->cmsg_len;
	if (sendmsg(afs_fd, &msg, 0) < 0) {
		int ret = -ERRNO_TO_PARA_ERROR(errno);
		close(spair[0]);
		close(spair[1]);
		return ret;
	}
	close(spair[0]);
	return spair[1];
}

/**
 * Ask the audio file selector process to call the given function.
 *
 * \param f The function to be called.
 * \param afs_fd Permanent server-afs command socket.
 * \param query Input for the callback.
 * \param result_handler Called for each shared memory area received from afs.
 * \param private_result_data Passed verbatim to the result handler.
 *
 * This function is called from command handler context. It creates a one-time
 * socket pair for the communication between this instance of the command
 * handler and the audio file selector process (afs). One socket of the pair
 * is used by the command handler, the other by afs. This second socket is
 * transferred to the afs process by means of SCM_RIGHTS via the permanent
 * command socket connecting the two processes.
 *
 * The command handler employs the one-time socket and a shared memory area
 * to transfer the address of the given callback function and the given
 * query buffer to the afs process. Specifically, the command handler
 * allocates a shared memory area and initializes it with these data,
 * then writes the identifier of the shared memory area, a 32 bit integer,
 * to the one-time socket.
 *
 * Shared memory identifiers are also employed for the (possibly large)
 * results from the callback. Data received from the one-time socket is
 * interpreted as a stream of 32bit shared memory identifiers. The given
 * result handler is called for each such identifier received, passing the
 * private result data pointer verbatim.
 *
 * \return Number of shared memory areas dispatched on success, negative on
 * errors.
 */
int send_callback_request(afs_callback *f, int afs_fd, struct osl_object *query,
		callback_result_handler *result_handler,
		void *private_result_data)
{
	struct callback_query *cq;
	int ret, fd = -1, query_shmid, result_shmid;
	void *query_shm;
	size_t query_shm_size = sizeof(*cq);
	int dispatch_error = 0, num_dispatched = 0;

	if (query)
		query_shm_size += query->size;
	ret = shm_new(query_shm_size);
	if (ret < 0)
		return ret;
	query_shmid = ret;
	ret = shm_attach(query_shmid, ATTACH_RW, &query_shm);
	if (ret < 0)
		goto out;
	cq = query_shm;
	cq->cb = f;
	cq->query_size = query_shm_size - sizeof(*cq);

	if (query)
		memcpy(query_shm + sizeof(*cq), query->data, query->size);
	ret = shm_detach(query_shm);
	if (ret < 0)
		goto out;
	ret = connect_afs(afs_fd);
	if (ret < 0)
		goto out;
	fd = ret;
	ret = write_all(fd, &query_shmid, sizeof(query_shmid));
	if (ret < 0)
		goto out;

	/*
	 * Read all shmids from afs.
	 *
	 * Even if the dispatcher returns an error we _must_ continue to read
	 * shmids from fd so that we can destroy all shared memory areas that
	 * have been created for us by the afs process.
	 */
	for (;;) {
		char buf[sizeof(int)];
		ret = recv_bin_buffer(fd, buf, sizeof(buf));
		if (ret <= 0)
			goto out;
		assert(ret == sizeof(int));
		ret = *(int *) buf;
		assert(ret > 0);
		result_shmid = ret;
		ret = dispatch_result(result_shmid, result_handler,
			private_result_data);
		if (ret < 0 && dispatch_error >= 0)
			dispatch_error = ret;
		ret = shm_destroy(result_shmid);
		if (ret < 0)
			PARA_CRIT_LOG("destroy result failed: %s\n",
				para_strerror(-ret));
		num_dispatched++;
	}
out:
	if (shm_destroy(query_shmid) < 0)
		PARA_CRIT_LOG("shm destroy error\n");
	if (fd >= 0)
		close(fd);
	if (dispatch_error < 0)
		return dispatch_error;
	if (ret < 0)
		return ret;
	return num_dispatched;
}

/**
 * Wrapper for \ref send_callback_request() which passes a lopsub parse result.
 *
 * \param f The callback function.
 * \param afs_fd Permanent server-afs command socket.
 * \param cmd Needed for (de-)serialization.
 * \param lpr Must match cmd.
 * \param cc Passed as private data to \ref afs_cb_result_handler().
 *
 * Serialize the parse result pointer into the query buffer and send this
 * buffer to the afs process by means of the callback mechanism. Results
 * are passed to \ref afs_cb_result_handler().
 *
 * \return The return value of the underlying call to \ref
 * send_callback_request().
 */
int send_lls_callback_request(afs_callback *f, int afs_fd,
		const struct lls_command * const cmd,
		struct lls_parse_result *lpr, struct command_context *cc)
{
	struct osl_object query;
	char *buf = NULL;
	int ret = lls_serialize_parse_result(lpr, cmd, &buf, &query.size);

	assert(ret >= 0);
	query.data = buf;
	ret = send_callback_request(f, afs_fd, &query, afs_cb_result_handler,
		cc);
	free(buf);
	return ret;
}

static int action_if_pattern_matches(struct osl_row *row, void *data)
{
	struct pattern_match_data *pmd = data;
	struct osl_object name_obj;
	const char *p, *name;
	int i, ret;

	ret = osl(osl_get_object(pmd->table, row, pmd->match_col_num,
		&name_obj));
	if (ret < 0)
		return ret;
	name = name_obj.data;
	assert(name);
	if (!*name && (pmd->pm_flags & PM_SKIP_EMPTY_NAME))
		return 1;
	if (lls_num_inputs(pmd->lpr) == 0) {
		if (pmd->pm_flags & PM_NO_PATTERN_MATCHES_EVERYTHING) {
			pmd->num_matches++;
			return pmd->action(pmd->table, row, name, pmd->data);
		}
	}
	i = pmd->input_skip;
	for (;;) {
		if (i >= lls_num_inputs(pmd->lpr))
			break;
		p = lls_input(i, pmd->lpr);
		ret = fnmatch(p, name, pmd->fnmatch_flags);
		if (ret != FNM_NOMATCH) {
			if (ret != 0)
				return -E_FNMATCH;
			ret = pmd->action(pmd->table, row, name, pmd->data);
			if (ret >= 0)
				pmd->num_matches++;
			return ret;

		}
		i++;
	}
	return 1;
}

/**
 * Execute the given function for each matching row.
 *
 * \param pmd Describes what to match and how.
 *
 * \return Standard.
 */
int for_each_matching_row(struct pattern_match_data *pmd)
{
	if (pmd->pm_flags & PM_REVERSE_LOOP)
		return osl(osl_rbtree_loop_reverse(pmd->table, pmd->loop_col_num, pmd,
			action_if_pattern_matches));
	return osl(osl_rbtree_loop(pmd->table, pmd->loop_col_num, pmd,
			action_if_pattern_matches));
}

/**
 * Compare two osl objects of string type.
 *
 * \param obj1 Pointer to the first object.
 * \param obj2 Pointer to the second object.
 *
 * At most MIN(obj1->size, obj2->size) characters of each string are taken
 * into account.
 *
 * \return An integer less than, equal to, or greater than zero if the string
 * that corresponds to obj1 is found, respectively, to be less than, to match,
 * or be greater than the one of obj2.
 *
 * \sa strcmp(3), strncmp(3).
 */
int string_compare(const struct osl_object *obj1, const struct osl_object *obj2)
{
	const char *str1 = obj1->data;
	const char *str2 = obj2->data;
	return strncmp(str1, str2, PARA_MIN(obj1->size, obj2->size));
}

static int pass_afd(int fd, char *buf, size_t size)
{
	struct iovec iov = {.iov_base = buf, .iov_len = size};
	char control[255] __a_aligned(8);
	struct msghdr msg = {
		.msg_iov = &iov,
		.msg_iovlen = 1,
		.msg_control = control,
		.msg_controllen = sizeof(control),
	};
	struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);

	cmsg->cmsg_level = SOL_SOCKET;
	cmsg->cmsg_type = SCM_RIGHTS;
	cmsg->cmsg_len = CMSG_LEN(sizeof(int));
	*(int *)CMSG_DATA(cmsg) = fd;
	/* Sum of the length of all control messages in the buffer */
	msg.msg_controllen = cmsg->cmsg_len;
	PARA_DEBUG_LOG("passing %zu bytes and fd %d\n", size, fd);
	if (sendmsg(server_socket, &msg, 0) < 0)
		return -ERRNO_TO_PARA_ERROR(errno);
	return 1;
}

/*
 * Send audio file information to the server process.
 *
 * This first calls \ref open_and_update_audio_file() to obtain a file
 * descriptor to the upcoming audio file and an id of a shared memory area
 * containing information for the virtual streaming system. The file descriptor
 * and the id are transferred to the server process through the Unix domain
 * socket that was passed to \ref afs_init().
 */
static int open_next_audio_file(void)
{
	int ret, shmid, fd;
	char buf[8];

	ret = open_and_update_audio_file(&fd);
	if (ret < 0) {
		if (ret != osl(-E_OSL_RB_KEY_NOT_FOUND))
			PARA_ERROR_LOG("%s\n", para_strerror(-ret));
		goto no_admissible_files;
	}
	shmid = ret;
	if (!write_ok(server_socket)) {
		ret = -E_AFS_SOCKET;
		goto destroy;
	}
	*(uint32_t *)buf = NEXT_AUDIO_FILE;
	*(uint32_t *)(buf + 4) = (uint32_t)shmid;
	ret = pass_afd(fd, buf, 8);
	close(fd);
	if (ret >= 0)
		return ret;
destroy:
	shm_destroy(shmid);
	return ret;
no_admissible_files:
	*(uint32_t *)buf = NO_ADMISSIBLE_FILES;
	*(uint32_t *)(buf + 4) = (uint32_t)0;
	return write_all(server_socket, buf, 8);
}

static int activate_mood_or_playlist(const char *arg, struct para_buffer *pb,
		struct afs_callback_arg *aca)
{
	enum selector_id new_id;
	int ret;
	struct selector_instance *si;

	if (!arg) { /* load dummy mood */
		ret = selector_ops[SEL_MOOD]->load(NULL, pb, aca, &si);
		new_id = SEL_MOOD;
	} else if (!strcmp(arg, ".")) { /* reload current mop */
		char *mop = current_mop? current_mop + 2 : NULL;
		ret = current_selector()->load(mop, pb, aca, &si);
		new_id = current_selector_id;
	} else if (!strcmp(arg, "-")) { /* load previous mop */
		char *mop = previous_mop? previous_mop + 2 : NULL;
		ret = selector_ops[previous_selector_id]->load(mop, pb, aca, &si);
		new_id = previous_selector_id;
	} else if (!strncmp(arg, "p/", 2)) {
		ret = selector_ops[SEL_PLAYLIST]->load(arg + 2, pb, aca, &si);
		new_id = SEL_PLAYLIST;
	} else if (!strncmp(arg, "m/", 2)) {
		ret = selector_ops[SEL_MOOD]->load(arg + 2, pb, aca, &si);
		new_id = SEL_MOOD;
	} else {
		ret = -ERRNO_TO_PARA_ERROR(EINVAL);
		afs_error(aca, "%s: parse error\n", arg);
	}
	if (ret < 0)
		return ret;
	current_selector()->unload(current_selector_instance);
	current_selector_instance = si;
	/*
	 * We get called with arg == current_mop from the signal dispatcher
	 * after SIGHUP and from the error path of the select command to
	 * re-select the current mood or playlist. Don't update the four global
	 * mop variables in these cases.
	 */
	if (arg != current_mop && arg && strcmp(arg, ".")) {
		previous_selector_id = current_selector_id;
		current_selector_id = new_id;
		if (arg && !strcmp(arg, "-")) {
			char *tmp = current_mop;
			current_mop = previous_mop;
			previous_mop = tmp;
		} else {
			free(previous_mop);
			previous_mop = current_mop;
			current_mop = arg? para_strdup(arg) : NULL;
		}
	}
	/* Notify the server about the mood/playlist change. */
	mutex_lock(mmd_mutex);
	snprintf(mmd->afs_mode_string, sizeof(mmd->afs_mode_string) - 1,
		"%s", current_mop? current_mop : "dummy");
	mmd->afs_mode_string[sizeof(mmd->afs_mode_string) - 1] = '\0';
	mmd->events++;
	mutex_unlock(mmd_mutex);
	return 1;
}

/**
 * Result handler for sending data to the para_client process.
 *
 * \param result The data to be sent.
 * \param band The band designator.
 * \param private Pointer to the command context.
 *
 * \return The return value of the underlying call to \ref send_sb.
 *
 * \sa \ref callback_result_handler
 */
int afs_cb_result_handler(struct osl_object *result, uint8_t band,
		void *private)
{
	struct command_context *cc = private;

	assert(cc);
	switch (band) {
	case SBD_OUTPUT:
	case SBD_DEBUG_LOG:
	case SBD_INFO_LOG:
	case SBD_NOTICE_LOG:
	case SBD_WARNING_LOG:
	case SBD_ERROR_LOG:
	case SBD_CRIT_LOG:
	case SBD_EMERG_LOG:
		assert(result->size > 0);
		return send_sb(&cc->scc, result->data, result->size, band, true);
	case SBD_AFS_CB_FAILURE:
		return *(int *)(result->data);
	default:
		return -E_BAD_BAND;
	}
}

static void flush_and_free_pb(struct para_buffer *pb)
{
	int ret;
	struct afs_max_size_handler_data *amshd = pb->private_data;

	if (pb->buf && pb->size > 0) {
		ret = pass_buffer_as_shm(amshd->fd, amshd->band, pb->buf,
			pb->offset);
		if (ret < 0)
			PARA_ERROR_LOG("%s\n", para_strerror(-ret));
	}
	free(pb->buf);
}

static void activate_mop_or_dummy(const char *arg)
{
	int ret = activate_mood_or_playlist(arg, NULL, NULL);

	if (ret < 0) {
		PARA_WARNING_LOG("could not activate %s: %s\n", arg?
			arg : "dummy", para_strerror(-ret));
		if (arg)
			assert(activate_mood_or_playlist(NULL, NULL, NULL) >= 0);
	}
}

static int setup_command_socket_or_die(void)
{
	int ret, socket_fd;
	const char *socket_name = OPT_STRING_VAL(AFS_SOCKET);

	unlink(socket_name);
	ret = create_local_socket(socket_name);
	if (ret < 0) {
		PARA_EMERG_LOG("%s: %s\n", para_strerror(-ret), socket_name);
		exit(EXIT_FAILURE);
	}
	socket_fd = ret;
	PARA_INFO_LOG("listening on socket %s (fd %d)\n", socket_name,
		socket_fd);
	return socket_fd;
}

static char *database_dir;

static void close_afs_tables(void)
{
	int i;
	PARA_NOTICE_LOG("closing afs tables\n");
	for (i = 0; i < NUM_AFS_TABLES; i++)
		afs_tables[i].ops->close();
	free(database_dir);
	database_dir = NULL;
}

static void get_database_dir(void)
{
	if (!database_dir) {
		if (OPT_GIVEN(AFS_DATABASE_DIR))
			database_dir = para_strdup(OPT_STRING_VAL(AFS_DATABASE_DIR));
		else {
			char *home = para_homedir();
			database_dir = make_message(
				"%s/.paraslash/afs_database-0.7", home);
			free(home);
		}
	}
	PARA_INFO_LOG("afs_database dir %s\n", database_dir);
}

static int open_afs_tables(bool reload)
{
	int i, ret;

	get_database_dir();
	if (OPT_GIVEN(INIT) && !reload) {
		PARA_NOTICE_LOG("creating afs database\n");
		if (mkdir(database_dir, 0777) < 0)
			return -ERRNO_TO_PARA_ERROR(errno);
		for (i = 0; i < NUM_AFS_TABLES; i++) {
			const struct afs_table *t = afs_tables + i;
			if (!t->ops->create)
				continue;
			ret = t->ops->create(database_dir);
			if (ret < 0) {
				PARA_ERROR_LOG("cannot create table %s\n",
					t->name);
				return ret;
			}
			PARA_INFO_LOG("created %s table\n", t->name);
		}
	}
	PARA_NOTICE_LOG("opening %zu osl tables in %s\n", NUM_AFS_TABLES,
		database_dir);
	for (i = 0; i < NUM_AFS_TABLES; i++) {
		ret = afs_tables[i].ops->open(database_dir);
		if (ret >= 0)
			continue;
		PARA_ERROR_LOG("could not open %s\n", afs_tables[i].name);
		break;
	}
	if (ret >= 0)
		return ret;
	while (i)
		afs_tables[--i].ops->close();
	return ret;
}

static int afs_signal_post_monitor(struct sched *s, __a_unused void *context)
{
	int signum, ret;
	struct signal_task *st = context;

	ret = task_get_notification(st->task);
	if (ret < 0)
		return ret;
	if (getppid() == 1) {
		PARA_EMERG_LOG("para_server died\n");
		goto shutdown;
	}
	signum = para_next_signal();
	if (signum == 0)
		return 0;
	if (signum == SIGHUP) {
		close_afs_tables();
		parse_config_or_die(1);
		ret = open_afs_tables(true);
		if (ret < 0) {
			PARA_ERROR_LOG("cannot re-open tables: %s\n",
				para_strerror(-ret));
			goto shutdown;
		}
		activate_mop_or_dummy(current_mop);
		return 0;
	}
	PARA_EMERG_LOG("terminating on signal %d\n", signum);
shutdown:
	task_notify_all(s, E_AFS_SIGNAL);
	return -E_AFS_SIGNAL;
}

static void register_signal_task(struct sched *s)
{
	static struct signal_task signal_task;

	signal_task.fd = signal_init();
	para_sigaction(SIGPIPE, SIG_IGN);
	para_install_sighandler(SIGINT);
	para_install_sighandler(SIGTERM);
	para_install_sighandler(SIGHUP);
	signal_task.task = task_register(&(struct task_info) {
		.name = "signal",
		.pre_monitor = signal_pre_monitor,
		.post_monitor = afs_signal_post_monitor,
		.context = &signal_task,

	}, s);
}


static void command_pre_monitor(struct sched *s, __a_unused void *context)
{
	sched_monitor_readfd(server_socket, s);
}

/**
 * Send data as shared memory identifiers.
 *
 * \param fd File descriptor to send the identifier to.
 * \param band The band designator for this data.
 * \param buf The contents to be sent.
 * \param size The size of the content buffer.
 *
 * This function copies the buffer to a newly created shared memory area
 * and sends the identifier of this area to the given file descriptor.
 *
 * It is called by the max_size handler of the audio file selector as well
 * as directly by the callbacks of certain afs commands.
 *
 * \return Zero if NULL or a zero-sized buffer was passed, negative error
 * code on failure, positive on success.
 */
int pass_buffer_as_shm(int fd, uint8_t band, const char *buf, size_t size)
{
	int ret, shmid;
	void *shm;
	struct callback_result *cr;

	if (size == 0)
		assert(band != SBD_OUTPUT);
	ret = shm_new(size + sizeof(*cr));
	if (ret < 0)
		return ret;
	shmid = ret;
	ret = shm_attach(shmid, ATTACH_RW, &shm);
	if (ret < 0)
		goto err;
	cr = shm;
	cr->result_size = size;
	cr->band = band;
	if (size > 0)
		memcpy(shm + sizeof(*cr), buf, size);
	ret = shm_detach(shm);
	if (ret < 0)
		goto err;
	ret = write_all(fd, (char *)&shmid, sizeof(int));
	if (ret >= 0)
		return ret;
err:
	if (shm_destroy(shmid) < 0)
		PARA_ERROR_LOG("destroy result failed\n");
	return ret;
}

/**
 * Format and send an error message to the command handler.
 *
 * To pass an error message from the callback of an afs command to the client,
 * this function should be called. It formats the message into a buffer which
 * is passed as a shared memory area identifier to the command handler from
 * where it propagates to the client.
 *
 * The message will be tagged with the ERROR_LOG sideband designator so that
 * the client writes it to its stderr stream rather than to stdout as with
 * aca->pbout. In analogy to the default Unix semantics of stderr, the message
 * is sent without buffering.
 *
 * If sending the error message fails, an error is logged on the server side,
 * but no other action is taken.
 *
 * \param aca Determines the file descriptor to send the identifier to.
 * \param fmt Usual format string.
 */
__printf_2_3 void afs_error(const struct afs_callback_arg *aca,
		const char *fmt,...)
{
	va_list argp;
	char *msg;
	unsigned n;
	int ret;

	va_start(argp, fmt);
	n = xvasprintf(&msg, fmt, argp);
	va_end(argp);
	if (!aca) {
		PARA_ERROR_LOG("%s", msg);
		free(msg);
		return;
	}
	ret = pass_buffer_as_shm(aca->fd, SBD_ERROR_LOG, msg, n + 1);
	if (ret < 0)
		PARA_ERROR_LOG("Could not send %s: %s\n", msg,
			para_strerror(-ret));
	free(msg);
}

/* Pass one chunk of output to the command handler as a shared memory area. */
static int afs_max_size_handler(char *buf, size_t size, void *private)
{
	struct afs_max_size_handler_data *amshd = private;
	return pass_buffer_as_shm(amshd->fd, amshd->band, buf, size);
}

static int call_callback(int fd, int query_shmid)
{
	void *query_shm;
	struct callback_query *cq;
	int ret, ret2;
	struct afs_callback_arg aca = {.fd = fd};

	ret = shm_attach(query_shmid, ATTACH_RW, &query_shm);
	if (ret < 0)
		return ret;
	cq = query_shm;
	aca.query.data = (char *)query_shm + sizeof(*cq);
	aca.query.size = cq->query_size;
	aca.pbout.max_size = shm_get_shmmax();
	aca.pbout.max_size_handler = afs_max_size_handler;
	aca.pbout.private_data = &(struct afs_max_size_handler_data) {
		.fd = fd,
		.band = SBD_OUTPUT
	};
	ret = cq->cb(&aca);
	ret2 = shm_detach(query_shm);
	if (ret2 < 0) {
		if (ret < 0) /* ignore (but log) detach error */
			PARA_ERROR_LOG("could not detach sma: %s\n",
				para_strerror(-ret2));
		else
			ret = ret2;
	}
	flush_and_free_pb(&aca.pbout);
	if (ret < 0) {
		ret2 = pass_buffer_as_shm(fd, SBD_AFS_CB_FAILURE,
			(const char *)&ret, sizeof(ret));
		if (ret2 < 0)
			PARA_ERROR_LOG("could not pass cb failure packet: %s\n",
				para_strerror(-ret2));
	}
	return ret;
}

/* returns 0 if no data available, 1 else */
static int execute_afs_command(int fd)
{
	int query_shmid;
	int ret;
	ssize_t sz = read(fd, &query_shmid, sizeof(query_shmid));

	if (sz < 0) {
		ret = -ERRNO_TO_PARA_ERROR(errno);
		goto out;
	}
	if (sz != sizeof(query_shmid)) {
		PARA_NOTICE_LOG("short read (%zd bytes, expected %lu)\n",
			sz, (long unsigned) sizeof(query_shmid));
		return 1;
	}
	ret = call_callback(fd, query_shmid);
	close(fd);
out:
	if (ret < 0)
		PARA_NOTICE_LOG("%s\n", para_strerror(-ret));
	return ret;
}

static int execute_command(void)
{
	int ret, fd = -1;
	char control[255] __a_aligned(8), buf[4];
	struct iovec iov = {.iov_base = buf, .iov_len = sizeof(buf)};
	struct msghdr msg = {
		.msg_iov = &iov,
		.msg_iovlen = 1,
		.msg_control = control,
		.msg_controllen = sizeof(control),
	};
	struct cmsghdr *cmsg;

	ret = recvmsg(server_socket, &msg, 0);
	if (ret < 0) {
		if (errno == EAGAIN || errno == EWOULDBLOCK)
			return 0;
		return -ERRNO_TO_PARA_ERROR(errno);
	}
	if (iov.iov_len != sizeof(buf))
		return -E_AFS_SHORT_READ;
	for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
		if (cmsg->cmsg_level != SOL_SOCKET
			|| cmsg->cmsg_type != SCM_RIGHTS)
			continue;
		if ((cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int) != 1)
			continue;
		fd = *(int *)CMSG_DATA(cmsg);
	}
	buf[3] = '\0';
	if (!strcmp(buf, "cmd")) {
		assert(fd > 0);
		ret = execute_afs_command(fd);
		if (ret < 0)
			PARA_INFO_LOG("afs command failed: %s\n",
				para_strerror(-ret));
		return 0;
	}
	if (!strcmp(buf, "new"))
		return open_next_audio_file();
	return -ERRNO_TO_PARA_ERROR(EINVAL);
}

static int command_post_monitor(struct sched *s, void *context)
{
	struct command_task *ct = context;
	int ret;

	ret = task_get_notification(ct->task);
	if (ret < 0)
		return ret;
	ret = execute_command();
	if (ret < 0) {
		PARA_EMERG_LOG("%s\n", para_strerror(-ret));
		task_notify_all(s, -ret);
	}
	return ret;
}

static void register_command_task(struct sched *s)
{
	struct command_task *ct = &command_task_struct;
	ct->fd = setup_command_socket_or_die();

	ct->task = task_register(&(struct task_info) {
		.name = "afs command",
		.pre_monitor = command_pre_monitor,
		.post_monitor = command_post_monitor,
		.context = ct,
	}, s);
}

static int afs_poll(struct pollfd *fds, nfds_t nfds, int timeout)
{
	mutex_lock(mmd_mutex);
	daemon_set_loglevel(mmd->loglevel);
	mutex_unlock(mmd_mutex);
	return xpoll(fds, nfds, timeout);
}
/**
 * Initialize the audio file selector.
 *
 * \param socket_fd File descriptor for communication with the server process.
 *
 * Open all tables of the afs database, then schedule the signal task and the
 * command task. The former reacts to signals by reloading the database or by
 * terminating after closing all database tables. The command task dispatches
 * commands from the server and from command handlers.
 *
 * The server connects the command task when the virtual streaming system
 * needs to stream the next audio file. The command task opens the highest
 * scoring admissible file and sends streaming information back to the
 * server. The command task also accepts connections from command handlers,
 * modifies or queries the afs database accordingly, and reports its findings
 * back to the command handler.
 */
__noreturn void afs_init(int socket_fd)
{
	struct sched *s = sched_new(afs_poll);
	int ret;

	register_signal_task(s);
	ret = open_afs_tables(false);
	if (ret < 0)
		goto out;
	server_socket = socket_fd;
	ret = mark_fd_nonblocking(server_socket);
	if (ret < 0)
		goto out_close;
	PARA_INFO_LOG("server_socket: %d\n", server_socket);
	activate_mop_or_dummy(OPT_STRING_VAL(AFS_INITIAL_MODE));
	register_command_task(s);
	ret = write(socket_fd, "\0", 1);
	if (ret != 1) {
		if (ret == 0)
			errno = EINVAL;
		ret = -ERRNO_TO_PARA_ERROR(errno);
		goto out_close;
	}
	ret = schedule(s);
	sched_shutdown(s);
	current_selector()->unload(current_selector_instance);
out_close:
	close_afs_tables();
out:
	free_status_items();
	free(current_mop);
	free(previous_mop);
	free_lpr();
	if (ret < 0)
		PARA_EMERG_LOG("%s\n", para_strerror(-ret));
	exit(EXIT_FAILURE);
}

static int com_select_callback(struct afs_callback_arg *aca)
{
	const struct lls_command *cmd = SERVER_CMD_CMD_PTR(SELECT);
	const char *arg;
	int ret;
	struct para_buffer *pbout;

	ret = lls_deserialize_parse_result(aca->query.data, cmd, &aca->lpr);
	assert(ret >= 0);
	arg = lls_input(0, aca->lpr);
	pbout = SERVER_CMD_OPT_GIVEN(SELECT, VERBOSE, aca->lpr)?
		&aca->pbout : NULL;
	ret = activate_mood_or_playlist(arg, pbout, aca);
	if (ret < 0)
		afs_error(aca, "cannot activate %s\n", arg);
	lls_free_parse_result(aca->lpr, cmd);
	return ret;
}

static int com_select(struct command_context *cc, struct lls_parse_result *lpr)
{
	const struct lls_command *cmd = SERVER_CMD_CMD_PTR(SELECT);
	char *errctx;
	int ret = lls(lls_check_arg_count(lpr, 1, 1, &errctx));

	if (ret < 0) {
		send_errctx(cc, errctx);
		return ret;
	}
	ret = send_lls_callback_request(com_select_callback, cc->afs_fd, cmd,
		lpr, cc);
	return ret == osl(-E_OSL_RB_KEY_NOT_FOUND)? -E_BAD_MOP : ret;
}
EXPORT_SERVER_CMD_HANDLER(select);

static int com_check(struct command_context *cc, struct lls_parse_result *lpr)
{
	const struct lls_opt_result *r_a = SERVER_CMD_OPT_RESULT(CHECK, AFT, lpr);
	const struct lls_opt_result *r_A = SERVER_CMD_OPT_RESULT(CHECK, ATTRIBUTE, lpr);
	const struct lls_opt_result *r_m = SERVER_CMD_OPT_RESULT(CHECK, MOOD, lpr);
	const struct lls_opt_result *r_p = SERVER_CMD_OPT_RESULT(CHECK, PLAYLIST, lpr);
	bool noopt = !lls_opt_given(r_a) && !lls_opt_given(r_m)
		&& !lls_opt_given(r_p);
	int ret;

	if (lls_opt_given(r_A))
		send_sb_va(&cc->scc, SBD_WARNING_LOG,
			"--attribute has no effect and will be removed\n");
	if (noopt || lls_opt_given(r_a)) {
		ret = send_callback_request(aft_check_callback, cc->afs_fd, NULL,
			afs_cb_result_handler, cc);
		if (ret < 0)
			return ret;
	}
	if (noopt || lls_opt_given(r_p)) {
		ret = send_callback_request(selector_ops[SEL_PLAYLIST]->check,
			cc->afs_fd, NULL, afs_cb_result_handler, cc);
		if (ret < 0)
			return ret;
	}
	if (noopt || lls_opt_given(r_m)) {
		ret = send_callback_request(selector_ops[SEL_MOOD]->check,
			cc->afs_fd, NULL, afs_cb_result_handler, cc);
		if (ret < 0)
			return ret;
	}
	return 1;
}
EXPORT_SERVER_CMD_HANDLER(check);

/**
 * The afs event dispatcher.
 *
 * \param event Type of the event.
 * \param data Size and contents depend on the event type.
 *
 * This function calls each table event handler, passing the buffer and the
 * data pointer verbatim. If an event handler returns negative, the loop
 * is aborted.
 *
 * \return The (negative) error code of the first handler that failed, or non-negative
 * if all handlers succeeded.
 */
__must_check int afs_event(enum afs_events event, void *data)
{
	int i, ret;

	for (i = 0; i < NUM_AFS_TABLES; i++) {
		const struct afs_table *t = afs_tables + i;
		if (!t->ops->event_handler)
			continue;
		ret = t->ops->event_handler(event, data);
		if (ret < 0) {
			PARA_CRIT_LOG("table %s, event %u: %s\n", t->name,
				event, para_strerror(-ret));
			return ret;
		}
	}
	return 1;
}