diff --git a/include/search_engine/crawler/CrawlerManager.h b/include/search_engine/crawler/CrawlerManager.h index 55c75f6..43bd496 100644 --- a/include/search_engine/crawler/CrawlerManager.h +++ b/include/search_engine/crawler/CrawlerManager.h @@ -9,6 +9,7 @@ #include #include #include "Crawler.h" +#include "SessionAnalyticsStore.h" #include "models/CrawlConfig.h" #include "models/CrawlResult.h" #include "../storage/ContentStorage.h" @@ -33,19 +34,29 @@ struct CrawlSession { std::atomic isCompleted{false}; std::thread crawlThread; CrawlCompletionCallback completionCallback; - - CrawlSession(const std::string& sessionId, std::unique_ptr crawlerInstance, + // Seed URL and domain (captured at start so the analytics record can + // reference them after the crawler is gone). #15 + std::string seedUrl; + std::string seedDomain; + // Wall-clock start moment of the underlying crawl thread (distinct from + // createdAt, which is set when the session struct is constructed). #15 + std::chrono::system_clock::time_point startedAt; + + CrawlSession(const std::string& sessionId, std::unique_ptr crawlerInstance, CrawlCompletionCallback callback = nullptr) : id(sessionId), crawler(std::move(crawlerInstance)), createdAt(std::chrono::system_clock::now()), completionCallback(std::move(callback)) {} - + CrawlSession(CrawlSession&& other) noexcept : id(std::move(other.id)) , crawler(std::move(other.crawler)) , createdAt(other.createdAt) , isCompleted(other.isCompleted.load()) , crawlThread(std::move(other.crawlThread)) - , completionCallback(std::move(other.completionCallback)) {} + , completionCallback(std::move(other.completionCallback)) + , seedUrl(std::move(other.seedUrl)) + , seedDomain(std::move(other.seedDomain)) + , startedAt(other.startedAt) {} CrawlSession(const CrawlSession&) = delete; CrawlSession& operator=(const CrawlSession&) = delete; @@ -77,8 +88,12 @@ class CrawlerManager { // Get access to storage for logging std::shared_ptr getStorage() const { return storage_; } - - + + // Access to the session-analytics store (issue #15). Always non-null; + // defaults to in-memory store created in the constructor. + ISessionAnalyticsStore* getAnalyticsStore() const { return analyticsStore_.get(); } + + // Limit concurrent sessions to prevent MongoDB connection issues static constexpr size_t MAX_CONCURRENT_SESSIONS = 5; @@ -89,9 +104,16 @@ class CrawlerManager { std::atomic sessionCounter_{0}; std::thread cleanupThread_; std::atomic shouldStop_{false}; + // Owned analytics store — captures a SessionMetricsRecord for every + // session that completes (success or failure). #15 + std::unique_ptr analyticsStore_; std::string generateSessionId(); void cleanupWorker(); std::unique_ptr createCrawler(const CrawlConfig& config, const std::string& sessionId = ""); + // Build a SessionMetricsRecord from a completed session and push it + // into the analytics store. #15 + void recordSessionAnalytics(const CrawlSession& session, + const std::vector& results); }; diff --git a/include/search_engine/crawler/SessionAnalytics.h b/include/search_engine/crawler/SessionAnalytics.h new file mode 100644 index 0000000..2a8961f --- /dev/null +++ b/include/search_engine/crawler/SessionAnalytics.h @@ -0,0 +1,217 @@ +#pragma once + +#include "SessionMetricsRecord.h" +#include "models/CrawlResult.h" +#include +#include +#include +#include +#include +#include +#include + +/** + * @brief Pure functions for session-metrics aggregation, comparison, and + * trend reporting (issue #15). + * + * None of these touch storage, threads, network — they take values in and + * return values out. That keeps everything trivially unit-testable. + */ +namespace SessionAnalytics { + +/** + * Build a SessionMetricsRecord from a finished session's CrawlResult vector + * plus session-level metadata (id, seed url/domain, start/finish times). + */ +inline SessionMetricsRecord buildFromResults(const std::string& sessionId, + const std::string& seedUrl, + const std::string& seedDomain, + std::chrono::system_clock::time_point startedAt, + std::chrono::system_clock::time_point finishedAt, + const std::vector& results) { + SessionMetricsRecord r; + r.sessionId = sessionId; + r.seedUrl = seedUrl; + r.seedDomain = seedDomain; + r.startedAt = startedAt; + r.finishedAt = finishedAt; + r.durationMs = std::chrono::duration_cast(finishedAt - startedAt).count(); + if (r.durationMs < 0) r.durationMs = 0; + + std::vector latencies; + latencies.reserve(results.size()); + + for (const auto& cr : results) { + r.totalUrls++; + if (cr.success) { + r.successfulUrls++; + r.totalBytes += cr.contentSize; + } else if (cr.crawlStatus == "failed") { + r.failedUrls++; + } + if (cr.retryCount > 0) { + r.retriedUrls++; + r.totalRetryAttempts += static_cast(cr.retryCount); + } + if (cr.statusCode != 0) { + r.statusCodeCounts[cr.statusCode]++; + } + r.failureTypeCounts[static_cast(cr.failureType)]++; + + // Compute per-URL latency only if we have a sane started/finished + // timestamp pair on the result. + auto perUrlMs = std::chrono::duration_cast( + cr.finishedAt - cr.startedAt).count(); + if (perUrlMs > 0) latencies.push_back(perUrlMs); + } + + if (!latencies.empty()) { + std::sort(latencies.begin(), latencies.end()); + int64_t sum = 0; + for (auto v : latencies) sum += v; + r.avgLatencyMs = sum / static_cast(latencies.size()); + auto percentile = [&](double p) -> int64_t { + // Nearest-rank percentile; safe for small N. + if (latencies.empty()) return 0; + size_t idx = static_cast(std::ceil(p * latencies.size())) ; + if (idx == 0) idx = 1; + if (idx > latencies.size()) idx = latencies.size(); + return latencies[idx - 1]; + }; + r.p50LatencyMs = percentile(0.50); + r.p95LatencyMs = percentile(0.95); + r.p99LatencyMs = percentile(0.99); + r.maxLatencyMs = latencies.back(); + } + + return r; +} + +/** + * Summary across many sessions — used as the response for "list all" and + * for compare. Sums and averages roll up cleanly. + */ +struct AggregateSummary { + size_t sessionCount{0}; + size_t totalUrls{0}; + size_t successfulUrls{0}; + size_t failedUrls{0}; + size_t retriedUrls{0}; + size_t totalRetryAttempts{0}; + size_t totalBytes{0}; + // Averaged across sessions (so a 0-URL session still pulls the average + // toward 0; callers can filter empty sessions if they want). + double avgSuccessRate{0.0}; + double avgRetryRate{0.0}; + double avgThroughput{0.0}; + int64_t avgDurationMs{0}; + int64_t avgLatencyMs{0}; +}; + +inline AggregateSummary summarize(const std::vector& records) { + AggregateSummary s; + s.sessionCount = records.size(); + if (records.empty()) return s; + + double sumSuccessRate = 0, sumRetryRate = 0, sumThroughput = 0; + int64_t sumDuration = 0, sumLatency = 0; + for (const auto& r : records) { + s.totalUrls += r.totalUrls; + s.successfulUrls += r.successfulUrls; + s.failedUrls += r.failedUrls; + s.retriedUrls += r.retriedUrls; + s.totalRetryAttempts += r.totalRetryAttempts; + s.totalBytes += r.totalBytes; + sumSuccessRate += r.getSuccessRate(); + sumRetryRate += r.getRetryRate(); + sumThroughput += r.getThroughput(); + sumDuration += r.durationMs; + sumLatency += r.avgLatencyMs; + } + s.avgSuccessRate = sumSuccessRate / static_cast(records.size()); + s.avgRetryRate = sumRetryRate / static_cast(records.size()); + s.avgThroughput = sumThroughput / static_cast(records.size()); + s.avgDurationMs = sumDuration / static_cast(records.size()); + s.avgLatencyMs = sumLatency / static_cast(records.size()); + return s; +} + +/** + * Pairwise comparison result between two sessions — used by the compare API. + */ +struct ComparisonResult { + std::string sessionA; + std::string sessionB; + double successRateDelta{0.0}; // B - A + double retryRateDelta{0.0}; // B - A + double throughputDelta{0.0}; // B - A + int64_t durationDeltaMs{0}; // B - A + int64_t avgLatencyDeltaMs{0}; // B - A + // > 0 means B is faster / better depending on the field. +}; + +inline ComparisonResult compare(const SessionMetricsRecord& a, const SessionMetricsRecord& b) { + ComparisonResult c; + c.sessionA = a.sessionId; + c.sessionB = b.sessionId; + c.successRateDelta = b.getSuccessRate() - a.getSuccessRate(); + c.retryRateDelta = b.getRetryRate() - a.getRetryRate(); + c.throughputDelta = b.getThroughput() - a.getThroughput(); + c.durationDeltaMs = b.durationMs - a.durationMs; + c.avgLatencyDeltaMs = b.avgLatencyMs - a.avgLatencyMs; + return c; +} + +/** + * Trend bucket — a time slice (e.g. one hour, one day) with rolled-up stats. + * Buckets are keyed by the start of the bucket (Unix epoch ms). + */ +struct TrendBucket { + int64_t bucketStartMs{0}; + int64_t bucketEndMs{0}; + AggregateSummary summary; +}; + +/** + * Bucket sessions by their `startedAt` into fixed-size time slices and + * summarize each bucket. Returns buckets in ascending time order. + * + * - `bucketWidth` controls the slice size (e.g. 1h, 24h, 7d). + * - Sessions outside [windowStart, windowEnd) are skipped. + */ +inline std::vector trends(const std::vector& records, + std::chrono::system_clock::time_point windowStart, + std::chrono::system_clock::time_point windowEnd, + std::chrono::milliseconds bucketWidth) { + std::vector out; + if (bucketWidth.count() <= 0) return out; + if (windowEnd <= windowStart) return out; + + auto winStartMs = std::chrono::duration_cast( + windowStart.time_since_epoch()).count(); + auto winEndMs = std::chrono::duration_cast( + windowEnd.time_since_epoch()).count(); + int64_t bw = bucketWidth.count(); + + // Group records into bucket indices. + std::map> buckets; + for (const auto& r : records) { + auto startedMs = std::chrono::duration_cast( + r.startedAt.time_since_epoch()).count(); + if (startedMs < winStartMs || startedMs >= winEndMs) continue; + int64_t idx = (startedMs - winStartMs) / bw; + buckets[idx].push_back(r); + } + + // Emit one bucket per index that actually has data. + for (const auto& [idx, recs] : buckets) { + TrendBucket tb; + tb.bucketStartMs = winStartMs + idx * bw; + tb.bucketEndMs = tb.bucketStartMs + bw; + tb.summary = summarize(recs); + out.push_back(tb); + } + return out; +} + +} // namespace SessionAnalytics diff --git a/include/search_engine/crawler/SessionAnalyticsStore.h b/include/search_engine/crawler/SessionAnalyticsStore.h new file mode 100644 index 0000000..c8b561c --- /dev/null +++ b/include/search_engine/crawler/SessionAnalyticsStore.h @@ -0,0 +1,121 @@ +#pragma once + +#include "SessionMetricsRecord.h" +#include +#include +#include +#include +#include +#include +#include + +/** + * @brief Thread-safe in-memory store for SessionMetricsRecord (issue #15). + * + * Records are inserted at session completion and queried by: + * - sessionId + * - time window + * - "all" (with optional cap) + * + * A capacity bound keeps memory bounded; oldest records evict first. + * + * The store is an interface (virtual methods) so it can later be backed by + * MongoDB or another durable backend without changing callers. The default + * in-memory implementation is provided here for the API endpoints, tests, + * and dev usage. A persistent backend can subclass and override. + */ +class ISessionAnalyticsStore { +public: + virtual ~ISessionAnalyticsStore() = default; + + virtual void put(const SessionMetricsRecord& record) = 0; + virtual std::optional get(const std::string& sessionId) const = 0; + virtual std::vector getAll(size_t limit = 0) const = 0; + virtual std::vector getInWindow( + std::chrono::system_clock::time_point from, + std::chrono::system_clock::time_point to) const = 0; + virtual size_t size() const = 0; + virtual void clear() = 0; +}; + +class InMemorySessionAnalyticsStore : public ISessionAnalyticsStore { +public: + explicit InMemorySessionAnalyticsStore(size_t capacity = 10000) + : capacity_(capacity) {} + + void put(const SessionMetricsRecord& record) override { + std::lock_guard lock(mutex_); + // Overwrite on duplicate id (idempotent updates allowed). + auto idxIt = idIndex_.find(record.sessionId); + if (idxIt != idIndex_.end()) { + records_[idxIt->second] = record; + return; + } + records_.push_back(record); + idIndex_[record.sessionId] = records_.size() - 1; + evictIfNeededLocked(); + } + + std::optional get(const std::string& sessionId) const override { + std::lock_guard lock(mutex_); + auto it = idIndex_.find(sessionId); + if (it == idIndex_.end()) return std::nullopt; + return records_[it->second]; + } + + std::vector getAll(size_t limit = 0) const override { + std::lock_guard lock(mutex_); + if (limit == 0 || limit >= records_.size()) { + return records_; + } + // Return the most recent `limit` entries. + return std::vector( + records_.end() - static_cast(limit), records_.end()); + } + + std::vector getInWindow( + std::chrono::system_clock::time_point from, + std::chrono::system_clock::time_point to) const override { + std::lock_guard lock(mutex_); + std::vector out; + out.reserve(records_.size()); + for (const auto& r : records_) { + if (r.startedAt >= from && r.startedAt < to) { + out.push_back(r); + } + } + return out; + } + + size_t size() const override { + std::lock_guard lock(mutex_); + return records_.size(); + } + + void clear() override { + std::lock_guard lock(mutex_); + records_.clear(); + idIndex_.clear(); + } + +private: + void evictIfNeededLocked() { + if (records_.size() <= capacity_) return; + size_t toDrop = records_.size() - capacity_; + // Erase oldest entries (front). + for (size_t i = 0; i < toDrop; ++i) { + idIndex_.erase(records_[i].sessionId); + } + records_.erase(records_.begin(), records_.begin() + static_cast(toDrop)); + // Rebuild index since positions shifted. + idIndex_.clear(); + for (size_t i = 0; i < records_.size(); ++i) { + idIndex_[records_[i].sessionId] = i; + } + } + + mutable std::mutex mutex_; + std::vector records_; + std::unordered_map idIndex_; + size_t capacity_; +}; diff --git a/include/search_engine/crawler/SessionMetricsRecord.h b/include/search_engine/crawler/SessionMetricsRecord.h new file mode 100644 index 0000000..411699c --- /dev/null +++ b/include/search_engine/crawler/SessionMetricsRecord.h @@ -0,0 +1,69 @@ +#pragma once + +#include +#include +#include +#include +#include + +/** + * @brief Immutable per-session metrics record (issue #15). + * + * Produced from CrawlResult vectors at session completion and persisted + * via SessionAnalyticsStore for historical reporting and comparisons. + * + * The struct is intentionally a plain value type — no atomics, no + * threading — so it can be passed around freely, serialized to JSON, + * compared, and bucketed for trend reports. + */ +struct SessionMetricsRecord { + // Identity + std::string sessionId; + std::string seedUrl; + std::string seedDomain; + + // Timing + std::chrono::system_clock::time_point startedAt{}; + std::chrono::system_clock::time_point finishedAt{}; + // Wall clock duration of the session in milliseconds. + int64_t durationMs{0}; + + // Counts + size_t totalUrls{0}; + size_t successfulUrls{0}; + size_t failedUrls{0}; + size_t retriedUrls{0}; + size_t totalRetryAttempts{0}; + + // Bytes downloaded (sum over successful results). + size_t totalBytes{0}; + + // Latency aggregates per-URL in milliseconds (finishedAt - startedAt). + int64_t avgLatencyMs{0}; + int64_t p50LatencyMs{0}; + int64_t p95LatencyMs{0}; + int64_t p99LatencyMs{0}; + int64_t maxLatencyMs{0}; + + // HTTP status code distribution (e.g. {200: 42, 404: 3, 500: 1}). + std::unordered_map statusCodeCounts; + + // Failure type distribution (FailureType enum as int, since storing + // the enum keeps this header free of the enum dependency). + std::unordered_map failureTypeCounts; + + // Derived rates (computed at build time for convenience). + double getSuccessRate() const { + return totalUrls > 0 ? static_cast(successfulUrls) / static_cast(totalUrls) : 0.0; + } + double getFailureRate() const { + return totalUrls > 0 ? static_cast(failedUrls) / static_cast(totalUrls) : 0.0; + } + double getRetryRate() const { + return totalUrls > 0 ? static_cast(retriedUrls) / static_cast(totalUrls) : 0.0; + } + // Throughput: URLs per second across session duration. + double getThroughput() const { + return durationMs > 0 ? (static_cast(totalUrls) * 1000.0) / static_cast(durationMs) : 0.0; + } +}; diff --git a/src/controllers/SearchController.cpp b/src/controllers/SearchController.cpp index 221889a..173ecd0 100644 --- a/src/controllers/SearchController.cpp +++ b/src/controllers/SearchController.cpp @@ -2,6 +2,9 @@ #include "../../include/Logger.h" #include "../../include/search_engine/crawler/Crawler.h" #include "../../include/search_engine/crawler/CrawlerManager.h" +#include "../../include/search_engine/crawler/SessionAnalytics.h" +#include "../../include/search_engine/crawler/SessionAnalyticsStore.h" +#include "../../include/search_engine/crawler/SessionMetricsRecord.h" #include "../../include/search_engine/crawler/PageFetcher.h" #include "../../include/search_engine/crawler/models/CrawlConfig.h" #include "../../include/search_engine/storage/ContentStorage.h" @@ -796,6 +799,265 @@ void SearchController::getCrawlDetails(uWS::HttpResponse* res, uWS::HttpR } } +// ============================================================================ +// Session analytics endpoints (issue #15) +// ============================================================================ + +namespace { + int64_t toEpochMs(std::chrono::system_clock::time_point tp) { + return std::chrono::duration_cast( + tp.time_since_epoch()).count(); + } + + nlohmann::json sessionRecordToJson(const SessionMetricsRecord& r) { + nlohmann::json j; + j["sessionId"] = r.sessionId; + j["seedUrl"] = r.seedUrl; + j["seedDomain"] = r.seedDomain; + j["startedAtMs"] = toEpochMs(r.startedAt); + j["finishedAtMs"] = toEpochMs(r.finishedAt); + j["durationMs"] = r.durationMs; + j["totalUrls"] = r.totalUrls; + j["successfulUrls"] = r.successfulUrls; + j["failedUrls"] = r.failedUrls; + j["retriedUrls"] = r.retriedUrls; + j["totalRetryAttempts"] = r.totalRetryAttempts; + j["totalBytes"] = r.totalBytes; + j["successRate"] = r.getSuccessRate(); + j["failureRate"] = r.getFailureRate(); + j["retryRate"] = r.getRetryRate(); + j["throughputUrlsPerSec"] = r.getThroughput(); + j["latency"] = { + {"avgMs", r.avgLatencyMs}, + {"p50Ms", r.p50LatencyMs}, + {"p95Ms", r.p95LatencyMs}, + {"p99Ms", r.p99LatencyMs}, + {"maxMs", r.maxLatencyMs} + }; + // status code map (key as string for JSON) + nlohmann::json codes = nlohmann::json::object(); + for (const auto& [code, count] : r.statusCodeCounts) { + codes[std::to_string(code)] = count; + } + j["statusCodeCounts"] = codes; + nlohmann::json failTypes = nlohmann::json::object(); + for (const auto& [ft, count] : r.failureTypeCounts) { + failTypes[std::to_string(ft)] = count; + } + j["failureTypeCounts"] = failTypes; + return j; + } + + nlohmann::json summaryToJson(const SessionAnalytics::AggregateSummary& s) { + return { + {"sessionCount", s.sessionCount}, + {"totalUrls", s.totalUrls}, + {"successfulUrls", s.successfulUrls}, + {"failedUrls", s.failedUrls}, + {"retriedUrls", s.retriedUrls}, + {"totalRetryAttempts", s.totalRetryAttempts}, + {"totalBytes", s.totalBytes}, + {"avgSuccessRate", s.avgSuccessRate}, + {"avgRetryRate", s.avgRetryRate}, + {"avgThroughputUrlsPerSec", s.avgThroughput}, + {"avgDurationMs", s.avgDurationMs}, + {"avgLatencyMs", s.avgLatencyMs} + }; + } + + // Split "a,b,c" into {"a","b","c"}. + std::vector splitCsv(const std::string& s) { + std::vector out; + std::string cur; + for (char c : s) { + if (c == ',') { + if (!cur.empty()) { out.push_back(cur); cur.clear(); } + } else { + cur += c; + } + } + if (!cur.empty()) out.push_back(cur); + return out; + } +} + +// GET /api/analytics/sessions?limit=N +void SearchController::getAnalyticsSessions(uWS::HttpResponse* res, uWS::HttpRequest* req) { + LOG_INFO("SearchController::getAnalyticsSessions called"); + try { + if (!g_crawlerManager) { serverError(res, "CrawlerManager not initialized"); return; } + auto store = g_crawlerManager->getAnalyticsStore(); + if (!store) { serverError(res, "Analytics store not initialized"); return; } + + auto params = parseQuery(req); + size_t limit = 0; + auto lit = params.find("limit"); + if (lit != params.end()) { + try { limit = std::stoull(lit->second); } catch (...) { limit = 0; } + } + + auto records = store->getAll(limit); + nlohmann::json arr = nlohmann::json::array(); + for (const auto& r : records) arr.push_back(sessionRecordToJson(r)); + + auto summary = SessionAnalytics::summarize(records); + nlohmann::json response; + response["count"] = arr.size(); + response["storeSize"] = store->size(); + response["summary"] = summaryToJson(summary); + response["sessions"] = arr; + json(res, response); + } catch (const std::exception& e) { + LOG_ERROR("Error in getAnalyticsSessions: " + std::string(e.what())); + serverError(res, "Failed to get analytics sessions"); + } +} + +// GET /api/analytics/sessions/detail?sessionId=... +void SearchController::getAnalyticsSession(uWS::HttpResponse* res, uWS::HttpRequest* req) { + LOG_INFO("SearchController::getAnalyticsSession called"); + try { + if (!g_crawlerManager) { serverError(res, "CrawlerManager not initialized"); return; } + auto store = g_crawlerManager->getAnalyticsStore(); + if (!store) { serverError(res, "Analytics store not initialized"); return; } + + auto params = parseQuery(req); + auto sit = params.find("sessionId"); + if (sit == params.end() || sit->second.empty()) { + badRequest(res, "sessionId query parameter is required"); + return; + } + auto rec = store->get(sit->second); + if (!rec.has_value()) { + notFound(res, "Session metrics not found"); + return; + } + json(res, sessionRecordToJson(*rec)); + } catch (const std::exception& e) { + LOG_ERROR("Error in getAnalyticsSession: " + std::string(e.what())); + serverError(res, "Failed to get session analytics"); + } +} + +// GET /api/analytics/sessions/compare?ids=a,b,c +void SearchController::getAnalyticsCompare(uWS::HttpResponse* res, uWS::HttpRequest* req) { + LOG_INFO("SearchController::getAnalyticsCompare called"); + try { + if (!g_crawlerManager) { serverError(res, "CrawlerManager not initialized"); return; } + auto store = g_crawlerManager->getAnalyticsStore(); + if (!store) { serverError(res, "Analytics store not initialized"); return; } + + auto params = parseQuery(req); + auto iit = params.find("ids"); + if (iit == params.end() || iit->second.empty()) { + badRequest(res, "ids query parameter required (comma-separated session ids)"); + return; + } + auto ids = splitCsv(iit->second); + if (ids.size() < 2) { + badRequest(res, "At least 2 session ids required for comparison"); + return; + } + + std::vector records; + nlohmann::json missing = nlohmann::json::array(); + for (const auto& id : ids) { + auto rec = store->get(id); + if (rec.has_value()) records.push_back(*rec); + else missing.push_back(id); + } + if (records.size() < 2) { + nlohmann::json err; + err["error"] = "Need at least 2 found sessions to compare"; + err["missing"] = missing; + json(res, err); + return; + } + + // Pairwise compare against records[0] for a "vs baseline" view. + nlohmann::json pairwise = nlohmann::json::array(); + for (size_t i = 1; i < records.size(); ++i) { + auto c = SessionAnalytics::compare(records[0], records[i]); + pairwise.push_back({ + {"baseline", c.sessionA}, + {"other", c.sessionB}, + {"successRateDelta", c.successRateDelta}, + {"retryRateDelta", c.retryRateDelta}, + {"throughputDelta", c.throughputDelta}, + {"durationDeltaMs", c.durationDeltaMs}, + {"avgLatencyDeltaMs", c.avgLatencyDeltaMs} + }); + } + + nlohmann::json response; + response["summary"] = summaryToJson(SessionAnalytics::summarize(records)); + response["missing"] = missing; + nlohmann::json sessionsArr = nlohmann::json::array(); + for (const auto& r : records) sessionsArr.push_back(sessionRecordToJson(r)); + response["sessions"] = sessionsArr; + response["pairwise"] = pairwise; + json(res, response); + } catch (const std::exception& e) { + LOG_ERROR("Error in getAnalyticsCompare: " + std::string(e.what())); + serverError(res, "Failed to compare sessions"); + } +} + +// GET /api/analytics/sessions/trends?windowMs=86400000&bucketMs=3600000 +void SearchController::getAnalyticsTrends(uWS::HttpResponse* res, uWS::HttpRequest* req) { + LOG_INFO("SearchController::getAnalyticsTrends called"); + try { + if (!g_crawlerManager) { serverError(res, "CrawlerManager not initialized"); return; } + auto store = g_crawlerManager->getAnalyticsStore(); + if (!store) { serverError(res, "Analytics store not initialized"); return; } + + auto params = parseQuery(req); + // Defaults: last 24 hours, hourly buckets. + int64_t windowMs = 24LL * 60 * 60 * 1000; + int64_t bucketMs = 60LL * 60 * 1000; + auto wit = params.find("windowMs"); + if (wit != params.end()) { + try { windowMs = std::stoll(wit->second); } catch (...) {} + } + auto bit = params.find("bucketMs"); + if (bit != params.end()) { + try { bucketMs = std::stoll(bit->second); } catch (...) {} + } + if (windowMs <= 0 || bucketMs <= 0) { + badRequest(res, "windowMs and bucketMs must be positive"); + return; + } + + auto now = std::chrono::system_clock::now(); + auto windowStart = now - std::chrono::milliseconds(windowMs); + auto records = store->getInWindow(windowStart, now); + + auto buckets = SessionAnalytics::trends(records, windowStart, now, + std::chrono::milliseconds(bucketMs)); + nlohmann::json bucketsArr = nlohmann::json::array(); + for (const auto& b : buckets) { + bucketsArr.push_back({ + {"bucketStartMs", b.bucketStartMs}, + {"bucketEndMs", b.bucketEndMs}, + {"summary", summaryToJson(b.summary)} + }); + } + + nlohmann::json response; + response["windowMs"] = windowMs; + response["bucketMs"] = bucketMs; + response["windowStartMs"] = toEpochMs(windowStart); + response["windowEndMs"] = toEpochMs(now); + response["totalSessionsInWindow"] = records.size(); + response["overallSummary"] = summaryToJson(SessionAnalytics::summarize(records)); + response["buckets"] = bucketsArr; + json(res, response); + } catch (const std::exception& e) { + LOG_ERROR("Error in getAnalyticsTrends: " + std::string(e.what())); + serverError(res, "Failed to compute trends"); + } +} + void SearchController::detectSpa(uWS::HttpResponse* res, uWS::HttpRequest* req) { LOG_INFO("SearchController::detectSpa called"); diff --git a/src/controllers/SearchController.h b/src/controllers/SearchController.h index 9ad9dd4..f3c72ef 100644 --- a/src/controllers/SearchController.h +++ b/src/controllers/SearchController.h @@ -23,6 +23,12 @@ class SearchController : public routing::Controller { void addSiteToCrawl(uWS::HttpResponse* res, uWS::HttpRequest* req); // Supports 'force' parameter void getCrawlStatus(uWS::HttpResponse* res, uWS::HttpRequest* req); void getCrawlDetails(uWS::HttpResponse* res, uWS::HttpRequest* req); // New endpoint + + // Session analytics (issue #15) + void getAnalyticsSessions(uWS::HttpResponse* res, uWS::HttpRequest* req); + void getAnalyticsSession(uWS::HttpResponse* res, uWS::HttpRequest* req); + void getAnalyticsCompare(uWS::HttpResponse* res, uWS::HttpRequest* req); + void getAnalyticsTrends(uWS::HttpResponse* res, uWS::HttpRequest* req); // SPA detection void detectSpa(uWS::HttpResponse* res, uWS::HttpRequest* req); @@ -77,5 +83,10 @@ ROUTE_CONTROLLER(SearchController) { REGISTER_ROUTE(HttpMethod::POST, "/api/crawl/add-site", addSiteToCrawl, SearchController); REGISTER_ROUTE(HttpMethod::GET, "/api/crawl/status", getCrawlStatus, SearchController); REGISTER_ROUTE(HttpMethod::GET, "/api/crawl/details", getCrawlDetails, SearchController); // New endpoint + // Session analytics endpoints (#15) + REGISTER_ROUTE(HttpMethod::GET, "/api/analytics/sessions", getAnalyticsSessions, SearchController); + REGISTER_ROUTE(HttpMethod::GET, "/api/analytics/sessions/detail", getAnalyticsSession, SearchController); + REGISTER_ROUTE(HttpMethod::GET, "/api/analytics/sessions/compare", getAnalyticsCompare, SearchController); + REGISTER_ROUTE(HttpMethod::GET, "/api/analytics/sessions/trends", getAnalyticsTrends, SearchController); REGISTER_ROUTE(HttpMethod::POST, "/api/spa/detect", detectSpa, SearchController); -} \ No newline at end of file +} \ No newline at end of file diff --git a/src/crawler/CrawlerManager.cpp b/src/crawler/CrawlerManager.cpp index dec8dfc..671bb1a 100644 --- a/src/crawler/CrawlerManager.cpp +++ b/src/crawler/CrawlerManager.cpp @@ -1,6 +1,7 @@ #include "CrawlerManager.h" #include "../../include/Logger.h" #include "../../include/crawler/CrawlLogger.h" +#include "../../include/search_engine/crawler/SessionAnalytics.h" #include "PageFetcher.h" #include #include @@ -9,9 +10,10 @@ #include CrawlerManager::CrawlerManager(std::shared_ptr storage) - : storage_(storage) { - LOG_INFO("CrawlerManager initialized"); - + : storage_(storage), + analyticsStore_(std::make_unique(10000)) { + LOG_INFO("CrawlerManager initialized (analytics store capacity=10000)"); + // Start background cleanup thread cleanupThread_ = std::thread(&CrawlerManager::cleanupWorker, this); } @@ -81,6 +83,22 @@ std::string CrawlerManager::startCrawl(const std::string& url, const CrawlConfig // Add seed URL to the crawler LOG_DEBUG("CrawlerManager::startCrawl - Adding seed URL for session: " + sessionId); session->crawler->addSeedURL(url, force); + + // Capture seed info on the session struct so the analytics record + // built at completion time can reference them. #15 + session->seedUrl = url; + try { + // The crawler exposes a URL frontier that knows how to extract + // the domain; fall back to empty if anything goes sideways. + // (Keeping this best-effort to avoid coupling.) + auto schemeEnd = url.find("://"); + std::string rest = (schemeEnd == std::string::npos) ? url : url.substr(schemeEnd + 3); + auto slash = rest.find('/'); + session->seedDomain = (slash == std::string::npos) ? rest : rest.substr(0, slash); + } catch (...) { + session->seedDomain.clear(); + } + session->startedAt = std::chrono::system_clock::now(); // Start crawling in a separate thread session->crawlThread = std::thread([sessionId, this]() { @@ -152,13 +170,23 @@ std::string CrawlerManager::startCrawl(const std::string& url, const CrawlConfig if (sessionIt != sessions_.end()) { auto& completedSession = sessionIt->second; completedSession->isCompleted = true; - + + std::vector finalResults; + if (completedSession->crawler) { + finalResults = completedSession->crawler->getResults(); + } + + try { + recordSessionAnalytics(*completedSession, finalResults); + } catch (const std::exception& e) { + LOG_ERROR("Error recording session analytics for " + sessionId + ": " + e.what()); + } + // Execute completion callback if provided if (completedSession->completionCallback) { LOG_INFO("Executing completion callback for session: " + sessionId); try { - auto results = completedSession->crawler->getResults(); - completedSession->completionCallback(sessionId, results, this); + completedSession->completionCallback(sessionId, finalResults, this); LOG_INFO("Completion callback executed successfully for session: " + sessionId); } catch (const std::exception& e) { LOG_ERROR("Error executing completion callback for session " + sessionId + ": " + e.what()); @@ -398,18 +426,39 @@ void CrawlerManager::cleanupWorker() { std::unique_ptr CrawlerManager::createCrawler(const CrawlConfig& config, const std::string& sessionId) { auto crawler = std::make_unique(config, storage_, sessionId); - + // Configure PageFetcher settings if (crawler->getPageFetcher()) { // Disable SSL verification for problematic sites crawler->getPageFetcher()->setVerifySSL(false); - + // Enable SPA rendering if configured if (config.spaRenderingEnabled) { crawler->getPageFetcher()->setSpaRendering(true, config.browserlessUrl); CrawlLogger::broadcastLog("🤖 SPA rendering enabled for session with browserless URL: " + config.browserlessUrl, "info"); } } - + return crawler; +} + +// Build a SessionMetricsRecord from the just-completed session and push it +// into the analytics store. Failures here must not propagate; this is a +// best-effort telemetry hook. #15 +void CrawlerManager::recordSessionAnalytics(const CrawlSession& session, + const std::vector& results) { + if (!analyticsStore_) return; + auto finishedAt = std::chrono::system_clock::now(); + auto startedAt = session.startedAt.time_since_epoch().count() == 0 + ? session.createdAt + : session.startedAt; + auto record = SessionAnalytics::buildFromResults( + session.id, session.seedUrl, session.seedDomain, + startedAt, finishedAt, results); + analyticsStore_->put(record); + LOG_INFO("Recorded analytics for session " + session.id + + " (urls=" + std::to_string(record.totalUrls) + + ", success=" + std::to_string(record.successfulUrls) + + ", failed=" + std::to_string(record.failedUrls) + + ", durationMs=" + std::to_string(record.durationMs) + ")"); } \ No newline at end of file diff --git a/src/crawler/CrawlerManager.h b/src/crawler/CrawlerManager.h index 0b16589..2c11633 100644 --- a/src/crawler/CrawlerManager.h +++ b/src/crawler/CrawlerManager.h @@ -9,6 +9,7 @@ #include #include #include "Crawler.h" +#include "../../include/search_engine/crawler/SessionAnalyticsStore.h" #include "models/CrawlConfig.h" #include "models/CrawlResult.h" #include "../../include/search_engine/storage/ContentStorage.h" @@ -33,19 +34,26 @@ struct CrawlSession { std::atomic isCompleted{false}; std::thread crawlThread; CrawlCompletionCallback completionCallback; - - CrawlSession(const std::string& sessionId, std::unique_ptr crawlerInstance, + // Seed URL/domain captured at start for analytics. #15 + std::string seedUrl; + std::string seedDomain; + std::chrono::system_clock::time_point startedAt; + + CrawlSession(const std::string& sessionId, std::unique_ptr crawlerInstance, CrawlCompletionCallback callback = nullptr) : id(sessionId), crawler(std::move(crawlerInstance)), createdAt(std::chrono::system_clock::now()), completionCallback(std::move(callback)) {} - + CrawlSession(CrawlSession&& other) noexcept : id(std::move(other.id)) , crawler(std::move(other.crawler)) , createdAt(other.createdAt) , isCompleted(other.isCompleted.load()) , crawlThread(std::move(other.crawlThread)) - , completionCallback(std::move(other.completionCallback)) {} + , completionCallback(std::move(other.completionCallback)) + , seedUrl(std::move(other.seedUrl)) + , seedDomain(std::move(other.seedDomain)) + , startedAt(other.startedAt) {} CrawlSession(const CrawlSession&) = delete; CrawlSession& operator=(const CrawlSession&) = delete; @@ -89,22 +97,31 @@ class CrawlerManager { // Get access to storage for logging std::shared_ptr getStorage() const { return storage_; } + // Access the session-analytics store (issue #15). Always non-null. + ISessionAnalyticsStore* getAnalyticsStore() const { return analyticsStore_.get(); } + private: std::shared_ptr storage_; std::unordered_map> sessions_; std::mutex sessionsMutex_; std::atomic sessionCounter_{0}; - + // Background cleanup thread std::thread cleanupThread_; std::atomic shouldStop_{false}; - + // Owned analytics store. #15 + std::unique_ptr analyticsStore_; + // Generate unique session ID std::string generateSessionId(); - + // Background cleanup worker void cleanupWorker(); - + // Create a new crawler instance with configuration std::unique_ptr createCrawler(const CrawlConfig& config, const std::string& sessionId = ""); -}; \ No newline at end of file + + // Build & push a SessionMetricsRecord at session completion. #15 + void recordSessionAnalytics(const CrawlSession& session, + const std::vector& results); +}; \ No newline at end of file diff --git a/tests/crawler/CMakeLists.txt b/tests/crawler/CMakeLists.txt index 6c8caa0..2e26439 100644 --- a/tests/crawler/CMakeLists.txt +++ b/tests/crawler/CMakeLists.txt @@ -5,6 +5,7 @@ add_executable(crawler_tests robots_txt_parser_tests.cpp url_frontier_tests.cpp page_fetcher_tests.cpp + session_analytics_tests.cpp ../../src/storage/ContentStorage.cpp ../../src/storage/MongoDBStorage.cpp ) diff --git a/tests/crawler/session_analytics_tests.cpp b/tests/crawler/session_analytics_tests.cpp new file mode 100644 index 0000000..da80c68 --- /dev/null +++ b/tests/crawler/session_analytics_tests.cpp @@ -0,0 +1,259 @@ +#include +#include "search_engine/crawler/SessionAnalytics.h" +#include "search_engine/crawler/SessionAnalyticsStore.h" + +#include +#include + +using namespace std::chrono_literals; + +// ---------- Helpers ---------- + +static CrawlResult makeResult(const std::string& url, + bool success, + int statusCode, + int64_t latencyMs, + int retryCount = 0, + size_t bytes = 0) { + CrawlResult r; + r.url = url; + r.statusCode = statusCode; + r.success = success; + r.crawlStatus = success ? "downloaded" : "failed"; + r.retryCount = retryCount; + r.contentSize = bytes; + r.domain = "example.com"; + auto now = std::chrono::system_clock::now(); + r.startedAt = now; + r.finishedAt = now + std::chrono::milliseconds(latencyMs); + return r; +} + +static SessionMetricsRecord makeRecord(const std::string& id, + size_t total, + size_t success, + size_t failed, + size_t retried, + int64_t durationMs, + int64_t avgLatencyMs, + std::chrono::system_clock::time_point startedAt) { + SessionMetricsRecord r; + r.sessionId = id; + r.startedAt = startedAt; + r.finishedAt = startedAt + std::chrono::milliseconds(durationMs); + r.durationMs = durationMs; + r.totalUrls = total; + r.successfulUrls = success; + r.failedUrls = failed; + r.retriedUrls = retried; + r.avgLatencyMs = avgLatencyMs; + return r; +} + +// ---------- buildFromResults ---------- + +TEST_CASE("buildFromResults computes counts and rates", "[SessionAnalytics]") { + auto now = std::chrono::system_clock::now(); + std::vector results = { + makeResult("u1", true, 200, 100, 0, 1000), + makeResult("u2", true, 200, 200, 0, 2000), + makeResult("u3", false, 500, 300, 1, 0), + makeResult("u4", true, 200, 400, 2, 500), + makeResult("u5", false, 404, 50, 0, 0), + }; + + auto r = SessionAnalytics::buildFromResults( + "s1", "https://example.com", "example.com", + now, now + 5s, results); + + REQUIRE(r.sessionId == "s1"); + REQUIRE(r.totalUrls == 5); + REQUIRE(r.successfulUrls == 3); + REQUIRE(r.failedUrls == 2); + REQUIRE(r.retriedUrls == 2); // u3 and u4 + REQUIRE(r.totalRetryAttempts == 3); // 1 + 2 + REQUIRE(r.totalBytes == 3500); // 1000+2000+500 + REQUIRE(r.statusCodeCounts[200] == 3); + REQUIRE(r.statusCodeCounts[500] == 1); + REQUIRE(r.statusCodeCounts[404] == 1); + REQUIRE(r.durationMs == 5000); + REQUIRE(r.getSuccessRate() == 0.6); + REQUIRE(r.getFailureRate() == 0.4); +} + +TEST_CASE("buildFromResults computes latency percentiles", "[SessionAnalytics]") { + auto now = std::chrono::system_clock::now(); + std::vector results = { + makeResult("u1", true, 200, 10), + makeResult("u2", true, 200, 20), + makeResult("u3", true, 200, 30), + makeResult("u4", true, 200, 40), + makeResult("u5", true, 200, 50), + makeResult("u6", true, 200, 60), + makeResult("u7", true, 200, 70), + makeResult("u8", true, 200, 80), + makeResult("u9", true, 200, 90), + makeResult("u10", true, 200, 100), + }; + auto r = SessionAnalytics::buildFromResults("s", "u", "d", now, now + 1s, results); + REQUIRE(r.avgLatencyMs == 55); // (10+20+...+100)/10 + REQUIRE(r.p50LatencyMs == 50); // 50th percentile, nearest-rank + REQUIRE(r.p95LatencyMs == 100); // ceil(0.95*10) = 10 -> idx 10 -> 100 + REQUIRE(r.maxLatencyMs == 100); +} + +TEST_CASE("buildFromResults with empty results yields empty record", "[SessionAnalytics]") { + auto now = std::chrono::system_clock::now(); + auto r = SessionAnalytics::buildFromResults("s", "u", "d", now, now + 1s, {}); + REQUIRE(r.totalUrls == 0); + REQUIRE(r.successfulUrls == 0); + REQUIRE(r.getSuccessRate() == 0.0); + REQUIRE(r.getThroughput() == 0.0); + REQUIRE(r.avgLatencyMs == 0); +} + +// ---------- summarize ---------- + +TEST_CASE("summarize rolls up multiple records", "[SessionAnalytics]") { + auto now = std::chrono::system_clock::now(); + std::vector recs = { + makeRecord("a", 10, 8, 2, 1, 1000, 100, now), + makeRecord("b", 20, 18, 2, 4, 2000, 200, now + 1s), + }; + // Manually set rates so summarize averages something meaningful. + auto s = SessionAnalytics::summarize(recs); + REQUIRE(s.sessionCount == 2); + REQUIRE(s.totalUrls == 30); + REQUIRE(s.successfulUrls == 26); + REQUIRE(s.failedUrls == 4); + REQUIRE(s.retriedUrls == 5); + REQUIRE(s.avgDurationMs == 1500); + REQUIRE(s.avgLatencyMs == 150); + // avgSuccessRate = (0.8 + 0.9) / 2 = 0.85 + REQUIRE(s.avgSuccessRate == 0.85); +} + +TEST_CASE("summarize handles empty input", "[SessionAnalytics]") { + auto s = SessionAnalytics::summarize({}); + REQUIRE(s.sessionCount == 0); + REQUIRE(s.totalUrls == 0); + REQUIRE(s.avgSuccessRate == 0.0); +} + +// ---------- compare ---------- + +TEST_CASE("compare reports B-A deltas", "[SessionAnalytics]") { + auto now = std::chrono::system_clock::now(); + auto a = makeRecord("a", 10, 5, 5, 1, 2000, 200, now); // success 0.5, retry 0.1 + auto b = makeRecord("b", 10, 9, 1, 3, 1000, 100, now); // success 0.9, retry 0.3 + auto c = SessionAnalytics::compare(a, b); + REQUIRE(c.sessionA == "a"); + REQUIRE(c.sessionB == "b"); + REQUIRE(c.successRateDelta == 0.4); // 0.9 - 0.5 + REQUIRE(c.retryRateDelta == 0.2); // 0.3 - 0.1 + REQUIRE(c.durationDeltaMs == -1000); // b is 1s shorter + REQUIRE(c.avgLatencyDeltaMs == -100); // b is 100ms faster +} + +// ---------- trends ---------- + +TEST_CASE("trends buckets sessions into hourly slices", "[SessionAnalytics]") { + // Pin a known start so bucket boundaries are predictable. + auto base = std::chrono::system_clock::time_point{} + std::chrono::hours(1000); + + std::vector recs = { + makeRecord("h0_a", 10, 8, 2, 0, 500, 50, base + 5min), + makeRecord("h0_b", 20, 16, 4, 0, 700, 70, base + 30min), + makeRecord("h1_a", 5, 5, 0, 0, 200, 20, base + 1h + 10min), + makeRecord("h2_a", 8, 4, 4, 0, 800, 80, base + 2h + 1min), + // Outside window (will be filtered out). + makeRecord("out", 1, 1, 0, 0, 100, 10, base + 24h), + }; + + auto buckets = SessionAnalytics::trends( + recs, base, base + 3h, std::chrono::hours(1)); + + REQUIRE(buckets.size() == 3); + REQUIRE(buckets[0].summary.sessionCount == 2); // h0 + REQUIRE(buckets[0].summary.totalUrls == 30); + REQUIRE(buckets[1].summary.sessionCount == 1); // h1 + REQUIRE(buckets[1].summary.totalUrls == 5); + REQUIRE(buckets[2].summary.sessionCount == 1); // h2 + REQUIRE(buckets[2].summary.totalUrls == 8); +} + +TEST_CASE("trends with zero-width bucket returns empty", "[SessionAnalytics]") { + auto now = std::chrono::system_clock::now(); + auto buckets = SessionAnalytics::trends({}, now, now + 1h, 0ms); + REQUIRE(buckets.empty()); +} + +// ---------- InMemorySessionAnalyticsStore ---------- + +TEST_CASE("InMemoryStore put/get/getAll", "[SessionAnalyticsStore]") { + InMemorySessionAnalyticsStore store; + auto now = std::chrono::system_clock::now(); + store.put(makeRecord("a", 1, 1, 0, 0, 100, 10, now)); + store.put(makeRecord("b", 2, 2, 0, 0, 200, 20, now + 1s)); + REQUIRE(store.size() == 2); + + auto g = store.get("a"); + REQUIRE(g.has_value()); + REQUIRE(g->totalUrls == 1); + REQUIRE_FALSE(store.get("missing").has_value()); + + auto all = store.getAll(); + REQUIRE(all.size() == 2); + + // limit returns most recent + auto recent = store.getAll(1); + REQUIRE(recent.size() == 1); + REQUIRE(recent[0].sessionId == "b"); +} + +TEST_CASE("InMemoryStore put is idempotent on same id", "[SessionAnalyticsStore]") { + InMemorySessionAnalyticsStore store; + auto now = std::chrono::system_clock::now(); + store.put(makeRecord("a", 1, 1, 0, 0, 100, 10, now)); + store.put(makeRecord("a", 5, 5, 0, 0, 500, 50, now)); // overwrite + REQUIRE(store.size() == 1); + auto g = store.get("a"); + REQUIRE(g.has_value()); + REQUIRE(g->totalUrls == 5); +} + +TEST_CASE("InMemoryStore evicts oldest beyond capacity", "[SessionAnalyticsStore]") { + InMemorySessionAnalyticsStore store(3); + auto now = std::chrono::system_clock::now(); + store.put(makeRecord("a", 1, 1, 0, 0, 100, 10, now)); + store.put(makeRecord("b", 1, 1, 0, 0, 100, 10, now)); + store.put(makeRecord("c", 1, 1, 0, 0, 100, 10, now)); + store.put(makeRecord("d", 1, 1, 0, 0, 100, 10, now)); // pushes a out + REQUIRE(store.size() == 3); + REQUIRE_FALSE(store.get("a").has_value()); + REQUIRE(store.get("b").has_value()); + REQUIRE(store.get("c").has_value()); + REQUIRE(store.get("d").has_value()); +} + +TEST_CASE("InMemoryStore getInWindow filters by startedAt", "[SessionAnalyticsStore]") { + InMemorySessionAnalyticsStore store; + auto base = std::chrono::system_clock::now(); + store.put(makeRecord("old", 1, 1, 0, 0, 100, 10, base - 2h)); + store.put(makeRecord("mid1", 1, 1, 0, 0, 100, 10, base - 30min)); + store.put(makeRecord("mid2", 1, 1, 0, 0, 100, 10, base - 10min)); + store.put(makeRecord("future", 1, 1, 0, 0, 100, 10, base + 10min)); + + auto inWin = store.getInWindow(base - 1h, base); + REQUIRE(inWin.size() == 2); +} + +TEST_CASE("InMemoryStore clear empties the store", "[SessionAnalyticsStore]") { + InMemorySessionAnalyticsStore store; + auto now = std::chrono::system_clock::now(); + store.put(makeRecord("a", 1, 1, 0, 0, 100, 10, now)); + REQUIRE(store.size() == 1); + store.clear(); + REQUIRE(store.size() == 0); + REQUIRE_FALSE(store.get("a").has_value()); +}