ud3tn/components/agents/sqlite/sqlite_agent.c

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

#include "cla/posix/cla_sqlite.h"

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

#include "ud3tn/bundle_processor.h"
#include "ud3tn/common.h"
#include "ud3tn/eid.h"
#include "ud3tn/result.h"

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

#include <sqlite3.h>

#include <pb_decode.h>

struct sqlite_agent_params {
	sqlite3 *db;
	QueueIdentifier_t cla_queue;
};

static enum ud3tn_result storage_operation_delete_bundles(sqlite3 *db, BundleFilter *filter)
{
	sqlite3_stmt *stmt;
	const char *sql = "DELETE FROM bundles WHERE destination GLOB ?";

	if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) {
		LOGF_ERROR("SQLiteAgent: Failed to prepare SQL statement: %s", sqlite3_errmsg(db));
		return UD3TN_FAIL;
	}

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

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

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

	return UD3TN_OK;
}

static enum ud3tn_result storage_operation_push_bundles(
	sqlite3 *db,
	QueueIdentifier_t cla_queue,
	BundleFilter *filter)
{
	sqlite3_stmt *stmt;

	const char *sql = "SELECT rowid from bundles WHERE destination GLOB ?";

	if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) {
		LOGF_ERROR("SQLiteAgent: Failed to prepare SQL statement: %s", sqlite3_errmsg(db));
		return UD3TN_FAIL;
	}

	if (sqlite3_bind_text(
		stmt, 1, filter->eid_glob, strlen(filter->eid_glob), NULL
	) != SQLITE_OK) {
		LOGF_ERROR(
			"SQLiteAgent: Failed to bind destination column: %s",
			sqlite3_errmsg(db)
		);
		return UD3TN_FAIL;
	}

	while (sqlite3_step(stmt) != SQLITE_DONE) {
		struct cla_sqlite_command cmd =  {
			.type = SQLITE_COMMAND_SELECT,
			.row_id = sqlite3_column_int64(stmt, 0),
		};

		LOGF_DEBUG("SQLiteAgent: Send SQLITE_COMMAND_SELECT (row_id = %u)", cmd.row_id);

		hal_queue_push_to_back(cla_queue, &cmd);
	}

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

	return UD3TN_OK;
}

static enum ud3tn_result storage_operation_list_bundles(
	sqlite3 *db,
	BundleFilter *filter)
{
	LOG_WARN("SQLiteAgent: STORAGE_OPERATION_LIST_BUNDLES unimplemented!");
	return UD3TN_OK;
}


static enum ud3tn_result parse_storage_call(uint8_t *data, size_t length, StorageCall *cmd)
{
	pb_istream_t stream = pb_istream_from_buffer(data, length);

	if (!pb_decode_ex(&stream, StorageCall_fields, cmd, PB_DECODE_DELIMITED)) {
		LOGF_ERROR("SQLiteAgent: Protobuf decode error: %s", PB_GET_ERROR(&stream));
		return UD3TN_FAIL;
	}

	return UD3TN_OK;
}

static void sqlite_agent_callback(struct bundle_adu data, void *p, const void *bp_context)
{
	sqlite3 *db = ((struct sqlite_agent_params *)p)->db;
	QueueIdentifier_t cla_queue = ((struct sqlite_agent_params *)p)->cla_queue;
	StorageCall cmd = StorageCall_init_default;

	parse_storage_call(data.payload, data.length, &cmd);

	switch (cmd.operation) {
	case StorageOperation_STORAGE_OPERATION_DELETE_BUNDLES:
		LOGF_DEBUG(
			"SQLiteAgent: Received STORAGE_OPERATION_DELETE_BUNDLES (filter.eid_glob = %s)",
			cmd.filter.eid_glob
		);
		storage_operation_delete_bundles(db, &cmd.filter);
		break;
	case StorageOperation_STORAGE_OPERATION_PUSH_BUNDLES:
		LOGF_DEBUG(
			"SQLiteAgent: Received STORAGE_OPERATION_PUSH_BUNDLES (filter.eid_glob = %s)",
			cmd.filter.eid_glob
		);
		storage_operation_push_bundles(db, cla_queue, &cmd.filter);
		break;
	case StorageOperation_STORAGE_OPERATION_LIST_BUNDLES:
		LOGF_DEBUG(
			"SQLiteAgent: Received STORAGE_OPERATION_LIST_BUNDLES (filter.eid_glob = %s)",
			cmd.filter.eid_glob
		);
		storage_operation_list_bundles(db, &cmd.filter);
		break;
	default:
		LOG_WARN("SQLiteAgent: Unsupported StorageOperation");
		return;
	}

	// Free allocated data
	pb_release(StorageCall_fields, &cmd);
	bundle_adu_free_members(data);
}

int sqlite_agent_setup(
	const struct bundle_agent_interface *const bai,
	sqlite3 *db,
	QueueIdentifier_t cla_queue)
{
	ASSERT(db);
	ASSERT(cla_queue);
	ASSERT(sqlite3_threadsafe() != 0);

	struct sqlite_agent_params *params = malloc(
		sizeof(struct sqlite_agent_params)
	);

	if (!params)
		return -1;

	params->db = db;
	params->cla_queue = cla_queue;

	bool is_ipn = get_eid_scheme(bai->local_eid) == EID_SCHEME_IPN;

	const struct agent agent = {
		.sink_identifier = is_ipn ? AGENT_ID_SQLITE_IPN : AGENT_ID_SQLITE_DTN,
		.callback = sqlite_agent_callback,
		.param = params,
	};

	if (bundle_processor_perform_agent_action(
		bai->bundle_signaling_queue,
		BP_SIGNAL_AGENT_REGISTER,
		agent,
		false
	)) {
		LOG_ERROR("SQLiteAgent: Failed to register with the bundle processor");
		free(params);
		return -1;
	}

	return 0;
}