voice-cat/tests/test_m3_multistream.cpp

/*
 * test_m3_multistream — M3 exit criterion, exercised through the real C ABI.
 *
 * Mirrors test_voice_client_abi.cpp's approach (real vc_client instances, not raw sockets —
 * the M2 lesson is that ABI-level coverage is what actually proves the client library works).
 * Covers the whole M3 milestone in one flow:
 *
 *   1. A starts two concurrent local streams (MIC + SCREEN_AUDIO) -- distinct stream ids,
 *      both visible to B as separate STREAM_STARTED events for the same user.
 *   2. Synthetic PCM (vc_test_inject_capture) flows into both of A's streams without crashing
 *      and without disrupting the control/voice plane; B observes a VC_EVENT_TALK_STATE
 *      talking=true edge for A's MIC stream while both are still in the same channel (voice
 *      only relays within a channel, so this must happen before step 4 moves A elsewhere).
 *   3. B independently gains/mutes/NS-toggles A's two streams (vc_set_remote_stream) --
 *      one call doesn't clobber the other's routing; a bogus stream_id is rejected.
 *   4. Per-channel Opus configurability: A joins "Music Room" (channel 2, stereo/128kbps/
 *      OPUS_AUDIO/no DTX) before announcing there, while B stays in "Lobby" (channel 1,
 *      mono/24kbps/OPUS_VOIP/DTX) -- vc_get_stream_audio_config shows the two streams'
 *      effective config differs exactly as the server enforces it.
 */
#include <cstdio>

#include <atomic>
#include <chrono>
#include <cmath>
#include <condition_variable>
#include <filesystem>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include "voicecat.h"
#include "server.h"
#include "db.h"

// ── Event tracking ────────────────────────────────────────────────────────────

struct StreamEvent {
    bool     started;  // true = STARTED, false = STOPPED
    uint32_t user_id;
    uint32_t stream_id;
};

struct TalkEvent {
    uint32_t user_id;
    uint32_t stream_id;
    bool     talking;
};

struct EventStore {
    std::mutex              mu;
    std::condition_variable cv;

    bool      auth_ok{false};
    uint32_t  self_user_id{0};
    bool      channel_list_received{false};
    std::vector<StreamEvent> stream_events;
    std::vector<TalkEvent>   talk_events;
    bool      voice_subscribed{false};
    bool      disconnected{false};

    const char* label{nullptr};

    // Set right after vc_client_create, before vc_connect — lets on_event auto-confirm the
    // M4 TOFU gate (VC_EVENT_SERVER_IDENTITY below) for this headless test.
    vc_client* client{nullptr};
};

static void on_event(void* user, const vc_event* ev) {
    auto* s = static_cast<EventStore*>(user);
    std::lock_guard lk(s->mu);
    switch (ev->type) {
        case VC_EVENT_SERVER_IDENTITY:
            // No human to ask in a headless test — trust on first connect unconditionally.
            vc_confirm_server_identity(s->client, 1);
            break;
        case VC_EVENT_AUTH_RESULT:
            s->auth_ok      = (ev->result == VC_OK);
            s->self_user_id = ev->user_id;
            if (!s->auth_ok) std::fprintf(stderr, "[%s] AUTH FAILED: %s\n",
                s->label ? s->label : "?", ev->text ? ev->text : "(no msg)");
            break;
        case VC_EVENT_CHANNEL_LIST:
            s->channel_list_received = true;
            break;
        case VC_EVENT_VOICE_STATE:
            s->voice_subscribed = (ev->u32a == 1);
            break;
        case VC_EVENT_STREAM_STARTED:
            s->stream_events.push_back({true, ev->user_id, ev->stream_id});
            break;
        case VC_EVENT_STREAM_STOPPED:
            s->stream_events.push_back({false, ev->user_id, ev->stream_id});
            break;
        case VC_EVENT_TALK_STATE:
            s->talk_events.push_back({ev->user_id, ev->stream_id, ev->u32a != 0});
            break;
        case VC_EVENT_ERROR:
            std::fprintf(stderr, "[%s] ERROR rc=%d: %s\n",
                s->label ? s->label : "?", ev->result, ev->text ? ev->text : "");
            break;
        case VC_EVENT_DISCONNECTED:
            s->disconnected = true;
            break;
        default:
            break;
    }
    s->cv.notify_all();
}

template <typename Pred>
static bool wait_for(EventStore& s, Pred pred, int timeout_ms) {
    auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);
    std::unique_lock lk(s.mu);
    return s.cv.wait_until(lk, deadline, [&] { return pred(s); });
}

static std::vector<int16_t> make_sine_frame(int frame_idx, float freq_hz,
                                            int frame_samples = 960) {
    std::vector<int16_t> pcm(frame_samples);
    for (int i = 0; i < frame_samples; ++i) {
        float t = static_cast<float>(frame_idx * frame_samples + i) / 48000.0f;
        pcm[i] = static_cast<int16_t>(std::sin(2.0f * 3.14159265f * freq_hz * t) * 16000.0f);
    }
    return pcm;
}

// ── Test harness ──────────────────────────────────────────────────────────────

static int g_failures = 0;
#define CHECK(cond)                                                        \
    do {                                                                   \
        if (!(cond)) {                                                     \
            std::printf("FAIL: %s (%s:%d)\n", #cond, __FILE__, __LINE__); \
            ++g_failures;                                                  \
        }                                                                  \
    } while (0)

int main() {
    auto tmp = std::filesystem::temp_directory_path() /
               ("vctest_m3_" + std::to_string(
                   std::chrono::steady_clock::now().time_since_epoch().count()));
    std::filesystem::create_directories(tmp);
    std::string data_dir = tmp.string();

    std::atomic<uint16_t> bound_port{0};
    std::mutex              ready_mu;
    std::condition_variable ready_cv;
    bool                    ready{false};

    voicecat::server::Config cfg;
    cfg.data_dir     = data_dir;
    cfg.bind_port    = 0;
    cfg.media_port   = 0;
    cfg.server_name  = "VoiceCat-M3Test";
    cfg.allow_guests = true;
    cfg.on_ready = [&](uint16_t p) {
        bound_port.store(p);
        { std::lock_guard lk(ready_mu); ready = true; }
        ready_cv.notify_all();
    };

    voicecat::server::Server server(cfg);
    std::thread server_thread([&] { server.run(); });

    {
        std::unique_lock lk(ready_mu);
        bool ok = ready_cv.wait_for(lk, std::chrono::seconds(10), [&] { return ready; });
        if (!ok) {
            std::printf("FAIL: server did not become ready within 10s\n");
            server.stop();
            server_thread.join();
            std::filesystem::remove_all(tmp);
            return 1;
        }
    }

    uint16_t port = bound_port.load();
    std::printf("m3_multistream: server ready on :%u\n", port);

    // ── Client A: guest "M3-A" ────────────────────────────────────────────────
    EventStore evA;
    evA.label = "clientA";
    vc_callbacks cbA{on_event, nullptr, &evA};
    vc_config cfgA{"test-clientA", "0.1", VC_LOG_OFF};
    vc_client* clientA = vc_client_create(&cfgA, cbA);
    CHECK(clientA != nullptr);
    evA.client = clientA;

    CHECK(vc_connect(clientA, "127.0.0.1", port) == VC_OK);
    CHECK(vc_authenticate_guest(clientA, "M3-A") == VC_OK);
    CHECK(wait_for(evA, [](EventStore& s) { return s.auth_ok; }, 8000));
    CHECK(wait_for(evA, [](EventStore& s) { return s.channel_list_received; }, 3000));

    CHECK(vc_join_voice(clientA) == VC_OK);
    CHECK(wait_for(evA, [](EventStore& s) { return s.voice_subscribed; }, 5000));

    // ── Client B: guest "M3-B" ────────────────────────────────────────────────
    EventStore evB;
    evB.label = "clientB";
    vc_callbacks cbB{on_event, nullptr, &evB};
    vc_config cfgB{"test-clientB", "0.1", VC_LOG_OFF};
    vc_client* clientB = vc_client_create(&cfgB, cbB);
    CHECK(clientB != nullptr);
    evB.client = clientB;

    CHECK(vc_connect(clientB, "127.0.0.1", port) == VC_OK);
    CHECK(vc_authenticate_guest(clientB, "M3-B") == VC_OK);
    CHECK(wait_for(evB, [](EventStore& s) { return s.auth_ok; }, 8000));
    CHECK(wait_for(evB, [](EventStore& s) { return s.channel_list_received; }, 3000));

    CHECK(vc_join_voice(clientB) == VC_OK);
    CHECK(wait_for(evB, [](EventStore& s) { return s.voice_subscribed; }, 5000));

    uint32_t a_uid = 0;
    { std::lock_guard lk(evA.mu); a_uid = evA.self_user_id; }

    // Both guests land in channel 1 (Lobby) automatically; give the async UDP binding
    // handshake a moment to complete on both clients before announcing streams.
    std::this_thread::sleep_for(std::chrono::milliseconds(500));

    // ── 1. A starts MIC + SCREEN_AUDIO concurrently ──────────────────────────
    vc_stream_desc mic_desc{};
    mic_desc.kind  = VC_STREAM_MIC;
    mic_desc.label = "mic";
    uint32_t mic_sid = 0;
    CHECK(vc_stream_start(clientA, &mic_desc, &mic_sid) == VC_OK);

    vc_stream_desc screen_desc{};
    screen_desc.kind  = VC_STREAM_SCREEN_AUDIO;
    screen_desc.label = "desktop audio";
    uint32_t screen_sid = 0;
    CHECK(vc_stream_start(clientA, &screen_desc, &screen_sid) == VC_OK);

    CHECK(mic_sid != 0 && screen_sid != 0 && mic_sid != screen_sid);

    // B observes two distinct STREAM_STARTED events for user A.
    bool b_saw_both = wait_for(evB, [&](EventStore& s) {
        bool saw_mic = false, saw_screen = false;
        for (auto& e : s.stream_events) {
            if (!e.started || e.user_id != a_uid) continue;
            if (e.stream_id == mic_sid) saw_mic = true;
            if (e.stream_id == screen_sid) saw_screen = true;
        }
        return saw_mic && saw_screen;
    }, 5000);
    CHECK(b_saw_both);

    // Also wait for A's own view of both streams (vc_test_inject_capture requires the
    // LocalStream to be active, which flips on A's io_thread_ independently of -- and not
    // necessarily before -- the broadcast B observes above).
    bool a_self_saw_both = wait_for(evA, [&](EventStore& s) {
        bool saw_mic = false, saw_screen = false;
        for (auto& e : s.stream_events) {
            if (!e.started || e.user_id != a_uid) continue;
            if (e.stream_id == mic_sid) saw_mic = true;
            if (e.stream_id == screen_sid) saw_screen = true;
        }
        return saw_mic && saw_screen;
    }, 5000);
    CHECK(a_self_saw_both);

    // ── 2. Inject synthetic PCM into both of A's local streams ──────────────
    for (int i = 0; i < 25; ++i) {
        auto mic_pcm = make_sine_frame(i, 440.0f);
        auto screen_pcm = make_sine_frame(i, 880.0f);
        CHECK(vc_test_inject_capture(clientA, mic_sid, mic_pcm.data(), mic_pcm.size()) == VC_OK);
        CHECK(vc_test_inject_capture(clientA, screen_sid, screen_pcm.data(), screen_pcm.size()) == VC_OK);
        std::this_thread::sleep_for(std::chrono::milliseconds(20));
    }

    // No disconnects/errors should have resulted from the dual-stream PCM flow.
    { std::lock_guard lk(evA.mu); CHECK(!evA.disconnected); }
    { std::lock_guard lk(evB.mu); CHECK(!evB.disconnected); }

    // ── 5. Talk indicators ────────────────────────────────────────────────────
    // While A and B are still both in Lobby (voice actually relays between them here --
    // the SFU forwards within a channel, so this must happen before A moves to Music Room
    // in step 4 below), confirm B observed a talking=true edge for A's MIC stream.
    bool b_saw_talking = wait_for(evB, [&](EventStore& s) {
        for (auto& e : s.talk_events)
            if (e.user_id == a_uid && e.stream_id == mic_sid && e.talking) return true;
        return false;
    }, 3000);
    CHECK(b_saw_talking);

    // ── 3. B independently controls gain/mute/NS on each of A's streams ─────
    CHECK(vc_set_remote_stream(clientB, a_uid, mic_sid, 1.0f, 0, 0) == VC_OK);
    CHECK(vc_set_remote_stream(clientB, a_uid, screen_sid, 0.3f, 1, 1) == VC_OK);
    CHECK(vc_set_remote_stream(clientB, a_uid, 0xDEADBEEF, 1.0f, 0, 0) == VC_ERR_INVALID_ARG);

    // Toggle NS on/off a few times -- plumbing should never fault or disrupt the stream.
    for (int i = 0; i < 3; ++i) {
        CHECK(vc_set_remote_stream(clientB, a_uid, mic_sid, 1.0f, 0, 1) == VC_OK);
        CHECK(vc_set_remote_stream(clientB, a_uid, mic_sid, 1.0f, 0, 0) == VC_OK);
    }
    { std::lock_guard lk(evB.mu); CHECK(!evB.disconnected); }

    // ── 3b. Read back what B just set (vc_get_remote_stream round-trips the recv state) ─
    {
        vc_remote_stream_state st{};
        // mic_sid: last write above was (1.0, mute=0, nr=0)
        CHECK(vc_get_remote_stream(clientB, a_uid, mic_sid, &st) == VC_OK);
        CHECK(fabsf(st.gain - 1.0f) < 1e-5f);
        CHECK(st.muted == 0);
        CHECK(st.noise_reduction == 0);

        // screen_sid: set to (0.3, mute=1, nr=1) at line ~282
        CHECK(vc_get_remote_stream(clientB, a_uid, screen_sid, &st) == VC_OK);
        CHECK(fabsf(st.gain - 0.3f) < 1e-5f);
        CHECK(st.muted == 1);
        CHECK(st.noise_reduction == 1);

        // Unknown stream_id on a known user -> INVALID_ARG.
        CHECK(vc_get_remote_stream(clientB, a_uid, 0xDEADBEEF, &st) == VC_ERR_INVALID_ARG);
        // Null out -> INVALID_ARG.
        CHECK(vc_get_remote_stream(clientB, a_uid, mic_sid, nullptr) == VC_ERR_INVALID_ARG);
    }


    // ── 4. Per-channel Opus configurability ──────────────────────────────────
    // A moves to "Music Room" (channel 2: stereo/128kbps/OPUS_AUDIO/no DTX) and announces a
    // fresh MIC stream there; B stays in "Lobby" (channel 1: mono/24kbps/OPUS_VOIP/DTX) with
    // its own MIC stream. Their effective_audio should differ exactly as configured server-side.
    CHECK(vc_stream_stop(clientA, mic_sid) == VC_OK);
    CHECK(vc_join_channel(clientA, 2, nullptr) == VC_OK);
    std::this_thread::sleep_for(std::chrono::milliseconds(300));

    vc_stream_desc music_mic_desc{};
    music_mic_desc.kind  = VC_STREAM_MIC;
    music_mic_desc.label = "music-mic";
    uint32_t a_music_mic_sid = 0;
    CHECK(vc_stream_start(clientA, &music_mic_desc, &a_music_mic_sid) == VC_OK);
    CHECK(wait_for(evA, [&](EventStore& s) {
        for (auto& e : s.stream_events)
            if (e.started && e.user_id == a_uid && e.stream_id == a_music_mic_sid) return true;
        return false;
    }, 5000));

    vc_stream_desc b_mic_desc{};
    b_mic_desc.kind  = VC_STREAM_MIC;
    b_mic_desc.label = "lobby-mic";
    uint32_t b_mic_sid = 0;
    CHECK(vc_stream_start(clientB, &b_mic_desc, &b_mic_sid) == VC_OK);
    CHECK(wait_for(evB, [&](EventStore& s) {
        uint32_t self = s.self_user_id;
        for (auto& e : s.stream_events)
            if (e.started && e.user_id == self && e.stream_id == b_mic_sid) return true;
        return false;
    }, 5000));

    vc_audio_config a_cfg{};
    vc_audio_config b_cfg{};
    CHECK(vc_get_stream_audio_config(clientA, a_uid, a_music_mic_sid, &a_cfg) == VC_OK);
    uint32_t b_uid = 0;
    { std::lock_guard lk(evB.mu); b_uid = evB.self_user_id; }
    CHECK(vc_get_stream_audio_config(clientB, b_uid, b_mic_sid, &b_cfg) == VC_OK);

    // Music Room: stereo, 128kbps, OPUS_AUDIO, DTX off. Lobby: mono, 24kbps, OPUS_VOIP, DTX on.
    CHECK(a_cfg.mode == 1 /* stereo */);
    CHECK(b_cfg.mode == 0 /* mono */);
    CHECK(a_cfg.bitrate_bps == 128000);
    CHECK(b_cfg.bitrate_bps == 24000);
    CHECK(a_cfg.application == 1 /* OPUS_AUDIO */);
    CHECK(b_cfg.application == 0 /* OPUS_VOIP */);
    CHECK(a_cfg.dtx == 0);
    CHECK(b_cfg.dtx != 0);

    // ── Cleanup ───────────────────────────────────────────────────────────────
    vc_disconnect(clientA);
    vc_disconnect(clientB);
    vc_client_destroy(clientA);
    vc_client_destroy(clientB);

    server.stop();
    server_thread.join();

    std::filesystem::remove_all(tmp);

    if (g_failures == 0) {
        std::printf("m3_multistream: all checks passed\n");
        return 0;
    }
    std::printf("m3_multistream: %d failure(s)\n", g_failures);
    return 1;
}