main.cpp

#include <chrono>
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <iomanip>
#include <iostream>
#include <string>
#include <thread>

#include "aggregator.h"
#include "mmap_reader.h"
#include "thread_pool.h"

namespace {

struct Options {
  std::string input;
  unsigned threads = 0;       // 0 -> hardware_concurrency
  std::size_t top_endpoints = 10;
  std::size_t chunk_mb = 16;  // spec recommends 16/32/64 MB chunks
};

void print_usage(const char* prog) {
  std::cerr << "Usage: " << prog << " --input <file> [--threads N] "
            << "[--top-endpoints N] [--chunk-mb N]\n";
}

// Returns the value for a flag given as "--flag value" or "--flag=value".
// Advances i when the value is a separate argv element.
bool take_value(int argc, char** argv, int& i, std::string_view flag, std::string& out) {
  std::string_view arg = argv[i];
  if (arg == flag) {
    if (i + 1 >= argc) return false;
    out = argv[++i];
    return true;
  }
  std::string eq = std::string(flag) + "=";
  if (arg.substr(0, eq.size()) == eq) {
    out = std::string(arg.substr(eq.size()));
    return true;
  }
  return false;
}

bool parse_args(int argc, char** argv, Options& opt) {
  for (int i = 1; i < argc; ++i) {
    std::string v;
    std::string_view arg = argv[i];
    if (arg == "-h" || arg == "--help") return false;
    if (take_value(argc, argv, i, "--input", v)) {
      opt.input = v;
    } else if (take_value(argc, argv, i, "--threads", v)) {
      opt.threads = static_cast<unsigned>(std::stoul(v));
    } else if (take_value(argc, argv, i, "--top-endpoints", v)) {
      opt.top_endpoints = static_cast<std::size_t>(std::stoul(v));
    } else if (take_value(argc, argv, i, "--chunk-mb", v)) {
      opt.chunk_mb = static_cast<std::size_t>(std::stoul(v));
    } else if (opt.input.empty() && arg[0] != '-') {
      opt.input = std::string(arg);  // allow positional input path
    } else {
      std::cerr << "Unknown or malformed argument: " << arg << "\n";
      return false;
    }
  }
  return !opt.input.empty();
}

void print_report(const Report& r, const Options& opt, unsigned threads,
                  double seconds, std::size_t bytes) {
  std::cout << std::fixed << std::setprecision(1);
  std::cout << "\n==== Log Analytics Report ====\n";
  std::cout << "Total Requests:  " << r.total_requests << "\n";
  std::cout << "Total Errors:    " << r.total_errors << "\n";
  std::cout << "Error Rate:      " << (r.error_rate * 100.0) << "%\n";
  if (r.skipped > 0) std::cout << "Skipped (bad):   " << r.skipped << "\n";
  std::cout << "Average Latency: " << r.avg_latency << " ms\n";
  std::cout << "P95 Latency:     " << r.p95_latency << " ms\n";

  std::cout << "\nTop " << opt.top_endpoints << " Endpoints (of "
            << r.distinct_endpoints << "):\n";
  for (const auto& e : r.top_endpoints) {
    std::cout << "  " << std::left << std::setw(16) << e.label << std::right
              << e.count << "\n";
  }

  std::cout << "\nRequests/min:    avg " << r.avg_requests_per_minute << " over "
            << r.distinct_minutes << " minutes";
  if (!r.peak_minute.empty()) {
    std::cout << " (peak " << r.peak_minute_requests << " at " << r.peak_minute << ")";
  }
  std::cout << "\n";

  if (!r.top_users.empty()) {
    std::cout << "\nTop Users (of " << r.distinct_users << "):\n";
    for (const auto& u : r.top_users) {
      std::cout << "  user=" << std::left << std::setw(12) << u.label << std::right
                << u.count << "\n";
    }
  }

  const double mb = static_cast<double>(bytes) / (1024.0 * 1024.0);
  std::cout << "\n---- Performance ----\n";
  std::cout << "Threads:    " << threads << "\n";
  std::cout << "Processed:  " << mb << " MB in " << seconds * 1000.0 << " ms\n";
  std::cout << "Throughput: " << (seconds > 0 ? mb / seconds : 0.0) << " MB/s\n";
}

}  // namespace

int main(int argc, char** argv) {
  Options opt;
  if (!parse_args(argc, argv, opt)) {
    print_usage(argv[0]);
    return 2;
  }

  unsigned threads = opt.threads;
  if (threads == 0) {
    threads = std::thread::hardware_concurrency();
    if (threads == 0) threads = 1;
  }

  try {
    const auto t0 = std::chrono::steady_clock::now();

    MmapReader reader(opt.input);
    const std::size_t target = opt.chunk_mb * 1024 * 1024;
    std::vector<Chunk> chunks = reader.chunks(target);

    ThreadPool pool(threads);
    for (const Chunk& c : chunks) pool.submit(c);
    pool.close_and_join();
    Report report = merge(pool.stats(), opt.top_endpoints);

    const auto t1 = std::chrono::steady_clock::now();
    const double seconds = std::chrono::duration<double>(t1 - t0).count();

    print_report(report, opt, threads, seconds, reader.size());
    return 0;
  } catch (const std::exception& ex) {
    std::cerr << "Error: " << ex.what() << "\n";
    return 1;
  }
}