/*
	This file is part of Bolixo.

	Bolixo is free software: you can redistribute it and/or modify
	it under the terms of the GNU General Public License as published by
	the Free Software Foundation, either version 3 of the License, or
	(at your option) any later version.

	Bolixo is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public License
	along with Bolixo.  If not, see <https://www.gnu.org/licenses/>.
*/
/*
	Publish content to other bolixo nodes
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <errno.h>
#include <signal.h>
#include <syslog.h>
#include <unistd.h>
#include <tlmpsql.h>
#include <tlmplib.h>
#include <tlmpnet.h>
#include <trlitool.h>
#include <deque>
#include <memory>
#include <string>
#include "filesystem.h"
#include "bolixo.h"
#include "bolixo.m"
#define INSTRUMENT_DONOTOPEN
#include "instrument.h"


using namespace std;
static DEBUG_KEY D_PROTO ("proto","Protocol information");

enum CONNECT_TYPE { TYPE_NONE, TYPE_CONTROL, TYPE_CLIENT, TYPE_WORKER};

struct HANDLE_INFO: public ARRAY_OBJ{
	CONNECT_TYPE type;
	REQUEST_INFO req;
	HANDLE_INFO(){
		type = TYPE_NONE;
	}
};
#include "proto/publishd_control.protoh"
#include "proto/publishd_client.protoh"

#include "proto/bod_client.protodef"
#define webapi_test_NOTNEED
#define webapi_login_NOTNEED
#define webapi_addfile_NOTNEED
#define webapi_addfile_bob_NOTNEED
#define webapi_delfile_NOTNEED
#define webapi_undelete_NOTNEED
#define webapi_modifyfile_NOTNEED
#define webapi_modifyfile_bob_NOTNEED
#define webapi_rename_NOTNEED
#define webapi_copy_NOTNEED
#define webapi_readfile_NOTNEED
#define webapi_readfile_bob_NOTNEED
#define webapi_readmore_NOTNEED
#define webapi_mkdir_NOTNEED
#define webapi_rmdir_NOTNEED
#define webapi_listdir_NOTNEED
#define webapi_stat_NOTNEED
#define webapi_set_access_NOTNEED
#define webapi_markview_NOTNEED
#define webapi_list_inboxes_NOTNEED
#define webapi_list_msgs_NOTNEED
#define webapi_sendmsg_NOTNEED
#define webapi_sendmsg_project_NOTNEED
#define webapi_replymsg_NOTNEED
#define webapi_replymsg_project_NOTNEED
#define webapi_sendattach_NOTNEED
#define webapi_sendtalk_file_NOTNEED
#define webapi_list_talk_NOTNEED
#define webapi_public_listdir_NOTNEED
#define webapi_public_readfile_NOTNEED
#define webapi_public_list_talk_NOTNEED
#define webapi_systempubkey_NOTNEED
#define webapi_verifysign_NOTNEED
#define webapi_getpubkey_NOTNEED
#define webapi_registernode_NOTNEED
#define webapi_remote_interest_set_NOTNEED
#define webapi_remote_interest_unset_NOTNEED
#define webapi_nodelogin_NOTNEED
#define webapi_nodepass_NOTNEED
#define webapi_config_read_NOTNEED
#define webapi_config_write_NOTNEED
#define webapi_contact_manage_NOTNEED
#define webapi_contact_request_NOTNEED
#define webapi_contact_list_NOTNEED
#define webapi_list_contacts_NOTNEED
#define webapi_list_lists_NOTNEED
#define webapi_list_groups_NOTNEED
#include "proto/webapi.protoch"


#define bo_keysd_control_genkey_NOTNEED
#define bo_keysd_control_setpassphrase_NOTNEED
#define bo_keysd_control_checkpassphrase_NOTNEED
#define bo_keysd_control_quit_NOTNEED
#define bo_keysd_control_debug_NOTNEED
#define bo_keysd_control_debugfile_NOTNEED
#define bo_keysd_control_runstatus_NOTNEED
#include "proto/bo-keysd_control.protoch"

struct PUBLISH_MESSAGE{
	string user;
	string name;
	unsigned userid;
	unsigned dirid;
	unsigned fileid;
	string modified;
	PUBLISH_MESSAGE(const char *_user, unsigned _userid, unsigned _dirid, unsigned _fileid, const char *_modified, const char *_name){
		user = _user;
		userid = _userid;
		dirid = _dirid;
		fileid = _fileid;
		modified = _modified;
		name = _name;
	}
	PUBLISH_MESSAGE(){
		userid = (unsigned)-1;
		dirid  = (unsigned)-1;
		fileid = (unsigned)-1;
	}
};

<mod>
static int publishd_remotelogin (
	CONNECT_HTTP_INFO &hcon,
	PARAM_STRING remote_user,
	const char *keysdsock,
	unsigned userid,
	string &sessionid)
{
	glocal int ret = -1;
	glocal string *sessionid = &sessionid;
	glocal CONNECT_HTTP_INFO *hcon = &hcon;
	glocal const char *keysdsock = keysdsock;
	glocal const char *remote_user = remote_user.ptr;
	glocal unsigned userid = userid;
	<call webapi_remotelogin>(hcon,remote_user);
	<f ok>
		if (!success){
			tlmp_error ("remotelogin failed for %s: %s\n"
				,glocal.remote_user,msg);
		}else{
			CONNECT_INFO con;
			con.port = glocal.keysdsock;
			glocal string sign;
			*glocal.sessionid = sessionid;
			<call bo_keysd_control_sign>(con,glocal.userid,BOB_TYPE(sessionid,strlen(sessionid),false));
			<f ok>
				if (status == ERR_CODE_NONE) glocal.sign = sign;
			</f>
			</call>
			// tlmp_error ("sign=%s\n",glocal.sign.c_str());
			if (glocal.sign.size() > 0){
				<call webapi_remotepass>(*glocal.hcon,sessionid,glocal.remote_user,glocal.sign);
				<f ok>
					if (success) glocal.ret = 0;
				</f>
				</call>
			}
		}
	</f>
	</call>
	return glocal.ret;
}
</mod>

struct STATS{
	unsigned nbcopied;
	unsigned nbfailed;
	unsigned nbtried;
	STATS(){
		nbcopied = nbfailed = nbtried = 0;
	}
};

<mod>
static void publishd_doone (
	_F_TCPSERVER_V1 &c,
	pid_t &pid,
	deque<PUBLISH_MESSAGE> &messages,
	NSQL &sq,
	const char *our_node,
	const char *keysdsock)
{
	glocal NSQL *sq = &sq;
	glocal const char *our_node = our_node;
	glocal const char *keysdsock = keysdsock;
	if (messages.size() > 0){
		int tb[2];
		if (pipe(tb)==-1){
			tlmp_error ("Can't setup pipe (%s)\n",strerror(errno));
		}else{
			glocal STATS stats;
			glocal PUBLISH_MESSAGE todo = messages.front();
			messages.pop_front();
			pid = fork();
			if (pid == (pid_t)0){
				// Do the job
				close (tb[0]);
				glocal BOB_TYPE content;
				glocal string sign;
				glocal FILE *fin = NULL;
				glocal bool data_loaded=false;
				glocal bool some_errors = false;
				<call sql_query>(*glocal.sq,"select nodename from interests_remote where userid=%u",glocal.todo.userid);
				<f onerow>
					//tlmp_error ("nodename=%s\n",row[0]);
					glocal CONNECT_HTTP_INFO hcon;
					// Load the data only once since it is copied to potentially more than one node
					if (!glocal.data_loaded){
						<call sql_query>(*glocal.sq,"select content,signature from files where id=%u and modified='%s'"
							,glocal.todo.fileid,glocal.todo.modified.c_str());
						<f onerow>
							if (row[1] != NULL) glocal.sign = row[1];
							if (row[0] != NULL){
								glocal.content.setbuffer((const void*)row[0],strlen(row[0]),true);
							}else{
								// The content is in a file
								string fname = fs_createpath(glocal.todo.fileid,glocal.todo.modified);
								glocal.fin = fopen (fname.c_str(),"r");
								if (glocal.fin == NULL){
									tlmp_error ("Can't open content file %s (%s)\n",fname.c_str(),strerror(errno));
									glocal.some_errors = true;
								}
							}
						</f>
						<f empty>
							tlmp_error ("No content for message: user=%s dirid=%u userid=%u fileid=%u modified=%s\n"
								,glocal.todo.user.c_str(),glocal.todo.dirid,glocal.todo.userid
								,glocal.todo.fileid,glocal.todo.modified.c_str());
							glocal.some_errors = true;
						</f>
						</call>
						glocal.data_loaded = true;
					}
					if (!glocal.some_errors && glocal.hcon.init (row[0]) != -1){
						glocal string remote_user = string_f("%s@%s",glocal.todo.user.c_str(),glocal.our_node);
						glocal string sessionid;
						glocal.stats.nbtried++;
						if (publishd_remotelogin(glocal.hcon,glocal.remote_user,glocal.keysdsock,glocal.todo.userid,glocal.sessionid)!=-1){
							glocal bool error = false;
							static vector<string> empty;
							if (glocal.fin == NULL){
								<call webapi_sendtalk>(glocal.hcon,glocal.sessionid,"",empty,"public",glocal.remote_user
									,glocal.content,false,glocal.todo.name,glocal.sign,"");
								<f ok>
									if (!success){
										tlmp_error ("message to %s internal=%d success=%d msg=%s\n"
											,glocal.remote_user.c_str(),internal_error,success,msg);
										glocal.error = true;
									}
								</f>
								</call>
							}else{
								// The content is in a file
								glocal string handle;
								char buf[REQ_CONTENT_CHUNK];
								int n;
								bool more = false;
								rewind(glocal.fin);
								while (!glocal.error && (n=fread(buf,1,sizeof(buf),glocal.fin))>0){
									more = n==sizeof(buf);
									BOB_TYPE content((const void*)buf,n,false);
									if (glocal.handle.size() == 0){
										<call webapi_sendtalk>(glocal.hcon,glocal.sessionid,"",empty,"public",glocal.remote_user
											,content,more,glocal.todo.name,glocal.sign,"");
										<f ok>
											glocal.handle = handle;
											if(!success){
												glocal.error = true;
												tlmp_error ("copying to %s internal=%d success=%d msg=%s\n"
													,glocal.remote_user.c_str(),internal_error,success,msg);
											}
										</f>
										</call>
									}else{
										<call webapi_appendfile>(glocal.hcon,glocal.sessionid,glocal.handle,content,more);
										<f ok>
											if(!success){
												glocal.error = true;
												tlmp_error ("appending to %s internal=%d success=%d msg=%s\n"
													,glocal.remote_user.c_str(),internal_error,success,msg);
											}
										</f>
										</call>
									}
								}
								if (more){
									BOB_TYPE content((const void*)"",0,false);
									<call webapi_appendfile>(glocal.hcon,glocal.sessionid,glocal.handle,content,false);
									<f ok>
										//tlmp_error ("internal=%d success=%d msg=%s\n",internal_error,success,msg);
									</f>
									</call>
								}
							}
							if (glocal.error){
								glocal.stats.nbfailed++;
							}else{
								glocal.stats.nbcopied++;
							}
							<call webapi_logout>(glocal.hcon,glocal.sessionid);
							<f ok>
							</f>
							</call>
						}
					}
					if (glocal.some_errors) end = true;
				</f>
				</call>
				if (glocal.fin != NULL) fclose (glocal.fin);
				write (tb[1],&glocal.stats,sizeof(glocal.stats));	
				_exit (0);
			}else if (pid == (pid_t)-1){
				tlmp_error ("Can't fork (%s)\n",strerror(errno));
			}else{
				HANDLE_INFO *n = new HANDLE_INFO;
				n->type = TYPE_WORKER;
				c.inject (tb[0],n);
				c.setrawmode(tb[0],true);
				close (tb[1]);
			}
		}
	}
}
</mod>
<mod>
int main (int argc, char *argv[])
{
	glocal int ret = -1;
	glocal const char *data_dbserv = "localhost";
	glocal const char *data_dbname = "files";
	glocal const char *data_dbuser = NULL;
	glocal const char *control = "/var/run/publishd.sock";
	glocal const char *clientsock = "/tmp/publishd_client.sock";
	glocal const char *user = "bolixo";
	glocal bool daemon = false;
	glocal const char *admin_secretfile = "/etc/bolixo/secrets.admin";
	glocal const char *pidfile = "/var/run/publishd.pid";
	glocal const char *hostname = NULL;
	glocal const char *keysdsock = "/dev/keysd.sock";

	static const char *tbdic[]={"bolixo","tlmpsql",NULL};
	glocal.ret = <call tlmpprogram>(argc,argv,tbdic);
	<f init>
		setproginfo ("publishd",VERSION,"Spread user messages to other bolixo nodes");
		setgrouparg ("Networking");
		setarg ('c',"control","Unix socket for publishd-control",glocal.control,false);
		setarg ('C',"clientsock","Unix socket for publishd-client",glocal.clientsock,false);
		setgrouparg ("Database");
		setarg (' ',"dbserv","Database server",glocal.data_dbserv,false);
		setarg (' ',"dbname","Database name",glocal.data_dbname,false);
		setarg (' ',"dbuser","Database user",glocal.data_dbuser,true);
		setgrouparg ("Misc.");
		setarg (' ',"hostname",MSG_U(O_HOSTNAME,"Host name"),glocal.hostname,true);
		setarg (' ',"keysdsock","Port to reach the bo-keysd server",glocal.keysdsock,false);
		setarg (' ',"admin_secrets",MSG_R(O_ADMINSECRETS),glocal.admin_secretfile,false);
		setarg (' ',"user","Run the program as this user",glocal.user,false);
		setarg (' ',"daemon","Run in background",glocal.daemon,false);
		setarg (' ',"pidfile","File holding the PID of the process",glocal.pidfile,false);
	</f>
	<f showerror>
		if (glocal.daemon){
			syslog (LOG_ERR,"%s",msg);
		}else{
			fprintf (stderr,"%s",msg);
		}
	</f>
	<f showwarn>
		if (glocal.daemon){
			syslog (LOG_WARNING,"%s",msg);
		}else{
			fprintf (stderr,"%s",msg);
		}
	</f>
	<f main_noarg>
		int ret = -1;
		glocal STATS stats;
		glocal unsigned messages_sent = 0;
		glocal string controlport = string_f("unix:%s",glocal.control);
		glocal string clientport = string_f("unix:%s",glocal.clientsock);
		glocal deque<PUBLISH_MESSAGE> messages;
		glocal map<string,string> admin_secrets;
		glocal pid_t pid = (pid_t)-1;
		glocal NSQL *sub_sq = NULL;
		const char *passwd = getenv("PUBLISHD_PWD");
		if (passwd == NULL){
			tlmp_error ("Can't get database password from environment, aborting\n");
			exit (-1);
		} 
		fdpass_readsecrets (glocal.admin_secretfile,glocal.admin_secrets);
		query_setdefaultdb (glocal.data_dbserv,glocal.data_dbname,glocal.data_dbuser,passwd);
		query_getdefaultdb()->showerrormode(true);
		// Connection for the sub-process
		NSQL sub_sq (glocal.data_dbserv,glocal.data_dbname,glocal.data_dbuser,passwd);
		sub_sq.showerrormode(true);
		glocal.sub_sq = &sub_sq;
		signal (SIGCHLD,SIG_IGN);
		<obj TCPSERVER s>(glocal.clientport,5);
		<f newclient>
			HANDLE_INFO *n = new HANDLE_INFO;
			info.data = n;
			// tlmp_error ("port=%s control=%s client=%s\n",info.port,glocal.controlport.c_str(),glocal.clientport.c_str());
			if (string_cmp(info.port,glocal.controlport)==0){
				n->type = TYPE_CONTROL;
			}else if (string_cmp(info.port,glocal.clientport)==0){
				n->req.secret = fdpass_findsecret (glocal.admin_secrets,info.port);
				n->type = TYPE_CLIENT;
			}
		</f>
		<f endclient>
			HANDLE_INFO *n = (HANDLE_INFO*)info.data;
			if (n->type == TYPE_WORKER){
				glocal.pid = (pid_t)-1;
				publishd_doone(*this,glocal.pid,glocal.messages,*glocal.sub_sq,glocal.hostname,glocal.keysdsock);
			}
		</f>
		<f receive>
			debug_printf (D_PROTO,"receive line: %s\n",line);
			HANDLE_INFO *c = (HANDLE_INFO*)info.data;
			static const char *tbtype[]={"none","control request","client request", "worker request"};
			ERROR_PREFIX prefix ("%s: ",tbtype[c->type]);
			if (c->type == TYPE_CONTROL){
				<call publishd_control>(this,c->req,line, info.linelen,endserver, endclient, no,c);
				<f status>
					vector<string> tb;
					tb.push_back(string_f ("Version %s",VERSION));
					tb.push_back(string_f("messages size: %lu",glocal.messages.size()));
					tb.push_back(string_f("sub-process: %s",glocal.pid == (pid_t)-1 ? "not running" : "running"));
					tb.push_back(string_f("hostname: %s",glocal.hostname));
					tb.push_back(string_f("messages_sent: %u",glocal.messages_sent));
					tb.push_back(string_f("stats: nbtried %u nbcopied %u nbfailed %u"
						,glocal.stats.nbtried,glocal.stats.nbcopied,glocal.stats.nbfailed));
					instrument_status(tb);
					rep_status(tb);
				</f>
				<f instrument>
					toggle_instrument_file(on);
				</f>
				<f quit>
					endserver = true;
				</f>
				<f debug>
					if (on){
						debug_seton();
					}else{
						debug_setoff();
					}
				</f>
				<f debugfile>
					debug_setfdebug (filename);
				</f>
				<f help_connect> // connectto port send = lines:v
					glocal const char *send = send;
					glocal vector<string> lines;
					// We want to test publishd connectivity to the outside
					<call tcpconnect>(connectto,port,5);
					<f init>
						sendf ("%s\n",glocal.send);
					</f>
					<f oneline>
						glocal.lines.push_back(line);
						end = true;
					</f>
					<f fail>
						glocal.lines.emplace_back(string_f("fail: %s\n",strerror(errno)));
					</f>
					<f end>
					</f>
					</call>
					rep_help_connect (glocal.lines);
				</f>
				<f test>
					glocal bool dbfiles=false;
					glocal bool keysd=false;
					<call sql_query>("select count(*) from id2name");
					<f onerow>
						glocal.dbfiles=true;
					</f>
					</call>
					CONNECT_INFO con;
					con.port = glocal.keysdsock;
					<call bo_keysd_control_status>(con);
					<f ok>
						glocal.keysd = lines.size() > 0;
					</f>
					</call>
					rep_test (glocal.dbfiles,glocal.keysd);
				</f>
				<f invalid>
					tlmp_error ("Invalid command: %s\n",line);
					endclient = true;
				</f>
				</call>
			}else if (c->type == TYPE_WORKER){
				if (info.linelen == sizeof(STATS)){
					const STATS *st = (const STATS *)line;
					glocal.stats.nbtried += st->nbtried;
					glocal.stats.nbfailed += st->nbfailed;
					glocal.stats.nbcopied += st->nbcopied;
				}
			}else if (c->type == TYPE_CLIENT){
				<call publishd_client>(this,c->req,line,info.linelen, endserver, endclient,no,c);
				<f test>
					glocal bool dbfiles = false;
					glocal bool fsok = file_type(testfile)==0;
					if (glocal.fsok && unlink(testfile)!= -1) glocal.fsok = false;
					<call sql_query>("select count(*) from id2name");
					<f onerow>
						glocal.dbfiles=true;
					</f>
					</call>
					rep_test (glocal.dbfiles,glocal.fsok);
				</f>
				<f sendmessage>
					glocal string msg;
					glocal bool public_view = false;
					// tlmp_error ("sendmessage userid=%u dirid=%u fileid=%u modified=%s\n",userid,dirid,fileid,modified);
					<call sql_query>("select public_view from config where userid=%u",userid);
					<f onerow>
						glocal.public_view = atoi(row[0]);
					</f>
					</call>
					if (glocal.public_view){
						glocal string user;
						<call sql_query>("select name from id2name where userid=%u",userid);
						<f onerow>
							glocal.user = row[0];
						</f>
						</call>
						if (glocal.user.size()==0){
							glocal.msg = string_f("No user for userid=%u",userid);
							tlmp_error ("No user for userid=%u\n",userid);
						}else{
							glocal.messages_sent++;
							glocal.messages.emplace_back(glocal.user.c_str(),userid,dirid,fileid,modified,name);
							if (glocal.pid == (pid_t)-1){
								publishd_doone (glocal.TCPSERVER,glocal.pid,glocal.messages
									,*glocal.sub_sq,glocal.hostname,glocal.keysdsock);
							}
						}
					}	
					if (glocal.msg.size() > 0){
						rep_sendmessage (false,glocal.msg);
					}else{
						rep_sendmessage (true,"");
					}
				</f>
				<f invalid>
					tlmp_error ("Invalid command: %s\n",line);
					endclient = true;
				</f>
				</call>
			}
		</f>
		</obj>
		bool some_errors = false;
		if (fdpass_setcontrol(s,glocal.control,glocal.user)==-1){
			some_errors = true;
		}

		if (!some_errors && s.is_ok()){
			chmod (glocal.clientsock,0666);
			s.setrawmode(true);
			if (glocal.daemon){
				daemon_init(glocal.pidfile,glocal.user);
			}
			open_instrument_file();
			s.loop();
			ret = 0;
		}
		return ret;
	</f>
	</call>
	return glocal.ret;
}
</mod>