<commit_before>#include "core/global.h"
#include "core/msgtypes.h"
#include <unordered_set>

#include "DBStateServer.h"
#include "LoadingObject.h"

// RoleConfig
static ConfigVariable<channel_t> database_channel("database", INVALID_CHANNEL);

// RangesConfig
static ConfigVariable<doid_t> range_min("min", INVALID_DO_ID);
static ConfigVariable<doid_t> range_max("max", DOID_MAX);

DBStateServer::DBStateServer(RoleConfig roleconfig) : StateServer(roleconfig),
	m_db_channel(database_channel.get_rval(m_roleconfig)), m_next_context(0)
{
	RangesConfig ranges = roleconfig["ranges"];
	for(auto it = ranges.begin(); it != ranges.end(); ++it)
	{
		channel_t min = range_min.get_rval(*it);
		channel_t max = range_max.get_rval(*it);
		MessageDirector::singleton.subscribe_range(this, min, max);
	}

	std::stringstream name;
	name << "DBSS(Database: " << m_db_channel << ")";
	m_log = new LogCategory("dbss", name.str());
	set_con_name(name.str());
}

DBStateServer::~DBStateServer()
{
	delete m_log;
}

void DBStateServer::handle_activate(DatagramIterator &dgi, bool has_other)
{
	doid_t do_id = dgi.read_uint32();
	doid_t parent_id = dgi.read_uint32();
	zone_t zone_id = dgi.read_uint32();

	// Check object is not already active
	if(m_objs.find(do_id) != m_objs.end() || m_loading.find(do_id) != m_loading.end())
	{
		m_log->warning() << "Received activate for already-active object"
		                 << " - id:" << do_id << std::endl;
		return;
	}

	if(!has_other)
	{
		auto load_it = m_inactive_loads.find(do_id);
		if(load_it == m_inactive_loads.end())
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id);
		}
		else
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id, load_it->second);
		}
		m_loading[do_id]->begin();
	}
	else
	{
		uint16_t dc_id = dgi.read_uint16();

		// Check dclass is valid
		if(dc_id >= g_dcf->get_num_classes())
		{
			m_log->error() << "Received activate_other with unknown dclass"
			               << " - id:" << dc_id << std::endl;
			return;
		}

		DCClass *dclass = g_dcf->get_class(dc_id);
		auto load_it = m_inactive_loads.find(do_id);
		if(load_it == m_inactive_loads.end())
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id, dclass, dgi);
		}
		else
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id, dclass, dgi, load_it->second);
		}
		m_loading[do_id]->begin();
	}
}

void DBStateServer::handle_datagram(Datagram &in_dg, DatagramIterator &dgi)
{
	channel_t sender = dgi.read_uint64();
	uint16_t msgtype = dgi.read_uint16();
	switch(msgtype)
	{
		case DBSS_OBJECT_ACTIVATE_WITH_DEFAULTS:
		{
			handle_activate(dgi, false);
			break;
		}
		case DBSS_OBJECT_ACTIVATE_WITH_DEFAULTS_OTHER:
		{
			handle_activate(dgi, true);
			break;
		}
		case DBSS_OBJECT_DELETE_DISK:
		{
			doid_t do_id = dgi.read_uint32();

			// If object exists broadcast the delete message
			auto obj_keyval = m_objs.find(do_id);
			if(obj_keyval != m_objs.end())
			{
				DistributedObject* obj = obj_keyval->second;
				std::set<channel_t> targets;

				// Add location to broadcast
				if(obj->get_location())
				{
					targets.insert(obj->get_location());
				}

				// Add AI to broadcast
				if(obj->get_ai())
				{
					targets.insert(obj->get_ai());
				}

				// Add owner to broadcast
				if(obj->get_owner())
				{
					targets.insert(obj->get_owner());
				}

				// Build and send datagram
				Datagram dg(targets, sender, DBSS_OBJECT_DELETE_DISK);
				dg.add_doid(do_id);
				route_datagram(dg);
			}

			// Send delete to database
			Datagram dg(m_db_channel, do_id, DBSERVER_OBJECT_DELETE);
			dg.add_doid(do_id);
			route_datagram(dg);

			break;
		}
		case STATESERVER_OBJECT_SET_FIELD:
		{
			doid_t do_id = dgi.read_doid();
			uint16_t field_id = dgi.read_uint16();

			DCField* field = g_dcf->get_field_by_index(field_id);
			if(field && field->is_db())
			{
				m_log->trace() << "Forwarding SetField for field with id " << field_id
				              << ", on object " << do_id << " to database." << std::endl;

				Datagram dg(m_db_channel, do_id, DBSERVER_OBJECT_SET_FIELD);
				dg.add_doid(do_id);
				dg.add_uint16(field_id);
				dg.add_data(dgi.read_remainder());
				route_datagram(dg);
			}
			break;
		}
		case STATESERVER_OBJECT_SET_FIELDS:
		{
			doid_t do_id = dgi.read_doid();
			uint16_t field_count = dgi.read_uint16();

			std::unordered_map<DCField*, std::vector<uint8_t> > db_fields;
			for(uint16_t i = 0; i < field_count; ++i)
			{
				uint16_t field_id = dgi.read_uint16();
				DCField* field = g_dcf->get_field_by_index(field_id);
				if(!field)
				{
					m_log->warning() << "Received invalid field in SetFields"
					                 " with id " << field_id << std::endl;
					break;
				}
				if(field->is_db())
				{
					dgi.unpack_field(field, db_fields[field]);
				}
			}

			if(db_fields.size() > 0)
			{
				m_log->trace() << "Forwarding SetFields on object " << do_id << " to database." << std::endl;

				Datagram dg(m_db_channel, do_id, DBSERVER_OBJECT_SET_FIELDS);
				dg.add_doid(do_id);
				dg.add_uint16(field_count);
				for(auto it = db_fields.begin(); it != db_fields.end(); ++it)
				{
					dg.add_uint16(it->first->get_number());
					dg.add_data(it->second);
				}
				route_datagram(dg);
			}

			break;
		}
		case STATESERVER_OBJECT_GET_FIELD:
		{
			uint32_t r_context = dgi.read_uint32();
			doid_t r_do_id = dgi.read_doid();
			uint16_t field_id = dgi.read_uint16();

			// If object is active or loading, the Object or Loader will handle it
			if(m_objs.find(r_do_id) != m_objs.end() ||
			        m_loading.find(r_do_id) != m_loading.end())
			{
				break;
			}

			m_log->trace() << "Received GetField for field with id " << field_id
			              << " on inactive object with id " << r_do_id << std::endl;<commit_msg>DBSS: Fix broken SET_FIELDS behavior; properly skip non-database fields.<commit_after>#include "core/global.h"
#include "core/msgtypes.h"
#include <unordered_set>

#include "DBStateServer.h"
#include "LoadingObject.h"

// RoleConfig
static ConfigVariable<channel_t> database_channel("database", INVALID_CHANNEL);

// RangesConfig
static ConfigVariable<doid_t> range_min("min", INVALID_DO_ID);
static ConfigVariable<doid_t> range_max("max", DOID_MAX);

DBStateServer::DBStateServer(RoleConfig roleconfig) : StateServer(roleconfig),
	m_db_channel(database_channel.get_rval(m_roleconfig)), m_next_context(0)
{
	RangesConfig ranges = roleconfig["ranges"];
	for(auto it = ranges.begin(); it != ranges.end(); ++it)
	{
		channel_t min = range_min.get_rval(*it);
		channel_t max = range_max.get_rval(*it);
		MessageDirector::singleton.subscribe_range(this, min, max);
	}

	std::stringstream name;
	name << "DBSS(Database: " << m_db_channel << ")";
	m_log = new LogCategory("dbss", name.str());
	set_con_name(name.str());
}

DBStateServer::~DBStateServer()
{
	delete m_log;
}

void DBStateServer::handle_activate(DatagramIterator &dgi, bool has_other)
{
	doid_t do_id = dgi.read_uint32();
	doid_t parent_id = dgi.read_uint32();
	zone_t zone_id = dgi.read_uint32();

	// Check object is not already active
	if(m_objs.find(do_id) != m_objs.end() || m_loading.find(do_id) != m_loading.end())
	{
		m_log->warning() << "Received activate for already-active object"
		                 << " - id:" << do_id << std::endl;
		return;
	}

	if(!has_other)
	{
		auto load_it = m_inactive_loads.find(do_id);
		if(load_it == m_inactive_loads.end())
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id);
		}
		else
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id, load_it->second);
		}
		m_loading[do_id]->begin();
	}
	else
	{
		uint16_t dc_id = dgi.read_uint16();

		// Check dclass is valid
		if(dc_id >= g_dcf->get_num_classes())
		{
			m_log->error() << "Received activate_other with unknown dclass"
			               << " - id:" << dc_id << std::endl;
			return;
		}

		DCClass *dclass = g_dcf->get_class(dc_id);
		auto load_it = m_inactive_loads.find(do_id);
		if(load_it == m_inactive_loads.end())
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id, dclass, dgi);
		}
		else
		{
			m_loading[do_id] = new LoadingObject(this, do_id, parent_id, zone_id, dclass, dgi, load_it->second);
		}
		m_loading[do_id]->begin();
	}
}

void DBStateServer::handle_datagram(Datagram &in_dg, DatagramIterator &dgi)
{
	channel_t sender = dgi.read_uint64();
	uint16_t msgtype = dgi.read_uint16();
	switch(msgtype)
	{
		case DBSS_OBJECT_ACTIVATE_WITH_DEFAULTS:
		{
			handle_activate(dgi, false);
			break;
		}
		case DBSS_OBJECT_ACTIVATE_WITH_DEFAULTS_OTHER:
		{
			handle_activate(dgi, true);
			break;
		}
		case DBSS_OBJECT_DELETE_DISK:
		{
			doid_t do_id = dgi.read_uint32();

			// If object exists broadcast the delete message
			auto obj_keyval = m_objs.find(do_id);
			if(obj_keyval != m_objs.end())
			{
				DistributedObject* obj = obj_keyval->second;
				std::set<channel_t> targets;

				// Add location to broadcast
				if(obj->get_location())
				{
					targets.insert(obj->get_location());
				}

				// Add AI to broadcast
				if(obj->get_ai())
				{
					targets.insert(obj->get_ai());
				}

				// Add owner to broadcast
				if(obj->get_owner())
				{
					targets.insert(obj->get_owner());
				}

				// Build and send datagram
				Datagram dg(targets, sender, DBSS_OBJECT_DELETE_DISK);
				dg.add_doid(do_id);
				route_datagram(dg);
			}

			// Send delete to database
			Datagram dg(m_db_channel, do_id, DBSERVER_OBJECT_DELETE);
			dg.add_doid(do_id);
			route_datagram(dg);

			break;
		}
		case STATESERVER_OBJECT_SET_FIELD:
		{
			doid_t do_id = dgi.read_doid();
			uint16_t field_id = dgi.read_uint16();

			DCField* field = g_dcf->get_field_by_index(field_id);
			if(field && field->is_db())
			{
				m_log->trace() << "Forwarding SetField for field with id " << field_id
				              << ", on object " << do_id << " to database." << std::endl;

				Datagram dg(m_db_channel, do_id, DBSERVER_OBJECT_SET_FIELD);
				dg.add_doid(do_id);
				dg.add_uint16(field_id);
				dg.add_data(dgi.read_remainder());
				route_datagram(dg);
			}
			break;
		}
		case STATESERVER_OBJECT_SET_FIELDS:
		{
			doid_t do_id = dgi.read_doid();
			uint16_t field_count = dgi.read_uint16();

			std::unordered_map<DCField*, std::vector<uint8_t> > db_fields;
			for(uint16_t i = 0; i < field_count; ++i)
			{
				uint16_t field_id = dgi.read_uint16();
				DCField* field = g_dcf->get_field_by_index(field_id);
				if(!field)
				{
					m_log->warning() << "Received invalid field in SetFields"
					                 " with id " << field_id << std::endl;
					break;
				}
				if(field->is_db())
				{
					dgi.unpack_field(field, db_fields[field]);
				}
				else
				{
					dgi.skip_field(field);
				}
			}

			if(db_fields.size() > 0)
			{
				m_log->trace() << "Forwarding SetFields on object " << do_id << " to database." << std::endl;

				Datagram dg(m_db_channel, do_id, DBSERVER_OBJECT_SET_FIELDS);
				dg.add_doid(do_id);
				dg.add_uint16(db_fields.size());
				for(auto it = db_fields.begin(); it != db_fields.end(); ++it)
				{
					dg.add_uint16(it->first->get_number());
					dg.add_data(it->second);
				}
				route_datagram(dg);
			}

			break;
		}
		case STATESERVER_OBJECT_GET_FIELD:
		{
			uint32_t r_context = dgi.read_uint32();
			doid_t r_do_id = dgi.read_doid();
			uint16_t field_id = dgi.read_uint16();

			// If object is active or loading, the Object or Loader will handle it
			if(m_objs.find(r_do_id) != m_objs.end() ||
			        m_loading.find(r_do_id) != m_loading.end())
			{
				break;
			}

			m_log->trace() << "Received GetField for field with id " << field_id
			              << " on inactive object with id " << r_do_id << std::endl;
{'fraction_non_alphanumeric': 0.14256497656582873, 'fraction_numerical': 0.008521516829995739, 'mean_word_length': 3.6059654631083204, 'pattern_counts': {'":': 0, '<': 135, '<?xml version=': 0, '>': 73, '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': '1988272', 'n_tokens_mistral': 5301, 'n_tokens_neox': 4528, 'n_words': 789}