Source file ThunkIo.ml

(** Make an abstraction over a file. The file does not need to be on disk; it
    can be a remote file accessible through a URL, or in-memory.

    {3 Usage}

    {[
      module BasicIo = struct
        module P = MlFront_Thunk.Promises.PromiseMinimal
        include MlFront_Thunk.ThunkIo.Make (P)
      end
    ]}

    and using it with ThunkCst:

    {[
      let module Parser = MlFront_Thunk.ThunkCst.Io (BasicIo.P) in
      let parse_promise =
        Parser.parse (module ResultObserver) (BasicIo.disk_file thunkfile)
      in
      let parse_result = BasicIo.P.run_promise parse_promise in
      match parse_result with _ -> ()
    ]}

    {3 Extending the abstraction}

    You will want an extension if you do asynchronous file reading or read
    "cloud" files from the Internet.

    Here is a functional equivalent of reading from disk:

    {[
      module ExtendedIo = struct
        module P = MlFront_Thunk.Promises.PromiseMinimal
        (** You need some promise implementation. *)

        include MlFront_Thunk.ThunkIo.Make (P)

        (** You need a creator of file objects. Your implementation will use
            [generic_file] to define all the messages that a file object must
            response to. *)
        let a_disk_file origin =
          generic_file ~origin ~read_all:(fun () ->
              let content =
                In_channel.with_open_bin origin In_channel.input_all
              in
              P.return (`Content content))
      end
    ]}

    and using it with ThunkCst:

    {[
      let module Parser = MlFront_Thunk.ThunkCst.Io (ExtendedIo.P) in
      let parse_promise =
        Parser.parse (module ResultObserver) (ExtendedIo.disk_file thunkfile)
      in
      let parse_result = ExtendedIo.P.run_promise parse_promise in
      match parse_result with _ -> ()
    ]} *)
module Make (M : BuildConstraints.MONAD_PROMISE) : sig
  type file_object = private {
    file_origin : string;
        (** The location of the file. It should be clickable if possible by
            modern IDEs in a terminal. For example, a file should just be a path
            to the file (no ["file://"] prefix), while a http URL should be a
            URL. *)
    is_local_file : bool;
        (** [is_local_file] is [true] if and only if the native operating system
            tools like {!In_channel} can manipulate the file. *)
    open_for_writing :
      unit ->
      [ `Node of Int64.t | `IsDirectory of directory_object | `Error of string ]
      M.t;
        (** [open_for_writing path] opens the file at [path] and returns a
            promise that resolves to a file node (ex. inode) if successful, or
            an error message if not. *)
    open_for_reading :
      unit ->
      [ `Node of Int64.t | `IsDirectory of directory_object | `Error of string ]
      M.t;
        (** [open_for_reading path] opens the file at [path] and returns a
            promise that resolves to a file node (ex. inode) if successful, or
            an error message if not. *)
    read_some :
      Int64.t -> [ `Bytes of bytes * int * int | `Eof | `Error of string ] M.t;
    write_all :
      Int64.t -> string -> int -> int -> [ `WroteBytes | `Error of string ] M.t;
    close : Int64.t -> unit M.t;
        (** [close node] closes the file opened with {!open_for_writing}. *)
    read_all :
      unit ->
      [ `Content of string | `Error of string | `ExceededSizeLimit of int64 ]
      M.t;
        (** [read_all source] gets the file, but may not give back the file if
            it exceeds memory.

            The [error] in [`Error error], if possible, should complete the
            phrase ["Cannot read the file because ___."]. For example,
            [`Error "the file was not found"]. *)
    prepare_as_copy_destination : unit -> [ `Ready | `Error of string ] M.t;
        (** [prepare_as_copy_destination ()] prepares the file for being the
            destination target of a copy. On Windows the file may be set to
            read-write. *)
    delete_file : unit -> [ `Deleted | `Error of string ] M.t;
        (** [delete_file node] deletes the file. *)
    checksum_file :
      strip_carriage_returns:bool ->
      algo:[ `Sha1 | `Sha256 ] ->
      unit ->
      [ `Checksum of string * Int64.t | `Error of string ] M.t;
  }

  and directory_object = private {
    dir_origin : string;
        (** The location of the directory. It should be clickable if possible by
            modern IDEs in a terminal. For example, a file should just be a path
            to the file (no ["file://"] prefix), while a http URL should be a
            URL. *)
    create_directory : unit -> [ `Created | `Error of string ] M.t;
        (** Create the directory and all parent directories if they do not
            exist. *)
    delete_directory : unit -> [ `Deleted | `Error of string ] M.t;
        (** Delete the directory and all of its content. *)
    zip_directory :
      ?intermediate:unit ->
      staging_dir:directory_object ->
      unit ->
      [ `ZipFile of file_object | `Error of string ] M.t;
        (** Create a zip archive of the directory, placing it in the
            [staging_dir] directory. *)
    spawn_in_directory :
      command:MlFront_Core.FilePath.t ->
      args:string list ->
      envmods:MlFront_Core.EnvMods.t ->
      stdout:file_object ->
      stderr:file_object ->
      unit ->
      [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
      M.t;
        (** Spawns the command in the directory. If the directory is remote, the
            semantics are to run the command remotely. *)
    interactive_shell_in_directory :
      ?promptname:string ->
      envmods:MlFront_Core.EnvMods.t ->
      unit ->
      [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
      M.t;
        (** Spawns a shell in the directory. If the directory is remote, it is
            an error.

            It is implementation dependent if this function returns or if the
            current program is replaced by the shell. *)
  }

  (** {1 Constructors and Accessors} *)

  val generic_file :
    origin:string ->
    is_local_file:bool ->
    open_for_writing:
      (unit ->
      [ `Node of Int64.t | `IsDirectory of directory_object | `Error of string ]
      M.t) ->
    open_for_reading:
      (unit ->
      [ `Node of Int64.t | `IsDirectory of directory_object | `Error of string ]
      M.t) ->
    read_some:
      (Int64.t -> [ `Bytes of bytes * int * int | `Eof | `Error of string ] M.t) ->
    write_all:
      (Int64.t ->
      string ->
      int ->
      int ->
      [ `WroteBytes | `Error of string ] M.t) ->
    close:(Int64.t -> unit M.t) ->
    read_all:
      (unit ->
      [ `Content of string | `Error of string | `ExceededSizeLimit of int64 ]
      M.t) ->
    prepare_as_copy_destination:(unit -> [ `Ready | `Error of string ] M.t) ->
    delete_file:(unit -> [ `Deleted | `Error of string ] M.t) ->
    checksum_file:
      (strip_carriage_returns:bool ->
      algo:[ `Sha1 | `Sha256 ] ->
      unit ->
      [ `Checksum of string * Int64.t | `Error of string ] M.t) ->
    unit ->
    file_object
  (** [generic_file ~origin ~is_local_file ~open_for_writing ~open_for_reading
       ~read_some ~write_all ~close ~read_all ~prepare_as_copy_destination
       ~delete_file ~checksum_file] creates a new file object. *)

  val generic_dir :
    origin:string ->
    create_directory:(unit -> [ `Created | `Error of string ] M.t) ->
    delete_directory:(unit -> [ `Deleted | `Error of string ] M.t) ->
    zip_directory:
      (?intermediate:unit ->
      staging_dir:directory_object ->
      unit ->
      [ `ZipFile of file_object | `Error of string ] M.t) ->
    spawn_in_directory:
      (command:MlFront_Core.FilePath.t ->
      args:string list ->
      envmods:MlFront_Core.EnvMods.t ->
      stdout:file_object ->
      stderr:file_object ->
      unit ->
      [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
      M.t) ->
    interactive_shell_in_directory:
      (?promptname:string ->
      envmods:MlFront_Core.EnvMods.t ->
      unit ->
      [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
      M.t) ->
    unit ->
    directory_object
  (** [generic_dir ~origin ~create_directory] creates a new directory object. *)

  val inmemory_file : origin:MlFront_Core.FilePath.t -> string -> file_object
  (** [inmemory_file ~origin contents] creates a new in-memory file object whose
      origin is [origin] and whose contents are [contents]. *)

  val inmemory_dir : origin:MlFront_Core.FilePath.t -> unit -> directory_object
  (** [inmemory_dir ~origin ()] creates a new in-memory directory object whose
      origin is [origin]. *)

  val file_origin : file_object -> string
  (** The location of the file. It should be clickable if possible by modern
      IDEs in a terminal. For example, a file should just be a path to the file
      (no ["file://"] prefix), while a http URL should be a URL. *)

  val directory_origin : directory_object -> string
  (** The location of the directory. It should be clickable if possible by
      modern IDEs in a terminal. For example, a file should just be a path to
      the file (no ["file://"] prefix), while a http URL should be a URL. *)

  (** {2 File Operations} *)

  val read_all :
    file_object ->
    [ `Content of string | `Error of string | `ExceededSizeLimit of int64 ] M.t
  (** [read_all source] gets the file, but may not give back the file if it
      exceeds memory.

      The [error] in [`Error error], if possible, should complete the phrase
      ["Cannot read the file because ___."]. For example,
      [`Error "the file was not found"]. *)

  val copy :
    src:file_object ->
    dest:file_object ->
    unit ->
    [ `Copied
    | `DestinationIsDirectory of directory_object
    | `Error of string
    | `SourceIsDirectory of directory_object ]
    M.t
  (** [copy ~src ~dest ()] copies the file [src] to the destination file [dest].

      On success, the file will have the executable bit set on Unix. *)

  val copy_or_fail :
    src:file_object ->
    dest:file_object ->
    on_error:(string -> 'a M.t) ->
    'a M.t ->
    'a M.t
  (** [copy_or_fail ~src ~dest ~on_error successvalue] copies the file [src] to
      the destination file [dest].

      [successvalue] is returned on success.

      On success, the file will have the executable bit set on Unix.

      The result of [on_error because] is returned on error, where [because] is
      a ["<subject phrase> was <adjective phrase>"] sentence fragment. *)

  val copy_but_error_if_dest_is_dir :
    src:file_object ->
    dest:file_object ->
    unit ->
    [ `Copied | `Error of string | `SourceIsDirectory of directory_object ] M.t
  (** [copy_but_error_if_dest_is_dir ~src ~dest ()] copies the file [src] to the
      destination file [dest]. If [dest] is a directory, an error is returned.

      On success, the file will have the executable bit set on Unix. *)

  val checksum_file :
    ?strip_carriage_returns:unit ->
    algo:[ `Sha1 | `Sha256 ] ->
    file_object ->
    [ `Error of string | `Checksum of string * Int64.t ] M.t
  (** [checksum_file ?strip_carriage_returns file] gets the SHA256 or SH1
      checksum of the file [file].

      If the flag [~strip_carriage_returns:()] is used, the checksum will be of
      the file {b after} it has been stripped of carriage returns (CR) with
      strip_carriage_returns. Only use this if you can guarantee removing
      carriage returns does not affect the semantics of the file. For example,
      OCaml source code can have multi-line strings where carriage returns
      should not be stripped. But JSON, where strings must have escaped carriage
      returns, can have all CR (ASCII 13) stripped. *)

  val replace_all_bytes :
    file_object ->
    bytes ->
    int ->
    int ->
    [ `Error of string | `IsDirectory of directory_object | `WroteBytes ] M.t
  (** [replace_all_bytes file bytes pos len] replaces the contents of the file
      [file] with the bytes [bytes] starting at position [pos] for length [len].

      On success, the file will have the executable bit set on Unix. *)

  val replace_all_string :
    file_object ->
    string ->
    int ->
    int ->
    [ `Error of string | `IsDirectory of directory_object | `WroteBytes ] M.t
  (** [replace_all_string file str pos len] replaces the contents of the file
      [file] with the string [str] starting at position [pos] for length [len].

      On success, the file will have the executable bit set on Unix. *)

  val delete_file : file_object -> [ `Deleted | `Error of string ] M.t
  val is_local_file : file_object -> bool

  (** {2 Directory Operations} *)

  val create_directory : directory_object -> [ `Created | `Error of string ] M.t
  val delete_directory : directory_object -> [ `Deleted | `Error of string ] M.t

  val spawn_in_directory :
    command:MlFront_Core.FilePath.t ->
    args:string list ->
    cwd:directory_object ->
    envmods:MlFront_Core.EnvMods.t ->
    stdout:file_object ->
    stderr:file_object ->
    unit ->
    [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
    M.t
  (** [spawn_in_directory ~command ~args ~cwd ~envmods ~stdout ~stderr] runs the
      command [command] with arguments [args] and environment modifications
      [envmods] in the working directory [cwd].

      A log of the standard output and standard error are written to the files
      [stdout] and [stderr], respectively.

      It returns a continuation that is one of:

      + [`Error error] if there was an error trying to run the command.
      + [`Exited code] if the process exited normally with exit code [code]
      + [`Signaled signal] if the process was killed by signal [signal]
      + [`Stopped signal] if the process was stopped by signal [signal]

      The [cwd] is the current working directory of the command. It may be a
      remote directory; if so, the command should run remotely.

      The [command] should be an absolute path to the command, or a path
      relative to the current working directory [cwd].

      The first element of [args] should be the path to the command or the
      basename of the command.

      The [envmods] are modifications to the environment variables. See
      {!MlFront_Core.EnvMods} for details. *)

  val interactive_shell_in_directory :
    ?promptname:string ->
    envmods:MlFront_Core.EnvMods.t ->
    cwd:directory_object ->
    unit ->
    [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
    M.t
  (** [interactive_shell_in_directory ?promptname ~envmods ~cwd ()] runs a shell
      in the working directory [cwd] with the environment modifications
      [envmods].

      The shell is interactive with access to the terminal's standard input,
      output, and error.

      [promptname] is used as as hint for inclusion into the shell prompt.

      It is implementation dependent if this function returns or if the current
      program is replaced by the shell *)

  val copy_file_or_dir_to_file_and_sha256_or_fail :
    ?intermediate:unit ->
    src:file_object ->
    dest:file_object ->
    staging_dir:directory_object ->
    on_error:(string -> 'b M.t) ->
    (string -> int64 -> 'b M.t) ->
    'b M.t
  (** [copy_file_or_dir_to_file_and_sha256_or_fail ?intermediate ~src ~dest
       ~staging_dir ~on_error successfunc] copies the file or directory [src] to
      the destination file [dest].

      The [src] is treated presumptively as a file. If and only if the copy
      operation fails specifically because [src] was detected to be a directory,
      then:

      + [staging_dir] is created if it does not exist
      + a zip archive file is created as a temporary file in [staging_dir] from
        the contents of [src] in the "deterministic" mode of the SPECIFICATION
        document
      + the temporary file is copied to [dest]
      + the temporary file is deleted (regardless of success or error)

      [successfunc sha256 file_sz] is returned on success, where [sha256] is the
      SHA256 hex-encoded hash of the copied file and [file_sz] is the size of
      the copied file.

      On success, the [dest] file will have the executable bit set on Unix.

      The result of [on_error because] is returned on error, where [because] is
      a ["<subject phrase> was <adjective phrase>"] sentence fragment. *)
end = struct
  type file_object = {
    file_origin : string;
    is_local_file : bool;
    open_for_writing :
      unit ->
      [ `Node of Int64.t | `IsDirectory of directory_object | `Error of string ]
      M.t;
    open_for_reading :
      unit ->
      [ `Node of Int64.t | `IsDirectory of directory_object | `Error of string ]
      M.t;
    read_some :
      Int64.t -> [ `Bytes of bytes * int * int | `Eof | `Error of string ] M.t;
    write_all :
      Int64.t -> string -> int -> int -> [ `WroteBytes | `Error of string ] M.t;
    close : Int64.t -> unit M.t;
    read_all :
      unit ->
      [ `Content of string | `Error of string | `ExceededSizeLimit of int64 ]
      M.t;
    prepare_as_copy_destination : unit -> [ `Ready | `Error of string ] M.t;
    delete_file : unit -> [ `Deleted | `Error of string ] M.t;
    checksum_file :
      strip_carriage_returns:bool ->
      algo:[ `Sha1 | `Sha256 ] ->
      unit ->
      [ `Checksum of string * Int64.t | `Error of string ] M.t;
  }

  and directory_object = {
    dir_origin : string;
    create_directory : unit -> [ `Created | `Error of string ] M.t;
    delete_directory : unit -> [ `Deleted | `Error of string ] M.t;
    zip_directory :
      ?intermediate:unit ->
      staging_dir:directory_object ->
      unit ->
      [ `ZipFile of file_object | `Error of string ] M.t;
    spawn_in_directory :
      command:MlFront_Core.FilePath.t ->
      args:string list ->
      envmods:MlFront_Core.EnvMods.t ->
      stdout:file_object ->
      stderr:file_object ->
      unit ->
      [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
      M.t;
    interactive_shell_in_directory :
      ?promptname:string ->
      envmods:MlFront_Core.EnvMods.t ->
      unit ->
      [ `Error of string | `Exited of int | `Signaled of int | `Stopped of int ]
      M.t;
  }

  let file_origin { file_origin; _ } = file_origin
  let read_all { read_all; _ } = read_all ()

  let checksum_file ?strip_carriage_returns ~algo { checksum_file; _ } =
    checksum_file
      ~strip_carriage_returns:(strip_carriage_returns = Some ())
      ~algo ()

  let delete_file { delete_file; _ } = delete_file ()
  let is_local_file { is_local_file; _ } = is_local_file
  let default_open_bufsize = 16_384
  let create_directory { create_directory; _ } = create_directory ()
  let delete_directory { delete_directory; _ } = delete_directory ()
  let directory_origin { dir_origin; _ } = dir_origin

  let spawn_in_directory ~command ~args ~cwd ~envmods ~stdout ~stderr () =
    cwd.spawn_in_directory ~command ~args ~envmods ~stdout ~stderr ()

  let interactive_shell_in_directory ?promptname ~envmods ~cwd () =
    cwd.interactive_shell_in_directory ?promptname ~envmods ()

  let replace_all_string { open_for_writing; write_all; close; _ } bytes pos len
      =
    let ( let* ) = M.bind in
    let* open_result = open_for_writing () in
    match open_result with
    | `IsDirectory d -> M.pure (`IsDirectory d)
    | `Error e -> M.pure (`Error e)
    | `Node n ->
        let* write_result = write_all n bytes pos len in
        let* () = close n in
        M.pure
          (write_result
            :> [ `Error of string
               | `IsDirectory of directory_object
               | `WroteBytes ])

  let replace_all_bytes io bytes = replace_all_string io (Bytes.to_string bytes)

  let generic_file ~origin ~is_local_file ~open_for_writing ~open_for_reading
      ~read_some ~write_all ~close ~read_all ~prepare_as_copy_destination
      ~delete_file ~checksum_file () =
    {
      file_origin = origin;
      is_local_file;
      open_for_writing;
      open_for_reading;
      read_some;
      write_all;
      close;
      read_all;
      prepare_as_copy_destination;
      delete_file;
      checksum_file;
    }

  let generic_dir ~origin ~create_directory ~delete_directory ~zip_directory
      ~spawn_in_directory ~interactive_shell_in_directory () =
    {
      dir_origin = origin;
      create_directory;
      delete_directory;
      zip_directory;
      spawn_in_directory;
      interactive_shell_in_directory;
    }

  let inmem_read_records = Hashtbl.create 1
  let inmem_write_records = Hashtbl.create 1
  let inmem_counter = ref 0L

  let inmemory_file ~origin contents =
    let origin_s = MlFront_Core.FilePath.to_string origin in
    {
      file_origin = origin_s;
      is_local_file = false;
      open_for_writing =
        (fun () ->
          let idx = !inmem_counter in
          inmem_counter := Int64.succ !inmem_counter;
          Hashtbl.add inmem_write_records idx
            (origin, Buffer.create default_open_bufsize);
          M.return (`Node 0L));
      open_for_reading =
        (fun () ->
          let idx = !inmem_counter in
          inmem_counter := Int64.succ !inmem_counter;
          Hashtbl.add inmem_read_records idx (origin, `Started);
          M.return (`Node idx));
      read_some =
        (fun idx ->
          match Hashtbl.find_opt inmem_read_records idx with
          | Some (_origin, `Started) ->
              Hashtbl.add inmem_read_records idx (origin, `Done);
              M.return
                (`Bytes (Bytes.of_string contents, 0, String.length contents))
          | Some (_origin, `Done) -> M.return `Eof
          | None ->
              M.return
                (`Error
                   (Printf.sprintf "in-memory file `%s` is not open" origin_s)));
      write_all =
        (fun idx str pos len ->
          match Hashtbl.find_opt inmem_write_records idx with
          | Some (_origin, buffer) ->
              Buffer.add_substring buffer str pos len;
              M.return `WroteBytes
          | None ->
              M.return
                (`Error
                   (Printf.sprintf "in-memory file `%s` is not open" origin_s)));
      close = (fun _ -> M.return ());
      read_all = (fun () -> M.return (`Content contents));
      prepare_as_copy_destination = (fun () -> M.return `Ready);
      delete_file =
        (fun () ->
          (* delete matching entries *)
          Hashtbl.filter_map_inplace
            (fun _idx (origin', status) ->
              if MlFront_Core.FilePath.compare origin origin' = 0 then None
              else Some (origin', status))
            inmem_read_records;
          Hashtbl.filter_map_inplace
            (fun _idx (origin', buffer) ->
              if MlFront_Core.FilePath.compare origin origin' = 0 then None
              else Some (origin', buffer))
            inmem_write_records;
          M.return `Deleted);
      checksum_file =
        (let do_strip_cr s =
           let buf = Buffer.create (String.length s) in
           String.iter
             (fun c -> if c <> '\r' then Buffer.add_char buf c else ())
             s;
           Buffer.contents buf
         in
         fun ~strip_carriage_returns ~algo () ->
           let contents =
             if strip_carriage_returns then do_strip_cr contents else contents
           in
           let len = String.length contents in
           match algo with
           | `Sha1 ->
               let cksum = Digestif.SHA1.digest_string contents in
               M.return
                 (`Checksum (Digestif.SHA1.to_hex cksum, Int64.of_int len))
           | `Sha256 ->
               let cksum = Digestif.SHA256.digest_string contents in
               M.return
                 (`Checksum (Digestif.SHA256.to_hex cksum, Int64.of_int len)));
    }

  let inmemory_dir ~origin () =
    let origin_s = MlFront_Core.FilePath.to_string origin in
    {
      dir_origin = origin_s;
      create_directory = (fun () -> M.return `Created);
      delete_directory =
        (fun () ->
          (* delete descendant file entries *)
          Hashtbl.filter_map_inplace
            (fun _idx (origin', status) ->
              if MlFront_Core.FilePath.is_subpath origin origin' then None
              else Some (origin', status))
            inmem_read_records;
          Hashtbl.filter_map_inplace
            (fun _idx (origin', buffer) ->
              if MlFront_Core.FilePath.is_subpath origin origin' then None
              else Some (origin', buffer))
            inmem_write_records;
          M.return `Deleted);
      zip_directory =
        (fun ?intermediate:_ ~staging_dir:_ () ->
          M.return
            (`Error
               (Printf.sprintf
                  "currently there is no support for zipping in-memory \
                   directories including `%s`"
                  origin_s)));
      spawn_in_directory =
        (fun ~command:_ ~args:_ ~envmods:_ ~stdout:_ ~stderr:_ () ->
          M.return
            (`Error
               "currently there is no support for spawning commands in \
                in-memory directories"));
      interactive_shell_in_directory =
        (fun ?promptname:_ ~envmods:_ () ->
          M.return
            (`Error
               "currently there is no support for launching shells in \
                in-memory directories"));
    }

  let copy ~src ~dest () =
    (* Copy file using buffered copy *)
    let ( let* ) = M.bind in
    let* prepare_result = dest.prepare_as_copy_destination () in
    match prepare_result with
    | `Error e -> M.pure (`Error e)
    | `Ready -> (
        let* openwrite_result = dest.open_for_writing () in
        match openwrite_result with
        | `Error e -> M.pure (`Error e)
        | `IsDirectory d -> M.pure (`DestinationIsDirectory d)
        | `Node dest_node -> (
            let finish () = dest.close dest_node in
            let* openread_result = src.open_for_reading () in
            match openread_result with
            | `Error e ->
                let* () = finish () in
                M.pure (`Error e)
            | `IsDirectory d ->
                let* () = finish () in
                M.pure (`SourceIsDirectory d)
            | `Node src_node ->
                (* We have both source and destination nodes open for reading/writing *)
                let finish () =
                  let* () = dest.close dest_node in
                  let* () = src.close src_node in
                  M.pure ()
                in
                let rec aux acc =
                  let* c = acc in
                  match c with
                  | `Error e -> M.pure (`Error e)
                  | `Copied -> (
                      let* read_result = src.read_some src_node in
                      match read_result with
                      | `Error e -> M.pure (`Error e)
                      | `Eof -> M.pure `Copied
                      | `Bytes (b, off, len) -> (
                          let* write_result =
                            dest.write_all dest_node (Bytes.to_string b) off len
                          in
                          match write_result with
                          | `Error e -> M.pure (`Error e)
                          | `WroteBytes -> aux (M.pure `Copied)))
                in
                let* final_result = aux (M.return `Copied) in
                let* () = finish () in
                M.pure final_result))

  let copy_or_fail ~src ~dest ~on_error successvalue =
    let ( let* ) = M.bind in
    let* copy_result = copy ~src ~dest () in
    let r =
      match copy_result with
      | `Copied -> `Copied
      | `DestinationIsDirectory _d ->
          `Error
            (Printf.sprintf "destination path `%s` is a directory"
               (file_origin dest))
      | `Error s -> `Error s
      | `SourceIsDirectory _d ->
          `Error
            (Printf.sprintf "source path `%s` is a directory" (file_origin src))
    in
    match r with `Copied -> successvalue | `Error because -> on_error because

  let copy_but_error_if_dest_is_dir ~src ~dest () =
    let ( let* ) = M.bind in
    let* copy_result = copy ~src ~dest () in
    match copy_result with
    | `Copied -> M.pure `Copied
    | `DestinationIsDirectory _d ->
        M.pure
          (`Error
             (Printf.sprintf "destination path `%s` is a directory"
                (file_origin dest)))
    | `Error s -> M.pure (`Error s)
    | `SourceIsDirectory d -> M.pure (`SourceIsDirectory d)

  let copy_file_or_dir_to_file_and_sha256_or_fail =
    let ( let* ) = M.bind in
    fun ?intermediate ~src ~dest ~staging_dir ~on_error successvalue ->
      let intermediatefile_pending_deletion = ref None in
      Fun.protect
        ~finally:(fun () ->
          if intermediate = None then
            try Option.iter Sys.remove !intermediatefile_pending_deletion
            with Sys_error _ -> ())
        (fun () ->
          let* reduced_result = copy_but_error_if_dest_is_dir ~src ~dest () in
          let* r =
            match reduced_result with
            | `Copied -> M.pure `Copied
            | `Error s -> M.pure (`Error s)
            | `SourceIsDirectory srcdir -> (
                let* dir_result = staging_dir.create_directory () in
                match dir_result with
                | `Error e -> M.pure (`Error e)
                | `Created -> (
                    let* zip_result =
                      srcdir.zip_directory ?intermediate ~staging_dir ()
                    in
                    match zip_result with
                    | `Error e -> M.pure (`Error e)
                    | `ZipFile intermediatezipfile -> (
                        (* copy the zipfile to the destination.
                          TODO: move instead of copy, or at least try to move first. *)
                        intermediatefile_pending_deletion :=
                          Some (file_origin intermediatezipfile);
                        let* copy_result =
                          copy_but_error_if_dest_is_dir ~src:intermediatezipfile
                            ~dest ()
                        in
                        match copy_result with
                        | `Error e ->
                            let* () =
                              if intermediate = Some () then
                                let* _delete_result :
                                    [ `Deleted | `Error of string ] =
                                  intermediatezipfile.delete_file ()
                                in
                                M.pure ()
                              else M.pure ()
                            in
                            M.pure (`Error e)
                        | `Copied -> M.pure `Copied
                        | `SourceIsDirectory _ ->
                            M.pure
                              (`Error
                                 (Printf.sprintf
                                    "intermediate zipfile path `%s` is a \
                                     directory"
                                    (file_origin intermediatezipfile))))))
          in
          match r with
          | `Error because -> on_error because
          | `Copied -> (
              let* sha256_result = checksum_file ~algo:`Sha256 dest in
              match sha256_result with
              | `Error e -> on_error e
              | `Checksum (sha256, filesz) -> successvalue sha256 filesz))
end