fumi/core/shim/smolmail_rns.cpp

/*
 * microReticulum bridge for the fumi RNS client carrier (upstream spec sec
 * 13). Owns the Reticulum instance, the UDP interface and the loop thread
 * that drives Reticulum::loop(). Every request/response exchange goes through
 * a single slot guarded by a condvar: the CLI has one request in flight at a
 * time, so no map of callbacks is needed.
 *
 * microReticulum splices the request payload into its msgpack envelope
 * verbatim (Link.cpp pack_request_envelope), so the request must be packed
 * as a msgpack binary by the caller; the response, by contrast, arrives
 * already decoded (unpack_response_envelope yields the payload itself), so
 * only the outbound direction touches msgpack here.
 */
#include "smolmail_rns.h"
#include "udp_interface.h"

#include <microStore/FileSystem.h>
#include <microStore/Adapters/UniversalFileSystem.h>
#include <MsgPack.h>

#include <microReticulum.h>

#include <chrono>
#include <condition_variable>
#include <cstring>
#include <mutex>
#include <thread>

// Anchored in src/rns/transport.rs and upstream spec sec 13.1/13.5.
static const char* APP_NAME   = "smolmail";
static const char* APP_ASPECT = "server";
static const char* REQ_PATH   = "smolmail/1";

static const double LOOP_SLEEP_SECS = 0.01;
static const double CONNECT_POLL_SECS = 0.05;
static const double PATH_POLL_SECS = 0.1;

static RNS::Reticulum reticulum({RNS::Type::NONE});
static RNS::Interface udp_interface({RNS::Type::NONE});
static RNS::Link active_link({RNS::Type::NONE});
static microStore::FileSystem filesystem{microStore::Adapters::UniversalFileSystem()};

static volatile bool running = false;
// Joinable, never detached: the loop thread must be joined by
// smolmail_rns_stop before the process tears down statics, or it keeps
// calling reticulum.loop() while their destructors run.
static std::thread loop_thread;

// The single response slot (plan: one request in flight at a time).
static std::mutex slot_mutex;
static std::condition_variable slot_cv;
static RNS::Bytes slot_response;
static volatile bool slot_ready = false;   // response or failure arrived
static volatile bool slot_failed = false;

// Link establishment, signalled from the loop thread.
static std::mutex link_mutex;
static std::condition_variable link_cv;
static volatile bool link_established = false;
static volatile bool link_closed_early = false;

static void on_link_established(RNS::Link& established) {
	(void)established;
	std::lock_guard<std::mutex> lock(link_mutex);
	link_established = true;
	link_cv.notify_all();
}

static void on_link_closed(RNS::Link& closed) {
	(void)closed;
	{
		std::lock_guard<std::mutex> lock(link_mutex);
		link_closed_early = true;
		link_cv.notify_all();
	}
	// A dead link fails any request waiting on the slot rather than letting
	// it run to the timeout (upstream spec sec 13.5: a local error).
	std::lock_guard<std::mutex> lock(slot_mutex);
	slot_failed = true;
	slot_ready = true;
	slot_cv.notify_all();
}

static void on_response(const RNS::RequestReceipt& receipt) {
	RNS::Bytes response = receipt.get_response();
	std::lock_guard<std::mutex> lock(slot_mutex);
	slot_response = response;
	slot_failed = false;
	slot_ready = true;
	slot_cv.notify_all();
}

static void on_failed(const RNS::RequestReceipt& receipt) {
	(void)receipt;
	std::lock_guard<std::mutex> lock(slot_mutex);
	slot_failed = true;
	slot_ready = true;
	slot_cv.notify_all();
}

// The unwrapped large-response payload, filled in per request below. It is
// deliberately local: a static here kept the last resource-path response
// alive past its request, so every later bare response -- an empty fetch
// page, an ack -- was misread as that stale page and the client looped
// FETCH/DELETE pairs forever.
static void loop_thread_main() {
	while (running) {
		reticulum.loop();
		RNS::Utilities::OS::sleep(LOOP_SLEEP_SECS);
	}
}

extern "C" int smolmail_rns_start(const char* storage_dir,
                                 const char* udp_listen_host, uint16_t udp_listen_port,
                                 const char* udp_forward_host, uint16_t udp_forward_port) {
	if (running) {
		return 0; // already started; the storage path cannot change mid-run
	}

	// Registered before anything else, as the interop examples do, so
	// persistence writes have a filesystem to go through.
	filesystem.init();
	RNS::Utilities::OS::register_filesystem(filesystem);
	RNS::Reticulum::storagepath(storage_dir);

	udp_interface = new UDPInterface("smolmail_rns_udp",
	                                 udp_listen_host ? udp_listen_host : "",
	                                 udp_listen_port,
	                                 udp_forward_host ? udp_forward_host : "",
	                                 udp_forward_port);
	udp_interface.mode(RNS::Type::Interface::MODE_GATEWAY);
	RNS::Transport::register_interface(udp_interface);
	if (!udp_interface.start()) {
		return -2;
	}

	reticulum = RNS::Reticulum();
	// Transport mode must be on for a client: the known-destinations store
	// (what Identity::recall reads, and what path discovery stores the
	// announced identity into) is only initialised inside it. The Python
	// client never trips this because its known-destinations map always
	// exists in memory; microReticulum with RNS_USE_FS keeps it in a
	// FileStore instead.
	reticulum.transport_enabled(true);
	reticulum.start();

	running = true;
	loop_thread = std::thread(loop_thread_main);
	return 0;
}

extern "C" int smolmail_rns_connect(const uint8_t* destination_hash, uint32_t timeout_ms,
                                    uint8_t* link_id_out) {
	const RNS::Bytes hash(destination_hash, 16);

	// Request a path and WAIT before creating the link (upstream spec sec
	// 13.3): with no path RNS assumes the maximum hop count and the link
	// fails after minutes instead of promptly. The deadline is the stack's
	// own path request timeout, not a number copied from the document.
	if (!RNS::Transport::has_path(hash)) {
		RNS::Transport::request_path(hash);
		const double deadline = RNS::Utilities::OS::time()
			+ (double)RNS::Type::Transport::PATH_REQUEST_TIMEOUT;
		while (!RNS::Transport::has_path(hash)
		       && RNS::Utilities::OS::time() < deadline) {
			RNS::Utilities::OS::sleep(PATH_POLL_SECS);
		}
		if (!RNS::Transport::has_path(hash)) {
			return -1;
		}
	}

	// Recall can return nothing immediately after a path appears; the caller
	// treats that as a retry, not an error.
	RNS::Identity identity = RNS::Identity::recall(hash);
	if (!identity) {
		return -2;
	}

	// One fresh link per session, never reused across authentications:
	// link_id is what stops an AUTH replaying on another link (upstream
	// spec sec 13.6).
	RNS::Destination destination(identity, RNS::Type::Destination::OUT,
	                              RNS::Type::Destination::SINGLE, APP_NAME, APP_ASPECT);
	{
		std::lock_guard<std::mutex> lock(link_mutex);
		link_established = false;
		link_closed_early = false;
	}
	{
		std::lock_guard<std::mutex> slot_lock(slot_mutex);
		slot_ready = false;
		slot_failed = false;
	}
	active_link = RNS::Link(destination, on_link_established, on_link_closed);

	const double deadline = RNS::Utilities::OS::time() + (double)timeout_ms / 1000.0;
	{
		std::unique_lock<std::mutex> lock(link_mutex);
		while (!link_established && !link_closed_early
		       && RNS::Utilities::OS::time() < deadline) {
			link_cv.wait_for(lock, std::chrono::duration<double>(CONNECT_POLL_SECS));
		}
	}
	if (!link_established || active_link.status() != RNS::Type::Link::ACTIVE) {
		active_link.teardown();
		active_link = RNS::Link({RNS::Type::NONE});
		return -3;
	}
	memcpy(link_id_out, active_link.link_id().data(), 16);
	return 0;
}

extern "C" int smolmail_rns_request(const uint8_t* request, size_t request_len,
                                    uint32_t timeout_ms, uint8_t* out, size_t cap,
                                    size_t* out_len) {
	if (!active_link || active_link.status() != RNS::Type::Link::ACTIVE) {
		return -1;
	}

	// The request payload must be msgpack-encoded itself: Link::request
	// splices it verbatim into the envelope's third element.
	MsgPack::Packer packer;
	packer.packBinary(request, request_len);
	RNS::Bytes encoded(packer.data(), packer.size());

	{
		std::lock_guard<std::mutex> lock(slot_mutex);
		slot_ready = false;
		slot_failed = false;
		slot_response.clear();
	}
	// A client MUST set its own request timeout (upstream spec sec 13.5):
	// Reticulum's default is derived from the round trip time and covers a
	// packet, not a FETCH page.
	RNS::RequestReceipt receipt = active_link.request(RNS::Bytes(REQ_PATH), encoded,
	                                           on_response, on_failed, nullptr,
	                                           (double)timeout_ms / 1000.0);
	if (!receipt) {
		return -2;
	}

	// No path, a dead link, a rejected resource and a timeout are local
	// errors; none of them is a status code.
	{
		std::unique_lock<std::mutex> lock(slot_mutex);
		const bool concluded = slot_cv.wait_for(lock,
			std::chrono::milliseconds(timeout_ms + 1000),
			[]() { return slot_ready; });
		if (!concluded) {
			return -3;
		}
		if (slot_failed) {
			return -4;
		}
	}

	// microReticulum decodes the response envelope differently for the two
	// transfer modes: a small response arrives as the bare smolmail payload
	// (`status u8 || payload`), while a large one, transferred as a
	// resource, is still wrapped in its msgpack binary (Link.cpp hands the
	// raw remainder to handle_response on that path). The wrapper is
	// unambiguous: a smolmail status is a single byte below 16, and the
	// msgpack bin headers are 0xC4..0xC6, so only those are unwrapped.
	RNS::Bytes unwrapped;
	const RNS::Bytes& payload = [&]() -> const RNS::Bytes& {
		if (slot_response.size() > 0 && slot_response.data()[0] >= 0xC4
		    && slot_response.data()[0] <= 0xC6) {
			MsgPack::Unpacker unpacker;
			unpacker.feed(slot_response.data(), slot_response.size());
			if (unpacker.isBin()) {
				MsgPack::bin_t<uint8_t> bin;
				unpacker.deserialize(bin);
				unwrapped = RNS::Bytes(bin.data(), bin.size());
			}
		}
		return unwrapped.size() > 0 ? unwrapped : slot_response;
	}();
	if (!payload || payload.size() == 0) {
		return -5;
	}
	if (payload.size() > cap) {
		return -6;
	}
	memcpy(out, payload.data(), payload.size());
	*out_len = payload.size();
	return 0;
}

extern "C" void smolmail_rns_close(void) {
	if (active_link) {
		active_link.teardown();
		active_link = RNS::Link({RNS::Type::NONE});
	}
}

extern "C" void smolmail_rns_stop(void) {
	if (!running) {
		return;
	}
	// Order matters: halt and join the loop thread first, so nothing is
	// inside Reticulum, Transport or the filesystem while they are torn
	// down; only then take the link, the interface and the instance apart.
	running = false;
	if (loop_thread.joinable()) {
		loop_thread.join();
	}
	if (active_link) {
		active_link.teardown();
		active_link = RNS::Link({RNS::Type::NONE});
	}
	RNS::Transport::deregister_interface(udp_interface);
	udp_interface.stop();
	reticulum = RNS::Reticulum({RNS::Type::NONE});
}