#include "control.h"
#include <cstdio>
#include <cstdlib>
#include <cstring>
extern "C" {
#include <compact.h>
#include <rpc.h>
}
namespace swarmd {
// The dispatcher's handlers are plain C function pointers with a void* ctx, so
// the instance travels through ctx rather than through a capture.
struct Ctx {
Control *self;
Control::Session *session;
ControlBackend *backend;
uint64_t *hellos;
uint64_t *client_id;
};
static int on_hello(void *vctx, const swarmd_hello_request_t *req,
swarmd_hello_response_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
// ⛔ COPY OUT. req's views point into the inbound frame buffer, which is
// reused the moment this returns.
memcpy(ctx->session->app_identity, req->app_identity, 32);
ctx->session->name.assign((const char *)req->name.data, req->name.len);
ctx->session->hello = true;
(*ctx->hellos)++;
// ⛔ THESE ARE WHAT HELLO IS FOR. §5 says the reply carries {client_id,
// daemon_pubkey}; §7 item 4 says client ids exist so a client can suppress
// its OWN echo once messages carry an origin. Both were zero — a placeholder
// with a comment saying someone else would fill it, and nobody did. A client
// that cannot learn its own id cannot tell its own traffic from a peer's,
// and a zeroed key tells it nothing about which daemon it reached.
res->client_id = *ctx->client_id;
memset(res->daemon_pubkey, 0, 32);
if (ctx->backend->daemon_pubkey) ctx->backend->daemon_pubkey(res->daemon_pubkey);
return 0;
}
static int on_status_get(void *vctx, const swarmd_none_t *,
swarmd_status_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
memset(res, 0, sizeof(*res));
if (ctx->backend->status) ctx->backend->status(res);
return 0;
}
static int on_swarm_join(void *vctx, const swarmd_swarm_join_t *req,
swarmd_ok_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
res->ok = ctx->backend->swarm_join
? ctx->backend->swarm_join(req->topic, req->server, req->client)
: false;
return 0;
}
static int on_swarm_leave(void *vctx, const swarmd_topic_ref_t *req,
swarmd_ok_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
res->ok = ctx->backend->swarm_leave ? ctx->backend->swarm_leave(req->topic) : false;
return 0;
}
static int on_channel_open(void *vctx, const swarmd_channel_open_t *req,
swarmd_channel_ref_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
res->channel = 0;
if (!ctx->backend->channel_open) return 0;
// ⛔ COPY OUT. protocol and id point into the inbound frame buffer, and the
// channel outlives this call by design.
std::string protocol((const char *)req->protocol.data, req->protocol.len);
std::vector<uint8_t> id(req->id, req->id + req->id_len);
res->channel = ctx->backend->channel_open(protocol, id, (uint32_t)req->slots,
*ctx->client_id);
return 0;
}
static int on_channel_close(void *vctx, const swarmd_channel_ref_t *req,
swarmd_ok_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
res->ok = ctx->backend->channel_close ? ctx->backend->channel_close(req->channel) : false;
return 0;
}
static int on_send(void *vctx, const swarmd_send_request_t *req,
swarmd_send_result_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
res->accepted = false;
res->writable = false;
if (!ctx->backend->send) return 0;
bool writable = true;
res->accepted = ctx->backend->send(req->channel, (uint32_t)req->slot, req->target,
req->bytes, req->bytes_len, &writable);
res->writable = writable;
return 0;
}
// ---- content -------------------------------------------------------------
//
// Nothing here moves bytes. CONTENT_HASH reads a file and answers with 32 bytes;
// the other three are orders, answered with ok. The file itself never touches
// this socket — docs/SWARMD-DESIGN.md §6.
static int on_content_hash(void *vctx, const swarmd_path_t *req,
swarmd_hash_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
memset(res->hash, 0, 32);
if (!ctx->backend->content_hash) return 0;
// ⛔ COPY OUT: the view points into the inbound frame buffer.
const std::string path((const char *)req->path.data, req->path.len);
ctx->backend->content_hash(path, res->hash);
return 0;
}
static int on_content_serve(void *vctx, const swarmd_content_serve_t *req,
swarmd_ok_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
const std::string path((const char *)req->path.data, req->path.len);
res->ok = ctx->backend->content_serve ? ctx->backend->content_serve(req->hash, path) : false;
return 0;
}
static int on_content_fetch(void *vctx, const swarmd_content_fetch_t *req,
swarmd_ok_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
const std::string dest((const char *)req->dest_path.data, req->dest_path.len);
res->ok = ctx->backend->content_fetch ? ctx->backend->content_fetch(req->hash, dest) : false;
return 0;
}
static int on_content_cancel(void *vctx, const swarmd_content_cancel_t *req,
swarmd_ok_t *res, hrpc_error_t *) {
Ctx *ctx = (Ctx *)vctx;
res->ok = ctx->backend->content_cancel
? ctx->backend->content_cancel(req->hash, req->discard)
: false;
return 0;
}
void Control::attach(UdsServer *uds, ControlBackend backend) {
_uds = uds;
_backend = std::move(backend);
_uds->on_connect = [this](uint64_t id) { _sessions[id] = Session(); };
_uds->on_disconnect = [this](uint64_t id) { _sessions.erase(id); };
_uds->on_frame = [this](uint64_t id, const uint8_t *d, size_t n) { _on_frame(id, d, n); };
}
const Control::Session *Control::session(uint64_t client_id) const {
auto it = _sessions.find(client_id);
return it == _sessions.end() ? nullptr : &it->second;
}
void Control::_on_frame(uint64_t client_id, const uint8_t *data, size_t len) {
auto sit = _sessions.find(client_id);
if (sit == _sessions.end()) return;
compact_state_t state = { 0, len, (uint8_t *)data };
rpc_message_t msg;
if (rpc_decode_message(&state, &msg) < 0) {
fprintf(stderr, "[swarmd] client %llu sent an undecodable message — dropping it\n",
(unsigned long long)client_id);
return;
}
Ctx ctx = { this, &sit->second, &_backend, &_hellos, &client_id };
swarmd_hrpc_handlers_t h;
memset(&h, 0, sizeof(h));
h.ctx = &ctx;
h.on_hello = on_hello;
h.on_status_get = on_status_get;
h.on_swarm_join = on_swarm_join;
h.on_swarm_leave = on_swarm_leave;
h.on_channel_open = on_channel_open;
h.on_channel_close = on_channel_close;
h.on_send = on_send;
h.on_content_hash = on_content_hash;
h.on_content_serve = on_content_serve;
h.on_content_fetch = on_content_fetch;
h.on_content_cancel = on_content_cancel;
uint8_t *reply = nullptr;
size_t reply_len = 0;
_current_client = client_id;
const int r = swarmd_hrpc_dispatch(&h, &msg, &reply, &reply_len);
_current_client = 0;
if (r == hrpc_dispatch_reply && reply != nullptr) {
_uds->send(client_id, reply, reply_len);
// ⛔ THE CALLER FREES. Every *_encode_* heap-allocates its output.
free(reply);
}
}
void Control::push_status(uint64_t client_id) {
swarmd_status_t st;
memset(&st, 0, sizeof(st));
if (_backend.status) _backend.status(&st);
uint8_t *out = nullptr;
size_t out_len = 0;
if (swarmd_encode_on_status(&st, &out, &out_len) < 0 || out == nullptr) return;
_uds->send(client_id, out, out_len);
free(out); // caller frees
}
} // namespace swarmd