#define _GNU_SOURCE

#include <ctype.h>
#include <errno.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <netdb.h>
#include <sys/syscall.h>
#include <sched.h>
#include <stdatomic.h>

#include <sys/stat.h>
#include <sched.h>
#include <mqueue.h>

#include "config_servers.h"

#include "darray.h"
#include "buf.h"
#include "mq_cmd.h"

/* 
no defence programming, no error checking, no argument checking just PoC 
nothing else
*/

#define MQ_MSG_SIZE 8192

/*
return fd!=ifconnection there
*/
int irc_connect( char *hostname, char *port )
{
	int fd=0;

	struct addrinfo serv, *res;

	memset(&serv, 0, sizeof(serv));
	serv.ai_family = AF_INET;
	serv.ai_socktype = SOCK_STREAM;
	getaddrinfo(hostname, port, &serv, &res);
	fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
	connect(fd, res->ai_addr, res->ai_addrlen);

	return fd;
}

//async io multiplexer
typedef struct mq_ntf_io_mplx
{
	struct timeval tv;
	fd_set io_fds;
} mq_ntf_io_mplx;

#define MQ_OUT  1
#define MQ_IN   2

/* 
supposed format agni-[id]-[in/out] 
in is for input of thread (send to in and thread will recieve),
out if for output of thread (read from out to recieve from thread)
*/
#define MQ_PREFIX "/agni-"
typedef struct mq_ntf_mdt
{
	int id;
	mqd_t mq_in;
	mqd_t mq_out;
	mq_ntf_io_mplx io_mplx;
	//_Atomic int used; nado?
} mq_ntf_mdt;


/* 
return -1 on error 
*/
int mq_ntf_open(mq_ntf_mdt *mq, int id)
{
	const int name_size = 32;
	char name[name_size];

	if (!mq)
	{
		PERM();
		return -1;
	}
	mq->id = id;
	snprintf(name, name_size, MQ_PREFIX"%d-in",id);
	/*create with default configs*/
	mq->mq_in = mq_open(name,
		O_RDWR | O_CREAT, 0666, NULL);
	if (mq->mq_in == -1)
	{
		ENL();
		perror("mq_open");
		return -1;
	}
	snprintf(name, name_size, MQ_PREFIX"%d-out",id);
	/*create with default configs*/
	mq->mq_out = mq_open(name,
		O_RDWR | O_CREAT, 0666, NULL);
	if (mq->mq_out == -1)
	{
		perror("mq_open");
		return -1;
	}

	PRINT("opened = %d %d\n", mq->mq_in, mq->mq_out);

	return 0;
}

int mq_ntf_mplx(mq_ntf_mdt *mq, int msec)
{
	fd_set io_fds;

	if (mq == NULL)
	{
		PERM();
		return -1;
	}

	if (msec < 0)
	{
		PERM();
		return -1;
	}

	//mq->io_mplx.tv.tv_sec  = msec/1000;
	//mq->io_mplx.tv.tv_usec = (msec%1000)*1000;
	mq->io_mplx.tv.tv_sec  = 2;
	mq->io_mplx.tv.tv_usec = 0;
	PRINT("mplx = %d %d\n", mq->mq_in, mq->mq_out);
	FD_ZERO(&io_fds);
	FD_SET(mq->mq_in,  &io_fds);//in first
	FD_SET(mq->mq_out, &io_fds);//out second
	mq->io_mplx.io_fds = io_fds;

	return 0;
}

/* return number if there is something to read
RETURN:
	-1        some error
	0x00      nothing to read
	0x01      input ready
	0x02      output ready
	0x02|0x01 input&output ready

*/
//if mplx havent done, everything could blow up
int mq_ntf_select(mq_ntf_mdt *mq)
{
	int fdnum, fd1, fd2;
	int ret=0x00;
	fd_set io_fds;
	struct timeval tv;

	if (mq == NULL)
	{
		PERM();
		return -1;
	}

	//we should allways have just 2 descriptors to listen
	//io_fds = mq->io_mplx.io_fds;
	fd1 = mq->mq_in;
	fd2 = mq->mq_out;
	FD_ZERO(&io_fds);
	FD_SET(fd1, &io_fds);
	FD_SET(fd2, &io_fds);
	tv.tv_sec = 1;
	tv.tv_usec = 0;
	fdnum = select(2, 
		&io_fds,
		NULL,
		NULL,
		&tv );

	if (fdnum == -1)
	{
		ENL();
		return -1;
	} else {
		if (FD_ISSET(mq->mq_in, &mq->io_mplx.io_fds))
		{
			ret = 0x01;
		}
		
		if (FD_ISSET(mq->mq_out, &mq->io_mplx.io_fds))
		{
			ret = 0x02;
		}
	}

	return ret;
}

int mq_ntf_read(mq_ntf_mdt *mq, int dir, char *buf, size_t size)
{
	struct mq_attr attr;
	size_t bytes;
	mqd_t mqd;

	unsigned int prio = 10;

	if (dir == MQ_IN)
	{
		mqd = mq->mq_in;
	} else if (dir == MQ_OUT)
	{
		mqd = mq->mq_out;
	} else
	{
		perror("Wrong direction");
	}

	mq_getattr(mqd, &attr);
	if (attr.mq_maxmsg == attr.mq_curmsgs)
	{
		printf("queue %d out full\n", mq->id);
		return -1;
	}
	bytes = mq_receive(mqd, buf, 
		size > attr.mq_msgsize ? attr.mq_msgsize : size,
		&prio);
	if (bytes == -1)
	{
		perror("mq_receive read out");
		return -1;
	}

	return bytes;
}

int mq_ntf_write(mq_ntf_mdt *mq, int dir, const char *buf, size_t size)
{
	struct mq_attr attr;
	size_t bytes;
	mqd_t mqd;

	unsigned int prio = 10;

	if (dir == MQ_IN)
	{
		mqd = mq->mq_in;
	} else if (dir == MQ_OUT)
	{
		mqd = mq->mq_out;
	} else
	{
		perror("Wrong direction");
	}

	mq_getattr(mqd, &attr);
	if (attr.mq_maxmsg == attr.mq_curmsgs)
	{
		printf("queue %d out full\n", mq->id);
		return -1;
	}
	bytes = mq_send(mqd, buf, 
		size > attr.mq_msgsize ? attr.mq_msgsize : size,
		prio);
	if (bytes == -1)
	{
		perror("mq_send");
		return -1;
	}

	return bytes;
}

//dir - dirction 1 in 2 out
int mq_ntf_count(mq_ntf_mdt *mq, int dir)
{
	struct mq_attr attr;
	int mq_fd;

	if (mq == NULL)
	{
		PERM();
		return -1;
	}

	if (dir == 1)
	{
		mq_fd = mq->mq_in;
	} else
	{
		mq_fd = mq->mq_out;
	}

	if (mq_getattr(mq_fd, &attr) != -1)
	{
		return attr.mq_curmsgs;
	}

	return -1;
}


/* close queue */
int mq_ntf_close(mq_ntf_mdt *mq)
{
	const int name_size = 32;
	char name[name_size];
	int id;

	id = mq->id;
	snprintf(name, name_size, MQ_PREFIX"%d-in",id);
	mq_unlink(name);

	snprintf(name, name_size, MQ_PREFIX"%d-out",id);
	mq_unlink(name);

	//clean file descriptors for io_mplx

	return 0;
}

//send command
int mq_ntf_cmd_send(mq_ntf_mdt *mq, mq_cmd *cmd)
{

	return 0;
}

//recieve command from other end
int mq_ntf_cmd_recv(mq_ntf_mdt *mq, mq_cmd *cmd)
{
	return 0;
}

int mq_drain_q(mqd_t mqd)
{
	struct mq_attr attr;
	char *buf = NULL;
	unsigned int prio = 0;
	int cnt_msg = 0;
	int num_read = 0;
	int i = 0;

	if (mq_getattr(mqd, &attr) == -1)
	{
		ENL();
		return -1;
	}

	buf = malloc(attr.mq_msgsize);
	if (buf == NULL)
	{
		ENL();
		return -1;
	}

	for (i=0; i<attr.mq_curmsgs; i++)
	{
		num_read = mq_receive(mqd, buf, attr.mq_msgsize, &prio);
		if (num_read == -1)
		{
			ENL();
			free(buf);
			return -1;
		}
		cnt_msg += 1;
	}

	free(buf);
	return cnt_msg;
}

//drain all mq buffers to zero
//to use at begining and clear all buffers to zero
int mq_ntf_drain(mq_ntf_mdt *mq)
{
	int err;

	err = mq_drain_q(mq->mq_in);
	if (err < 0)
	{
		PERM();
		return -1;
	}
	printf("Drained in %d messages\n", err);

	err = mq_drain_q(mq->mq_out);
	if (err < 0)
	{
		PERM();
		return -1;
	}
	printf("Drained out %d messages\n", err);

	return 0;
}



int mq_ntf_getattr(mq_ntf_mdt *mq, int dir, struct mq_attr **attr)
{
	struct mq_attr gattr;
	mqd_t mqdt;

	if (mq == NULL)
	{
		PERM();
		return -1;
	}

	if (dir == MQ_OUT)
	{
		mqdt = mq->mq_out;
	} else if (dir == MQ_IN)
	{
		mqdt = mq->mq_in;
	} else
	{
		return -1;
	}

	if (mq_getattr(mqdt, &gattr) == -1)
	{
		ENL();
		return -1;
	}

	memcpy(*attr,&gattr,sizeof(struct mq_attr));

	return 0;
}



int send_mq_cmd(mq_ntf_mdt *mq, 
	int io,
	int cmd_id, 
	char *cmd, 
	size_t cmd_sz, 
	char *param, 
	size_t param_sz)
{
	int err = 0;
	mq_cmd *cmd_send = NULL;
	cmd_send = mq_cmd_create(cmd_id,cmd,cmd_sz,param,param_sz);
	if (cmd_send == NULL)
	{
		printf("Cant create command\n");
		return -1;
	}

	err = -1;
	if (io == MQ_OUT)
	{
		err = mq_ntf_write(mq, MQ_OUT, cmd_send->buf, cmd_send->sz);
	} else if (io == MQ_IN)
	{
		err = mq_ntf_write(mq, MQ_IN, cmd_send->buf, cmd_send->sz);
	} else
	{
		printf("Unknown direction");
		return -1;
	}

	if (err == -1)
	{
		printf("Couldnt recieve command\n");
		mq_cmd_free(cmd_send);
		return -1;
	}

	mq_cmd_free(cmd_send);

	return 0;
}

int recv_mq_cmd(mq_ntf_mdt *mq, 
	int io,
	mq_cmd **cmd) //return back recieved command
{
	int err = 0;
	mq_cmd *cmd_recv = NULL;
	int recv_sz=-1;
	char *in_buf=NULL;

	in_buf = malloc(MQ_MSG_SIZE);
	if (in_buf == NULL)
	{
		ENL();
		return -1;
	}

	err = -1;
	if (io == MQ_OUT)
	{
		err = mq_ntf_read(mq, MQ_OUT, in_buf, MQ_MSG_SIZE);
	} else if (io == MQ_IN)
	{
		err = mq_ntf_read(mq, MQ_IN, in_buf, MQ_MSG_SIZE);
	} else
	{
		printf("Unknown direction");
		return -1;
	}

	if (err == -1)
	{
		printf("Couldnt recieve command\n");
		free(in_buf);
		return -1;
	}
	recv_sz = err;

	//in err should be still the size of recieved buffer
	cmd_recv = mq_cmd_creates(in_buf, recv_sz);
	if (cmd_recv == NULL)
	{
		printf("Cannot create cmd\n");
		free(in_buf);
		return -1;
	}

	*cmd = cmd_recv;
	free(in_buf);

	return 0;
}

#define SEND_CMD_IN(MQ,ID,CMD,PARAM) \
	send_mq_cmd(MQ,MQ_IN,ID,CMD,strlen(CMD),PARAM,strlen(PARAM));
#define SEND_CMD_OUT(MQ,ID,CMD,PARAM) \
	send_mq_cmd(MQ,MQ_OUT,ID,CMD,strlen(CMD),PARAM,strlen(PARAM));
#define RECV_CMD_IN(MQ,CMD) \
	recv_mq_cmd(MQ,MQ_IN,CMD);
#define RECV_CMD_OUT(MQ,CMD) \
	recv_mq_cmd(MQ,MQ_OUT,CMD);

/*

|--------|<--IN---|---------|
| SERVER |        | MANAGER |
|--------|--OUT-->|---------|

*/
#define STACK_SIZE (1024*16)
/* not supposed to be changed. */
typedef struct server_cfg
{
	/* thread params */
	int tid;
	void *stack;
	//atomic_int running;
	_Atomic int running;
	mq_ntf_mdt *mq;
	/* irc server config */
	char *user;
	char *password;
	char *server;
	char **channels;
	int port;
	int ssl;
} server_cfg;


/* server_cfg struct as input */
#define TH_STATE_INIT        0
#define TH_STATE_START       1
#define TH_STATE_LISTEN_IN   2
#define TH_STATE_LISTEN_OUT  3
#define TH_STATE_SEND_IN     4
#define TH_STATE_SEND_OUT    5
#define TH_STATE_TERMINATION 6
#define TH_STATE_EXIT        7

int th_start_client(void *data)
{
	int cmd_id = 1;
	int err;
	//char cmd_buf[MQ_MSG_SIZE];
	//mq_cmd *cmd=NULL;
	int run;

	server_cfg *cfg = data;
	mq_ntf_mdt *mq = cfg->mq;
	atomic_fetch_add(&cfg->running,1);
	printf("Start client\n");
	printf("Server %d\n",cfg->tid);
	sleep(1);

	//send command wait for response

	run = 1;
	while (run)
	{
		//printf("Send msg\n");
		err = SEND_CMD_OUT(mq,cmd_id,"INIT","NO");
		printf("err = %d\n",err);
		cmd_id += 1;
		run += 1;
		sleep(1);
		if (run == 10)
		{
			break;
		}
	}


	printf("End client\n");
	atomic_fetch_sub( &cfg->running,1);

	return 0;
}


#define EVENT_HND_STACK_SIZE (16*1024)
typedef struct event_handler_cfg {
	/* thread params*/
	void *stack;
	mq_ntf_mdt *mq_listen;
	int mq_num;
	_Atomic int running;

} event_handler_cfg;

/* server_cfg struct as input */
#define EH_STATE_INIT        0
#define EH_STATE_START       1
#define EH_STATE_LISTEN_IN   2
#define EH_STATE_LISTEN_OUT  3
#define EH_STATE_SEND_IN     4
#define EH_STATE_SEND_OUT    5
#define EH_STATE_TERMINATION 6
#define EH_STATE_EXIT        7

/* Thread to reacieve messages and return them back */
int th_event_manager(void *data)
{
	event_handler_cfg *cfg = data;
	atomic_fetch_add(&cfg->running,1);
	mq_ntf_mdt *mq=NULL;
	char *buf = NULL;
	int run;
	int err;
	int mq_event;
	//read any command
	//mq_cmd *cmd = NULL;
	struct mq_attr out_attr, *ptr_out_attr=&out_attr;

	printf("Start event thread\n");
	mq = cfg->mq_listen;
	err = mq_ntf_mplx(mq, 10000);
	if (err == -1)
	{
		printf("Ups something whent wrong\n");
	}

	//get mq attributes
	if (mq_ntf_getattr(mq, MQ_OUT, &ptr_out_attr) == -1)
	{
		printf("Cant get attribute\n");
	}

	buf = malloc(out_attr.mq_msgsize);
	//maybe its not null

	printf("Start event loop\n");
	run = 1;
	while(run)
	{
		//check for messages
		run += 1;
		mq_event = mq_ntf_select(mq);
		switch(mq_event)
		{
		case 0:
			break;
		case 1:
			break;
		case 2:
			if (mq_ntf_read(mq, MQ_OUT, buf, out_attr.mq_msgsize) == -1)
			{
				printf("Cant read message\n");
			} else
			{
				buf[out_attr.mq_msgsize-1] = 0x0;
				printf("Recieve %s\n", buf);
			}
			break;
		default:
			printf("Unknown event type\n");
		}
		sleep(1);
		if (run == 10)
			break;
	}

	printf("End event thread\n");

	atomic_fetch_sub( &cfg->running,1);
	return 0;
}

int main(int argc, char **argv)
{
	int i;

	int cnt_servers,cnt_running;
	server_cfg *cfg_list;
	event_handler_cfg *evhnd_cfg;
	mq_ntf_mdt *mq_array;

	/* Load configuration */
	cnt_servers = SIZEOF_SERVER_LIST;
	for (i=0;i<SIZEOF_SERVER_LIST;i++)
	{
		printf("Load config for server %s\n",server_list[i].server);
	}

	cfg_list = malloc(sizeof(server_cfg)*cnt_servers);
	mq_array = malloc(sizeof(mq_ntf_mdt)*cnt_servers);

	/* For each configuration create listener */

	for (i=0;i<SIZEOF_SERVER_LIST;i++)
	{
		/*initialise server config*/
		server_cfg *srvc = cfg_list+i;
		irc_server_conf *isrvc = &server_list[i];
		srvc->tid = i;
		srvc->stack = malloc(STACK_SIZE); //NULL will brake everything ;)
		srvc->user = isrvc->user;
		srvc->password = isrvc->password;
		srvc->server = isrvc->server;
		srvc->channels = isrvc->channels;
		srvc->port = isrvc->port;
		srvc->ssl = isrvc->ssl;

		//atomic_init( &srvc->running, 1);
		atomic_store(&srvc->running, 0);

		/* initalise posix mq */
		if (0 != mq_ntf_open(&mq_array[i], i))
		{
			printf("Couldnt open mq_ntf_open\n");
		}
		srvc->mq = &mq_array[i];
		//try to drain mq
		mq_ntf_drain(&mq_array[i]);

		/* clone new proc */
		clone(th_start_client, srvc->stack+STACK_SIZE, CLONE_VM|CLONE_FILES, (void *)srvc);
	}

	/* event handler thread */
	evhnd_cfg = malloc(sizeof(event_handler_cfg));
	memset(evhnd_cfg, 0, sizeof(event_handler_cfg));
	evhnd_cfg->stack = malloc(EVENT_HND_STACK_SIZE);
	atomic_store(&evhnd_cfg->running, 0);
	evhnd_cfg->mq_num = cnt_servers;
	evhnd_cfg->mq_listen = mq_array;
	clone(th_event_manager, evhnd_cfg->stack+EVENT_HND_STACK_SIZE, CLONE_VM|CLONE_FILES, (void *)evhnd_cfg);

	/* run until all threads are up */
	cnt_running = 1; 
	while(cnt_running != 0)
	{
		//printf("Count\n");
		/*count how many proceses is running there*/
		int val;
		cnt_running = 0;
		for (i=0;i<cnt_servers;i++)
		{
			val = atomic_load(&cfg_list[i].running);
			if (val != 0)
				cnt_running += 1;
		}
		//PRINT("cnt_running %d\n",cnt_running);
		sleep(1);
	}

	free(cfg_list);

	return 0;
}