type capture = {
  stdout : string;
  stderr : string;
  status : Report.process_status;
}

let read_pipe ~sw source =
  let promise, resolver = Eio.Promise.create () in
  Eio.Fiber.fork ~sw (fun () ->
      let result =
        try
          let reader =
            Eio.Buf_read.of_flow ~max_size:(128 * 1024 * 1024) source
          in
          Ok (Eio.Buf_read.take_all reader)
        with exn -> Error (Printexc.to_string exn)
      in
      Eio.Promise.resolve resolver result);
  promise

let process_status = function
  | `Exited code -> Report.Exited code
  | `Signaled signal -> Report.Signaled signal

let capture_process ~clock ~process_mgr ~timeout command =
  try
    Eio.Switch.run @@ fun sw ->
    let stdout_source, stdout_sink = Eio.Process.pipe ~sw process_mgr in
    let stderr_source, stderr_sink = Eio.Process.pipe ~sw process_mgr in
    let process =
      Eio.Process.spawn ~sw process_mgr ~stdout:stdout_sink ~stderr:stderr_sink
        command
    in
    Eio.Flow.close stdout_sink;
    Eio.Flow.close stderr_sink;
    let stdout = read_pipe ~sw stdout_source in
    let stderr = read_pipe ~sw stderr_source in
    let status =
      match
        Eio.Time.with_timeout clock timeout (fun () ->
            Ok (Eio.Process.await process))
      with
      | Ok status -> process_status status
      | Error `Timeout ->
          Eio.Process.signal process Sys.sigkill;
          ignore (Eio.Process.await process);
          Report.Timed_out
    in
    let captured promise stream =
      match Eio.Promise.await promise with
      | Ok bytes -> (bytes, "")
      | Error message -> ("", stream ^ " capture failed: " ^ message ^ "\n")
    in
    let stdout, stdout_error = captured stdout "stdout" in
    let stderr, stderr_error = captured stderr "stderr" in
    { stdout; stderr = stderr ^ stdout_error ^ stderr_error; status }
  with exn ->
    {
      stdout = "";
      stderr = "";
      status = Report.Spawn_error (Printexc.to_string exn);
    }

let parse_observations ~client stdout =
  let observations = ref [] in
  let invalid = ref [] in
  String.split_on_char '\n' stdout
  |> List.iteri (fun index input ->
      let line = index + 1 in
      if String.trim input <> "" then
        match
          Json.decode
            ~file:(Printf.sprintf "%s:stdout:%d" client line)
            Json.observation input
        with
        | Ok observation ->
            observations := (line, input, observation) :: !observations
        | Error error ->
            invalid :=
              { Report.line = Some line; case_id = None; error; input }
              :: !invalid);
  (List.rev !observations, List.rev !invalid)

let parse_conformance_observations ~client stdout =
  let observations = ref [] in
  let invalid = ref [] in
  String.split_on_char '\n' stdout
  |> List.iteri (fun index input ->
      let line = index + 1 in
      if String.trim input <> "" then
        match
          Json.decode
            ~file:(Printf.sprintf "%s:stdout:%d" client line)
            Conformance_json.observation input
        with
        | Ok observation ->
            observations := (line, input, observation) :: !observations
        | Error error ->
            invalid :=
              { Conformance.line = Some line; case_id = None; error; input }
              :: !invalid);
  (List.rev !observations, List.rev !invalid)

let run ~net ~clock ~process_mgr ~client ~command ?(cases = Corpus.all)
    ?(selected_tags = []) ?(timeout = 120.) () =
  if command = [] then invalid_arg "command must not be empty";
  if timeout <= 0. then invalid_arg "timeout must be positive";
  Eio.Switch.run @@ fun sw ->
  let listening, set_listening = Eio.Promise.create () in
  Eio.Fiber.fork_daemon ~sw (fun () ->
      Server.run ~sw ~net ~clock ~cases
        ~addr:(`Tcp (Eio.Net.Ipaddr.V4.loopback, 0))
        ~on_listening:(Eio.Promise.resolve set_listening)
        ());
  let port =
    match Eio.Promise.await listening with
    | `Tcp (_, port) -> port
    | `Unix _ -> assert false
  in
  let base_url = Printf.sprintf "http://127.0.0.1:%d" port in
  let command = command @ [ base_url ] in
  let started = Eio.Time.now clock in
  let capture = capture_process ~clock ~process_mgr ~timeout command in
  let duration_ms = (Eio.Time.now clock -. started) *. 1000. in
  let observations, invalid_observations =
    parse_observations ~client capture.stdout
  in
  Report.make ~cases ~selected_tags ~client ~command ~base_url ~duration_ms
    ~process_status:capture.status ~stderr:capture.stderr ~observations
    ~invalid_observations ()

let run_conformance ~net ~clock ~process_mgr ~profile ~client ~command
    ?(cases = Conformance.cases profile) ?(selected_tags = []) ?(timeout = 120.)
    () =
  if command = [] then invalid_arg "command must not be empty";
  if timeout <= 0. then invalid_arg "timeout must be positive";
  Eio.Switch.run @@ fun sw ->
  let listening, set_listening = Eio.Promise.create () in
  Eio.Fiber.fork_daemon ~sw (fun () ->
      Server.run ~sw ~net ~clock
        ~addr:(`Tcp (Eio.Net.Ipaddr.V4.loopback, 0))
        ~on_listening:(Eio.Promise.resolve set_listening)
        ());
  let port =
    match Eio.Promise.await listening with
    | `Tcp (_, port) -> port
    | `Unix _ -> assert false
  in
  let base_url = Printf.sprintf "http://127.0.0.1:%d" port in
  let command = command @ [ base_url ] in
  let started = Eio.Time.now clock in
  let capture = capture_process ~clock ~process_mgr ~timeout command in
  let duration_ms = (Eio.Time.now clock -. started) *. 1000. in
  let observations, invalid_observations =
    parse_conformance_observations ~client capture.stdout
  in
  Conformance.make_report ~profile ~cases ~selected_tags ~client ~command
    ~base_url ~duration_ms ~process_status:capture.status ~stderr:capture.stderr
    ~observations ~invalid_observations ()