<commit_before>/*
**
** Copyright (c) 2014, Eneo Tecnologia
** Author: <NAME> <<EMAIL>>
** All rights reserved.
**
** This program is free software; you can redistribute it and/or modify
** 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 General Public License
** along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "rb_http2k_decoder.h"
#include "rb_mac.h"
#include "kafka.h"
#include "global_config.h"
#include "rb_json.h"
#include "topic_database.h"

#include <yajl/yajl_parse.h>
#include <yajl/yajl_gen.h>
#include <librd/rdlog.h>
#include <librd/rdmem.h>
#include <assert.h>
#include <jansson.h>
#include <stdint.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include <librd/rd.h>
#include <librdkafka/rdkafka.h>

static const char RB_HTTP2K_CONFIG_KEY[] = "rb_http2k_config";
static const char RB_SENSOR_UUID_ENRICHMENT_KEY[] = "uuids";
static const char RB_TOPICS_KEY[] = "topics";
static const char RB_SENSOR_UUID_KEY[] = "uuid";

static const char RB_TOPIC_PARTITIONER_KEY[] = "partition_key";
static const char RB_TOPIC_PARTITIONER_ALGORITHM_KEY[] = "partition_algo";

typedef int32_t (*partitioner_cb) (const rd_kafka_topic_t *rkt,
                                   const void *keydata, size_t keylen, int32_t partition_cnt,
                                   void *rkt_opaque, void *msg_opaque);

static int32_t (mac_partitioner) (const rd_kafka_topic_t *rkt,
                                  const void *keydata, size_t keylen, int32_t partition_cnt,
                                  void *rkt_opaque, void *msg_opaque);

/** Algorithm of messages partitioner */
enum partitioner_algorithm {
	/** Random partitioning */
	none,
	/** Mac partitioning */
	mac,
};

struct {
	enum partitioner_algorithm algoritm;
	const char *name;
	partitioner_cb partitioner;
} partitioner_algorithm_list[] = {
	{mac, "mac", mac_partitioner},
};

void init_rb_database(struct rb_database *db) {
	memset(db, 0, sizeof(*db));
	pthread_rwlock_init(&db->rwlock, 0);
}

#ifndef NDEBUG
#define RB_OPAQUE_MAGIC 0x0B0A3A1C0B0A3A1CL
#endif

struct rb_opaque {
#ifdef RB_OPAQUE_MAGIC
	uint64_t magic;
#endif

	struct rb_config *rb_config;
};

static int32_t mac_partitioner (const rd_kafka_topic_t *rkt,
                                const void *keydata, size_t keylen, int32_t partition_cnt,
                                void *rkt_opaque, void *msg_opaque) {
	size_t toks = 0;
	uint64_t intmac = 0;
	char mac_key[sizeof("00:00:00:00:00:00")];

	if (keylen != strlen("00:00:00:00:00:00")) {
		rdlog(LOG_WARNING, "Invalid mac %.*s len", (int)keylen, (const char *)keydata);
		goto fallback_behavior;
	}

	mac_key[0] = '\0';
	strncat(mac_key, (const char *)keydata, sizeof(mac_key) - 1);

	for (toks = 1; toks < 6; ++toks) {
		const size_t semicolon_pos = 3 * toks - 1;
		if (':' != mac_key[semicolon_pos]) {
			rdlog(LOG_WARNING, "Invalid mac %.*s (it does not have ':' in char %zu.",
			      (int)keylen, mac_key, semicolon_pos);
			goto fallback_behavior;
		}
	}

	for (toks = 0; toks < 6; ++toks) {
		char *endptr = NULL;
		intmac = (intmac << 8) + strtoul(&mac_key[3 * toks], &endptr, 16);
		/// The key should end with '"' in json format
		if ((toks < 5 && *endptr != ':') || (toks == 5 && *endptr != '\0')) {
			rdlog(LOG_WARNING, "Invalid mac %.*s, unexpected %c end of %zu token",
			      (int)keylen, mac_key, *endptr, toks);
			goto fallback_behavior;

		}

		if (endptr != mac_key + 3 * (toks + 1) - 1) {
			rdlog(LOG_WARNING, "Invalid mac %.*s, unexpected token length at %zu token",
			      (int)keylen, mac_key, toks);
			goto fallback_behavior;
		}
	}

	return intmac % (uint32_t)partition_cnt;

fallback_behavior:
	return rd_kafka_msg_partitioner_random(rkt, keydata, keylen, partition_cnt,
	                                       rkt_opaque, msg_opaque);
}

/** Parsing of per uuid enrichment.
	@param config original config with
	RB_SENSOR_UUID_ENRICHMENT_KEY to extract it.
	@param uuid_enrichment per-uuid enrichment object.
	@param dangerous_values Json hsahtable with values that could could belong
	to any client
	@return 0 if ok. Any other value should be checked against jerr.
	*/
static int parse_per_uuid_opaque_config(json_t *config,
	                json_t **uuid_enrichment,json_t **dangerous_values) {
	json_error_t jerr;
	const char *uuid_key = NULL,*uuid_enrichment_key = NULL;
	json_t *uuid_value = NULL,*uuid_enrichment_value = NULL;

	assert(config);
	assert(uuid_enrichment);
	assert(dangerous_values);

	const int json_unpack_rc = json_unpack_ex(config, &jerr, 0, "{s:O}",
	                           RB_SENSOR_UUID_ENRICHMENT_KEY, uuid_enrichment);

	if (0 != json_unpack_rc) {
		rdlog(LOG_ERR,"Couldn't unpack %s key: %s",
			RB_SENSOR_UUID_ENRICHMENT_KEY,jerr.text);
	}

	*dangerous_values = json_object();
	if(NULL == *dangerous_values) {
		rdlog(LOG_ERR,"Couldn't create dangerous values table (out of memory?)");
	}

	json_object_foreach(*uuid_enrichment,uuid_key,uuid_value) {
		if(!json_is_object(uuid_value)) {
			continue;
		}

		json_object_foreach(uuid_value,uuid_enrichment_key,
		                            uuid_enrichment_value) {
			if(NULL == json_object_get(*dangerous_values,uuid_enrichment_key)) {
				// New empty object.
				json_t *new_object = json_null();
				if(NULL == new_object) {
					rdlog(LOG_ERR,"Couldn't allocate new dangerous values object");
					continue;
				}

				json_object_set_new_nocheck(*dangerous_values,
					uuid_enrichment_key,new_object);
			}
		}
	}

	return json_unpack_rc;
}

static partitioner_cb partitioner_of_name(const char *name) {
	size_t i = 0;

	assert(name);

	for (i = 0; i < RD_ARRAYSIZE(partitioner_algorithm_list); ++i) {
		if (0 == strcmp(partitioner_algorithm_list[i].name, name)) {
			return partitioner_algorithm_list[i].partitioner;
		}
	}

	return NULL;
}

static int parse_topic_list_config(json_t *config, struct topics_db *new_topics_db) {
	const char *key;
	json_t *value;
	json_error_t jerr;
	json_t *topic_list = NULL;
	int pass = 0;

	assert(config);

	const int json_unpack_rc = json_unpack_ex(config, &jerr, 0, "{s:o}",
	                           RB_TOPICS_KEY, &topic_list);

	if (0 != json_unpack_rc) {
		rdlog(LOG_ERR, "%s", jerr.text);
		return json_unpack_rc;
	}

	if (!json_is_object(topic_list)) {
		rdlog(LOG_ERR, "%s is not an object", RB_TOPICS_KEY);
		return -1;
	}

	const size_t array_size = json_object_size(topic_list);
	if (0 == array_size) {
		rdlog(LOG_ERR, "%s has no childs", RB_TOPICS_KEY);
		return -1;
	}

	json_object_foreach(topic_list, key, value) {
		const char *partition_key = NULL, *partition_algo = NULL;
		size_t partition_key_len = 0;
		const char *topic_name = key;
		rd_kafka_topic_t *rkt = NULL;

		if (!json_is_object(value)) {
			if (pass == 0) {
				rdlog(LOG_ERR, "Topic %s is not an object. Discarding.", topic_name);
			}
			continue;
		}

		const int topic_unpack_rc = json_unpack_ex(value, &jerr, 0, "{s?s%,s?s}",
		                            RB_TOPIC_PARTITIONER_KEY, &partition_key, &partition_key_len,
		                            RB_TOPIC_PARTITIONER_ALGORITHM_KEY, &partition_algo);

		if (0 != topic_unpack_rc) {
			rdlog(LOG_ERR, "Can't extract information of topic %s (%s). Discarding.",
			      topic_name, jerr.text);
			continue;
		}

		rd_kafka_topic_conf_t *my_rkt_conf = rd_kafka_topic_conf_dup(
                                global_config.kafka_topic_conf);
		if(NULL == my_rkt_conf) {
			rdlog(LOG_ERR,"Couldn't topic_conf_dup in topic %s",topic_name);
			continue;
		}

		if (NULL != partition_algo) {
			partitioner_cb partitioner = partitioner_of_name(partition_algo);
			if (NULL != partitioner) {
				rd_kafka_topic_conf_set_partitioner_cb(my_rkt_conf, partitioner);
			} else {
				rdlog(LOG_ERR, 
					"Can't found partitioner algorithm %s for topic %s", 
					partition_algo,topic_name);
			}
		}

		rkt = rd_kafka_topic_new(global_config.rk, topic_name, my_rkt_conf);
		if (NULL == rkt) {
			char buf[BUFSIZ];
			strerror_r(errno, buf, sizeof(buf));
			rdlog(LOG_ERR, "Can't create topic %s: %s", topic_name, buf);
			rd_kafka_topic_conf_destroy(my_rkt_conf);
			continue;
		}

		topics_db_add(new_topics_db,rkt,partition_key,partition_key_len);
	}

	return 0;
}

int rb_opaque_creator(json_t *config __attribute__((unused)), void **_opaque) {
	assert(_opaque);

	struct rb_opaque *opaque = (*_opaque) = calloc(1, sizeof(*opaque));
	if (NULL == opaque) {
		rdlog(LOG_ERR, "Can't alloc RB_HTTP2K opaque (out of memory?)");
		return -1;
	}

#ifdef RB_OPAQUE_MAGIC
	opaque->magic = RB_OPAQUE_MAGIC;
#endif

	/// @TODO move global_config to static allocated buffer
	opaque->rb_config = &global_config.rb;

	return 0;
}

int rb_opaque_reload(json_t *config, void *_opaque) {
	int rc = 0;
	struct rb_opaque *opaque = _opaque;
	struct topics_db *old_topics_db = NULL, *my_new_topics_db = NULL;
	json_t *new_uuid_enrichment = NULL, *old_uuid_enrichment = NULL;
	json_t *new_dangerous_values_table = NULL,
		*old_dangerous_values_table = NULL;

	assert(opaque);
	assert(config);
#ifdef RB_OPAQUE_MAGIC
	assert(RB_OPAQUE_MAGIC == opaque->magic);
#endif

	const int per_sensor_uuid_enrichment_rc = parse_per_uuid_opaque_config(config,
	        &new_uuid_enrichment,&new_dangerous_values_table);
	if (per_sensor_uuid_enrichment_rc != 0 || NULL == new_uuid_enrichment) {
		rc = -1;
		goto err;
	}

	my_new_topics_db = topics_db_new();

	const int topic_list_rc = parse_topic_list_config(config,my_new_topics_db);
	if (topic_list_rc != 0) {
		rc = -1;
		goto err;
	}

	pthread_rwlock_wrlock(&opaque->rb_config->database.rwlock);

	old_uuid_enrichment = opaque->rb_config->database.uuid_enrichment;
	opaque->rb_config->database.uuid_enrichment = new_uuid_enrichment;
	new_uuid_enrichment = NULL;

	old_dangerous_values_table = opaque->rb_config->database.dangerous_values;
	opaque->rb_config->database.dangerous_values = new_dangerous_values_table;
	new_dangerous_values_table = NULL;

	old_topics_db = opaque->rb_config->database.topics_db;
	opaque->rb_config->database.topics_db = my_new_topics_db;
	my_new_topics_db = NULL;

	pthread_rwlock_unlock(&opaque->rb_config->database.rwlock);

err:
	if (my_new_topics_db) {
		topics_db_done(my_new_topics_db);
	}

	if (old_topics_db) {
		topics_db_done(old_topics_db);
	}

	if (old_uuid_enrichment) {
		json_decref(old_uuid_enrichment);
	}

	if (new_uuid_enrichment) {
		json_decref(new_uuid_enrichment);
	}

	if (new_dangerous_values_table) {
		json_decref(new_dangerous_values_table);
	}

	if (old_dangerous_values_table) {
		json_decref(old_dangerous_values_table);
	}

	return rc;
}

void rb_opaque_done(void *_opaque) {
	assert(_opaque);

	struct rb_opaque *opaque = _opaque;
#ifdef RB_OPAQUE_MAGIC
	assert(RB_OPAQUE_MAGIC == opaque->magic);
#endif

	free(opaque);
}

int parse_rb_config(void *void_db, const struct json_t *config) {
	assert(void_db);
	char errbuf[BUFSIZ];
	struct rb_config *rb_config = void_db;

	struct rb_opaque dummy_opaque = {
#ifdef RB_OPAQUE_MAGIC
		.magic = RB_OPAQUE_MAGIC,<commit_msg>Deleting dangerous values, not valid anymore<commit_after>/*
**
** Copyright (c) 2014, Eneo Tecnologia
** Author: <NAME> <<EMAIL>>
** All rights reserved.
**
** This program is free software; you can redistribute it and/or modify
** 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 General Public License
** along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "rb_http2k_decoder.h"
#include "rb_mac.h"
#include "kafka.h"
#include "global_config.h"
#include "rb_json.h"
#include "topic_database.h"

#include <yajl/yajl_parse.h>
#include <yajl/yajl_gen.h>
#include <librd/rdlog.h>
#include <librd/rdmem.h>
#include <assert.h>
#include <jansson.h>
#include <stdint.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include <librd/rd.h>
#include <librdkafka/rdkafka.h>

static const char RB_HTTP2K_CONFIG_KEY[] = "<KEY>";
static const char RB_SENSOR_UUID_ENRICHMENT_KEY[] = "uuids";
static const char RB_TOPICS_KEY[] = "topics";
static const char RB_SENSOR_UUID_KEY[] = "uuid";

static const char RB_TOPIC_PARTITIONER_KEY[] = "partition_key";
static const char RB_TOPIC_PARTITIONER_ALGORITHM_KEY[] = "partition_algo";

typedef int32_t (*partitioner_cb) (const rd_kafka_topic_t *rkt,
                                   const void *keydata, size_t keylen, int32_t partition_cnt,
                                   void *rkt_opaque, void *msg_opaque);

static int32_t (mac_partitioner) (const rd_kafka_topic_t *rkt,
                                  const void *keydata, size_t keylen, int32_t partition_cnt,
                                  void *rkt_opaque, void *msg_opaque);

/** Algorithm of messages partitioner */
enum partitioner_algorithm {
	/** Random partitioning */
	none,
	/** Mac partitioning */
	mac,
};

struct {
	enum partitioner_algorithm algoritm;
	const char *name;
	partitioner_cb partitioner;
} partitioner_algorithm_list[] = {
	{mac, "mac", mac_partitioner},
};

void init_rb_database(struct rb_database *db) {
	memset(db, 0, sizeof(*db));
	pthread_rwlock_init(&db->rwlock, 0);
}

#ifndef NDEBUG
#define RB_OPAQUE_MAGIC 0x0B0A3A1C0B0A3A1CL
#endif

struct rb_opaque {
#ifdef RB_OPAQUE_MAGIC
	uint64_t magic;
#endif

	struct rb_config *rb_config;
};

static int32_t mac_partitioner (const rd_kafka_topic_t *rkt,
                                const void *keydata, size_t keylen, int32_t partition_cnt,
                                void *rkt_opaque, void *msg_opaque) {
	size_t toks = 0;
	uint64_t intmac = 0;
	char mac_key[sizeof("00:00:00:00:00:00")];

	if (keylen != strlen("00:00:00:00:00:00")) {
		rdlog(LOG_WARNING, "Invalid mac %.*s len", (int)keylen, (const char *)keydata);
		goto fallback_behavior;
	}

	mac_key[0] = '\0';
	strncat(mac_key, (const char *)keydata, sizeof(mac_key) - 1);

	for (toks = 1; toks < 6; ++toks) {
		const size_t semicolon_pos = 3 * toks - 1;
		if (':' != mac_key[semicolon_pos]) {
			rdlog(LOG_WARNING, "Invalid mac %.*s (it does not have ':' in char %zu.",
			      (int)keylen, mac_key, semicolon_pos);
			goto fallback_behavior;
		}
	}

	for (toks = 0; toks < 6; ++toks) {
		char *endptr = NULL;
		intmac = (intmac << 8) + strtoul(&mac_key[3 * toks], &endptr, 16);
		/// The key should end with '"' in json format
		if ((toks < 5 && *endptr != ':') || (toks == 5 && *endptr != '\0')) {
			rdlog(LOG_WARNING, "Invalid mac %.*s, unexpected %c end of %zu token",
			      (int)keylen, mac_key, *endptr, toks);
			goto fallback_behavior;

		}

		if (endptr != mac_key + 3 * (toks + 1) - 1) {
			rdlog(LOG_WARNING, "Invalid mac %.*s, unexpected token length at %zu token",
			      (int)keylen, mac_key, toks);
			goto fallback_behavior;
		}
	}

	return intmac % (uint32_t)partition_cnt;

fallback_behavior:
	return rd_kafka_msg_partitioner_random(rkt, keydata, keylen, partition_cnt,
	                                       rkt_opaque, msg_opaque);
}

/** Parsing of per uuid enrichment.
	@param config original config with
	RB_SENSOR_UUID_ENRICHMENT_KEY to extract it.
	@param uuid_enrichment per-uuid enrichment object.
	@param dangerous_values Json hsahtable with values that could could belong
	to any client
	@return 0 if ok. Any other value should be checked against jerr.
	*/
static int parse_per_uuid_opaque_config(json_t *config,
	                json_t **uuid_enrichment) {
	json_error_t jerr;

	assert(config);
	assert(uuid_enrichment);

	const int json_unpack_rc = json_unpack_ex(config, &jerr, 0, "{s:O}",
	                           RB_SENSOR_UUID_ENRICHMENT_KEY, uuid_enrichment);

	if (0 != json_unpack_rc) {
		rdlog(LOG_ERR,"Couldn't unpack %s key: %s",
			RB_SENSOR_UUID_ENRICHMENT_KEY,jerr.text);
	}

	return json_unpack_rc;
}

static partitioner_cb partitioner_of_name(const char *name) {
	size_t i = 0;

	assert(name);

	for (i = 0; i < RD_ARRAYSIZE(partitioner_algorithm_list); ++i) {
		if (0 == strcmp(partitioner_algorithm_list[i].name, name)) {
			return partitioner_algorithm_list[i].partitioner;
		}
	}

	return NULL;
}

static int parse_topic_list_config(json_t *config, struct topics_db *new_topics_db) {
	const char *key;
	json_t *value;
	json_error_t jerr;
	json_t *topic_list = NULL;
	int pass = 0;

	assert(config);

	const int json_unpack_rc = json_unpack_ex(config, &jerr, 0, "{s:o}",
	                           RB_TOPICS_KEY, &topic_list);

	if (0 != json_unpack_rc) {
		rdlog(LOG_ERR, "%s", jerr.text);
		return json_unpack_rc;
	}

	if (!json_is_object(topic_list)) {
		rdlog(LOG_ERR, "%s is not an object", RB_TOPICS_KEY);
		return -1;
	}

	const size_t array_size = json_object_size(topic_list);
	if (0 == array_size) {
		rdlog(LOG_ERR, "%s has no childs", RB_TOPICS_KEY);
		return -1;
	}

	json_object_foreach(topic_list, key, value) {
		const char *partition_key = NULL, *partition_algo = NULL;
		size_t partition_key_len = 0;
		const char *topic_name = key;
		rd_kafka_topic_t *rkt = NULL;

		if (!json_is_object(value)) {
			if (pass == 0) {
				rdlog(LOG_ERR, "Topic %s is not an object. Discarding.", topic_name);
			}
			continue;
		}

		const int topic_unpack_rc = json_unpack_ex(value, &jerr, 0, "{s?s%,s?s}",
		                            RB_TOPIC_PARTITIONER_KEY, &partition_key, &partition_key_len,
		                            RB_TOPIC_PARTITIONER_ALGORITHM_KEY, &partition_algo);

		if (0 != topic_unpack_rc) {
			rdlog(LOG_ERR, "Can't extract information of topic %s (%s). Discarding.",
			      topic_name, jerr.text);
			continue;
		}

		rd_kafka_topic_conf_t *my_rkt_conf = rd_kafka_topic_conf_dup(
                                global_config.kafka_topic_conf);
		if(NULL == my_rkt_conf) {
			rdlog(LOG_ERR,"Couldn't topic_conf_dup in topic %s",topic_name);
			continue;
		}

		if (NULL != partition_algo) {
			partitioner_cb partitioner = partitioner_of_name(partition_algo);
			if (NULL != partitioner) {
				rd_kafka_topic_conf_set_partitioner_cb(my_rkt_conf, partitioner);
			} else {
				rdlog(LOG_ERR, 
					"Can't found partitioner algorithm %s for topic %s", 
					partition_algo,topic_name);
			}
		}

		rkt = rd_kafka_topic_new(global_config.rk, topic_name, my_rkt_conf);
		if (NULL == rkt) {
			char buf[BUFSIZ];
			strerror_r(errno, buf, sizeof(buf));
			rdlog(LOG_ERR, "Can't create topic %s: %s", topic_name, buf);
			rd_kafka_topic_conf_destroy(my_rkt_conf);
			continue;
		}

		topics_db_add(new_topics_db,rkt,partition_key,partition_key_len);
	}

	return 0;
}

int rb_opaque_creator(json_t *config __attribute__((unused)), void **_opaque) {
	assert(_opaque);

	struct rb_opaque *opaque = (*_opaque) = calloc(1, sizeof(*opaque));
	if (NULL == opaque) {
		rdlog(LOG_ERR, "Can't alloc RB_HTTP2K opaque (out of memory?)");
		return -1;
	}

#ifdef RB_OPAQUE_MAGIC
	opaque->magic = RB_OPAQUE_MAGIC;
#endif

	/// @TODO move global_config to static allocated buffer
	opaque->rb_config = &global_config.rb;

	return 0;
}

int rb_opaque_reload(json_t *config, void *_opaque) {
	int rc = 0;
	struct rb_opaque *opaque = _opaque;
	struct topics_db *old_topics_db = NULL, *my_new_topics_db = NULL;
	json_t *new_uuid_enrichment = NULL, *old_uuid_enrichment = NULL;

	assert(opaque);
	assert(config);
#ifdef RB_OPAQUE_MAGIC
	assert(RB_OPAQUE_MAGIC == opaque->magic);
#endif

	const int per_sensor_uuid_enrichment_rc = parse_per_uuid_opaque_config(config,
	        &new_uuid_enrichment);
	if (per_sensor_uuid_enrichment_rc != 0 || NULL == new_uuid_enrichment) {
		rc = -1;
		goto err;
	}

	my_new_topics_db = topics_db_new();

	const int topic_list_rc = parse_topic_list_config(config,my_new_topics_db);
	if (topic_list_rc != 0) {
		rc = -1;
		goto err;
	}

	pthread_rwlock_wrlock(&opaque->rb_config->database.rwlock);

	old_uuid_enrichment = opaque->rb_config->database.uuid_enrichment;
	opaque->rb_config->database.uuid_enrichment = new_uuid_enrichment;
	new_uuid_enrichment = NULL;

	old_topics_db = opaque->rb_config->database.topics_db;
	opaque->rb_config->database.topics_db = my_new_topics_db;
	my_new_topics_db = NULL;

	pthread_rwlock_unlock(&opaque->rb_config->database.rwlock);

err:
	if (my_new_topics_db) {
		topics_db_done(my_new_topics_db);
	}

	if (old_topics_db) {
		topics_db_done(old_topics_db);
	}

	if (old_uuid_enrichment) {
		json_decref(old_uuid_enrichment);
	}

	if (new_uuid_enrichment) {
		json_decref(new_uuid_enrichment);
	}

	return rc;
}

void rb_opaque_done(void *_opaque) {
	assert(_opaque);

	struct rb_opaque *opaque = _opaque;
#ifdef RB_OPAQUE_MAGIC
	assert(RB_OPAQUE_MAGIC == opaque->magic);
#endif

	free(opaque);
}

int parse_rb_config(void *void_db, const struct json_t *config) {
	assert(void_db);
	char errbuf[BUFSIZ];
	struct rb_config *rb_config = void_db;

	struct rb_opaque dummy_opaque = {
#ifdef RB_OPAQUE_MAGIC
		.magic = RB_OPAQUE_MAGIC,
{'fraction_non_alphanumeric': 0.1166312678917721, 'fraction_numerical': 0.009927047742173793, 'mean_word_length': 4.068804118886028, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 74, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7566027', 'n_tokens_mistral': 8818, 'n_tokens_neox': 8135, 'n_words': 1820}