Talon/voice-cat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(M2): UDP voice/media plane -- SFU relay, Opus, AEAD, jitter buffer

Browse Source 
Adds the full voice pipeline: 14-byte binary frame header, ChaCha20-Poly1305
AEAD keyed from the TLS exporter, libopus encode/decode with FEC/PLC/DTX,
an adaptive per-ssrc jitter buffer, a miniaudio capture/playback engine, an
APM passthrough stub, and the UdpBinding/StreamAnnounce signaling chain
wired through ConnSession/SessionRegistry into a new server-side SFU
(MediaRelay) that decrypts and re-encrypts frames per channel member.

Exit criterion verified: test_m2_voice — two headless clients relay 50
encrypted Opus frames through the server; ctest --preset m1-dev is 9/9
green. Also corrects protocol.md's UdpBinding diagram, which described the
UDP-side binding packet as AEAD-sealed when it is in fact a plaintext
bootstrap frame (separate from the TCP/TLS UdpBinding ack).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Talon
2026-06-16 01:31:14 +02:00
parent 63f457fc54
commit 694494a5be
29 changed files with 2548 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
server/src/conn_session.cpp
View File

@@ -4,6 +4,9 @@
#include <chrono>
#include <cstdio>
#include <cstring>
#include <sodium.h>
#include "db.h"
#include "session_registry.h"
@@ -22,12 +25,16 @@ ConnSession::ConnSession(std::shared_ptr<Database> db,
std::shared_ptr<SessionRegistry> registry,
std::shared_ptr<voicecat::WorkerPool> workers,
const std::array<uint8_t, 32>& server_fp,
bool allow_guests)
bool allow_guests,
uint16_t udp_media_port)
: db_(std::move(db)),
registry_(std::move(registry)),
workers_(std::move(workers)),
server_fp_(server_fp),
allow_guests_(allow_guests) {}
allow_guests_(allow_guests),
udp_media_port_(udp_media_port) {
randombytes_buf(udp_token_.data(), udp_token_.size());
}
void ConnSession::set_io(SendFn send_fn, CloseFn close_fn) {
send_fn_ = std::move(send_fn);
@@ -67,6 +74,14 @@ void ConnSession::on_frame(std::vector<uint8_t> frame) {
if (st == State::Authenticated)
registry_->set_user_channel(user_id_.load(), 1);
break;
case voicecat::v1::Envelope::kUdpBinding:
if (st == State::Authenticated)
handle_udp_binding(env.request_id(), env.udp_binding());
break;
case voicecat::v1::Envelope::kStreamAnnounce:
if (st == State::Authenticated)
handle_stream_announce(env.request_id(), env.stream_announce());
break;
default:
break;
}
@@ -93,6 +108,44 @@ void ConnSession::close() {
if (close_fn_) close_fn_();
}
// ── M2: media crypto ─────────────────────────────────────────────────────────
void ConnSession::set_media_crypto(
std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> send,
std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> recv) {
std::lock_guard lk(crypto_mu_);
send_crypto_ = std::move(send);
recv_crypto_ = std::move(recv);
}
voicecat::crypto::SodiumMediaCrypto* ConnSession::send_crypto() {
std::lock_guard lk(crypto_mu_);
return send_crypto_.get();
}
voicecat::crypto::SodiumMediaCrypto* ConnSession::recv_crypto() {
std::lock_guard lk(crypto_mu_);
return recv_crypto_.get();
}
// ── M2: UDP endpoint ─────────────────────────────────────────────────────────
void ConnSession::set_udp_endpoint(asio::ip::udp::endpoint ep) {
{
std::lock_guard lk(udp_ep_mu_);
udp_ep_ = ep;
}
has_udp_ep_.store(true, std::memory_order_release);
registry_->register_udp_endpoint(ep, session_id_);
}
asio::ip::udp::endpoint ConnSession::udp_endpoint() const {
std::lock_guard lk(udp_ep_mu_);
return udp_ep_;
}
// ── Handlers ─────────────────────────────────────────────────────────────────
void ConnSession::handle_client_hello(uint64_t req_id, const voicecat::v1::ClientHello& msg) {
if (msg.proto_version() != 1) {
send_disconnect_and_close(1, "unsupported protocol version");
@@ -106,6 +159,7 @@ void ConnSession::handle_client_hello(uint64_t req_id, const voicecat::v1::Clien
if (allow_guests_) hello->add_auth_methods("guest");
hello->add_auth_methods("password");
hello->set_server_identity_fingerprint(server_fp_.data(), server_fp_.size());
if (udp_media_port_) hello->set_udp_port(udp_media_port_);
send_envelope(env);
state_.store(State::WaitingAuth, std::memory_order_release);
}
@@ -141,12 +195,15 @@ void ConnSession::finish_guest_auth(const voicecat::v1::GuestAuth& guest, uint64
user_id_.store(uid, std::memory_order_relaxed);
state_.store(State::Authenticated, std::memory_order_release);
registry_->register_udp_token(udp_token_, session_id_);
{
auto env = make_env(req_id);
auto* res = env.mutable_auth_result();
res->set_ok(true);
res->set_session_id(session_id_);
*res->mutable_self() = user;
res->set_udp_token(udp_token_.data(), udp_token_.size());
send_envelope(env);
}
broadcast_user_joined(user);
@@ -176,6 +233,8 @@ void ConnSession::finish_password_auth(const std::string& username,
self->user_id_.store(uid, std::memory_order_relaxed);
self->state_.store(State::Authenticated, std::memory_order_release);
self->registry_->register_udp_token(self->udp_token_, self->session_id_);
{
auto env = make_env(req_id);
auto* res = env.mutable_auth_result();
@@ -184,6 +243,7 @@ void ConnSession::finish_password_auth(const std::string& username,
*res->mutable_self() = user;
auto* perms = res->mutable_permissions();
perms->set_is_admin(acc->is_admin);
res->set_udp_token(self->udp_token_.data(), self->udp_token_.size());
self->send_envelope(env);
}
self->broadcast_user_joined(user);
@@ -247,6 +307,48 @@ void ConnSession::handle_ping(const voicecat::v1::Ping& msg) {
send_envelope(env);
}
void ConnSession::handle_udp_binding(uint64_t req_id, const voicecat::v1::UdpBinding& msg) {
if (msg.ack()) return; // server→client direction; ignore if echoed back
const std::string& tok = msg.udp_token();
if (tok.size() != 16 || std::memcmp(tok.data(), udp_token_.data(), 16) != 0) {
// Bad token — silently ignore (don't leak timing information)
return;
}
// Ack over TCP; MediaRelay will set the UDP endpoint when the UDP binding packet arrives.
auto env = make_env(req_id);
env.mutable_udp_binding()->set_ack(true);
send_envelope(env);
}
void ConnSession::handle_stream_announce(uint64_t req_id,
const voicecat::v1::StreamAnnounce& msg) {
uint32_t ssrc = registry_->assign_ssrc(session_id_);
auto env = make_env(req_id);
auto* res = env.mutable_stream_announce_result();
res->set_ok(true);
res->set_stream_id(1);
res->set_ssrc(ssrc);
auto* eff = res->mutable_effective_audio();
if (msg.has_requested_audio()) {
*eff = msg.requested_audio();
} else {
eff->set_codec(0); // OPUS
eff->set_sample_rate(48000);
eff->set_bitrate_bps(24000);
eff->set_frame_ms(20);
eff->set_fec(true);
}
if (eff->sample_rate() == 0) eff->set_sample_rate(48000);
if (eff->bitrate_bps() == 0) eff->set_bitrate_bps(24000);
if (eff->frame_ms() == 0) eff->set_frame_ms(20);
send_envelope(env);
}
void ConnSession::send_disconnect_and_close(uint32_t code, const std::string& reason) {
auto env = make_env();
auto* d = env.mutable_disconnect();

84
server/src/conn_session.h
View File

@@ -21,9 +21,13 @@
#include <string>
#include <vector>
#define ASIO_STANDALONE 1
#include <asio.hpp>
#include "crypto/crypto.h"
#include "proto/voicecat.pb.h"
namespace voicecat { class WorkerPool; } // defined in core/worker_pool.h
namespace voicecat { class WorkerPool; }
namespace voicecat::server {
@@ -34,36 +38,42 @@ class ConnSession : public std::enable_shared_from_this<ConnSession> {
public:
enum class State { WaitingHello, WaitingAuth, Authenticated, Disconnecting };
using SendFn = std::function<void(std::vector<uint8_t>)>;
using SendFn = std::function<void(std::vector<uint8_t>)>;
using CloseFn = std::function<void()>;
ConnSession(std::shared_ptr<Database> db,
std::shared_ptr<SessionRegistry> registry,
std::shared_ptr<voicecat::WorkerPool> workers,
const std::array<uint8_t, 32>& server_fp,
bool allow_guests);
ConnSession(std::shared_ptr<Database> db,
std::shared_ptr<SessionRegistry> registry,
std::shared_ptr<voicecat::WorkerPool> workers,
const std::array<uint8_t, 32>& server_fp,
bool allow_guests,
uint16_t udp_media_port = 0);
// Called after construction: gives the session its send + close handles.
void set_io(SendFn send_fn, CloseFn close_fn);
// Called by server after it has registered the session id.
void set_session_id(uint64_t id) { session_id_ = id; }
// Entry point: send ServerHello and begin reading.
void begin();
// Deliver a received frame (called from TcpServerConn's strand).
void on_frame(std::vector<uint8_t> frame);
// Called when the TCP connection drops.
void on_disconnect();
// Thread-safe send.
void send_envelope(const voicecat::v1::Envelope& env);
// Graceful close (can be called from any thread).
void close();
// ── M2: media key injection (called from on_tls_ready) ───────────────────
void set_media_crypto(std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> send,
std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> recv);
// ── M2: UDP endpoint (set by MediaRelay on UdpBinding) ────────────────────
void set_udp_endpoint(asio::ip::udp::endpoint ep);
asio::ip::udp::endpoint udp_endpoint() const;
bool has_udp_endpoint() const { return has_udp_ep_.load(); }
// ── M2: media crypto access (for SFU relay) ──────────────────────────────
voicecat::crypto::SodiumMediaCrypto* send_crypto();
voicecat::crypto::SodiumMediaCrypto* recv_crypto();
// ── M2: UDP token (for binding) ───────────────────────────────────────────
const std::array<uint8_t, 16>& udp_token() const { return udp_token_; }
// ── Accessors ──────────────────────────────────────────────────────────────
State state() const { return state_.load(); }
uint64_t session_id() const { return session_id_; }
uint32_t user_id() const { return user_id_; }
@@ -74,25 +84,41 @@ class ConnSession : public std::enable_shared_from_this<ConnSession> {
void handle_join_channel(uint64_t req_id, const voicecat::v1::JoinChannelRequest& msg);
void handle_text_message(const voicecat::v1::TextMessage& msg);
void handle_ping(const voicecat::v1::Ping& msg);
void handle_udp_binding(uint64_t req_id, const voicecat::v1::UdpBinding& msg);
void handle_stream_announce(uint64_t req_id, const voicecat::v1::StreamAnnounce& msg);
void finish_guest_auth(const voicecat::v1::GuestAuth& guest, uint64_t req_id);
void finish_password_auth(const std::string& username, const std::string& password,
uint64_t req_id);
void send_auth_result_ok(uint64_t req_id, const voicecat::v1::User& user,
const voicecat::v1::Permissions* perms = nullptr);
void send_state_snapshot();
void broadcast_user_joined(const voicecat::v1::User& user);
void send_disconnect_and_close(uint32_t code, const std::string& reason);
std::shared_ptr<Database> db_;
std::shared_ptr<SessionRegistry> registry_;
std::shared_ptr<Database> db_;
std::shared_ptr<SessionRegistry> registry_;
std::shared_ptr<voicecat::WorkerPool> workers_;
std::array<uint8_t, 32> server_fp_;
bool allow_guests_;
std::array<uint8_t, 32> server_fp_;
bool allow_guests_;
uint16_t udp_media_port_;
SendFn send_fn_;
CloseFn close_fn_;
std::atomic<State> state_{State::WaitingHello};
uint64_t session_id_{0}; // set once before begin(), then read-only
std::atomic<uint32_t> user_id_{0};
std::atomic<bool> closed_{false};
SendFn send_fn_;
CloseFn close_fn_;
std::atomic<State> state_{State::WaitingHello};
uint64_t session_id_{0};
std::atomic<uint32_t> user_id_{0};
std::atomic<bool> closed_{false};
// M2 UDP / media
std::array<uint8_t, 16> udp_token_{};
mutable std::mutex udp_ep_mu_;
asio::ip::udp::endpoint udp_ep_;
std::atomic<bool> has_udp_ep_{false};
mutable std::mutex crypto_mu_;
std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> send_crypto_;
std::unique_ptr<voicecat::crypto::SodiumMediaCrypto> recv_crypto_;
};
} // namespace voicecat::server

129
server/src/media_relay.cpp Normal file
View File

@@ -0,0 +1,129 @@
#include "media_relay.h"
#ifdef VOICECAT_HAS_NET
#include <array>
#include <cstdio>
#include "conn_session.h"
#include "crypto/crypto.h"
#include "net/voice_frame.h"
#include "session_registry.h"
namespace voicecat::server {
MediaRelay::MediaRelay(asio::io_context& io, std::shared_ptr<SessionRegistry> registry)
: io_(io), registry_(std::move(registry)) {}
MediaRelay::~MediaRelay() { stop(); }
bool MediaRelay::bind(uint16_t port) {
return udp_.bind(io_, port);
}
void MediaRelay::start() {
udp_.start_recv([this](const uint8_t* data, size_t len, asio::ip::udp::endpoint sender) {
on_udp_frame(data, len, sender);
});
}
void MediaRelay::stop() {
udp_.close();
}
uint16_t MediaRelay::media_port() const {
return static_cast<uint16_t>(udp_.local_endpoint().port());
}
void MediaRelay::on_udp_frame(const uint8_t* data, size_t len,
asio::ip::udp::endpoint sender) {
if (len < 1) return;
const uint8_t frame_type = data[0];
if (frame_type == voicecat::net::kFrameUdpBinding) {
// Payload = 16-byte token after the 14-byte header.
if (len < voicecat::net::kVoiceHeaderSize + 16) return;
std::array<uint8_t, 16> token{};
std::memcpy(token.data(), data + voicecat::net::kVoiceHeaderSize, 16);
auto session = registry_->find_by_udp_token(token);
if (!session) return;
session->set_udp_endpoint(sender);
return;
}
if (frame_type == voicecat::net::kFrameVoice) {
if (len < voicecat::net::kVoiceHeaderSize + crypto_aead_chacha20poly1305_ietf_ABYTES) return;
// Resolve sender session.
auto sender_session = registry_->find_by_udp_endpoint(sender);
if (!sender_session) return;
auto* recv_crypto = sender_session->recv_crypto();
if (!recv_crypto) return;
// AAD = 14-byte header (authenticated, not encrypted).
const uint8_t* aad = data;
const uint8_t* sealed = data + voicecat::net::kVoiceHeaderSize;
size_t sealed_len = len - voicecat::net::kVoiceHeaderSize;
if (plain_buf_.size() < sealed_len) plain_buf_.resize(sealed_len);
long plain_len = recv_crypto->open(sealed, sealed_len, aad, voicecat::net::kVoiceHeaderSize,
plain_buf_.data(), plain_buf_.size());
if (plain_len < 0) return; // auth failure or replay
// Parse the voice frame header to find the source ssrc/channel.
voicecat::net::VoiceFrame hdr{};
if (!voicecat::net::parse_header(data, len, hdr)) return;
// Find the channel and get all other members.
uint32_t uid = sender_session->user_id();
uint32_t channel = registry_->user_channel(uid);
if (channel == 0) return;
auto members = registry_->find_channel_sessions(channel, sender_session->session_id());
// Re-encrypt and relay to each member.
for (auto& member : members) {
if (!member->has_udp_endpoint()) continue;
auto* send_crypto = member->send_crypto();
if (!send_crypto) continue;
// Build an outgoing frame with the same 14-byte header.
if (seal_buf_.size() < voicecat::net::kVoiceHeaderSize +
static_cast<size_t>(plain_len) +
crypto_aead_chacha20poly1305_ietf_ABYTES) {
seal_buf_.resize(voicecat::net::kVoiceHeaderSize +
static_cast<size_t>(plain_len) +
crypto_aead_chacha20poly1305_ietf_ABYTES);
}
// Copy header (re-use sender's header verbatim — ssrc, seq, ts pass through).
std::memcpy(seal_buf_.data(), data, voicecat::net::kVoiceHeaderSize);
uint8_t* out_payload = seal_buf_.data() + voicecat::net::kVoiceHeaderSize;
long sealed_out = send_crypto->seal(
plain_buf_.data(), static_cast<size_t>(plain_len),
seal_buf_.data(), voicecat::net::kVoiceHeaderSize,
out_payload,
static_cast<size_t>(plain_len) + crypto_aead_chacha20poly1305_ietf_ABYTES);
if (sealed_out < 0) continue;
udp_.send_to(seal_buf_.data(),
voicecat::net::kVoiceHeaderSize + static_cast<size_t>(sealed_out),
member->udp_endpoint());
}
return;
}
// kFrameKeepalive or unknown: silently discard.
}
} // namespace voicecat::server
#endif // VOICECAT_HAS_NET

65
server/src/media_relay.h Normal file
View File

@@ -0,0 +1,65 @@
/*
* server/media_relay.h — UDP SFU relay for M2 voice.
*
* Design: docs/architecture.md §5, docs/voice.md §2.
* Receives encrypted UDP voice frames from clients, decrypts+authenticates them,
* re-encrypts for each channel member, and forwards.
*
* Flow:
* 1. Client sends kFrameUdpBinding UDP packet (plaintext) → MediaRelay looks up
* the 16-byte token in SessionRegistry, associates the sender endpoint with
* the ConnSession, and calls session->set_udp_endpoint().
* 2. Client sends kFrameVoice UDP packets → MediaRelay decrypts via recv_crypto(),
* finds channel members via find_channel_sessions(), re-encrypts via send_crypto(),
* and sends to each member's UDP endpoint.
*/
#ifndef VOICECAT_SERVER_MEDIA_RELAY_H
#define VOICECAT_SERVER_MEDIA_RELAY_H
#ifdef VOICECAT_HAS_NET
#include <memory>
#define ASIO_STANDALONE 1
#include <asio.hpp>
#include "net/transport.h"
namespace voicecat::server {
class SessionRegistry;
class MediaRelay {
public:
MediaRelay(asio::io_context& io, std::shared_ptr<SessionRegistry> registry);
~MediaRelay();
// Bind the UDP socket. port=0 lets the OS pick. Must be called before start().
bool bind(uint16_t port = 0);
// Begin async receive loop. Call once after bind().
void start();
// Stop receiving and close the socket.
void stop();
// Actual bound port (after bind()).
uint16_t media_port() const;
private:
void on_udp_frame(const uint8_t* data, size_t len, asio::ip::udp::endpoint sender);
asio::io_context& io_;
std::shared_ptr<SessionRegistry> registry_;
voicecat::net::UdpMediaChannel udp_;
// Scratch buffer for re-encrypted payloads (size = max_frame + 16 MAC)
static constexpr size_t kMaxPayload = 1500;
std::vector<uint8_t> seal_buf_ = std::vector<uint8_t>(kMaxPayload + 16, uint8_t{0});
std::vector<uint8_t> plain_buf_ = std::vector<uint8_t>(kMaxPayload, uint8_t{0});
};
} // namespace voicecat::server
#endif // VOICECAT_HAS_NET
#endif // VOICECAT_SERVER_MEDIA_RELAY_H

25
server/src/server.cpp
View File

@@ -15,6 +15,7 @@
#include "crypto/crypto.h"
#include "db.h"
#include "identity.h"
#include "media_relay.h"
#include "net/transport.h"
#include "session_registry.h"
@@ -62,6 +63,16 @@ int Server::run() {
// ── Asio io_context ──────────────────────────────────────────────────────
asio::io_context io;
// ── UDP media relay (M2) ─────────────────────────────────────────────────
auto media_relay = std::make_shared<MediaRelay>(io, registry);
if (!media_relay->bind(cfg_.media_port)) {
std::fprintf(stderr, "[server] failed to bind UDP media port %u\n", cfg_.media_port);
return 1;
}
media_relay->start();
uint16_t media_bound = media_relay->media_port();
if (cfg_.on_media_ready) cfg_.on_media_ready(media_bound);
// Capture all locals by reference for the factory lambda (io lifetime is > factory)
voicecat::net::TcpAcceptor acceptor(
io, cfg_.bind_port,
@@ -69,7 +80,8 @@ int Server::run() {
auto session = std::make_shared<ConnSession>(
db, registry, workers,
id_mgr.identity().fingerprint,
cfg_.allow_guests);
cfg_.allow_guests,
media_bound);
// Use shared_ptr (not weak_ptr) so TcpServerConn keeps ConnSession alive.
// cycle is broken by weak_tcp in the send/close fns below.
@@ -83,6 +95,12 @@ int Server::run() {
cbs.on_error = [session](std::error_code) {
session->on_disconnect();
};
// Derive media keys right after TLS handshake (server is not the client).
cbs.on_tls_ready = [session](voicecat::crypto::TlsContext& tls) {
auto send = voicecat::crypto::SodiumMediaCrypto::derive_send(tls, false);
auto recv = voicecat::crypto::SodiumMediaCrypto::derive_recv(tls, false);
if (send && recv) session->set_media_crypto(std::move(send), std::move(recv));
};
// Create a TLS context for this connection (server role).
auto tls = std::make_unique<voicecat::crypto::TlsContext>(
@@ -126,11 +144,12 @@ int Server::run() {
signals.async_wait([&](std::error_code, int sig) {
std::printf("\n[server] signal %d — shutting down\n", sig);
acceptor.stop();
media_relay->stop();
io.stop();
});
std::printf("[voicecat-server] %s — listening on :%u\n",
cfg_.server_name.c_str(), bound);
std::printf("[voicecat-server] %s — TCP :%u UDP :%u\n",
cfg_.server_name.c_str(), bound, media_bound);
std::printf("[voicecat-server] fingerprint: %s\n",
id_mgr.fingerprint_display().c_str());

11
server/src/server.h
View File

@@ -15,12 +15,15 @@
namespace voicecat::server {
struct Config {
std::string server_name = "VoiceCat Server";
std::string data_dir = "voicecat-data";
uint16_t bind_port = 8384; // 0 = let OS pick (useful for tests)
std::string server_name = "VoiceCat Server";
std::string data_dir = "voicecat-data";
uint16_t bind_port = 8384; // 0 = let OS pick (useful for tests)
uint16_t media_port = 0; // M2 UDP media port; 0 = OS-assigned
bool allow_guests = true;
// Called with the actual bound port once the acceptor is ready (m1-dev only).
// Called with the actual bound TCP port once the acceptor is ready.
std::function<void(uint16_t)> on_ready;
// Called with the actual bound UDP media port once the relay is ready.
std::function<void(uint16_t)> on_media_ready;
};
class Server {

62
server/src/session_registry.cpp
View File

@@ -2,6 +2,7 @@
#ifdef VOICECAT_HAS_NET
#include <atomic>
#include <mutex>
#include <shared_mutex>
@@ -110,6 +111,67 @@ void SessionRegistry::broadcast(const voicecat::v1::Envelope& env,
}
}
// ── M2: UDP / media ──────────────────────────────────────────────────────────
void SessionRegistry::register_udp_token(const std::array<uint8_t, 16>& token,
uint64_t session_id) {
std::unique_lock lk(mu_);
udp_tokens_[token] = session_id;
}
std::shared_ptr<ConnSession> SessionRegistry::find_by_udp_token(
const std::array<uint8_t, 16>& token) const {
std::shared_lock lk(mu_);
auto it = udp_tokens_.find(token);
if (it == udp_tokens_.end()) return nullptr;
auto sit = sessions_.find(it->second);
if (sit == sessions_.end()) return nullptr;
return sit->second.lock();
}
void SessionRegistry::register_udp_endpoint(asio::ip::udp::endpoint ep,
uint64_t session_id) {
std::unique_lock lk(mu_);
udp_endpoints_[ep] = session_id;
}
std::shared_ptr<ConnSession> SessionRegistry::find_by_udp_endpoint(
const asio::ip::udp::endpoint& ep) const {
std::shared_lock lk(mu_);
auto it = udp_endpoints_.find(ep);
if (it == udp_endpoints_.end()) return nullptr;
auto sit = sessions_.find(it->second);
if (sit == sessions_.end()) return nullptr;
return sit->second.lock();
}
uint32_t SessionRegistry::assign_ssrc(uint64_t session_id) {
uint32_t ssrc = next_ssrc_.fetch_add(1, std::memory_order_relaxed);
std::unique_lock lk(mu_);
ssrc_to_session_[ssrc] = session_id;
return ssrc;
}
std::vector<std::shared_ptr<ConnSession>> SessionRegistry::find_channel_sessions(
uint32_t channel_id, uint64_t exclude_session_id) const {
std::shared_lock lk(mu_);
std::vector<std::shared_ptr<ConnSession>> result;
for (auto& [uid, entry] : users_) {
if (entry.proto.channel_id() != channel_id) continue;
if (entry.session_id == exclude_session_id) continue;
auto sit = sessions_.find(entry.session_id);
if (sit == sessions_.end()) continue;
if (auto sess = sit->second.lock()) result.push_back(sess);
}
return result;
}
uint32_t SessionRegistry::user_channel(uint32_t user_id) const {
std::shared_lock lk(mu_);
auto it = users_.find(user_id);
return (it == users_.end()) ? 0 : it->second.proto.channel_id();
}
} // namespace voicecat::server
#endif // VOICECAT_HAS_NET

60
server/src/session_registry.h
View File

@@ -1,7 +1,8 @@
/*
* server/session_registry.h — In-memory session, channel, and user registry.
*
* Tracks all authenticated sessions, the channel tree, and user<→>channel assignments.
* Tracks all authenticated sessions, the channel tree, user<→>channel assignments,
* UDP endpoint bindings (M2), and SSRC<→>session mappings (M2).
* Protected by a shared_mutex (many readers, few writers). All methods are thread-safe.
*/
#ifndef VOICECAT_SERVER_SESSION_REGISTRY_H
@@ -9,6 +10,8 @@
#ifdef VOICECAT_HAS_NET
#include <array>
#include <atomic>
#include <cstdint>
#include <memory>
#include <shared_mutex>
@@ -16,6 +19,9 @@
#include <unordered_map>
#include <vector>
#define ASIO_STANDALONE 1
#include <asio.hpp>
#include "proto/voicecat.pb.h"
namespace voicecat::server {
@@ -31,6 +37,14 @@ struct UserEntry {
uint64_t session_id{};
};
// Hashes asio::ip::udp::endpoint by "addr:port" string.
struct UdpEndpointHash {
size_t operator()(const asio::ip::udp::endpoint& ep) const {
std::string key = ep.address().to_string() + ':' + std::to_string(ep.port());
return std::hash<std::string>{}(key);
}
};
class SessionRegistry {
public:
SessionRegistry() = default;
@@ -58,14 +72,36 @@ class SessionRegistry {
std::vector<voicecat::v1::User> user_snapshot() const;
// Resolve target sessions for a text message relay.
// TEXT_CHANNEL: all users in that channel (except sender's session).
// TEXT_PRIVATE: the session for that user_id.
std::vector<std::shared_ptr<ConnSession>> resolve_text_targets(
uint64_t sender_session_id, voicecat::v1::TextScope scope, uint32_t target_id) const;
// Broadcast an envelope to all sessions except the excluded one.
void broadcast(const voicecat::v1::Envelope& env, uint64_t exclude_session_id = 0) const;
// ── M2: UDP / media ──────────────────────────────────────────────────────
// Register a session's UDP token (called at auth success).
void register_udp_token(const std::array<uint8_t, 16>& token, uint64_t session_id);
// Locate a session by its UDP binding token (called by MediaRelay on UDP_BINDING).
std::shared_ptr<ConnSession> find_by_udp_token(const std::array<uint8_t, 16>& token) const;
// Associate a UDP endpoint with a session (called by MediaRelay after token verification).
void register_udp_endpoint(asio::ip::udp::endpoint ep, uint64_t session_id);
// Locate the session that owns a UDP sender endpoint (called per incoming voice packet).
std::shared_ptr<ConnSession> find_by_udp_endpoint(const asio::ip::udp::endpoint& ep) const;
// Assign an SSRC for a new stream. Returns the assigned SSRC.
uint32_t assign_ssrc(uint64_t session_id);
// Get all sessions in a channel except the one excluded (for SFU relay).
std::vector<std::shared_ptr<ConnSession>> find_channel_sessions(
uint32_t channel_id, uint64_t exclude_session_id = 0) const;
// Return the channel_id of a user (0 if not found).
uint32_t user_channel(uint32_t user_id) const;
private:
mutable std::shared_mutex mu_;
@@ -76,6 +112,24 @@ class SessionRegistry {
std::unordered_map<uint64_t, std::weak_ptr<ConnSession>> sessions_;
std::unordered_map<uint32_t, UserEntry> users_;
std::unordered_map<uint32_t, ChannelEntry> channels_;
// M2: token → session_id (populated at auth, cleared on disconnect)
struct TokenHash {
size_t operator()(const std::array<uint8_t, 16>& t) const {
// FNV-1a over 16 bytes
size_t h = 14695981039346656037ULL;
for (auto b : t) { h ^= b; h *= 1099511628211ULL; }
return h;
}
};
std::unordered_map<std::array<uint8_t, 16>, uint64_t, TokenHash> udp_tokens_;
// M2: UDP endpoint → session_id (populated after UDP binding packet arrives)
std::unordered_map<asio::ip::udp::endpoint, uint64_t, UdpEndpointHash> udp_endpoints_;
// M2: ssrc → session_id (populated when StreamAnnounce is processed)
std::unordered_map<uint32_t, uint64_t> ssrc_to_session_;
std::atomic<uint32_t> next_ssrc_{1};
};
} // namespace voicecat::server