/* --------------------------------
 * 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;
}

void
conn_notifies_process(connectionObject *self)
{
    PGnotify *pgn = NULL;
    PyObject *notify = NULL;
    PyObject *pid = NULL, *channel = NULL, *payload = NULL;
    PyObject *tmp = NULL;

    static PyObject *append;

    if (!append) {
        if (!(append = Text_FromUTF8("append"))) {
            goto error;
        }
    }

    while ((pgn = PQnotifies(self->pgconn)) != NULL) {

        Dprintf("conn_notifies_process: got NOTIFY from pid %d, msg = %s",
                (int) pgn->be_pid, pgn->relname);

        if (!(pid = PyInt_FromLong((long)pgn->be_pid))) { goto error; }
        if (!(channel = conn_text_from_chars(self, pgn->relname))) { goto error; }
        if (!(payload = conn_text_from_chars(self, pgn->extra))) { goto error; }

        if (!(notify = PyObject_CallFunctionObjArgs((PyObject *)&notifyType,
                pid, channel, payload, NULL))) {
            goto error;
        }

        Py_DECREF(pid); pid = NULL;
        Py_DECREF(channel); channel = NULL;
        Py_DECREF(payload); payload = NULL;

        if (!(tmp = PyObject_CallMethodObjArgs(
                self->notifies, append, notify, NULL))) {
            goto error;
        }
        Py_DECREF(tmp); tmp = NULL;

        Py_DECREF(notify); notify = NULL;
        PQfreemem(pgn); pgn = NULL;
    }
    return;  /* no error */

error:
    if (pgn) { PQfreemem(pgn); }
    Py_XDECREF(tmp);
    Py_XDECREF(notify);
    Py_XDECREF(pid);
    Py_XDECREF(channel);
    Py_XDECREF(payload);

    /* TODO: callers currently don't expect an error from us */
    PyErr_Clear();

}

/* Establishes two network connections to a Postgres server: one for SQL, and one
 * for replication. context->conninfo contains the connection string or URL to connect
 * to, and context->app_name is the client name (which appears, for example, in
 * pg_stat_activity). Returns 0 on success. */
int client_connect(client_context_t context) {
    if (!context->conninfo || context->conninfo[0] == '\0') {
        client_error(context, "conninfo must be set in client context");
        return EINVAL;
    }
    if (!context->app_name || context->app_name[0] == '\0') {
        client_error(context, "app_name must be set in client context");
        return EINVAL;
    }

    context->sql_conn = PQconnectdb(context->conninfo);
    if (PQstatus(context->sql_conn) != CONNECTION_OK) {
        client_error(context, "Connection to database failed: %s", PQerrorMessage(context->sql_conn));
        return EIO;
    }

    /* Parse the connection string into key-value pairs */
    char *error = NULL;
    PQconninfoOption *parsed_opts = PQconninfoParse(context->conninfo, &error);
    if (!parsed_opts) {
        client_error(context, "Replication connection info: %s", error);
        PQfreemem(error);
        return EIO;
    }

    /* Copy the key-value pairs into a new structure with added replication options */
    PQconninfoOption *option;
    int optcount = 2; /* replication, fallback_application_name */
    for (option = parsed_opts; option->keyword != NULL; option++) {
        if (option->val != NULL && option->val[0] != '\0') optcount++;
    }

    const char **keys = malloc((optcount + 1) * sizeof(char *));
    const char **values = malloc((optcount + 1) * sizeof(char *));
    int i = 0;

    for (option = parsed_opts; option->keyword != NULL; option++) {
        if (option->val != NULL && option->val[0] != '\0') {
            keys[i] = option->keyword;
            values[i] = option->val;
            i++;
        }
    }

    keys[i] = "replication";               values[i] = "database";        i++;
    keys[i] = "fallback_application_name"; values[i] = context->app_name; i++;
    keys[i] = NULL;                        values[i] = NULL;

    int err = 0;
    context->repl.conn = PQconnectdbParams(keys, values, true);
    if (PQstatus(context->repl.conn) != CONNECTION_OK) {
        client_error(context, "Replication connection failed: %s", PQerrorMessage(context->repl.conn));
        err = EIO;
    }

    free(keys);
    free(values);
    PQconninfoFree(parsed_opts);
    return err;
}

/*
 * call-seq:
 *    conn.subtransaction( name, *args) { |conn,sp| ... }
 *
 * Open and close a transaction savepoint.  The savepoints name +nam+ may
 * contain % directives that will be expanded by +args+.
 */
VALUE
pgconn_subtransaction( int argc, VALUE *argv, VALUE self)
{
    struct pgconn_data *c;
    int a;
    VALUE sp, par, cmd, ya;
    const char *q;
    char *p;
    int n;

    Data_Get_Struct( self, struct pgconn_data, c);
    a = rb_scan_args( argc, argv, "1*", &sp, &par);
    StringValue( sp);
    if (a > 1)
        sp = rb_str_format(RARRAY_LEN(par), RARRAY_PTR(par), sp);

    cmd = rb_str_buf_new2( "savepoint ");
    q = pgconn_destring( c, sp, &n);
    p = PQescapeIdentifier( c->conn, q, n);
    rb_str_buf_cat2( cmd, p);
    ya = rb_ary_new3( 2, self, rb_str_new2( p));
    PQfreemem( p);
    rb_str_buf_cat2( cmd, ";");

    pgresult_clear( pg_statement_exec( self, cmd, Qnil));
    return rb_ensure( yield_subtransaction, ya, release_subtransaction, ya);
}

VALUE typecast_detect(const char *data, size_t size, int type) {
    VALUE value;
    char *bytea;
    size_t bytea_len;
    switch (type) {
        case SWIFT_TYPE_INT:
            return rb_cstr2inum(data, 10);
        case SWIFT_TYPE_FLOAT:
            return rb_float_new(atof(data));
        case SWIFT_TYPE_NUMERIC:
            return rb_funcall(cBigDecimal, fnew, 1, rb_str_new(data, size));
        case SWIFT_TYPE_BOOLEAN:
            return (data && (data[0] =='t' || data[0] == '1')) ? Qtrue : Qfalse;
        case SWIFT_TYPE_BLOB:
            bytea = PQunescapeBytea(data, &bytea_len);
            value = rb_str_new(bytea, bytea_len);
            PQfreemem(bytea);
            return rb_funcall(cStringIO, fnew, 1, value);
        case SWIFT_TYPE_TIMESTAMP:
            return datetime_parse(cSwiftDateTime, data, size);
        case SWIFT_TYPE_DATE:
            return date_parse(cSwiftDateTime, data, size);
        default:
            return rb_enc_str_new(data, size, rb_utf8_encoding());
    }
}

void pgsql_update_lease(const uint8_t* mac, const struct in_addr* yip, const char* ifname, const uint32_t expiresAt, const enum t_lease_update_src reason)
{
	/* only write DHCP ACK packet changes back */
	if (reason != UPDATED_LEASE_FROM_DHCP)
		return;

	/* the pgsql commands are both run always, as the initial entry might have been created on another device. */
	/* though, we restrict ACKs to be received on APs that saw the request - no roaming between REQ/ACK */
	/* add to pgsql */
	if (!pgsql_connected())
		return;
	
	const uint32_t now = reltime();
	eprintf(DEBUG_VERBOSE, "sql: update lease: MAC: %s IP: %s VLAN: %s expiresIn: %d", ether_ntoa_z((struct ether_addr *)mac), inet_ntoa(*yip), ifname, expiresAt - now);

	char *sql_esc_bridge = PQescapeLiteral(pgsql, ifname, strlen(ifname));
	if (!sql_esc_bridge) return;

	char sql[2048];
	if (expiresAt > now) {
		snprintf(sql, sizeof(sql), "INSERT INTO " PGSQLLEASETABLE " (bridge, mac, ip, validUntil) VALUES(%s, '%s', '%s', CURRENT_TIMESTAMP + interval '%d seconds') ON CONFLICT (bridge, mac, ip) DO UPDATE SET validUntil = CURRENT_TIMESTAMP + interval '%d seconds';", sql_esc_bridge, ether_ntoa_z((struct ether_addr *)mac), inet_ntoa(*yip), expiresAt - now, expiresAt - now);
	} else {
		snprintf(sql, sizeof(sql), "UPDATE " PGSQLLEASETABLE " SET validUntil = CURRENT_TIMESTAMP WHERE bridge = %s AND mac = '%s';", sql_esc_bridge, ether_ntoa_z((struct ether_addr *)mac));
	}
	PQfreemem(sql_esc_bridge); sql_esc_bridge = NULL;
	eprintf(DEBUG_GENERAL, "write sql: %s", sql);
	pgsql_query_errprint(sql);
}

static VALUE typecast(const char *value, long length, const VALUE type, int encoding) {

  if (type == rb_cInteger) {
    return rb_cstr2inum(value, 10);
  } else if (type == rb_cString) {
    return DO_STR_NEW(value, length, encoding);
  } else if (type == rb_cFloat) {
    return rb_float_new(rb_cstr_to_dbl(value, Qfalse));
  } else if (type == rb_cBigDecimal) {
    return rb_funcall(rb_cBigDecimal, ID_NEW, 1, rb_str_new(value, length));
  } else if (type == rb_cDate) {
    return parse_date(value);
  } else if (type == rb_cDateTime) {
    return parse_date_time(value);
  } else if (type == rb_cTime) {
    return parse_time(value);
  } else if (type == rb_cTrueClass) {
    return *value == 't' ? Qtrue : Qfalse;
  } else if (type == rb_cByteArray) {
    size_t new_length = 0;
    char* unescaped = (char *)PQunescapeBytea((unsigned char*)value, &new_length);
    VALUE byte_array = rb_funcall(rb_cByteArray, ID_NEW, 1, rb_str_new(unescaped, new_length));
    PQfreemem(unescaped);
    return byte_array;
  } else if (type == rb_cClass) {
    return rb_funcall(rb_cObject, rb_intern("full_const_get"), 1, rb_str_new(value, length));
  } else if (type == rb_cObject) {
    return rb_marshal_load(rb_str_new(value, length));
  } else if (type == rb_cNilClass) {
    return Qnil;
  } else {
    return DO_STR_NEW(value, length, encoding);
  }

}

static int pgsql_handle_quoter(pdo_dbh_t *dbh, const char *unquoted, int unquotedlen, char **quoted, int *quotedlen, enum pdo_param_type paramtype)
{
	unsigned char *escaped;
	pdo_pgsql_db_handle *H = (pdo_pgsql_db_handle *)dbh->driver_data;
	size_t tmp_len;

	switch (paramtype) {
		case PDO_PARAM_LOB:
			/* escapedlen returned by PQescapeBytea() accounts for trailing 0 */
			escaped = PQescapeByteaConn(H->server, (unsigned char *)unquoted, (size_t)unquotedlen, &tmp_len);
			*quotedlen = (int)tmp_len + 1;
			*quoted = emalloc(*quotedlen + 1);
			memcpy((*quoted)+1, escaped, *quotedlen-2);
			(*quoted)[0] = '\'';
			(*quoted)[*quotedlen-1] = '\'';
			(*quoted)[*quotedlen] = '\0';
			PQfreemem(escaped);
			break;
		default:
			*quoted = safe_emalloc(2, unquotedlen, 3);
			(*quoted)[0] = '\'';
			*quotedlen = PQescapeStringConn(H->server, *quoted + 1, unquoted, (size_t)unquotedlen, NULL);
			(*quoted)[*quotedlen + 1] = '\'';
			(*quoted)[*quotedlen + 2] = '\0';
			*quotedlen += 2;
	}
	return 1;
}

static VALUE cConnection_quote_byte_array(VALUE self, VALUE string) {
  PGconn *db = DATA_PTR(rb_iv_get(self, "@connection"));

  const unsigned char *source = (unsigned char*) RSTRING_PTR(string);
  size_t source_len     = RSTRING_LEN(string);

  unsigned char *escaped;
  unsigned char *escaped_quotes;
  size_t quoted_length = 0;
  VALUE result;

  // Allocate space for the escaped version of 'string'
  // http://www.postgresql.org/docs/8.3/static/libpq-exec.html#LIBPQ-EXEC-ESCAPE-STRING
  escaped = PQescapeByteaConn(db, source, source_len, &quoted_length);
  escaped_quotes = (unsigned char *)calloc(quoted_length + 1, sizeof(unsigned char));
  memcpy(escaped_quotes + 1, escaped, quoted_length);

  // Wrap the escaped string in single-quotes, this is DO's convention (replace trailing \0)
  escaped_quotes[quoted_length] = escaped_quotes[0] = '\'';

  result = rb_str_new((const char *)escaped_quotes, quoted_length + 1);
  PQfreemem(escaped);
  free(escaped_quotes);
  return result;
}

static VALUE typecast(const char* data, uint64_t len, int pgtype) {
  size_t bytea_len;
  unsigned char* bytea;
  VALUE rv;

  switch(pgtype) {
    case 16:
      return *data == 't' ? Qtrue : Qfalse;
    case 17:
      bytea = PQunescapeBytea(data, &bytea_len);
      rv = rb_funcall(cStringIO, fnew, 1, rb_str_new(bytea, bytea_len));
      PQfreemem(bytea);
      return rv;
    case 20:
    case 21:
    case 22:
    case 23:
    case 26:
      return rb_cstr2inum(data, 10);
    case 700:
    case 701:
    case 790:
      return rb_float_new(atof(data));
    case 1700:
      return rb_funcall(cBigDecimal, fnew, 1, rb_str_new(data, len));
    case 1082:
      return typecast_date(data, len);
    case 1114:
    case 1184:
      return typecast_timestamp(data, len);
    default:
      return rb_str_new(data, len);
  }
}

static PyObject *
psyco_parse_dsn(PyObject *self, PyObject *args, PyObject *kwargs)
{
    char *err = NULL;
    PQconninfoOption *options = NULL;
    PyObject *res = NULL, *dsn;

    static char *kwlist[] = {"dsn", NULL};
    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O", kwlist, &dsn)) {
        return NULL;
    }

    Py_INCREF(dsn); /* for ensure_bytes */
    if (!(dsn = psycopg_ensure_bytes(dsn))) { goto exit; }

    options = PQconninfoParse(Bytes_AS_STRING(dsn), &err);
    if (options == NULL) {
        if (err != NULL) {
            PyErr_Format(ProgrammingError, "invalid dsn: %s", err);
            PQfreemem(err);
        } else {
            PyErr_SetString(OperationalError, "PQconninfoParse() failed");
        }
        goto exit;
    }

    res = psycopg_dict_from_conninfo_options(options, /* include_password = */ 1);

exit:
    PQconninfoFree(options);    /* safe on null */
    Py_XDECREF(dsn);

    return res;
}

/*
 * call-seq:
 *    conn.getline( async = nil)         -> str
 *    conn.getline( async = nil) { ... } -> str
 *
 * Reads a line from the backend server after a +COPY+ command.
 * Returns +nil+ for EOF.
 *
 * If async is +true+ and no data is available then the block will be called
 * and its value will be returned.
 *
 * Call this method inside a block passed to +copy_stdout+.  See
 * there for an example.
 */
VALUE
pgconn_getline( int argc, VALUE *argv, VALUE self)
{
    struct pgconn_data *c;
    VALUE as;
    int async;
    char *b;
    int r;

    async = rb_scan_args( argc, argv, "01", &as) > 0 && !NIL_P( as) ? 1 : 0;

    Data_Get_Struct( self, struct pgconn_data, c);
    r = PQgetCopyData( c->conn, &b, async);
    if (r > 0) {
        VALUE ret;

        ret = pgconn_mkstringn( c, b, r);
        PQfreemem( b);
        rb_lastline_set( ret);
        return ret;
    } else if (r == 0)
        return rb_yield( Qnil);
    else {
        /* PQgetResult() will be called in the ensure block. */
    }
    return Qnil;
}

/*
 * Receive a message available from XLOG stream, blocking for
 * maximum of 'timeout' ms.
 *
 * Returns:
 *
 *	 True if data was received. *type, *buffer and *len are set to
 *	 the type of the received data, buffer holding it, and length,
 *	 respectively.
 *
 *	 False if no data was available within timeout, or wait was interrupted
 *	 by signal.
 *
 * The buffer returned is only valid until the next call of this function or
 * libpq_connect/disconnect.
 *
 * ereports on error.
 */
static bool
libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len)
{
	int			rawlen;

	if (recvBuf != NULL)
		PQfreemem(recvBuf);
	recvBuf = NULL;

	/*
	 * If the caller requested to block, wait for data to arrive. But if this
	 * is the first call after connecting, don't wait, because there might
	 * already be some data in libpq buffer that we haven't returned to
	 * caller.
	 */
	if (timeout > 0 && !justconnected)
	{
		if (!libpq_select(timeout))
			return false;

		if (PQconsumeInput(streamConn) == 0)
			ereport(ERROR,
					(errmsg("could not receive data from WAL stream: %s",
							PQerrorMessage(streamConn))));
	}
	justconnected = false;

	/* Receive CopyData message */
	rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
	if (rawlen == 0)			/* no data available yet, then return */
		return false;
	if (rawlen == -1)			/* end-of-streaming or error */
	{
		PGresult   *res;

		res = PQgetResult(streamConn);
		if (PQresultStatus(res) == PGRES_COMMAND_OK)
		{
			PQclear(res);
			ereport(ERROR,
					(errmsg("replication terminated by primary server")));
		}
		PQclear(res);
		ereport(ERROR,
				(errmsg("could not receive data from WAL stream: %s",
						PQerrorMessage(streamConn))));
	}
	if (rawlen < -1)
		ereport(ERROR,
				(errmsg("could not receive data from WAL stream: %s",
						PQerrorMessage(streamConn))));

	/* Return received messages to caller */
	*type = *((unsigned char *) recvBuf);
	*buffer = recvBuf + sizeof(*type);
	*len = rawlen - sizeof(*type);

	return true;
}

static inline QByteArray byteArrayFromData(const char *data)
{
    size_t unescapedLen;
    unsigned char *unescapedData = PQunescapeBytea((const unsigned char*)data, &unescapedLen);
    const QByteArray result((const char*)unescapedData, unescapedLen);
    //! @todo avoid deep copy; QByteArray does not allow passing ownership of data; maybe copy PQunescapeBytea code?
    PQfreemem(unescapedData);
    return result;
}

void pp_export(PGconn * conn, const char * query)
{

	const int ascFmt = 0;
	int len = 0, i;
	int rows, ncol, icol;
	char * val;
	char * file;
	PGresult * res;
	const char * const * ival = (const char * const *) &val;
	FILE * fimg;

	/* execute the user provided query */
	res = PQexecParams(conn, query,
			0, //n. of params
			NULL, //oids guessed by backend
			NULL,
			NULL,
			NULL,
			ascFmt);
	if (PQresultStatus(res) != PGRES_TUPLES_OK) {
		pp_print_error(PQerrorMessage(conn));
		PQclear(res);
		return;
	}

	/* some check */
	icol = PQfnumber(res, opts.imagecol);
	if(icol==-1) {
		fprintf(stderr, "ERROR: Image column '%s' does not exist in result set.\n", opts.imagecol);
		PQclear(res);
		return;
	}
	if (!opts.nameprefix) {
		ncol = PQfnumber(res, opts.namecol);
		if(ncol==-1) {
			fprintf(stderr, "ERROR: Name column '%s' does not exist in result set.\n", opts.namecol);
			PQclear(res);
			return;
		}
	}
	rows = PQntuples(res);
	
	/* fetch the data and save */
	for (i = 0; i < rows; ++i) {
		val = PQunescapeBytea( PQgetvalue(res, i, icol), &len);
		if(opts.namecol) file = PQgetvalue(res, i, ncol);
		else file = image_name(opts.nameprefix, i);
		fimg = fopen(file, "w");
		fwrite(val, len, 1, fimg);
		fclose(fimg);
		PQfreemem(val);
		fprintf(stderr, "INFO: exported file %s\n", file);
	}
	PQclear(res);
}

const char* PgSQLResult::getDataStream(const std::string& s, uint64_t& size)
{
	std::string buf = PQgetvalue(m_handle, m_cursor, PQfnumber(m_handle, s.c_str()));
	uint8_t* temp = PQunescapeBytea( (const uint8_t*)buf.c_str(), (size_t*)&size);

	char* value = new char[buf.size()];
	strcpy(value, (char*)temp);

	PQfreemem(temp);
	return value;
}

// Encrypt a password using the appropriate encoding conversion
wxString pgConn::EncryptPassword(const wxString &user, const wxString &password)
{
	char *chrPassword;
	wxString strPassword;

	chrPassword = PQencryptPassword(password.mb_str(*conv), user.mb_str(*conv));
	strPassword = wxString::FromAscii(chrPassword);

	PQfreemem(chrPassword);

	return strPassword;
}

/*!
 * \brief Convert a str to a db value, copy strings
 *
 * Convert a str to a db value, copy strings.
 * The postgresql module uses a custom escape function for BLOBs.
 * If the _s is linked in the db_val result, it will be returned zero
 * \param _t destination value type
 * \param _v destination value
 * \param _s source string
 * \param _l string length
 * \return 0 on success, negative on error
 */
int db_postgres_str2val(const db_type_t _t, db_val_t* _v, const char* _s,
		const int _l)
{
	/* use common function for non BLOB, NULL setting and input
	 * parameter checking */
	if ( _t != DB1_BLOB || _s == NULL || _v == NULL) {
		return db_str2val(_t, _v, _s, _l, 1);
	} else {
		char * tmp_s = NULL;
		LM_DBG("converting BLOB [%.*s]\n", _l, _s);
		/*
		 * The string is stored in new allocated memory, which we could
		 * not free later thus we need to copy it to some new memory here.
		 */
		tmp_s = (char*)PQunescapeBytea((unsigned char*)_s,
					(size_t*)(void*)&(VAL_BLOB(_v).len));
		if(tmp_s==NULL) {
			LM_ERR("PQunescapeBytea failed\n");
			return -7;
		}
		VAL_BLOB(_v).s = pkg_malloc(VAL_BLOB(_v).len + 1);
		if (VAL_BLOB(_v).s == NULL) {
			LM_ERR("no private memory left\n");
			PQfreemem(tmp_s);
			return -8;
		}
		LM_DBG("allocate %d+1 bytes memory for BLOB at %p",
				VAL_BLOB(_v).len, VAL_BLOB(_v).s);
		memcpy(VAL_BLOB(_v).s, tmp_s, VAL_BLOB(_v).len);
		PQfreemem(tmp_s);

		VAL_BLOB(_v).s[VAL_BLOB(_v).len] = '\0';
		VAL_TYPE(_v) = DB1_BLOB;
		VAL_FREE(_v) = 1;

		LM_DBG("got blob len %d\n", _l);
		return 0;

	}
}

/*
 * handleCopyOut
 * receives data as a result of a COPY ... TO STDOUT command
 *
 * conn should be a database connection that you just issued COPY TO on
 * and got back a PGRES_COPY_OUT result.
 * copystream is the file stream for the data to go to.
 *
 * result is true if successful, false if not.
 */
bool
handleCopyOut(PGconn *conn, FILE *copystream)
{
	bool		OK = true;
	char	   *buf;
	int			ret;
	PGresult   *res;

	for (;;)
	{
		ret = PQgetCopyData(conn, &buf, 0);

		if (ret < 0)
			break;				/* done or error */

		if (buf)
		{
			if (fwrite(buf, 1, ret, copystream) != ret)
			{
				if (OK)			/* complain only once, keep reading data */
					psql_error("could not write COPY data: %s\n",
							   strerror(errno));
				OK = false;
			}
			PQfreemem(buf);
		}
	}

	if (OK && fflush(copystream))
	{
		psql_error("could not write COPY data: %s\n",
				   strerror(errno));
		OK = false;
	}

	if (ret == -2)
	{
		psql_error("COPY data transfer failed: %s", PQerrorMessage(conn));
		OK = false;
	}

	/* Check command status and return to normal libpq state */
	res = PQgetResult(conn);
	if (PQresultStatus(res) != PGRES_COMMAND_OK)
	{
		psql_error("%s", PQerrorMessage(conn));
		OK = false;
	}
	PQclear(res);

	return OK;
}

bool queryCallback(void *p_context, int p_placeholder, DBBuffer &p_output)
{
    QueryMetadata *t_query_metadata;
    t_query_metadata = (QueryMetadata *)p_context;

    DBString t_parameter_value;
    t_parameter_value = t_query_metadata -> arguments[p_placeholder - 1];

    void *t_escaped_string;
    t_escaped_string = NULL;

    size_t t_escaped_string_length;
    t_escaped_string_length = 0;

    if (t_parameter_value . isbinary)
    {
        t_escaped_string = PQescapeBytea((const unsigned char *)t_parameter_value . sptr, t_parameter_value . length, &t_escaped_string_length);
        if (t_escaped_string == NULL)
            return false;

        // PQescapeBytea appends an extra null char to the end of the escaped string, disregard this.
        t_escaped_string_length--;
    }
    else
    {
        if (t_parameter_value . length != 0)
        {
            t_escaped_string = malloc((t_parameter_value . length * 2) + 1);
            t_escaped_string_length = PQescapeString((char *)t_escaped_string, t_parameter_value . sptr, t_parameter_value . length);
        }
    }

    p_output . ensure(t_escaped_string_length + 2);
    memcpy(p_output . getFrontier(), "'", 1);
    p_output . advance(1);

    if (t_escaped_string != NULL)
    {
        memcpy(p_output . getFrontier(), t_escaped_string, t_escaped_string_length);
        p_output . advance(t_escaped_string_length);
    }

    memcpy(p_output . getFrontier(), "'", 1);
    p_output . advance(1);

    if (t_parameter_value . isbinary)
        PQfreemem(t_escaped_string);
    else
        free(t_escaped_string);

    return true;
}