path: root/lib/disk.ml

module type Configuration = sig
  val base_dir : string
  val storage : string
  val transactions : string
  val assure : unit -> unit
end

module DevelopmentConfiguration : Configuration = struct
  let base_dir = "/tmp/relational-engine/"
  let storage = base_dir ^ "storage/"
  let transactions = base_dir ^ "transactions/"

  let assure () =
    if Sys.file_exists base_dir then Sys.remove base_dir;
    Sys.mkdir base_dir 0o700;
    Sys.mkdir storage 0o700;
    Sys.mkdir transactions 0o700
end

module type FileSystem = sig
  type target = Storage | Transaction

  val append : target -> bytes -> (int * int64, string) result
  val wipe_and_write : string -> bytes -> (int * int64, string) result
  val read : target -> int64 -> int -> (bytes, string) result
end

module FileSystem (C : Configuration) : FileSystem = struct
  type target = Storage | Transaction

  let target_to_string = function
    | Storage -> C.storage ^ "storage.re"
    | Transaction -> C.transactions ^ "transaction.re"

  (** [append target content] writes content to file in append only format,
      returning [content_length * offset_written] *)
  let append target content =
    let register channel =
      let offset_written =
        try
          let stats = Unix.stat (target_to_string target) in
          Int64.of_int stats.Unix.st_size
        with _ -> 0L
      in
      (* Out_channel.seek channel offset_written; *)
      (* let offset_written = Out_channel.length channel in *)
      Out_channel.output_bytes channel content;
      Out_channel.flush channel;
      Out_channel.close channel;
      Ok (Bytes.length content, offset_written)
    in
    try
      Out_channel.with_open_gen
        [ Open_append; Open_binary; Open_creat ]
        0o666 (target_to_string target) register
    with e ->
      Error
        (Printf.sprintf "Error on appending to target '%s': %s"
           (target_to_string target) (Printexc.to_string e))

  let wipe_and_write path content =
    let offset_written =
      try
        let stats = Unix.stat path in
        Int64.of_int stats.Unix.st_size
      with _ -> 0L
    in
    let register channel =
      Out_channel.output_bytes channel content;
      Out_channel.flush channel;
      Ok (Bytes.length content, offset_written)
    in
    try
      Out_channel.with_open_gen
        [ Open_trunc; Open_binary; Open_creat; Open_wronly ]
        0o666 path register
    with e ->
      Error (
          Printf.sprintf "Error on appending to path '%s': %s"
            path (Printexc.to_string e)
        )

  let read target (offset : int64) size =
    print_endline ("READ_OFFSET: " ^ Int64.to_string offset);
    print_endline ("READ_SIZE: " ^ string_of_int (Int64.to_int offset + size));
    let register (channel : in_channel) =
      let buffer = Bytes.create size in
      In_channel.seek channel offset;
      let _ = In_channel.really_input channel buffer 0 size in
      print_bytes buffer;
      print_newline ();
      In_channel.close channel;
      Ok buffer
    in
    try
      In_channel.with_open_gen
        [ Open_binary; Open_rdonly ]
        0o666 (target_to_string target) register
    with e ->
      Error
        (Printf.sprintf "Error on reading from target '%s': %s"
           (target_to_string target) (Printexc.to_string e))
end

module Executor = struct
  module StringMap = Map.Make (String) [@@deriving show]
  module IntMap = Map.Make (Int64) [@@deriving show]
  module FS = FileSystem (DevelopmentConfiguration)

  module Location = struct
    type t = { offset : int64; size : int; hash : string } [@@deriving show]
    let encoding =
      let open Data_encoding in
      conv
        (fun { offset; size; hash } -> (offset, size, hash))
        (fun (offset, size, hash) -> { offset; size; hash })
        (tup3 int64 int31 string)
  end

  module Hashes = struct
    type t = { values : string list; hash : string } [@@deriving show]
    type history = t list [@@deriving show]

    let encoder: t Data_encoding.t =
      let open Data_encoding in
      conv
        (fun {values; hash} -> (values, hash))
        (fun (values, hash) -> {values; hash})
        (tup2 (list string) string)
  end

  open Sexplib0.Sexp_conv
  open Protocol_conv_xml

  type relational_type =
    | Text
    | Integer32
    | Integer64
    | Boolean
    (* | DiscriminatedUnion of (string * relational_type option) list * string *)
      (* [("MemberA", None); ("MemberB", Some Integer32)], "NameOfDU" *)
    | Relation of string
  [@@deriving show, sexp, protocol ~driver:(module Xml_light)]

  type relational_literal =
    | LText of string
    | LInteger32 of int32
    | LInteger64 of int64
    | LBoolean of bool
    | LRelation of int64*string
  [@@deriving show, sexp, protocol ~driver:(module Xml_light)]

  type schema = (string * relational_type) list StringMap.t
  (* entity * type * attribute_name *)
  type references = (string * relational_type * string) list IntMap.t StringMap.t

  type commit = {
    state : string;
    files : Hashes.history StringMap.t;
    references : references;
    schema : schema;
  }

  type history = commit list
  type locations = Location.t StringMap.t

  let relational_type_encoding =
    let open Data_encoding in
    union
      [
        case ~title:"text" (Tag 0) Data_encoding.empty
          (function Text -> Some () | _ -> None)
          (function () -> Text);
        case ~title:"integer32" (Tag 1) Data_encoding.empty
          (function Integer32 -> Some () | _ -> None)
          (function () -> Integer32);
        case ~title:"integer64" (Tag 2) Data_encoding.empty
          (function Integer64 -> Some () | _ -> None)
          (function () -> Integer64);
        case ~title:"boolean" (Tag 3) Data_encoding.empty
          (function Boolean -> Some () | _ -> None)
          (function () -> Boolean);
        case ~title:"relation" (Tag 4) Data_encoding.string
          (function Relation reference -> Some reference | _ -> None)
          (function reference -> Relation reference);
      ]

  let relational_literal_encoding =
    let open Data_encoding in
    union
      [
        case ~title:"text" (Tag 0) Data_encoding.string
          (function LText x -> Some x | _ -> None)
          (function x -> LText x);
        case ~title:"integer32" (Tag 1) Data_encoding.int32
          (function LInteger32 x -> Some x | _ -> None)
          (function x -> LInteger32 x);
        case ~title:"integer64" (Tag 2) Data_encoding.int64
          (function LInteger64 x -> Some x | _ -> None)
          (function x -> LInteger64 x);
        case ~title:"boolean" (Tag 3) Data_encoding.bool
          (function LBoolean x -> Some x | _ -> None)
          (function x -> LBoolean x);
        case ~title:"relation" (Tag 4) (Data_encoding.tup2 Data_encoding.int64 Data_encoding.string)
          (function LRelation (reference, name) -> Some (reference, name) | _ -> None)
          (function (reference, name) -> LRelation (reference, name));
      ]
  
  let encode_commit =
    let open Data_encoding in
    (* let references_encoder = (list (list (tup2 string (list (tup2 int64 (tup3 string relational_type_encoding string)))))) in *)
    let references_encoder = list (tup2 string (list (tup2 int64 (list (tup3 string relational_type_encoding string))))) in
    let schema_encoder = list (tup2 string (list (tup2 string relational_type_encoding))) in
    (* let schema_encoder = list (tup2 string (list (tup2 string relational_type_encoding))) in *)
    let files_encoder = (list (tup2 string (list Hashes.encoder))) in
    conv
      (fun {state; files; references; schema} -> (state, StringMap.to_list files, (let x = List.map (fun (k, v) -> k, IntMap.to_list v) (StringMap.to_list references) in x), StringMap.to_list schema))
      (fun (state, files, references, schema) -> {state; files = StringMap.of_list files; references = StringMap.of_list (List.map (fun (k, v) -> k, IntMap.of_list v) references); schema = StringMap.of_list schema})
      (tup4 string files_encoder references_encoder schema_encoder)


  let encode_locations =
    let open Data_encoding in
    conv
      StringMap.to_list
      StringMap.of_list
      (list (tup2 string Location.encoding))
  
  let _EMPTY_SHA_HASH_ =
    "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"

  (** Initializes the virtual file or opens it. *)
  let init_file (files : Hashes.history StringMap.t) filename :
      Hashes.history StringMap.t =
    match StringMap.find_opt filename files with
    | None ->
        let default : Hashes.t = { values = []; hash = _EMPTY_SHA_HASH_ } in
        let new_files = StringMap.add filename [ default ] files in
        new_files
    | Some _ -> files

  let compose_new_state files =
    let history = StringMap.to_list files in
    let all_hashes =
      List.concat
      @@ List.map
           (fun (_, (history : Hashes.history)) ->
             List.map (fun ({ hash; _ } : Hashes.t) -> hash) history)
           history
    in
    Interop.Merkle.merkle_generate_root all_hashes

  let write ({ files; references; schema; _ } as commit : commit)
      (locations : locations) ~filename ?hash_to_replace (content : Bytes.t) =
    let files = init_file files filename in
    let computed_hash : string = Interop.Sha256.compute_hash content in
    match StringMap.find_opt computed_hash locations with
    | None -> (
        print_endline "-----------NOT EXISTS-----------";
        match hash_to_replace with
        | None ->
            (* This instead should store a list of list to keep the history *)
            let files =
              StringMap.update filename
                (function
                  | Some (file_hashes : Hashes.history) ->
                      let { values; _ } : Hashes.t = List.hd file_hashes in
                      let new_state =
                        Interop.Merkle.merkle_generate_root
                          (computed_hash :: values)
                      in
                      Some
                        ({ values = computed_hash :: values; hash = new_state }
                        :: file_hashes)
                  | None ->
                      Some
                        [ { values = [ computed_hash ]; hash = computed_hash } ])
                files
            in
            let open Extensions.Result in
            let+ size, offset = FS.append Storage content in
            let () = print_endline @@ "OFFSET: " ^ Int64.to_string offset in
            let locations =
              StringMap.add computed_hash
                ({ size; offset; hash = computed_hash } : Location.t)
                locations
            in
            let commit =
              { state = compose_new_state files; files; references; schema }
            in
            Ok ((commit, locations), Some computed_hash)
        | Some hash_to_replace ->
            let update_fun = function
              | Some (file_hashes : Hashes.history) ->
                  let { values; _ } : Hashes.t = List.hd file_hashes in
                  let updated_entry =
                    List.map
                      (fun hash ->
                        if hash = hash_to_replace then computed_hash else hash)
                      values
                  in
                  let new_entry : Hashes.t =
                    {
                      values = updated_entry;
                      hash = Interop.Merkle.merkle_generate_root updated_entry;
                    }
                  in
                  Some (new_entry :: file_hashes)
              | None ->
                  Some [ { values = [ computed_hash ]; hash = computed_hash } ]
            in
            let files = StringMap.update filename update_fun files in
            let open Extensions.Result in
            let+ size, offset = FS.append Storage content in
            let () = print_endline @@ "OFFSET: " ^ Int64.to_string offset in
            let locations =
              StringMap.add computed_hash
                ({ size; offset; hash = computed_hash } : Location.t)
                locations
            in
            let commit =
              { state = compose_new_state files; files; references; schema }
            in
            Ok ((commit, locations), Some computed_hash))
    | Some _ -> (
        print_endline "-----------EXISTS-----------";
        match hash_to_replace with
        | Some hash_to_replace ->
            let update_fun = function
              | Some (file_hashes : Hashes.history) ->
                  let { values; _ } : Hashes.t = List.hd file_hashes in
                  let updated_entry =
                    List.map
                      (fun hash ->
                        if hash = hash_to_replace then computed_hash else hash)
                      values
                  in
                  let new_entry : Hashes.t =
                    {
                      values = updated_entry;
                      hash = Interop.Merkle.merkle_generate_root updated_entry;
                    }
                  in
                  Some (new_entry :: file_hashes)
              | None ->
                  Some [ { values = [ computed_hash ]; hash = computed_hash } ]
            in
            (* let location_update_fun = function
                 | Some (location: Location.t) ->
                    Some { location with references = references@location.references }
                 | None -> None (* This case is a bit weird to exist. Can we ever already add and get nothing? Refactor with the location passed on the `Some _` already *)
               in *)
            let files = StringMap.update filename update_fun files in
            let commit =
              { state = compose_new_state files; files; references; schema }
            in
            (* let locations = StringMap.update computed_hash location_update_fun locations in *)
            Ok ((commit, locations), Some computed_hash)
        | None -> Ok ((commit, locations), None))

  let read ({ files; _ } as _commit : commit) (locations : locations) ~filename
      =
    let open Extensions.Option in
    let+ history : Hashes.history = StringMap.find_opt filename files in
    let+ { values = location_hashes; _ } : Hashes.t =
      match history with [] -> None | history -> Some (List.hd history)
    in
    let physical_locations =
      List.map (fun hash -> StringMap.find hash locations) location_hashes
    in
    (* Terrible! Here I read everything opening a new IO every time. We should be able to read with only one call, but changing the offset every time *)
    let content =
      List.map
        (fun ({ offset; size; _ } : Location.t) ->
          match FS.read FS.Storage offset size with
          | Ok x -> x
          | Error err -> failwith err)
        physical_locations
    in
    (*Bytes.concat Bytes.empty @@*)
    Some (List.rev content)

  let cast_to_type (type': relational_type) (content: bytes): relational_literal =
    match type' with
    | Text -> LText (Bytes.to_string content)
    | Integer32 -> LInteger32 (Data_encoding.Binary.of_bytes_exn Data_encoding.int32 content)
    | Relation _ ->
       begin
         let (reference, name) = (Data_encoding.Binary.of_bytes_exn (Data_encoding.tup2 Data_encoding.int64 Data_encoding.string) content) in
         LRelation (reference, name)
       end
    | _ -> failwith "Not implemented"

  let relational_literal_to_bytes (literal: relational_literal): bytes =
    match literal with
    | LText x -> Bytes.of_string x
    | LInteger32 x -> Data_encoding.Binary.to_bytes_exn Data_encoding.int32 x
    | LInteger64 x -> Data_encoding.Binary.to_bytes_exn Data_encoding.int64 x
    | LRelation (x,y) ->
       begin match Data_encoding.Binary.to_bytes (Data_encoding.tup2 Data_encoding.int64 Data_encoding.string) (x, y) with
       | Error x -> let open Data_encoding.Binary in
                    begin match x with
                    | Size_limit_exceeded -> print_endline "A"; Bytes.empty;
                    | No_case_matched -> print_endline "B"; Bytes.empty;
                    | Invalid_int _ -> print_endline "C"; Bytes.empty;
                    | Invalid_float _ -> print_endline "D"; Bytes.empty;
                    | Invalid_bytes_length _ -> print_endline "E"; Bytes.empty;
                    | Invalid_string_length _ -> print_endline "F"; Bytes.empty;
                    | Invalid_natural -> print_endline "G"; Bytes.empty;
                    | List_invalid_length -> print_endline "H"; Bytes.empty;
                    | Array_invalid_length -> print_endline "I"; Bytes.empty;
                    | Exception_raised_in_user_function _x -> print_endline "J"; Bytes.empty;
                    end
       | Ok x -> x
       end
    | LBoolean x -> Data_encoding.Binary.to_bytes_exn Data_encoding.bool x

  let read_location ~hash (locations: locations) (cast_type: relational_type): relational_literal =
    match StringMap.find_opt hash locations with
    | Some ({ offset; size; _ } : Location.t) ->
       begin
         match FS.read FS.Storage offset size with
         | Ok x -> cast_to_type cast_type x
         | Error err -> failwith err
        end
    | None -> failwith "Location could not be found."
end

module Startup = struct
  (* TODO: Weird behavior with a system error on the dir /tmp/relational-engine opening. Disabling for now *)
  (* let () = DevelopmentConfiguration.assure() *)
end

module Command = struct
  module FS = FileSystem (DevelopmentConfiguration)
  open Data_encoding

  type transaction = {
    (* kind : command_kind; *)
    timestamp : float;
    (* hash : string; *)
    attribute : string;
    entity_id : int64 option;
    (* branch must be added here later *)
    content : Executor.relational_literal;
    type': Executor.relational_type;
  }

  open Sexplib0.Sexp_conv
  open Protocol_conv_xml

  type t =
    | SequentialRead of { relation_name : string }
    | SpecifyRelation of { relation_name: string;
                           attributes: (string*Executor.relational_type) list }
    | NaturalJoin of {left_relation: string;
                      right_relation: string}
  [@@deriving sexp, protocol ~driver:(module Xml_light)]
  
  let command_encoding =
    conv
      (fun { timestamp; attribute; entity_id; content; type' } ->
        (timestamp, attribute, entity_id, content, type'))
      (fun (timestamp, attribute, entity_id, content, type') ->
        { timestamp; attribute; entity_id; content; type' })
      Data_encoding.(tup5 float string (option int64) Executor.relational_literal_encoding Executor.relational_type_encoding)

  let parse_command ~data =
    let contract_encoding =
      Data_encoding.(tup4 string (option int64) Executor.relational_literal_encoding Executor.relational_type_encoding)
    in
    match Binary.of_bytes_opt contract_encoding data with
    | Some (attribute, entity_id, content, type') ->
        Ok { timestamp = Unix.time (); attribute; entity_id; content; type' }
    | None -> Error "Failed to parse command"

  type return =
    | ComputedHash of string
    | Read of (int64 * (string*Executor.relational_literal) list) list
    | Nothing
  [@@deriving show, sexp, protocol ~driver:(module Xml_light)]

  (** Writes the commit and location to the disk, but cleans it before.
      Needless to say this is absolutely just a dirty trick to test persistency of the structures.
      To implement this properly, we need to mark the new appended objects, flush the dirty ones to
      the disk without touching the rest, and do it atomically with unix file swaps (for example). **)
  let write_context commit locations =
    let open Extensions.Result in
    let+ commit = Result.map_error (fun _ -> "Failed to serialize 'commit'") @@ Data_encoding.Binary.to_bytes Executor.encode_commit commit in
    let+ locations = Result.map_error (fun _ -> "Failed to serialize 'locations'") @@ Data_encoding.Binary.to_bytes Executor.encode_locations locations in
    let+ _ = Result.map_error (fun msg -> Printf.sprintf "Failed to write 'commit' to the disk: %s" msg) @@ FS.wipe_and_write "/tmp/relational-engine/commit" commit in
    let+ _ = Result.map_error (fun msg -> Printf.sprintf "Failed to write 'locations' to the disk: %s" msg) @@ FS.wipe_and_write "/tmp/relational-engine/locations" locations in
    Ok ()
  
  (** TODO: This does not write atomically. If the system crashes while it
      attempts to write, it will corrupt. Solve this with a temporary file and
      move later. *)
  let transact (commit : Executor.commit) (locations : Executor.locations)
      (command : transaction) =
    let open Extensions.Result in
    let+ serialized_command =
      Result.map_error (fun _ ->
          "Failed to serialize command to binary format.")
      @@ Binary.to_bytes command_encoding command
    in
    let+ _ = FS.append FS.Transaction serialized_command in
    let open Extensions.Result in
    let+ () = write_context commit locations in
    let+ (commit, locations), computed_hash_handle =
      Executor.write commit locations ~filename:command.attribute
      @@ Executor.relational_literal_to_bytes command.content
    in
    match (computed_hash_handle, command.entity_id) with 
    | Some computed_hash_handle, Some entity_id ->
        let references: Executor.references =
          let relation_name::[attribute_name] =
            String.split_on_char '/' command.attribute
          [@@warning "-8"] in
          Executor.StringMap.update relation_name
            (function
              | Some entity ->
                  Some
                    (Executor.IntMap.update entity_id
                       (function
                         | Some locations ->
                             Some ((computed_hash_handle, command.type', attribute_name) :: locations)
                         | None -> Some [ (computed_hash_handle, command.type', attribute_name) ])
                       entity)
              | None ->
                  Some
                    (Executor.IntMap.empty
                    |> Executor.IntMap.add entity_id [ (computed_hash_handle, command.type', attribute_name) ]))
            commit.references
        in
        Ok
          ( ({ commit with references }, locations),
            ComputedHash computed_hash_handle )
    | Some _, None -> Error "Cannot write an entity without its referential."
    | None, _ -> Ok ((commit, locations), Nothing)
  
  let perform (commit : Executor.commit) (locations : Executor.locations) (command : t) =
    match command with
    | NaturalJoin {left_relation = _; right_relation = _} ->
       Ok ((commit, locations), Nothing)
    | SpecifyRelation {relation_name; attributes} ->
       (* If it already exists under the relation name key, it will be replaced *)
       Ok (({ commit with
              schema = commit.schema
                       |> Executor.StringMap.add relation_name attributes}, locations), Nothing)
    | SequentialRead { relation_name } ->
        let relation_name = List.hd @@ String.split_on_char '/' relation_name in
        let entities : (string * Executor.relational_type * string) list Executor.IntMap.t =
          print_endline relation_name;
          Executor.StringMap.find_opt relation_name commit.references
          |> function
          | Some entities -> entities
          | None -> Executor.IntMap.empty
        in

        let content: (int64 * (string*Executor.relational_literal) list) list =
          Executor.IntMap.fold
            (fun key elems acc ->
              (key,
                List.map
                  (fun (location, type', attribute_name) ->
                    (attribute_name, Executor.read_location ~hash:location locations type'))
                  elems)
              :: acc)
            entities []
        in
        
        Ok ((commit, locations), Read content)
end