#include "CS.h"

NodeP thisP;

DebugP deP;

DataP dataP;

void sighandle(int sig) { return; }

void bindN() {

	LOG(4, "binding...");

	struct addrinfo *sa = (struct addrinfo*)malloc(sizeof(struct addrinfo));

	memset(sa, 0, sizeof(struct addrinfo));

	sa->ai_family = AF_INET;

	sa->ai_socktype = SOCK_STREAM;

	sa->ai_protocol = 0;

	sa->ai_flags = AI_PASSIVE;

	char s[64];

	sprintf(s, "%d", thisP->locPort);

	int e;

	if((e = getaddrinfo(NULL, s, sa, &sa)) != 0) HARDERR(gai_strerror(e));

	GAI(4, sa);

	if((thisP->ssc = socket(sa->ai_family, sa->ai_socktype, sa->ai_protocol)) < 0) SYSERR("socket alloc");

   	int opt = 1;

   	if(setsockopt(thisP->ssc, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) SYSERR("set socket options");

	if(bind(thisP->ssc, sa->ai_addr, sa->ai_addrlen) < 0) SYSERR("bind");

	if(listen(thisP->ssc, 1) < 0) SYSERR("listen");

	LOG(2, "bound to %d", thisP->locPort);

}

void conn(int i, int remPort) {

	DebugT debug, *deP = &debug;

	DEBID("%s to %u", thisP->debug.id, remPort);

	thisP->cliSides[i].remPort = remPort;

	int e;

	LOG(4, "connecting to %u...", remPort);

	int retry = csP->connThreshold;

	struct addrinfo *ai = (struct addrinfo*)malloc(sizeof(struct addrinfo));

	memset(ai, 0, sizeof(struct addrinfo));

	ai->ai_family = AF_INET;

	ai->ai_socktype = SOCK_STREAM;

	ai->ai_protocol = 0;

	ai->ai_flags = 0;

	char port[6];

	sprintf(port, "%d", remPort);

	if((e = getaddrinfo("localhost", port, ai, &ai)) != 0) HARDERR(gai_strerror(e));

	GAI(3, ai);

	if((thisP->cliSides[i].sc = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) < 0) SYSERR("socket alloc");

	while(retry--) {

		if(connect(thisP->cliSides[i].sc, ai->ai_addr, ai->ai_addrlen) < 0) {

			if(errno != ECONNREFUSED) SYSERR("connect");

			usleep(csP->connTO);

		}

		else break;

	}

	if(retry < 1) {

		LOG(0, "connection refused threshold %d reached", csP->connThreshold);

		exit(EXIT_FAILURE);

	}

	if(sem_wait(&(csP->shP->counterSem)) < 0) SYSERR("sem_wait");

	csP->shP->conns++;

	if(sem_post(&(csP->shP->counterSem)) < 0) SYSERR("sem_post");

	socklen_t l = sizeof(struct sockaddr);

	struct sockaddr *sa = malloc(l);

	if(getpeername(thisP->cliSides[i].sc, sa, &l) < 0) SYSERR("getpeername");

	LOG(4, "peer: %s on sc=%d", gpa(sa), thisP->cliSides[i].sc); free(sa);

	if(thisP->ssl) {

		ERR_clear_error();

		if(!(thisP->cliSides[i].sslP = SSL_new(thisP->ctx))) SSLERR("new SSL");

		if(!SSL_set_fd(thisP->cliSides[i].sslP, thisP->cliSides[i].sc)) SSLERR("client SSL set fd");

		if((e = SSL_connect(thisP->cliSides[i].sslP)) < 1) {

			switch(SSL_get_error(thisP->cliSides[i].sslP, e)) {

				case SSL_ERROR_SYSCALL: SYSERR("SSL connect"); break;

				default:	SSLERR("SSL connect"); break;

			}

		}

	}

	LOG(2, "connected via sc=%d after %d retries", thisP->cliSides[i].sc, csP->connThreshold - (retry + 1));

}

void acc(int i) {

	LOG(4, "accepting...");

	if((thisP->srvSides[i].sc = accept(thisP->ssc, NULL, NULL)) < 0) SYSERR("accept");

	socklen_t l = sizeof(struct sockaddr);

	struct sockaddr *sa = malloc(l);

	if(getpeername(thisP->srvSides[i].sc, sa, &l) < 0) SYSERR("getpeername");

	LOG(4, "peer: %s on sc=%d", gpa(sa), thisP->srvSides[i].sc); free(sa);

	if(thisP->ssl) {

		int e;

		ERR_clear_error();

		if(!(thisP->srvSides[i].sslP = SSL_new(thisP->ctx))) SSLERR("new SSL");

		if(!SSL_set_fd(thisP->srvSides[i].sslP, thisP->srvSides[i].sc)) SSLERR("server SSL set fd");

		if((e = SSL_accept(thisP->srvSides[i].sslP)) < 1) {

			switch(SSL_get_error(thisP->srvSides[i].sslP, e)) {

				case SSL_ERROR_SYSCALL: SYSERR("SSL accept"); break;

				default:	SSLERR("SSL accept"); break;

			}

		}

	}

	LOG(2, "accepted");

}

void closeN(int i, nodeside side) {

	SocketP sc;

	if(side) sc = thisP->srvSides; else sc = thisP->cliSides;

	LOG(5, "closing sc=%d...", sc[i].sc);

	if(thisP->ssl) {

		int e;

		if((e = SSL_shutdown(sc[i].sslP)) < 0) SYSERR("SSL shutdown (1)");

		if(!e) {

			LOG(5, "SSL shutdown rc=0");

			if((e = SSL_shutdown(sc[i].sslP)) < 0) {

				switch(SSL_get_error(sc[i].sslP, e)) {

					case SSL_ERROR_SYSCALL:

						if(!(e = ERR_get_error())) {

							if(errno) SYSERR("SSL shutdown (2)");

							break;

						}

						break;

					default:	SSLERR("SSL shutdown (2)"); break;

				}

			}

		}

	}

	close(sc[i].sc);

	LOG(4, "closed sc=%d", sc[i].sc);

	sc[i].sc = -1;

}

void *close_clients() {

	DebugT debug, *deP = &debug;

	DEBID("%s CLOSE clients", thisP->debug.id);

	LOG(5, "start...");

	for (int i = 0; i < thisP->nodes; i++) if(thisP->cliSides[i].sc > -1) closeN(i, client);

	LOG(4, "all clients closed");

	pthread_exit(NULL);

}

void close_node() {

	if(!thisP->closing) {

		thisP->closingThread = 0;

		if(pthread_create(&thisP->closingThread, NULL, &close_clients, NULL) != 0) SYSERR("create thread");

	thisP->closing = 1;

	}

}

int readN(int i) {

	LOG(5, "ready to read len=%d from sc=%d...", dataP->dataLen, thisP->srvSides[i].sc);

	int n, rest = dataP->dataLen;

	void *buf = dataP->contP;

	while(rest > 0) {

		if(thisP->ssl) { if((n = SSL_read(thisP->srvSides[i].sslP, buf, rest)) < 0) SSLERR("read socket"); }

		else { if((n = read(thisP->srvSides[i].sc, buf, rest)) < 0) SYSERR("read socket"); }

		if(n == 0) {

			LOG(4, "read EOF");

			return 0;

		}

		else {

			buf += n; rest -= n;

			if(rest > 0) LOG(5, "partly read %d bytes", n);

		}

	}

	if(sem_wait(&(csP->shP->counterSem)) < 0) SYSERR("sem_wait");

	csP->shP->msgs++;

	if(sem_post(&(csP->shP->counterSem)) < 0) SYSERR("sem_post");

	LOG(5, "read %d from %u", dataP->dataLen, dataP->contP->hdr.listPort);

	return dataP->dataLen;

}

int writeN(int i) {

	DebugT debug, *deP = &debug;

	DEBID("%s to %u", thisP->debug.id, thisP->cliSides[i].remPort);

	LOG(5, "ready to write len=%d to sc=%d...", dataP->dataLen, thisP->cliSides[i].sc);

	int n, rest = dataP->dataLen;

	void *buf = dataP->contP;

	while(rest > 0) {

		if(thisP->ssl) { if((n = SSL_write(thisP->cliSides[i].sslP, buf, rest)) < 0) SSLERR("socket write"); }

		else { if((n = write(thisP->cliSides[i].sc, buf, rest)) < 0) SYSERR("socket write"); }

		buf += n; rest -= n;

		if(rest > 0) LOG(5, "partly written %d bytes", n);

	}

	LOG(5, "written %d", dataP->dataLen);

	return dataP->dataLen;

}

int getN(int i) {

	return readN(i) > 0;

}

int putN(int i) {

	dataP->contP->hdr.listPort = thisP->locPort;

	return writeN(i) > 0;

}

int next_node() {

	int next;

	if(thisP->topo == ring) {

		next = thisP->locPort + 1;

		if(next > thisP->last) next = thisP->first;

	}

	else while((next = thisP->first + thisP->nodes * ((float)random() / RAND_MAX)) == thisP->locPort);

	return next;

}

void forward(int sci) {

	int next, scn;

	char digest[24];

	if(getN(sci)) {

   	LOG(5, "received data from %u", remPortData(dataP));

		if(thisP->kicker) {

			LOG(4, "received from node %u: %s, ttl=%d",

						remPortData(dataP), digest24Data(dataP, digest), ttlData(dataP));

//			if(ttlData(dataP) == 2) sabotageData(dataP);

//			if(ttlData(dataP) == 2) errno=0, SYSERR("signal test");

			if(dttlData(dataP) <= 0) {

      		LOG(1, "received after passing all %s: %s", thisP->topo==mash ? "mashes" : "rings",digest24Data(dataP, digest));

				close_node();

				*(thisP->forw) = 0;

				LOG(4, "leaving forward closing");

				return;

			}

		}

		next = next_node(); scn = next - thisP->first;

		LOG(5, "forwarding to %d, len=%d, ttl=%d --->", next, dataP->dataLen, ttlData(dataP));

		if(thisP->cliSides[scn].sc < 0) conn(scn, next);

		if(*(thisP->forw) && csP->pacing) { LOG(5, "pacing..."); nanosleep(&(csP->pace), NULL); }

		putN(scn);

      LOG(5, "forwarded to %u", next);

	}

	else {

		close_node();

		closeN(sci, server);

	}

	return;

}

void main_loop() {

	sigset_t pacing;

	sigemptyset(&pacing);

   sigaddset(&pacing, PACING);

	union {		// simple select mask debug

	   fd_set rs;

	   uint mask;

	} u;

	int nfds;

	FD_ZERO(&(u.rs)); nfds = 0;

	if(*(thisP->forw)) {

		FD_SET(thisP->ssc, &(u.rs)); if(thisP->ssc >= nfds) nfds = thisP->ssc + 1; }

	while(nfds) {

		struct timeval t = {csP->selTO, 0};

		LOG(5, "selecting, mask=%08x", u.mask);

		int rc;

		rc = select(nfds, &(u.rs), NULL, NULL, &t);

		if(rc < 0 && errno != EINTR) SYSERR("select");

		if(rc > 0) {

			LOG(5, "return from select, mask=%08x", u.mask);

			if(FD_ISSET(thisP->ssc, &(u.rs))) {												// ssc posted: accept & forward

				int i;

				for(i=0; thisP->srvSides[i].sc > -1 && i < thisP->nodes; i++);		// find unused slot for accept

				if(i == thisP->nodes) HARDERR("can't accept, all slots in use");

				LOG(5, "slot for accept=%d", i);

				acc(i);

				forward(i);

			}

			else																						// check which connected socket is posted

				for(int i = 0; i < thisP->nodes; i++)

					if(thisP->srvSides[i].sc > -1 && FD_ISSET(thisP->srvSides[i].sc, &(u.rs))) forward(i);

		}

		FD_ZERO(&(u.rs)); nfds = 0;

		if(*(thisP->forw)) { FD_SET(thisP->ssc, &(u.rs)); if(thisP->ssc >= nfds) nfds = thisP->ssc + 1; }

		for(int i = 0; i < thisP->nodes; i++) {											// mask all connected client side sockets for select

			int sc = thisP->srvSides[i].sc;

			if(sc > -1) { FD_SET(sc, &(u.rs)); if(sc >= nfds) nfds = sc + 1; }

		}

	}

}

void Node(topology topo, int *forw, int port, int first, int n, int ssl) {

	NodeT this;

	thisP = &this;

	deP = &(thisP->debug);

	DEBID("%sSSL %s node %d", ssl ? "" : "non", topo==mash ? "MASH" : "RING", port);

	LOG(4, "initializing...");

	dataP = &thisP->data;

	Data(dataP, deP);

	thisP->topo = topo;

	thisP->locPort = port;

	thisP->first = first;

	thisP->last = first + n - 1;

	thisP->nodes = n;

	thisP->cliSides = malloc(n*sizeof(SocketT));

	thisP->srvSides = malloc(n*sizeof(SocketT));

	for(int k=0; k<n; k++) 	thisP->cliSides[k].sc = thisP->srvSides[k].sc = -1;

	thisP->kicker = (port == first);

	thisP->forw = forw;

	thisP->nodeIdx = port - first;

	thisP->closing = 0;

	thisP->ssl = ssl;

	if(thisP->ssl) {

		char s[64];

		SSL_load_error_strings();

		SSL_library_init();

		LOG(4, "setting SSL contex...");

		if(!(thisP->ctx = SSL_CTX_new(TLS_method()))) SSLERR("new SSL CTX");

		SSL_CTX_set_mode(thisP->ctx, SSL_MODE_AUTO_RETRY);

		SSL_CTX_set_verify(thisP->ctx, SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);

		sprintf(s, "%s/keys/%d.key", csP->ceP, thisP->locPort);

		if(SSL_CTX_use_PrivateKey_file(thisP->ctx, s, SSL_FILETYPE_PEM) != 1) SSLERR("hh's key file");

		sprintf(s, "%s/certs/%d.pem", csP->ceP, thisP->locPort);

		LOG(5, "SSL private key used: %s", s);

		if(SSL_CTX_use_certificate_file(thisP->ctx, s, SSL_FILETYPE_PEM) != 1) SSLERR("hh's cert file");

		LOG(5, "SSL certificate used: %s", s);

		if(SSL_CTX_load_verify_locations(thisP->ctx, NULL, csP->caP) != 1)	SSLERR("hh's thrusted certs path");

	}

	LOG(5, "initalized");

	bindN(thisP);

	if(thisP->kicker) {

		loadData(dataP, csP->ttl, csP->text);

		int sci, next;

		next = next_node(); sci = next - thisP->first;

		char digest[24];

		LOG(1, "KICKER: ready to initial send %s, len=%d to node %d", digest24Data(dataP, digest), dataP->dataLen, next);

		conn(sci, next);

		putN(sci);

	}

	main_loop();

	LOG(5, "closing ssc");

	close(thisP->ssc);

	if(thisP->closing) {		// wait for closing thread

		if(pthread_join(thisP->closingThread, NULL) != 0) SYSERR("join closing thread"); }

	struct sigaction sigact;

	sigfillset(&sigact.sa_mask);

	sigact.sa_handler=sighandle;

	if(sigaction(SIGUSR2,&sigact,NULL) < 0) SYSERR("sigaction");

	if(sem_wait(&csP->shP->counterSem) < 0) SYSERR("sem_wait");

	int active = --csP->shP->act;

	if(sem_post(&csP->shP->counterSem) < 0) SYSERR("sem_post");

	if(active > 0)	pause();

	else kill(0, SIGUSR2);

	int exitRc = EXIT_SUCCESS;

	if(thisP->kicker && !chkData(dataP)) {

		SOFTERR("INPUT AND OUTPUT DIFFER");

		exitRc = EXIT_FAILURE; }

	LOG(2, "ended");

	exit(exitRc);

}