// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
#include "core_queue_sink.h"
#include "queue_flac_metadata.h"
#include "queue_lms_cli.h"
#include "queue_report.h"
#include "third_party/yeney-core/core/player.h"
#include <algorithm>
#include <atomic>
#include <cmath>
#include <csignal>
#include <ctime>
#include <cstdlib>
#include <filesystem>
#include <fcntl.h>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <sstream>
#include <sys/file.h>
#include <thread>
#include <unistd.h>
namespace {
std::atomic_bool stopped{false}, resetMeasurement{false};
extern "C" void measureSignal(int) { resetMeasurement = true; }
extern "C" void stopSignal(int) { stopped = true; }
} // namespace
int main(int argc, char **argv) {
    std::signal(SIGINT, stopSignal);
    std::signal(SIGTERM, stopSignal);
    std::signal(SIGHUP, stopSignal);
    std::string server;
    unsigned seconds = 900;
    double speed = 1;
    bool keep = false, transcoded = false, controlled = false;
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        if (a == "--help") {
            std::cout << "yeney-queue-dryrun --server <ip> [--seconds 900] [--speed 1.0] "
                         "[--keep] [--transcoded]\nOwner-run only; select Queue Test "
                         "(YeneY) in "
                         "LMS. No Sonos or room-service access.\n";
            return 0;
        }
        if (a == "--suite-controlled")
            controlled = true;
        else if (a == "--transcoded")
            transcoded = true;
        else if (a == "--keep")
            keep = true;
        else if (a == "--server" && i + 1 < argc)
            server = argv[++i];
        else if (a.rfind("--server=", 0) == 0)
            server = a.substr(9);
        else if (a == "--speed" && i + 1 < argc) {
            try {
                std::string value = argv[++i];
                size_t used = 0;
                speed = std::stod(value, &used);
                if (used != value.size() || !std::isfinite(speed) || speed <= 0)
                    return 2;
            } catch (...) {
                return 2;
            }
        } else if (a == "--seconds" && i + 1 < argc) {
            try {
                std::string value = argv[++i];
                if (value.empty() || value.find_first_not_of("0123456789") != std::string::npos)
                    return 2;
                auto n = std::stoull(value);
                if (!n || n > UINT32_MAX)
                    return 2;
                seconds = unsigned(n);
            } catch (...) {
                return 2;
            }
        } else {
            std::cerr << "Invalid option: " << a << '\n';
            return 2;
        }
    }
    if (server.empty() || server.find(':') != std::string::npos || !seconds) {
        std::cerr << "--server <IPv4 address or hostname> is required; never "
                     "append :9000\n";
        return 2;
    }
    const auto env = [](const char *key, const std::string &fallback) {
        const char *value = std::getenv(key);
        return value && *value ? std::string(value) : fallback;
    };
    const std::string tempRoot = env("TMPDIR", "/var/tmp");
    const std::string cache = env("YENEY_TEST_CACHE", env("HOME", ".") + "/.cache/yeney-test");
    const std::filesystem::path reports = std::filesystem::path(cache) / "dryrun";
    std::filesystem::create_directories(reports);
    int lock = open("/var/tmp/yeney-queue-dryrun.lock", O_RDWR | O_CREAT | O_CLOEXEC | O_NOFOLLOW, 0600);
    if (lock < 0 || flock(lock, LOCK_EX | LOCK_NB)) {
        std::cerr << "Another dry-run owns the fixed test player\n";
        return 1;
    }
    auto time = std::time(nullptr);
    std::tm local{};
    localtime_r(&time, &local);
    char stamp[32];
    std::strftime(stamp, sizeof(stamp), "%Y%m%d-%H%M%S", &local);
    const std::string runName = std::string("yeney-queue-dryrun-") + stamp + "-" + std::to_string(getpid());
    const auto reportDir = reports / runName;
    std::filesystem::create_directory(reportDir);
    std::vector<std::filesystem::path> oldReports;
    for (const auto &entry : std::filesystem::directory_iterator(reports))
        if (entry.is_directory()) oldReports.push_back(entry.path());
    std::sort(oldReports.begin(), oldReports.end(), [](const auto &a, const auto &b) {
        return std::filesystem::last_write_time(a) > std::filesystem::last_write_time(b);
    });
    for (size_t i = 5; i < oldReports.size(); ++i) std::filesystem::remove_all(oldReports[i]);
    const std::string base = (reportDir / "report").string();
    const std::string storeRoot = keep ? (reportDir / "items").string() : tempRoot + "/" + runName + "-store";
    std::ofstream log(base + ".log");
    try {
        queue_audio::Config storage;
        storage.root = storeRoot;
        queue_audio::CoreQueueSink sink("QueueTest", storage, 30, speed);
        yeney::Config cfg;
        cfg.server = server;
        cfg.sourceIsTranscoded = [transcoded](uint64_t) { return transcoded; };
        cfg.name = "Queue Test (YeneY)";
        cfg.mac = {0x02, 0x59, 0x51, 0x54, 0, 1};
        std::mutex logMutex;
        cfg.log = [&](const std::string &s) {
            std::lock_guard<std::mutex> lock(logMutex);
            log << s << std::endl;
        };
        queue_audio::FlacMetadata flacMetadata(server, cfg.log);
        cfg.observeCommand = [&](const yeney::Command &command) {
            if (command.letter == 'f' || command.letter == 'q')
                flacMetadata.reset();
        };
        cfg.flacSourceInfo = [&](uint64_t id, std::optional<yeney::FlacSourceInfo> info) {
            auto rows = sink.snapshot();
            const auto audible = sink.audibleOccurrence();
            std::optional<uint64_t> audibleIndex;
            for (const auto &r : rows)
                if (r.source.occurrence == audible)
                    audibleIndex = r.expectedIndex;
            for (const auto &r : rows)
                if (r.source.occurrence == id)
                    return flacMetadata.resolve(r.expectedIndex, info, audibleIndex);
            cfg.log("FLAC metadata sealed fallback: occurrence identity unavailable");
            return info;
        };
        yeney::Player player(cfg, sink);
        if (controlled)
            std::signal(SIGUSR1, measureSignal);
        std::exception_ptr playerError;
        std::thread core([&] {
            try {
                player.run(stopped);
            } catch (...) {
                playerError = std::current_exception();
                stopped = true;
            }
        });
        struct JoinOnExit {
            std::thread &thread;
            ~JoinOnExit() {
                stopped = true;
                if (thread.joinable())
                    thread.join();
            }
        } joinOnExit{core};
        std::cout << "Play an album to Queue Test (YeneY), first crossfade off, "
                     "then on.\nReport: "
                  << base << ".md" << std::endl;
        using Clock = std::chrono::steady_clock;
        auto start = Clock::now(), last = start;
        double progress = 0;
        uint64_t active = 0;
        std::vector<uint64_t> completed;
        struct CliResult {
            uint64_t id;
            std::vector<std::string> values;
            double elapsed, stamp;
        };
        std::mutex cliMutex;
        std::vector<CliResult> cliResults;
        std::atomic<uint64_t> cliActive{0};
        std::atomic<double> cliElapsed{0};
        std::atomic_bool cliStop{false};
        std::thread cli([&] {
            auto next = Clock::now();
            auto nextRaw = next;
            while (!cliStop) {
                if (Clock::now() >= next) {
                    next += std::chrono::seconds(1);
                    const auto id = cliActive.load();
                    const double elapsed = cliElapsed.load(), stamp = sink.secondsSinceStart();
                    std::string raw;
                    auto values = queue_audio::pollLms(server, 9090, &raw);
                    {
                        std::lock_guard<std::mutex> lock(logMutex);
                        if (Clock::now() >= nextRaw) {
                            log << "LMS CLI raw: " << raw << std::endl;
                            nextRaw = Clock::now() + std::chrono::minutes(1);
                        }
                        log << "LMS CLI poll stamp=" << sink.secondsSinceStart() << " audible=" << id;
                        if (values.size() == 3)
                            log << " index=" << values[0] << " time=" << values[1] << " title=" << values[2];
                        else
                            log << " unavailable";
                        log << std::endl;
                    }
                    if (id && id == cliActive.load()) {
                        std::lock_guard<std::mutex> lock(cliMutex);
                        cliResults.push_back({id, std::move(values), elapsed, stamp});
                    }
                }
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        });
        struct CliJoin {
            std::thread &thread;
            std::atomic_bool &stop;
            ~CliJoin() {
                stop = true;
                if (thread.joinable())
                    thread.join();
            }
        } cliJoin{cli, cliStop};
        double measurementStart = 0;
        auto measuredRows = [&] {
            auto all = sink.snapshot();
            if (measurementStart > 0) {
                std::vector<uint64_t> owners;
                for (const auto &r : all)
                    if (r.source.occurrence < UINT64_MAX / 2 && r.strm >= measurementStart)
                        owners.push_back(r.source.occurrence);
                all.erase(std::remove_if(all.begin(), all.end(),
                                         [&](const auto &r) {
                                             if (r.source.occurrence < UINT64_MAX / 2)
                                                 return r.strm < measurementStart;
                                             return std::find(owners.begin(), owners.end(),
                                                              UINT64_MAX - r.source.occurrence) ==
                                                    owners.end();
                                         }),
                          all.end());
            }
            return all;
        };
        size_t reportedJoins = SIZE_MAX;
        auto nextProgress = Clock::now();
        while (!stopped && Clock::now() - start < std::chrono::seconds(seconds)) {
            auto now = Clock::now();
            if (resetMeasurement.exchange(false)) {
                measurementStart = sink.secondsSinceStart();
                reportedJoins = SIZE_MAX;
                std::cout << "Measurement ready" << std::endl;
            }
            double dt = std::chrono::duration<double>(now - last).count();
            last = now;
            auto rows = sink.snapshot();
            std::sort(rows.begin(), rows.end(),
                      [](const auto &a, const auto &b) { return a.first < b.first; });
            for (const auto &r : rows) {
                if (!r.item ||
                    std::find(completed.begin(), completed.end(), r.source.occurrence) != completed.end())
                    continue;
                std::optional<uint64_t> length;
                try {
                    length = r.item->contentLength();
                } catch (...) {
                    // A playlist replacement cancels the old active asset.
                    // Let the new occurrence become the next audible epoch.
                    if (active == r.source.occurrence) {
                        active = 0;
                        progress = 0;
                    }
                    continue;
                }
                if (!length || !r.item->producedFrames())
                    break;
                if (!active) {
                    active = r.source.occurrence;
                    progress = 0;
                    sink.audibleFrom(active, r.first);
                    dt = 0;
                }
                if (r.source.occurrence != active)
                    continue;
                if (!sink.isPaused())
                    progress += dt * speed * r.source.rate;
                const uint64_t total = (*length - 44) / 6, available = r.item->producedFrames();
                if (progress > available)
                    progress = available; // real underrun, never advance through absent PCM
                auto played = std::min<uint64_t>(uint64_t(progress), total);
                sink.advanceAudible(r.first + played);
                if (played == total) {
                    sink.audibleEnd(active);
                    completed.push_back(active);
                    active = 0;
                    progress = 0;
                }
                break;
            }
            const auto audibleId = active ? sink.audibleOccurrence() : 0;
            for (const auto &r : rows)
                if (r.source.occurrence == audibleId && r.source.rate) {
                    const auto frame = sink.audibleFrames();
                    cliElapsed = frame > r.first + r.prefixFrames
                                     ? double(frame - r.first - r.prefixFrames) / r.source.rate
                                     : 0;
                }
            cliActive = audibleId;
            std::vector<CliResult> samples;
            {
                std::lock_guard<std::mutex> lock(cliMutex);
                samples.swap(cliResults);
            }
            for (const auto &sample : samples) {
                auto it = std::find_if(rows.begin(), rows.end(),
                                       [&](const auto &r) { return r.source.occurrence == sample.id; });
                if (it == rows.end())
                    continue;
                queue_audio::QueueRecord::LmsSample value;
                value.stamp = sample.stamp;
                value.audibleElapsed = sample.elapsed;
                if (sample.values.size() == 3) {
                    value.index = sample.values[0];
                    value.time = sample.values[1];
                    value.title = sample.values[2];
                    try {
                        size_t used = 0;
                        double displayed = std::stod(value.time, &used);
                        if (used == value.time.size() && std::isfinite(displayed))
                            value.timeErrorWall = (displayed - sample.elapsed) / speed;
                    } catch (...) {
                    }
                    value.matches = value.index == std::to_string(it->expectedIndex);
                }
                sink.recordLms(sample.id, value);
                std::lock_guard<std::mutex> lock(logMutex);
                log << "LMS CLI stamp=" << value.stamp << " audible=" << sample.id
                    << " expected_index=" << it->expectedIndex << " index=" << value.index
                    << " time=" << value.time << " title=" << value.title << " lms_matches_audible="
                    << (value.matches ? (*value.matches ? "true" : "false") : "unknown") << std::endl;
            }
            if (now >= nextProgress) {
                std::ostringstream md, tsv;
                auto verdict = queue_audio::writeReadAheadReport(measuredRows(), md, tsv, speed);
                if (verdict.pairs != reportedJoins) {
                    reportedJoins = verdict.pairs;
                    std::cout << "Progress: completed_joins=" << reportedJoins << std::endl;
                }
                nextProgress = now + std::chrono::milliseconds(100);
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        cliStop = true;
        cli.join();
        try {
            sink.waitIdle();
        } catch (const std::exception &e) {
            std::lock_guard<std::mutex> lock(logMutex);
            log << "worker fault: " << e.what() << std::endl;
        }
        auto rows = measuredRows();
        // Snapshot before Player shutdown flushes pending assets; timestamps remain
        // explicit.
        stopped = true;
        core.join();
        try {
            sink.waitIdle();
        } catch (const std::exception &e) {
            log << "worker fault: " << e.what() << '\n';
        }
        std::ofstream tsv(base + ".tsv"), report(base + ".md");
        auto verdict =
            queue_audio::writeReadAheadReport(rows, report, tsv, speed, sink.wallSecondsSinceStart());
        std::cout << verdict.line() << "; worst lead " << verdict.worst << " s\n";
        if (playerError)
            std::rethrow_exception(playerError);
    } catch (const std::exception &e) {
        std::cerr << e.what() << '\n';
        if (!keep) {
            std::error_code ignored;
            std::filesystem::remove_all(storeRoot, ignored);
        }
        close(lock);
        return 1;
    }
    if (!keep)
        std::filesystem::remove_all(storeRoot);
    close(lock);
    return 0;
}