/*
 * src/bin/pgcopydb/ld_apply.c
 *     Implementation of a CLI to copy a database between two Postgres instances
 */

#include <errno.h>
#include <getopt.h>
#include <inttypes.h>
#include <sys/wait.h>
#include <unistd.h>

#include "postgres.h"
#include "postgres_fe.h"
#include "access/xlog_internal.h"
#include "access/xlogdefs.h"

#include "parson.h"

#include "cli_common.h"
#include "cli_root.h"
#include "copydb.h"
#include "env_utils.h"
#include "ld_stream.h"
#include "lock_utils.h"
#include "log.h"
#include "parsing_utils.h"
#include "pidfile.h"
#include "pg_utils.h"
#include "schema.h"
#include "signals.h"
#include "string_utils.h"
#include "summary.h"


GUC applySettings[] = {
	COMMON_GUC_SETTINGS,
	{ "session_replication_role", "'replica'" },
	{ NULL, NULL },
};


/*
 * stream_apply_catchup catches up with SQL files that have been prepared by
 * either the `pgcopydb stream prefetch` command.
 */
bool
stream_apply_catchup(StreamSpecs *specs)
{
	StreamApplyContext context = { 0 };

	if (!stream_apply_setup(specs, &context))
	{
		log_error("Failed to setup for catchup, see above for details");
		return false;
	}

	if (!context.apply)
	{
		/* errors have already been logged */
		return true;
	}

	/*
	 * Our main loop reads the current SQL file, applying all the queries from
	 * there and tracking progress, and then goes on to the next file, until no
	 * such file exists.
	 */
	char currentSQLFileName[MAXPGPATH] = { 0 };

	for (;;)
	{
		strlcpy(currentSQLFileName, context.sqlFileName, MAXPGPATH);

		if (asked_to_stop || asked_to_stop_fast || asked_to_quit)
		{
			break;
		}

		/*
		 * It might be the expected file doesn't exist already, in that case
		 * exit successfully so that the main process may switch from catchup
		 * mode to replay mode.
		 */
		if (!file_exists(context.sqlFileName))
		{
			log_info("File \"%s\" does not exists yet, exit",
					 context.sqlFileName);

			(void) pgsql_finish(&(context.pgsql));
			return true;
		}

		/*
		 * The SQL file exists already, apply it now.
		 */
		if (!stream_apply_file(&context))
		{
			/* errors have already been logged */
			(void) pgsql_finish(&(context.pgsql));
			return false;
		}

		/*
		 * Each time we are done applying a file, we update our progress and
		 * fetch new values from the pgcopydb sentinel. Errors are warning
		 * here, we'll update next time.
		 */
		(void) stream_apply_sync_sentinel(&context);

		/*
		 * When syncing with the pgcopydb sentinel we might receive a new
		 * endpos, and it might mean we're done already.
		 */
		if (!context.reachedEndPos &&
			context.endpos != InvalidXLogRecPtr &&
			context.endpos <= context.previousLSN)
		{
			context.reachedEndPos = true;

			log_info("Apply reached end position %X/%X at %X/%X",
					 LSN_FORMAT_ARGS(context.endpos),
					 LSN_FORMAT_ARGS(context.previousLSN));
		}

		if (context.reachedEndPos)
		{
			/* information has already been logged */
			break;
		}

		if (!computeSQLFileName(&context))
		{
			/* errors have already been logged */
			(void) pgsql_finish(&(context.pgsql));
			return false;
		}

		/*
		 * If we reached the end of the file and the current LSN still belongs
		 * to the same file (a SWITCH did not occur), then we exit so that the
		 * calling process may switch from catchup mode to live replay mode.
		 */
		if (strcmp(context.sqlFileName, currentSQLFileName) == 0)
		{
			log_info("Reached end of file \"%s\" at %X/%X.",
					 currentSQLFileName,
					 LSN_FORMAT_ARGS(context.previousLSN));

			(void) pgsql_finish(&(context.pgsql));
			return true;
		}
	}

	/* we might still have to disconnect now */
	(void) pgsql_finish(&(context.pgsql));

	return true;
}


/*
 * stream_apply_setup does the required setup for then starting to catchup or
 * to replay changes from the SQL input (files or Unix PIPE) to the target
 * database.
 */
bool
stream_apply_setup(StreamSpecs *specs, StreamApplyContext *context)
{
	/* wait until the sentinel enables the apply process */
	if (!stream_apply_wait_for_sentinel(specs, context))
	{
		/* errors have already been logged */
		return false;
	}

	if (!context->apply)
	{
		log_notice("Apply mode is still disabled, quitting now");
		return true;
	}

	if (specs->system.timeline == 0)
	{
		if (!stream_read_context(&(specs->paths),
								 &(specs->system),
								 &(specs->WalSegSz)))
		{
			log_error("Failed to read the streaming context information "
					  "from the source database, see above for details");
			return false;
		}
	}

	context->system = specs->system;
	context->WalSegSz = specs->WalSegSz;

	log_debug("Source database wal_segment_size is %u", context->WalSegSz);
	log_debug("Source database timeline is %d", context->system.timeline);

	if (!setupReplicationOrigin(context,
								&(specs->paths),
								specs->connStrings,
								specs->origin,
								specs->endpos,
								context->apply,
								specs->logSQL))
	{
		log_error("Failed to setup replication origin on the target database");
		return false;
	}

	char *process =
		specs->mode == STREAM_MODE_CATCHUP ? "Catchup-up with" : "Replaying";

	if (context->endpos != InvalidXLogRecPtr)
	{
		if (context->endpos <= context->previousLSN)
		{
			log_info("Current endpos %X/%X was previously reached at %X/%X",
					 LSN_FORMAT_ARGS(context->endpos),
					 LSN_FORMAT_ARGS(context->previousLSN));

			return true;
		}

		log_info("%s changes from LSN %X/%X up to endpos LSN %X/%X",
				 process,
				 LSN_FORMAT_ARGS(context->previousLSN),
				 LSN_FORMAT_ARGS(context->endpos));
	}
	else
	{
		log_info("%s changes from LSN %X/%X",
				 process,
				 LSN_FORMAT_ARGS(context->previousLSN));
	}

	return true;
}


/*
 * stream_apply_wait_for_sentinel fetches the current pgcopydb sentinel values
 * on the source database: the catchup processing only gets to start when the
 * sentinel "apply" column has been set to true.
 */
bool
stream_apply_wait_for_sentinel(StreamSpecs *specs, StreamApplyContext *context)
{
	PGSQL src = { 0 };
	bool firstLoop = true;
	CopyDBSentinel sentinel = { 0 };

	if (!pgsql_init(&src, specs->connStrings->source_pguri, PGSQL_CONN_SOURCE))
	{
		/* errors have already been logged */
		return false;
	}

	/* skip logging the sentinel queries, we log_debug the values fetched */
	src.logSQL = context->logSQL;

	/* make sure context->apply is false before entering the loop */
	context->apply = false;

	while (!context->apply)
	{
		if (asked_to_stop || asked_to_stop_fast || asked_to_quit)
		{
			log_info("Apply process received a shutdown signal "
					 "while waiting for apply mode, "
					 "quitting now");
			return true;
		}

		/* this reconnects on each loop iteration, every 10s by default */
		if (!pgsql_get_sentinel(&src, &sentinel))
		{
			log_warn("Retrying to fetch pgcopydb sentinel values in %ds",
					 CATCHINGUP_SLEEP_MS / 10);
			pg_usleep(CATCHINGUP_SLEEP_MS * 1000);

			continue;
		}

		/* always update our context from the sentinel values */
		context->startpos = sentinel.startpos;
		context->endpos = sentinel.endpos;
		context->apply = sentinel.apply;

		context->previousLSN = sentinel.replay_lsn;

		log_debug("startpos %X/%X endpos %X/%X apply %s",
				  LSN_FORMAT_ARGS(context->startpos),
				  LSN_FORMAT_ARGS(context->endpos),
				  context->apply ? "enabled" : "disabled");

		if (context->apply)
		{
			break;
		}

		if (firstLoop)
		{
			firstLoop = false;

			log_info("Waiting until the pgcopydb sentinel apply is enabled");
		}

		/* avoid buzy looping and avoid hammering the source database */
		pg_usleep(CATCHINGUP_SLEEP_MS * 1000);
	}

	/* when apply was already set on first loop, don't even mention it */
	if (!firstLoop)
	{
		log_info("The pgcopydb sentinel has enabled applying changes");
	}

	return true;
}


/*
 * stream_apply_sync_sentinel sync with the pgcopydb sentinel table, sending
 * the current replay LSN position and fetching the maybe new endpos and apply
 * values.
 */
bool
stream_apply_sync_sentinel(StreamApplyContext *context)
{
	PGSQL src = { 0 };
	CopyDBSentinel sentinel = { 0 };

	if (!pgsql_init(&src, context->connStrings->source_pguri, PGSQL_CONN_SOURCE))
	{
		/* errors have already been logged */
		return false;
	}

	/* limit the amount of logging of the apply process */
	src.logSQL = context->logSQL;

	if (!pgsql_sync_sentinel_apply(&src, context->previousLSN, &sentinel))
	{
		log_warn("Failed to sync progress with the pgcopydb sentinel");
	}

	context->apply = sentinel.apply;
	context->endpos = sentinel.endpos;
	context->startpos = sentinel.startpos;

	return true;
}


/*
 * stream_apply_send_sync_sentinel sends a query to sync with the pgcopydb
 * sentinel table using the libpq async API. Use the associated function
 * stream_apply_fetch_sync_sentinel to check for availability of the result and
 * fetch it when it's available.
 */
bool
stream_apply_send_sync_sentinel(StreamApplyContext *context)
{
	PGSQL *src = &(context->src);

	if (context->sentinelQueryInProgress)
	{
		log_error("BUG: stream_apply_send_sync_sentinel already in progress");
		return false;
	}

	/* we're going to keep the connection around */
	context->src.connectionStatementType = PGSQL_CONNECTION_MULTI_STATEMENT;

	/* limit the amount of logging of the apply process */
	src->logSQL = true;

	if (!pgsql_send_sync_sentinel_apply(src, context->previousLSN))
	{
		log_error("Failed to sync progress with the pgcopydb sentinel");
		return false;
	}

	context->sentinelSyncTime = time(NULL);
	context->sentinelQueryInProgress = true;

	return true;
}


/*
 * stream_apply_fetch_sync_sentinel checks to see if the result of the
 * pgcopydb sentinel sync query is available and fetches it then.
 */
bool
stream_apply_fetch_sync_sentinel(StreamApplyContext *context)
{
	PGSQL *src = &(context->src);

	bool retry;
	CopyDBSentinel sentinel = { 0 };

	if (!pgsql_fetch_sync_sentinel_apply(src, &retry, &sentinel))
	{
		log_error("Failed to fetch sentinel update query results");
		return false;
	}

	if (!retry)
	{
		context->sentinelQueryInProgress = false;

		/* also disconnect between async queries */
		(void) pgsql_finish(&(context->src));

		context->apply = sentinel.apply;
		context->endpos = sentinel.endpos;
		context->startpos = sentinel.startpos;
		context->replay_lsn = sentinel.replay_lsn;
	}

	return true;
}


/*
 * stream_apply_file connects to the target database system and applies the
 * given SQL file as prepared by the stream_transform_file function.
 */
bool
stream_apply_file(StreamApplyContext *context)
{
	StreamContent content = { 0 };
	long size = 0L;

	strlcpy(content.filename, context->sqlFileName, sizeof(content.filename));

	if (!read_file(content.filename, &(content.buffer), &size))
	{
		/* errors have already been logged */
		return false;
	}

	content.count = countLines(content.buffer);
	content.lines = (char **) calloc(content.count, sizeof(char *));
	content.count = splitLines(content.buffer, content.lines, content.count);

	if (content.lines == NULL)
	{
		log_error(ALLOCATION_FAILED_ERROR);
		return false;
	}

	log_info("Replaying changes from file \"%s\"", context->sqlFileName);

	log_debug("Read %d lines in file \"%s\"",
			  content.count,
			  content.filename);

	/* replay the SQL commands from the SQL file */
	for (int i = 0; i < content.count && !context->reachedEndPos; i++)
	{
		const char *sql = content.lines[i];

		LogicalMessageMetadata metadata = { 0 };

		if (!parseSQLAction(sql, &metadata))
		{
			/* errors have already been logged */
			free(content.buffer);
			free(content.lines);

			return false;
		}

		/*
		 * The SWITCH WAL command should always be the last line of the file.
		 */
		if (metadata.action == STREAM_ACTION_SWITCH &&
			i != (content.count - 1))
		{
			log_error("SWITCH command for LSN %X/%X found in \"%s\" line %d, "
					  "before last line %d",
					  LSN_FORMAT_ARGS(metadata.lsn),
					  content.filename,
					  i + 1,
					  content.count);

			free(content.buffer);
			free(content.lines);

			return false;
		}

		if (!stream_apply_sql(context, &metadata, sql))
		{
			/* errors have already been logged */
			free(content.buffer);
			free(content.lines);

			return false;
		}
	}

	/* free dynamic memory that's not needed anymore */
	free(content.buffer);
	free(content.lines);

	return true;
}


/*
 * stream_apply_sql connects to the target database system and applies the
 * given SQL command as prepared by the stream_transform_file or
 * stream_transform_stream function.
 */
bool
stream_apply_sql(StreamApplyContext *context,
				 LogicalMessageMetadata *metadata,
				 const char *sql)
{
	PGSQL *pgsql = &(context->pgsql);

	switch (metadata->action)
	{
		case STREAM_ACTION_SWITCH:
		{
			log_debug("SWITCH from %X/%X to %X/%X",
					  LSN_FORMAT_ARGS(context->previousLSN),
					  LSN_FORMAT_ARGS(metadata->lsn));

			context->previousLSN = metadata->lsn;

			break;
		}

		case STREAM_ACTION_BEGIN:
		{
			if (metadata->lsn == InvalidXLogRecPtr ||
				IS_EMPTY_STRING_BUFFER(metadata->timestamp))
			{
				log_fatal("Failed to parse BEGIN message: %s", sql);
				return false;
			}

			/* did we reach the starting LSN positions now? */
			if (!context->reachedStartPos)
			{
				/*
				 * compare previousLSN with COMMIT LSN to safely include
				 * complete transactions while skipping already applied changes.
				 *
				 * this is particularly useful at the beginnig where BEGIN LSN
				 * of some transactions could be less than `consistent_point`,
				 * but COMMIT LSN of those transactions is guaranteed to be
				 * greater.
				 *
				 * in case of interruption and this is the first transaction to
				 * be applied, previousLSN should be equal to the last
				 * transaction's COMMIT LSN or the LSN of non-transaction
				 * action. Therefore, this condition will still hold true.
				 */

				if (!readTxnCommitLSN(context, metadata))
				{
					/* errors have already been logged */
					return false;
				}

				context->reachedStartPos =
					context->previousLSN < metadata->txnCommitLSN;
			}

			log_trace("BEGIN %lld LSN %X/%X @%s, previous LSN %X/%X, COMMIT LSN %X/%X %s",
					  (long long) metadata->xid,
					  LSN_FORMAT_ARGS(metadata->lsn),
					  metadata->timestamp,
					  LSN_FORMAT_ARGS(context->previousLSN),
					  LSN_FORMAT_ARGS(metadata->txnCommitLSN),
					  context->reachedStartPos ? "" : "[skipping]");

			/*
			 * Check if we reached the endpos LSN already.
			 */
			if (context->endpos != InvalidXLogRecPtr &&
				context->endpos <= metadata->lsn)
			{
				context->reachedEndPos = true;
				break;
			}

			/* actually skip this one if we didn't reach start pos yet */
			if (!context->reachedStartPos)
			{
				return true;
			}

			/*
			 * We're all good to replay that transaction, let's BEGIN and
			 * register our origin tracking on the target database.
			 */
			if (!pgsql_begin(pgsql))
			{
				/* errors have already been logged */
				return false;
			}

			if (!pgsql_set_gucs(pgsql, applySettings))
			{
				log_error("Failed to set the apply GUC settings, "
						  "see above for details");
				return false;
			}

			context->transactionInProgress = true;

			break;
		}

		case STREAM_ACTION_COMMIT:
		{
			if (!context->reachedStartPos)
			{
				return true;
			}

			/*
			 * update replication progress with metadata->lsn, that is,
			 * transaction COMMIT LSN
			 */
			char lsn[PG_LSN_MAXLENGTH] = { 0 };

			sformat(lsn, sizeof(lsn), "%X/%X",
					LSN_FORMAT_ARGS(metadata->lsn));

			if (!pgsql_replication_origin_xact_setup(pgsql,
													 lsn,
													 metadata->timestamp))
			{
				/* errors have already been logged */
				return false;
			}

			log_trace("COMMIT %lld LSN %X/%X",
					  (long long) metadata->xid,
					  LSN_FORMAT_ARGS(metadata->lsn));


			/* calling pgsql_commit() would finish the connection, avoid */
			if (!pgsql_execute(pgsql, "COMMIT"))
			{
				/* errors have already been logged */
				return false;
			}

			context->transactionInProgress = false;
			context->previousLSN = metadata->lsn;

			/*
			 * At COMMIT time we might have reached the endpos: we know
			 * that already when endpos <= lsn. It's important to check
			 * that at COMMIT record time, because that record might be the
			 * last entry of the file we're applying.
			 */
			if (context->endpos != InvalidXLogRecPtr &&
				context->endpos <= context->previousLSN)
			{
				context->reachedEndPos = true;

				log_notice("Apply reached end position %X/%X at %X/%X",
						   LSN_FORMAT_ARGS(context->endpos),
						   LSN_FORMAT_ARGS(context->previousLSN));
				break;
			}

			break;
		}

		case STREAM_ACTION_ENDPOS:
		{
			if (!context->reachedStartPos)
			{
				return true;
			}

			log_debug("ENDPOS %X/%X found at %X/%X",
					  LSN_FORMAT_ARGS(metadata->lsn),
					  LSN_FORMAT_ARGS(context->previousLSN));

			context->previousLSN = metadata->lsn;

			/*
			 * It could be the current endpos, or the endpos of a previous
			 * run.
			 */
			if (context->endpos != InvalidXLogRecPtr &&
				context->endpos <= context->previousLSN)
			{
				context->reachedEndPos = true;

				log_notice("Apply reached end position %X/%X at %X/%X",
						   LSN_FORMAT_ARGS(context->endpos),
						   LSN_FORMAT_ARGS(context->previousLSN));

				if (context->transactionInProgress)
				{
					if (!pgsql_execute(pgsql, "ROLLBACK"))
					{
						/* errors have already been logged */
						return false;
					}

					context->transactionInProgress = false;
				}

				break;
			}

			break;
		}

		/*
		 * A KEEPALIVE message is replayed as its own transaction where the
		 * only thgin we do is call into the replication origin tracking
		 * API to advance our position on the target database.
		 */
		case STREAM_ACTION_KEEPALIVE:
		{
			/* did we reach the starting LSN positions now? */
			if (!context->reachedStartPos)
			{
				context->reachedStartPos =
					context->previousLSN < metadata->lsn;
			}

			/* in a transaction only the COMMIT LSN is tracked */
			if (context->transactionInProgress)
			{
				return true;
			}

			log_trace("KEEPALIVE LSN %X/%X @%s, previous LSN %X/%X %s",
					  LSN_FORMAT_ARGS(metadata->lsn),
					  metadata->timestamp,
					  LSN_FORMAT_ARGS(context->previousLSN),
					  context->reachedStartPos ? "" : "[skipping]");

			if (metadata->lsn == InvalidXLogRecPtr ||
				IS_EMPTY_STRING_BUFFER(metadata->timestamp))
			{
				log_fatal("Failed to parse KEEPALIVE message: %s", sql);
				return false;
			}

			/*
			 * Check if we reached the endpos LSN already. If the keepalive
			 * message is the endpos, still apply it: its only purpose is
			 * to maintain our replication origin tracking on the target
			 * database.
			 */
			if (context->endpos != InvalidXLogRecPtr &&
				context->endpos < metadata->lsn)
			{
				context->reachedEndPos = true;
				break;
			}

			/* actually skip this one if we didn't reach start pos yet */
			if (!context->reachedStartPos)
			{
				return true;
			}

			/* skip KEEPALIVE message that won't make progress */
			if (metadata->lsn == context->previousLSN)
			{
				return true;
			}

			if (!pgsql_begin(pgsql))
			{
				/* errors have already been logged */
				return false;
			}

			char lsn[PG_LSN_MAXLENGTH] = { 0 };

			sformat(lsn, sizeof(lsn), "%X/%X",
					LSN_FORMAT_ARGS(metadata->lsn));

			if (!pgsql_replication_origin_xact_setup(pgsql,
													 lsn,
													 metadata->timestamp))
			{
				/* errors have already been logged */
				return false;
			}

			/* calling pgsql_commit() would finish the connection, avoid */
			if (!pgsql_execute(pgsql, "COMMIT"))
			{
				/* errors have already been logged */
				return false;
			}

			context->previousLSN = metadata->lsn;

			/*
			 * At COMMIT time we might have reached the endpos: we know
			 * that already when endpos <= lsn. It's important to check
			 * that at COMMIT record time, because that record might be the
			 * last entry of the file we're applying.
			 */
			if (context->endpos != InvalidXLogRecPtr &&
				context->endpos <= context->previousLSN)
			{
				context->reachedEndPos = true;

				log_notice("Apply reached end position %X/%X at %X/%X",
						   LSN_FORMAT_ARGS(context->endpos),
						   LSN_FORMAT_ARGS(context->previousLSN));
				break;
			}

			break;
		}

		case STREAM_ACTION_INSERT:
		case STREAM_ACTION_UPDATE:
		case STREAM_ACTION_DELETE:
		{
			if (!context->reachedStartPos)
			{
				return true;
			}

			uint32_t hash = metadata->hash;
			PreparedStmt *stmtHashTable = context->preparedStmt;
			PreparedStmt *stmt = NULL;

			HASH_FIND(hh, stmtHashTable, &hash, sizeof(hash), stmt);

			if (stmt == NULL)
			{
				char name[NAMEDATALEN] = { 0 };
				sformat(name, sizeof(name), "%x", metadata->hash);

				if (!pgsql_prepare(pgsql, name, metadata->stmt, 0, NULL))
				{
					/* errors have already been logged */
					return false;
				}

				PreparedStmt *stmt =
					(PreparedStmt *) calloc(1, sizeof(PreparedStmt));

				stmt->hash = hash;
				stmt->prepared = true;

				HASH_ADD(hh, stmtHashTable, hash, sizeof(hash), stmt);

				/* HASH_ADD can change the pointer in place, update */
				context->preparedStmt = stmtHashTable;
			}

			break;
		}

		case STREAM_ACTION_EXECUTE:
		{
			if (!context->reachedStartPos)
			{
				return true;
			}

			uint32_t hash = metadata->hash;
			PreparedStmt *stmtHashTable = context->preparedStmt;
			PreparedStmt *stmt = NULL;

			HASH_FIND(hh, stmtHashTable, &hash, sizeof(hash), stmt);

			if (stmt == NULL)
			{
				log_warn("BUG: Failed to find statement %x in stmtHashTable",
						 hash);
			}

			char name[NAMEDATALEN] = { 0 };
			sformat(name, sizeof(name), "%x", metadata->hash);

			JSON_Value *js = json_parse_string(metadata->jsonBuffer);

			if (json_value_get_type(js) != JSONArray)
			{
				log_error("Failed to parse EXECUTE array: %s",
						  metadata->jsonBuffer);
				return false;
			}

			JSON_Array *jsArray = json_value_get_array(js);

			int count = json_array_get_count(jsArray);
			const char **paramValues =
				(const char **) calloc(count, sizeof(char *));

			if (paramValues == NULL)
			{
				log_error(ALLOCATION_FAILED_ERROR);
				return false;
			}

			for (int i = 0; i < count; i++)
			{
				const char *value = json_array_get_string(jsArray, i);
				paramValues[i] = value;
			}

			if (!pgsql_execute_prepared(pgsql, name,
										count, paramValues,
										NULL, NULL))
			{
				/* errors have already been logged */
				return false;
			}

			free(paramValues);
			free(metadata->jsonBuffer);
			json_value_free(js);

			break;
		}

		case STREAM_ACTION_TRUNCATE:
		{
			if (!context->reachedStartPos)
			{
				return true;
			}

			/* chomp the final semi-colon that we added */
			int len = strlen(sql);

			if (sql[len - 1] == ';')
			{
				char *ptr = (char *) sql + len - 1;
				*ptr = '\0';
			}

			if (!pgsql_execute(pgsql, sql))
			{
				/* errors have already been logged */
				return false;
			}
			break;
		}

		default:
		{
			log_error("Failed to parse action %c for SQL query: %s",
					  metadata->action,
					  sql);

			return false;
		}
	}

	return true;
}


/*
 * setupReplicationOrigin ensures that a replication origin has been created on
 * the target database, and if it has been created previously then fetches the
 * previous LSN position it was at.
 *
 * Also setupReplicationOrigin calls pg_replication_origin_setup() in the
 * current connection, that is set to be
 */
bool
setupReplicationOrigin(StreamApplyContext *context,
					   CDCPaths *paths,
					   ConnStrings *connStrings,
					   char *origin,
					   uint64_t endpos,
					   bool apply,
					   bool logSQL)
{
	PGSQL *pgsql = &(context->pgsql);
	char *nodeName = context->origin;

	/*
	 * We have to consider both the --endpos command line option and the
	 * pgcopydb sentinel endpos value. Typically the sentinel is updated after
	 * the fact, but we still give precedence to --endpos.
	 *
	 * The endpos parameter here comes from the --endpos command line option,
	 * the context->endpos might have been set by calling
	 * stream_apply_wait_for_sentinel() earlier (when in STREAM_MODE_PREFETCH).
	 */
	if (endpos != InvalidXLogRecPtr)
	{
		if (context->endpos != InvalidXLogRecPtr && context->endpos != endpos)
		{
			log_warn("Option --endpos %X/%X is used, "
					 "even when the pgcopydb sentinel endpos was set to %X/%X",
					 LSN_FORMAT_ARGS(endpos),
					 LSN_FORMAT_ARGS(context->endpos));
		}
		context->endpos = endpos;
	}

	context->paths = *paths;
	context->apply = apply;

	context->reachedStartPos = false;

	context->connStrings = connStrings;

	strlcpy(context->origin, origin, sizeof(context->origin));

	if (!pgsql_init(pgsql, context->connStrings->target_pguri, PGSQL_CONN_TARGET))
	{
		/* errors have already been logged */
		return false;
	}

	/* we're going to send several replication origin commands */
	pgsql->connectionStatementType = PGSQL_CONNECTION_MULTI_STATEMENT;

	/* we also might want to skip logging any SQL query that we apply */
	pgsql->logSQL = logSQL;

	uint32_t oid = 0;

	if (!pgsql_replication_origin_oid(pgsql, nodeName, &oid))
	{
		/* errors have already been logged */
		return false;
	}

	log_debug("setupReplicationOrigin: oid == %u", oid);

	if (oid == 0)
	{
		log_error("Failed to fetch progress for replication origin \"%s\": "
				  "replication origin not found on target database",
				  nodeName);
		(void) pgsql_finish(pgsql);
		return false;
	}

	if (!pgsql_replication_origin_progress(pgsql,
										   nodeName,
										   true,
										   &(context->previousLSN)))
	{
		/* errors have already been logged */
		return false;
	}

	if (IS_EMPTY_STRING_BUFFER(context->sqlFileName))
	{
		if (!computeSQLFileName(context))
		{
			/* errors have already been logged */
			return false;
		}
	}

	/* compute the WAL filename that would host the previous LSN */
	log_debug("setupReplicationOrigin: replication origin \"%s\" "
			  "found at %X/%X, expected in file \"%s\"",
			  nodeName,
			  LSN_FORMAT_ARGS(context->previousLSN),
			  context->sqlFileName);

	if (!pgsql_replication_origin_session_setup(pgsql, nodeName))
	{
		/* errors have already been logged */
		return false;
	}

	return true;
}


/*
 * computeSQLFileName updates the StreamApplyContext structure with the current
 * LSN applied to the target system, and computed
 */
bool
computeSQLFileName(StreamApplyContext *context)
{
	XLogSegNo segno;

	if (context->WalSegSz == 0)
	{
		log_error("Failed to compute the SQL filename for LSN %X/%X "
				  "without context->wal_segment_size",
				  LSN_FORMAT_ARGS(context->previousLSN));
		return false;
	}

	XLByteToSeg(context->previousLSN, segno, context->WalSegSz);
	XLogFileName(context->wal, context->system.timeline, segno, context->WalSegSz);

	sformat(context->sqlFileName, sizeof(context->sqlFileName),
			"%s/%s.sql",
			context->paths.dir,
			context->wal);

	log_debug("computeSQLFileName: %X/%X \"%s\"",
			  LSN_FORMAT_ARGS(context->previousLSN),
			  context->sqlFileName);

	return true;
}


/*
 * parseSQLAction returns the action that is implemented in the given SQL
 * query.
 */
bool
parseSQLAction(const char *query, LogicalMessageMetadata *metadata)
{
	metadata->action = STREAM_ACTION_UNKNOWN;

	if (strcmp(query, "") == 0)
	{
		return true;
	}

	char *message = NULL;
	char *begin = strstr(query, OUTPUT_BEGIN);
	char *commit = strstr(query, OUTPUT_COMMIT);
	char *switchwal = strstr(query, OUTPUT_SWITCHWAL);
	char *keepalive = strstr(query, OUTPUT_KEEPALIVE);
	char *endpos = strstr(query, OUTPUT_ENDPOS);

	/* do we have a BEGIN or a COMMIT message to parse metadata of? */
	if (query == begin)
	{
		metadata->action = STREAM_ACTION_BEGIN;
		message = begin + strlen(OUTPUT_BEGIN);
	}
	else if (query == commit)
	{
		metadata->action = STREAM_ACTION_COMMIT;
		message = commit + strlen(OUTPUT_COMMIT);
	}
	else if (query == switchwal)
	{
		metadata->action = STREAM_ACTION_SWITCH;
		message = switchwal + strlen(OUTPUT_SWITCHWAL);
	}
	else if (query == keepalive)
	{
		metadata->action = STREAM_ACTION_KEEPALIVE;
		message = keepalive + strlen(OUTPUT_KEEPALIVE);
	}
	else if (query == endpos)
	{
		metadata->action = STREAM_ACTION_ENDPOS;
		message = endpos + strlen(OUTPUT_ENDPOS);
	}

	if (message != NULL)
	{
		JSON_Value *json = json_parse_string(message);

		if (!parseMessageMetadata(metadata, message, json, true))
		{
			/* errors have already been logged */
			json_value_free(json);
			return false;
		}

		json_value_free(json);

		return true;
	}

	/*
	 * So the SQL Action is a DML (or a TRUNCATE).
	 */
	size_t tLen = sizeof(TRUNCATE) - 1;
	size_t pLen = sizeof(PREPARE) - 1;
	size_t eLen = sizeof(EXECUTE) - 1;

	if (strncmp(query, TRUNCATE, tLen) == 0)
	{
		metadata->action = STREAM_ACTION_TRUNCATE;
	}
	else if (strncmp(query, PREPARE, pLen) == 0)
	{
		char *spc = strchr(query + pLen, ' ');

		if (spc == NULL)
		{
			log_error("Failed to parse PREPARE statement: %s", query);
			return false;
		}

		/* make a copy of just the hexadecimal string */
		int len = spc - (query + pLen);
		char str[BUFSIZE] = { 0 };

		sformat(str, sizeof(str), "%.*s", len, query + pLen);

		uint32_t hash = 0;

		if (!hexStringToUInt32(str, &hash))
		{
			log_error("Failed to parse PREPARE statement name: %s", query);
			return false;
		}

		metadata->hash = hash;

		size_t iLen = sizeof(INSERT) - 1;
		size_t uLen = sizeof(UPDATE) - 1;
		size_t dLen = sizeof(DELETE) - 1;

		if (strncmp(spc + 1, INSERT, iLen) == 0)
		{
			/* skip ' AS ' and point to INSERT */
			metadata->stmt = spc + 1 + 3;
			metadata->action = STREAM_ACTION_INSERT;
		}
		else if (strncmp(spc + 1, UPDATE, uLen) == 0)
		{
			/* skip ' AS ' and point to UPDATE */
			metadata->stmt = spc + 1 + 3;
			metadata->action = STREAM_ACTION_UPDATE;
		}
		else if (strncmp(spc + 1, DELETE, dLen) == 0)
		{
			/* skip ' AS ' and point to DELETE */
			metadata->stmt = spc + 1 + 3;
			metadata->action = STREAM_ACTION_DELETE;
		}
	}
	else if (strncmp(query, EXECUTE, eLen) == 0)
	{
		metadata->action = STREAM_ACTION_EXECUTE;

		char *json = strchr(query + eLen, '[');

		if (json == NULL)
		{
			log_error("Failed to parse EXECUTE statement: %s", query);
			return false;
		}

		/* make a copy of just the hexadecimal string */
		int len = json - (query + eLen);
		char str[BUFSIZE] = { 0 };

		sformat(str, sizeof(str), "%.*s", len, query + pLen);

		uint32_t hash = 0;

		if (!hexStringToUInt32(str, &hash))
		{
			log_error("Failed to parse EXECUTE statement name: %s", query);
			return false;
		}

		metadata->hash = hash;

		/* chomp ; at the end of the query string */
		len = strlen(json) - 1;
		size_t bytes = len + 1;

		metadata->jsonBuffer = (char *) calloc(bytes, sizeof(char));

		if (metadata->jsonBuffer == NULL)
		{
			log_error(ALLOCATION_FAILED_ERROR);
			return false;
		}

		sformat(metadata->jsonBuffer, bytes, "%.*s", len, json);
	}

	if (metadata->action == STREAM_ACTION_UNKNOWN)
	{
		log_error("Failed to parse action from query: %s", query);
		return false;
	}

	return true;
}


/*
 * readTxnCommitLSN ensures metadata has transaction COMMIT LSN by fetching it
 * from metadata file if it is not present
 */
bool
readTxnCommitLSN(StreamApplyContext *context,
				 LogicalMessageMetadata *metadata)
{
	/* if txnCommitLSN is invalid, then fetch it from txn metadata file */
	if (metadata->txnCommitLSN != InvalidXLogRecPtr)
	{
		return true;
	}

	char txnfilename[MAXPGPATH] = { 0 };

	if (!computeTxnMetadataFilename(metadata->xid,
									context->paths.dir,
									txnfilename))
	{
		/* errors have already been logged */
		return false;
	}

	log_debug("stream_apply_sql: BEGIN message without a commit LSN, "
			  "fetching commit LSN from transaction metadata file \"%s\"",
			  txnfilename);

	LogicalMessageMetadata txnMetadata = { .xid = metadata->xid };

	if (!parseTxnMetadataFile(txnfilename, &txnMetadata))
	{
		/* errors have already been logged */
		return false;
	}

	metadata->txnCommitLSN = txnMetadata.txnCommitLSN;

	return true;
}


/*
 * parseTxnMetadataFile returns the transaction metadata content for the given
 * metadata filename.
 */
bool
parseTxnMetadataFile(const char *filename, LogicalMessageMetadata *metadata)
{
	/* store xid as it will be overwritten while parsing metadata */
	uint32_t xid = metadata->xid;

	if (xid == 0)
	{
		log_error("BUG: parseTxnMetadataFile is called with "
				  "transaction xid: %lld", (long long) xid);
		return false;
	}

	/*
	 * Read the transaction metadata file created by the transform process for
	 * transactions spanning multiple WAL files. The metadata json file is
	 * generated upon encountering the COMMIT statement, but it may take some
	 * time to become available for transformation. Therefore, we retry here.
	 */

	ConnectionRetryPolicy retryPolicy = { 0 };

	int maxT = 900;             /* 15 mins */
	int maxSleepTime = 3000;    /* 2s */
	int baseSleepTime = 100;    /* 100ms */

	(void) pgsql_set_retry_policy(&retryPolicy,
								  maxT,
								  -1, /* unbounded number of attempts */
								  maxSleepTime,
								  baseSleepTime);

	while (!pgsql_retry_policy_expired(&retryPolicy))
	{
		if (file_exists(filename))
		{
			break;
		}

		int sleepTimeMs =
			pgsql_compute_connection_retry_sleep_time(&retryPolicy);

		log_debug("parseTxnMetadataFile: waiting for transaction metadata "
				  "file %s to be created, retrying in %dms",
				  filename, sleepTimeMs);

		/* we have milliseconds, pg_usleep() wants microseconds */
		(void) pg_usleep(sleepTimeMs * 1000);
	}

	char *txnMetadataContent = NULL;
	long size = 0L;

	/* we don't want to retry anymore, error out if files still don't exist */
	if (!read_file(filename, &txnMetadataContent, &size))
	{
		/* errors have already been logged */
		return false;
	}

	JSON_Value *json = json_parse_string(txnMetadataContent);

	if (!parseMessageMetadata(metadata, txnMetadataContent, json, true))
	{
		/* errors have already been logged */
		json_value_free(json);
		return false;
	}

	json_value_free(json);

	if (metadata->txnCommitLSN == InvalidXLogRecPtr ||
		metadata->xid != xid ||
		IS_EMPTY_STRING_BUFFER(metadata->timestamp))
	{
		log_error("Failed to parse metadata for transaction metadata file "
				  "%s: %s", filename, txnMetadataContent);
		return false;
	}

	return true;
}
