#include "prism_engineering/asset_pipeline.hpp"
#include <iostream>
#include <limits>
#include <stdexcept>
#include <string_view>
using namespace prism_engineering;
namespace {
int checks{};
void check(bool value, std::string_view message) {
  if (!value) throw std::runtime_error{std::string{message}};
  ++checks;
}
std::unique_ptr<AssetPipeline> make(PipelineOptions options = {}) {
  auto result = AssetPipeline::create(options);
  check(result.has_value(), "pipeline creation");
  return std::move(*result);
}
void limits_and_shutdown() {
  check(!AssetPipeline::create({0,{}}), "zero workers rejected");
  check(!AssetPipeline::create({AssetPipeline::worker_limit + 1,{}}), "worker upper bound");
  for (const auto index : {0U, 1U}) {
    auto failed = AssetPipeline::create({2,index});
    check(!failed && failed.error() == Error::thread_start, "partial launch cleanup");
  }
  auto pipeline = make({1,{}});
  check(!pipeline->try_submit({1,1,8,64,{},false}), "gradient extent minimum");
  check(!pipeline->try_submit({1,65,8,64,{},false}), "extent upper bound");
  check(!pipeline->try_submit({0,8,8,64,{},false}), "generation zero rejected");
  auto gate = std::make_shared<WorkGate>();
  check(pipeline->try_submit({1,8,8,64,gate}).has_value(), "admission");
  check(gate->wait_for_arrivals(1), "worker is inside closed gate");
  check(!pipeline->try_take(), "poll returns while work is blocked");
  for (std::uint64_t generation = 2; generation <= AssetPipeline::capacity; ++generation)
    check(pipeline->try_submit({generation,8,8,64,{},false}).has_value(), "fill bounded admission");
  auto full = pipeline->try_submit({9,8,8,64,{},false});
  check(!full && full.error() == Admission::full, "full rejects without waiting");
  pipeline->shutdown(); // Closed gate observes stop; no sleep or forced release.
  check(!pipeline->try_submit({10,8,8,64,{},false}), "closed rejects work");
  std::size_t cancelled{};
  while (auto result = pipeline->try_take()) {
    check(result->status == Status::cancelled && result->image.pixels.empty(), "queued and active cancellation");
    ++cancelled;
  }
  check(cancelled == AssetPipeline::capacity, "all admitted requests have terminal result");
  check(!pipeline->wait_take(), "joined empty service does not wait");
  pipeline->shutdown(); // Idempotent owner-thread shutdown.
  check(!pipeline->runtime_failed(), "no worker runtime failure");
}
void ordered_publication() {
  auto pipeline = make();
  Gallery gallery;
  auto old_gate = std::make_shared<WorkGate>();
  check(gallery.request(*pipeline,{1,8,8,10,old_gate}).has_value(), "request A");
  check(old_gate->wait_for_arrivals(1), "A started but cannot finish");
  check(gallery.request(*pipeline,{2,8,8,20,{},false}).has_value(), "request B");
  check(!gallery.apply({1,Status::failed,{}}) && gallery.latest_status()==Status::pending, "stale failure cannot settle current request");
  auto b = pipeline->wait_take();
  check(b && b->generation == 2 && b->status == Status::succeeded, "B completes first");
  check(gallery.apply(std::move(*b)), "publish B");
  check(gallery.image().pixels.size() == 64, "owned 8x8 output");
  check(gallery.image().pixels.front() == Rgb{0,0,20}, "independent first pixel");
  check(gallery.image().pixels.back() == Rgb{255,255,20}, "independent last pixel");
  old_gate->open();
  auto a = pipeline->wait_take();
  check(a && a->generation == 1, "A completes last");
  check(!gallery.apply(std::move(*a)) && gallery.published_generation() == 2, "stale A cannot replace B");
  check(!gallery.apply({2,Status::succeeded,{}}), "duplicate completion rejected");
  check(gallery.request(*pipeline,{3,8,8,30,{},true}).has_value(), "failed candidate accepted for preparation");
  auto failed = pipeline->wait_take();
  check(failed && failed->status == Status::failed, "worker exception contained");
  check(!gallery.apply(std::move(*failed)), "failure is not publication");
  check(gallery.latest_status() == Status::failed && gallery.published_generation() == 2 &&
        gallery.image().pixels.front() == Rgb{0,0,20}, "prior valid image retained");
  auto invalid = gallery.request(*pipeline,{4,0,8,64,{},false});
  check(!invalid && gallery.requested_generation() == 3, "rejected request does not advance revision");
  auto cancel_gate = std::make_shared<WorkGate>();
  check(gallery.request(*pipeline,{4,8,8,40,cancel_gate}).has_value(), "replacement awaiting cancel");
  check(cancel_gate->wait_for_arrivals(1), "replacement in gate");
  pipeline->request_stop();
  pipeline->shutdown();
  auto cancelled = pipeline->try_take();
  check(cancelled && cancelled->status == Status::cancelled, "cancelled replacement result");
  check(!gallery.apply(std::move(*cancelled)) && gallery.published_generation() == 2, "cancel preserves B");
}
void retained_results_and_generation() {
  auto pipeline = make({1,{}});
  auto last_gate=std::make_shared<WorkGate>();
  // With one worker and no consumption, slots are claimed in index order.
  // Arrival at slot 8 therefore proves the preceding 7 results are retained.
  for (std::uint64_t generation=1; generation<AssetPipeline::capacity; ++generation)
    check(pipeline->try_submit({generation,2,2,64,{},false}).has_value(), "bounded retained slots");
  check(pipeline->try_submit({8,2,2,64,last_gate,false}).has_value(), "last slot gate");
  check(last_gate->wait_for_arrivals(1), "seven completed results retained before final gate");
  auto full = pipeline->try_submit({9,2,2,64,{},false});
  check(!full && full.error()==Admission::full, "completed results do not silently free admission");
  auto result = pipeline->wait_take();
  check(result.has_value(), "consume one slot");
  check(pipeline->try_submit({9,2,2,64,{},false}).has_value(), "consumption restores one slot");
  last_gate->open();
  pipeline->shutdown();
  auto exhaustion = make({1,{}});
  const auto maximum = std::numeric_limits<std::uint64_t>::max();
  check(exhaustion->try_submit({maximum,2,2,64,{},false}).has_value(), "maximum identifier accepted without increment");
  check(exhaustion->wait_take().has_value(), "maximum generation completes");
  auto repeated = exhaustion->try_submit({maximum,2,2,64,{},false});
  check(!repeated && repeated.error()==Admission::invalid_generation, "identifier reuse rejected");
  check(!exhaustion->try_submit({0,2,2,64,{},false}), "wrapped identifier rejected");
}
}
int main() {
  try {
    limits_and_shutdown(); ordered_publication(); retained_results_and_generation();
    std::cout << checks << " async image pipeline checks passed\n";
    return 0;
  } catch (const std::exception& error) { std::cerr << error.what() << '\n'; return 1; }
}