<commit_before>/*
 * Copyright (C) 2015 deipi.com LLC and contributors. All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */

#include "database.h"

#include "database_autocommit.h"
#include "generate_terms.h"
#include "log.h"
#include "msgpack_patcher.h"
#include "multivaluerange.h"

#include <bitset>
#include <fcntl.h>
#include <limits>
#include <sysexits.h>

#define XAPIAN_LOCAL_DB_FALLBACK 1

#define DATABASE_UPDATE_TIME 10

#define DEFAULT_OFFSET "0" /* Replace for the real offset */

#define WAL_MAX_SLOT 1000 /* Max commit number for file */

#define PATH_WAL "/.wal/"

#define FILE_WAL "wal."

#define MAGIC 0xC0DE

#define SIZE_UUID 36
#define SIZE_WAL_HEADER sizeof(int) + SIZE_UUID + sizeof(uint64_t)


static const std::regex find_field_re("(([_a-z][_a-z0-9]*):)?(\"[^\"]+\"|[^\": ]+)[ ]*", std::regex::icase | std::regex::optimize);


static auto getPos = [](size_t pos, size_t size) noexcept {
	return pos < size ? pos : size - 1;
};


constexpr const char* const DatabaseWAL::names[];


DatabaseWAL::DatabaseWAL(Database* _database)
	: current_file_rev(0),
	  fd_revision(-1),
	  database(_database) { }


DatabaseWAL::~DatabaseWAL()
{
	if (fd_revision != -1) {
		close(fd_revision);
	}
}


bool
DatabaseWAL::execute(const std::string& line)
{
	const char *p = line.data();
	const char *p_end = p + line.size();

	if (!(database->flags & DB_WRITABLE)) {
		throw MSG_Error("Database is read-only");
	}

	if (!database->local) {
		throw MSG_Error("Can not execute WAL on a remote database!");
	}

	size_t size = decode_length(&p, p_end, true);
	std::string revision(p, size);
	p += size;

	if (revision != database->get_revision_info()) {
		return false;
	}

	Type type = static_cast<Type>(decode_length(&p, p_end));

	size = decode_length(&p, p_end, true);
	std::string data(p, size);

	Xapian::docid did;
	Xapian::termcount freq;

	p = data.data();
	p_end = p + data.size();

	Xapian::WritableDatabase *wdb = static_cast<Xapian::WritableDatabase *>(database->db.get());
	switch (type) {
		case Type::ADD_DOCUMENT:
			wdb->add_document(Xapian::Document::unserialise(data));
			break;
		case Type::CANCEL:
			wdb->begin_transaction(false);
			wdb->cancel_transaction();
			break;
		case Type::DELETE_DOCUMENT_TERM:
			wdb->delete_document(data);
			break;
		case Type::COMMIT:
			wdb->commit();
			break;
		case Type::REPLACE_DOCUMENT:
			did = static_cast<Xapian::docid>(decode_length(&p, p_end));
			wdb->replace_document(did, Xapian::Document::unserialise(std::string(p, p_end - p)));
			break;
		case Type::REPLACE_DOCUMENT_TERM:
			size = decode_length(&p, p_end, true);
			wdb->replace_document(std::string(p, size), Xapian::Document::unserialise(std::string(p + size, p_end - p - size)));
			break;
		case Type::DELETE_DOCUMENT:
			did = static_cast<Xapian::docid>(decode_length(&p, p_end));
			wdb->delete_document(did);
			break;
		case Type::SET_METADATA:
			size = decode_length(&p, p_end, true);
			wdb->set_metadata(std::string(p, size), std::string(p + size, p_end - p - size));
			break;
		case Type::ADD_SPELLING:
			freq = static_cast<Xapian::termcount>(decode_length(&p, p_end));
			wdb->add_spelling(std::string(p, p_end - p), freq);
			break;
		case Type::REMOVE_SPELLING:
			freq = static_cast<Xapian::termcount>(decode_length(&p, p_end));
			wdb->remove_spelling(std::string(p, p_end - p), freq);
			break;
		default:
			throw MSG_Error("Invalid WAL message!");
	}

	return true;
}


void
DatabaseWAL::write(Type type, const std::string& data)
{
	if (!(database->flags & DB_WRITABLE)) {
		throw MSG_Error("Database is read-only");
	}

	if (!database->local) {
		throw MSG_Error("Can not execute WAL on a remote database!");
	}

	if (database->flags & DB_NOWAL) {
		throw MSG_Error("Can not execute WAL on a database with DB_NOWAL flag");
	}

	auto endpoint = database->endpoints.cbegin();
	std::string revision = database->get_revision_info();
	std::string uuid = database->get_uuid();
	std::string line = encode_length(revision.size()) + revision + encode_length(toUType(type)) + encode_length(data.size()) + data;

	L_DATABASE_WAL(this, "%s on %s: '%s'", names[toUType(type)], endpoint->path.c_str(), repr(line).c_str());

	uint64_t rev = 0;
	memcpy(&rev, revision.data(), revision.size());
	++rev; //Revision of the operation

	if (current_file_rev + WAL_MAX_SLOT <= rev) {
		close(fd_revision);
		open(revision, endpoint->path);
	}

	off_t last_write_off = lseek(fd_revision, 0, SEEK_CUR);
	::write(fd_revision, line.data(), line.size());
	uint64_t slot = (rev - current_file_rev) + 1;
	++slot; //Starting to write the next slot with the size of the written

	off_t off_slot = SIZE_WAL_HEADER + (sizeof(off_t) * slot);<commit_msg>Unnecesary increment<commit_after>/*
 * Copyright (C) 2015 deipi.com LLC and contributors. All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */

#include "database.h"

#include "database_autocommit.h"
#include "generate_terms.h"
#include "log.h"
#include "msgpack_patcher.h"
#include "multivaluerange.h"

#include <bitset>
#include <fcntl.h>
#include <limits>
#include <sysexits.h>

#define XAPIAN_LOCAL_DB_FALLBACK 1

#define DATABASE_UPDATE_TIME 10

#define DEFAULT_OFFSET "0" /* Replace for the real offset */

#define WAL_MAX_SLOT 1000 /* Max commit number for file */

#define PATH_WAL "/.wal/"

#define FILE_WAL "wal."

#define MAGIC 0xC0DE

#define SIZE_UUID 36
#define SIZE_WAL_HEADER sizeof(int) + SIZE_UUID + sizeof(uint64_t)


static const std::regex find_field_re("(([_a-z][_a-z0-9]*):)?(\"[^\"]+\"|[^\": ]+)[ ]*", std::regex::icase | std::regex::optimize);


static auto getPos = [](size_t pos, size_t size) noexcept {
	return pos < size ? pos : size - 1;
};


constexpr const char* const DatabaseWAL::names[];


DatabaseWAL::DatabaseWAL(Database* _database)
	: current_file_rev(0),
	  fd_revision(-1),
	  database(_database) { }


DatabaseWAL::~DatabaseWAL()
{
	if (fd_revision != -1) {
		close(fd_revision);
	}
}


bool
DatabaseWAL::execute(const std::string& line)
{
	const char *p = line.data();
	const char *p_end = p + line.size();

	if (!(database->flags & DB_WRITABLE)) {
		throw MSG_Error("Database is read-only");
	}

	if (!database->local) {
		throw MSG_Error("Can not execute WAL on a remote database!");
	}

	size_t size = decode_length(&p, p_end, true);
	std::string revision(p, size);
	p += size;

	if (revision != database->get_revision_info()) {
		return false;
	}

	Type type = static_cast<Type>(decode_length(&p, p_end));

	size = decode_length(&p, p_end, true);
	std::string data(p, size);

	Xapian::docid did;
	Xapian::termcount freq;

	p = data.data();
	p_end = p + data.size();

	Xapian::WritableDatabase *wdb = static_cast<Xapian::WritableDatabase *>(database->db.get());
	switch (type) {
		case Type::ADD_DOCUMENT:
			wdb->add_document(Xapian::Document::unserialise(data));
			break;
		case Type::CANCEL:
			wdb->begin_transaction(false);
			wdb->cancel_transaction();
			break;
		case Type::DELETE_DOCUMENT_TERM:
			wdb->delete_document(data);
			break;
		case Type::COMMIT:
			wdb->commit();
			break;
		case Type::REPLACE_DOCUMENT:
			did = static_cast<Xapian::docid>(decode_length(&p, p_end));
			wdb->replace_document(did, Xapian::Document::unserialise(std::string(p, p_end - p)));
			break;
		case Type::REPLACE_DOCUMENT_TERM:
			size = decode_length(&p, p_end, true);
			wdb->replace_document(std::string(p, size), Xapian::Document::unserialise(std::string(p + size, p_end - p - size)));
			break;
		case Type::DELETE_DOCUMENT:
			did = static_cast<Xapian::docid>(decode_length(&p, p_end));
			wdb->delete_document(did);
			break;
		case Type::SET_METADATA:
			size = decode_length(&p, p_end, true);
			wdb->set_metadata(std::string(p, size), std::string(p + size, p_end - p - size));
			break;
		case Type::ADD_SPELLING:
			freq = static_cast<Xapian::termcount>(decode_length(&p, p_end));
			wdb->add_spelling(std::string(p, p_end - p), freq);
			break;
		case Type::REMOVE_SPELLING:
			freq = static_cast<Xapian::termcount>(decode_length(&p, p_end));
			wdb->remove_spelling(std::string(p, p_end - p), freq);
			break;
		default:
			throw MSG_Error("Invalid WAL message!");
	}

	return true;
}


void
DatabaseWAL::write(Type type, const std::string& data)
{
	if (!(database->flags & DB_WRITABLE)) {
		throw MSG_Error("Database is read-only");
	}

	if (!database->local) {
		throw MSG_Error("Can not execute WAL on a remote database!");
	}

	if (database->flags & DB_NOWAL) {
		throw MSG_Error("Can not execute WAL on a database with DB_NOWAL flag");
	}

	auto endpoint = database->endpoints.cbegin();
	std::string revision = database->get_revision_info();
	std::string uuid = database->get_uuid();
	std::string line = encode_length(revision.size()) + revision + encode_length(toUType(type)) + encode_length(data.size()) + data;

	L_DATABASE_WAL(this, "%s on %s: '%s'", names[toUType(type)], endpoint->path.c_str(), repr(line).c_str());

	uint64_t rev = 0;
	memcpy(&rev, revision.data(), revision.size());
	++rev; //Revision of the operation

	if (current_file_rev + WAL_MAX_SLOT <= rev) {
		close(fd_revision);
		open(revision, endpoint->path);
	}

	off_t last_write_off = lseek(fd_revision, 0, SEEK_CUR);
	::write(fd_revision, line.data(), line.size());
	uint64_t slot = (rev - current_file_rev) + 1;

	off_t off_slot = SIZE_WAL_HEADER + (sizeof(off_t) * slot);
{'fraction_non_alphanumeric': 0.1421288564647974, 'fraction_numerical': 0.005449591280653951, 'mean_word_length': 4.87706611570248, 'pattern_counts': {'":': 2, '<': 27, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '58759', 'n_tokens_mistral': 4234, 'n_tokens_neox': 3916, 'n_words': 1065}