/*
 * namespace.c
 *
 * Copyright (C) 2012-2020 Aerospike, Inc.
 *
 * Portions may be licensed to Aerospike, Inc. under one or more contributor
 * license agreements.
 *
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License as published by the Free
 * Software Foundation, either version 3 of the License, or (at your option) any
 * later version.
 *
 * 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 Affero General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see http://www.gnu.org/licenses/
 */

//==========================================================
// Includes.
//

#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <string.h>

#include "aerospike/as_atomic.h"
#include "citrusleaf/alloc.h"

#include "cf_mutex.h"
#include "dynbuf.h"
#include "hist.h"
#include "linear_hist.h"
#include "log.h"
#include "vmapx.h"
#include "xmem.h"

#include "base/cfg.h"
#include "base/datamodel.h"
#include "base/index.h"
#include "base/proto.h"
#include "base/thr_info.h"
#include "fabric/partition.h"
#include "sindex/sindex.h"
#include "storage/storage.h"


//==========================================================
// Forward declarations.
//

static void append_set_props(as_set *p_set, cf_dyn_buf *db);


//==========================================================
// Public API.
//

as_namespace *
as_namespace_create(char *name)
{
	cf_assert_nostack(strlen(name) < AS_ID_NAMESPACE_SZ,
			AS_NAMESPACE, "{%s} namespace name too long (max length is %u)",
			name, AS_ID_NAMESPACE_SZ - 1);

	cf_assert_nostack(g_config.n_namespaces < AS_NAMESPACE_SZ,
			AS_NAMESPACE, "too many namespaces (max is %u)", AS_NAMESPACE_SZ);

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		as_namespace *ns = g_config.namespaces[ns_ix];

		if (strcmp(ns->name, name) == 0) {
			cf_crash_nostack(AS_NAMESPACE, "{%s} duplicate namespace", name);
		}
	}

	as_namespace *ns = cf_malloc(sizeof(as_namespace));

	g_config.namespaces[g_config.n_namespaces] = ns;

	// Set all members 0/NULL/false to start with.
	memset(ns, 0, sizeof(as_namespace));

	strcpy(ns->name, name);
	ns->ix = g_config.n_namespaces++;

	cf_mutex_init(&ns->query_rsvs_lock); // nowhere better to do this

	//--------------------------------------------
	// Non-0/NULL/false configuration defaults.
	//

	ns->pi_xmem_type = CF_XMEM_TYPE_MEM;
	ns->si_xmem_type = CF_XMEM_TYPE_MEM;

	ns->background_query_max_rps = 10000; // internal write generation limit
	ns->conflict_resolution_policy = AS_NAMESPACE_CONFLICT_RESOLUTION_POLICY_UNDEF;
	ns->evict_hist_buckets = 10000; // for 30 day TTL, bucket width is 4 minutes 20 seconds
	ns->evict_tenths_pct = 5; // default eviction amount is 0.5%
	ns->index_stage_size = 1024L * 1024L * 1024L; // 1G
	ns->max_record_size = 1024 * 1024; // 1M
	ns->migrate_order = 5;
	ns->migrate_retransmit_ms = 1000 * 5; // 5 seconds
	ns->migrate_sleep = 1;
	ns->mrt_duration = 0; // will be  set later to DEFAULT_MRT_DURATION (10 sec) if applicable and not explicitly configured
	ns->nsup_hist_period = 60 * 60; // 1 hour
	ns->n_nsup_threads = 1;
	ns->tree_shared.n_sprigs = NUM_LOCK_PAIRS; // can't be less than number of lock pairs, 256 per partition
	ns->read_consistency_level = AS_READ_CONSISTENCY_LEVEL_PROTO;
	ns->cfg_replication_factor = 2;
	ns->replication_factor = 0; // gets set on rebalance
	ns->sindex_stage_size = 1024L * 1024L * 1024L; // 1G
	ns->n_single_query_threads = 4; // maximum number of threads a single query may run
	ns->stop_writes_sys_memory_pct = 90; // stop writes when 90% of system memory is used
	ns->tomb_raider_eligible_age = 60 * 60 * 24; // 1 day
	ns->tomb_raider_period = 60 * 60 * 24; // 1 day
	ns->transaction_pending_limit = 20;
	ns->n_truncate_threads = 4;
	ns->write_commit_level = AS_WRITE_COMMIT_LEVEL_PROTO;
	ns->xdr_bin_tombstone_ttl_ms = 60 * 60 * 24 * 1000UL; // 1 day
	ns->xdr_tomb_raider_period = 2 * 60; // 2 minutes
	ns->n_xdr_tomb_raider_threads = 1;

	ns->storage_type = AS_STORAGE_ENGINE_UNDEFINED;

	ns->storage_defrag_lwm_pct = 50; // defrag if occupancy of block is < 50%
	ns->storage_defrag_sleep = 1000; // sleep this many microseconds between each wblock
	ns->storage_encryption = AS_ENCRYPTION_AES_128;
	ns->storage_flush_max_us = 1000 * 1000; // wait this many microseconds before flushing inactive current write buffer (0 = never)
	ns->storage_flush_size = 0; // will be set later to DEFAULT_FLUSH_SIZE (1M) if applicable and not explicitly configured
	ns->storage_max_write_cache = DEFAULT_MAX_WRITE_CACHE;
	ns->storage_post_write_cache = DEFAULT_POST_WRITE_CACHE; // bytes per device used as post-write cache
	ns->storage_stop_writes_avail_pct = 5; // stop writes when < 5% disk is writable
	ns->storage_stop_writes_used_pct = 70; // stop writes when > 70% disk is used
	ns->storage_tomb_raider_sleep = 1000; // sleep this many microseconds between each device read

	ns->geo2dsphere_within_strict = true;
	ns->geo2dsphere_within_min_level = 1;
	ns->geo2dsphere_within_max_level = 20;
	ns->geo2dsphere_within_max_cells = 12;
	ns->geo2dsphere_within_level_mod = 1;
	ns->geo2dsphere_within_earth_radius_meters = 6371000;  // Wikipedia, mean

	// Special defaults that differ between CE and EE.
	as_config_init_namespace(ns);

	return ns;
}


void
as_namespaces_init(bool cold_start_cmd, uint32_t instance)
{
	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		as_namespace *ns = g_config.namespaces[ns_ix];

		// Done with temporary sets configuration array.
		if (ns->sets_cfg_array) {
			cf_free(ns->sets_cfg_array);
		}

		for (uint32_t pid = 0; pid < AS_PARTITIONS; pid++) {
			as_partition_init(ns, pid);
		}

		as_namespace_finish_setup(ns, instance);
	}
}


bool
as_namespace_configure_sets(as_namespace *ns)
{
	for (uint32_t i = 0; i < ns->sets_cfg_count; i++) {
		uint32_t idx;
		cf_vmapx_err result = cf_vmapx_put_unique(ns->p_sets_vmap,
				ns->sets_cfg_array[i].name, &idx);

		if (result == CF_VMAPX_OK || result == CF_VMAPX_ERR_NAME_EXISTS) {
			as_set* p_set = NULL;

			if ((result = cf_vmapx_get_by_index(ns->p_sets_vmap, idx,
					(void**)&p_set)) != CF_VMAPX_OK) {
				// Should be impossible - just verified idx.
				cf_crash(AS_NAMESPACE, "vmap error %d", result);
			}

			// Transfer configurable metadata.
			p_set->stop_writes_count = ns->sets_cfg_array[i].stop_writes_count;
			p_set->stop_writes_size = ns->sets_cfg_array[i].stop_writes_size;
			p_set->default_ttl = ns->sets_cfg_array[i].default_ttl;
			p_set->default_read_touch_ttl_pct =
					ns->sets_cfg_array[i].default_read_touch_ttl_pct;
			p_set->eviction_disabled = ns->sets_cfg_array[i].eviction_disabled;
			p_set->index_enabled = ns->sets_cfg_array[i].index_enabled;
		}
		else {
			// Maybe exceeded max sets allowed, but try failing gracefully.
			cf_warning(AS_NAMESPACE, "vmap error %d", result);
			return false;
		}
	}

	return true;
}


as_namespace *
as_namespace_get_byname(const char *name)
{
	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		as_namespace *ns = g_config.namespaces[ns_ix];

		if (strcmp(ns->name, name) == 0) {
			return ns;
		}
	}

	return NULL;
}


as_namespace *
as_namespace_get_bybuf(const uint8_t *buf, size_t len)
{
	if (len >= AS_ID_NAMESPACE_SZ) {
		return NULL;
	}

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		as_namespace *ns = g_config.namespaces[ns_ix];

		if (memcmp(buf, ns->name, len) == 0 && ns->name[len] == 0) {
			return ns;
		}
	}

	return NULL;
}


as_namespace *
as_namespace_get_bymsgfield(as_msg_field *fp)
{
	return as_namespace_get_bybuf(fp->data, as_msg_field_get_value_sz(fp));
}


const char *
as_namespace_get_set_name(const as_namespace *ns, uint16_t set_id)
{
	// Note that set_id is 1-based, but cf_vmap index is 0-based.
	// (This is because 0 in the index structure means 'no set'.)

	if (set_id == INVALID_SET_ID) {
		return NULL;
	}

	as_set *p_set;

	return cf_vmapx_get_by_index(ns->p_sets_vmap, (uint32_t)set_id - 1,
			(void**)&p_set) == CF_VMAPX_OK ? p_set->name : NULL;
}


uint16_t
as_namespace_get_set_id(as_namespace *ns, const char *set_name)
{
	uint32_t idx;

	return cf_vmapx_get_index(ns->p_sets_vmap, set_name, &idx) == CF_VMAPX_OK ?
			(uint16_t)(idx + 1) : INVALID_SET_ID;
}


// At the moment this is only used by the enterprise security & xdr features.
uint16_t
as_namespace_get_create_set_id(as_namespace *ns, const char *set_name)
{
	if (! set_name) {
		// Should be impossible.
		cf_warning(AS_NAMESPACE, "null set name");
		return INVALID_SET_ID;
	}

	uint32_t idx;
	cf_vmapx_err result = cf_vmapx_get_index(ns->p_sets_vmap, set_name, &idx);

	if (result == CF_VMAPX_OK) {
		return (uint16_t)(idx + 1);
	}

	if (result == CF_VMAPX_ERR_NAME_NOT_FOUND) {
		result = cf_vmapx_put_unique(ns->p_sets_vmap, set_name, &idx);

		if (result == CF_VMAPX_ERR_NAME_EXISTS) {
			return (uint16_t)(idx + 1);
		}

		if (result == CF_VMAPX_ERR_BAD_PARAM) {
			cf_warning(AS_NAMESPACE, "set name %s too long", set_name);
			return INVALID_SET_ID;
		}

		if (result == CF_VMAPX_ERR_FULL) {
			cf_warning(AS_NAMESPACE, "can't add %s (at sets limit)", set_name);
			return INVALID_SET_ID;
		}

		if (result != CF_VMAPX_OK) {
			// Currently, remaining errors are all some form of out-of-memory.
			cf_warning(AS_NAMESPACE, "can't add %s (%d)", set_name, result);
			return INVALID_SET_ID;
		}

		return (uint16_t)(idx + 1);
	}

	// Should be impossible.
	cf_warning(AS_NAMESPACE, "unexpected error %d", result);
	return INVALID_SET_ID;
}


int
as_namespace_set_set_w_len(as_namespace *ns, const char *set_name, size_t len,
		uint16_t *p_set_id, bool apply_count_limit)
{
	as_set *p_set;

	if (as_namespace_get_create_set_w_len(ns, set_name, len, &p_set,
			p_set_id) != 0) {
		return AS_ERR_PARAMETER;
	}

	if (apply_count_limit && as_set_count_stop_writes(p_set)) {
		cf_ticker_warning(AS_NAMESPACE, "{%s|%s} at stop-writes-count - can't add record",
				ns->name, p_set->name);
		return AS_ERR_FORBIDDEN;
	}

	as_incr_uint64(&p_set->n_objects);

	return AS_OK;
}


int
as_namespace_get_create_set_w_len(as_namespace *ns, const char *set_name,
		size_t len, as_set **pp_set, uint16_t *p_set_id)
{
	cf_assert(set_name, AS_NAMESPACE, "null set name");
	cf_assert(len != 0, AS_NAMESPACE, "empty set name");

	uint32_t idx;
	cf_vmapx_err result = cf_vmapx_get_index_w_len(ns->p_sets_vmap, set_name,
			len, &idx);

	if (result == CF_VMAPX_ERR_NAME_NOT_FOUND) {
		// Special case handling for name too long.
		if (len >= AS_SET_NAME_MAX_SIZE) {
			char bad_name[AS_SET_NAME_MAX_SIZE];

			memcpy(bad_name, set_name, AS_SET_NAME_MAX_SIZE - 1);
			bad_name[AS_SET_NAME_MAX_SIZE - 1] = 0;

			cf_warning(AS_NAMESPACE, "set name %s... too long", bad_name);
			return -1;
		}

		result = cf_vmapx_put_unique_w_len(ns->p_sets_vmap, set_name, len,
				&idx);

		// Since this function can be called via many functions simultaneously.
		// Need to handle race, So handle CF_VMAPX_ERR_NAME_EXISTS.
		if (result == CF_VMAPX_ERR_FULL) {
			cf_warning(AS_NAMESPACE, "at set names limit, can't add set");
			return -1;
		}

		if (result != CF_VMAPX_OK && result != CF_VMAPX_ERR_NAME_EXISTS) {
			cf_warning(AS_NAMESPACE, "error %d, can't add set", result);
			return -1;
		}
	}
	else if (result != CF_VMAPX_OK) {
		// Should be impossible.
		cf_warning(AS_NAMESPACE, "unexpected error %d", result);
		return -1;
	}

	if (pp_set) {
		if ((result = cf_vmapx_get_by_index(ns->p_sets_vmap, idx,
				(void**)pp_set)) != CF_VMAPX_OK) {
			// Should be impossible - just verified idx.
			cf_warning(AS_NAMESPACE, "unexpected error %d", result);
			return -1;
		}
	}

	if (p_set_id) {
		*p_set_id = (uint16_t)(idx + 1);
	}

	return 0;
}


as_set *
as_namespace_get_set_by_name(as_namespace *ns, const char *set_name)
{
	uint32_t idx;

	if (cf_vmapx_get_index(ns->p_sets_vmap, set_name, &idx) != CF_VMAPX_OK) {
		return NULL;
	}

	as_set *p_set;

	if (cf_vmapx_get_by_index(ns->p_sets_vmap, idx, (void**)&p_set) !=
			CF_VMAPX_OK) {
		// Should be impossible - just verified idx.
		cf_crash(AS_NAMESPACE, "unexpected vmap error");
	}

	return p_set;
}


as_set *
as_namespace_get_set_by_id(const as_namespace *ns, uint16_t set_id)
{
	if (set_id == INVALID_SET_ID) {
		return NULL;
	}

	as_set *p_set;

	if (cf_vmapx_get_by_index(ns->p_sets_vmap, set_id - 1, (void**)&p_set) !=
			CF_VMAPX_OK) {
		// Should be impossible.
		cf_warning(AS_NAMESPACE, "unexpected - record with set-id not in vmap");
		return NULL;
	}

	return p_set;
}


as_set *
as_namespace_get_record_set(const as_namespace *ns, const as_record *r)
{
	return as_namespace_get_set_by_id(ns, as_index_get_set_id(r));
}


void
as_namespace_get_set_info(as_namespace *ns, const char *set_name,
		cf_dyn_buf *db)
{
	as_set *p_set;

	if (set_name) {
		if (cf_vmapx_get_by_name(ns->p_sets_vmap, set_name, (void**)&p_set) ==
				CF_VMAPX_OK) {
			append_set_props(p_set, db);
		}

		return;
	}

	for (uint32_t idx = 0; idx < cf_vmapx_count(ns->p_sets_vmap); idx++) {
		if (cf_vmapx_get_by_index(ns->p_sets_vmap, idx, (void**)&p_set) ==
				CF_VMAPX_OK) {
			cf_dyn_buf_append_string(db, "ns=");
			cf_dyn_buf_append_string(db, ns->name);
			cf_dyn_buf_append_char(db, ':');
			cf_dyn_buf_append_string(db, "set=");
			cf_dyn_buf_append_string(db, p_set->name);
			cf_dyn_buf_append_char(db, ':');
			append_set_props(p_set, db);
		}
	}
}


void
as_namespace_adjust_set_data_used_bytes(as_namespace *ns, uint16_t set_id,
		int64_t delta_bytes)
{
	if (set_id == INVALID_SET_ID) {
		return;
	}

	as_set *p_set;

	if (cf_vmapx_get_by_index(ns->p_sets_vmap, set_id - 1, (void**)&p_set) !=
			CF_VMAPX_OK) {
		cf_warning(AS_NAMESPACE, "set-id %u - failed vmap get", set_id);
		return;
	}

	uint64_t n_bytes = as_aaf_uint64(&p_set->data_used_bytes, delta_bytes);

	cf_assert((int64_t)n_bytes >= 0, AS_NAMESPACE,
			"{%s} set-id %u - negative n_bytes_device %ld (delta_bytes %ld)",
			ns->name, set_id, (int64_t)n_bytes, delta_bytes);
}


void
as_namespace_release_set_id(as_namespace *ns, uint16_t set_id)
{
	if (set_id == INVALID_SET_ID) {
		return;
	}

	as_set *p_set;

	if (cf_vmapx_get_by_index(ns->p_sets_vmap, set_id - 1, (void**)&p_set) !=
			CF_VMAPX_OK) {
		return;
	}

	uint64_t n_objects = as_aaf_uint64(&p_set->n_objects, -1);

	cf_assert(n_objects != (uint64_t)-1, AS_NAMESPACE,
			"{%s} set-id %u - n_objects underflow", ns->name, set_id);
}


// e.g. for ttl:
// units=seconds:hist-width=2582800:bucket-width=25828:buckets=0,0,0 ...
//
// e.g. for object-size-linear:
// units=bytes:hist-width=131072:bucket-width=128:buckets=16000,8000,0,0 ...
//
// e.g. for object-size:
// units=bytes:[64-128)=16000:[128-256)=8000 ...
void
as_namespace_get_hist_info(as_namespace *ns, char *set_name, char *hist_name,
		cf_dyn_buf *db)
{
	if (set_name[0] == '\0') {
		if (strcmp(hist_name, "ttl") == 0) {
			linear_hist_get_info(ns->ttl_hist, db);
		}
		else if (strcmp(hist_name, "object-size") == 0) {
			histogram_get_info(ns->obj_size_log_hist, db);
		}
		else if (strcmp(hist_name, "object-size-linear") == 0) {
			linear_hist_get_info(ns->obj_size_lin_hist, db);
		}
		else {
			as_info_respond_error(db, AS_ERR_PARAMETER, "unknown hist name");
		}

		return;
	}

	uint16_t set_id = as_namespace_get_set_id(ns, set_name);

	if (set_id == INVALID_SET_ID) {
		as_info_respond_error(db, AS_ERR_PARAMETER, "unknown set name");
		return;
	}

	if (strcmp(hist_name, "ttl") == 0) {
		if (ns->set_ttl_hists[set_id]) {
			linear_hist_get_info(ns->set_ttl_hists[set_id], db);
		}
		else {
			cf_dyn_buf_append_string(db, "hist-unavailable");
		}
	}
	else if (strcmp(hist_name, "object-size") == 0) {
		if (ns->set_obj_size_log_hists[set_id] != NULL) {
			histogram_get_info(ns->set_obj_size_log_hists[set_id], db);
		}
		else {
			cf_dyn_buf_append_string(db, "hist-unavailable");
		}
	}
	else if (strcmp(hist_name, "object-size-linear") == 0) {
		if (ns->set_obj_size_lin_hists[set_id] != NULL) {
			linear_hist_get_info(ns->set_obj_size_lin_hists[set_id], db);
		}
		else {
			cf_dyn_buf_append_string(db, "hist-unavailable");
		}
	}
	else {
		as_info_respond_error(db, AS_ERR_PARAMETER, "unknown hist name");
	}
}


//==========================================================
// Local helpers.
//

static void
append_set_props(as_set *p_set, cf_dyn_buf *db)
{
	// Statistics:

	cf_dyn_buf_append_string(db, "objects=");
	cf_dyn_buf_append_uint64(db, p_set->n_objects);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "tombstones=");
	cf_dyn_buf_append_uint64(db, p_set->n_tombstones);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "data_used_bytes=");
	cf_dyn_buf_append_uint64(db, p_set->data_used_bytes);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "truncate_lut=");
	cf_dyn_buf_append_uint64(db, p_set->truncate_lut);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "sindexes=");
	cf_dyn_buf_append_uint32(db, p_set->n_sindexes);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "index_populating=");
	cf_dyn_buf_append_bool(db, p_set->index_populating);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "truncating=");
	cf_dyn_buf_append_bool(db, p_set->truncating);
	cf_dyn_buf_append_char(db, ':');

	// Configuration:

	cf_dyn_buf_append_string(db, "default-read-touch-ttl-pct=");
	cf_dyn_buf_append_uint32(db, p_set->default_read_touch_ttl_pct);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "default-ttl=");
	cf_dyn_buf_append_uint32(db, p_set->default_ttl);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "disable-eviction=");
	cf_dyn_buf_append_bool(db, p_set->eviction_disabled);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "enable-index=");
	cf_dyn_buf_append_bool(db, p_set->index_enabled);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "stop-writes-count=");
	cf_dyn_buf_append_uint64(db, p_set->stop_writes_count);
	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "stop-writes-size=");
	cf_dyn_buf_append_uint64(db, p_set->stop_writes_size);
	cf_dyn_buf_append_char(db, ';');
}
