path: root/lib/kernel/BranchManager.ml

module Error = struct
  open Concepts.Condition

  let invalid_root head =
    condition "invalid-root"
      "The root hash for the database state 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 head))

  let refused violations =
    condition "refused" "The state is not admitted: a denial in it holds tuples"
      (List.fold_left
         (fun ps (path, tuples) ->
           ps
           & path
             |=| Concepts.Value.String
                   (String.concat "; " (List.map Concepts.Tuple.to_string tuples)) )
         empty violations )
end

let admitted state =
  let open Utilities.Result in
  let* admission = Protocols.Admission.require state in
  let* violations = Protocols.Admission.check admission in
  if violations = [] then Ok () else Error (Error.refused violations)

module Make (S : Abstract.Storage.STORAGE) = struct
  module M = Merkle.Make (S) (Merkle.StringKey)
  module SI = Storage.Make (S)
  module B = Branch.Make (S)

  let root_for tx label =
    let open Utilities.Result in
    let* label = S.get tx (S.Label label) in
    let data =
      Option.map
        (fun blob ->
          Concepts.Blob.bytes_of_blob blob |> Bytes.to_string |> Concepts.Hash.of_raw_string )
        label
    in
    Ok data

  let persist_root tx label addr =
    let data = Concepts.Hash.blob_of_hash addr in
    S.put tx (S.Label label) data

  class manager ?root storage label head =
    object (self)
      inherit Lifecycle.null
      inherit Identity.of_id
      method to_string = "branch-manager"
      val storage : S.connection = storage
      val label : string = label
      val head : M.node Atomic.t = Atomic.make head

      method fetch branch_name =
        SI.with_transaction storage (fun tx ->
            let head = Atomic.get head in
            M.lookup tx branch_name head )

      method protocols = Protocols.[Directory.make self; Registry.make self]
      method list = SI.with_transaction storage (fun tx -> Atomic.get head |> M.keys tx)

      method find key =
        let open Utilities.Result in
        SI.with_transaction storage (fun tx ->
            let* head = Atomic.get head |> M.lookup tx key in
            match head with
            | None -> Ok None
            | Some head -> B.load ?root tx storage head |> Result.map Option.some )

      (* a head moves only to an admitted state, judged before the head is
         taken, since judging may run stored programs that read storage *)
      method update key reference value =
        let open Utilities.Result in
        let open Protocols in
        let* () = Option.fold ~none:(Ok ()) ~some:admitted value in
        SI.with_transaction storage (fun tx ->
            match
              Utilities.Atomic.mswap head (fun head ->
                  let address_of h =
                    let* i = Handle.require Addressable.from h in
                    Ok (Addressable.address i)
                  in
                  let* branch = M.lookup tx key head |> Result.map_error (fun e -> Some e) in
                  let* reference =
                    Option.map address_of reference
                    |> Utilities.Option.sequence
                    |> Result.map_error (fun e -> Some e)
                  in
                  let* value =
                    Option.map address_of value
                    |> Utilities.Option.sequence
                    |> Result.map_error (fun e -> Some e)
                  in
                  if branch = reference then
                    begin
                      let* new_head =
                        match value with
                        | Some value -> M.insert tx key value head
                        | None -> M.remove tx key head
                      in
                      let* _ = persist_root tx label (M.hash_of new_head) in
                      Ok new_head
                    end
                    |> Result.map_error (fun e -> Some e)
                  else Error None )
            with
            | Ok _ -> Ok true
            | Error None -> Ok false
            | Error (Some e) -> Error e )
    end

  let make ?root storage label =
    let open Utilities.Result in
    SI.with_transaction storage (fun tx ->
        let* addr = root_for tx label in
        match addr with
        | None -> Ok (new manager ?root storage label M.empty |> Protocols.Handle.make)
        | Some addr ->
            let* head = M.find tx addr in
            let* head = Option.to_result ~none:(Error.invalid_root addr) head in
            Ok (new manager ?root storage label head |> Protocols.Handle.make) )
end