/*
 * delete.c
 *
 * Copyright (C) 2016-2021 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 "transaction/delete.h"

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

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

#include "cf_mutex.h"
#include "dynbuf.h"
#include "log.h"

#include "base/batch.h"
#include "base/cfg.h"
#include "base/datamodel.h"
#include "base/exp.h"
#include "base/index.h"
#include "base/mrt_monitor.h"
#include "base/proto.h"
#include "base/set_index.h"
#include "base/transaction.h"
#include "base/transaction_policy.h"
#include "fabric/fabric.h"
#include "fabric/partition.h"
#include "sindex/sindex.h"
#include "storage/storage.h"
#include "transaction/duplicate_resolve.h"
#include "transaction/mrt_utils.h"
#include "transaction/proxy.h"
#include "transaction/replica_write.h"
#include "transaction/rw_request.h"
#include "transaction/rw_request_hash.h"
#include "transaction/rw_utils.h"


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

static void delete_dup_res_start_cb(rw_request* rw, as_transaction* tr, as_record* r);
static void start_delete_repl_write(rw_request* rw, as_transaction* tr);
static void start_delete_repl_write_forget(rw_request* rw, as_transaction* tr);
static bool delete_dup_res_cb(rw_request* rw);
static void delete_repl_write_after_dup_res(rw_request* rw, as_transaction* tr);
static void delete_repl_write_forget_after_dup_res(rw_request* rw, as_transaction* tr);
static void delete_repl_write_cb(rw_request* rw);

static void send_delete_response(as_transaction* tr);
static void delete_timeout_cb(rw_request* rw);


//==========================================================
// Inlines & macros.
//

static inline void
client_delete_update_stats(as_namespace* ns, uint8_t result_code,
		bool is_xdr_op)
{
	switch (result_code) {
	case AS_OK:
		as_incr_uint64(&ns->n_client_delete_success);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_client_delete_success);
		}
		break;
	default:
		as_incr_uint64(&ns->n_client_delete_error);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_client_delete_error);
		}
		break;
	case AS_ERR_TIMEOUT:
		as_incr_uint64(&ns->n_client_delete_timeout);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_client_delete_timeout);
		}
		break;
	case AS_ERR_NOT_FOUND:
		as_incr_uint64(&ns->n_client_delete_not_found);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_client_delete_not_found);
		}
		break;
	case AS_ERR_FILTERED_OUT:
		// Can't be an XDR delete.
		as_incr_uint64(&ns->n_client_delete_filtered_out);
		break;
	}
}

static inline void
from_proxy_delete_update_stats(as_namespace* ns, uint8_t result_code,
		bool is_xdr_op)
{
	switch (result_code) {
	case AS_OK:
		as_incr_uint64(&ns->n_from_proxy_delete_success);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_from_proxy_delete_success);
		}
		break;
	default:
		as_incr_uint64(&ns->n_from_proxy_delete_error);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_from_proxy_delete_error);
		}
		break;
	case AS_ERR_TIMEOUT:
		as_incr_uint64(&ns->n_from_proxy_delete_timeout);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_from_proxy_delete_timeout);
		}
		break;
	case AS_ERR_NOT_FOUND:
		as_incr_uint64(&ns->n_from_proxy_delete_not_found);
		if (is_xdr_op) {
			as_incr_uint64(&ns->n_xdr_from_proxy_delete_not_found);
		}
		break;
	case AS_ERR_FILTERED_OUT:
		// Can't be an XDR delete.
		as_incr_uint64(&ns->n_from_proxy_delete_filtered_out);
		break;
	}
}

// Can't be an XDR delete.
static inline void
batch_sub_delete_update_stats(as_namespace* ns, uint8_t result_code)
{
	switch (result_code) {
	case AS_OK:
		as_incr_uint64(&ns->n_batch_sub_delete_success);
		break;
	default:
		as_incr_uint64(&ns->n_batch_sub_delete_error);
		break;
	case AS_ERR_TIMEOUT:
		as_incr_uint64(&ns->n_batch_sub_delete_timeout);
		break;
	case AS_ERR_NOT_FOUND:
		as_incr_uint64(&ns->n_batch_sub_delete_not_found);
		break;
	case AS_ERR_FILTERED_OUT:
		as_incr_uint64(&ns->n_batch_sub_delete_filtered_out);
		break;
	}
}

// Can't be an XDR delete.
static inline void
from_proxy_batch_sub_delete_update_stats(as_namespace* ns, uint8_t result_code)
{
	switch (result_code) {
	case AS_OK:
		as_incr_uint64(&ns->n_from_proxy_batch_sub_delete_success);
		break;
	default:
		as_incr_uint64(&ns->n_from_proxy_batch_sub_delete_error);
		break;
	case AS_ERR_TIMEOUT:
		as_incr_uint64(&ns->n_from_proxy_batch_sub_delete_timeout);
		break;
	case AS_ERR_NOT_FOUND:
		as_incr_uint64(&ns->n_from_proxy_batch_sub_delete_not_found);
		break;
	case AS_ERR_FILTERED_OUT:
		as_incr_uint64(&ns->n_from_proxy_batch_sub_delete_filtered_out);
		break;
	}
}


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

transaction_status
as_delete_start(as_transaction* tr)
{
	// Apply XDR filter.
	if (! xdr_allows_write(tr)) {
		tr->result_code = AS_ERR_FORBIDDEN;
		send_delete_response(tr);
		return TRANS_DONE;
	}

	int result = validate_delete_durability(tr);

	if (result != AS_OK) {
		tr->result_code = result;
		send_delete_response(tr);
		return TRANS_DONE;
	}

	if (delete_storage_overloaded(tr)) {
		tr->result_code = AS_ERR_DEVICE_OVERLOAD;
		send_delete_response(tr);
		return TRANS_DONE;
	}

	// Create rw_request and add to hash.
	rw_request_hkey hkey = { tr->rsv.ns->ix, tr->keyd };
	rw_request* rw = rw_request_create(&tr->keyd);
	transaction_status status = rw_request_hash_insert(&hkey, rw, tr);

	// If rw_request wasn't inserted in hash, transaction is finished.
	if (status != TRANS_IN_PROGRESS) {
		rw_request_release(rw);

		if (status != TRANS_WAITING) {
			send_delete_response(tr);
		}

		return status;
	}
	// else - rw_request is now in hash, continue...

	if (tr->rsv.ns->write_dup_res_disabled) {
		// Note - preventing duplicate resolution this way allows
		// rw_request_destroy() to handle dup_msg[] cleanup correctly.
		tr->rsv.n_dupl = 0;
	}

	// If there are duplicates to resolve, start doing so.
	// TODO - should we bother if there's no generation check?
	if (tr->rsv.n_dupl != 0 && dup_res_start(rw, tr, delete_dup_res_start_cb)) {
		return TRANS_IN_PROGRESS; // started duplicate resolution
	}
	// else - no duplicate resolution phase, apply operation to master.

	// If error, transaction is finished.
	if ((status = delete_master(tr, rw)) != TRANS_IN_PROGRESS) {
		rw_request_hash_delete(&hkey, rw);

		if (status != TRANS_WAITING) {
			send_delete_response(tr);
		}

		return status;
	}

	// If we don't need replica writes, transaction is finished.
	if (rw->n_dest_nodes == 0) {
		finished_replicated(tr);
		rw_request_hash_delete(&hkey, rw);
		send_delete_response(tr);
		return TRANS_DONE;
	}

	// If we don't need to wait for replica write acks, fire and forget.
	if (respond_on_master_complete(tr)) {
		start_delete_repl_write_forget(rw, tr);
		rw_request_hash_delete(&hkey, rw);
		send_delete_response(tr);
		return TRANS_DONE;
	}

	start_delete_repl_write(rw, tr);

	// Started replica write.
	return TRANS_IN_PROGRESS;
}


//==========================================================
// Local helpers - transaction flow.
//

static void
delete_dup_res_start_cb(rw_request* rw, as_transaction* tr, as_record* r)
{
	// Finish initializing rw, construct and send dup-res message.

	dup_res_make_message(rw, tr, r);

	cf_mutex_lock(&rw->lock);

	dup_res_setup_rw(rw, tr, delete_dup_res_cb, delete_timeout_cb);
	send_rw_messages(rw);

	cf_mutex_unlock(&rw->lock);
}

static void
start_delete_repl_write(rw_request* rw, as_transaction* tr)
{
	// Finish initializing rw, construct and send repl-delete message.

	repl_write_make_message(rw, tr);

	cf_mutex_lock(&rw->lock);

	repl_write_setup_rw(rw, tr, delete_repl_write_cb, delete_timeout_cb);
	send_rw_messages(rw);

	cf_mutex_unlock(&rw->lock);
}

static void
start_delete_repl_write_forget(rw_request* rw, as_transaction* tr)
{
	// Construct and send repl-write message. No need to finish rw setup.

	repl_write_make_message(rw, tr);
	send_rw_messages_forget(rw);
}

static bool
delete_dup_res_cb(rw_request* rw)
{
	as_transaction tr;
	as_transaction_init_from_rw(&tr, rw);

	if (tr.result_code != AS_OK) {
		send_delete_response(&tr);
		return true;
	}

	transaction_status status = delete_master(&tr, rw);

	if (status == TRANS_WAITING) {
		// Note - new tr now owns msgp, make sure rw destructor doesn't free it.
		// Also, rw will release rsv - new tr will get a new one.
		rw->msgp = NULL;
		return true;
	}

	if (status == TRANS_DONE) {
		send_delete_response(&tr);
		return true;
	}

	// If we don't need replica writes, transaction is finished.
	if (rw->n_dest_nodes == 0) {
		finished_replicated(&tr);
		send_delete_response(&tr);
		return true;
	}

	// If we don't need to wait for replica write acks, fire and forget.
	// (Remember that nsup deletes can't get here, so no need to check.)
	if (respond_on_master_complete(&tr)) {
		delete_repl_write_forget_after_dup_res(rw, &tr);
		send_delete_response(&tr);
		return true;
	}

	delete_repl_write_after_dup_res(rw, &tr);

	// Started replica write - don't delete rw_request from hash.
	return false;
}

static void
delete_repl_write_after_dup_res(rw_request* rw, as_transaction* tr)
{
	// Recycle rw_request that was just used for duplicate resolution to now do
	// replica writes. Note - we are under the rw_request lock here!

	repl_write_make_message(rw, tr);
	repl_write_reset_rw(rw, tr, delete_repl_write_cb);
	send_rw_messages(rw);
}

static void
delete_repl_write_forget_after_dup_res(rw_request* rw, as_transaction* tr)
{
	// Send replica writes. Not waiting for acks, so need to reset rw_request.
	// Note - we are under the rw_request lock here!

	repl_write_make_message(rw, tr);
	send_rw_messages_forget(rw);
}

static void
delete_repl_write_cb(rw_request* rw)
{
	as_transaction tr;
	as_transaction_init_from_rw(&tr, rw);

	finished_replicated(&tr);
	send_delete_response(&tr);

	// Finished transaction - rw_request cleans up reservation and msgp!
}


//==========================================================
// Local helpers - transaction end.
//

static void
send_delete_response(as_transaction* tr)
{
	// Paranoia - shouldn't get here on losing race with timeout.
	if (! tr->from.any) {
		cf_warning(AS_RW, "transaction origin %u has null 'from'", tr->origin);
		return;
	}

	// Note - if tr was setup from rw, rw->from.any has been set null and
	// informs timeout it lost the race.

	as_record_version v;

	switch (tr->origin) {
	case FROM_CLIENT:
		as_msg_send_reply(tr->from.proto_fd_h, tr->result_code, 0, 0, NULL,
				NULL, 0, tr->rsv.ns, mrt_write_fill_version(&v, tr));
		client_delete_update_stats(tr->rsv.ns, tr->result_code,
				as_transaction_is_xdr(tr));
		break;
	case FROM_PROXY:
		as_proxy_send_response(tr->from.proxy_node, tr->from_data.proxy_tid,
				tr->result_code, 0, 0, NULL, NULL, 0, tr->rsv.ns,
				mrt_write_fill_version(&v, tr));
		if (as_transaction_is_batch_sub(tr)) {
			from_proxy_batch_sub_delete_update_stats(tr->rsv.ns,
					tr->result_code);
		}
		else {
			from_proxy_delete_update_stats(tr->rsv.ns, tr->result_code,
					as_transaction_is_xdr(tr));
		}
		break;
	case FROM_BATCH:
		// TODO - maybe future generic add_ack() can take these as params?
		tr->generation = 0;
		tr->void_time = 0;
		as_batch_add_ack(tr, mrt_write_fill_version(&v, tr));
		batch_sub_delete_update_stats(tr->rsv.ns, tr->result_code);
		break;
	case FROM_MONITOR_DELETE:
		// Nothing needed.
		break;
	default:
		cf_crash(AS_RW, "unexpected transaction origin %u", tr->origin);
		break;
	}

	tr->from.any = NULL; // pattern, not needed
}

static void
delete_timeout_cb(rw_request* rw)
{
	if (! rw->from.any) {
		return; // lost race against dup-res or repl-write callback
	}

	finished_not_replicated(rw);

	switch (rw->origin) {
	case FROM_CLIENT:
		as_msg_send_reply(rw->from.proto_fd_h, AS_ERR_TIMEOUT, 0, 0, NULL, NULL,
				0, rw->rsv.ns, NULL);
		client_delete_update_stats(rw->rsv.ns, AS_ERR_TIMEOUT,
				as_msg_is_xdr(&rw->msgp->msg));
		break;
	case FROM_PROXY:
		if (rw_request_is_batch_sub(rw)) {
			from_proxy_batch_sub_delete_update_stats(rw->rsv.ns,
					AS_ERR_TIMEOUT);
		}
		else {
			from_proxy_delete_update_stats(rw->rsv.ns, AS_ERR_TIMEOUT,
					as_msg_is_xdr(&rw->msgp->msg));
		}
		break;
	case FROM_BATCH:
		as_batch_add_error(rw->from.batch_shared, rw->from_data.batch_index,
				AS_ERR_TIMEOUT);
		// Timeouts aren't included in histograms.
		batch_sub_delete_update_stats(rw->rsv.ns, AS_ERR_TIMEOUT);
		break;
	case FROM_MONITOR_DELETE:
		// Nothing needed.
		break;
	default:
		cf_crash(AS_RW, "unexpected transaction origin %u", rw->origin);
		break;
	}

	rw->from.any = NULL; // inform other callback it lost the race
}


//==========================================================
// Local helpers - delete master.
//

// Not static - called by split.
transaction_status
drop_master(as_transaction* tr, as_index_ref* r_ref, rw_request* rw)
{
	as_msg* m = &tr->msgp->msg;
	as_namespace* ns = tr->rsv.ns;
	as_index_tree* tree = tr->rsv.tree;
	as_record* r = r_ref->r;

	// Check generation requirement, if any.
	if (! generation_check(r, m, ns)) {
		as_record_done(r_ref, ns);
		as_incr_uint64(&ns->n_fail_generation);
		tr->result_code = AS_ERR_GENERATION;
		return TRANS_DONE;
	}

	// Apply predexp metadata filter if present.

	as_exp* filter_exp = NULL;
	int result = handle_meta_filter(tr, r, &filter_exp);

	if (result != 0) {
		as_record_done(r_ref, ns);
		tr->result_code = result;
		return TRANS_DONE;
	}

	// Set up the nodes to which we'll write replicas.
	if (! set_replica_destinations(tr, rw)) {
		as_record_done(r_ref, ns);
		tr->result_code = AS_ERR_UNAVAILABLE;
		return TRANS_DONE;
	}

	// Fire and forget can overload the fabric send queues - check.
	if (respond_on_master_complete(tr) &&
			as_fabric_is_overloaded(rw->dest_nodes, rw->n_dest_nodes,
					AS_FABRIC_CHANNEL_RW, 8)) {
		tr->flags |= AS_TRANSACTION_FLAG_SWITCH_TO_COMMIT_ALL;
	}

	bool check_key = as_transaction_has_key(tr);

	if (filter_exp != NULL || check_key) {
		as_storage_rd rd;
		as_storage_record_open(ns, r, &rd);

		// Apply predexp record bins filter if present.
		if (filter_exp != NULL) {
			if ((result = read_and_filter_bins(&rd, filter_exp)) != 0) {
				destroy_filter_exp(tr, filter_exp);
				as_storage_record_close(&rd);
				as_record_done(r_ref, ns);
				tr->result_code = result;
				return TRANS_DONE;
			}

			destroy_filter_exp(tr, filter_exp);
		}

		// Check the key if required.
		// Note - for data-not-in-memory a key check is expensive!
		if (check_key && as_storage_rd_load_key(&rd) &&
				! check_msg_key(m, &rd)) {
			as_storage_record_close(&rd);
			as_record_done(r_ref, ns);
			tr->result_code = AS_ERR_KEY_MISMATCH;
			return TRANS_DONE;
		}

		as_storage_record_close(&rd);
	}

	if (ns->storage_type != AS_STORAGE_ENGINE_SSD ||
			ns->pi_xmem_type == CF_XMEM_TYPE_FLASH) {
		remove_from_sindex(ns, r_ref);
	}

	as_set_index_delete(ns, tree, as_index_get_set_id(r), r_ref->r_h);
	as_index_delete(tree, &tr->keyd);
	as_record_done(r_ref, ns);

	return TRANS_IN_PROGRESS;
}
