int rl_list_iterator_next(rl_list_iterator *iterator, void **element)
{
	int retval;
	if (iterator->node == NULL) {
		retval = RL_END;
		goto cleanup;
	}
	if (element) {
		RL_MALLOC(*element, iterator->list->type->element_size);
		memcpy(*element, iterator->node->elements[iterator->node_position], iterator->list->type->element_size);
	}
	iterator->node_position += iterator->direction;
	if (iterator->node_position < 0 || iterator->node_position == iterator->node->size) {
		long next_node_page = iterator->direction == 1 ? iterator->node->right : iterator->node->left;
		RL_CALL(rl_list_node_nocache_destroy, RL_OK, iterator->db, iterator->node);
		iterator->node = NULL;
		if (next_node_page) {
			void *_node;
			RL_CALL(rl_read, RL_FOUND, iterator->db, iterator->list->type->list_node_type, next_node_page, iterator->list, &_node, 0);
			iterator->node = _node;
			iterator->node_position = iterator->direction == 1 ? 0 : (iterator->node->size - 1);
		}
	}
	if (iterator->node && iterator->node_position < -1) {
		retval = RL_UNEXPECTED;
		goto cleanup;
	}
	retval = RL_OK;
cleanup:
	if (retval != RL_OK) {
		rl_list_iterator_destroy(iterator->db, iterator);
	}
	return retval;
}

int rl_rename(struct rlite *db, const unsigned char *src, long srclen, const unsigned char *target, long targetlen, int overwrite)
{
	int retval;
	unsigned char type;
	unsigned long long expires;
	long value_page;
	long version = 0;
	if (overwrite) {
		RL_CALL2(rl_key_get, RL_FOUND, RL_NOT_FOUND, db, target, targetlen, NULL, NULL, NULL, NULL, &version);
		if (retval == RL_FOUND) {
			RL_CALL(rl_key_delete_with_value, RL_OK, db, target, targetlen);
			version++;
		}
	}
	else {
		RL_CALL(rl_key_get, RL_NOT_FOUND, db, target, targetlen, NULL, NULL, NULL, NULL, NULL);
	}
	// this could be more efficient, if we don't delete the value page
	RL_CALL(rl_key_get, RL_FOUND, db, src, srclen, &type, NULL, &value_page, &expires, NULL);
	RL_CALL(rl_key_delete, RL_OK, db, src, srclen);
	RL_CALL(rl_key_set, RL_OK, db, target, targetlen, type, value_page, expires, version);
	retval = RL_OK;
cleanup:
	return retval;
}

int rl_list_iterator_create(rlite *db, rl_list_iterator **_iterator, rl_list *list, int direction)
{
	void *_node;
	int retval;
	rl_list_iterator *iterator = NULL;
	RL_MALLOC(iterator, sizeof(*iterator));
	iterator->db = db;
	iterator->list = list;
	iterator->node = NULL;
	if (direction < 0) {
		iterator->direction = -1;
		RL_CALL(rl_read, RL_FOUND, db, list->type->list_node_type, list->right, list, &_node, 0);
		iterator->node = _node;
		iterator->node_position = iterator->node->size - 1;
	}
	else {
		iterator->direction = 1;
		RL_CALL(rl_read, RL_FOUND, db, list->type->list_node_type, list->left, list, &_node, 0);
		iterator->node = _node;
		iterator->node_position = 0;
	}
	*_iterator = iterator;
	retval = RL_OK;
cleanup:
	if (iterator && retval != RL_OK) {
		rl_list_iterator_destroy(db, iterator);
	}
	return retval;
}

int rl_multi_string_cpyrange(struct rlite *db, long number, unsigned char *data, long *_size, long start, long stop)
{
	long totalsize;
	rl_list *list = NULL;
	rl_list_node *node = NULL;
	void *_list, *tmp;
	int retval;
	RL_CALL(rl_read, RL_FOUND, db, &rl_data_type_list_long, number, &rl_list_type_long, &_list, 0);
	list = _list;
	unsigned char *tmp_data;
	long i, pos = 0, pagesize, pagestart;
	long size;

	RL_CALL(rl_list_get_element, RL_FOUND, db, list, &tmp, 0);
	totalsize = *(long *)tmp;
	if (totalsize == 0) {
		if (_size) {
			*_size = 0;
		}
		retval = RL_OK;
		goto cleanup;
	}
	rl_normalize_string_range(totalsize, &start, &stop);
	if (stop < start) {
		if (_size) {
			*_size = 0;
		}
		retval = RL_OK;
		goto cleanup;
	}
	size = stop - start + 1;
	if (_size) {
		*_size = size;
	}

	i = start / db->page_size;
	pagestart = start % db->page_size;
	// pos = i * db->page_size + pagestart;
	// the first element in the list is the length of the array, skip to the second
	for (i++; i < list->size; i++) {
		RL_CALL(rl_list_get_element, RL_FOUND, db, list, &tmp, i);
		RL_CALL(rl_string_get, RL_OK, db, &tmp_data, *(long *)tmp);
		pagesize = db->page_size - pagestart;
		if (pos + pagesize > size) {
			pagesize = size - pos;
		}
		memcpy(&data[pos], &tmp_data[pagestart], sizeof(unsigned char) * pagesize);
		pos += pagesize;
		pagestart = 0;
	}
	retval = RL_OK;
cleanup:
	if (list) {
		rl_list_nocache_destroy(db, list);
	}
	if (node) {
		rl_list_node_nocache_destroy(db, node);
	}
	return retval;
}

int rl_zremrangebylex(rlite *db, const unsigned char *key, long keylen, unsigned char *min, long minlen, unsigned char *max, long maxlen, long *changed)
{
	rl_zset_iterator *iterator;
	rl_btree *scores;
	rl_skiplist *skiplist;
	long scores_page, skiplist_page, start, end, levels_page_number;
	int retval;
	RL_CALL(validate_lex_range, RL_OK, min, minlen, max, maxlen);
	RL_CALL(rl_zset_get_objects, RL_OK, db, key, keylen, &levels_page_number, &scores, &scores_page, &skiplist, &skiplist_page, 1, 1);
	retval = lex_get_range(db, min, minlen, max, maxlen, skiplist, &start, &end);
	if (retval == RL_NOT_FOUND) {
		*changed = 0;
		retval = RL_OK;
		goto cleanup;
	}
	if (retval != RL_OK) {
		goto cleanup;
	}

	RL_CALL(_rl_zrange, RL_OK, db, skiplist, start, end, 1, &iterator);
	RL_CALL(_zremiterator, RL_OK, db, key, keylen, levels_page_number, iterator, scores, scores_page, skiplist, skiplist_page, changed);
	retval = RL_OK;
cleanup:
	if (retval != RL_OK && changed) {
		*changed = 0;
	}
	return retval;
}

int rl_key_set(rlite *db, const unsigned char *key, long keylen, unsigned char type, long value_page, unsigned long long expires, long version)
{
	int retval;

	rl_key *key_obj = NULL;
	unsigned char *digest = NULL;
	RL_CALL2(rl_key_delete, RL_OK, RL_NOT_FOUND, db, key, keylen);
	RL_MALLOC(digest, sizeof(unsigned char) * 20);
	RL_CALL(sha1, RL_OK, key, keylen, digest);
	rl_btree *btree;
	RL_CALL(rl_get_key_btree, RL_OK, db, &btree, 1);
	RL_MALLOC(key_obj, sizeof(*key_obj))
	RL_CALL(rl_multi_string_set, RL_OK, db, &key_obj->string_page, key, keylen);
	key_obj->type = type;
	key_obj->value_page = value_page;
	key_obj->expires = expires;
	// reserving version=0 for non existent keys
	if (version == 0) {
		version = 1;
	}
	key_obj->version = version;

	RL_CALL(rl_btree_add_element, RL_OK, db, btree, db->databases[rl_get_selected_db(db)], digest, key_obj);
	retval = RL_OK;
cleanup:
	if (retval != RL_OK) {
		rl_free(digest);
		rl_free(key_obj);
	}
	return retval;
}

int rl_print_list(rlite *db, rl_list *list)
{
	if (!list->type->formatter) {
		fprintf(stderr, "Trying to print an element without formatter\n");
		return RL_UNEXPECTED;
	}
	printf("-------\n");
	rl_list_node *node;
	void *_node;
	char *element;
	int size;
	long i, number = list->left;
	int retval = RL_OK;
	while (number != 0) {
		RL_CALL(rl_read, RL_FOUND, db, list->type->list_node_type, number, list, &_node, 1);
		node = _node;
		for (i = 0; i < node->size; i++) {
			RL_CALL(list->type->formatter, RL_OK, node->elements[i], &element, &size);
			fwrite(element, sizeof(char), size, stdout);
			rl_free(element);
			printf("\n");
		}
		number = node->right;
	}
	printf("-------\n");
cleanup:
	return retval;
}

int rl_read_header(rlite *db)
{
	db->page_size = HEADER_SIZE;
	int retval;
	if (db->driver_type == RL_MEMORY_DRIVER) {
		db->page_size = DEFAULT_PAGE_SIZE;
		RL_CALL(rl_create_db, RL_OK, db);
	}
	else if (db->driver_type == RL_FILE_DRIVER) {
		RL_CALL(file_driver_fp, RL_OK, db);
		RL_CALL(rl_apply_wal, RL_OK, db);
		retval = rl_read(db, &rl_data_type_header, 0, NULL, NULL, 1);
		if (retval == RL_NOT_FOUND && rl_has_flag(db, RLITE_OPEN_CREATE)) {
			db->page_size = DEFAULT_PAGE_SIZE;
			RL_CALL(rl_create_db, RL_OK, db);
			RL_CALL(rl_write, RL_OK, db, &rl_data_type_header, 0, NULL);
		}
		else if (retval != RL_FOUND) {
			goto cleanup;
		}
	} else {
		fprintf(stderr, "Unknown driver type %d\n", db->driver_type);
		retval = RL_UNEXPECTED;
		goto cleanup;
	}
	retval = RL_OK;
cleanup:
	return retval;
}

static int _rl_zrange(rlite *db, rl_skiplist *skiplist, long start, long end, int direction, rl_zset_iterator **iterator)
{
	int retval = RL_OK;
	long size, node_page;
	long card = skiplist->size;

	if (start < 0) {
		start += card;
		if (start < 0) {
			start = 0;
		}
	}
	if (end < 0) {
		end += card;
	}
	if (start > end || start >= card) {
		retval = RL_NOT_FOUND;
		goto cleanup;
	}
	if (end >= card) {
		end = card - 1;
	}

	size = end - start + 1;

	RL_CALL(rl_skiplist_node_by_rank, RL_OK, db, skiplist, direction > 0 ? start : end, NULL, &node_page);
	RL_CALL(rl_skiplist_iterator_create, RL_OK, db, iterator, skiplist, node_page, direction, size);
cleanup:
	return retval;
}

int rl_key_delete(struct rlite *db, const unsigned char *key, long keylen)
{
	int retval;
	void *tmp;
	unsigned char *digest;
	rl_btree *btree = NULL;
	rl_key *key_obj = NULL;
	RL_MALLOC(digest, sizeof(unsigned char) * 20);
	RL_CALL(sha1, RL_OK, key, keylen, digest);
	RL_CALL(rl_get_key_btree, RL_OK, db, &btree, 0);
	retval = rl_btree_find_score(db, btree, digest, &tmp, NULL, NULL);
	if (retval == RL_FOUND) {
		int selected_database = rl_get_selected_db(db);
		key_obj = tmp;
		RL_CALL(rl_multi_string_delete, RL_OK, db, key_obj->string_page);
		retval = rl_btree_remove_element(db, btree, db->databases[selected_database], digest);
		if (retval == RL_DELETED) {
			db->databases[selected_database] = 0;
			retval = RL_OK;
		}
		else if (retval != RL_OK) {
			goto cleanup;
		}
	}
cleanup:
	rl_free(digest);
	return retval;
}

static int remove_member_score(rlite *db, const unsigned char *key, long keylen, long levels_page_number, rl_btree *scores, long scores_page, rl_skiplist *skiplist, long skiplist_page, unsigned char *member, long member_len, double score)
{
	unsigned char digest[20];
	int retval;
	RL_CALL(sha1, RL_OK, member, member_len, digest);
	RL_CALL(remove_member_score_sha1, RL_OK, db, key, keylen, levels_page_number, scores, scores_page, skiplist, skiplist_page, member, member_len, score, digest);
cleanup:
	return retval;
}

int rl_zscore(rlite *db, const unsigned char *key, long keylen, unsigned char *member, long memberlen, double *score)
{
	rl_btree *scores;
	int retval;
	RL_CALL(rl_zset_get_objects, RL_OK, db, key, keylen, NULL, &scores, NULL, NULL, NULL, 0, 0);
	RL_CALL(rl_get_zscore, RL_FOUND, db, scores, member, memberlen, score);
cleanup:
	return retval;
}

int rl_key_expires(struct rlite *db, const unsigned char *key, long keylen, unsigned long long expires)
{
	int retval;
	unsigned char type;
	long string_page, value_page, version;
	RL_CALL(rl_key_get, RL_FOUND, db, key, keylen, &type, &string_page, &value_page, NULL, &version);
	RL_CALL(rl_key_set, RL_OK, db, key, keylen, type, value_page, expires, version + 1);
cleanup:
	return retval;
}

static int getScript(rliteClient *c, char hash[40], char **script, long *scriptlen) {
	int retval;
	RL_CALL(rl_select_internal, RL_OK, c->context->db, RLITE_INTERNAL_DB_LUA);

	RL_CALL(rl_get, RL_OK, c->context->db, (unsigned char *)hash, 40, (unsigned char **)script, scriptlen);

cleanup:
	rl_select_internal(c->context->db, RLITE_INTERNAL_DB_NO);
	return retval;
}

int rl_zrange(rlite *db, const unsigned char *key, long keylen, long start, long end, rl_zset_iterator **iterator)
{
	rl_skiplist *skiplist;

	int retval;
	RL_CALL(rl_zset_get_objects, RL_OK, db, key, keylen, NULL, NULL, NULL, &skiplist, NULL, 0, 0);
	RL_CALL(_rl_zrange, RL_OK, db, skiplist, start, end, 1, iterator);
cleanup:
	return retval;
}

int rl_refresh(rlite *db)
{
	int retval = RL_OK;
	if (db->driver_type == RL_FILE_DRIVER) {
		RL_CALL(rl_discard, RL_OK, db);
		RL_CALL(rl_read_header, RL_OK, db);
	}
cleanup:
	return retval;
}

int rl_randomkey(struct rlite *db, unsigned char **key, long *keylen)
{
	int retval;
	rl_btree *btree;
	rl_key *key_obj;
	RL_CALL(rl_get_key_btree, RL_OK, db, &btree, 0);
	RL_CALL(rl_btree_random_element, RL_OK, db, btree, NULL, (void **)&key_obj);
	RL_CALL(rl_multi_string_get, RL_OK, db, key_obj->string_page, key, keylen);
cleanup:
	return retval;
}

static int rl_find_element_by_position(rlite *db, rl_list *list, long *position, long *_pos, rl_list_node **_node, long *_number, int add)
{
	rl_list_node *node;
	void *tmp_node;
	long pos = 0, number;
	if (*position >= list->size + add || *position <= - list->size - 1 - add) {
		return RL_INVALID_PARAMETERS;
	}
	int retval = RL_OK;
	if (*position >= 0) {
		number = list->left;
		while (1) {
			RL_CALL(rl_read, RL_FOUND, db, list->type->list_node_type, number, list, &tmp_node, 1);
			node = tmp_node;
			if (pos + node->size > *position) {
				break;
			}
			if (node->right != 0) {
				number = node->right;
				pos += node->size;
			}
			else {
				break;
			}
		}
	}
	else {
		*position = list->size + *position + add;
		pos = list->size;
		number = list->right;
		while (1) {
			RL_CALL(rl_read, RL_FOUND, db, list->type->list_node_type, number, list, &tmp_node, 1);
			node = tmp_node;
			pos -= node->size;
			if (pos <= *position) {
				break;
			}
			if (node->left != 0) {
				number = node->left;
			}
			else {
				break;
			}
		}
	}
	*_pos = pos;
	*_node = node;
	if (_number) {
		*_number = number;
	}
cleanup:
	return retval;
}

int rl_zrank(rlite *db, const unsigned char *key, long keylen, unsigned char *member, long memberlen, long *rank)
{
	double score;
	rl_btree *scores;
	rl_skiplist *skiplist;
	int retval;
	RL_CALL(rl_zset_get_objects, RL_OK, db, key, keylen, NULL, &scores, NULL, &skiplist, NULL, 0, 0);
	RL_CALL(rl_get_zscore, RL_FOUND, db, scores, member, memberlen, &score);
	RL_CALL(rl_skiplist_first_node, RL_FOUND, db, skiplist, score, RL_SKIPLIST_INCLUDE_SCORE, member, memberlen, NULL, rank);
cleanup:
	return retval;
}

int rl_key_delete_with_value(struct rlite *db, const unsigned char *key, long keylen)
{
	int retval;
	unsigned char identifier;
	long value_page;
	unsigned long long expires;
	RL_CALL(rl_key_get_ignore_expire, RL_FOUND, db, key, keylen, &identifier, NULL, &value_page, &expires, NULL, 1);
	RL_CALL(rl_key_delete_value, RL_OK, db, identifier, value_page);
	RL_CALL(rl_key_delete, RL_OK, db, key, keylen);
	retval = expires != 0 && expires <= rl_mstime() ? RL_NOT_FOUND : RL_OK;
cleanup:
	return retval;
}