// 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 #include #include #include #include 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; }