path: root/lib/kernel/SubstantialRelation.ml

module Make (S : Abstract.Storage.STORAGE) = struct
  module SI = Storage.Make (S)

  module Error = struct
    open Concepts.Condition

    let malformed_relation () =
      condition "malformed-substantial-relation"
        "The on-disk representation of a substantial relation did not conform to what was \
         expected. Is your database corrupted?"
        empty

    let malformed_tuple_key length =
      condition "malformed-tuple-key" "A tuple key did not have the expected hash size"
        ("length" |=| Concepts.Value.Integer length)

    let invalid_tuple_root root =
      condition "invalid-tuple-root"
        "The tuple-set root for a substantial relation is missing from the backend. Either your \
         storage is corrupted, or this is a bug in RNT!"
        ("hash" |=| Concepts.Value.String (Concepts.Hash.to_hum_string root))

    let missing_attribute name =
      condition "missing-attribute" "A tuple to store has no value for an attribute of its relation"
        ("attribute" |=| Concepts.Value.String name)

    let unindexed_attribute name =
      condition "unindexed-attribute" "A relation was looked up on an attribute it has no index for"
        ("attribute" |=| Concepts.Value.String name)

    let invalid_index_root root =
      condition "invalid-index-root"
        "An index root for a substantial relation is missing from the backend. Either your storage \
         is corrupted, or this is a bug in RNT!"
        ("hash" |=| Concepts.Value.String (Concepts.Hash.to_hum_string root))
  end

  module TupleKey = struct
    type t = Concepts.Hash.hash

    let encode = Concepts.Hash.blob_of_hash

    let decode blob =
      let raw = Concepts.Blob.bytes_of_blob blob |> Bytes.to_string in
      if String.length raw = Concepts.Hash.size then Ok (Concepts.Hash.of_raw_string raw)
      else Error (Error.malformed_tuple_key (String.length raw))

    let compare = Concepts.Hash.compare
  end

  module TupleValue = struct
    type t = Concepts.Tuple.t

    let encode = Concepts.Tuple.Representation.to_blob
    let decode = Concepts.Tuple.Representation.of_blob
  end

  module TupleSet = Merkle.Interface (S) (TupleKey) (TupleValue)
  module Index = Merkle.Make (S) (Merkle.StringKey)

  type t =
    { schematics: Concepts.Hash.hash;
      tuples: TupleSet.address;
      indexes: Index.address BatMap.String.t }

  module rec Representation : (Concepts.Encoding.Record.S with type t = t) =
    Concepts.Encoding.Record.Make (Body)

  and Body : Concepts.Encoding.Record.BODY = struct
    type nonrec t = t

    let tag = 'R'
    let malformed = Error.malformed_relation

    let fields {schematics; tuples; indexes} =
      let open Concepts.Encoding in
      [ "schema", Value.bencode_of_hash schematics;
        "tuples", Value.bencode_of_hash tuples;
        ( "indexes",
          Bencode.Dict
            (BatMap.String.bindings indexes |> List.map (fun (a, h) -> a, Value.bencode_of_hash h))
        ) ]

    let of_fields fields =
      let open Utilities.Result in
      let open Concepts.Encoding in
      let* schematics = Bencode.field "schema" fields |>> Value.hash_of_bencode in
      let* tuples = Bencode.field "tuples" fields |>> Value.hash_of_bencode in
      let* indexes = Bencode.field "indexes" fields |>> Bencode.as_dict in
      let* indexes =
        List.map (fun (a, h) -> Value.hash_of_bencode h |> Result.map (fun h -> a, h)) indexes
        |> Utilities.List.sequence
      in
      Ok {schematics; tuples; indexes= BatMap.String.of_list indexes}
  end

  let encode = Representation.to_blob
  let decode = Representation.of_blob

  let tuple_node tx relation =
    let open Utilities.Result in
    let* tuples = TupleSet.find tx relation.tuples in
    Option.to_result ~none:(Error.invalid_tuple_root relation.tuples) tuples

  let store_schema tx description =
    Concepts.Encoding.Bencode.(
      Dict
        ( BatMap.String.bindings description
        |> List.map (fun (name, {Protocols.Schematics.domain; _}) -> name, String domain) )
      |> to_blob )
    |> SI.store_blob tx

  let schema_of tx schematics =
    let open Utilities.Result in
    let* data = SI.get_req tx (S.Hash schematics) in
    let* fields = Concepts.Encoding.Bencode.of_blob data in
    let* kvs = Concepts.Encoding.Bencode.as_dict fields in
    kvs
    |> List.map (fun (name, domain) ->
        Concepts.Encoding.Bencode.as_string domain
        |> Result.map (fun domain -> name, {Protocols.Schematics.domain; provenance= []}) )
    |> Utilities.List.sequence
    |> Result.map BatMap.String.of_list

  let prefix value =
    Concepts.Encoding.Bencode.to_bytes (Concepts.Tuple.bencode_of_value value) |> Bytes.to_string

  let reading connection start walk =
    let open Utilities.Result in
    let* tx = S.start_read connection in
    let close () = ignore (S.abort tx) in
    let* node = start tx |> Result.map_error (fun c -> close (); c) in
    Ok (Generator.cursor_of ~finally:close (walk tx node))

  let enumerate connection relation =
    reading connection
      (fun tx -> tuple_node tx relation)
      (fun tx node ~yield -> TupleSet.iter tx (fun _ tuple -> yield tuple) node)

  let lookup connection relation attribute value =
    let open Utilities.Result in
    let* root =
      BatMap.String.find_opt attribute relation.indexes
      |> Option.to_result ~none:(Error.unindexed_attribute attribute)
    in
    let prefix = prefix value in
    reading connection
      (fun tx -> Index.find tx root |>> Option.to_result ~none:(Error.invalid_index_root root))
      (fun tx node ~yield ->
        Index.fold_left ~from:prefix
          ~until:(fun key -> not (String.starts_with ~prefix key))
          tx
          (fun read _ address ->
            let* () = read in
            let* tuple = SI.get_req tx (S.Hash address) |>> TupleValue.decode in
            Ok (yield tuple) )
          (Ok ()) node
        |> Result.join )

  class substantial_relation connection value description =
    object (self)
      inherit Lifecycle.null
      method to_string = "relation"
      val connection : S.connection = connection
      val relation : t = value

      method contains (tuple : Concepts.Tuple.t) =
        SI.with_read connection (fun tx ->
            let open Utilities.Result in
            let* node = tuple_node tx relation in
            TupleSet.mem tx (Concepts.Tuple.hash tuple) node )

      method modes =
        Ok
          Concepts.(
            Mode.mode [] Cardinality.Finite
            :: Mode.mode (List.map fst (BatMap.String.bindings description)) (Cardinality.bounded 1)
            :: List.map (fun a -> Mode.mode [a] Cardinality.Finite) self#indexes )

      method indexes = List.map fst (BatMap.String.bindings relation.indexes)
      method lookup attribute value = lookup connection relation attribute value
      method describe () = Ok (Protocols.Schematics.Relation description)
      method enumerate = enumerate connection relation

      method protocols : Protocols.Handle.protocol list =
        [ Protocols.Addressable.make self;
          Protocols.Relation.make self;
          Protocols.Enumerable.make self;
          Protocols.Indexed.make self;
          Protocols.Schematics.make self ]

      method hash = encode relation |> Concepts.Hash.hash_of_blob
      method address = self#hash
    end

  let load connection relation =
    SI.with_read connection (fun tx -> schema_of tx relation.schematics)
    |> Result.map (fun description ->
        new substantial_relation connection relation description |> Protocols.Handle.make )

  let index tx attribute hashed =
    let open Utilities.Result in
    let* entries =
      List.map
        (fun (hash, tuple) ->
          match BatMap.String.find_opt attribute tuple.Concepts.Tuple.attributes with
          | Some value -> Ok (prefix value ^ Concepts.Hash.to_raw_string hash, hash)
          | None -> Error (Error.missing_attribute attribute) )
        hashed
      |> Utilities.List.sequence
    in
    Index.of_list tx entries |> Result.map (fun node -> attribute, Index.hash_of node)

  let store tx description tuples =
    let open Utilities.Result in
    let* schematics = store_schema tx description in
    let hashed = List.map (fun tuple -> Concepts.Tuple.hash tuple, tuple) tuples in
    let* node = TupleSet.of_list tx hashed in
    let* indexes =
      BatMap.String.bindings description
      |> List.map (fun (attribute, _) -> index tx attribute hashed)
      |> Utilities.List.sequence
    in
    let relation =
      {schematics; tuples= TupleSet.hash_of node; indexes= BatMap.String.of_list indexes}
    in
    let* _ = SI.store_blob tx (encode relation) in
    Ok relation

  let instantiate connection ~schematics =
    let open Utilities.Result in
    let* relation =
      SI.with_transaction connection (fun tx ->
          let* description = schema_of tx schematics in
          store tx description [] )
    in
    load connection relation
end