/*
streamer-tools OBS Camera Plugin - LiveKit end-to-end integration test
Copyright (C) 2026 CyberCoveLLC <jknapp85@gmail.com>

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0
*/

// The one test that proves the session wrapper actually receives media.
//
// It publishes a synthetic camera into a real LiveKit room using the same
// SDK, subscribes to it through stplugin::LiveKitSession, and asserts that
// decoded frames arrive with the geometry the OBS adapter is going to hand
// to obs_source_output_video. It also drives the publisher-swap sequence
// (unpublish, republish) that motivated this whole plugin, and asserts the
// wrapper recovers instead of going stale or erroring out.
//
// It needs a reachable LiveKit server, so it SKIPS (exit 0) unless these are
// set -- CI on the three build runners has no server, and this must not turn
// into a red build there:
//
//   STPLUGIN_IT_URL                ws://127.0.0.1:7880
//   STPLUGIN_IT_PUBLISH_TOKEN      JWT: roomJoin + canPublish for the room
//   STPLUGIN_IT_SUBSCRIBE_TOKEN    JWT: roomJoin + canSubscribe for the room
//   STPLUGIN_IT_PUBLISHER_IDENTITY the identity in the publish token
//
// scripts/livekit-dev-room.py mints all four against a `livekit-server --dev`.

#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <functional>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include <livekit/audio_frame.h>
#include <livekit/audio_source.h>
#include <livekit/livekit.h>
#include <livekit/local_audio_track.h>
#include <livekit/local_participant.h>
#include <livekit/local_track_publication.h>
#include <livekit/local_video_track.h>
#include <livekit/room.h>
#include <livekit/video_frame.h>
#include <livekit/video_source.h>

#include "stplugin/session.h"
#include "test_util.h"

using namespace stplugin;

namespace {

constexpr int kWidth = 320;
constexpr int kHeight = 240;

std::string envOrEmpty(const char *name)
{
    const char *value = std::getenv(name);
    return value ? std::string(value) : std::string();
}

/// A moving horizontal band, so a frozen or stale frame is distinguishable
/// from a live one by luma alone.
livekit::VideoFrame makeFrame(int tick)
{
    livekit::VideoFrame frame = livekit::VideoFrame::create(kWidth, kHeight, livekit::VideoBufferType::I420);
    std::uint8_t *data = frame.data();
    const std::size_t luma = static_cast<std::size_t>(kWidth) * kHeight;
    std::memset(data, 16, luma);
    const int band = (tick * 7) % kHeight;
    std::memset(data + static_cast<std::size_t>(band) * kWidth, 235, kWidth);
    std::memset(data + luma, 128, frame.dataSize() - luma);
    return frame;
}

void step(const char *what)
{
    std::printf("  step: %s\n", what);
    std::fflush(stdout);
}

bool waitFor(const std::function<bool()> &predicate, int timeout_ms)
{
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);
    while (std::chrono::steady_clock::now() < deadline) {
        if (predicate())
            return true;
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
    return predicate();
}

struct Publisher {
    livekit::Room room;
    std::shared_ptr<livekit::VideoSource> video_source;
    std::shared_ptr<livekit::LocalVideoTrack> video_track;
    std::shared_ptr<livekit::AudioSource> audio_source;
    std::shared_ptr<livekit::LocalAudioTrack> audio_track;
    std::thread pump;
    std::atomic<bool> stop{false};
    std::string video_sid;

    bool connect(const std::string &url, const std::string &token)
    {
        livekit::RoomOptions options;
        options.auto_subscribe = false;
        options.connect_timeout = std::chrono::milliseconds(10000);
        return room.connect(url, token, options);
    }

    bool publishVideo()
    {
        auto local = room.localParticipant().lock();
        if (!local)
            return false;
        video_source = std::make_shared<livekit::VideoSource>(kWidth, kHeight);
        video_track = livekit::LocalVideoTrack::createLocalVideoTrack("camera", video_source);
        livekit::TrackPublishOptions options;
        options.source = livekit::TrackSource::SOURCE_CAMERA;
        options.simulcast = false;
        local->publishTrack(video_track, options);
        // publishTrack is async server-side; the SID appears on the track once
        // the publication lands.
        for (int i = 0; i < 100 && video_track->sid().empty(); ++i)
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        video_sid = video_track->sid();
        return !video_sid.empty();
    }

    bool publishAudio()
    {
        auto local = room.localParticipant().lock();
        if (!local)
            return false;
        audio_source = std::make_shared<livekit::AudioSource>(48000, 1);
        audio_track = livekit::LocalAudioTrack::createLocalAudioTrack("microphone", audio_source);
        livekit::TrackPublishOptions options;
        options.source = livekit::TrackSource::SOURCE_MICROPHONE;
        local->publishTrack(audio_track, options);
        for (int i = 0; i < 100 && audio_track->sid().empty(); ++i)
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        return !audio_track->sid().empty();
    }

    bool unpublishVideo()
    {
        auto local = room.localParticipant().lock();
        if (!local)
            return false;
        // The FFI keys local publications by the *publication* SID, which is
        // not necessarily the track SID -- unpublishing by track SID throws
        // "track not found".
        std::string sid = video_sid;
        if (video_track && video_track->publication())
            sid = video_track->publication()->sid();
        if (sid.empty())
            return false;
        try {
            local->unpublishTrack(sid);
        } catch (const std::exception &e) {
            std::printf("  unpublishTrack(%s) threw: %s\n", sid.c_str(), e.what());
            std::fflush(stdout);
            return false;
        }
        video_sid.clear();
        video_track.reset();
        video_source.reset();
        return true;
    }

    void startPump()
    {
        stop.store(false);
        pump = std::thread([this] {
            int tick = 0;
            std::vector<std::int16_t> pcm(480, 0); // 10ms of 48kHz mono
            while (!stop.load()) {
                if (video_source) {
                    livekit::VideoFrame frame = makeFrame(tick);
                    video_source->captureFrame(frame, static_cast<std::int64_t>(tick) * 33333);
                }
                if (audio_source) {
                    for (std::size_t i = 0; i < pcm.size(); ++i)
                        pcm[i] = static_cast<std::int16_t>(((tick * 480 + static_cast<int>(i)) % 100) * 100);
                    livekit::AudioFrame audio(pcm, 48000, 1, static_cast<int>(pcm.size()));
                    audio_source->captureFrame(audio);
                }
                ++tick;
                std::this_thread::sleep_for(std::chrono::milliseconds(33));
            }
        });
    }

    void stopPump()
    {
        stop.store(true);
        if (pump.joinable())
            pump.join();
    }

    ~Publisher()
    {
        stopPump();
        room.disconnect(livekit::DisconnectReason::ClientInitiated);
    }
};

} // namespace

int main()
{
    const std::string url = envOrEmpty("STPLUGIN_IT_URL");
    const std::string publish_token = envOrEmpty("STPLUGIN_IT_PUBLISH_TOKEN");
    const std::string subscribe_token = envOrEmpty("STPLUGIN_IT_SUBSCRIBE_TOKEN");
    const std::string publisher_identity = envOrEmpty("STPLUGIN_IT_PUBLISHER_IDENTITY");

    if (url.empty() || publish_token.empty() || subscribe_token.empty() || publisher_identity.empty()) {
        std::printf("integration_livekit: SKIPPED (STPLUGIN_IT_* not set)\n");
        return 0;
    }

    LiveKitSession::globalInitialize();

    auto publisher_holder = std::make_unique<Publisher>();
    Publisher &publisher = *publisher_holder;
    step("publisher connect");
    ST_ASSERT(publisher.connect(url, publish_token));
    step("publish video");
    ST_ASSERT(publisher.publishVideo());
    step("publish audio");
    ST_ASSERT(publisher.publishAudio());
    publisher.startPump();

    // --- subscribe through the wrapper under test --------------------------

    std::atomic<int> video_frames{0};
    std::atomic<int> audio_frames{0};
    std::atomic<int> bad_frames{0};
    std::atomic<int> full_size_frames{0};
    std::atomic<int> last_width{0};
    std::atomic<int> last_height{0};
    std::atomic<int> last_planes{0};
    std::atomic<long long> last_timestamp{0};
    std::atomic<int> last_sample_rate{0};
    std::atomic<int> last_channels{0};

    LiveKitSession session;
    session.setVideoHandler([&](const VideoFrameData &frame) {
        // Everything the OBS adapter is about to dereference must be sane --
        // checked against the frame's OWN geometry, not the publisher's.
        // WebRTC ramps a new subscription up from a downscaled spatial layer,
        // so the first frames after (re)subscribing legitimately arrive
        // smaller than what is being published; the adapter has to cope with
        // a mid-stream resolution change, and so does this assertion.
        const std::uint32_t chroma_stride = static_cast<std::uint32_t>((frame.width + 1) / 2);
        const bool sane = frame.width > 0 && frame.height > 0 && frame.format == PixelFormat::I420 &&
                          frame.plane_count == 3 && frame.data != nullptr &&
                          frame.size >= expectedFrameBytes(frame.format, frame.width, frame.height) &&
                          frame.planes[0].data != nullptr && frame.planes[1].data != nullptr &&
                          frame.planes[2].data != nullptr &&
                          frame.planes[0].stride >= static_cast<std::uint32_t>(frame.width) &&
                          frame.planes[1].stride >= chroma_stride && frame.planes[2].stride >= chroma_stride;
        if (!sane)
            bad_frames.fetch_add(1);
        if (frame.width == kWidth && frame.height == kHeight)
            full_size_frames.fetch_add(1);
        last_width.store(frame.width);
        last_height.store(frame.height);
        last_planes.store(frame.plane_count);
        last_timestamp.store(static_cast<long long>(frame.timestamp_us));
        video_frames.fetch_add(1);
    });
    session.setAudioHandler([&](const AudioFrameData &frame) {
        last_sample_rate.store(frame.sample_rate);
        last_channels.store(frame.channels);
        audio_frames.fetch_add(1);
    });

    std::atomic<int> state_changes{0};
    session.setStateHandler([&](SessionState, const std::string &) { state_changes.fetch_add(1); });

    SessionConfig config;
    config.ws_url = url;
    config.token = subscribe_token;
    config.participant_identity = publisher_identity;
    config.connect_timeout_ms = 10000;

    step("subscriber connect");
    ST_ASSERT(session.connect(config));
    ST_ASSERT(session.state() == SessionState::Connected);

    ST_ASSERT(waitFor([&] { return video_frames.load() >= 15; }, 25000));
    ST_ASSERT(session.hasVideo());
    ST_ASSERT(!session.waitingForCamera());
    ST_ASSERT_EQ(bad_frames.load(), 0);
    ST_ASSERT_EQ(last_planes.load(), 3);
    // The stream must actually reach the published resolution, not just
    // deliver ramp-up frames forever.
    ST_ASSERT(waitFor([&] { return full_size_frames.load() > 0; }, 20000));
    ST_ASSERT_EQ(last_width.load(), kWidth);
    ST_ASSERT_EQ(last_height.load(), kHeight);
    ST_ASSERT(last_timestamp.load() > 0);
    ST_ASSERT(session.videoFrameCount() >= 15);

    ST_ASSERT(waitFor([&] { return audio_frames.load() >= 10; }, 20000));
    ST_ASSERT(session.hasAudio());
    ST_ASSERT_EQ(last_sample_rate.load(), 48000);
    ST_ASSERT(last_channels.load() >= 1);

    // --- publisher swap: the bug this plugin exists to make impossible -----

    const int before_swap = video_frames.load();
    publisher.stopPump();
    step("unpublish video");
    ST_ASSERT(publisher.unpublishVideo());

    ST_ASSERT(waitFor([&] { return !session.hasVideo(); }, 15000));
    // An unpublished camera is the placeholder state, never a failure: the
    // room connection itself is untouched.
    ST_ASSERT(session.state() == SessionState::Connected);
    ST_ASSERT(session.waitingForCamera());

    // No frames may keep arriving from the dead publisher.
    const int after_unpublish = video_frames.load();
    std::this_thread::sleep_for(std::chrono::milliseconds(1500));
    ST_ASSERT_EQ(video_frames.load(), after_unpublish);
    ST_ASSERT(after_unpublish >= before_swap);

    // Republish, exactly as a reconnecting browser would.
    step("republish video");
    ST_ASSERT(publisher.publishVideo());
    publisher.startPump();

    ST_ASSERT(waitFor([&] { return video_frames.load() >= after_unpublish + 15; }, 25000));
    ST_ASSERT(session.hasVideo());
    ST_ASSERT(session.state() == SessionState::Connected);
    ST_ASSERT_EQ(bad_frames.load(), 0);

    // --- teardown ----------------------------------------------------------

    step("teardown");
    publisher.stopPump();
    session.disconnect();
    ST_ASSERT(session.state() == SessionState::Disconnected);
    ST_ASSERT(!session.hasVideo());

    // The publisher's Room must be torn down while the SDK is still
    // initialized, or its FFI disconnect fails on the way out.
    publisher_holder.reset();
    LiveKitSession::globalShutdown();

    std::printf("integration_livekit: %d video frames, %d audio frames, %d state changes\n", video_frames.load(),
                audio_frames.load(), state_changes.load());
    return st_test_report("integration_livekit");
}