<commit_before>open Core.Std
open Lwt
open Cohttp_lwt_unix

type http_response = Response.t * Cohttp_lwt_body.t
type http_request = Request.t * Cohttp_lwt_body.t
type request_handler = Uri.t list -> http_request -> http_response Lwt.t

let (<*>) f g x = f (g x)
let locked = Hashtbl.create ~hashable:String.hashable ()
let lock_timeout = 1.5

let get_filename req =
  let path = Uri.path (Request.uri req) in
  Sys.getcwd () ^ path

let critical_error e =
  let body = Exn.to_string_mach e in
  print_endline body;
  Server.respond_string ~status:`Internal_server_error ~body ()

let conflict = Server.respond_string ~status:`Conflict ~body:""
let not_found = Server.respond_string ~status:`Not_found ~body:""
let ok = Server.respond_string ~status:`OK ~body:""

let forward_to_others ips meth (req, body) =
  match Cohttp.Header.get (Request.headers req) "forwarded" with
  | Some _ -> None
  | None -> Some (
    Lwt_list.map_p (fun remote ->
      let uri = Request.uri req
      |> Fn.flip Uri.with_host (Uri.host remote)
      |> Fn.flip Uri.with_port (Uri.port remote)
      in
      let headers = Cohttp.Header.init_with "forwarded" "true" in
      Client.call ~headers ~body ~chunked:false meth uri
    ) ips)

let only_one_response ls =
  List.filter ~f:(function (resp, _) -> Response.status resp = `OK) ls
  |> function
  | [] -> not_found ()
  | x :: [] -> return x
  | xs ->
    let xs' = List.map ~f:(Sexp.to_string <*> Response.sexp_of_t <*> fst) xs in
    critical_error (Failure (List.to_string ~f:Fn.id xs'))

let get ips (req, body) =
  let add_trailing_slash_if_directory root filename =
    Lwt_unix.stat (root ^ "/" ^ filename)
    >|= (fun stats ->
      match stats.Lwt_unix.st_kind with
      | Unix.S_DIR -> filename ^ "/"
      | _ -> filename)
  in
  let get_directory_content path =
    Lwt_unix.files_of_directory path
    |> Lwt_stream.map_s (add_trailing_slash_if_directory path)
    |> Lwt_stream.to_list
  in
  let fname = get_filename req in
  Lwt_unix.stat fname
  >>= fun stats ->
    match Lwt_unix.(stats.st_kind) with
    | Unix.S_DIR ->
      Option.value ~default:(return []) (forward_to_others ips `GET (req, body))
      >>= fun contents ->
        if List.for_all ~f:((=) `OK <*> Response.status <*> fst) contents then
          (get_directory_content fname
          >>= (fun local_content ->
              contents
              |> Lwt_list.map_p (Cohttp_lwt_body.to_string <*> snd)
              >|= (List.join <*> List.map ~f:String.split_lines)
              >|= List.append local_content
              >|= List.dedup ~compare:String.compare
              >|= List.fold ~init:"" ~f:(fun acc -> fun file -> file ^ "\n" ^ acc))
          >>= fun body -> Server.respond_string
            ~headers:(Cohttp.Header.init_with "is-directory" "true")
            ~status:`OK
            ~body
            ())
        else
          let contents' =
            List.map ~f:(Sexp.to_string <*> Response.sexp_of_t <*> fst) contents
          in
          critical_error (Failure (List.to_string ~f:Fn.id contents'))
    | _ -> Server.respond_file ~fname ()
  >>= (function (r, _) as resp ->
    match Response.status r with
    | `Not_found ->
      Option.value_map
        ~default:(not_found ())
        ~f:(Fn.flip Lwt.bind only_one_response)
        (forward_to_others ips `GET (req, body))
    | _ -> return resp)

let lock ips (req, body) =
  let fname = get_filename req in
  try_lwt (
    Lwt_io.with_file ~flags:[Unix.O_RDONLY] ~mode:Lwt_io.input fname (fun _ -> conflict ())
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
    (* The file doesn't exist, lock it and return OK *)
    Hashtbl.set locked fname true;
    ignore (Lwt_unix.timeout lock_timeout >|= fun () -> Hashtbl.remove locked fname);
    ok ()
  | e -> critical_error e

let post ips (req, body) =
  let fname = get_filename req in
  match Hashtbl.find locked fname with
  | Some el -> conflict ()
  | None ->
    match forward_to_others ips (`Other "LOCK") ((Request.make ~meth:(`Other "LOCK") (Request.uri req)), body) with
    | None -> critical_error (Failure "Impossible case, LOCK cannot be forwarded")
    | Some x ->
      x >>= fun responses ->
        if List.for_all ~f:((=) `OK <*> Response.status <*> fst) responses then
          (try_lwt (
            Lwt_io.with_file ~flags:[Unix.O_WRONLY; Unix.O_CREAT] ~mode:Lwt_io.output fname (fun ch ->
              Lwt_io.write ch "" >>= ok
            )
          ) with e -> critical_error e)
        else conflict ()

let put ips (req, body) =
  let filename = get_filename req in
  let flags = [Unix.O_WRONLY; Unix.O_TRUNC] in
  let mode = Lwt_io.output in
  try_lwt (
    Lwt_io.with_file ~flags ~mode filename (Fn.flip Cohttp_lwt_body.write_body body <*> Lwt_io.write)
    >>= ok
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
    Option.value_map
      ~default:(not_found ())
      ~f:(Fn.flip Lwt.bind only_one_response)
      (forward_to_others ips `PUT (req, body))
  | e -> critical_error e

let delete ips (req, body) =
  let filename = get_filename req in
  try_lwt (
    Lwt_unix.unlink filename >>= ok
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
    Option.value_map
      ~default:(not_found ())
      ~f:(Fn.flip Lwt.bind only_one_response)
      (forward_to_others ips `DELETE (req, body))
  | e -> critical_error e

let callback _ ips http_request =
  ignore (List.map ~f:Uri.to_string ips |> String.concat ~sep:"  " |> print_endline);
  try_lwt (
    match Request.meth (fst http_request) with
    | `GET -> get ips http_request
    | `Other "LOCK" -> lock ips http_request
    | `POST -> Lwt.pick [post ips http_request; Lwt_unix.timeout lock_timeout >>= fun () -> critical_error (Failure "Timeout while acquiring lock")]
    | `PUT -> put ips http_request
    | `DELETE -> delete ips http_request
    | meth -> Server.respond_string ~status:`Method_not_allowed ~body:("Method unimplemented: " ^ Cohttp.Code.string_of_method meth) ()
  ) with e -> critical_error e

let make_server port ips () =<commit_msg>Working 'ls' on directories, Yeahgit add http.ml !<commit_after>open Core.Std
open Lwt
open Cohttp_lwt_unix

type http_response = Response.t * Cohttp_lwt_body.t
type http_request = Request.t * Cohttp_lwt_body.t
type request_handler = Uri.t list -> http_request -> http_response Lwt.t

let (<*>) f g x = f (g x)
let locked = Hashtbl.create ~hashable:String.hashable ()
let lock_timeout = 1.5

let get_filename req =
  let path = Uri.path (Request.uri req) in
  Sys.getcwd () ^ path

let critical_error e =
  let body = Exn.to_string_mach e in
  print_endline body;
  Server.respond_string ~status:`Internal_server_error ~body ()

let conflict = Server.respond_string ~status:`Conflict ~body:""
let not_found = Server.respond_string ~status:`Not_found ~body:""
let ok = Server.respond_string ~status:`OK ~body:""

let forward_to_others ips meth (req, body) =
  match Cohttp.Header.get (Request.headers req) "forwarded" with
  | Some _ -> None
  | None -> Some (
    Lwt_list.map_p (fun remote ->
      let uri = Request.uri req
      |> Fn.flip Uri.with_scheme (Uri.scheme remote)
      |> Fn.flip Uri.with_host (Uri.host remote)
      |> Fn.flip Uri.with_port (Uri.port remote)
      in
      let headers = Cohttp.Header.init_with "forwarded" "true" in
      Client.call ~headers ~body ~chunked:false meth uri
    ) ips)

let only_one_response ls =
  List.filter ~f:(function (resp, _) -> Response.status resp = `OK) ls
  |> function
  | [] -> not_found ()
  | x :: [] -> return x
  | xs ->
    let xs' = List.map ~f:(Sexp.to_string <*> Response.sexp_of_t <*> fst) xs in
    critical_error (Failure (List.to_string ~f:Fn.id xs'))

let get ips (req, body) =
  let add_trailing_slash_if_directory root filename =
    Lwt_unix.stat (root ^ "/" ^ filename)
    >|= (fun stats ->
      match stats.Lwt_unix.st_kind with
      | Unix.S_DIR -> filename ^ "/"
      | _ -> filename)
  in
  let get_directory_content path =
    Lwt_unix.files_of_directory path
    |> Lwt_stream.map_s (add_trailing_slash_if_directory path)
    |> Lwt_stream.to_list
  in
  let fname = get_filename req in
  try_lwt (
    Lwt_unix.stat fname
    >>= fun stats ->
      match Lwt_unix.(stats.st_kind) with
      | Unix.S_DIR ->
        Option.value ~default:(return []) (forward_to_others ips `GET (req, body))
        >>= fun contents ->
          if List.for_all ~f:((List.mem [`OK; `Not_found]) <*> Response.status <*> fst) contents then
            (get_directory_content fname
            >>= (fun local_content ->
                Lwt_list.map_p (Cohttp_lwt_body.to_string <*> snd) contents
                >|= (List.join <*> List.map ~f:String.split_lines)
                >|= List.append local_content
                >|= List.dedup ~compare:String.compare
                >|= List.fold ~init:"" ~f:(fun acc -> fun file -> file ^ "\n" ^ acc))
            >>= fun body -> Server.respond_string
              ~headers:(Cohttp.Header.init_with "is-directory" "true")
              ~status:`OK
              ~body
              ())
          else
            let contents' =
              List.map ~f:(Sexp.to_string <*> Response.sexp_of_t <*> fst) contents
            in
            critical_error (Failure (List.to_string ~f:Fn.id contents'))
      | Unix.S_REG | Unix.S_LNK -> Server.respond_file ~fname ()
      | _ -> critical_error (Failure "Unsupported file kind")
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
      Option.value_map
        ~default:(not_found ())
        ~f:(Fn.flip Lwt.bind only_one_response)
        (forward_to_others ips `GET (req, body))
  | e -> critical_error e

let lock ips (req, body) =
  let fname = get_filename req in
  try_lwt (
    Lwt_io.with_file ~flags:[Unix.O_RDONLY] ~mode:Lwt_io.input fname (fun _ -> conflict ())
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
    (* The file doesn't exist, lock it and return OK *)
    Hashtbl.set locked fname true;
    ignore (Lwt_unix.timeout lock_timeout >|= fun () -> Hashtbl.remove locked fname);
    ok ()
  | e -> critical_error e

let post ips (req, body) =
  let fname = get_filename req in
  match Hashtbl.find locked fname with
  | Some el -> conflict ()
  | None ->
    match forward_to_others ips (`Other "LOCK") ((Request.make ~meth:(`Other "LOCK") (Request.uri req)), body) with
    | None -> critical_error (Failure "Impossible case, LOCK cannot be forwarded")
    | Some x ->
      x >>= fun responses ->
        if List.for_all ~f:((=) `OK <*> Response.status <*> fst) responses then
          (try_lwt (
            Lwt_io.with_file ~flags:[Unix.O_WRONLY; Unix.O_CREAT] ~mode:Lwt_io.output fname (fun ch ->
              Lwt_io.write ch "" >>= ok
            )
          ) with e -> critical_error e)
        else conflict ()

let put ips (req, body) =
  let filename = get_filename req in
  let flags = [Unix.O_WRONLY; Unix.O_TRUNC] in
  let mode = Lwt_io.output in
  try_lwt (
    Lwt_io.with_file ~flags ~mode filename (Fn.flip Cohttp_lwt_body.write_body body <*> Lwt_io.write)
    >>= ok
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
    Option.value_map
      ~default:(not_found ())
      ~f:(Fn.flip Lwt.bind only_one_response)
      (forward_to_others ips `PUT (req, body))
  | e -> critical_error e

let delete ips (req, body) =
  let filename = get_filename req in
  try_lwt (
    Lwt_unix.unlink filename >>= ok
  ) with
  | Unix.Unix_error (Unix.ENOENT, _, _) ->
    Option.value_map
      ~default:(not_found ())
      ~f:(Fn.flip Lwt.bind only_one_response)
      (forward_to_others ips `DELETE (req, body))
  | e -> critical_error e

let callback _ ips ((req, _) as http_request) =
  try_lwt (
    ignore (Lwt_io.printf "%s %s\n%s"
      (Cohttp.Code.string_of_method (Request.meth req))
      (Uri.to_string (Request.uri req))
      (Cohttp.Header.to_string (Request.headers req)));
    match Request.meth (fst http_request) with
    | `GET -> get ips http_request
    | `Other "LOCK" -> lock ips http_request
    | `POST -> Lwt.pick [post ips http_request; Lwt_unix.timeout lock_timeout >>= fun () -> critical_error (Failure "Timeout while acquiring lock")]
    | `PUT -> put ips http_request
    | `DELETE -> delete ips http_request
    | meth -> Server.respond_string ~status:`Method_not_allowed ~body:("Method unimplemented: " ^ Cohttp.Code.string_of_method meth) ()
  ) with e -> critical_error e

let make_server port ips () =
{'fraction_non_alphanumeric': 0.12885222033487018, 'fraction_numerical': 0.000323546064870986, 'mean_word_length': 2.7798838275756648, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 183, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '427238', 'n_tokens_mistral': 4407, 'n_tokens_neox': 4230, 'n_words': 1085}