//
// See the file LICENSE for redistribution information.
//
// Copyright (c) 2002-2003
//	Sleepycat Software.  All rights reserved.
//

static const char revid[] = "$Id: Database.cpp,v 1.79 2003/10/13 21:45:12 merrells Exp $";

#include "dbxml_config.h"
#include "dbxml/XmlPortability.hpp"
#include "Database.hpp"
#include "Cursor.hpp"
#include "Key.hpp"
#include "ID.hpp"
#include "OperationContext.hpp"
#include "db_rdbt.h"

#include <cassert>
#include <cerrno>

extern "C" int __db_prheader(DB *, char *, int, int, void *,
				     int (*)(void *, const void *),
				    /* VRFY_DBINFO */ void *,  db_pgno_t);
extern "C" int __db_prdbt (DBT *, int, const char *, void *,
				   int (*)(void *, const void *), int,
				  /* VRFY_DBINFO */ void *);
extern "C" int __db_prfooter(void *, int (*)(void *, const void *));

using namespace std;
using namespace DbXml;

// Database

Database::Database(DbEnv *environment, const std::string &containerName, const std::string &prefixName, const std::string &databaseName, u_int32_t pageSize, u_int32_t flags)
	: needsToBeClosed_(true),
	containerName_(containerName),
	prefixName_(prefixName),
	databaseName_(databaseName),
	pageSize_(pageSize),
	db_(environment, flags),
	swapBytes_(false),
	environment_(environment),
	duplicates_(false)
{
}

Database::~Database()
{
	if (needsToBeClosed_) {
		try {
			close(0);
		} catch (...) {
			// Exceptions thrown from destructors are a bad thing.
			assert(0);
		}
	}
}

#define	I_DBXML_CHKSUM 0x04000000 // mapped onto db->set_flags(DB_CHKSUM);
#define	I_DBXML_ENCRYPT 0x08000000 // mapped onto db->set_flags(DB_ENCRYPT);

int Database::open(DbTxn *txn, DBTYPE type, u_int32_t flags, int mode)
{
	int err = 0;
	if (pageSize_ > 0) {
		db_.set_pagesize(pageSize_);
	} else {
		pageSize_ = 16 * 1024; // Default page size of 16k, if the user hasn't specified otherwise.
	}

	if(flags&I_DBXML_CHKSUM)
#if (DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR < 2)
		db_.set_flags(DB_CHKSUM_SHA1);
#else
		db_.set_flags(DB_CHKSUM);
#endif

	if(flags&I_DBXML_ENCRYPT)
		db_.set_flags(DB_ENCRYPT);

	string dbname(getDatabaseName());

	// The file is opened in the directory specified by the environment.
	const char *fileName= 0;
	const char *databaseName= 0;
	if(containerName_.length()>0)
	{
		fileName= containerName_.c_str();
		databaseName= dbname.c_str();
	}
	err = db_.open(txn, fileName, databaseName, type, flags, mode);
	if (err == 0) {
		// Have we opened a database with different endianness?
		int isSwapped = 0;
		db_.get_byteswapped(&isSwapped); // almost certainly won't fail.
		swapBytes_ = (isSwapped == 0 ? false : true);

		if (pageSize_ == 0) {
			// Find out the page size of the underlying database.
			// We don't use stat because it will fail with
			// 'Invalid argument' if the open is within a txn.
			//
			pageSize_ = db_.get_DB()->pgsize;
		}
	}
	return err;
}

int Database::close(u_int32_t flags)
{
	// We are forgiving and allow close to be called even if the db
	// isn't actually open. This is for the error handling code in
	// Container. It calls close on all the Database objects to make
	// sure they are all in the closed state before they are destroyed.
	//
	int err = 0;
	if (needsToBeClosed_) {
		needsToBeClosed_ = false; // db.close kills the db handle even if it fails.
		err = db_.close(flags);
	}
	return err;
}

/*
 * pr_callback - C++ callback function for using pr_* functions from C++.
 */
extern "C"
{
	static int pr_callback(void *handle, const void *str_arg) {
		std::ostream &out = *(std::ostream *)handle;
		const char *str = (const char *)str_arg;

		out << str;
		return (0);
	}
} /* extern "C" */

int Database::dump(std::ostream *out, u_int32_t flags)
{
	UNUSED(flags);

	int ret, t_ret;

	if ((ret = open(NULL, DB_UNKNOWN, 0, 0)) == 0) {
		/*
		 * Get a cursor and step through the database, printing out each
		 * key/data pair.
		 */
		Cursor cursor(environment_, db_, NULL);
		DbtOut key, data;
		DBTYPE db_type;
		int is_recno;

		if ((ret = db_.get_type(&db_type)) != 0)
			goto err;

		is_recno = (db_type == DB_RECNO || db_type == DB_QUEUE);

		if ((ret = __db_prheader(db_.get_DB(), NULL, 0, 1, out, pr_callback, NULL, 0)) != 0)
			goto err;

		while ((ret = cursor.get(&key, &data, DB_NEXT)) == 0) {
			if ((ret = __db_prdbt(key.get_DBT(), 0, " ", out, pr_callback, is_recno, NULL)) != 0 ||
			    (ret = __db_prdbt(data.get_DBT(), 0, " ", out, pr_callback, 0, NULL)) != 0) {
				db_.errx("Failed to print a DBT");
				goto err;
			}
		}

		if (ret == DB_NOTFOUND)
			ret = 0; /* normal case */
		else
			db_.err(ret, "cursor.get");

		(void)__db_prfooter(out, pr_callback);
	}

	if ((t_ret = close(0)) != 0 && ret == 0)
		ret = t_ret;

err:
	return (ret);
}

extern "C"
{
	/*
	 * Implementation of READ_FN for reading from a C++ istream.
	 * Reads at most 'len' characters into 'buf' up to the first 'until' characater
	 * (if non-zero). The terminator (if found) is discarded, and the string is nul
	 * terminated if 'len' > 1.
	 * Returns: zero on success, DB_NOTFOUND if 'until' != 0 and not found, or
	 * EOF on EOF
	 */
	static int
	read_callback(char *buf, size_t len, char until, void *handle) {
		std::istream &in = *(std::istream *)handle;
		char *p = buf;
		char c;
		size_t bytes_read;

		for (bytes_read = 0; bytes_read < len; bytes_read++) {
			if (!in.get(c) || (until != 0 && c == until))
				break;
			else
				*p++ = c;
		}

		if (bytes_read < len)
			*p = '\0';

		if (!in && bytes_read == 0)
			return (EOF);
		else if (until != 0 && c != until)
			return (DB_NOTFOUND);
		else
			return (0);
	}
} /* extern "C" */

int Database::load(std::istream *in, unsigned long *lineno, u_int32_t flags)
{
	UNUSED(flags);

	int version, ret, t_ret;
	DBTYPE dbtype;
	char *subdb;
	u_int32_t read_flags;
	Dbt key, data;
	db_recno_t recno, datarecno;
	DB_ENV *dbenv = environment_ ? environment_->get_DB_ENV() : 0;

	if ((ret = __db_rheader(dbenv, db_.get_DB(), &dbtype,
				&subdb, &version, &read_flags, read_callback, in, lineno)) != 0)
		goto err;

	/* We always print with keys */
	if (!(read_flags & DB_READ_HASKEYS)) {
		db_.errx("Invalid DbXml dump: keys missing");
		ret = EINVAL;
		goto err;
	}

	if ((ret = open(NULL, dbtype, DB_CREATE, 0)) != 0)
		goto err;

	/* Initialize the key/data pair. */
	if (dbtype == DB_RECNO || dbtype == DB_QUEUE) {
		key.set_size(sizeof(recno));
		key.set_data(&datarecno);
	} else {
		key.set_ulen(1024);
		key.set_data((void *)malloc(key.get_ulen()));
	}
	data.set_ulen(1024);
	data.set_data((void *)malloc(data.get_ulen()));
	if (key.get_data() == NULL || data.get_data() == NULL) {
		db_.err(ENOMEM, NULL);
		goto err;
	}

	/* Get each key/data pair and add them to the database. */
	for (recno = 1;; ++recno) {
		if ((ret = __db_rdbt(dbenv, key.get_DBT(), data.get_DBT(),
				     read_flags, read_callback, in, lineno)) != 0) {
			if (ret == EOF)
				ret = 0;
			break;
		}

		switch (ret = db_.put(NULL, &key, &data, 0)) {
		case 0:
			break;
		case DB_KEYEXIST:
			db_.errx("line %d: key already exists, not loaded:", lineno);

			(void)__db_prdbt(key.get_DBT(), read_flags & DB_READ_PRINTABLE,
					 0, &std::cerr, pr_callback, 0, NULL);
			break;
		default:
			db_.err(ret, NULL);
			goto err;
		}
	}

err:	/* Close the database. */
	if ((t_ret = close(0)) != 0 && ret == 0)
		ret = t_ret;

	/* Free allocated memory. */
	if (subdb != NULL)
		free(subdb);
	if (dbtype != DB_RECNO && dbtype != DB_QUEUE)
		free(key.get_data());
	if (data.get_data() != NULL)
		free(data.get_data());

	return (ret);
}

int Database::verify(std::ostream *out, u_int32_t flags)
{
#if DB_VERSION_MAJOR > 4 || (DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 2)
	needsToBeClosed_ = false; // db.verify kills the db handle even if it fails.
#endif
	return 
		db_.verify(
			containerName_.length() == 0 ? 0 : containerName_.c_str(), 
			getDatabaseName().c_str(), 
			out, 
			flags);
}

const char *Database::operationToString(Database::Operation operation)
{
	switch (operation) {
	case NONE:
		return "none";
	case ALL:
		return "all";
	case EQUALITY:
		return "=";
	case LTX:
		return "<";
	case LTE:
		return "<=";
	case GTX:
		return ">";
	case GTE:
		return ">=";
	case RANGE:
		return "range";
	case PREFIX:
		return "prefix";
	case SUBSTRING:
		return "substring";
	}
	return "unknown";
}

u_int32_t Database::getPageSize()
{
	return pageSize_;
}

unsigned long Database::getNumberOfPages() const
{
	return 0xFFFF; // JCM - Let's just assume that a sequential scan is very expensive.
}

// PrimaryDatabase

PrimaryDatabase::PrimaryDatabase(DbEnv *environment, const std::string &containerName, const std::string& databaseName, u_int32_t pageSize, u_int32_t flags)
	: Database(environment, containerName, "primary_", databaseName, pageSize, flags)
{}

PrimaryDatabase::~PrimaryDatabase()
{}

int PrimaryDatabase::open(DbTxn *txn, u_int32_t flags, int mode)
{
	return Database::open(txn, DB_RECNO, flags, mode);
}

int PrimaryDatabase::putPrimary(DbTxn *txn, ID &id, Dbt *data, u_int32_t flags)
{
	u_int32_t idt = 0;
	DbtIn key(&idt, sizeof(idt));
	int err = db_.put(txn, &key, data, flags | DB_APPEND);
	if (err == 0) {
		id.setThisFromDbt(key, swap());
	}
	return err;
}

int PrimaryDatabase::deletePrimary(DbTxn *txn, ID id, u_int32_t flags)
{
	DbtOut key;
	id.setDbtFromThis(key, swap());
	return db_.del(txn, &key, flags);
}

int PrimaryDatabase::createCursor(DbTxn *txn, scoped_ptr<PrimaryCursor> &cursor)
{
	cursor.reset(new PrimaryCursor(*this, txn, ALL, 0));
	return cursor->error();
}

// SecondaryDatabase

SecondaryDatabase::SecondaryDatabase(DbEnv *environment, const std::string &containerName, const std::string& databaseName, const Syntax *syntax, u_int32_t pageSize, u_int32_t flags)
	: Database(environment, containerName, "secondary_", databaseName, pageSize, flags),
	syntax_(syntax)
{
	if (syntax != 0) {
		Database::bt_compare_fn fn = syntax->get_bt_compare();
		if (fn != 0) {
			db_.set_bt_compare(fn);
		}
	}
}

SecondaryDatabase::~SecondaryDatabase()
{
	syntax_ = 0;
}

int SecondaryDatabase::open(DbTxn *txn, bool duplicates, u_int32_t flags, int mode)
{
	if (duplicates) {
		duplicates_= true;
		db_.set_flags(DB_DUP);
	}
	return Database::open(txn, DB_BTREE, flags, mode);
}

// We assume that key has the correct endianness.
int SecondaryDatabase::putID(OperationContext &context, const void *key, size_t length, const ID &id, bool *duplicate)
{
	int err = 0;
	DbtIn dbkey((void*)key, length);
	if (duplicate != 0) {
		// Check if the key already exists.
		// DB_DBT_PARTIAL and len=0 prevents retrieval of the data.
		DbtIn dbt;
		dbt.set_flags(DB_DBT_PARTIAL);
		err = db_.get(context.txn(), &dbkey, &dbt, /*no flags*/0);
		*duplicate = (err != DB_NOTFOUND);
	}
	if (err == 0 || err == DB_NOTFOUND) {
		// Add the key and data.
		id.setDbtFromThis(context.data(), swap());
		err = db_.put(context.txn(), &dbkey, &context.data(), /*no flags*/0);
	}
	return err;
}

// We assume that key has the correct endianness.
int SecondaryDatabase::delID(OperationContext &context, const void *key, size_t length, ID id, bool *duplicate)
{
	Cursor cursor(environment_, db_, context.txn());
	int err = cursor.error();
	if (err == 0) {
		// If a document contains multiple nodes with the same value, then
		// the indexer will generate duplicate keys for the same document.
		// We loop here to make sure we delete all of them.
		//
		context.key().set((void*)key, length);
		id.setDbtFromThis(context.data(), swap());
		while (err == 0) {
			err = cursor.get(&context.key(), &context.data(), DB_GET_BOTH);
			if (err == 0) {
				err = cursor.del( /*no flags*/0);
			}
		}
		if (duplicate != 0) {
			// Check if the key still exists.
			// DB_DBT_PARTIAL and len=0 prevents retrieval of the data.
			DbtIn dbt;
			dbt.set_flags(DB_DBT_PARTIAL);
			err = cursor.get(&context.key(), &dbt, DB_SET);
			*duplicate = (err != DB_NOTFOUND);
		}
		err = 0;
	}
	return err;
}

double SecondaryDatabase::cost(OperationContext &context, Operation operation, Operation gto, Operation lto, const Key &key1, const Key &key2)
{
	DbtOut &dbt1 = context.key();
	DbtOut &dbt2 = context.data();

	DB_KEY_RANGE krMin;
	Key keyMin;
	keyMin.set(key1, syntax_->minimumValue(), strlen(syntax_->minimumValue()));
	keyMin.setDbtFromThis(dbt1, swap());
	db_.key_range(context.txn(), &dbt1, &krMin, 0);

	DB_KEY_RANGE krMax;
	Key keyMax;
	keyMax.set(key1, syntax_->maximumValue(), strlen(syntax_->maximumValue()));
	keyMax.setDbtFromThis(dbt1, swap());
	db_.key_range(context.txn(), &dbt1, &krMax, 0);

	// range is the % of the database keys that the keys for this index occupy.
	double range = krMax.less + krMax.equal - krMin.less;
	double extent = 0.0;

	if (range > 0.0) {
		// extent is the % of the database keys that the keys for this index match this operation.
		DB_KEY_RANGE kr1;
		DB_KEY_RANGE kr2;
		if (operation == Database::PREFIX) {
			key1.setDbtFromThis(dbt1, swap());
			db_.key_range(context.txn(), &dbt1, &kr1, 0);
			Key t;
			t.set(key1, key1.getValue(), key1.getValueSize());
			unsigned char ff = 0xff;
			t.addValue((const char *)&ff, 1);
			t.setDbtFromThis(dbt2, swap());
			db_.key_range(context.txn(), &dbt2, &kr2, 0);
			extent = kr2.less - kr1.less;
		} else if (operation == Database::LTX || operation == Database::LTE) {
			key1.setDbtFromThis(dbt2, swap());
			db_.key_range(context.txn(), &dbt2, &kr2, 0);
			extent = kr2.less - krMin.less + (operation == Database::LTE ? kr2.equal : 0);
		} else if (operation == Database::GTX || operation == Database::GTE) {
			key1.setDbtFromThis(dbt1, swap());
			db_.key_range(context.txn(), &dbt1, &kr1, 0);
			extent = krMax.less + krMax.equal - kr1.less + (operation == Database::GTX ? kr1.equal : 0);
		} else if (operation == Database::RANGE) {
			key1.setDbtFromThis(dbt1, swap());
			db_.key_range(context.txn(), &dbt1, &kr1, 0);
			key2.setDbtFromThis(dbt2, swap());
			db_.key_range(context.txn(), &dbt2, &kr2, 0);
			extent = kr2.less - kr1.less + (lto == Database::LTE ? kr2.equal : 0) + (gto == Database::GTX ? kr1.equal : 0);
		}
	}

	// extent/range is the % of keys within this index that match this operation.
	return (range == 0 ? 0 : extent / range);
}

int SecondaryDatabase::createCursor(DbTxn *txn, scoped_ptr<SecondaryCursor> &cursor, Database::Operation operation, Key *key)
{
	cursor.reset(new SecondaryCursor(*this, txn, operation, key, syntax_));
	return cursor->error();
}

int SecondaryDatabase::createCursor(DbTxn *txn, scoped_ptr<SecondaryCursor> &cursor, Database::Operation gto, Key *gtk, Database::Operation lto, Key *ltk)
{
	cursor.reset(new SecondaryCursor(*this, txn, gto, gtk, lto, ltk, syntax_));
	return cursor->error();
}
