/******************************************************
Copyright (c) 2011-2017 Percona LLC and/or its affiliates.

The xbstream format writer implementation.

This program 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; version 2 of the License.

This program 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 this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA

*******************************************************/

#include <my_global.h>
#include <my_base.h>
#include <zlib.h>
#include <stdint.h>
#include "common.h"
#include "xbstream.h"

/* Group writes smaller than this into a single chunk */
#define XB_STREAM_MIN_CHUNK_SIZE (10 * 1024 * 1024)

struct xb_wstream_struct {
	pthread_mutex_t	mutex;
	xb_stream_write_callback *write;
	void *user_data;
};

struct xb_wstream_file_struct {
	xb_wstream_t	*stream;
	char		*path;
	size_t		path_len;
	char		chunk[XB_STREAM_MIN_CHUNK_SIZE];
	char		*chunk_ptr;
	size_t		chunk_free;
	my_off_t	offset;
  bool rewrite;
};

static int xb_stream_flush(xb_wstream_file_t *file);
static int xb_stream_write_chunk(xb_wstream_file_t *file,
				 const void *buf, size_t len);
static int xb_stream_write_eof(xb_wstream_file_t *file);

static
ssize_t
xb_stream_default_write_callback(
				 void *userdata __attribute__((unused)),
				 const void *buf, size_t len)
{
	if (my_write(my_fileno(stdout), (const uchar *)buf, len, MYF(MY_WME | MY_NABP)))
		return -1;
	return len;
}

xb_wstream_t *
xb_stream_write_new(
	xb_stream_write_callback *write_callback, void *user_data)
{
	xb_wstream_t	*stream;

	stream = (xb_wstream_t *) my_malloc(PSI_NOT_INSTRUMENTED, sizeof(xb_wstream_t), MYF(MY_FAE));
	pthread_mutex_init(&stream->mutex, NULL);
	if (write_callback) {
#ifdef _WIN32
		setmode(fileno(stdout), _O_BINARY);
#endif
		stream->write = write_callback;
		stream->user_data = user_data;
	}
	else {
		stream->write = xb_stream_default_write_callback;
		stream->user_data = user_data;
	}

	return stream;;
}

xb_wstream_file_t *
xb_stream_write_open(xb_wstream_t *stream, const char *path,
		const MY_STAT *mystat __attribute__((unused)), bool rewrite)
{
	xb_wstream_file_t	*file;
	size_t			path_len;

	path_len = strlen(path);

	if (path_len > FN_REFLEN) {
		msg("xb_stream_write_open(): file path is too long.");
		return NULL;
	}

	file = (xb_wstream_file_t *) my_malloc(PSI_NOT_INSTRUMENTED, sizeof(xb_wstream_file_t) +
					       path_len + 1, MYF(MY_FAE));

	file->path = (char *) (file + 1);
#ifdef _WIN32
	/* Normalize path on Windows, so we can restore elsewhere.*/
	{
		int i;
		for (i = 0; ; i++) {
			file->path[i] = (path[i] == '\\') ? '/' : path[i];
			if (!path[i])
				break;
		}
	}
#else
	memcpy(file->path, path, path_len + 1);
#endif
	file->path_len = path_len;

	file->stream = stream;
	file->offset = 0;
	file->chunk_ptr = file->chunk;
	file->chunk_free = XB_STREAM_MIN_CHUNK_SIZE;
	file->rewrite = rewrite;

	return file;
}

int
xb_stream_write_data(xb_wstream_file_t *file, const void *buf, size_t len)
{
	if (len < file->chunk_free) {
		memcpy(file->chunk_ptr, buf, len);
		file->chunk_ptr += len;
		file->chunk_free -= len;

		return 0;
	}

	if (xb_stream_flush(file))
		return 1;

	return xb_stream_write_chunk(file, buf, len);
}

int
xb_stream_write_close(xb_wstream_file_t *file)
{
	if (xb_stream_flush(file) ||
	    xb_stream_write_eof(file)) {
		my_free(file);
		return 1;
	}

	my_free(file);

	return 0;
}

int
xb_stream_write_done(xb_wstream_t *stream)
{
	pthread_mutex_destroy(&stream->mutex);

	my_free(stream);

	return 0;
}

static
int
xb_stream_flush(xb_wstream_file_t *file)
{
	if (file->chunk_ptr == file->chunk) {
		return 0;
	}

	if (xb_stream_write_chunk(file, file->chunk,
				  file->chunk_ptr - file->chunk)) {
		return 1;
	}

	file->chunk_ptr = file->chunk;
	file->chunk_free = XB_STREAM_MIN_CHUNK_SIZE;

	return 0;
}

static
int
xb_stream_write_chunk(xb_wstream_file_t *file, const void *buf, size_t len)
{
	/* Chunk magic + flags + chunk type + path_len + path + len + offset +
	checksum */
	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 +
			       FN_REFLEN + 8 + 8 + 4];
	uchar		*ptr;
	xb_wstream_t	*stream = file->stream;
	ulong		checksum;

	/* Write xbstream header */
	ptr = tmpbuf;

	/* Chunk magic */
	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;

	*ptr++ =
    file->rewrite ? XB_STREAM_FLAG_REWRITE : 0; /* Chunk flags */

	*ptr++ = (uchar) XB_CHUNK_TYPE_PAYLOAD;  /* Chunk type */

	int4store(ptr, file->path_len);          /* Path length */
	ptr += 4;

	memcpy(ptr, file->path, file->path_len); /* Path */
	ptr += file->path_len;

	int8store(ptr, len);                     /* Payload length */
	ptr += 8;

	checksum = my_checksum(0, buf, len);

	pthread_mutex_lock(&stream->mutex);

	int8store(ptr, file->offset);            /* Payload offset */
	ptr += 8;

	int4store(ptr, checksum);
	ptr += 4;

	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));

	if (stream->write(stream->user_data, tmpbuf, ptr-tmpbuf) == -1)
		goto err;


	if (stream->write(stream->user_data, buf, len) == -1) /* Payload */
		goto err;

	file->offset+= len;

	pthread_mutex_unlock(&stream->mutex);

	return 0;

err:

	pthread_mutex_unlock(&stream->mutex);

	return 1;
}

int xb_stream_write_seek_set(xb_wstream_file_t *file, my_off_t offset)
{
	/* Chunk magic + flags + chunk type + path_len + path + offset */
	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 +
			       FN_REFLEN + 8];
	int error = 0;
	xb_wstream_t	*stream = file->stream;
	uchar		*ptr = tmpbuf;
	/* Chunk magic */
	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
	*ptr++ = 0;                              /* Chunk flags */
	*ptr++ = (uchar) XB_CHUNK_TYPE_SEEK;   /* Chunk type */
	int4store(ptr, file->path_len);          /* Path length */
	ptr += 4;
	memcpy(ptr, file->path, file->path_len); /* Path */
	ptr += file->path_len;
	int8store(ptr, static_cast<int64_t>(offset));  /* Offset */
	ptr += 8;
	if (xb_stream_flush(file))
		return 1;
	pthread_mutex_lock(&stream->mutex);
	if (stream->write(stream->user_data, tmpbuf, ptr-tmpbuf) == -1)
		error = 1;
	if (!error)
		file->offset = offset;
	pthread_mutex_unlock(&stream->mutex);
	if (xb_stream_flush(file))
		return 1;
	return error;
}

static
int
xb_stream_write_eof(xb_wstream_file_t *file)
{
	/* Chunk magic + flags + chunk type + path_len + path */
	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 +
			       FN_REFLEN];
	uchar		*ptr;
	xb_wstream_t	*stream = file->stream;

	pthread_mutex_lock(&stream->mutex);

	/* Write xbstream header */
	ptr = tmpbuf;

	/* Chunk magic */
	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;

	*ptr++ = 0;                              /* Chunk flags */

	*ptr++ = (uchar) XB_CHUNK_TYPE_EOF;      /* Chunk type */

	int4store(ptr, file->path_len);          /* Path length */
	ptr += 4;

	memcpy(ptr, file->path, file->path_len); /* Path */
	ptr += file->path_len;

	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));

	if (stream->write(stream->user_data, tmpbuf,
			(ulonglong) (ptr - tmpbuf)) == -1)
		goto err;

	pthread_mutex_unlock(&stream->mutex);

	return 0;
err:

	pthread_mutex_unlock(&stream->mutex);

	return 1;
}


int
xb_stream_write_remove(xb_wstream_t *stream, const char *path) {
	/* Chunk magic + flags + chunk type + path_len + path */
	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 + FN_REFLEN];
	uchar		*ptr = tmpbuf;
	/* Chunk magic */
	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;

	*ptr++ = 0;                              /* Chunk flags */

	*ptr++ = (uchar) XB_CHUNK_TYPE_REMOVE;   /* Chunk type */
	size_t path_len = strlen(path);
	int4store(ptr, path_len);               /* Path length */
	ptr += 4;

	memcpy(ptr, path, path_len);            /* Path */
	ptr += path_len;

	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));

	pthread_mutex_lock(&stream->mutex);

	ssize_t result = stream->write(stream->user_data, tmpbuf,
		(ulonglong) (ptr - tmpbuf));

	pthread_mutex_unlock(&stream->mutex);

	return result < 0;

}

int
xb_stream_write_rename(
	xb_wstream_t *stream, const char *old_path, const char *new_path) {
	/* Chunk magic + flags + chunk type + path_len + path + path_len + path*/
	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 +
		4 + FN_REFLEN + 4 + FN_REFLEN];
	uchar		*ptr = tmpbuf;
	/* Chunk magic */
	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;

	*ptr++ = 0;                              /* Chunk flags */

	*ptr++ = (uchar) XB_CHUNK_TYPE_RENAME;   /* Chunk type */
	size_t path_len = strlen(old_path);
	int4store(ptr, path_len);               /* Path length */
	ptr += 4;

	memcpy(ptr, old_path, path_len);            /* Path */
	ptr += path_len;

	path_len = strlen(new_path);
	int4store(ptr, path_len);               /* Path length */
	ptr += 4;

	memcpy(ptr, new_path, path_len);            /* Path */
	ptr += path_len;

	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));

	pthread_mutex_lock(&stream->mutex);

	ssize_t result = stream->write(stream->user_data, tmpbuf,
		(ulonglong) (ptr - tmpbuf));

	pthread_mutex_unlock(&stream->mutex);

	return result < 0;
}