ud3tn/components/cla/sqlite/cla_sqlite.c

// SPDX-License-Identifier: BSD-3-Clause OR Apache-2.0
#include "agents/sqlite_agent.h"

#include "cla/cla.h"
#include "cla/cla_contact_tx_task.h"
#include "cla/posix/cla_sqlite.h"

#include "platform/hal_io.h"
#include "platform/hal_queue.h"
#include "platform/hal_semaphore.h"
#include "platform/hal_task.h"
#include "platform/hal_types.h"
#include "platform/posix/hal_types.h"

#include "ud3tn/eid.h"
#include "ud3tn/result.h"
#include "ud3tn/simplehtab.h"

#include <stddef.h>
#include <stdlib.h>
#include <stdint.h>
#include <unistd.h>

#include <sqlite3.h>

static const char *CLA_NAME = "sqlite";

static const char *SQL_CREATE_TABLE = (
	"CREATE TABLE IF NOT EXISTS bundles(source TEXT NOT NULL, destination TEXT, creation_timestamp INTEGER NOT NULL, bundle BLOB, PRIMARY KEY(source, creation_timestamp));"
);

static const char *SQL_INSERT_INTO = (
	"INSERT INTO bundles(source, destination, creation_timestamp, bundle) VALUES (?, ?, ?, ?);"
);

struct sqlite_bundle_blob {
	sqlite3_blob *blob;
	sqlite3_int64 size;
	sqlite3_int64 offset;
};

static enum ud3tn_result sqlite_bundle_blob_open(
	struct sqlite3 *db,
	sqlite3_int64 row_id,
	struct sqlite_bundle_blob *bundle_blob)
{
	if (sqlite3_blob_open(
		db,
		"main",
		"bundles",
		"bundle",
		row_id,
		1,
		&bundle_blob->blob
	) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to open BLOB: %s", sqlite3_errmsg(db));
		return UD3TN_FAIL;
	}

	bundle_blob->size = sqlite3_blob_bytes(bundle_blob->blob);
	bundle_blob->offset = 0;

	return UD3TN_OK;
}

static void sqlite_bundle_blob_reset(struct sqlite_bundle_blob *bundle_blob)
{
	bundle_blob->blob = NULL;
	bundle_blob->size = 0;
	bundle_blob->offset = 0;
}

static enum ud3tn_result sqlite_bundle_blob_close(
	struct sqlite3 *db,
	struct sqlite_bundle_blob *bundle_blob)
{
	int32_t res = sqlite3_blob_close(bundle_blob->blob);
	if (res != SQLITE_OK)
		LOGF_ERROR("SQLiteCLA: Failed to close BLOB: %s", sqlite3_errmsg(db));

	sqlite_bundle_blob_reset(bundle_blob);

	return res != SQLITE_OK ? UD3TN_FAIL : UD3TN_OK;
}

struct sqlite_link {
	struct cla_link base;
	struct sqlite3 *db;
	QueueIdentifier_t agent_queue;
	struct sqlite_bundle_blob read_bundle_blob;
	struct sqlite_bundle_blob send_bundle_blob;
};

struct sqlite_config {
	struct cla_config base;
	struct sqlite_link *link;
	pthread_t connection_task;
};

const char *sqlite_name_get(void)
{
	return CLA_NAME;
}


static enum ud3tn_result sqlite_launch(struct cla_config *const config)
{
	return UD3TN_OK;
}

static void sqlite_disconnect_handler(struct cla_link *link)
{
	// UNUSED
	(void)link;
}

static enum ud3tn_result sqlite_terminate(struct cla_config *config)
{
	struct sqlite_config *sqlite_config = (struct sqlite_config *)config;

	// Signal the termination of RX/TX task
	cla_generic_disconnect_handler((struct cla_link *)sqlite_config->link);

	// Signal the termination of sqlite_read()
	struct cla_sqlite_command cmd = { .type = SQLITE_COMMAND_FINALIZE };
	hal_queue_push_to_back(sqlite_config->link->agent_queue, &cmd);

	// Close database connection
	sqlite3_close(sqlite_config->link->db);

	// Cleanup allocated resources
	cla_link_wait_cleanup(&sqlite_config->link->base);

	hal_queue_delete(sqlite_config->link->agent_queue);

	free(sqlite_config->link);
	free(sqlite_config);

	LOG_INFO("SQLiteCLA: Terminated gracefully");

	return UD3TN_OK;
}

static size_t sqlite_mbs_get(struct cla_config *const config)
{
	// UNUSED
	(void)config;

	return SIZE_MAX;
}

static struct cla_tx_queue sqlite_get_tx_queue(
	struct cla_config *config,
	const char *eid,
	const char *cla_addr)
{
	// UNUSED
	(void)eid;
	(void)cla_addr;

	struct sqlite_link *const link = ((struct sqlite_config *)config)->link;

	// No active link!
	if (!link)
		return (struct cla_tx_queue){ NULL, NULL };

	hal_semaphore_take_blocking(link->base.tx_queue_sem);

	// Freed while trying to obtain it
	if (!link->base.tx_queue_handle)
		return (struct cla_tx_queue){ NULL, NULL };

	return (struct cla_tx_queue){
		.tx_queue_handle = link->base.tx_queue_handle,
		.tx_queue_sem = link->base.tx_queue_sem,
	};
}

static enum ud3tn_result sqlite_start_scheduled_contact(
	struct cla_config *config,
	const char *eid,
	const char *cla_addr)
{
	// UNUSED
	(void)eid;
	(void)cla_addr;
	(void)config;

	return UD3TN_OK;
}

static enum ud3tn_result sqlite_end_scheduled_contact(
	struct cla_config *config,
	const char *eid,
	const char *cla_addr)
{
	// UNUSED
	(void)eid;
	(void)cla_addr;
	(void)config;

	return UD3TN_OK;
}

static void sqlite_begin_packet(
	struct cla_link *link,
	const struct bundle *const bundle,
	size_t length,
	char *cla_addr)
{
	// UNUSED
	(void)cla_addr;

	LOGF_DEBUG("SQLiteCLA: Storing bundle %p with destination %s", bundle, bundle->destination);

	struct sqlite_link *sqlite_link = (struct sqlite_link *)link;
	sqlite3 *db = sqlite_link->db;

	sqlite3_stmt *stmt;

	if (sqlite3_prepare_v2(db, SQL_INSERT_INTO, -1, &stmt, 0) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to prepare INSERT statement: %s", sqlite3_errmsg(db));
		return;
	}

	if (sqlite3_bind_text(stmt, 1, bundle->source, strlen(bundle->source), NULL) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to bind source column: %s", sqlite3_errmsg(db));
		return;
	}

	if (sqlite3_bind_text(
		stmt, 2, bundle->destination, strlen(bundle->destination), NULL
	) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to bind destination column: %s", sqlite3_errmsg(db));
		return;
	}

	if (sqlite3_bind_int64(stmt, 3, bundle->creation_timestamp_ms) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to bind destination column: %s", sqlite3_errmsg(db));
		return;
	}

	if (sqlite3_bind_zeroblob(stmt, 4, length) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to bind bundle column: %s", sqlite3_errmsg(db));
		return;
	}

	if (sqlite3_step(stmt) != SQLITE_DONE) {
		LOGF_ERROR("SQLiteCLA: Failed to execute INSERT statement: %s", sqlite3_errmsg(db));
		return;
	};

	if (sqlite3_finalize(stmt) != SQLITE_OK) {
		LOGF_ERROR(
			"SQLiteCLA: Failed to finalize INSERT statement: %s",
			sqlite3_errmsg(db)
		);
		return;
	}

	/* Open the bundle blob to write the data incrementally when the sqlite_send_packet_data
	 * function is called.
	 */
	sqlite_bundle_blob_open(db, sqlite3_last_insert_rowid(db), &sqlite_link->send_bundle_blob);
}

static void sqlite_send_packet_data(
	struct cla_link *link,
	const void *data,
	const size_t length)
{
	struct sqlite_link *sqlite_link = (struct sqlite_link *)link;
	sqlite3 *db = sqlite_link->db;
	struct sqlite_bundle_blob *send_bundle_blob = &sqlite_link->send_bundle_blob;

	if (sqlite3_blob_write(
		send_bundle_blob->blob,
		data,
		length,
		send_bundle_blob->offset
	) != SQLITE_OK)
		LOGF_ERROR("SQLiteCLA: Failed to write BLOB: %s", sqlite3_errmsg(db));

	send_bundle_blob->offset += length;
}

static void sqlite_end_packet(struct cla_link *link)
{
	struct sqlite_link *sqlite_link = (struct sqlite_link *)link;

	sqlite_bundle_blob_close(sqlite_link->db, &sqlite_link->send_bundle_blob);
}

static void sqlite_reset_parsers(struct cla_link *const link)
{
	struct sqlite_link *sqlite_link = (struct sqlite_link *)link;

	if (sqlite_link->read_bundle_blob.blob)
		sqlite_bundle_blob_close(sqlite_link->db, &sqlite_link->read_bundle_blob);

	rx_task_reset_parsers(&link->rx_task_data);
	link->rx_task_data.cur_parser =
		(struct parser *)&link->rx_task_data.bundle7_parser.basedata;
}

static size_t sqlite_forward_to_specific_parser(
	struct cla_link *const link,
	const uint8_t *const buffer,
	const size_t length)
{
	struct rx_task_data *const rx_data = &link->rx_task_data;
	size_t result = 0;

	switch (rx_data->payload_type) {
	case PAYLOAD_UNKNOWN:
		result = select_bundle_parser_version(rx_data, buffer, length);
		if (result == 0)
			sqlite_reset_parsers(link);
		break;
	case PAYLOAD_BUNDLE6:
		rx_data->cur_parser = rx_data->bundle6_parser.basedata;
		result = bundle6_parser_read(
			&rx_data->bundle6_parser,
			buffer,
			length
		);
		break;
	case PAYLOAD_BUNDLE7:
		rx_data->cur_parser = rx_data->bundle7_parser.basedata;
		result = bundle7_parser_read(
			&rx_data->bundle7_parser,
			buffer,
			length
		);
		break;
	default:
		sqlite_reset_parsers(link);
		return 0;
	}

	return result;
}


static enum ud3tn_result sqlite_read(
	struct cla_link *link,
	uint8_t *buffer,
	size_t length,
	size_t *bytes_read)
{
	struct sqlite_link *sqlite_link = (struct sqlite_link *)link;
	sqlite3 *db = sqlite_link->db;
	struct sqlite_bundle_blob *read_bundle_blob = &sqlite_link->read_bundle_blob;
	QueueIdentifier_t agent_queue = sqlite_link->agent_queue;

	// Check queue for new bundles if no one is in progress.
	if (!read_bundle_blob->blob) {
		struct cla_sqlite_command cmd;
		if (hal_queue_receive(agent_queue, &cmd, -1) == UD3TN_FAIL) {
			LOG_ERROR("SQLiteCLA: Failed to receive bundle from agent_queue");
			return UD3TN_FAIL;
		}
		switch (cmd.type) {
		case SQLITE_COMMAND_SELECT:
			LOGF_DEBUG(
				"SQLiteCLA: Received SQLITE_COMMAND_SELECT (row_id = %u)",
				cmd.row_id
			);
			if (sqlite_bundle_blob_open(
				db,
				cmd.row_id,
				read_bundle_blob
			) == UD3TN_FAIL)
				return UD3TN_FAIL;
			break;
		case SQLITE_COMMAND_FINALIZE:
			LOG_DEBUG("SQLiteCLA: Received SQLITE_COMMAND_FINALIZE: Terminate");
			return UD3TN_OK;
		case SQLITE_COMMAND_UNDEFINED:
			LOG_DEBUG("SQLiteCLA: Received SQLITE_COMMAND_UNDEFINED");
			return UD3TN_FAIL;
		}
	}

	// Recalucate length, if the remaining bundle is smaller than the buffer.
	if (read_bundle_blob->offset + (sqlite3_int64)length > read_bundle_blob->size)
		length = read_bundle_blob->size - read_bundle_blob->offset;

	// Read
	if (sqlite3_blob_read(
		read_bundle_blob->blob,
		buffer,
		length,
		read_bundle_blob->offset
	) == SQLITE_ERROR) {
		LOGF_ERROR(
			"SQLiteCLA: Failed to read Bundle BLOB: %s",
			sqlite3_errmsg(db)
		);
		sqlite3_blob_close(read_bundle_blob->blob);
		return UD3TN_FAIL;
	}

	// Update the offset
	read_bundle_blob->offset += length;

	// The bundle is completely read, reset the buffers to parse the next one.
	if (read_bundle_blob->offset >= read_bundle_blob->size)
		sqlite_bundle_blob_close(db, read_bundle_blob);

	if (bytes_read)
		*bytes_read = length;

	return UD3TN_OK;
}

const struct cla_vtable sqlite_vtable = {
	.cla_name_get = sqlite_name_get,

	.cla_launch = sqlite_launch,
	.cla_terminate = sqlite_terminate,

	.cla_mbs_get = sqlite_mbs_get,

	.cla_get_tx_queue = sqlite_get_tx_queue,
	.cla_start_scheduled_contact = sqlite_start_scheduled_contact,
	.cla_end_scheduled_contact = sqlite_end_scheduled_contact,

	.cla_begin_packet = sqlite_begin_packet,
	.cla_end_packet = sqlite_end_packet,
	.cla_send_packet_data = sqlite_send_packet_data,

	.cla_rx_task_reset_parsers = sqlite_reset_parsers,
	.cla_rx_task_forward_to_specific_parser =
		sqlite_forward_to_specific_parser,

	.cla_read = sqlite_read,

	.cla_disconnect_handler = sqlite_disconnect_handler,
};

static enum ud3tn_result sqlite_link_init(
	struct sqlite_link *link,
	struct cla_config *base,
	sqlite3 *db,
	QueueIdentifier_t agent_queue)
{
	link->db = db;
	link->agent_queue = agent_queue;
	sqlite_bundle_blob_reset(&link->read_bundle_blob);
	sqlite_bundle_blob_reset(&link->send_bundle_blob);

	if (cla_link_init(
		&link->base,
		base,
		NULL,
		true,
		true
	) != UD3TN_OK) {
		LOG_ERROR("SQLiteCLA: Failed to initialize link");
		return UD3TN_FAIL;
	}

	return UD3TN_OK;
}

static enum ud3tn_result sqlite_init(
	struct sqlite_config *config,
	const char *filename,
	const struct bundle_agent_interface *bundle_agent_interface)
{
	// Initialize the CLA base configuration
	if (cla_config_init(&config->base, bundle_agent_interface) != UD3TN_OK) {
		LOG_ERROR("SQLiteCLA: Failed to initialize CLA base config");
		return UD3TN_FAIL;
	}
	config->base.vtable = &sqlite_vtable;

	// Create a message queue to connect the sqlite agent with the CLA
	QueueIdentifier_t agent_queue = hal_queue_create(
		CLA_SQLITE_AGENT_QUEUE_LENGTH,
		sizeof(struct cla_sqlite_command)
	);
	if (!agent_queue) {
		LOG_ERROR("SQLiteCLA: Failed to create agent queue");
		return UD3TN_FAIL;
	}

	// Open the database and create the tables
	sqlite3 *db;
	if (sqlite3_config(SQLITE_CONFIG_SERIALIZED) != SQLITE_OK) {
		LOG_ERROR(
			"SQLiteCLA: Failed to set SQLITE_CONFIG_SERIALIZED (SQLite needs to be compiled with the SQLITE_THREADSAFE=1 compile-time option)"
		);
		goto failed_queue_delete;
	}
	if (sqlite3_open(filename, &db) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to open database file: %s", sqlite3_errmsg(db));
		goto failed_queue_delete;
	}
	if (sqlite3_exec(db, SQL_CREATE_TABLE, 0, 0, NULL) != SQLITE_OK) {
		LOGF_ERROR("SQLiteCLA: Failed to create table: %s", sqlite3_errmsg(db));
		goto failed_db_close;
	}


	// Initialize the link configuration
	config->link = malloc(sizeof(struct sqlite_link));
	if (!config->link) {
		LOG_ERROR("SQLiteCLA: Failed to allocate sqlite_link");
		goto failed_db_close;
	}
	if (sqlite_link_init(config->link, &config->base, db, agent_queue) != UD3TN_OK) {
		LOG_ERROR("SQLiteCLA: Failed to initialize sqlite_link");
		goto failed_link_free;
	}

	// Register the CLA
	bundle_processor_inform(
		bundle_agent_interface->bundle_signaling_queue,
		(struct bundle_processor_signal) {
			.type = BP_SIGNAL_NEW_LINK_ESTABLISHED,
			.peer_cla_addr = cla_get_cla_addr_from_link(
				&config->link->base
			),
		}
	);

	// Initialize the SQLite agent
	if (sqlite_agent_setup(bundle_agent_interface, db, agent_queue) != 0) {
		LOG_ERROR("SQLiteCLA: Failed to initialize SQLiteAgent");
		goto failed_link_free;
	}

	return UD3TN_OK;

failed_link_free:
	free(config->link);
failed_db_close:
	sqlite3_close(db);
failed_queue_delete:
	hal_queue_delete(agent_queue);

	return UD3TN_FAIL;
}

struct cla_config *sqlite_create(
	const char *const options[],
	const size_t option_count,
	const struct bundle_agent_interface *bundle_agent_interface)
{
	LOGF_DEBUG("SQLiteCLA: Using library version %s.", sqlite3_libversion());

	if (option_count != 1) {
		LOG_ERROR("SQLiteCLA: Option has to be a file path.");
		return NULL;
	}

	struct sqlite_config *config = malloc(sizeof(struct sqlite_config));

	if (!config) {
		LOG_ERROR("SQLiteCLA: Memory allocation failed!");
		return NULL;
	}

	if (sqlite_init(config, options[0],
		bundle_agent_interface) != UD3TN_OK) {
		free(config);
		LOG_ERROR("SQLiteCLA: Initialization failed!");
		return NULL;
	}

	return &config->base;
}