commits
tags
#include "torlinkc/ui/search_aggregator.hpp"
#include <algorithm>
#include <unordered_map>
#include <ftxui/component/event.hpp>
#include "torlinkc/sources/registry.hpp"
#include "torlinkc/util/net.hpp"
namespace torlinkc::ui {
namespace {
std::string errorCode(const std::exception& e) {
if (const auto* http = dynamic_cast<const HttpError*>(&e); http && http->status > 0) {
return "HTTP " + std::to_string(http->status);
}
return "no response";
}
} // namespace
SearchAggregator::SearchAggregator(ftxui::ScreenInteractive& screen, AppState& state)
: screen_(screen), state_(state) {}
SearchAggregator::~SearchAggregator() {
threads_.clear(); // requests stop + joins every jthread
}
void SearchAggregator::search(std::string query) {
const auto sources = allSources();
{
std::lock_guard<std::mutex> lock(dataMutex_);
generation_++;
collected_.clear();
perSource_.clear();
doneCount_ = 0;
totalCount_ = static_cast<int>(sources.size());
for (const auto& s : sources) perSource_[s.id] = SourceState{true, std::nullopt, std::nullopt, 0};
}
const int myGeneration = generation_;
// Destroying the old jthreads requests_stop() + joins each one. Real
// cancellation (util/net.cpp's libcurl progress callback) is what keeps
// this from blocking on a slow request's full timeout -- see the class
// comment.
threads_.clear();
notifier_ = std::make_unique<CoalescingNotifier>(std::chrono::milliseconds(150), [this] { flush(); });
state_.query = query;
state_.searching = true;
// OnAnimation only fires in response to a pending animation-frame request,
// not on an ordinary redraw -- without this kick, the spinner component
// would sit on its first frame until something else happened to already
// have one in flight.
screen_.RequestAnimationFrame();
state_.doneSources = 0;
state_.totalSources = static_cast<int>(sources.size());
state_.perSource.clear();
for (const auto& s : sources) state_.perSource[s.id] = SourceState{true, std::nullopt, std::nullopt, 0};
threads_.reserve(sources.size());
for (const auto& source : sources) {
threads_.emplace_back([this, source, query, myGeneration](std::stop_token stopToken) {
runSource(source, query, std::move(stopToken), myGeneration);
});
}
}
void SearchAggregator::runSource(const Source& source, const std::string& query, std::stop_token stopToken,
int generation) {
std::vector<TorrentResult> results;
std::optional<std::string> error;
std::optional<std::string> code;
try {
SearchOptions opts;
opts.stopToken = stopToken;
results = cache_.cachedSearch(source, query, opts);
} catch (const std::exception& e) {
if (stopToken.stop_requested()) return; // cancelled: drop silently, a newer search superseded this one
error = e.what();
code = errorCode(e);
}
if (stopToken.stop_requested()) return;
bool shouldFlushNow = false;
{
std::lock_guard<std::mutex> lock(dataMutex_);
if (generation != generation_) return; // belt-and-suspenders: see the header comment
if (error) {
perSource_[source.id] = SourceState{false, error, code, 0};
} else {
collected_.insert(collected_.end(), results.begin(), results.end());
perSource_[source.id] = SourceState{false, std::nullopt, std::nullopt, static_cast<int>(results.size())};
}
doneCount_++;
shouldFlushNow = doneCount_ >= totalCount_;
}
if (shouldFlushNow) {
notifier_->flushNow();
} else {
notifier_->markDirty();
}
}
void SearchAggregator::flush() {
std::vector<TorrentResult> resultsCopy;
std::unordered_map<std::string, SourceState> perSourceCopy;
int doneCopy = 0, totalCopy = 0;
{
std::lock_guard<std::mutex> lock(dataMutex_);
resultsCopy = collected_;
perSourceCopy = perSource_;
doneCopy = doneCount_;
totalCopy = totalCount_;
}
// Dedupe by infoHash keeping the higher-seeder copy, then default-order:
// healthiest first. Ported from useConcurrentSearch.ts's dedupe/defaultOrder.
std::unordered_map<std::string, TorrentResult> byHash;
for (auto& r : resultsCopy) {
auto it = byHash.find(r.infoHash);
if (it == byHash.end() || r.seeders > it->second.seeders) byHash[r.infoHash] = r;
}
std::vector<TorrentResult> deduped;
deduped.reserve(byHash.size());
for (auto& [hash, r] : byHash) deduped.push_back(std::move(r));
std::stable_sort(deduped.begin(), deduped.end(), [](const TorrentResult& a, const TorrentResult& b) {
if (a.seeders != b.seeders) return a.seeders > b.seeders;
return a.added.value_or(0) > b.added.value_or(0);
});
screen_.Post([this, deduped = std::move(deduped), perSourceCopy = std::move(perSourceCopy), doneCopy,
totalCopy]() mutable {
const bool hadNoResultsBefore = state_.results.empty();
state_.results = std::move(deduped);
state_.perSource = std::move(perSourceCopy);
state_.doneSources = doneCopy;
state_.totalSources = totalCopy;
state_.searching = doneCopy < totalCopy;
// main.cpp's onResultsChanged hook recomputes visibleResults and owns
// the "jump focus to the results list the first time this
// search produces any results" decision -- it knows the container's
// focus-index scheme, which SearchAggregator deliberately doesn't.
if (onResultsChanged) onResultsChanged(hadNoResultsBefore);
});
screen_.PostEvent(ftxui::Event::Custom);
}
} // namespace torlinkc::ui