/*
 * Add a sent message to sent message list
 */
void pool_add_sent_message(POOL_SENT_MESSAGE *message)
{
	POOL_SENT_MESSAGE *old_msg;
	POOL_SENT_MESSAGE_LIST *msglist;

	if (!session_context)
	{
		pool_error("pool_add_sent_message: session context is not initialized");
		return;
	}

	if (!message)
	{
		pool_debug("pool_add_sent_message: message is NULL");
		return;
	}

	old_msg = pool_get_sent_message(message->kind, message->name);
	msglist = &session_context->message_list;

	if (old_msg)
	{
		if (message->kind == 'B')
			pool_debug("pool_add_sent_message: portal \"%s\" already exists",
					   message->name);
		else
			pool_debug("pool_add_sent_message: prepared statement \"%s\" already exists",
					   message->name);

		if (*message->name == '\0')
			pool_remove_sent_message(old_msg->kind, old_msg->name);
		else
			return;
	}

	if (msglist->size == msglist->capacity)
	{
		msglist->capacity *= 2;
		msglist->sent_messages = realloc(msglist->sent_messages,
										 sizeof(POOL_SENT_MESSAGE *) * msglist->capacity);
		if (!msglist->sent_messages)
		{
			pool_error("pool_add_sent_message: realloc failed: %s", strerror(errno));
			exit(1);
		}
	}

	msglist->sent_messages[msglist->size++] = message;
}

/*
 * Walker function to find a function call which is supposed to write
 * database.
 */
static bool function_call_walker(Node *node, void *context)
{
	SelectContext	*ctx = (SelectContext *) context;

	if (node == NULL)
		return false;

	if (IsA(node, FuncCall))
	{
		FuncCall *fcall = (FuncCall *)node;
		char *fname;
		int length = list_length(fcall->funcname);

		if (length > 0)
		{
			if (length == 1)	/* no schema qualification? */
			{
				fname = strVal(linitial(fcall->funcname));
			}
			else
			{
				fname = strVal(lsecond(fcall->funcname));		/* with schema qualification */
			}

			pool_debug("function_call_walker: function name: %s", fname);

			/*
			 * Check white list if any.
			 */
			if (pool_config->num_white_function_list > 0)
			{
				/* Search function in the white list regex patterns */
				if (pattern_compare(fname, WHITELIST, "white_function_list") == 1) {
					/* If the function is found in the white list, we can ignore it */
					return raw_expression_tree_walker(node, function_call_walker, context);
				}
				/*
				 * Since the function was not found in white list, we
				 * have found a writing function.
				 */
				ctx->has_function_call = true;
				return false;
			}

			/*
			 * Check black list if any.
			 */
			if (pool_config->num_black_function_list > 0)
			{
				/* Search function in the black list regex patterns */
				if (pattern_compare(fname, BLACKLIST, "black_function_list") == 1) {
					/* Found. */
					ctx->has_function_call = true;
					return false;
				}
			}
		}
	}
	return raw_expression_tree_walker(node, function_call_walker, context);
}

void pool_session_remove_all_cursors()
{
	if(session_context == NULL) return;
	pool_debug("Removing all cursors");

	memset(&session_context->opened_cursors, 0, sizeof(session_context->opened_cursors));
}

int
wd_hb_recv(int sock, WdHbPacket * pkt)
{
	int rtn;
	struct sockaddr_in senderinfo;
	socklen_t addrlen;
	WdHbPacket buf;

	addrlen = sizeof(senderinfo);

	rtn = recvfrom(sock, &buf, sizeof(WdHbPacket), 0,
	               (struct sockaddr *)&senderinfo, &addrlen);
	if (rtn < 0)
	{	
		pool_error("wd_hb_recv: failed to receive packet");
		return WD_NG;
	}
	else if (rtn == 0)
	{
		pool_error("wd_hb_recv: received zero bytes");
		return WD_NG;
	}
	else
	{
		pool_debug("wd_hb_recv: received %d byte packet", rtn);
	}

	ntoh_wd_hb_packet(pkt, &buf);

	return WD_OK;
}

/* --------------------------------
 * write_cache() - append result data to buffer
 *
 * returns 0 on success, -1 otherwise
 * --------------------------------
 */
static int
write_cache(void *buf, int len)
{
	int required_len;

	if (len < 0)
		return -1;

	required_len = query_cache_info->cache_offset + len;
	if (required_len > query_cache_info->cache_size)
	{
		char *ptr;

		required_len = query_cache_info->cache_size * 2;
		ptr = (char *)realloc(query_cache_info->cache, required_len);
		if (malloc_failed(ptr))
			return -1;

		query_cache_info->cache = ptr;
		query_cache_info->cache_size = required_len;

		pool_debug("pool_query_cache: extended cache buffer size to %d", query_cache_info->cache_size);
	}

	memcpy(query_cache_info->cache + query_cache_info->cache_offset, buf, len);
	query_cache_info->cache_offset += len;

	return 0;
}

/* --------------------------------
 * ForwardCacheToFrontend - simply forwards cached data to the frontend
 *
 * since the cached data passed from the caller is in escaped binary string
 * format, unescape it and send it to the frontend appending 'Z' at the end.
 * returns 0 on success, -1 otherwise.
 * --------------------------------
 */
static int ForwardCacheToFrontend(POOL_CONNECTION *frontend, char *cache, char tstate)
{
	int sendlen;
	size_t sz;
	char *binary_cache = NULL;

	binary_cache = (char *)PQunescapeBytea((unsigned char *)cache, &sz);
	sendlen = (int) sz;
	if (malloc_failed(binary_cache))
		return -1;

	pool_debug("ForwardCacheToFrontend: query cache found (%d bytes)", sendlen);

	/* forward cache to the frontend */
	pool_write(frontend, binary_cache, sendlen);

	/* send ReadyForQuery to the frontend*/
	pool_write(frontend, "Z", 1);
	sendlen = htonl(5);
	pool_write(frontend, &sendlen, sizeof(int));
	if (pool_write_and_flush(frontend, &tstate, 1) < 0)
	{
		pool_error("pool_query_cache_lookup: error while writing data to the frontend");
		PQfreemem(binary_cache);
		return -1;
	}

	PQfreemem(binary_cache);
	return 0;
}

/*
 * Forget transaction isolation mode
 */
void pool_unset_transaction_isolation(void)
{
	if (!session_context)
	{
		pool_error("pool_unset_transaction_isolation: session context is not initialized");
		return;
	}
	pool_debug("pool_unset_transaction_isolation: done");
	session_context->transaction_isolation = POOL_UNKNOWN;
}

/*
 * The command in progress has succeeded.
 */
void pool_set_command_success(void)
{
	if (!session_context)
	{
		pool_error("pool_set_command_success: session context is not initialized");
		return;
	}
	pool_debug("pool_set_command_success: done");
	session_context->command_success = true;
}

/*
 * Set transaction isolation mode
 */
void pool_set_transaction_isolation(POOL_TRANSACTION_ISOLATION isolation_level)
{
	if (!session_context)
	{
		pool_error("pool_set_transaction_isolation: session context is not initialized");
		return;
	}
	pool_debug("pool_set_transaction_isolation: done");
	session_context->transaction_isolation = isolation_level;
}

/*
 * Call pgpool_recovery() function.
 */
static int exec_recovery(PGconn *conn, BackendInfo *backend, char stage)
{
	PGresult *result;
	char *hostname;
	char *script;
	int r;

	if (strlen(backend->backend_hostname) == 0 || *(backend->backend_hostname) == '/')
		hostname = "localhost";
	else
		hostname = backend->backend_hostname;

	script = (stage == FIRST_STAGE) ?
		pool_config->recovery_1st_stage_command : pool_config->recovery_2nd_stage_command;

	if (script == NULL || strlen(script) == 0)
	{
		/* do not execute script */
		return 0;
	}

	snprintf(recovery_command,
			 sizeof(recovery_command),
			 "SELECT pgpool_recovery('%s', '%s', '%s')",
			 script,
			 hostname,
			 backend->backend_data_directory);

	pool_log("starting recovery command: \"%s\"", recovery_command);

	pool_debug("exec_recovery: start recovery");
	result = PQexec(conn, recovery_command);
	r = (PQresultStatus(result) !=  PGRES_TUPLES_OK);
	if (r != 0)
	{
		pool_error("exec_recovery: %s command failed at %s",
				   script,
				   (stage == FIRST_STAGE) ? "1st stage" : "2nd stage");
	}
	PQclear(result);
	pool_debug("exec_recovery: finish recovery");
	return r;
}

/*
 * We don't have a write query in this transaction yet.
 */
void pool_unset_writing_transaction(void)
{
	if (!session_context)
	{
		pool_error("pool_unset_writing_transaction: session context is not initialized");
		return;
	}
	pool_debug("pool_unset_writing_transaction: done");
	session_context->writing_transaction = false;
}

int
wd_stand_for_master(void)
{
	int rtn;

	/* send stand for master packet */
	pool_debug("wd_stand_for_master: send the packet to be the new master");
	rtn = wd_send_packet_no(WD_STAND_FOR_MASTER);
	return rtn;
}

/*
 * Error occurred in this transaction.
 */
void pool_set_failed_transaction(void)
{
	if (!session_context)
	{
		pool_error("pool_set_failed_transaction: session context is not initialized");
		return;
	}
	pool_debug("pool_set_failed_transaction: done");
	session_context->failed_transaction = true;
}

static RETSIGTYPE exit_handler(int sig)
{
	int i;

	POOL_SETMASK(&AuthBlockSig);

	/*
	 * this could happen in a child process if a signal has been sent
	 * before resetting signal handler
	 */
	if (getpid() != mypid)
	{
		pool_debug("exit_handler: I am not parent");
		POOL_SETMASK(&UnBlockSig);
		pool_shmem_exit(0);
		exit(0);
	}

	if (sig == SIGTERM)
		pool_log("received smart shutdown request");
	else if (sig == SIGINT)
		pool_log("received fast shutdown request");
	else if (sig == SIGQUIT)
		pool_log("received immediate shutdown request");
	else
	{
		pool_error("exit_handler: unknown signal received %d", sig);
		POOL_SETMASK(&UnBlockSig);
		return;
	}

	exiting = 1;

	for (i = 0; i < pool_config->num_init_children; i++)
	{
		pid_t pid = pids[i].pid;
		if (pid)
		{
			kill(pid, sig);
		}
	}

	kill(pcp_pid, sig);

	POOL_SETMASK(&UnBlockSig);

	while (wait(NULL) > 0)
		;

	if (errno != ECHILD)
		pool_error("wait() failed. reason:%s", strerror(errno));

	pids = NULL;
	myexit(0);
}

int
wd_declare(void)
{
	int rtn;

	/* send declare new master packet */
	pool_debug("wd_declare: send the packet to declare the new master");

	rtn = wd_send_packet_no(WD_DECLARE_NEW_MASTER);
	return rtn;
}

/*
 * signal handler for SIGUSR1
 * close all idle connections
 */
static RETSIGTYPE close_idle_connection(int sig)
{
	int i, j;
	POOL_CONNECTION_POOL *p = pool_connection_pool;
	ConnectionInfo *info;

	pool_debug("child receives close connection request");

	for (j=0;j<pool_config->max_pool;j++, p++)
	{
		if (!MASTER_CONNECTION(p))
			continue;
 		if (!MASTER_CONNECTION(p)->sp)
			continue;
		if (MASTER_CONNECTION(p)->sp->user == NULL)
			continue;

		if (MASTER_CONNECTION(p)->closetime > 0)		/* idle connection? */
		{
			pool_debug("close_idle_connection: close idle connection: user %s database %s", MASTER_CONNECTION(p)->sp->user, MASTER_CONNECTION(p)->sp->database);
			pool_send_frontend_exits(p);

			for (i=0;i<NUM_BACKENDS;i++)
			{
				if (!VALID_BACKEND(i))
					continue;

				if (i == 0)
				{
					/* only first backend allocated the memory for the start up packet */
					pool_free_startup_packet(CONNECTION_SLOT(p, i)->sp);
				}
				pool_close(CONNECTION(p, i));
			}
			info = p->info;
			memset(p, 0, sizeof(POOL_CONNECTION_POOL));
			p->info = info;
			memset(p->info, 0, sizeof(ConnectionInfo));
		}
	}
}

/*
 * Walker function to extract table oids from SELECT statement.
 */
static bool
select_table_walker(Node *node, void *context)
{
	SelectContext	*ctx = (SelectContext *) context;
	int num_oids;

	if (node == NULL)
		return false;

	if (IsA(node, RangeVar))
	{
		RangeVar *rgv = (RangeVar *)node;
		char *table;
		int oid;
		char *s;

		table = make_table_name_from_rangevar(rgv);
		oid = pool_table_name_to_oid(table);

		if (oid)
		{
			if (POOL_MAX_SELECT_OIDS <= ctx->num_oids)
			{
				pool_debug("select_table_walker: number of oids exceeds");
				return false;
			}

			num_oids = ctx->num_oids++;

			ctx->table_oids[num_oids] = oid;
			s = strip_quote(table);
			strlcpy(ctx->table_names[num_oids], s, POOL_NAMEDATALEN);
			free(s);

			pool_debug("select_table_walker: ctx->table_names[%d] = %s",
			           num_oids, ctx->table_names[num_oids]);
		}
	}

	return raw_expression_tree_walker(node, select_table_walker, context);
}

/*
 * Walker function to find non immutable function call.
 */
static bool non_immutable_function_call_walker(Node *node, void *context)
{
	SelectContext	*ctx = (SelectContext *) context;

	if (node == NULL)
		return false;

	if (IsA(node, FuncCall))
	{
		FuncCall *fcall = (FuncCall *)node;
		char *fname;
		int length = list_length(fcall->funcname);

		if (length > 0)
		{
			if (length == 1)	/* no schema qualification? */
			{
				fname = strVal(linitial(fcall->funcname));
			}
			else
			{
				fname = strVal(lsecond(fcall->funcname));		/* with schema qualification */
			}

			pool_debug("non_immutable_function_call_walker: function name: %s", fname);

			/* Check system catalog if the function is immutable */
			if (is_immutable_function(fname) == false)
			{
				/* Non immutable function call found */
				ctx->has_non_immutable_function_call = true;
				return false;
			}
		}
	}
	else if (IsA(node, TypeCast))
	{
		/* CURRENT_DATE, CURRENT_TIME, LOCALTIMESTAMP, LOCALTIME etc.*/
		TypeCast	*tc = (TypeCast *) node;

		if ((isSystemType((Node *) tc->typeName, "date") ||
			 isSystemType((Node *) tc->typeName, "timestamp") ||
			 isSystemType((Node *) tc->typeName, "timestamptz") ||
			 isSystemType((Node *) tc->typeName, "time") ||
			 isSystemType((Node *) tc->typeName, "timetz")))
		{
			ctx->has_non_immutable_function_call = true;
			return false;
		}
	}

	return raw_expression_tree_walker(node, non_immutable_function_call_walker, context);
}

/**
 * Get average round-trip time of ping result.
 */
static double
get_result (char * ping_data)
{
	char * sp = NULL;
	char * ep = NULL;
	int i;
	double msec = 0;

	if (ping_data == NULL)
	{
		pool_error("get_result: no ping data");
		return -1;
	}

	pool_debug("get_result: ping data: %s", ping_data);

	/*
	 skip result until average data
	 typical result of ping is as follows,
	 "rtt min/avg/max/mdev = 0.045/0.045/0.046/0.006 ms"
	 we can find the average data beyond the 4th '/'.
	 */
	sp = ping_data;
	for ( i = 0 ; i < 4 ; i ++)
	{
		sp = strchr(sp,'/');
		if (sp == NULL)
		{
			return -1;
		}
		sp ++;
	}

	ep = strchr (sp,'/');
	if (ep == NULL)
	{
		return -1;
	}

	*ep = '\0';
	errno = 0;

	/* convert to numeric data from text */
	msec = strtod(sp,(char **)NULL);

	if (errno != 0)
	{
		pool_error("get_result: strtod() failed. reason: %s", strerror(errno));
		return -1;
	}

	return msec;
}

/*
* worker child main loop
*/
void do_worker_child(void)
{
	pool_debug("I am %d", getpid());

	/* Identify myself via ps */
	init_ps_display("", "", "", "");
	set_ps_display("worker process", false);

	/* set up signal handlers */
	signal(SIGALRM, SIG_DFL);
	signal(SIGTERM, my_signal_handler);
	signal(SIGINT, my_signal_handler);
	signal(SIGHUP, reload_config_handler);
	signal(SIGQUIT, my_signal_handler);
	signal(SIGCHLD, SIG_IGN);
	signal(SIGUSR1, my_signal_handler);
	signal(SIGUSR2, SIG_IGN);
	signal(SIGPIPE, SIG_IGN);

	/* Initialize my backend status */
	pool_initialize_private_backend_status();

	/* Initialize per process context */
	pool_init_process_context();

	for (;;)
	{
		CHECK_REQUEST;

		if (pool_config->sr_check_period <= 0)
		{
			sleep(30);
		}

		/*
		 * If streaming replication mode, do time lag checking
		 */

		if (pool_config->sr_check_period > 0 && MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
		{
			/* Check and establish persistent connections to the backend */
			establish_persistent_connection();

			/* Do replication time lag checking */
			check_replication_time_lag();

			/* Discard persistent connections */
			discard_persistent_connection();
		}
		sleep(pool_config->sr_check_period);
	}
	exit(0);
}